content-find-agent-new/content_agent/integrations/gemini_video.py

"""Gemini 视频相关性 client (V4-M3).

实现 interfaces.GeminiVideoClient.analyze:取视频(video_fetch)→ 多模态投给
Gemini(OpenRouter image_url data URL)→ 解析 V4 query relevance-only 结构。
任何失败一律返回 V4 failed 结构,不抛、不卡 run。
"""

from __future__ import annotations

import json
import os
import re
import time
from typing import Any, Callable, Mapping

import httpx

from content_agent.integrations import timeout_config, video_fetch

DEFAULT_OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
DEFAULT_VIDEO_MODEL = "google/gemini-3-flash-preview"
DEFAULT_VIDEO_TIMEOUT_SECONDS = timeout_config.total_timeout("video_llm")  # 10min(原 30min)
V4_GEMINI_QUERY_RELEVANCE_SCHEMA_VERSION = "v4_gemini_query_relevance.v1"

_SYSTEM_PROMPT = "你是视频内容与搜索需求相关性审核助手。只输出一个 JSON 对象,不要任何解释或 markdown。"
_USER_PROMPT = (
    "判断这条视频,严格按以下 JSON 结构输出(只输出 JSON):\n"
    '{{"query_relevance_score": 0~100 的整数或小数, "query_relevance_reason": "中文简述理由"}}\n'
    "字段含义:\n"
    "- query_relevance_score: 视频内容与搜索需求【{query_text}】的相关程度,0 表示完全无关,100 表示高度相关。\n"
    "- query_relevance_reason: 一句话说明相关或不相关的关键原因。"
)


def _fail(
    failure_type: str,
    *,
    query_text: str = "(未指定)",
    exception_type: str | None = None,
    error_message: str | None = None,
    http_status_code: int | None = None,
    retry_count: int = 1,
    response_body_summary: dict[str, Any] | None = None,
) -> dict[str, Any]:
    result = {
        "schema_version": V4_GEMINI_QUERY_RELEVANCE_SCHEMA_VERSION,
        "query_text": query_text,
        "query_relevance_score": None,
        "query_relevance_reason": "",
        "final_status": "failed",
        "failure_type": failure_type,
        "exception_type": exception_type,
        "http_status_code": http_status_code,
        "retry_count": retry_count,
    }
    if error_message:
        result["error_message"] = error_message
    if response_body_summary:
        result["response_body_summary"] = response_body_summary
    return result


def _video_url_candidates(media: dict[str, Any]) -> list[tuple[str, str]]:
    raw_payload = media.get("raw_payload") if isinstance(media.get("raw_payload"), dict) else {}
    candidates: list[tuple[str, str]] = []
    seen: set[str] = set()
    for source, value in [
        ("oss_url", media.get("oss_url")),
        ("cdn_url", media.get("cdn_url")),
        ("raw_payload.cdn_url", raw_payload.get("cdn_url")),
        ("play_url", media.get("play_url")),
    ]:
        if value:
            url = str(value)
            if url not in seen:
                seen.add(url)
                candidates.append((url, source))
    return candidates


def _with_media_update(result: dict[str, Any], update: dict[str, Any] | None) -> dict[str, Any]:
    if update:
        return {**result, "media_storage_update": update}
    return result


def _with_timing(result: dict[str, Any], timing: dict[str, Any]) -> dict[str, Any]:
    return {**result, "timing_metrics": timing}


def _clamp_score(value: Any) -> float:
    try:
        number = float(value)
    except (TypeError, ValueError):
        return 0.0
    return max(0.0, min(100.0, number))


def _parse(payload: dict[str, Any], query_text: str) -> dict[str, Any]:
    content = payload["choices"][0]["message"]["content"]
    text = str(content).strip()
    if text.startswith("```"):
        text = text.split("```", 2)[1]
        if text.startswith("json"):
            text = text[4:]
    data = json.loads(text)
    return {
        "schema_version": V4_GEMINI_QUERY_RELEVANCE_SCHEMA_VERSION,
        "query_text": query_text,
        "query_relevance_score": _clamp_score(data["query_relevance_score"]),
        "query_relevance_reason": str(data.get("query_relevance_reason") or ""),
        "final_status": "ok",
        "retry_count": 0,
    }


