"""Input Bus — handler registry · enrichments · LLM context blocks (v0.18 M3)."""

from __future__ import annotations

import base64
import hashlib
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Protocol

from niannian.chat.attachments import get_attachment_blob_path, get_attachment_meta
from niannian.chat.stickers_catalog import PACK_LABELS, StickerRow, build_sticker_catalog
from niannian.chat.user_stickers import USER_STICKER_PACK_ID, get_user_sticker, user_sticker_label
from niannian.llm.types import LLMUnavailable
from niannian.vault.text_fetch import extract_text_from_html, fetch_url_html, is_text_url_allowed, load_text_hosts_config


class PartHandler(Protocol):
    def process(self, ctx: InputBusContext, part: dict[str, Any]) -> dict[str, Any] | None: ...


@dataclass
class InputBusContext:
    user_id: str
    config: dict[str, Any]
    home: Path | None = None
    llm: Any | None = None
    async_enrichment: bool = False
    job_store: Any | None = None
    session_id: str | None = None
    turn_id: str | None = None
    sessions: Any | None = None


def enrichment_key_for_part(part: dict[str, Any]) -> str | None:
    ptype = part.get("type")
    if ptype == "text":
        return None
    if ptype == "sticker":
        return f"sticker:{part['sticker_id']}"
    if ptype == "link":
        digest = hashlib.sha256(part["url"].encode("utf-8")).hexdigest()[:16]
        return f"link:{digest}"
    if ptype in ("image", "file"):
        return part["attachment_id"]
    return None


class TextHandler:
    def process(self, ctx: InputBusContext, part: dict[str, Any]) -> dict[str, Any] | None:
        return None


class StickerHandler:
    def __init__(self, catalog: dict[str, StickerRow] | None = None) -> None:
        self._catalog = catalog if catalog is not None else build_sticker_catalog()

    def process(self, ctx: InputBusContext, part: dict[str, Any]) -> dict[str, Any] | None:
        sticker_id = part["sticker_id"]
        row = self._catalog.get(sticker_id)
        if row is None and ctx.home is not None:
            user_row = get_user_sticker(ctx.home, ctx.user_id, sticker_id)
            if user_row is not None:
                return {
                    "kind": "sticker",
                    "status": "ok",
                    "label": user_sticker_label(ctx.home, ctx.user_id, sticker_id),
                    "pack": USER_STICKER_PACK_ID,
                    "tags": str(user_row.get("tags") or "自定义"),
                    "persisted": False,
                }
        if row is None:
            raise ValueError(f"unknown sticker_id: {sticker_id!r}")
        return {
            "kind": "sticker",
            "status": "ok",
            "label": row["label"],
            "pack": row["pack"],
            "tags": row["tags"],
            "persisted": False,
        }


class ImageHandler:
    def process(self, ctx: InputBusContext, part: dict[str, Any]) -> dict[str, Any] | None:
        attachment_id = part["attachment_id"]
        if ctx.home is None:
            return _image_failed("no_home")
        meta = get_attachment_meta(ctx.home, ctx.user_id, attachment_id)
        blob_path = get_attachment_blob_path(ctx.home, ctx.user_id, attachment_id)
        if meta is None or blob_path is None:
            return _image_failed("attachment_not_found")
        if ctx.llm is None or not ctx.llm.is_module_configured("ocr"):
            return _image_failed("ocr_unavailable")
        mime = str(meta.get("mime") or "application/octet-stream")
        image_b64 = base64.b64encode(blob_path.read_bytes()).decode("ascii")
        try:
            result = ctx.llm.ocr("ocr", image_b64, mime=mime)
        except LLMUnavailable:
            return _image_failed("ocr_unavailable")
        except Exception as exc:
            return _image_failed(type(exc).__name__)
        ocr_text = (result.text or "").strip()
        if not ocr_text:
            return _image_failed("ocr_empty")
        out: dict[str, Any] = {
            "kind": "image",
            "status": "ok",
            "ocr_text": ocr_text,
            "persisted": False,
        }
        vision = _optional_vision_summary(ctx, image_b64, mime=mime, ocr_text=ocr_text)
        if vision:
            out["vision_summary"] = vision
        return out


