"""Central model-registry CONSUMER (docs/provenance/00 §3, provenance item 6).

Read-only Python mirror of the ArcGIS Pro ``SharedModelRegistryService`` /
``ModelRegistrySettingsService`` so both products read the IDENTICAL registry
SQLite databases (byte-identical schema, SQL, ordering, path scheme and import
behavior).

The registry DBs are *owned and created by the editors*. This module is strictly
READ-ONLY: it never creates, migrates or writes the ``models`` / ``model_versions``
tables. It only:

  * lists published versions for a model kind (same JOIN/SQL/order as Pro),
  * opens the DB defensively for OneDrive/SQLite safety (``journal_mode=DELETE``
    + ``quick_check`` + conflicted-copy rejection — see §6),
  * imports a chosen version into the project's ``model_snapshots`` table as an
    immutable embed-+-reference snapshot, and
  * materializes the version bytes to a temp file so a task can use it as a
    model path.

Pure stdlib (``sqlite3`` only) — no qgis / PyQt imports — so it unit-tests
cleanly and so the picker dialog's Qt surface stays separate.
"""

from __future__ import annotations

import os
import sqlite3
import tempfile
import uuid
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional

# Settings keys — identical to Pro's ModelRegistrySettingsService so a value set
# by one product is honored by the other.
CDP_REGISTRY_PATH_KEY = "CdpRegistryPath"
NETWEAVER_REGISTRY_PATH_KEY = "NetWeaverRegistryPath"

# Per-engine default registry locations (Pro: <UserProfile>\Documents\<folder>\model_registry.db).
# NOTE: per-kind paths, NOT a single consolidated path — this is what the
# editors and ArcGIS Pro use, so the DBs stay byte-identical across products.
_DEFAULT_PRODUCT_FOLDER = {"cdp": "CDP", "netweaver": "NW3"}
_REGISTRY_FILE_NAME = "model_registry.db"

_SETTINGS_KEY = {"cdp": CDP_REGISTRY_PATH_KEY, "netweaver": NETWEAVER_REGISTRY_PATH_KEY}

_DEFAULT_SOURCE_FORMAT = {"cdp": ".cdpx", "netweaver": ".nw2"}


def normalize_model_kind(value: Any) -> str:
    """Map common variants of a model-kind string to 'cdp' or 'netweaver'.

    Mirrors Pro's ``ModelSnapshotService.NormalizeModelKind`` for the two kinds
    the registry distinguishes. Unknown values are returned lowercased/trimmed.
    """
    text = str(value or "").strip().lower()
    if text in ("cdp", "cdpx", "cdp6", "criteria", ".cdpx", ".cdp"):
        return "cdp"
    if text in ("netweaver", "nw", "nw2", "nw3", "net weaver", ".nw2", ".nw"):
        return "netweaver"
    return text


def _normalize_source_format(value: Any) -> str:
    text = str(value or "").strip().lower()
    if not text:
        return ""
    return text if text.startswith(".") else "." + text


@dataclass
class RegistryVersion:
    """One published model version, mirroring Pro's ``ModelRegistryVersion``."""

    version_id: str
    model_id: str
    model_kind: str
    model_name: str
    version: str
    content_blob: bytes
    content_sha256: str
    source_format: Optional[str]
    schema_version: Optional[str]
    authoring_tool: Optional[str]
    authoring_tool_version: Optional[str]
    model_summary_json: Optional[str]
    is_current: bool


# Same column list / JOIN / WHERE / ORDER BY as Pro's SharedModelRegistryService.
_LIST_VERSIONS_SQL = """
    SELECT
        v.id AS version_id,
        m.id AS model_id,
        m.model_kind,
        m.name AS model_name,
        m.current_version_id,
        v.version,
        v.content_blob,
        v.content_sha256,
        v.source_format,
        v.schema_version,
        v.authoring_tool,
        v.authoring_tool_version,
        v.model_summary_json,
        v.created_at
    FROM model_versions v
    JOIN models m ON m.id = v.model_id
    WHERE lower(m.model_kind) = lower(?)
    ORDER BY v.created_at DESC, v.version DESC
"""


