"""Seed requestable DAX BigQuery datasets from an ops JSON inventory.

Input can be either a JSON array or an object with a top-level ``datasets``
array. Each entry uses human-operable owner emails and dataset locators:

[
  {
    "project_id": "gcp-digital-bi-dev",
    "dataset_id": "dwh_core",
    "owner_email": "owner@pingcap.com",
    "display_name": "DWH Core",
    "description": "Curated BI warehouse core dataset.",
    "labels": {"risk": "finance"},
    "requestable": true
  }
]

Run inside the backend environment after migrations and identity sync:

    /app/.venv/bin/python scripts/seed_data_access_resources.py \
      /tmp/dax-datasets.json --integration-slug dax-bigquery-prod --commit
"""

from __future__ import annotations

import argparse
import asyncio
import hashlib
import json
import sys
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from sqlalchemy import func, select

# Register mapper classes before touching relationships.
import app.approval.models  # noqa: F401
import app.audit.models  # noqa: F401
import app.auth.models  # noqa: F401
import app.domains.apikey.models  # noqa: F401
import app.domains.asset.models  # noqa: F401
import app.domains.cmdb.models  # noqa: F401
import app.domains.data_access.models  # noqa: F401
import app.domains.email_group.models  # noqa: F401
import app.domains.hr.models  # noqa: F401
import app.domains.offboarding.models  # noqa: F401
import app.domains.permission.models  # noqa: F401
import app.domains.provisioning.models  # noqa: F401
import app.identity.person_models  # noqa: F401
from app.core.database import async_session_factory
from app.domains.data_access import constants as c
from app.domains.data_access.catalog_feed import (
    OWNER_RESOLUTION_RESOLVED,
    OWNER_RESOLUTION_UNRESOLVED,
    project_catalog_dataset_feed,
)
from app.domains.data_access.lineage_inventory import sync_lineage_root_datasets
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


def _load_inventory(
    path: Path,
    *,
    lineage_catalog_path: Path | None = None,
) -> list[dict[str, Any]]:
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        print(f"ERROR: failed to read {path}: {exc}", file=sys.stderr)
        sys.exit(2)
    if lineage_catalog_path is not None:
        if not isinstance(raw, dict):
            print(
                "ERROR: automatic lineage-root projection requires an "
                "inventory object with a datasets array.",
                file=sys.stderr,
            )
            sys.exit(2)
        try:
            lineage_catalog = json.loads(
                lineage_catalog_path.read_text(encoding="utf-8"),
            )
        except (OSError, json.JSONDecodeError) as exc:
            print(
                f"ERROR: failed to read {lineage_catalog_path}: {exc}",
                file=sys.stderr,
            )
            sys.exit(2)
        if not isinstance(lineage_catalog, dict):
            print("ERROR: lineage catalog must be a JSON object.", file=sys.stderr)
            sys.exit(2)
        try:
            raw, additions = sync_lineage_root_datasets(raw, lineage_catalog)
        except ValueError as exc:
            print(
                f"ERROR: automatic lineage-root projection failed: {exc}",
                file=sys.stderr,
            )
            sys.exit(2)
        print(
            "AUTOMATIC LINEAGE ROOTS: "
            f"referenced={raw['metadata']['lineage_root_dataset_count']} "
            f"added={len(additions)}",
        )
    if isinstance(raw, dict):
        raw = raw.get("datasets")
    if not isinstance(raw, list):
        print(
            "ERROR: inventory must be a JSON array or an object with a datasets array.",
            file=sys.stderr,
        )
        sys.exit(2)
    out: list[dict[str, Any]] = []
    for index, item in enumerate(raw):
        if not isinstance(item, dict):
            print(f"ERROR: dataset #{index} must be an object.", file=sys.stderr)
            sys.exit(2)
        out.append(dict(item))
    return out