def _query_text(content: dict[str, Any], source_context: dict[str, Any]) -> str:
    matched = content.get("matched_search_queries")
    if isinstance(matched, list) and matched:
        return "、".join(str(item) for item in matched if item)
    query_sources = content.get("query_sources")
    if isinstance(query_sources, list):
        queries = [
            str(item.get("search_query"))
            for item in query_sources
            if isinstance(item, dict) and item.get("search_query")
        ]
        if queries:
            return "、".join(queries)
    if content.get("search_query"):
        return str(content["search_query"])
    evidence = source_context.get("ext_data", {}).get("evidence_pack", {})
    terms = evidence.get("seed_terms") or []
    return "、".join(str(t) for t in terms) or "(未指定)"


def _http_status(exc: httpx.HTTPError) -> int | None:
    response = getattr(exc, "response", None)
    status_code = getattr(response, "status_code", None)
    return int(status_code) if isinstance(status_code, int) else None


def _response_body_summary(response: Any | None) -> dict[str, Any] | None:
    if response is None:
        return None
    summary: dict[str, Any] = {
        "http_status_code": getattr(response, "status_code", None),
    }
    headers = getattr(response, "headers", None)
    if headers is not None:
        summary["content_type"] = headers.get("content-type", "")
        retry_after = headers.get("retry-after")
        if retry_after:
            summary["retry_after"] = retry_after
    try:
        content = response.content or b""
    except Exception:
        content = b""
    if content:
        summary["body_bytes"] = len(content)
    try:
        payload = response.json()
    except Exception:
        payload = None
    if isinstance(payload, dict):
        summary["json_top_level_keys"] = sorted(str(key) for key in payload.keys())[:20]
        text_source = payload.get("error") or payload.get("message") or payload.get("msg")
        if isinstance(text_source, dict):
            text_source = text_source.get("message") or text_source.get("msg") or text_source.get("code")
        if text_source is not None:
            summary["text_excerpt"] = _scrub_response_text(str(text_source))
    elif content:
        summary["text_excerpt"] = _scrub_response_text(content.decode("utf-8", errors="replace"))
    return {key: value for key, value in summary.items() if value not in (None, "", [], {})}


def _scrub_response_text(text: str, limit: int = 500) -> str:
    text = " ".join(text.split())
    text = re.sub(r"Bearer\s+[A-Za-z0-9._~+/=-]+", "Bearer [REDACTED]", text, flags=re.IGNORECASE)
    text = re.sub(r"sk-[A-Za-z0-9_-]{16,}", "sk-[REDACTED]", text)
    text = re.sub(r"data:[^,\s]+;base64,[A-Za-z0-9+/=]{128,}", "data:[REDACTED]", text)
    text = re.sub(r"(https?://[^\s?]+)\?[^\s]+", r"\1?[REDACTED]", text)
    return text[:limit] + ("..." if len(text) > limit else "")


def _retryable_http(exc: httpx.HTTPError) -> bool:
    if isinstance(exc, (httpx.ConnectError, httpx.TimeoutException, httpx.NetworkError)):
        return True
    status = _http_status(exc)
    return status in {408, 429} or (status is not None and 500 <= status <= 599)


