"""High-level Alibaba Cloud billing analyzer.""" from __future__ import annotations import csv import json from collections import defaultdict from datetime import date, datetime, timedelta from pathlib import Path from typing import Dict, Iterable, List, Optional, Sequence, Tuple from ..export.environments import EnvironmentManager from .anomaly import AnomalyDetector from ...account_details import load_account_metadata from ..notification.feishu import FeishuNotifier ALIBABA_DIR = Path(__file__).resolve().parents[2] class AlibabaCloudBillingAnalyzer: """Load exported CSVs, calculate baselines, and raise alerts.""" def __init__( self, config_path: Optional[str] = None, data_dir: Optional[str] = None, ) -> None: self.config = self._load_config(config_path) data_dir_value = data_dir or self.config.get("dataDir") or "alibaba_cloud_billing_exports" self.data_dir = self._resolve_path(data_dir_value) default_env = self.config.get("defaultEnvironment", "unassigned") self.environments_config: Dict[str, Dict] = self.config.get("environments", {}) self.env_manager = EnvironmentManager( environments_config=self.environments_config, output_dir=None, default_environment=default_env, cloud_webhook_url=self.config.get("cloudWebhookUrl") or self.config.get("cloud_webhook_url"), ) metadata_result = load_account_metadata(self.config.get("accountDetailsFile")) self.account_profiles: Dict[str, Dict] = {} self.active_accounts = None if metadata_result: self.account_profiles, details_path = metadata_result active_ids = { account_id for account_id, profile in self.account_profiles.items() if profile.get("is_active") } self.active_accounts = active_ids if active_ids: print( f"Loaded {len(active_ids)} active Alibaba Cloud accounts " f"from {details_path.name}" ) else: print(f"Warning: No active Alibaba Cloud accounts marked in {details_path}") self.default_thresholds = { "accountWarning": 1.5, "accountCritical": 2.0, "serviceWarning": 2.0, "serviceCritical": 3.0, "minCostThreshold": 10.0, "newServiceThreshold": 20.0, } # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def analyze( self, target_date: Optional[str] = None, environments: Optional[Sequence[str]] = None, account_ids: Optional[Sequence[str]] = None, dry_run: bool = False, send_notifications: bool = True, ) -> Dict[str, Dict]: if target_date: target = datetime.strptime(target_date, "%Y-%m-%d").date() else: target = datetime.now().date() - timedelta(days=2) environment_names = self._normalize_environments(environments) account_filters = {str(a).strip() for a in account_ids or [] if str(a).strip()} results: Dict[str, Dict] = {} for env_name in environment_names: env_result = self._analyze_environment(env_name, target, account_filters) results[env_name] = env_result webhook = self.env_manager.webhook_for_environment(env_name) display_name = self.env_manager.display_name_for_environment(env_name) is_default_env = env_name == self.env_manager.default_environment # Skip notification for default environment (accounts are analyzed but not alerted) if is_default_env: continue if send_notifications and webhook: notifier = FeishuNotifier(webhook_url=webhook, environment_name=display_name) notifier.send(env_result, dry_run=dry_run) return results # ------------------------------------------------------------------ # Environment analysis # ------------------------------------------------------------------ def _analyze_environment( self, environment: str, target_date: date, account_filters: Iterable[str], ) -> Dict: current_date = target_date.isoformat() historical_start = target_date - timedelta(days=30) thresholds = self._resolve_thresholds(environment) detector = AnomalyDetector(thresholds) account_records = self._load_environment_records( environment, months=self._months_to_load(target_date), ) findings = [] total_cost = 0.0 total_anomalies = 0 currency_hint = "" service_totals = defaultdict(float) for account_id, records in account_records.items(): if account_filters and account_id not in account_filters: continue if self.active_accounts is not None and account_id not in self.active_accounts: continue current_records = [row for row in records if row["date"] == current_date] if not current_records: continue currency_hint = current_records[0].get("currency", "") account_name = self._account_display_name( account_id, current_records[0].get("account_name", account_id), ) account_cost = sum(float(row["cost"]) for row in current_records) total_cost += account_cost service_costs = defaultdict(float) for row in current_records: service_costs[row["product"]] += float(row["cost"]) historical_records = [ row for row in records if historical_start <= datetime.strptime(row["date"], "%Y-%m-%d").date() < target_date ] account_baseline = detector.account_baseline(historical_records) account_anomaly = detector.detect_account(account_cost, account_baseline) service_baselines = detector.service_baselines(historical_records) service_anomalies = [] for service, cost in service_costs.items(): anomaly = detector.detect_service(service, cost, service_baselines) if anomaly: anomaly.update({"service": service, "cost": cost}) service_anomalies.append(anomaly) new_services = detector.detect_new_services(service_costs, service_baselines) total_anomalies += len(service_anomalies) if account_anomaly: total_anomalies += 1 for service_name, cost in service_costs.items(): service_totals[service_name] += cost findings.append( { "account_id": account_id, "account_name": account_name, "date": current_date, "currency": currency_hint, "total_cost": account_cost, "baseline": account_baseline, "account_anomaly": account_anomaly, "service_costs": dict(service_costs), "service_anomalies": service_anomalies, "new_services": new_services, } ) findings.sort(key=lambda item: item["total_cost"], reverse=True) top_accounts = [ { "account_id": item["account_id"], "account_name": item["account_name"], "total_cost": item["total_cost"], } for item in findings[:5] ] top_services = [ {"service": name, "total_cost": cost} for name, cost in sorted(service_totals.items(), key=lambda kv: kv[1], reverse=True)[:5] ] return { "environment": environment, "environment_name": self.env_manager.display_name_for_environment(environment), "target_date": current_date, "currency": currency_hint or self.config.get("settings", {}).get("defaultCurrency", "CNY"), "total_cost": total_cost, "total_anomalies": total_anomalies, "findings": findings, "top_accounts": top_accounts, "top_services": top_services, "success": True, } def _account_display_name(self, account_id: str, fallback: str) -> str: profile = self.account_profiles.get(account_id) if profile: name = (profile.get("account_name") or "").strip() if name: return name return fallback or account_id # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ def _resolve_path(self, candidate: str) -> Path: path = Path(candidate) if not path.is_absolute(): path = ALIBABA_DIR / path return path def _load_config(self, config_path: Optional[str]) -> Dict: if config_path: path = Path(config_path) if not path.is_absolute(): candidate = Path.cwd() / path if candidate.exists(): path = candidate else: path = ALIBABA_DIR / path else: path = ALIBABA_DIR / "config.json" if not path.exists(): print(f"Warning: Alibaba Cloud analyzer config not found at {path}, using defaults.") return {} with path.open("r", encoding="utf-8") as handle: return json.load(handle) def _normalize_environments(self, environments: Optional[Sequence[str]]) -> List[str]: available = self.env_manager.list_environments() if not environments or "all" in environments: return available resolved = [env for env in environments if env in available] if not resolved: raise ValueError(f"Unknown environments {environments}. Available: {', '.join(available)}") return resolved def _months_to_load(self, target_date: date) -> List[str]: months = {target_date.strftime("%Y-%m")} previous = (target_date.replace(day=1) - timedelta(days=1)).strftime("%Y-%m") months.add(previous) historical = (target_date - timedelta(days=30)).strftime("%Y-%m") months.add(historical) return list(months) def _load_environment_records(self, environment: str, months: Iterable[str]) -> Dict[str, List[Dict]]: env_dir = self.data_dir / environment if not env_dir.exists(): return {} target_months = set(months) records: Dict[str, List[Dict]] = defaultdict(list) for csv_path in env_dir.glob("*.csv"): try: with csv_path.open("r", encoding="utf-8") as handle: reader = csv.DictReader(handle) for row in reader: month = row.get("date", "")[:7] if target_months and month not in target_months: continue account_id = (row.get("account_id") or "").strip() if not account_id: continue record = { "date": row.get("date", ""), "product": row.get("product", ""), "cost": float(row.get("cost") or 0), "currency": row.get("currency", ""), "account_name": row.get("account_name", account_id), } records[account_id].append(record) except Exception as exc: # pragma: no cover - defensive print(f"Warning: failed to load {csv_path}: {exc}") return records def _resolve_thresholds(self, environment: str) -> Dict[str, float]: env_config = self.environments_config.get(environment, {}) thresholds = env_config.get("thresholds", {}) combined = self.default_thresholds.copy() combined.update(thresholds) return combined