"""ExtractSpec v1 — 文本站抓取 · text · md_import → vault MD。"""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any, Callable

from niannian.llm.types import LLMUnavailable, Message
from niannian.vault.frontmatter import serialize_md
from niannian.vault.image_fetch import load_image_b64, max_image_bytes
from niannian.vault.import_service import VaultImportService
from niannian.vault.text_fetch import (
    extract_text_from_html,
    fetch_url_html,
    is_text_url_allowed,
    load_text_hosts_config,
)
from niannian.vault.taxonomy import TaxonomyService


class ExtractService:
    def __init__(
        self,
        importer: VaultImportService,
        taxonomy: TaxonomyService,
        *,
        llm=None,
        config: dict[str, Any] | None = None,
        fetch_html: Callable[..., str] | None = None,
        fetch_image_b64_fn=None,
    ) -> None:
        self._importer = importer
        self._taxonomy = taxonomy
        self._llm = llm
        self._config = config or {}
        self._fetch_html = fetch_html or fetch_url_html
        self._fetch_image_b64 = fetch_image_b64_fn
        vault_cfg = self._config.get("vault") or {}
        extract_cfg = vault_cfg.get("extract") if isinstance(vault_cfg, dict) else {}
        self._llm_module = str((extract_cfg or {}).get("llm_module") or "left")
        self._default_extract_tier = str((extract_cfg or {}).get("default_tier") or "light")
        self._max_image_bytes = max_image_bytes(self._config)

    def extract(
        self,
        *,
        namespace: str = "user",
        input_type: str = "url",
        payload: dict[str, Any] | None = None,
        hints: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        payload = payload or {}
        hints = hints or {}
        itype = (input_type or "url").lower()

        if itype == "md_import":
            md_body = str(payload.get("md_body") or "")
            if not md_body.strip():
                raise ValueError("md_body required")
            return self._importer.import_md(md_body, namespace=namespace, hints=hints)

        if itype == "text":
            text = str(payload.get("text") or "").strip()
            if not text:
                raise ValueError("text required")
            title = str(hints.get("title") or payload.get("title") or "粘贴文本")
            md = self._build_md(
                title=title,
                body_text=text,
                source_type="text",
                source_url=payload.get("url"),
                namespace=namespace,
                hints=hints,
            )
            return self._importer.import_md(md, namespace=namespace, hints=hints)

        if itype == "url":
            url = str(payload.get("url") or "").strip()
            if not url:
                raise ValueError("url required")
            if not is_text_url_allowed(url):
                raise ValueError("host_not_allowed")
            html = self._fetch_html(url)
            title, body_text = extract_text_from_html(html)
            if hints.get("title"):
                title = str(hints["title"])
            md = self._build_md(
                title=title,
                body_text=body_text,
                source_type="url",
                source_url=url,
                namespace=namespace,
                hints=hints,
            )
            result = self._importer.import_md(md, namespace=namespace, hints=hints)
            result["source_url"] = url
            result["extracted_by"] = "niannian"
            return result

        if itype == "image":
            image_b64, mime = load_image_b64(
                payload,
                max_bytes=self._max_image_bytes,
                fetch_fn=self._fetch_image_b64,
            )
            ocr_text = self._run_ocr(image_b64, mime=mime)
            title = str(hints.get("title") or payload.get("title") or "图片资料")
            source_url = payload.get("url")
            md = self._build_md(
                title=title,
                body_text=ocr_text,
                source_type="image",
                source_url=str(source_url) if source_url else None,
                namespace=namespace,
                hints={**hints, "trust_tier": hints.get("trust_tier") or "web"},
            )
            result = self._importer.import_md(md, namespace=namespace, hints=hints)
            result["ocr_text_len"] = len(ocr_text)
            return result

        raise ValueError(f"unsupported input_type: {input_type}")

    def _run_ocr(self, image_b64: str, *, mime: str) -> str:
        if self._llm is None or not self._llm.is_module_configured("ocr"):
            raise ValueError("ocr_unavailable")
        try:
            result = self._llm.ocr("ocr", image_b64, mime=mime)
        except LLMUnavailable as exc:
            raise ValueError("ocr_unavailable") from exc
        text = (result.text or "").strip()
        if not text:
            raise ValueError("ocr_unavailable")
        return text

    def _build_md(
        self,
        *,
        title: str,
        body_text: str,
        source_type: str,
        source_url: str | None,
        namespace: str,
        hints: dict[str, Any],
    ) -> str:
        tier = str(hints.get("extract_tier") or self._default_extract_tier).lower()
        if tier == "full":
            llm_md = self._try_llm_md(title=title, body_text=body_text, source_url=source_url)
            if llm_md:
                return llm_md
        summary = body_text[:200].replace("\n", " ").strip()
        points = [ln.strip() for ln in body_text.splitlines() if ln.strip()][:5]
        points_md = "\n".join(f"- {p[:200]}" for p in points) or "- （无）"
        fm: dict[str, Any] = {
            "vault_version": 1,
            "namespace": namespace,
            "title": title,
            "category": hints.get("category"),
            "subproject": hints.get("subproject"),
            "source_type": source_type,
            "source_url": source_url,
            "extracted_by": "niannian",
            "extracted_at": datetime.now(timezone.utc).isoformat(),
            "trust_tier": "web",
            "tags": hints.get("tags") or [],
        }
        if hints.get("taxonomy_status"):
            fm["taxonomy_status"] = hints["taxonomy_status"]
        body = (
            f"## 摘要\n\n{summary}\n\n"
            f"## 要点\n\n{points_md}\n\n"
            f"## 来源说明\n\n{source_url or '粘贴/本地'}\n\n"
            f"## 正文\n\n{body_text[:20_000]}"
        )
        return serialize_md({k: v for k, v in fm.items() if v is not None}, body)

    def _try_llm_md(self, *, title: str, body_text: str, source_url: str | None) -> str | None:
        if self._llm is None:
            return None
        module = self._llm_module
        if module not in ("left", "right") or not self._llm.is_module_configured(module):
            alt = "right" if module == "left" else "left"
            if self._llm.is_module_configured(alt):
                module = alt
            else:
                return None
        prompt = (
            "将以下网页正文整理为 vault MD（含 YAML frontmatter + ## 摘要/要点/来源说明/正文）。"
            f"title: {title}\nsource_url: {source_url or ''}\n"
            f"trust_tier: web\nextracted_by: niannian\n\n正文：\n{body_text[:8000]}"
        )
        try:
            result = self._llm.complete(module, [Message(role="user", content=prompt)], max_tokens=4096)
        except Exception:
            return None
        text = (result.text or "").strip()
        if text.startswith("---"):
            return text
        return None