class GeminiVideoClient:
    def __init__(
        self,
        *,
        api_key: str,
        model: str = DEFAULT_VIDEO_MODEL,
        base_url: str = DEFAULT_OPENROUTER_BASE_URL,
        timeout_seconds: float = DEFAULT_VIDEO_TIMEOUT_SECONDS,
        fetch_fn: Callable[..., str] = video_fetch.fetch_and_compress,
        http_post: Callable[..., Any] = httpx.post,
    ) -> None:
        self.api_key = api_key
        self.model = model
        self.base_url = base_url.rstrip("/")
        self.timeout_seconds = timeout_seconds
        self.fetch_fn = fetch_fn
        self.http_post = http_post

    @classmethod
    def from_env(cls, env: Mapping[str, str] | None = None) -> "GeminiVideoClient":
        source = os.environ if env is None else env
        api_key = source.get("OPENROUTER_API_KEY") or source.get("OPEN_ROUTER_API_KEY")
        if not api_key:
            return MissingGeminiVideoClient("gemini video config missing: OPENROUTER_API_KEY")
        return cls(
            api_key=api_key,
            model=source.get("CONTENT_AGENT_VIDEO_LLM_MODEL") or DEFAULT_VIDEO_MODEL,
            base_url=source.get("OPENROUTER_BASE_URL") or DEFAULT_OPENROUTER_BASE_URL,
            timeout_seconds=float(source.get("CONTENT_AGENT_VIDEO_LLM_TIMEOUT_SECONDS") or DEFAULT_VIDEO_TIMEOUT_SECONDS),
        )

    def analyze(
        self,
        content: dict[str, Any],
        media: dict[str, Any],
        source_context: dict[str, Any],
    ) -> dict[str, Any]:
        query_text = _query_text(content, source_context)
        video_candidates = _video_url_candidates(media)
        if not video_candidates:
            media_raw = media.get("raw_payload") if isinstance(media.get("raw_payload"), dict) else {}
            failure_reason = (
                media.get("failure_reason")
                or media_raw.get("failure_reason")
                or "no_play_url"
            )
            return _with_media_update(
                _with_timing(
                    _fail(str(failure_reason), query_text=query_text),
                    {"video_fetch": {"skipped": str(failure_reason)}},
                ),
                _media_unavailable_update(str(failure_reason)),
            )
        timing_metrics: dict[str, Any] = {}
        data_url = ""
        fetch_attempts: list[dict[str, Any]] = []
        last_fetch_exc: Exception | None = None
        for video_url, video_source in video_candidates:
            try:
                data_url, fetch_metrics = self._fetch_video(video_url, content.get("platform", "douyin"))
                timing_metrics["video_fetch"] = {
                    **fetch_metrics,
                    "gemini_video_source": video_source,
                    "attempts": fetch_attempts,
                }
                break
            except Exception as exc:
                last_fetch_exc = exc
                fetch_metrics = getattr(exc, "metrics", {}) or {}
                fetch_attempts.append(
                    {
                        **fetch_metrics,
                        "gemini_video_source": video_source,
                        "status": "failed",
                        "exception_type": type(exc).__name__,
                        "error_message": str(exc),
                    }
                )
        else:
            assert last_fetch_exc is not None
            fetch_metrics = getattr(last_fetch_exc, "metrics", {}) or {}
            timing_metrics["video_fetch"] = {
                **fetch_metrics,
                "gemini_video_source": video_candidates[-1][1],
                "attempts": fetch_attempts,
            }
            return _with_timing(
                _fail(
                    "video_fetch_failed",
                    query_text=query_text,
                    exception_type=type(last_fetch_exc).__name__,
                    error_message=str(last_fetch_exc),
                ),
                timing_metrics,
            )

        messages = [
            {"role": "system", "content": _SYSTEM_PROMPT},
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": _USER_PROMPT.format(query_text=query_text)},
                    {"type": "image_url", "image_url": {"url": data_url}},
                ],
            },
        ]
        last_failure: dict[str, Any] | None = None
        request_attempts: list[dict[str, Any]] = []
        request_started_at = time.monotonic()
        for attempt in range(2):
            retry_count = attempt + 1
            attempt_started_at = time.monotonic()
            response_summary: dict[str, Any] | None = None
            try:
                response = self.http_post(
                    f"{self.base_url}/chat/completions",
                    headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"},
                    json={"model": self.model, "messages": messages},
                    timeout=timeout_config.as_httpx_timeout(
                        self.timeout_seconds, read=timeout_config.read_timeout("video_llm")
                    ),
                )
                response_summary = _response_body_summary(response)
                response.raise_for_status()
                payload = response.json()
                request_attempts.append(
                    {
                        "attempt": retry_count,
                        "duration_ms": _elapsed_ms(attempt_started_at, time.monotonic()),
                        "http_status_code": getattr(response, "status_code", None),
                        "status": "ok",
                    }
                )
                timing_metrics["gemini_request"] = {
                    "total_duration_ms": _elapsed_ms(request_started_at, time.monotonic()),
                    "attempts": request_attempts,
                }
                return _with_timing(_parse(payload, query_text), timing_metrics)
            except httpx.HTTPError as exc:
                failure_type = "gemini_client_timeout" if isinstance(exc, httpx.TimeoutException) else "gemini_http_error"
                response_summary = response_summary or _response_body_summary(getattr(exc, "response", None))
                request_attempts.append(
                    {
                        "attempt": retry_count,
                        "duration_ms": _elapsed_ms(attempt_started_at, time.monotonic()),
                        "http_status_code": _http_status(exc),
                        "status": "failed",
                        "failure_type": failure_type,
                        "exception_type": type(exc).__name__,
                        **({"response_body_summary": response_summary} if response_summary else {}),
                    }
                )
                timing_metrics["gemini_request"] = {
                    "total_duration_ms": _elapsed_ms(request_started_at, time.monotonic()),
                    "attempts": request_attempts,
                }
                last_failure = _fail(
                    failure_type,
                    query_text=query_text,
                    exception_type=type(exc).__name__,
                    error_message=str(exc),
                    http_status_code=_http_status(exc),
                    retry_count=retry_count,
                    response_body_summary=response_summary,
                )
                if attempt == 0 and _retryable_http(exc):
                    continue
                return _with_timing(last_failure, timing_metrics)
            except (KeyError, IndexError, TypeError, ValueError, json.JSONDecodeError) as exc:
                request_attempts.append(
                    {
                        "attempt": retry_count,
                        "duration_ms": _elapsed_ms(attempt_started_at, time.monotonic()),
                        "status": "failed",
                        "failure_type": "gemini_response_invalid",
                        "exception_type": type(exc).__name__,
                        **({"response_body_summary": response_summary} if response_summary else {}),
                    }
                )
                timing_metrics["gemini_request"] = {
                    "total_duration_ms": _elapsed_ms(request_started_at, time.monotonic()),
                    "attempts": request_attempts,
                }
                last_failure = _fail(
                    "gemini_response_invalid",
                    query_text=query_text,
                    exception_type=type(exc).__name__,
                    error_message=str(exc),
                    retry_count=retry_count,
                    response_body_summary=response_summary,
                )
                if attempt == 0:
                    continue
                return _with_timing(last_failure, timing_metrics)
        return _with_timing(last_failure or _fail("gemini_unknown_error", query_text=query_text), timing_metrics)

    def _fetch_video(self, play_url: str, platform: str) -> tuple[str, dict[str, Any]]:
        if self.fetch_fn is video_fetch.fetch_and_compress:
            return video_fetch.fetch_and_compress_with_metrics(play_url, platform)
        started_at = time.monotonic()
        result = self.fetch_fn(play_url, platform)
        if isinstance(result, tuple) and len(result) == 2 and isinstance(result[1], dict):
            return result[0], result[1]
        return result, {"total_duration_ms": _elapsed_ms(started_at, time.monotonic())}


class MissingGeminiVideoClient:
    def __init__(self, reason: str) -> None:
        self.reason = reason

    def analyze(
        self,
        content: dict[str, Any],
        media: dict[str, Any],
        source_context: dict[str, Any],
    ) -> dict[str, Any]:
        return _fail(
            "gemini_config_missing",
            query_text=_query_text(content, source_context),
            exception_type=self.reason,
            error_message=self.reason,
        )


def _media_unavailable_update(failure_reason: str) -> dict[str, Any]:
    return {
        "content_media_status": "unavailable",
        "oss_url": None,
        "local_path": None,
        "failure_reason": failure_reason,
        "raw_payload": {"failure_reason": failure_reason},
    }


def _elapsed_ms(start: float, end: float) -> int:
    return max(0, int((end - start) * 1000))