class ModelRegistryError(RuntimeError):
    """Raised when a registry DB is missing, corrupt, or has conflicted copies."""


class ModelRegistryReader:
    """Read-only consumer of a central model-registry SQLite database."""

    def __init__(self, registry_path: str):
        if not str(registry_path or "").strip():
            raise ValueError("Registry path is required.")
        self.registry_path = str(registry_path)

    # ------------------------------------------------------------------ open
    def _open(self) -> sqlite3.Connection:
        """Open the registry defensively (mirrors Pro's OpenRegistryAsync)."""
        if not os.path.isfile(self.registry_path):
            raise ModelRegistryError(
                f"Model registry database was not found: {self.registry_path}"
            )
        self._raise_if_conflicted_copy_exists()

        # journal_mode=DELETE avoids WAL sidecars that sync inconsistently on
        # OneDrive (§6). Open the DB, set the journal mode, then quick_check.
        conn = sqlite3.connect(self.registry_path)
        conn.row_factory = sqlite3.Row
        try:
            conn.execute("PRAGMA journal_mode=DELETE")
            row = conn.execute("PRAGMA quick_check").fetchone()
            result = (row[0] if row is not None else "").strip()
            if result.lower() != "ok":
                raise ModelRegistryError(
                    f"Model registry quick_check failed: {result or '(no result)'}"
                )
        except ModelRegistryError:
            conn.close()
            raise
        except sqlite3.DatabaseError as exc:
            conn.close()
            raise ModelRegistryError(f"Model registry could not be opened: {exc}") from exc
        return conn

    def _raise_if_conflicted_copy_exists(self) -> None:
        """Reject OneDrive '... (conflicted copy) ...' sidecars (§6)."""
        directory = os.path.dirname(self.registry_path)
        if not directory or not os.path.isdir(directory):
            return
        base_no_ext = os.path.splitext(os.path.basename(self.registry_path))[0]
        target_real = os.path.realpath(self.registry_path)
        conflicts: List[str] = []
        try:
            entries = os.listdir(directory)
        except OSError:
            return
        for entry in entries:
            if not entry.startswith(base_no_ext):
                continue
            if "conflicted copy" not in entry.lower():
                continue
            full = os.path.join(directory, entry)
            if os.path.realpath(full) == target_real:
                continue
            conflicts.append(full)
        if conflicts:
            raise ModelRegistryError(
                "Conflicted model registry copies were found (OneDrive sync conflict): "
                + ", ".join(conflicts)
            )

    # --------------------------------------------------------------- queries
    def list_versions(self, model_kind: str) -> List[RegistryVersion]:
        """All published versions for ``model_kind``, newest-first (Pro order)."""
        normalized = normalize_model_kind(model_kind)
        conn = self._open()
        try:
            rows = conn.execute(_LIST_VERSIONS_SQL, (normalized,)).fetchall()
        finally:
            conn.close()

        versions: List[RegistryVersion] = []
        for row in rows:
            version_id = row["version_id"]
            current_version_id = row["current_version_id"]
            blob = row["content_blob"]
            versions.append(
                RegistryVersion(
                    version_id=version_id,
                    model_id=row["model_id"],
                    model_kind=normalize_model_kind(row["model_kind"]),
                    model_name=row["model_name"],
                    version=row["version"],
                    content_blob=bytes(blob) if blob is not None else b"",
                    content_sha256=row["content_sha256"],
                    source_format=_normalize_source_format(row["source_format"]) or None,
                    schema_version=row["schema_version"],
                    authoring_tool=row["authoring_tool"],
                    authoring_tool_version=row["authoring_tool_version"],
                    model_summary_json=row["model_summary_json"],
                    is_current=(
                        version_id is not None
                        and current_version_id is not None
                        and str(version_id).lower() == str(current_version_id).lower()
                    ),
                )
            )
        return versions

    def find_version(self, model_kind: str, version_id: str) -> Optional[RegistryVersion]:
        """The version with ``version_id`` for ``model_kind`` (or None)."""
        if not str(version_id or "").strip():
            return None
        for version in self.list_versions(model_kind):
            if str(version.version_id or "").lower() == str(version_id).lower():
                return version
        return None


