"""VP import-source — fetch web/image URL · summary to sources / deep_learn runs."""

from __future__ import annotations

import json
import re
import uuid
from datetime import datetime, timezone
from html import unescape
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable
from urllib.parse import urlparse

from niannian.soul.search_providers import url_allowed
from niannian.virtual_persona.assembly_master import append_section_body
from niannian.virtual_persona.deep_learn import DEEP_LEARN_DIR, RUNS_DIR
from niannian.virtual_persona.enrich import _allowed_image_url, _fetch_url_bytes
from niannian.virtual_persona.pack import read_pack
from niannian.virtual_persona.profile_md import merge_profile_sections

if TYPE_CHECKING:
    from niannian.virtual_persona.store import VirtualPersonaStore

IMPORT_SOURCE_TOPIC = "链接导入"
DEFAULT_MERGE_INTO = "sources"
VALID_KINDS = frozenset({"web", "image", "video"})
VALID_MERGE_INTO = frozenset({"profile", "sources", "assembly_master"})
_PAGE_MAX_BYTES = 256_000
_IMAGE_MAX_BYTES = 512 * 1024
_SUMMARY_MAX_CHARS = 2000
_TAG_RE = re.compile(r"<(script|style)[^>]*>.*?</\1>", re.IGNORECASE | re.DOTALL)
_STRIP_TAG_RE = re.compile(r"<[^>]+>")

FetchFn = Callable[[str, int], bytes]
SummarizeFn = Callable[[str, str, str], str]


class ImportSourceError(Exception):
    def __init__(self, code: str, message: str, *, status: str | None = None, hint: str | None = None) -> None:
        super().__init__(message)
        self.code = code
        self.message = message
        self.status = status
        self.hint = hint


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _slug(text: str) -> str:
    slug = re.sub(r"[^\w\-]+", "_", str(text or "source").strip())[:48]
    return slug or "source"


def _url_slug(url: str) -> str:
    parsed = urlparse(str(url or "").strip())
    base = parsed.path.strip("/").replace("/", "_") or parsed.netloc.replace(".", "_")
    return _slug(base)


def _normalize_locked_headings(locked_sections: list[str] | None) -> set[str]:
    locked: set[str] = set()
    for item in locked_sections or []:
        name = str(item).strip()
        if name.startswith("## "):
            name = name[3:].strip()
        elif name.startswith("#"):
            name = name.lstrip("#").strip()
        if name:
            locked.add(name)
    return locked


def _locked_sections(store: VirtualPersonaStore, vp_id: str) -> list[str] | None:
    try:
        pack = read_pack(store, vp_id)
    except FileNotFoundError:
        return None
    raw = pack.get("locked_sections")
    return list(raw) if isinstance(raw, list) else None


def _validate_https_url(url: str, *, allowlist: list[str] | None = None) -> str:
    clean = str(url or "").strip()
    parsed = urlparse(clean)
    if parsed.scheme != "https" or not parsed.netloc:
        raise ImportSourceError("invalid_url", "only https URLs are allowed")
    if allowlist and not url_allowed(clean, allowlist):
        raise ImportSourceError("url_not_allowed", "URL not in allowlist")
    return clean


def _runs_dir(store: VirtualPersonaStore, vp_id: str) -> Path:
    path = store.vp_dir(vp_id) / DEEP_LEARN_DIR / RUNS_DIR
    path.mkdir(parents=True, exist_ok=True)
    return path


def _sources_dir(store: VirtualPersonaStore, vp_id: str) -> Path:
    path = store.vp_dir(vp_id) / "sources"
    path.mkdir(parents=True, exist_ok=True)
    return path


def _html_to_text(html: str, *, max_chars: int = _SUMMARY_MAX_CHARS) -> str:
    stripped = _TAG_RE.sub(" ", str(html or ""))
    text = unescape(_STRIP_TAG_RE.sub(" ", stripped))
    text = re.sub(r"\s+", " ", text).strip()
    if len(text) > max_chars:
        text = text[: max_chars - 3] + "..."
    return text or "（无正文摘要）"


def _default_summarize(label: str, url: str, raw_text: str) -> str:
    body = (raw_text or "").strip() or "（待补充）"
    if len(body) > 240:
        body = body[:237] + "..."
    return f"链接导入 · {label} · {url}\n{body}"


