"""LLM 分析服务 (MVP-3 / D29)

调用多模态大模型 (GPT-4V / Qwen-VL / GLM-4V) 对可疑事件进行二次判断。

设计要点:

1. **provider 抽象**: ``BaseLLMProvider`` 定义统一接口，
   ``OpenAICompatibleProvider`` 覆盖所有 OpenAI 协议兼容的 vendor，
   ``MockLLMProvider`` 提供完全离线的可预测行为，便于测试与本地开发。
2. **并发控制**: 使用 ``asyncio.Semaphore`` 限制同时进行的 LLM 调用数量，
   避免大量并发请求拖垮 API 配额。
3. **降级策略**: 上层调用方可通过 ``analyze_with_fallback`` 在 LLM 失败时
   返回 ``confirmed=None`` 的"未确认"结果，由 ``ResultFusion`` 决定如何处理。
4. **图像编码**: 自动按最大边长缩放后编码为 base64 JPEG，控制 token 成本。
5. **结构化输出**: prompt 强制 LLM 输出 JSON，解析失败时降级为
   ``confirmed=None`` 而不是抛异常。

依赖说明:

- HTTP 客户端使用 ``httpx`` (已在 requirements 中)
- 图像编码使用 ``opencv-python`` 已存在的 cv2.imencode

无需新增 Python 依赖即可运行 (mock provider)，仅当 ``provider != mock`` 时
才会真正发起 HTTP 请求。
"""

from __future__ import annotations

import asyncio
import base64
import json
import logging
import re
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Sequence

import numpy as np

from models.event_schemas import CandidateEvent, EventType

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# 数据结构
# ---------------------------------------------------------------------------


@dataclass
class LLMAnalysisRequest:
    """LLM 分析请求。

    Attributes:
        candidate: 触发分析的候选事件 (含事件类型 / bbox)
        frames: 关键帧序列 (BGR ``np.ndarray``)，
            通常取累积窗口内的 1-3 张代表帧
        prompt_extra: 业务侧附加提示词 (可选)
        request_id: 请求标识 (用于日志关联)
    """

    candidate: CandidateEvent
    frames: Sequence[np.ndarray]
    prompt_extra: Optional[str] = None
    request_id: Optional[str] = None


@dataclass
class LLMAnalysisResult:
    """LLM 分析结果。

    Attributes:
        confirmed: True=LLM 确认事件成立，False=否决，None=未知/调用失败
        confidence: LLM 给出的置信度 [0, 1]
        reasoning: LLM 的简短推理说明
        provider: 实际调用的 provider 名称
        model: 实际调用的模型名
        latency_ms: 调用耗时 (毫秒)
        error: 失败时的错误信息
        raw: 原始返回 (调试用)
    """

    confirmed: Optional[bool]
    confidence: float = 0.0
    reasoning: str = ""
    provider: str = ""
    model: str = ""
    latency_ms: float = 0.0
    error: Optional[str] = None
    raw: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)


# ---------------------------------------------------------------------------
# Provider 抽象
# ---------------------------------------------------------------------------


class BaseLLMProvider:
    """LLM provider 抽象。"""

    name: str = "base"

    async def analyze(self, request: LLMAnalysisRequest) -> LLMAnalysisResult:
        raise NotImplementedError


class MockLLMProvider(BaseLLMProvider):
    """离线 Mock provider。

    根据候选事件置信度生成可预测的判定结果，便于测试与无 API Key 环境。

    规则:
    - confidence >= 0.7  -> confirmed=True
    - confidence <  0.3  -> confirmed=False
    - 其余 -> confirmed=None
    """

    name = "mock"

    def __init__(self, model: str = "mock-vlm") -> None:
        self.model = model

    async def analyze(self, request: LLMAnalysisRequest) -> LLMAnalysisResult:
        start = time.time()
        await asyncio.sleep(0)  # 模拟异步
        conf = float(request.candidate.confidence)
        if conf >= 0.7:
            confirmed: Optional[bool] = True
        elif conf < 0.3:
            confirmed = False
        else:
            confirmed = None

        reasoning = (
            f"mock provider: candidate confidence={conf:.3f}, "
            f"event_type={request.candidate.event_type.value}"
        )
        return LLMAnalysisResult(
            confirmed=confirmed,
            confidence=conf,
            reasoning=reasoning,
            provider=self.name,
            model=self.model,
            latency_ms=(time.time() - start) * 1000.0,
        )


