"""Vault MD import → pointer atom。"""

from __future__ import annotations

from typing import Any, Callable

from niannian.vault.frontmatter import parse_md
from niannian.vault.overview import touch_overview
from niannian.vault.store import VaultStore
from niannian.vault.taxonomy import TaxonomyService

_DEDUP_WINDOW_HOURS = 24


class DuplicateImportError(Exception):
    """同 source_url 或 vault_id 在 dedup 窗口内已导入。"""

    def __init__(self, *, vault_id: str, source_url: str | None = None) -> None:
        self.vault_id = vault_id
        self.source_url = source_url
        super().__init__("duplicate_import")


def dedup_key(*, source_url: str | None, vault_id: str | None) -> str | None:
    url = str(source_url or "").strip()
    if url:
        return f"url:{url}"
    vid = str(vault_id or "").strip()
    if vid:
        return f"vault:{vid}"
    return None


_TRUST_MAP = {
    "user": "user_paste",
    "web": "web",
    "curated": "curated",
    "user_paste": "user_paste",
    "guest_reference": "guest_reference",
}


class VaultImportService:
    def __init__(
        self,
        store: VaultStore,
        *,
        ingest_fn: Callable[..., Any] | None = None,
        taxonomy: TaxonomyService | None = None,
        memory_store=None,
    ) -> None:
        self._store = store
        self._ingest = ingest_fn
        self._taxonomy = taxonomy
        self._memory_store = memory_store

    def import_md(
        self,
        md_body: str,
        *,
        namespace: str = "user",
        hints: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        fm, body = parse_md(md_body)
        hints = hints or {}
        title = str(fm.get("title") or hints.get("title") or "未命名资料")
        ns = str(fm.get("namespace") or namespace or "user")
        if ns not in ("user", "niannian"):
            ns = "user"
        merged = {**fm}
        for k, v in hints.items():
            if v is not None and k not in merged:
                merged[k] = v
        merged.setdefault("source_type", "md_import")
        merged.setdefault("extracted_by", "codex")
        merged.setdefault("trust_tier", "user")
        merged.setdefault("taxonomy_status", merged.get("taxonomy_status") or "confirmed")
        if self._taxonomy is not None and ns == "user":
            routed = self._taxonomy.match_routing(merged)
            if routed:
                merged["category"] = routed["category"]
                merged["subproject"] = routed["subproject"]
                merged["taxonomy_status"] = "confirmed"
        self._check_dedup(merged, namespace=ns)
        doc = self._store.write_document(
            namespace=ns,
            title=title,
            body=body or merged.get("body") or "",
            frontmatter=merged,
        )
        try:
            atom_id = None
            if self._ingest is not None:
                atom_id = self._compile_pointer_atom(doc, merged)
                if atom_id:
                    self._store.update_entry_atom(doc["vault_id"], atom_id)
            touch_overview(
                self._store,
                namespace=ns,
                category=str(doc["frontmatter"].get("category") or ""),
                subproject=str(doc["frontmatter"].get("subproject") or ""),
                entry={"vault_id": doc["vault_id"], "title": title},
            )
            self._supersede_prior_pointers(merged, doc, atom_id)
            return {
                "vault_id": doc["vault_id"],
                "rel_path": doc["rel_path"],
                "atom_id": atom_id,
                "title": title,
            }
        except Exception:
            self._store.rollback_import(doc["vault_id"], doc["rel_path"])
            raise

    def _check_dedup(self, merged: dict[str, Any], *, namespace: str) -> None:
        version = int(merged.get("vault_version") or 1)
        if version > 1:
            return
        source_url = str(merged.get("source_url") or "").strip() or None
        vault_id = str(merged.get("id") or "").strip() or None
        if source_url:
            existing = self._store.find_recent_import_by_source_url(
                source_url,
                namespace=namespace,
                window_hours=_DEDUP_WINDOW_HOURS,
            )
            if existing and existing.get("vault_id"):
                raise DuplicateImportError(
                    vault_id=str(existing["vault_id"]),
                    source_url=source_url,
                )
        if vault_id:
            existing = self._store.find_recent_import_by_vault_id(
                vault_id,
                window_hours=_DEDUP_WINDOW_HOURS,
            )
            if existing:
                raise DuplicateImportError(
                    vault_id=vault_id,
                    source_url=str(existing.get("source_url") or "") or None,
                )

    def _compile_pointer_atom(self, doc: dict[str, Any], fm: dict[str, Any]) -> str | None:
        title = str(fm.get("title") or doc.get("title") or "资料")
        body = doc.get("body") or ""
        summary = ""
        for line in body.splitlines():
            if line.strip().startswith("## 摘要"):
                continue
            if line.strip().startswith("##"):
                break
            if line.strip():
                summary = line.strip()
                break
        preview = (summary or body[:120] or title)[:120]
        trust = _TRUST_MAP.get(str(fm.get("trust_tier") or "user"), "user_paste")
        rel = doc["rel_path"]
        tags = ["vault", str(fm.get("category") or "")]
        extra_tags = fm.get("tags") or []
        if isinstance(extra_tags, list):
            tags.extend(str(t) for t in extra_tags if t)
        meta = {
            "memory_scope": "subject",
            "subject_eligible": True,
            "vault": {
                "namespace": fm.get("namespace") or "user",
                "path": rel,
                "vault_id": doc["vault_id"],
            },
            "provenance": {
                "type": "external_web" if fm.get("source_url") else "vault_import",
                "url": fm.get("source_url"),
                "trust_tier": trust,
            },
        }
        atom = self._ingest(
            f"【资料】{title}：{preview}",
            tags=sorted(set(t for t in tags if t)),
            source="vault:import",
            meta=meta,
        )
        return getattr(atom, "id", None)

    def _supersede_prior_pointers(
        self,
        merged: dict[str, Any],
        doc: dict[str, Any],
        new_atom_id: str | None,
    ) -> None:
        if self._memory_store is None:
            return
        source_url = str(merged.get("source_url") or "").strip()
        version = int(merged.get("vault_version") or 1)
        if not source_url or version <= 1:
            return
        current_vid = str(doc.get("vault_id") or "")
        for entry in self._store.load_index().get("entries") or []:
            if not isinstance(entry, dict):
                continue
            old_vid = str(entry.get("vault_id") or "")
            if not old_vid or old_vid == current_vid:
                continue
            rel = str(entry.get("rel_path") or "")
            try:
                old_doc = self._store.read_document(rel)
            except (FileNotFoundError, OSError):
                continue
            old_url = str((old_doc.get("frontmatter") or {}).get("source_url") or "")
            if old_url != source_url:
                continue
            old_atom = entry.get("atom_id")
            if old_atom and old_atom != new_atom_id:
                self._memory_store.patch_atom_meta(
                    str(old_atom),
                    {"belief_status": "superseded"},
                )

    def import_batch(
        self,
        items: list[dict[str, Any]],
        *,
        default_namespace: str = "user",
    ) -> dict[str, Any]:
        results: list[dict[str, Any]] = []
        errors: list[dict[str, str]] = []
        for i, item in enumerate(items):
            md_body = str(item.get("md_body") or "").strip()
            if not md_body:
                errors.append({"index": str(i), "error": "md_body required"})
                continue
            ns = str(item.get("namespace") or default_namespace)
            hints = item.get("hints") if isinstance(item.get("hints"), dict) else None
            try:
                row = self.import_md(md_body, namespace=ns, hints=hints)
                results.append(row)
            except Exception as exc:
                errors.append({"index": str(i), "error": str(exc)})
        return {"count": len(results), "items": results, "errors": errors}