def _write_run_record(
    store: VirtualPersonaStore,
    vp_id: str,
    *,
    kind: str,
    url: str,
    summary: str,
    source_file: str | None = None,
) -> dict[str, Any]:
    run_id = f"run_{uuid.uuid4().hex[:12]}"
    record: dict[str, Any] = {
        "run_id": run_id,
        "task_id": f"import_{_url_slug(url)}",
        "topic": IMPORT_SOURCE_TOPIC,
        "section": "作品与经历",
        "source_kind": kind,
        "source_url": url,
        "started_at": _now_iso(),
        "completed_at": _now_iso(),
        "summary": summary,
        "draft_body": summary,
        "import_source": True,
    }
    if source_file:
        record["source_file"] = source_file
    run_path = _runs_dir(store, vp_id) / f"{run_id}.json"
    with run_path.open("w", encoding="utf-8") as f:
        json.dump(record, f, ensure_ascii=False, indent=2)
        f.write("\n")
    return {"run_id": run_id, "run_path": str(run_path), "record": record}


def _append_meta_source(store: VirtualPersonaStore, vp_id: str, entry: dict[str, Any]) -> list[dict[str, Any]]:
    dest = store.vp_dir(vp_id)
    meta_path = dest / "meta.yaml"
    import yaml

    with meta_path.open(encoding="utf-8") as f:
        meta = yaml.safe_load(f) or {}
    if not isinstance(meta, dict):
        meta = {}
    rows = list(meta.get("sources") or [])
    rows.append(entry)
    meta["sources"] = rows
    meta["updated_at"] = _now_iso()
    with meta_path.open("w", encoding="utf-8") as f:
        yaml.safe_dump(meta, f, allow_unicode=True, sort_keys=False)
    return rows


def _merge_summary_into_target(
    store: VirtualPersonaStore,
    vp_id: str,
    *,
    merge_into: str,
    summary: str,
    url: str,
    respect_locked_sections: bool,
) -> dict[str, Any]:
    merged: dict[str, Any] = {"merge_into": merge_into, "applied": False}
    if merge_into == "sources":
        return merged

    locked_list = _locked_sections(store, vp_id) if respect_locked_sections else None
    locked = _normalize_locked_headings(locked_list)
    bullet = f"- [{url}]({url})\n  {summary[:400]}"

    if merge_into == "profile":
        row = store.get(vp_id)
        profile_md = str(row.get("profile_md") or "")
        heading = "作品与经历"
        if heading in locked:
            merged["skipped_locked"] = [heading]
            return merged
        updates = {heading: bullet}
        new_md = merge_profile_sections(profile_md, updates, only_empty=False)
        store.update(vp_id, profile_md=new_md)
        merged["applied"] = True
        merged["profile_section"] = heading
        return merged

    if merge_into == "assembly_master":
        heading = IMPORT_SOURCE_TOPIC
        if heading in locked:
            merged["skipped_locked"] = [heading]
            return merged
        ok = append_section_body(
            store,
            vp_id,
            heading,
            bullet,
            locked_sections=locked_list if respect_locked_sections else None,
        )
        merged["applied"] = ok
        merged["assembly_heading"] = heading
        return merged

    return merged


