"""Keep one day of the exact text sent to a model and the text it sent back.

The admin page reads this store. Rows older than a day are dropped. Bodies are
capped so a long tool argument cannot grow the file without limit. Recording
never raises into the model call.
"""

from __future__ import annotations

import contextvars
import json
import logging
import os
import tempfile
import threading
import uuid
from collections.abc import Mapping
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

logger = logging.getLogger(__name__)

_RETENTION = timedelta(days=1)
_STORE_NAME = "model-exchanges.jsonl"
_MAX_RECORDS = 100
_MAX_FIELD_CHARS = 200_000
_PREVIEW_CHARS = 80
_TRUNCATED = "\n…（超出保存长度，后面已截断）"

_LOCK = threading.Lock()
_ACTIVE: contextvars.ContextVar[ExchangeCapture | None] = contextvars.ContextVar(
    "deerflow_model_exchange",
    default=None,
)
_store_override: Path | None = None


@dataclass
class ExchangeCapture:
    """One in-flight model call. Text stays here until the call finishes."""

    model: str = ""
    enable_thinking: str = "unset"
    request_text: str = ""
    request_body: str = ""
    request_preview: str = ""
    content_parts: list[str] = field(default_factory=list)
    reasoning_by_field: dict[str, list[str]] = field(default_factory=dict)
    reasoning_fields: list[str] = field(default_factory=list)
    raw_chunks: list[str] = field(default_factory=list)
    owned: bool = False
    token: contextvars.Token | None = None
    saved: bool = False


def use_exchange_store(path: Path | None) -> None:
    """Point the store at a test file. ``None`` restores the runtime data dir."""
    global _store_override
    _store_override = path


def open_exchange() -> ExchangeCapture:
    return ExchangeCapture()


def bind_exchange(capture: ExchangeCapture) -> contextvars.Token:
    return _ACTIVE.set(capture)


def unbind_exchange(token: contextvars.Token) -> None:
    _ACTIVE.reset(token)


def activate_exchange() -> tuple[ExchangeCapture, contextvars.Token]:
    capture = open_exchange()
    return capture, bind_exchange(capture)


def deactivate_exchange(capture: ExchangeCapture, token: contextvars.Token) -> None:
    try:
        finish_exchange(capture)
    finally:
        unbind_exchange(token)


def remember_request(payload: Mapping[str, Any]) -> None:
    """Copy the outbound messages once, after the payload is final."""
    try:
        capture = _ACTIVE.get()
        if capture is None:
            capture = ExchangeCapture(owned=True)
            capture.token = _ACTIVE.set(capture)
        elif capture.owned and capture.request_text and not capture.saved:
            _discard_owned(capture)
            capture = ExchangeCapture(owned=True)
            capture.token = _ACTIVE.set(capture)
        if capture.request_text:
            return
        messages = payload.get("messages")
        capture.model = str(payload.get("model") or "")
        capture.enable_thinking = _thinking_flag(payload)
        capture.request_preview = _preview(_last_user_text(messages))
        capture.request_text = _request_text(payload, messages)
        capture.request_body = _replay_request(payload)
    except Exception:
        logger.debug("model exchange request was not captured", exc_info=True)


def remember_provider_chunk(chunk: Mapping[str, Any]) -> None:
    """Append one provider chunk exactly as it arrived."""
    try:
        capture = _ACTIVE.get()
        if capture is None:
            return
        _store_raw(capture, chunk)
        for choice in _choices(chunk):
            delta = choice.get("delta") if isinstance(choice.get("delta"), Mapping) else {}
            message = choice.get("message") if isinstance(choice.get("message"), Mapping) else {}
            _append_channel(capture.content_parts, _text_field(delta, "content") or _text_field(message, "content"))
            _take_reasoning(capture, delta)
            _take_reasoning(capture, message)
    except Exception:
        logger.debug("model exchange chunk was not captured", exc_info=True)


def remember_raw_response(payload: Mapping[str, Any]) -> None:
    """Keep a non-streaming response body exactly as the provider sent it."""
    try:
        capture = _ACTIVE.get()
        if capture is None:
            return
        _store_raw(capture, payload, only_if_empty=True)
    except Exception:
        logger.debug("model exchange response was not captured", exc_info=True)


def remember_final_message(message: Mapping[str, Any]) -> None:
    """Record a non-streaming message when no chunks were captured."""
    try:
        capture = _ACTIVE.get()
        if capture is None:
            return
        _store_raw(capture, message, only_if_empty=True)
        if capture.content_parts or _has_reasoning(capture):
            return
        _append_channel(capture.content_parts, _text_field(message, "content"))
        _take_reasoning(capture, message)
    except Exception:
        logger.debug("model exchange message was not captured", exc_info=True)


def finish_owned_exchange() -> None:
    capture = _ACTIVE.get()
    if capture is None or not capture.owned:
        return
    token = capture.token
    try:
        finish_exchange(capture)
    finally:
        if token is not None:
            _ACTIVE.reset(token)