class OpenAICompatibleProvider(BaseLLMProvider):
    """OpenAI 协议兼容 provider (覆盖 GPT-4V / Qwen-VL / GLM-4V 等)。"""

    name = "openai"

    def __init__(
        self,
        api_base: str,
        api_key: str,
        model: str,
        timeout: float = 15.0,
        max_retries: int = 2,
        max_tokens: int = 512,
        temperature: float = 0.0,
        provider_name: Optional[str] = None,
    ) -> None:
        if not api_base:
            raise ValueError("api_base 不能为空")
        if not api_key:
            raise ValueError("api_key 不能为空")
        self.api_base = api_base.rstrip("/")
        self.api_key = api_key
        self.model = model
        self.timeout = timeout
        self.max_retries = max_retries
        self.max_tokens = max_tokens
        self.temperature = temperature
        if provider_name:
            self.name = provider_name

    async def analyze(self, request: LLMAnalysisRequest) -> LLMAnalysisResult:
        try:
            import httpx  # noqa: WPS433
        except ImportError as exc:  # pragma: no cover - 依赖在 requirements 已声明
            return LLMAnalysisResult(
                confirmed=None,
                provider=self.name,
                model=self.model,
                error=f"httpx 未安装: {exc}",
            )

        url = f"{self.api_base}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        payload = {
            "model": self.model,
            "messages": _build_messages(request),
            "max_tokens": self.max_tokens,
            "temperature": self.temperature,
        }

        attempt = 0
        last_error: Optional[str] = None
        start = time.time()
        while attempt <= self.max_retries:
            attempt += 1
            try:
                async with httpx.AsyncClient(timeout=self.timeout) as client:
                    resp = await client.post(url, headers=headers, json=payload)
                    resp.raise_for_status()
                    data = resp.json()
                content = (
                    data.get("choices", [{}])[0]
                    .get("message", {})
                    .get("content", "")
                )
                parsed = _parse_llm_json(content)
                return LLMAnalysisResult(
                    confirmed=parsed.get("confirmed"),
                    confidence=float(parsed.get("confidence", 0.0) or 0.0),
                    reasoning=str(parsed.get("reasoning", "")),
                    provider=self.name,
                    model=self.model,
                    latency_ms=(time.time() - start) * 1000.0,
                    raw=content,
                )
            except Exception as exc:  # noqa: BLE001
                last_error = f"{type(exc).__name__}: {exc}"
                logger.warning(
                    "LLM 调用失败 attempt=%d/%d err=%s",
                    attempt,
                    self.max_retries + 1,
                    last_error,
                )
                if attempt > self.max_retries:
                    break
                await asyncio.sleep(min(2 ** (attempt - 1), 5))

        return LLMAnalysisResult(
            confirmed=None,
            provider=self.name,
            model=self.model,
            latency_ms=(time.time() - start) * 1000.0,
            error=last_error,
        )


# ---------------------------------------------------------------------------
# Service
# ---------------------------------------------------------------------------