def _image_failed(error_code: str) -> dict[str, Any]:
    return {
        "kind": "image",
        "status": "failed",
        "error_code": error_code,
        "persisted": False,
    }


def _optional_vision_summary(
    ctx: InputBusContext,
    image_b64: str,
    *,
    mime: str,
    ocr_text: str,
) -> str | None:
    """Heavy-tier optional one-line image describe via OCR module (no extra vision slot)."""
    from niannian.token.budget import load_token_budget, resolve_chat_llm_level

    if ctx.home is None or ctx.llm is None:
        return None
    budget = load_token_budget(ctx.home, ctx.user_id)
    if resolve_chat_llm_level(budget) != "heavy":
        return None
    if not ctx.llm.is_module_configured("ocr"):
        return None
    prompt = "用一句话概括这张图片的内容（中文，不超过40字）。"
    try:
        result = ctx.llm.ocr("ocr", image_b64, mime=mime, prompt=prompt)
    except (LLMUnavailable, Exception):
        return None
    summary = (result.text or "").strip()
    if not summary or summary == ocr_text:
        return None
    return summary[:120]


class LinkHandler:
    def __init__(self, fetch_html: Callable[..., str] | None = None) -> None:
        self._fetch_html = fetch_html or fetch_url_html

    def process(self, ctx: InputBusContext, part: dict[str, Any]) -> dict[str, Any] | None:
        url = part["url"]
        cfg = load_text_hosts_config()
        if not is_text_url_allowed(url, cfg):
            return _link_failed("host_not_allowed")
        try:
            html = self._fetch_html(url, cfg=cfg)
            title, body = extract_text_from_html(html)
            snippet = body[:400].replace("\n", " ").strip()
            return {
                "kind": "link",
                "status": "ok",
                "title": title,
                "snippet": snippet,
                "md_preview": body[:2000],
                "fetched_at": _now_iso(),
                "persisted": False,
            }
        except ValueError as exc:
            return _link_failed(str(exc))
        except Exception as exc:
            return _link_failed(type(exc).__name__)


class FileHandler:
    def process(self, ctx: InputBusContext, part: dict[str, Any]) -> dict[str, Any] | None:
        attachment_id = part["attachment_id"]
        name = str(part.get("name") or "")
        mime = str(part.get("mime") or "")
        if ctx.home is None:
            return _file_failed("no_home")
        blob_path = get_attachment_blob_path(ctx.home, ctx.user_id, attachment_id)
        if blob_path is None:
            return _file_failed("attachment_not_found")
        try:
            body_text = _read_file_text(blob_path.read_bytes(), mime=mime)
            summary = body_text[:200].replace("\n", " ").strip()
            return {
                "kind": "file",
                "status": "ok",
                "name": name,
                "mime": mime,
                "extract_summary": summary,
                "persisted": False,
            }
        except ValueError as exc:
            return _file_failed(str(exc))
        except Exception as exc:
            return _file_failed(type(exc).__name__)


def _now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def _link_failed(error_code: str) -> dict[str, Any]:
    return {
        "kind": "link",
        "status": "failed",
        "error_code": error_code,
        "persisted": False,
    }


def _file_failed(error_code: str) -> dict[str, Any]:
    return {
        "kind": "file",
        "status": "failed",
        "error_code": error_code,
        "persisted": False,
    }


def _read_file_text(data: bytes, *, mime: str) -> str:
    mt = mime.lower()
    if mt.startswith("text/") or mt in ("application/json", "application/xml"):
        text = data.decode("utf-8", errors="replace").strip()
        if not text:
            raise ValueError("empty_file")
        return text
    raise ValueError("unsupported_mime")


HANDLERS: dict[str, PartHandler] = {
    "text": TextHandler(),
    "sticker": StickerHandler(),
    "image": ImageHandler(),
    "link": LinkHandler(),
    "file": FileHandler(),
}


def is_sticker_only_parts(parts: list[dict[str, Any]]) -> bool:
    if not parts:
        return False
    if any(p.get("type") == "text" and str(p.get("text", "")).strip() for p in parts):
        return False
    return all(p.get("type") == "sticker" for p in parts)


