"""Utilities for synchronizing exported billing data into MySQL.""" from __future__ import annotations from collections import defaultdict from datetime import date from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple from urllib.parse import quote_plus import ssl from sqlalchemy import create_engine, text from sqlalchemy.engine import Engine from sqlalchemy.exc import SQLAlchemyError def _safe_decimal(value: Any) -> Decimal: """Convert arbitrary numeric input into Decimal safely.""" if value is None: return Decimal("0") if isinstance(value, Decimal): return value try: return Decimal(str(value)) except (InvalidOperation, ValueError, TypeError): return Decimal("0") def _first_day(input_date: date) -> date: """Normalize a date to the first day of its month.""" return input_date.replace(day=1) class BillingDataUploader: """Pushes normalized billing exports into MySQL tables.""" def __init__(self, provider: str, settings: Dict[str, Any]): self.provider = provider self.settings = settings or {} self.enabled = bool(self.settings.get("enabled")) self.currency = self.settings.get("defaultCurrency", "USD") self.engine: Optional[Engine] = None if self.enabled: self._init_engine() @classmethod def from_config(cls, config_manager, provider: str) -> "BillingDataUploader": """Instantiate uploader using a ConfigManager instance.""" settings = config_manager.get("mysqlConfig", {}) or {} return cls(provider=provider, settings=settings) def _init_engine(self) -> None: """Create a SQLAlchemy engine based on settings.""" user = self.settings.get("username") password = self.settings.get("password", "") host = self.settings.get("host", "127.0.0.1") port = self.settings.get("port", 3306) database = self.settings.get("databaseName") if not user or not database: print("⚠️ MySQL 同步未启用: 配置缺少用户名或数据库") self.enabled = False return if password is None: password = "" query_params = {"charset": "utf8mb4"} connect_args: Dict[str, Any] = {} if self.settings.get("useTls"): ca_path_setting = self.settings.get("caCertificatePath") ca_path: Optional[str] = None default_ca_candidates = [ Path("/etc/ssl/certs/ca-certificates.crt"), # Debian / Ubuntu Path("/etc/ssl/cert.pem"), # macOS / Alpine Path("/etc/pki/tls/certs/ca-bundle.crt"), # RedHat / Fedora Path("/etc/ssl/ca-bundle.pem"), # openSUSE ] if ca_path_setting: provided_path = Path(ca_path_setting).expanduser() if provided_path.exists(): ca_path = str(provided_path) else: print( f"⚠️ 指定的 CA 证书路径不存在 ({provided_path}); " "尝试使用系统默认路径。" ) if not ca_path: for candidate in default_ca_candidates: if candidate.exists(): ca_path = str(candidate) break if not ca_path: print("✗ MySQL TLS 已启用，但缺少 CA 证书路径 (caCertificatePath)。") self.enabled = False return client_cert = self.settings.get("clientCertificatePath") client_key = self.settings.get("clientKeyPath") verify_identity = self.settings.get("verifyServerIdentity", True) ssl_options: Dict[str, Any] = {"ca": ca_path, "cert_reqs": ssl.CERT_REQUIRED} if client_cert and client_key: ssl_options["cert"] = client_cert ssl_options["key"] = client_key if verify_identity: ssl_options["check_hostname"] = True connect_args["ssl"] = ssl_options print("🔒 已启用 MySQL TLS，并校验服务器证书。") query_string = "&".join(f"{key}={value}" for key, value in query_params.items()) safe_user = quote_plus(user) safe_password = quote_plus(password) url = f"mysql+pymysql://{safe_user}:{safe_password}@{host}:{port}/{database}?{query_string}" try: self.engine = create_engine( url, pool_pre_ping=True, pool_recycle=3600, pool_size=self.settings.get("poolSize", 5), max_overflow=self.settings.get("maxOverflow", 5), connect_args=connect_args, ) except SQLAlchemyError as exc: print(f"✗ 无法初始化 MySQL 连接: {exc}") self.enabled = False # Public API --------------------------------------------------------- def sync_account_costs( self, account_identifier: str, account_name: str, environment: str, records: Iterable[Dict[str, Any]], start_date: date, end_date: date, source_file: Optional[Path] = None, metadata: Optional[Dict[str, Any]] = None, ) -> None: """Persist daily and monthly aggregates for an account.""" if not self.enabled or not self.engine: return prepared = self._prepare_records(records) if not prepared: self._record_run( account_identifier, environment, start_date, end_date, processed_rows=0, status="skipped", message="no rows to ingest", source_file=source_file, ) return try: with self.engine.begin() as conn: account_id = self._upsert_account( conn, account_identifier=account_identifier, account_name=account_name, environment=environment, metadata=metadata or {}, ) self._upsert_daily_costs(conn, account_id, environment, prepared) self._upsert_monthly_summary(conn, account_id, environment, prepared) self._record_run( account_identifier, environment, start_date, end_date, processed_rows=len(prepared), status="success", source_file=source_file, metadata=metadata, connection=conn, ) except SQLAlchemyError as exc: print(f"✗ MySQL 同步失败 ({self.provider} {account_identifier}): {exc}") self._record_run( account_identifier, environment, start_date, end_date, processed_rows=len(prepared), status="failed", message=str(exc), source_file=source_file, ) # Internal helpers --------------------------------------------------- def _prepare_records(self, records: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]: """Normalize export rows into the format needed for inserts.""" normalized: List[Dict[str, Any]] = [] for row in records: raw_date = row.get("date") product = row.get("product") or row.get("service") or "Uncategorized" cost_value = _safe_decimal(row.get("cost")) if not raw_date: continue try: parsed_date = date.fromisoformat(str(raw_date)) except ValueError: continue normalized.append( { "billing_date": parsed_date, "service_category": product, "cost": cost_value, "currency": row.get("currency", self.currency), } ) return normalized def _environment_group(self, environment: str) -> str: prod_aliases = {"prod", "production", "misplaced_prod"} if environment in prod_aliases: return "prod" return "dev_test" def _upsert_account( self, connection, account_identifier: str, account_name: str, environment: str, metadata: Dict[str, Any], ) -> int: """Ensure the account row exists and return its primary key.""" group = self._environment_group(environment) params = { "provider": self.provider, "account_identifier": account_identifier, "account_name": account_name, "environment": environment, "environment_group": group, "currency": metadata.get("currency", self.currency), "is_active": metadata.get("is_active", True), } connection.execute( text( """ INSERT INTO billing_accounts ( provider, account_identifier, account_name, environment, environment_group, currency, is_active, created_at, updated_at ) VALUES ( :provider, :account_identifier, :account_name, :environment, :environment_group, :currency, :is_active, NOW(), NOW() ) ON DUPLICATE KEY UPDATE account_name = VALUES(account_name), environment = VALUES(environment), environment_group = VALUES(environment_group), currency = VALUES(currency), is_active = VALUES(is_active), updated_at = NOW() """ ), params, ) account_id = connection.execute( text( """ SELECT id FROM billing_accounts WHERE provider = :provider AND account_identifier = :account_identifier LIMIT 1 """ ), { "provider": self.provider, "account_identifier": account_identifier, }, ).scalar_one() return int(account_id) def _upsert_daily_costs( self, connection, account_id: int, environment: str, rows: List[Dict[str, Any]], ) -> None: """Insert or update daily costs for the account.""" buckets: Dict[Tuple[date, str], Dict[str, Any]] = {} for row in rows: key = (row["billing_date"], row["service_category"]) if key not in buckets: buckets[key] = { "billing_date": row["billing_date"], "service_category": row["service_category"], "cost": Decimal("0"), "currency": row["currency"], } buckets[key]["cost"] += row["cost"] for bucket in buckets.values(): params = { "account_id": account_id, "provider": self.provider, "environment": environment, "billing_date": bucket["billing_date"], "service_category": bucket["service_category"], "cost": bucket["cost"], "currency": bucket["currency"], } connection.execute( text( """ INSERT INTO billing_daily_costs ( account_id, provider, environment, billing_date, service_category, cost, currency, created_at, updated_at ) VALUES ( :account_id, :provider, :environment, :billing_date, :service_category, :cost, :currency, NOW(), NOW() ) ON DUPLICATE KEY UPDATE cost = VALUES(cost), currency = VALUES(currency), environment = VALUES(environment), updated_at = NOW() """ ), params, ) def _upsert_monthly_summary( self, connection, account_id: int, environment: str, rows: List[Dict[str, Any]], ) -> None: """Insert or update monthly service-level totals.""" buckets: Dict[Tuple[date, str], Dict[str, Any]] = defaultdict( lambda: {"cost": Decimal("0"), "currency": self.currency} ) for row in rows: first_day = _first_day(row["billing_date"]) key = (first_day, row["service_category"]) buckets[key]["cost"] += row["cost"] buckets[key]["currency"] = row["currency"] for (month_start, service_category), payload in buckets.items(): params = { "account_id": account_id, "provider": self.provider, "environment": environment, "billing_month": month_start, "service_category": service_category, "cost": payload["cost"], "currency": payload["currency"], "credits": Decimal("0"), "net_cost": payload["cost"], } connection.execute( text( """ INSERT INTO billing_monthly_summary ( account_id, provider, environment, billing_month, service_category, cost, credits, net_cost, currency, created_at, updated_at ) VALUES ( :account_id, :provider, :environment, :billing_month, :service_category, :cost, :credits, :net_cost, :currency, NOW(), NOW() ) ON DUPLICATE KEY UPDATE cost = VALUES(cost), credits = VALUES(credits), net_cost = VALUES(net_cost), currency = VALUES(currency), environment = VALUES(environment), updated_at = NOW() """ ), params, ) def _record_run( self, account_identifier: str, environment: str, start_date: date, end_date: date, processed_rows: int, status: str, message: Optional[str] = None, source_file: Optional[Path] = None, metadata: Optional[Dict[str, Any]] = None, connection=None, ) -> None: """Persist export run metadata for observability.""" if not self.enabled or not self.engine: return payload = { "provider": self.provider, "account_identifier": account_identifier, "environment": environment, "export_start": start_date, "export_end": end_date, "source_file": str(source_file) if source_file else None, "status": status, "processed_rows": processed_rows, "message": message, "extra": (metadata or {}).get("notes"), } stmt = text( """ INSERT INTO billing_export_runs ( provider, account_identifier, environment, export_start, export_end, source_file, status, processed_rows, message, extra, created_at ) VALUES ( :provider, :account_identifier, :environment, :export_start, :export_end, :source_file, :status, :processed_rows, :message, :extra, NOW() ) """ ) if connection is not None: connection.execute(stmt, payload) return with self.engine.begin() as conn: conn.execute(stmt, payload)