"""Helpers for reading Alibaba Cloud account metadata.""" from __future__ import annotations import csv from pathlib import Path from typing import Any, Dict, Optional, Set, Tuple ALIBABA_DIR = Path(__file__).resolve().parent _TRUTHY_MARKERS = { "true", "1", "yes", "y", "on", "active", "enabled", } _TRUTHY_NATIVE = {"是", "启用"} def _normalise_flag(value: Optional[str]) -> bool: if value is None: return False raw = str(value).strip() if not raw: return False lowered = raw.lower() if lowered in _TRUTHY_MARKERS: return True return raw in _TRUTHY_NATIVE def resolve_account_details_path(account_details_file: Optional[str] = None) -> Optional[Path]: candidates = [] if account_details_file: path = Path(account_details_file) if not path.is_absolute(): path = ALIBABA_DIR / path candidates.append(path) candidates.append(ALIBABA_DIR / "alibaba_cloud_account_details.csv") for path in candidates: if path.exists(): return path return None def load_account_metadata( account_details_file: Optional[str] = None, ) -> Optional[Tuple[Dict[str, Dict[str, Any]], Path]]: """Read the account metadata CSV and return structured records with the file path.""" details_path = resolve_account_details_path(account_details_file) if details_path is None: print("Warning: Alibaba Cloud account details file not found. Skipping account filtering.") return None metadata: Dict[str, Dict[str, Any]] = {} try: with details_path.open("r", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) if not reader.fieldnames or "account_id" not in reader.fieldnames: print( "Warning: account details file missing 'account_id' column: " f"{details_path}" ) return None for row in reader: account_id = (row.get("account_id") or "").strip() if not account_id: continue metadata[account_id] = { "account_name": (row.get("account_name") or row.get("accountName") or "").strip(), "owner": (row.get("owner") or row.get("owner_account") or "").strip(), "description": (row.get("description") or "").strip(), "is_active": _normalise_flag(row.get("is_active") or row.get("isActive")), } except Exception as exc: # pragma: no cover - defensive logging only print(f"Warning: Failed to load Alibaba Cloud account details from {details_path}: {exc}") return None return metadata, details_path def load_active_accounts(account_details_file: Optional[str] = None) -> Optional[Set[str]]: metadata_result = load_account_metadata(account_details_file) if metadata_result is None: return None metadata, details_path = metadata_result active_accounts: Set[str] = { account_id for account_id, profile in metadata.items() if profile.get("is_active") } if not active_accounts: print(f"Warning: No active Alibaba Cloud accounts marked in {details_path}") else: print(f"Loaded {len(active_accounts)} active Alibaba Cloud accounts from {details_path.name}") return active_accounts