from __future__ import annotations

import json
import logging
import os
from pathlib import Path
from typing import Any

from niannian.external_data.capability import CapabilityMeta, assert_capability_allowed
from niannian.external_data.learn import distill_function, learned_path, maybe_learn_after_write
from niannian.external_data.schema import ExternalDataRequest, ExternalDataResult
from niannian.external_data.store import ExternalDataStore, rel_path_for
from niannian.external_data.tool_audit import append_tool_audit, audit_path as tool_audit_path, build_audit_row
from niannian.memory.space import ensure_memory_space

logger = logging.getLogger(__name__)


def _resolve_learn_mode(data: dict[str, Any], meta: CapabilityMeta, req: ExternalDataRequest) -> str:
    """Explicit learn.mode wins; otherwise apply capability auto_learn_default."""
    raw_learn = data.get("learn")
    if raw_learn is None or (isinstance(raw_learn, dict) and "mode" not in raw_learn):
        return str(meta.auto_learn_default or "off").strip().lower()
    return str((req.learn or {}).get("mode") or "off").strip().lower()


class ExternalDataService:
    """外部数据源 — add / delete / overwrite；默认 L1 三云，不归 L2。"""

    def __init__(self, home: Path, *, config: dict[str, Any] | None = None, memory_store=None) -> None:
        self._home = home
        self._config = config or {}
        self._store = ExternalDataStore(home)
        self._memory_store = memory_store

    def apply(self, data: dict[str, Any]) -> ExternalDataResult:
        capability_id = str(data.get("capability_id") or "").strip()
        caller = data.get("caller") or {}
        client = str(caller.get("client") or "").strip()
        operation = str(data.get("operation") or "add").strip().lower()
        function = str(data.get("function") or "").strip()
        meta = assert_capability_allowed(
            capability_id,
            client,
            operation,
            function=function,
            config=self._config,
        )

        ext_cfg = self._config.get("external_tools") or {}
        strict_v2 = ext_cfg.get("strict_v2_provenance", True)
        req = ExternalDataRequest.from_dict(data, strict_v2=bool(strict_v2))
        ensure_memory_space(self._home, req.project)

        record_id = req.record_id

        if req.operation == "add" and req.dedup_key:
            existing = self._store.find_by_dedup_key(
                req.project,
                req.function,
                req.dedup_key,
                storage_target=req.storage_target,
                local_dir=req.local_dir,
            )
            if existing:
                raise ValueError(f"dedup_key already exists: {req.dedup_key}")

        if req.operation == "delete":
            if not record_id and req.dedup_key:
                found = self._store.find_by_dedup_key(
                    req.project,
                    req.function,
                    req.dedup_key,
                    storage_target=req.storage_target,
                    local_dir=req.local_dir,
                )
                if found:
                    record_id = found
            if not record_id:
                raise ValueError("delete requires record_id or dedup_key")
            existed = self._store.delete_record(
                req.project,
                req.function,
                record_id,
                storage_target=req.storage_target,
                local_dir=req.local_dir,
            )
            rel = rel_path_for(req.project, req.function, record_id)
            if req.storage_target == "cloud":
                self._sync_manifest(req.project, req.function, record_id, deleted=True)
            result = ExternalDataResult(
                ok=existed,
                operation="delete",
                record_id=record_id,
                rel_path=rel,
                storage_target=req.storage_target,
                project=req.project,
                function=req.function,
                message="deleted" if existed else "not_found",
            )
            self._append_audit(req, result, learn_applied=False)
            return result

        if not record_id:
            record_id = self._store.resolve_record_id(req.record_id, req.dedup_key, req.project, req.function)

        if req.operation == "overwrite" and req.dedup_key and not req.record_id:
            found = self._store.find_by_dedup_key(
                req.project,
                req.function,
                req.dedup_key,
                storage_target=req.storage_target,
                local_dir=req.local_dir,
            )
            if found:
                record_id = found

        envelope = req.to_envelope(record_id=record_id)
        path = self._store.write_record(envelope, storage_target=req.storage_target, local_dir=req.local_dir)
        rel = rel_path_for(req.project, req.function, record_id)

        synced = False
        if req.storage_target == "cloud":
            synced = self._sync_manifest(req.project, req.function, record_id, deleted=False)

        result = ExternalDataResult(
            ok=True,
            operation=req.operation,
            record_id=record_id,
            rel_path=rel,
            storage_target=req.storage_target,
            project=req.project,
            function=req.function,
            synced_to_cloud=synced,
            message=str(path),
        )
        learn_mode = _resolve_learn_mode(data, meta, req)
        learn_applied = maybe_learn_after_write(
            self._home,
            req.project,
            req.function,
            envelope,
            learn_mode,
        )
        self._append_audit(req, result, learn_applied=learn_applied)
        return result

    def _append_audit(
        self,
        req: ExternalDataRequest,
        result: ExternalDataResult,
        *,
        learn_applied: bool = False,
    ) -> None:
        try:
            append_tool_audit(
                self._home,
                req.project,
                build_audit_row(req, result, learn_applied=learn_applied),
            )
        except Exception:
            logger.exception("external_data tool audit append failed")

    def _sync_manifest(self, project: str, function: str, record_id: str, *, deleted: bool) -> bool:
        if self._memory_store is None:
            return False
        try:
            rel = rel_path_for(project, function, record_id)
            path = self._home / rel.replace("/", os.sep)
            manifest = getattr(self._memory_store, "manifest", None)
            if manifest is None:
                manifest = getattr(self._memory_store, "_manifest", None)
            if manifest is None:
                return False
            if deleted and not path.is_file():
                manifest.update_file_entry(rel, path)
            elif path.is_file():
                manifest.update_file_entry(rel, path)
            idx_rel = f"external/{project}/{function}/_index.jsonl"
            idx_path = self._home / "external" / project / function / "_index.jsonl"
            if idx_path.is_file():
                manifest.update_file_entry(idx_rel, idx_path)
            self._schedule_cloud_push()
            return True
        except Exception:
            logger.exception("external_data manifest sync failed")
            return False

    def _schedule_cloud_push(self) -> None:
        if self._memory_store is None:
            return
        from niannian.storage.sync.push_after_ingest import schedule_push_after_ingest

        schedule_push_after_ingest(self._memory_store, self._config)

    def archive_to_l2(
        self,
        project: str,
        function: str,
        *,
        storage_target: str = "cloud",
        local_dir: str = "",
    ) -> dict[str, Any]:
        """仅在 Chat 明确要求时调用 — 路径 {archive_root}/{project}/{function}/"""
        archive_cfg = (self._config.get("storage") or {}).get("archive") or {}
        archive_root = Path(os.environ.get("NIANNIAN_ARCHIVE_DIR", archive_cfg.get("local_dir", "/data/archive")))
        result = self._store.archive_to_l2(
            project,
            function,
            archive_root,
            storage_target=storage_target,
            local_dir=local_dir,
        )
        webdav_cfg = archive_cfg.get("webdav") or {}
        if webdav_cfg.get("enabled") and result.get("ok"):
            try:
                from niannian.storage.webdav import webdav_put

                dest = Path(result["dest"])
                base = f"/niannian/archive/{project}/{function}"
                for name in result.get("files") or []:
                    fp = dest / name
                    if fp.is_file():
                        webdav_put(f"{base}/{name}", fp.read_bytes(), config=self._config)
            except Exception as exc:
                result["webdav_error"] = str(exc)
        return result

    def list_records(self, project: str, function: str) -> list[dict[str, Any]]:
        base = self._home / "external" / project / function
        if not base.is_dir():
            return []
        out: list[dict[str, Any]] = []
        for p in sorted(base.glob("*.json")):
            if p.name.startswith("_"):
                continue
            try:
                out.append(json.loads(p.read_text(encoding="utf-8")))
            except Exception:
                continue
        return out

    def list_external(self, project: str, function: str = "") -> dict[str, Any]:
        base = self._home / "external" / project
        fn = str(function or "").strip()
        if not fn:
            functions: list[str] = []
            if base.is_dir():
                for path in sorted(base.iterdir()):
                    if path.is_dir() and not path.name.startswith("_"):
                        functions.append(path.name)
            return {"project": project, "functions": functions}
        records = self.list_records(project, fn)
        return {"project": project, "function": fn, "count": len(records), "records": records}

    def recall_external(
        self,
        project: str,
        function: str,
        query: str,
        *,
        limit: int = 10,
    ) -> dict[str, Any]:
        q = str(query or "").strip().lower()
        cap = max(1, int(limit))
        hits: list[dict[str, Any]] = []
        learned_snippet: str | None = None

        learned = learned_path(self._home, project, function)
        if learned.is_file():
            text = learned.read_text(encoding="utf-8")
            if not q or q in text.lower():
                learned_snippet = text[:2000]

        for record in self.list_records(project, function):
            blob = json.dumps(record, ensure_ascii=False).lower()
            if not q or q in blob:
                hits.append(record)
            if len(hits) >= cap:
                break

        return {
            "project": project,
            "function": function,
            "query": query,
            "count": len(hits),
            "records": hits[:cap],
            "learned": learned_snippet,
        }

    def learn_external(self, project: str, function: str) -> dict[str, Any]:
        path = distill_function(self._home, project, function)
        return {"ok": True, "project": project, "function": function, "path": path}

    def query_tool_audit(
        self,
        project: str,
        *,
        since: str = "",
        actor: str = "",
    ) -> dict[str, Any]:
        path = tool_audit_path(self._home, project)
        since_val = str(since or "").strip()
        actor_val = str(actor or "").strip()
        entries: list[dict[str, Any]] = []
        if path.is_file():
            for line in path.read_text(encoding="utf-8").strip().split("\n"):
                if not line:
                    continue
                try:
                    row = json.loads(line)
                except json.JSONDecodeError:
                    continue
                if since_val:
                    ts = str(row.get("received_at") or row.get("occurred_at") or "")
                    if ts and ts < since_val:
                        continue
                if actor_val:
                    user_id = str((row.get("actor") or {}).get("user_id") or "")
                    if user_id != actor_val:
                        continue
                entries.append(row)
        return {"project": project, "count": len(entries), "entries": entries}

    def _find_record(self, project: str, record_id: str) -> dict[str, Any] | None:
        base = self._home / "external" / project
        if not base.is_dir():
            return None
        for fn_dir in sorted(base.iterdir()):
            if not fn_dir.is_dir() or fn_dir.name.startswith("_"):
                continue
            path = fn_dir / f"{record_id}.json"
            if path.is_file():
                try:
                    return json.loads(path.read_text(encoding="utf-8"))
                except (json.JSONDecodeError, OSError):
                    return None
        return None

    def trace_external(self, project: str, record_id: str) -> dict[str, Any]:
        chain: list[dict[str, Any]] = []
        visited: set[str] = set()
        current = str(record_id or "").strip()
        while current and current not in visited:
            visited.add(current)
            record = self._find_record(project, current)
            if record is None:
                break
            chain.append(record)
            prov = record.get("provenance") or {}
            current = str(prov.get("parent_record_id") or "").strip()
        return {"project": project, "record_id": record_id, "chain": chain}