def process_parts(ctx: InputBusContext, parts: list[dict[str, Any]]) -> dict[str, Any]:
    enrichments: dict[str, Any] = {}
    for part in parts:
        ptype = str(part.get("type") or "")
        handler = HANDLERS.get(ptype)
        if handler is None:
            continue
        key = enrichment_key_for_part(part)
        if (
            ctx.async_enrichment
            and ctx.job_store is not None
            and ctx.session_id
            and ctx.turn_id
            and ctx.sessions is not None
            and ptype in ("link", "file")
            and key is not None
        ):
            job_id = ctx.job_store.enqueue(
                session_id=ctx.session_id,
                turn_id=ctx.turn_id,
                enrichment_key=key,
                kind=ptype,
                worker=lambda h=handler, p=part: h.process(ctx, p),
                on_complete=_make_turn_patch_callback(ctx, key),
            )
            enrichments[key] = {
                "kind": ptype,
                "status": "pending",
                "job_id": job_id,
                "persisted": False,
            }
            continue
        try:
            result = handler.process(ctx, part)
        except Exception as exc:
            if key is None:
                continue
            enrichments[key] = {
                "kind": ptype,
                "status": "failed",
                "error_code": type(exc).__name__,
                "persisted": False,
            }
            continue
        if key is not None and result is not None:
            enrichments[key] = result
    return enrichments


def _make_turn_patch_callback(ctx: InputBusContext, enrichment_key: str):
    def _patch(_job_id: str, enrichment: dict[str, Any] | None, error: str | None) -> None:
        if ctx.sessions is None or not ctx.session_id or not ctx.turn_id:
            return
        try:
            session = ctx.sessions.require(ctx.session_id)
        except Exception:
            return
        for turn in session.turns:
            if turn.id != ctx.turn_id:
                continue
            meta = dict(turn.meta)
            enr = dict(meta.get("enrichments") or {})
            if enrichment is not None:
                enr[enrichment_key] = enrichment
            else:
                enr[enrichment_key] = {
                    "kind": enr.get(enrichment_key, {}).get("kind") or "link",
                    "status": "failed",
                    "error_code": error or "job_failed",
                    "persisted": False,
                }
            meta["enrichments"] = enr
            turn.meta = meta
            break

    return _patch


def _sticker_context_line(enrichment: dict[str, Any]) -> str:
    label = enrichment.get("label") or ""
    pack_id = str(enrichment.get("pack") or "")
    pack_label = PACK_LABELS.get(pack_id, pack_id)
    tags = enrichment.get("tags") or ""
    return f"[用户发送贴纸：{label} · {pack_label}系 · {tags}]"


def _enrichment_context_line(kind: str, enrichment: dict[str, Any]) -> str | None:
    status = enrichment.get("status")
    if status == "pending":
        return None
    if status == "failed":
        if kind == "image":
            return "（图片识别失败）"
        if kind == "link":
            return "（链接解析失败）"
        if kind == "file":
            return "（文件解析失败）"
        return None
    if status != "ok":
        return None
    if kind == "sticker":
        return _sticker_context_line(enrichment)
    if kind == "image":
        parts: list[str] = []
        if enrichment.get("ocr_text"):
            parts.append(str(enrichment["ocr_text"]))
        if enrichment.get("vision_summary"):
            parts.append(str(enrichment["vision_summary"]))
        return "\n".join(parts) if parts else None
    if kind == "link":
        title = enrichment.get("title") or ""
        snippet = enrichment.get("snippet") or ""
        if title and snippet:
            return f"{title}\n{snippet}"
        return title or snippet or None
    if kind == "file":
        name = enrichment.get("name") or ""
        summary = enrichment.get("extract_summary") or ""
        if name and summary:
            return f"{name}\n{summary}"
        return name or summary or None
    return None


def build_llm_context_block(parts: list[dict[str, Any]], enrichments: dict[str, Any]) -> str:
    blocks: list[str] = []
    for part in parts:
        ptype = part.get("type")
        if ptype == "text":
            text = str(part.get("text") or "").strip()
            if text:
                blocks.append(text)
            continue
        key = enrichment_key_for_part(part)
        if key is None:
            continue
        enrichment = enrichments.get(key)
        if not enrichment:
            continue
        line = _enrichment_context_line(str(enrichment.get("kind") or ptype), enrichment)
        if line:
            blocks.append(line)
    return "\n".join(blocks)
