from __future__ import annotations import os import uuid from collections.abc import Iterable, Mapping from datetime import UTC, date, datetime, timedelta from typing import Any, Protocol from zoneinfo import ZoneInfo from google.cloud import bigquery from .month_end_arr import ( TARGET_OBJECT, MonthEndARRRecord, RowValidationError, bind_month_end_arr_account, map_month_end_arr_row, validate_target_contract, ) from .nexuscrm_client import NexusCRMClient, UpsertResult from .outbound_config import OutboundConfig from .outbound_state import BigQueryOutboundState class OutboundRunError(RuntimeError): def __init__(self, message: str, summary: dict[str, Any]) -> None: super().__init__(message) self.summary = summary def month_window(now: datetime) -> tuple[date, date]: """Previous month inclusive, next month exclusive, fixed at run startup.""" current = now.astimezone(ZoneInfo("Asia/Shanghai")).date().replace(day=1) previous = (current - timedelta(days=1)).replace(day=1) following = (current + timedelta(days=32)).replace(day=1) return previous, following class CRMClient(Protocol): def describe_object(self, object_name: str) -> dict[str, Any]: ... def resolve_account_ids(self, external_ids: list[str]) -> dict[str, str]: ... def upsert_collection( self, *, object_name: str, external_id_field: str, records: list[dict[str, Any]] ) -> list[UpsertResult]: ... class StateStore(Protocol): def ensure_tables(self) -> None: ... def acquire_lease(self, job_name: str, run_id: str, lease_seconds: int) -> None: ... def release_lease(self, job_name: str, run_id: str) -> None: ... def start_run(self, job_name: str, run_id: str) -> None: ... def finish_run( self, job_name: str, run_id: str, *, status: str, metrics: dict[str, Any], error_message: str | None = None, ) -> None: ... def latest_record_states(self, job_name: str) -> dict[str, dict[str, Any]]: ... def append_attempts(self, rows: list[dict[str, Any]]) -> None: ... def run_outbound( config: OutboundConfig, *, dry_run: bool = False, validate_only: bool = False, run_id: str | None = None, limit: int | None = None, source_rows: Iterable[Mapping[str, Any]] | None = None, bigquery_client: bigquery.Client | None = None, crm_client: CRMClient | None = None, state: StateStore | None = None, as_of: datetime | None = None, ) -> dict[str, Any]: if dry_run and validate_only: raise ValueError("--dry-run and --validate-only are mutually exclusive") if config.nexuscrm_object != TARGET_OBJECT: raise ValueError(f"month-end ARR outbound requires target.object={TARGET_OBJECT}") current_run_id = run_id or str(uuid.uuid4()) window_start, window_end = month_window(as_of or datetime.now(UTC)) if source_rows is None: client = bigquery_client or bigquery.Client(project=config.source_project, location=config.location) source_rows = _read_source_rows(client, config, limit=limit, window_start=window_start, window_end=window_end) records, source_count, skipped_blank_accounts = _map_and_validate_rows( source_rows, external_id_field=config.nexuscrm_external_id_field, ) if any( not window_start <= date.fromisoformat(record.payload["Revenue_Month__c"]) < window_end for record in records ): raise RowValidationError("source preflight failed: record outside current/previous Shanghai month window") summary: dict[str, Any] = { "job_name": config.name, "run_id": current_run_id, "mode": "dry_run" if dry_run else "validate_only" if validate_only else "write", "window_start": window_start.isoformat(), "window_end_exclusive": window_end.isoformat(), "window_timezone": "Asia/Shanghai", "source_rows": source_count, "skipped_blank_account": skipped_blank_accounts, "valid_rows": len(records), "unchanged_rows": 0, "attempted_rows": 0, "created_rows": 0, "updated_rows": 0, "failed_rows": 0, } if dry_run: summary["status"] = "validated" return summary owned_crm_client = crm_client is None target_client = crm_client or _build_crm_client(config) try: describe = target_client.describe_object(config.nexuscrm_object) validate_target_contract( describe, records, external_id_field=config.nexuscrm_external_id_field, required_fields=tuple(config.required_target_fields), ) records = _resolve_accounts(target_client, records) if validate_only: summary["status"] = "validated" return summary if state is None: bq_client = bigquery_client or bigquery.Client(project=config.state_project, location=config.location) target_state: StateStore = BigQueryOutboundState( bq_client, project=config.state_project, dataset=config.state_dataset, location=config.location, ) else: target_state = state return _write_records(config, target_client, target_state, current_run_id, records, summary) finally: if owned_crm_client and isinstance(target_client, NexusCRMClient): target_client.close() def _read_source_rows( client: bigquery.Client, config: OutboundConfig, *, limit: int | None, window_start: date, window_end: date ) -> Iterable[Mapping[str, Any]]: sql = config.source_query_path.read_text().strip().rstrip(";") if limit is not None: if limit < 1: raise ValueError("limit must be at least 1") sql = f"SELECT * FROM (\n{sql}\n) LIMIT {int(limit)}" job_config = bigquery.QueryJobConfig(query_parameters=[ bigquery.ScalarQueryParameter("window_start", "DATE", window_start), bigquery.ScalarQueryParameter("window_end", "DATE", window_end), ]) return client.query(sql, location=config.location, job_config=job_config).result() def _map_and_validate_rows( rows: Iterable[Mapping[str, Any]], *, external_id_field: str ) -> tuple[list[MonthEndARRRecord], int, int]: records: list[MonthEndARRRecord] = [] errors: list[str] = [] source_count = 0 skipped_blank_accounts = 0 keys: dict[str, int] = {} for row_number, row in enumerate(rows, start=1): source_count += 1 try: record = map_month_end_arr_row(row, external_id_field=external_id_field) except RowValidationError as error: if len(errors) < 20: errors.append(f"row {row_number}: {error}") continue if record is None: skipped_blank_accounts += 1 continue if record.external_key in keys: if len(errors) < 20: errors.append( f"row {row_number}: duplicate business key (first seen at row {keys[record.external_key]})" ) continue keys[record.external_key] = row_number records.append(record) if errors: suffix = " (showing first 20)" if len(errors) == 20 else "" raise RowValidationError(f"source preflight failed{suffix}: " + "; ".join(errors)) return records, source_count, skipped_blank_accounts def _write_records( config: OutboundConfig, crm_client: CRMClient, state: StateStore, run_id: str, records: list[MonthEndARRRecord], summary: dict[str, Any], ) -> dict[str, Any]: state.ensure_tables() lease_acquired = False run_started = False try: state.acquire_lease(config.name, run_id, config.lease_seconds) lease_acquired = True state.start_run(config.name, run_id) run_started = True prior = state.latest_record_states(config.name) pending = [ record for record in records if not ( prior.get(record.external_key, {}).get("status") == "succeeded" and prior[record.external_key].get("source_hash") == record.source_hash ) ] summary["unchanged_rows"] = len(records) - len(pending) for batch in _batches(pending, config.batch_size): results = crm_client.upsert_collection( object_name=config.nexuscrm_object, external_id_field=config.nexuscrm_external_id_field, records=[record.payload for record in batch], ) attempts = [ _attempt_row(config.name, run_id, record, result) for record, result in zip(batch, results, strict=True) ] state.append_attempts(attempts) summary["attempted_rows"] += len(batch) for result in results: if result.success and result.created is True: summary["created_rows"] += 1 elif result.success: summary["updated_rows"] += 1 else: summary["failed_rows"] += 1 if summary["failed_rows"]: summary["status"] = "failed" state.finish_run( config.name, run_id, status="failed", metrics=summary, error_message=f"{summary['failed_rows']} NexusCRM record writes failed", ) run_started = False raise OutboundRunError("one or more NexusCRM records failed", summary) summary["status"] = "succeeded" state.finish_run(config.name, run_id, status="succeeded", metrics=summary) run_started = False return summary except Exception as error: if run_started: summary["status"] = "failed" state.finish_run( config.name, run_id, status="failed", metrics=summary, error_message=str(error)[:1000], ) raise finally: if lease_acquired: state.release_lease(config.name, run_id) def _resolve_accounts(crm_client: CRMClient, records: list[MonthEndARRRecord]) -> list[MonthEndARRRecord]: external_ids = sorted({record.source_account_id for record in records}) resolved = crm_client.resolve_account_ids(external_ids) missing = sorted(set(external_ids) - resolved.keys()) if missing: preview = ", ".join(missing[:20]) suffix = " (showing first 20)" if len(missing) > 20 else "" raise RowValidationError(f"account_id values do not resolve to NexusCRM Account records{suffix}: {preview}") return [bind_month_end_arr_account(record, resolved[record.source_account_id]) for record in records] def _attempt_row( job_name: str, run_id: str, record: MonthEndARRRecord, result: UpsertResult ) -> dict[str, Any]: first_error = result.errors[0] if result.errors else {} return { "job_name": job_name, "run_id": run_id, "external_key": record.external_key, "source_hash": record.source_hash, "target_id": result.id or None, "status": "succeeded" if result.success else "failed", "created": result.created, "error_code": str(first_error.get("statusCode") or first_error.get("errorCode") or "") or None, "error_message": str(first_error.get("message") or "")[:1000] or None, } def _batches(records: list[MonthEndARRRecord], size: int) -> Iterable[list[MonthEndARRRecord]]: for start in range(0, len(records), size): yield records[start : start + size] def _build_crm_client(config: OutboundConfig) -> NexusCRMClient: base_url = os.environ.get(config.nexuscrm_base_url_env, config.nexuscrm_base_url).strip().rstrip("/") client_id = os.environ.get(config.nexuscrm_client_id_env, "").strip() client_secret = os.environ.get(config.nexuscrm_client_secret_env, "").strip() missing = [ name for name, value in ( (config.nexuscrm_client_id_env, client_id), (config.nexuscrm_client_secret_env, client_secret), ) if not value ] if missing: raise RuntimeError(f"missing NexusCRM credential environment variables: {', '.join(missing)}") return NexusCRMClient( base_url=base_url, client_id=client_id, client_secret=client_secret, api_version=config.nexuscrm_api_version, timeout_seconds=config.request_timeout_seconds, max_attempts=config.max_http_attempts, )