class LLMAnalysisService:
    """LLM 二次判断服务。

    Args:
        provider: LLM provider 实例 (Mock / OpenAI 兼容)
        max_concurrency: 同时进行的最大调用数
        image_max_side: 编码前图像最大边长 (像素)
    """

    def __init__(
        self,
        provider: BaseLLMProvider,
        max_concurrency: int = 2,
        image_max_side: int = 768,
    ) -> None:
        if max_concurrency < 1:
            raise ValueError("max_concurrency 必须 >= 1")
        if image_max_side < 64:
            raise ValueError("image_max_side 必须 >= 64")
        self.provider = provider
        self._semaphore = asyncio.Semaphore(max_concurrency)
        self.image_max_side = image_max_side

        # 统计
        self._call_count = 0
        self._success_count = 0
        self._failure_count = 0
        self._total_latency_ms = 0.0

    async def analyze(self, request: LLMAnalysisRequest) -> LLMAnalysisResult:
        """调用 LLM 分析单个事件 (并发受信号量限制)。"""

        # 图像缩放预处理 (mock provider 也走一遍，保持行为一致)
        prepared_frames = [
            _resize_frame(f, self.image_max_side) for f in request.frames
        ]
        if any(p is not o for p, o in zip(prepared_frames, request.frames)):
            request = LLMAnalysisRequest(
                candidate=request.candidate,
                frames=prepared_frames,
                prompt_extra=request.prompt_extra,
                request_id=request.request_id,
            )

        async with self._semaphore:
            self._call_count += 1
            try:
                result = await self.provider.analyze(request)
            except Exception as exc:  # noqa: BLE001
                logger.error("LLM provider 异常: %s", exc)
                result = LLMAnalysisResult(
                    confirmed=None,
                    provider=getattr(self.provider, "name", "unknown"),
                    model=getattr(self.provider, "model", ""),
                    error=f"{type(exc).__name__}: {exc}",
                )

        if result.error or result.confirmed is None:
            self._failure_count += 1
        else:
            self._success_count += 1
        self._total_latency_ms += result.latency_ms
        return result

    async def analyze_with_fallback(
        self,
        request: LLMAnalysisRequest,
    ) -> LLMAnalysisResult:
        """带降级的分析: 任意异常都不会向上抛出。"""

        try:
            return await self.analyze(request)
        except Exception as exc:  # noqa: BLE001
            logger.error("LLM 分析降级: %s", exc)
            return LLMAnalysisResult(
                confirmed=None,
                provider=getattr(self.provider, "name", "unknown"),
                model=getattr(self.provider, "model", ""),
                error=f"{type(exc).__name__}: {exc}",
            )

    @property
    def stats(self) -> Dict[str, Any]:
        avg_latency = (
            self._total_latency_ms / self._call_count
            if self._call_count > 0
            else 0.0
        )
        return {
            "provider": getattr(self.provider, "name", "unknown"),
            "model": getattr(self.provider, "model", ""),
            "call_count": self._call_count,
            "success_count": self._success_count,
            "failure_count": self._failure_count,
            "avg_latency_ms": round(avg_latency, 2),
        }


# ---------------------------------------------------------------------------
# Factory
# ---------------------------------------------------------------------------


def create_llm_service_from_settings(settings: Any) -> Optional[LLMAnalysisService]:
    """根据 ``Settings.llm`` 配置构造 LLMAnalysisService。

    返回 None 表示未启用，调用方应跳过 LLM 分析。
    """

    llm_cfg = getattr(settings, "llm", None)
    if llm_cfg is None or not getattr(llm_cfg, "enabled", False):
        return None

    provider_name = (llm_cfg.provider or "mock").lower()
    if provider_name == "mock":
        provider: BaseLLMProvider = MockLLMProvider(model=llm_cfg.model or "mock-vlm")
    else:
        api_base = llm_cfg.api_base
        api_key = llm_cfg.api_key
        if not api_base or not api_key:
            logger.warning(
                "LLM provider=%s 缺少 api_base/api_key，回退到 mock", provider_name
            )
            provider = MockLLMProvider(model=llm_cfg.model or "mock-vlm")
        else:
            provider = OpenAICompatibleProvider(
                api_base=api_base,
                api_key=api_key,
                model=llm_cfg.model,
                timeout=llm_cfg.timeout,
                max_retries=llm_cfg.max_retries,
                max_tokens=llm_cfg.max_tokens,
                temperature=llm_cfg.temperature,
                provider_name=provider_name,
            )

    return LLMAnalysisService(
        provider=provider,
        max_concurrency=llm_cfg.max_concurrency,
        image_max_side=llm_cfg.image_max_side,
    )


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


