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

"""通义千问(DashScope)视频理解 client (V4-M3 同契约).

与 GeminiVideoClient 同一 analyze 契约,但两点不同:
1. 由模型侧 fetch 视频 URL(优先 oss_url),不在本地下载/压缩/base64;
2. 走 DashScope OpenAI 兼容端点(默认中国节点 dashscope.aliyuncs.com)。
失败一律返回 V4 failed 结构,不抛、不卡 run。复用 gemini_video 的解析/失败/打分机器。
"""

from __future__ import annotations

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

import httpx

from content_agent.integrations import timeout_config
from content_agent.integrations.gemini_video import (
    DEFAULT_VIDEO_TIMEOUT_SECONDS,
    MissingGeminiVideoClient,
    _SYSTEM_PROMPT,
    _USER_PROMPT,
    _elapsed_ms,
    _fail,
    _http_status,
    _media_unavailable_update,
    _parse,
    _query_text,
    _response_body_summary,
    _retryable_http,
    _video_url_candidates,
    _with_media_update,
    _with_timing,
)

DEFAULT_DASHSCOPE_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
DEFAULT_QWEN_VIDEO_MODEL = "qwen3.6-plus"  # 通义千问 3.6 Plus(多模态,支持视频理解)
DEFAULT_QWEN_MAX_ATTEMPTS = 2