def _handle_video_stub(store: VirtualPersonaStore, vp_id: str, url: str) -> None:
    """Record pending video URL without fetch."""
    pending_dir = store.vp_dir(vp_id) / "import_source"
    pending_dir.mkdir(parents=True, exist_ok=True)
    pending_path = pending_dir / "pending.json"
    payload = {
        "url": url,
        "kind": "video",
        "status": "unsupported",
        "hint": "待视频解析 AI",
        "recorded_at": _now_iso(),
    }
    with pending_path.open("w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
        f.write("\n")
    _append_meta_source(
        store,
        vp_id,
        {
            "type": "import_source_video",
            "url": url,
            "status": "unsupported",
            "fetched_at": _now_iso(),
        },
    )


def run_import_source(
    store: VirtualPersonaStore,
    vp_id: str,
    *,
    url: str,
    kind: str = "web",
    merge_into: str = DEFAULT_MERGE_INTO,
    respect_locked_sections: bool = True,
    allowlist: list[str] | None = None,
    fetch_fn: FetchFn | None = None,
    summarize_fn: SummarizeFn | None = None,
) -> dict[str, Any]:
    """Fetch web/image URL · write sources + deep_learn run · video is unsupported."""
    clean_kind = str(kind or "web").strip().lower()
    if clean_kind not in VALID_KINDS:
        raise ImportSourceError("invalid_kind", f"kind must be one of {sorted(VALID_KINDS)}")

    clean_url = _validate_https_url(url, allowlist=allowlist)
    merge_target = str(merge_into or DEFAULT_MERGE_INTO).strip().lower()
    if merge_target not in VALID_MERGE_INTO:
        raise ImportSourceError("invalid_merge_into", f"merge_into must be one of {sorted(VALID_MERGE_INTO)}")

    row = store.get(vp_id)
    label = str(row.get("label") or vp_id)

    if clean_kind == "video":
        _handle_video_stub(store, vp_id, clean_url)
        raise ImportSourceError(
            "import_source_unsupported",
            "video import is not supported yet",
            status="unsupported",
            hint="待视频解析 AI",
        )

    def _fetch(target: str, limit: int) -> bytes:
        if fetch_fn is not None:
            return fetch_fn(target, limit)
        return _fetch_url_bytes(target, max_bytes=limit)

    fn = summarize_fn or _default_summarize
    source_file: str | None = None
    avatar_saved = False

    if clean_kind == "web":
        try:
            raw = _fetch(clean_url, _PAGE_MAX_BYTES)
        except ValueError as exc:
            if str(exc) == "response_too_large":
                raise ImportSourceError("response_too_large", "page exceeds size limit") from exc
            raise ImportSourceError("fetch_failed", str(exc)) from exc
        except Exception as exc:
            raise ImportSourceError("fetch_failed", str(exc) or "fetch failed") from exc
        html = raw.decode("utf-8", errors="replace")
        text = _html_to_text(html)
        summary = fn(label, clean_url, text)
        fname = f"import_{_url_slug(clean_url)}.txt"
        (_sources_dir(store, vp_id) / fname).write_text(
            f"{clean_url}\n\n{text}\n",
            encoding="utf-8",
        )
        source_file = fname
    else:
        if not _allowed_image_url(clean_url):
            raise ImportSourceError("invalid_image_url", "only https image URLs are allowed")
        try:
            data = _fetch(clean_url, _IMAGE_MAX_BYTES)
        except ValueError as exc:
            if str(exc) == "response_too_large":
                raise ImportSourceError("image_too_large", "image exceeds size limit") from exc
            raise ImportSourceError("fetch_failed", str(exc)) from exc
        except Exception as exc:
            raise ImportSourceError("fetch_failed", str(exc) or "fetch failed") from exc
        ext = ".png"
        lower = clean_url.lower()
        if ".jpg" in lower or ".jpeg" in lower:
            ext = ".jpg"
        elif ".webp" in lower:
            ext = ".webp"
        elif ".gif" in lower:
            ext = ".gif"
        fname = f"import_{_url_slug(clean_url)}{ext}"
        (_sources_dir(store, vp_id) / fname).write_bytes(data)
        source_file = fname
        summary = fn(label, clean_url, f"图片 {len(data)} bytes")
        if merge_target == "profile" and not store.has_avatar(vp_id):
            try:
                store.save_avatar(vp_id, data, kind="import_source", source_url=clean_url)
                avatar_saved = True
            except Exception:
                pass

    run_info = _write_run_record(
        store,
        vp_id,
        kind=clean_kind,
        url=clean_url,
        summary=summary,
        source_file=source_file,
    )
    meta_entry = {
        "type": f"import_source_{clean_kind}",
        "url": clean_url,
        "file": source_file,
        "run_id": run_info["run_id"],
        "fetched_at": _now_iso(),
    }
    meta_sources = _append_meta_source(store, vp_id, meta_entry)

    merge_result = _merge_summary_into_target(
        store,
        vp_id,
        merge_into=merge_target,
        summary=summary,
        url=clean_url,
        respect_locked_sections=respect_locked_sections,
    )

    return {
        "ok": True,
        "vp_id": vp_id,
        "kind": clean_kind,
        "url": clean_url,
        "summary": summary,
        "source_file": source_file,
        "run_id": run_info["run_id"],
        "run_path": run_info["run_path"],
        "meta_sources": meta_sources,
        "merge": merge_result,
        "avatar_saved": avatar_saved,
    }
