from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Any

import yaml

from ingestion.common.env import load_secrets


@dataclass(frozen=True)
class SourceConfig:
    """Standalone BigQuery extract-source definition.

    Mirrors the TiDB Cloud extract config: it declares which BigQuery tables to snapshot,
    where to land the Parquet export in GCS, and the governance rules used to classify
    tables. BigQuery is an externally-managed source — we do not own the table — so this
    stage owns only the GCS-snapshot *source* contract for the objects it writes (ADR 0015).
    """

    path: Path
    data: dict[str, Any]

    @property
    def root(self) -> Path:
        return self.path.parent

    @property
    def source_name(self) -> str:
        return self.data["source"]["name"]

    @property
    def project(self) -> str:
        """The Google Cloud project that owns the source dataset/table."""
        return str(self.data["source"]["project"])

    @property
    def dataset(self) -> str:
        return str(self.data["source"]["dataset"])

    @property
    def domain(self) -> str:
        return self.data["source"]["domain"]

    @property
    def owner(self) -> str:
        return self.data["source"]["owner"]

    @property
    def gcloud_project(self) -> str:
        """Billing/quota project for ``bq`` and ``gcloud storage`` calls. Defaults to the
        source project, but may differ when the GCS bucket lives in another project."""
        return str(self.data.get("gcloud", {}).get("project") or self.project)

    @property
    def watermark_candidates(self) -> list[str]:
        return list(self.data["source"].get("watermark_candidates", []))

    @property
    def declared_primary_keys(self) -> dict[str, list[str]]:
        """BigQuery has no enforced primary keys, so a snapshot source declares them per
        table here; the loader needs a key for dedup detection on the raw snapshot."""
        return {k: list(v) for k, v in self.data["source"].get("primary_keys", {}).items()}

    def primary_key_for(self, table_name: str) -> list[str]:
        return list(self.declared_primary_keys.get(table_name, []))

    @property
    def selected_tables(self) -> list[str]:
        return list(self.data.get("tables", {}).get("include", []))

    @property
    def excluded_tables(self) -> list[str]:
        return list(self.data.get("tables", {}).get("exclude", []))

    @property
    def require_primary_key(self) -> bool:
        return bool(self.data.get("tables", {}).get("require_primary_key", False))

    @property
    def classification_rules(self) -> dict[str, Any]:
        governance = self.data.get("governance", {})
        return {
            "baseline": list(governance.get("classifications", [])),
            "patterns": dict(governance.get("classification_patterns", {})),
            "name_rules": list(governance.get("classification_name_rules", [])),
        }

    @property
    def gcs_bucket(self) -> str:
        return self.data["gcs"]["bucket"]

    @property
    def gcs_location(self) -> str:
        return self.data["gcs"]["location"]

    @property
    def snapshot_prefix(self) -> str:
        return self.data["gcs"]["snapshot_prefix"].rstrip("/")

    @property
    def manifest_prefix(self) -> str:
        return self.data["gcs"]["manifest_prefix"].rstrip("/")

    @property
    def runtime_log_prefix(self) -> str:
        return self.data["gcs"]["runtime_log_prefix"].rstrip("/")

    @property
    def compression(self) -> str:
        return self.data.get("export", {}).get("compression", "ZSTD")

    @property
    def contracts_output_dir(self) -> str:
        return self.data.get("contracts", {}).get("output_dir", "contracts/generated")

    def gcs_run_uri(self, run_date: str, run_id: str) -> str:
        return (
            f"gs://{self.gcs_bucket}/{self.snapshot_prefix}/"
            f"dataset={self.dataset}/run_date={run_date}/run_id={run_id}/"
        )

    def table_gcs_uri(self, table: str, run_date: str, run_id: str) -> str:
        return f"{self.gcs_run_uri(run_date, run_id)}table={table}/"

    def manifest_gcs_uri(self, run_date: str, run_id: str) -> str:
        return (
            f"gs://{self.gcs_bucket}/{self.manifest_prefix}/"
            f"dataset={self.dataset}/run_date={run_date}/run_id={run_id}/manifest.json"
        )

    def runtime_log_gcs_uri(self, run_date: str, run_id: str) -> str:
        return (
            f"gs://{self.gcs_bucket}/{self.runtime_log_prefix}/"
            f"dataset={self.dataset}/run_date={run_date}/run_id={run_id}/events.jsonl"
        )


def load_source_config(path: str | Path) -> SourceConfig:
    config_path = Path(path)
    data = yaml.safe_load(config_path.read_text())
    secrets_file = data["source"].get("secrets_file", "secrets.env")
    load_secrets(config_path.parent / secrets_file)
    return SourceConfig(path=config_path, data=data)