class QwenVideoClient:
    def __init__(
        self,
        *,
        api_key: str,
        model: str = DEFAULT_QWEN_VIDEO_MODEL,
        base_url: str = DEFAULT_DASHSCOPE_BASE_URL,
        timeout_seconds: float = DEFAULT_VIDEO_TIMEOUT_SECONDS,
        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.http_post = http_post

    @classmethod
    def from_env(cls, env: Mapping[str, str] | None = None) -> Any:
        source = os.environ if env is None else env
        api_key = (
            source.get("CONTENT_AGENT_VIDEO_LLM_API_KEY")
            or source.get("DASHSCOPE_API_KEY")
            or source.get("QWEN_API_KEY")
        )
        if not api_key:
            return MissingGeminiVideoClient("qwen video config missing: CONTENT_AGENT_VIDEO_LLM_API_KEY")
        return cls(
            api_key=api_key,
            model=source.get("CONTENT_AGENT_VIDEO_LLM_MODEL") or DEFAULT_QWEN_VIDEO_MODEL,
            base_url=source.get("CONTENT_AGENT_VIDEO_LLM_BASE_URL") or DEFAULT_DASHSCOPE_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]:
        result = self._no_fetchable_result(content, media, source_context)
        if result is not None:
            return result

        request_attempts: list[dict[str, Any]] = []
        started_at = time.monotonic()
        last_result: dict[str, Any] | None = None
        for attempt in range(1, 4):
            result = self.analyze_once(
                content,
                media,
                source_context,
                attempt_number=attempt,
                previous_attempts=request_attempts,
                request_started_at=started_at,
            )
            request_attempts = _qwen_attempts(result)
            last_result = result
            if result.get("final_status") != "failed":
                return result
            if not result.get("qwen_retryable") or attempt >= _max_attempts_for_failure(result):
                return result
        return last_result or _fail("qwen_unknown_error", query_text=_query_text(content, source_context))

    def analyze_once(
        self,
        content: dict[str, Any],
        media: dict[str, Any],
        source_context: dict[str, Any],
        *,
        attempt_number: int = 1,
        previous_attempts: list[dict[str, Any]] | None = None,
        request_started_at: float | None = None,
    ) -> dict[str, Any]:
        query_text = _query_text(content, source_context)
        fetchable = [(u, s) for (u, s) in _video_url_candidates(media) if s != "play_url"]
        if not fetchable:
            return self._no_fetchable_result(content, media, source_context) or _fail("no_oss_url", query_text=query_text)

        timing_metrics: dict[str, Any] = {}
        request_attempts = list(previous_attempts or [])
        started_at = request_started_at if request_started_at is not None else time.monotonic()
        video_url, video_source = fetchable[0]
        messages = [
            {"role": "system", "content": _SYSTEM_PROMPT},
            {
                "role": "user",
                "content": [
                    {"type": "video_url", "video_url": {"url": video_url}},
                    {"type": "text", "text": _USER_PROMPT.format(query_text=query_text)},
                ],
            },
        ]
        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()
            parsed = _parse(payload, query_text)
            request_attempts.append(
                {
                    "attempt": attempt_number,
                    "video_source": video_source,
                    "duration_ms": _elapsed_ms(attempt_started_at, time.monotonic()),
                    "http_status_code": getattr(response, "status_code", None),
                    "status": "ok",
                }
            )
            timing_metrics["qwen_request"] = {
                "total_duration_ms": _elapsed_ms(started_at, time.monotonic()),
                "attempts": request_attempts,
                "video_source": video_source,
            }
            return _with_timing(parsed, timing_metrics)
        except httpx.HTTPError as exc:
            failure_type = "qwen_client_timeout" if isinstance(exc, httpx.TimeoutException) else "qwen_http_error"
            response_summary = response_summary or _response_body_summary(getattr(exc, "response", None))
            if _is_content_inspection_failure(failure_type, response_summary):
                failure_type = "content_inspection_blocked"
            request_attempts.append(
                {
                    "attempt": attempt_number,
                    "video_source": video_source,
                    "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 {}),
                }
            )
            retryable = failure_type != "content_inspection_blocked" and _retryable_http(exc)
            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=attempt_number,
                response_body_summary=response_summary,
            )
            return _with_timing(
                {
                    **failure,
                    "qwen_retryable": retryable,
                    "qwen_retry_after_seconds": _retry_after_seconds(response_summary),
                },
                {
                    "qwen_request": {
                        "total_duration_ms": _elapsed_ms(started_at, time.monotonic()),
                        "attempts": request_attempts,
                        "video_source": video_source,
                    }
                },
            )
        except (KeyError, IndexError, TypeError, ValueError, json.JSONDecodeError) as exc:
            request_attempts.append(
                {
                    "attempt": attempt_number,
                    "video_source": video_source,
                    "duration_ms": _elapsed_ms(attempt_started_at, time.monotonic()),
                    "status": "failed",
                    "failure_type": "qwen_response_invalid",
                    "exception_type": type(exc).__name__,
                    **({"response_body_summary": response_summary} if response_summary else {}),
                }
            )
            failure = _fail(
                "qwen_response_invalid",
                query_text=query_text,
                exception_type=type(exc).__name__,
                error_message=str(exc),
                retry_count=attempt_number,
                response_body_summary=response_summary,
            )
            return _with_timing(
                {**failure, "qwen_retryable": True},
                {
                    "qwen_request": {
                        "total_duration_ms": _elapsed_ms(started_at, time.monotonic()),
                        "attempts": request_attempts,
                        "video_source": video_source,
                    }
                },
            )

    def _no_fetchable_result(
        self,
        content: dict[str, Any],
        media: dict[str, Any],
        source_context: dict[str, Any],
    ) -> dict[str, Any] | None:
        query_text = _query_text(content, source_context)
        fetchable = [(u, s) for (u, s) in _video_url_candidates(media) if s != "play_url"]
        if fetchable:
            return None
        media_raw = media.get("raw_payload") if isinstance(media.get("raw_payload"), dict) else {}
        reason = media.get("failure_reason") or media_raw.get("failure_reason") or "no_oss_url"
        return _with_media_update(
            _with_timing(
                _fail(str(reason), query_text=query_text),
                {"video_fetch": {"skipped": str(reason)}},
            ),
            _media_unavailable_update(str(reason)),
        )


def _qwen_attempts(result: dict[str, Any]) -> list[dict[str, Any]]:
    timing = result.get("timing_metrics") if isinstance(result.get("timing_metrics"), dict) else {}
    request = timing.get("qwen_request") if isinstance(timing.get("qwen_request"), dict) else {}
    attempts = request.get("attempts")
    return list(attempts) if isinstance(attempts, list) else []


def _is_content_inspection_failure(failure_type: str, response_summary: dict[str, Any] | None) -> bool:
    if failure_type != "qwen_http_error":
        return False
    excerpt = str((response_summary or {}).get("text_excerpt", "")).lower()
    return "datainspection" in excerpt or "inappropriate content" in excerpt


def _retry_after_seconds(response_summary: dict[str, Any] | None) -> float | None:
    raw = (response_summary or {}).get("retry_after")
    if raw is None:
        return None
    try:
        return max(0.0, float(raw))
    except (TypeError, ValueError):
        return None


def _max_attempts_for_failure(result: dict[str, Any]) -> int:
    if result.get("failure_type") == "qwen_http_error" and result.get("http_status_code") == 429:
        return 3
    return DEFAULT_QWEN_MAX_ATTEMPTS