_EVENT_TYPE_DESCRIPTIONS: Dict[EventType, str] = {
    EventType.FIRE: "fire / open flame",
    EventType.SMOKE: "smoke / haze",
    EventType.SMOKING: "a person smoking a cigarette",
    EventType.FIGHT: "a fight or physical altercation",
    EventType.LOITERING: "a person loitering / lingering suspiciously",
    EventType.STATIONARY: "a stationary person (possibly fallen or unconscious)",
    EventType.INTRUSION: "an intrusion into a restricted area",
    EventType.ILLEGAL_PARKING: "a vehicle illegally parked",
    EventType.VEHICLE: "a vehicle of interest",
    EventType.PERSON: "a person of interest",
    EventType.UNKNOWN: "an unspecified suspicious event",
}


def _build_messages(request: LLMAnalysisRequest) -> List[Dict[str, Any]]:
    """构造 OpenAI ChatCompletion messages。"""

    candidate = request.candidate
    event_desc = _EVENT_TYPE_DESCRIPTIONS.get(candidate.event_type, "an event")
    bbox = candidate.detection.bbox.to_list()

    system_prompt = (
        "You are a strict video surveillance auditor. "
        "Given one or more frames and a candidate event reported by an AI detector, "
        "decide whether the event is truly present. "
        "Respond with a single JSON object: "
        '{"confirmed": true|false, "confidence": 0.0-1.0, "reasoning": "..."} '
        "and nothing else."
    )

    user_text_parts: List[str] = [
        f"Candidate event: {event_desc}.",
        f"Detector confidence: {candidate.confidence:.3f}.",
        f"Bounding box (x1,y1,x2,y2): {bbox}.",
    ]
    if request.prompt_extra:
        user_text_parts.append(request.prompt_extra)
    user_text_parts.append(
        "Please answer strictly in JSON: "
        '{"confirmed": <bool>, "confidence": <float>, "reasoning": <str>}'
    )

    content: List[Dict[str, Any]] = [
        {"type": "text", "text": "\n".join(user_text_parts)},
    ]
    for frame in request.frames:
        b64 = _encode_frame_base64(frame)
        if b64:
            content.append(
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/jpeg;base64,{b64}"},
                }
            )

    return [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": content},
    ]


def _encode_frame_base64(frame: np.ndarray) -> Optional[str]:
    """将 BGR 帧编码为 base64 JPEG 字符串。"""

    try:
        import cv2  # noqa: WPS433
    except ImportError:  # pragma: no cover
        return None
    if frame is None or frame.size == 0:
        return None
    ok, buf = cv2.imencode(".jpg", frame, [int(cv2.IMWRITE_JPEG_QUALITY), 80])
    if not ok:
        return None
    return base64.b64encode(buf.tobytes()).decode("ascii")


def _resize_frame(frame: np.ndarray, max_side: int) -> np.ndarray:
    """按最大边长等比缩放 (节省 token)。"""

    if frame is None or frame.size == 0:
        return frame
    try:
        import cv2  # noqa: WPS433
    except ImportError:  # pragma: no cover
        return frame
    h, w = frame.shape[:2]
    longest = max(h, w)
    if longest <= max_side:
        return frame
    scale = max_side / float(longest)
    new_size = (max(1, int(w * scale)), max(1, int(h * scale)))
    return cv2.resize(frame, new_size, interpolation=cv2.INTER_AREA)


_JSON_PATTERN = re.compile(r"\{.*\}", re.DOTALL)


def _parse_llm_json(content: str) -> Dict[str, Any]:
    """从 LLM 文本输出中尽力解析 JSON。"""

    if not content:
        return {}
    try:
        return json.loads(content)
    except json.JSONDecodeError:
        match = _JSON_PATTERN.search(content)
        if match:
            try:
                return json.loads(match.group(0))
            except json.JSONDecodeError:
                logger.debug("LLM JSON 解析失败: %s", content[:200])
    return {}


__all__ = [
    "LLMAnalysisService",
    "LLMAnalysisRequest",
    "LLMAnalysisResult",
    "BaseLLMProvider",
    "MockLLMProvider",
    "OpenAICompatibleProvider",
    "create_llm_service_from_settings",
]