def finish_exchange(capture: ExchangeCapture) -> None:
    if capture.saved or not capture.request_text:
        capture.saved = True
        return
    capture.saved = True
    content = "".join(capture.content_parts)
    reasoning_by_field = {
        name: "".join(capture.reasoning_by_field.get(name, []))
        for name in capture.reasoning_fields
        if "".join(capture.reasoning_by_field.get(name, []))
    }
    reasoning = "".join(reasoning_by_field.values())
    reasoning_field = "、".join(reasoning_by_field) or "（没有单独的思考字段）"
    response_raw = _cap("\n\n".join(capture.raw_chunks))
    preview_source = content or reasoning
    record = {
        "id": uuid.uuid4().hex,
        "at": datetime.now(UTC).isoformat(timespec="seconds"),
        "model": capture.model,
        "user_id": _current_user_id(),
        "enable_thinking": capture.enable_thinking,
        "request_preview": capture.request_preview,
        "response_preview": _preview(preview_source),
        "request_text": _cap(capture.request_text),
        "request_body": capture.request_body,
        "response_text": response_raw,
        "response_raw": response_raw,
        "response_content": _cap(content),
        "response_reasoning": _cap(reasoning),
        "reasoning_field": reasoning_field,
    }
    _append_record(record)


def query_model_exchanges() -> list[dict[str, Any]]:
    """Newest first. Entries older than one day are dropped."""
    with _LOCK:
        records = _read_records()
        _write_records(records)
    return list(reversed(records))


def _discard_owned(capture: ExchangeCapture) -> None:
    capture.saved = True
    token = capture.token
    if token is not None:
        _ACTIVE.reset(token)


def _store_path() -> Path:
    if _store_override is not None:
        return _store_override
    from deerflow.config.paths import get_paths

    return get_paths().base_dir / _STORE_NAME


def _append_record(record: dict[str, Any]) -> None:
    try:
        with _LOCK:
            records = _read_records()
            records.append(record)
            if len(records) > _MAX_RECORDS:
                records = records[-_MAX_RECORDS:]
            _write_records(records)
    except Exception:
        logger.debug("model exchange was not saved", exc_info=True)


def _read_records() -> list[dict[str, Any]]:
    path = _store_path()
    if not path.is_file():
        return []
    cutoff = datetime.now(UTC) - _RETENTION
    kept: list[dict[str, Any]] = []
    for line in path.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        try:
            record = json.loads(line)
        except json.JSONDecodeError:
            continue
        if isinstance(record, dict) and _record_time(record) >= cutoff:
            kept.append(record)
    if len(kept) > _MAX_RECORDS:
        kept = kept[-_MAX_RECORDS:]
    return kept


def _write_records(records: list[dict[str, Any]]) -> None:
    path = _store_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".model-exchanges.", suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            for record in records:
                handle.write(json.dumps(record, ensure_ascii=False) + "\n")
        os.replace(tmp, path)
    except Exception:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise


def _record_time(record: Mapping[str, Any]) -> datetime:
    raw = str(record.get("at") or "")
    try:
        parsed = datetime.fromisoformat(raw)
    except ValueError:
        return datetime.min.replace(tzinfo=UTC)
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=UTC)
    return parsed.astimezone(UTC)


def _current_user_id() -> str:
    try:
        from deerflow.runtime.user_context import get_effective_user_id

        return get_effective_user_id()
    except Exception:
        return ""


def _thinking_flag(payload: Mapping[str, Any]) -> str:
    extra_body = payload.get("extra_body")
    value = extra_body.get("enable_thinking", payload.get("enable_thinking")) if isinstance(extra_body, Mapping) else payload.get("enable_thinking")
    if value is True:
        return "true"
    if value is False:
        return "false"
    if value is None:
        return "unset"
    return "other"


def _choices(chunk: Mapping[str, Any]) -> list[Mapping[str, Any]]:
    choices = chunk.get("choices")
    if not isinstance(choices, list) or not choices:
        nested = chunk.get("chunk")
        if isinstance(nested, Mapping):
            choices = nested.get("choices")
    if not isinstance(choices, list):
        return []
    return [choice for choice in choices if isinstance(choice, Mapping)]


def _has_reasoning(capture: ExchangeCapture) -> bool:
    return any(parts for parts in capture.reasoning_by_field.values())


def _take_reasoning(capture: ExchangeCapture, source: Mapping[str, Any]) -> None:
    """Keep thinking text under the wire field that actually carried it."""
    for name in ("reasoning_content", "reasoning"):
        text = _text_field(source, name)
        if not text:
            continue
        if name not in capture.reasoning_fields:
            capture.reasoning_fields.append(name)
        _append_channel(capture.reasoning_by_field.setdefault(name, []), text)


def _text_field(payload: Mapping[str, Any], key: str) -> str:
    value = payload.get(key)
    if isinstance(value, str):
        return value
    return _content_to_text(value) if key == "content" else ""