def _required_string(item: dict[str, Any], key: str, *, index: int) -> str:
    value = str(item.get(key) or "").strip()
    if not value:
        print(f"ERROR: dataset #{index} missing required {key}.", file=sys.stderr)
        sys.exit(2)
    return value


def _source_etag(item: dict[str, Any]) -> str:
    explicit = str(item.get("source_etag") or "").strip()
    if explicit:
        return explicit
    stable_payload = {key: value for key, value in item.items() if key not in {"source_etag"}}
    digest = hashlib.sha256(
        json.dumps(
            stable_payload,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        ).encode(),
    ).hexdigest()
    return f"manual:{digest}"


def _optional_bool(value: Any, *, default: bool, field: str, index: int) -> bool:
    if value is None:
        return default
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        normalized = value.strip().lower()
        if normalized in {"1", "true", "yes", "on"}:
            return True
        if normalized in {"0", "false", "no", "off"}:
            return False
    print(
        f"ERROR: dataset #{index} {field} must be a boolean.",
        file=sys.stderr,
    )
    sys.exit(2)


async def _integration_id(*, integration_slug: str | None) -> str:
    async with async_session_factory() as db:
        stmt = select(Integration).where(
            Integration.kind == c.DOMAIN_DATA_ACCESS,
            Integration.provider == c.PROVIDER_BIGQUERY,
            Integration.enabled.is_(True),
        )
        if integration_slug:
            stmt = stmt.where(Integration.slug == integration_slug)
        rows = list((await db.scalars(stmt)).all())
    if not rows:
        target = integration_slug or "enabled data_access/bigquery integration"
        print(f"ERROR: no {target} found.", file=sys.stderr)
        sys.exit(3)
    if len(rows) > 1 and not integration_slug:
        print(
            "ERROR: multiple enabled data_access/bigquery integrations found; "
            "pass --integration-slug.",
            file=sys.stderr,
        )
        sys.exit(3)
    return str(rows[0].id)


async def _owner_person_id(owner_email: str) -> str | None:
    if not owner_email:
        return None
    async with async_session_factory() as db:
        person = await db.scalar(
            select(IdentityPerson).where(
                func.lower(IdentityPerson.work_email) == owner_email.lower(),
            ),
        )
    if person is None:
        return None
    return str(person.id)


async def _catalog_entries(
    raw_items: list[dict[str, Any]],
    *,
    source_version: str,
    allow_unresolved: bool,
    unresolved_owner_fallback_email: str | None,
) -> list[dict[str, Any]]:
    fallback_email = str(unresolved_owner_fallback_email or "").strip().casefold()
    fallback_person_id: str | None = None
    if fallback_email:
        if "@" not in fallback_email:
            print(
                "ERROR: unresolved owner fallback must be a valid email.",
                file=sys.stderr,
            )
            sys.exit(4)
        fallback_person_id = await _owner_person_id(fallback_email)
        if fallback_person_id is None:
            print(
                "ERROR: unresolved owner fallback email did not resolve to "
                "identity_persons.work_email.",
                file=sys.stderr,
            )
            sys.exit(4)

    entries: list[dict[str, Any]] = []
    fallback_count = 0
    for index, item in enumerate(raw_items):
        project_id = _required_string(item, "project_id", index=index)
        dataset_id = _required_string(item, "dataset_id", index=index)
        owner_email = str(item.get("owner_email") or "").strip()
        owner_person_id = await _owner_person_id(owner_email)
        source_etag_item = item
        if owner_person_id is None and fallback_person_id is not None:
            owner_person_id = fallback_person_id
            source_etag_item = {**item, "owner_email": fallback_email}
            fallback_count += 1
        if owner_person_id is None and not allow_unresolved:
            print(
                f"ERROR: dataset #{index} owner_email={owner_email!r} "
                "did not resolve to identity_persons.work_email. "
                "Run identity sync or pass --allow-unresolved.",
                file=sys.stderr,
            )
            sys.exit(4)
        entry = {
            "provider": c.PROVIDER_BIGQUERY,
            "resource_type": c.RESOURCE_TYPE_DATASET,
            "locator": {
                "project_id": project_id,
                "dataset_id": dataset_id,
            },
            "catalog_id": item.get("catalog_id") or f"{project_id}.{dataset_id}",
            c.FIELD_OWNER_PERSON_ID: owner_person_id,
            "owner_resolution_status": (
                OWNER_RESOLUTION_RESOLVED
                if owner_person_id is not None
                else OWNER_RESOLUTION_UNRESOLVED
            ),
            c.FIELD_SOURCE_VERSION: str(
                item.get("source_version") or source_version,
            ),
            c.FIELD_SOURCE_ETAG: _source_etag(source_etag_item),
            "display_name": item.get("display_name"),
            "description": item.get("description"),
            "labels": item.get("labels"),
            "requestable": _optional_bool(
                item.get("requestable"),
                default=True,
                field="requestable",
                index=index,
            ),
            "status": str(item.get("status") or "active"),
        }
        entries.append(entry)
    if fallback_count:
        print(
            f"OWNER FALLBACKS: count={fallback_count} fallback_email={fallback_email}",
        )
    return entries