def resolve_registry_path(
    model_kind: str,
    settings_get: Optional[Callable[[str], Optional[str]]] = None,
) -> str:
    """Resolve the per-kind registry path (Pro's ModelRegistrySettingsService).

    Returns a configured override (via ``settings_get(key)`` using the
    ``CdpRegistryPath`` / ``NetWeaverRegistryPath`` keys) when present, otherwise
    the per-engine default ``<UserProfile>/Documents/<CDP|NW3>/model_registry.db``.

    ``settings_get`` is injected so callers can supply a QgsSettings-backed getter
    WITHOUT this module importing qgis.
    """
    normalized = normalize_model_kind(model_kind)
    if settings_get is not None:
        key = _SETTINGS_KEY.get(normalized)
        if key:
            try:
                configured = settings_get(key)
            except Exception:
                configured = None
            if configured and str(configured).strip():
                return str(configured).strip()

    folder = _DEFAULT_PRODUCT_FOLDER.get(normalized)
    if folder is None:
        raise ValueError(f"Unknown model kind for registry path: {model_kind!r}")
    user_profile = (
        os.environ.get("USERPROFILE")
        or os.path.expanduser("~")
    )
    return os.path.join(user_profile, "Documents", folder, _REGISTRY_FILE_NAME)


def _parse_summary_json(model_summary_json: Optional[str]) -> Dict[str, Any]:
    if not model_summary_json:
        return {}
    try:
        import json

        parsed = json.loads(model_summary_json)
        return parsed if isinstance(parsed, dict) else {}
    except Exception:
        return {}


def _default_source_format(model_kind: str) -> str:
    return _DEFAULT_SOURCE_FORMAT.get(normalize_model_kind(model_kind), ".model")


def import_version_as_snapshot(
    manager: Any,
    version: RegistryVersion,
    *,
    registry_path: str = "",
    created_by: str = "",
) -> Optional[str]:
    """Import a registry version into the project's ``model_snapshots`` table.

    Embed-+-reference (§3): copies the version bytes into a snapshot row, with
    ``registry_id`` = model_id and ``registry_version`` = version pointing back at
    the registry. Reuses an existing snapshot by content sha (``reuse_by_sha``).

    ``manager`` is a WorkbenchMetadataManager (or compatible) exposing
    ``save_model_snapshot(...)``. Returns the snapshot id (or None on error).
    """
    source_format = _normalize_source_format(version.source_format) or _default_source_format(version.model_kind)
    return manager.save_model_snapshot(
        model_kind=normalize_model_kind(version.model_kind),
        content_blob=version.content_blob or b"",
        display_name=version.model_name or "",
        source_path=str(registry_path or ""),
        source_format=source_format,
        schema_version=version.schema_version or "",
        authoring_tool=version.authoring_tool or "",
        authoring_tool_version=version.authoring_tool_version or "",
        model_summary=_parse_summary_json(version.model_summary_json),
        registry_id=version.model_id or "",
        registry_version=version.version or "",
        created_by=created_by or "",
        reuse_by_sha=True,
    )


def _sanitize_file_name(value: str) -> str:
    cleaned = "".join(c if c not in '\\/:*?"<>|' else "-" for c in str(value or "")).strip()
    return cleaned or "model"


def materialize_version(version: RegistryVersion, temp_root: Optional[str] = None) -> str:
    """Write the version bytes to a temp file and return its path.

    The extension matches the version's source_format (Pro's MaterializeVersion),
    so a task can use the result directly as a model file path.
    """
    root = temp_root or os.path.join(tempfile.gettempdir(), "emds-registry-models")
    os.makedirs(root, exist_ok=True)
    ext = _normalize_source_format(version.source_format) or _default_source_format(version.model_kind)
    file_name = (
        f"emds-registry-{_sanitize_file_name(version.model_name)}-"
        f"{version.version_id}-{uuid.uuid4().hex}{ext}"
    )
    path = os.path.join(root, file_name)
    with open(path, "wb") as handle:
        handle.write(version.content_blob or b"")
    return path