def _content_to_text(content: Any) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for item in content:
            if isinstance(item, str):
                parts.append(item)
            elif isinstance(item, Mapping):
                text = item.get("text")
                if isinstance(text, str) and text:
                    parts.append(text)
                elif item.get("type") == "image_url":
                    parts.append("[image]")
        return "\n".join(parts)
    if content is None:
        return ""
    return str(content)


def _last_user_text(messages: Any) -> str:
    if not isinstance(messages, list):
        return ""
    for message in reversed(messages):
        if isinstance(message, Mapping) and message.get("role") == "user":
            return _content_to_text(message.get("content"))
    return ""


_SECRET_KEYS = frozenset({"api_key", "openai_api_key", "authorization", "token"})


def _replay_request(payload: Mapping[str, Any]) -> str:
    """JSON body the OpenAI client actually posts. ``extra_body`` is flattened."""
    ready = _json_ready(payload)
    if not isinstance(ready, dict):
        return "{}"
    extra = ready.pop("extra_body", None)
    if isinstance(extra, dict):
        ready.update(extra)
    return _fit_replay(ready)


def _store_raw(capture: ExchangeCapture, payload: Mapping[str, Any], *, only_if_empty: bool = False) -> None:
    if only_if_empty and capture.raw_chunks:
        return
    if sum(len(part) for part in capture.raw_chunks) >= _MAX_FIELD_CHARS:
        return
    capture.raw_chunks.append(json.dumps(payload, ensure_ascii=False, indent=2, default=str))


def _json_ready(value: Any) -> Any:
    if value is None or isinstance(value, (str, int, float, bool)):
        return value
    if isinstance(value, Mapping):
        return {
            str(key): _json_ready(item)
            for key, item in value.items()
            if item is not None and str(key) not in _SECRET_KEYS
        }
    if isinstance(value, (list, tuple)):
        return [_json_ready(item) for item in value]
    return str(value)


def _fit_replay(body: dict[str, Any]) -> str:
    encoded = json.dumps(body, ensure_ascii=False, indent=2)
    if len(encoded) <= _MAX_FIELD_CHARS:
        return encoded
    tools = body.get("tools")
    if isinstance(tools, list) and tools:
        names: list[str] = []
        for tool in tools:
            function = tool.get("function") if isinstance(tool, dict) else None
            name = function.get("name") if isinstance(function, dict) else None
            if isinstance(name, str) and name and name not in names:
                names.append(name)
        body["tools"] = [{"_omitted": "工具定义过长，这里只保留工具名", "names": names}]
        encoded = json.dumps(body, ensure_ascii=False, indent=2)
        if len(encoded) <= _MAX_FIELD_CHARS:
            return encoded
    messages = body.get("messages")
    if isinstance(messages, list):
        for _ in range(12):
            if len(encoded) <= _MAX_FIELD_CHARS:
                break
            longest: dict[str, Any] | None = None
            longest_len = 0
            for message in messages:
                if not isinstance(message, dict):
                    continue
                content = message.get("content")
                if isinstance(content, str) and len(content) > longest_len:
                    longest = message
                    longest_len = len(content)
            if longest is None or longest_len < 200:
                break
            longest["content"] = longest["content"][: max(80, longest_len // 2)] + _TRUNCATED.strip()
            encoded = json.dumps(body, ensure_ascii=False, indent=2)
    if len(encoded) > _MAX_FIELD_CHARS:
        body["messages"] = [{"role": "user", "content": _TRUNCATED.strip()}]
        encoded = json.dumps(body, ensure_ascii=False, indent=2)
    return encoded


def _request_text(payload: Mapping[str, Any], messages: Any) -> str:
    tool_names: list[str] = []
    tools = payload.get("tools")
    if isinstance(tools, list):
        for tool in tools:
            function = tool.get("function") if isinstance(tool, Mapping) else None
            name = function.get("name") if isinstance(function, Mapping) else None
            if isinstance(name, str) and name and name not in tool_names:
                tool_names.append(name)
    lines = [
        f"enable_thinking: {_thinking_flag(payload)}",
        f"tools: {', '.join(tool_names) if tool_names else '（无）'}",
        "",
    ]
    if isinstance(messages, list):
        for message in messages:
            if not isinstance(message, Mapping):
                continue
            role = str(message.get("role") or "unknown")
            lines.append(f"[{role}]")
            lines.append(_content_to_text(message.get("content")))
            lines.append("")
    return "\n".join(lines).rstrip()


def _append_channel(parts: list[str], incoming: str) -> None:
    if not incoming:
        return
    current = "".join(parts)
    if current and (incoming == current or incoming.startswith(current)):
        parts.clear()
        parts.append(incoming)
        return
    if current and current.endswith(incoming):
        return
    parts.append(incoming)


def _preview(text: str) -> str:
    compact = " ".join(text.split())
    if len(compact) <= _PREVIEW_CHARS:
        return compact
    return compact[: _PREVIEW_CHARS - 1] + "…"


def _cap(text: str) -> str:
    if len(text) <= _MAX_FIELD_CHARS:
        return text
    return text[: _MAX_FIELD_CHARS - len(_TRUNCATED)] + _TRUNCATED