async def _run(
    *,
    inventory_path: Path,
    lineage_catalog_path: Path | None,
    integration_slug: str | None,
    source_version: str,
    allow_unresolved: bool,
    unresolved_owner_fallback_email: str | None,
    commit: bool,
) -> None:
    raw_items = _load_inventory(
        inventory_path,
        lineage_catalog_path=lineage_catalog_path,
    )
    integration_id = await _integration_id(integration_slug=integration_slug)
    entries = await _catalog_entries(
        raw_items,
        source_version=source_version,
        allow_unresolved=allow_unresolved,
        unresolved_owner_fallback_email=unresolved_owner_fallback_email,
    )
    async with async_session_factory() as db:
        result = await project_catalog_dataset_feed(
            db,
            integration_id=integration_id,
            entries=entries,
            trace_id=f"dax_dataset_seed:{source_version}",
        )
        if commit:
            await db.commit()
        else:
            await db.rollback()
    prefix = "OK" if commit else "DRY RUN"
    print(
        f"{prefix}: entries={result.entry_count} "
        f"created={result.created_count} updated={result.updated_count} "
        f"unchanged={result.unchanged_count} suspended={result.suspended_count}",
    )
    if not commit:
        print("Re-run with --commit to write.")


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Seed DAX BigQuery dataset resources from JSON.",
    )
    parser.add_argument("inventory", type=Path)
    parser.add_argument(
        "--lineage-catalog",
        type=Path,
        help=(
            "Assemble referenced non-dbt source datasets into the inventory "
            "before projection. Missing owners default to the lineage-root "
            "owner policy; existing owners are preserved."
        ),
    )
    parser.add_argument("--integration-slug")
    parser.add_argument(
        "--source-version",
        default="manual-dax-seed-" + datetime.now(UTC).strftime("%Y%m%d"),
    )
    parser.add_argument(
        "--allow-unresolved",
        action="store_true",
        help=(
            "Allow unresolved owner_email values. Such rows are not requestable "
            "by resource-owner authority until repaired."
        ),
    )
    parser.add_argument(
        "--unresolved-owner-fallback-email",
        help=(
            "Staging-only fallback for inventory owner emails absent from a "
            "sanitized identity cohort. Production should omit this option."
        ),
    )
    parser.add_argument(
        "--commit",
        action="store_true",
        help="Write changes. Without this flag the script is a dry run.",
    )
    args = parser.parse_args()
    asyncio.run(
        _run(
            inventory_path=args.inventory,
            lineage_catalog_path=args.lineage_catalog,
            integration_slug=args.integration_slug,
            source_version=args.source_version,
            allow_unresolved=args.allow_unresolved,
            unresolved_owner_fallback_email=(args.unresolved_owner_fallback_email),
            commit=args.commit,
        ),
    )


if __name__ == "__main__":
    main()