feat(video): 视频理解支持通义千问(DashScope)+ provider 开关

新增 QwenVideoClient:走 DashScope OpenAI 兼容端点,模型侧直接 fetch 视频 URL(优先 oss_url,
不再本地下载/压缩/base64);复用 gemini_video 的解析/失败/打分机器,同 analyze 契约。
run_service 按 CONTENT_AGENT_VIDEO_LLM_PROVIDER 分发(dashscope→Qwen / 否则 OpenRouter-Gemini)。
海外保持 OpenRouter 不变;国内设 provider=dashscope + qwen3-vl-plus,贴近视频源(抖音中国)+模型(北京)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

content_agent/integrations/qwen_video.py

@@ -0,0 +1,191 @@
+"""通义千问(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.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-vl-plus"
+
+
+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]:
+        query_text = _query_text(content, source_context)
+        # 千问由模型侧 fetch URL,只用可公网访问的 oss/cdn;play_url(抖音签名/过期)模型取不到,排除。
+        fetchable = [(u, s) for (u, s) in _video_url_candidates(media) if s != "play_url"]
+        if not fetchable:
+            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)),
+            )
+
+        timing_metrics: dict[str, Any] = {}
+        request_attempts: list[dict[str, Any]] = []
+        last_failure: dict[str, Any] | None = None
+        started_at = time.monotonic()
+        for video_url, video_source in fetchable:
+            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)},
+                    ],
+                },
+            ]
+            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=self.timeout_seconds,
+                    )
+                    response_summary = _response_body_summary(response)
+                    response.raise_for_status()
+                    payload = response.json()
+                    request_attempts.append(
+                        {
+                            "attempt": retry_count,
+                            "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(_parse(payload, query_text), 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))
+                    request_attempts.append(
+                        {
+                            "attempt": retry_count,
+                            "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 {}),
+                        }
+                    )
+                    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
+                    break  # 不可重试 → 换下一个候选 URL
+                except (KeyError, IndexError, TypeError, ValueError, json.JSONDecodeError) as exc:
+                    request_attempts.append(
+                        {
+                            "attempt": retry_count,
+                            "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 {}),
+                        }
+                    )
+                    last_failure = _fail(
+                        "qwen_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
+                    break
+        timing_metrics["qwen_request"] = {
+            "total_duration_ms": _elapsed_ms(started_at, time.monotonic()),
+            "attempts": request_attempts,
+        }
+        return _with_timing(last_failure or _fail("qwen_unknown_error", query_text=query_text), timing_metrics)

content_agent/run_service.py

@@ -18,6 +18,7 @@ from content_agent.integrations.database_runtime import ContentSupplyDbConfig, D
 from content_agent.integrations.demand_source import DemandSourceService
 from content_agent.integrations.douyin import CrawapiDouyinClient
 from content_agent.integrations.gemini_video import GeminiVideoClient as RealGeminiVideoClient
+from content_agent.integrations.qwen_video import QwenVideoClient
 from content_agent.integrations.kuaishou import CrawapiKuaishouClient
 from content_agent.integrations.mock_platform import MockPlatformClient
 from content_agent.integrations import oss_archive
@@ -728,6 +729,10 @@ def _query_failures_from_error(error: ContentAgentError) -> list[dict[str, Any]]
 
 
 def _gemini_video_client_from_env(env: dict[str, str]) -> GeminiVideoClient:
+    # 视频理解 provider 开关:dashscope=通义千问(国内,video_url 直取 oss);否则 OpenRouter/Gemini(海外)。
+    provider = (env.get("CONTENT_AGENT_VIDEO_LLM_PROVIDER") or "openrouter").strip().lower()
+    if provider in {"dashscope", "qwen", "bailian", "tongyi"}:
+        return QwenVideoClient.from_env(env)
     return RealGeminiVideoClient.from_env(env)
 
 

tests/test_qwen_video.py

@@ -0,0 +1,59 @@
+"""通义千问视频 client:provider 开关 + video_url 取 oss + 解析。"""
+
+import json
+
+from content_agent.integrations.qwen_video import QwenVideoClient
+from content_agent.run_service import _gemini_video_client_from_env
+from content_agent.integrations.gemini_video import GeminiVideoClient
+
+
+class _FakeResponse:
+    def __init__(self, payload):
+        self._payload = payload
+        self.status_code = 200
+        self.headers = {"content-type": "application/json"}
+        self.content = json.dumps(payload).encode()
+
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return self._payload
+
+
+def test_provider_switch_picks_qwen_vs_openrouter():
+    qwen = _gemini_video_client_from_env({"CONTENT_AGENT_VIDEO_LLM_PROVIDER": "dashscope", "CONTENT_AGENT_VIDEO_LLM_API_KEY": "k"})
+    assert isinstance(qwen, QwenVideoClient)
+    openrouter = _gemini_video_client_from_env({"OPENROUTER_API_KEY": "k"})
+    assert isinstance(openrouter, GeminiVideoClient)
+
+
+def test_qwen_analyze_sends_video_url_oss_and_parses():
+    captured = {}
+
+    def fake_post(url, headers=None, json=None, timeout=None):
+        captured["url"] = url
+        captured["json"] = json
+        return _FakeResponse(
+            {"choices": [{"message": {"content": '{"query_relevance_score": 88, "query_relevance_reason": "贴合"}'}}]}
+        )
+
+    client = QwenVideoClient(api_key="k", model="qwen3-vl-plus", http_post=fake_post)
+    out = client.analyze(
+        {"platform": "douyin", "search_query": "拉伸"},
+        {"oss_url": "https://res.example.com/v.mp4", "play_url": "https://douyinvod.com/x"},
+        {},
+    )
+    assert out["final_status"] == "ok"
+    assert out["query_relevance_score"] == 88
+    # 发的是 dashscope chat/completions,且用 video_url 取 oss(不是 play_url、不是 base64)
+    assert captured["url"].endswith("/chat/completions")
+    content = captured["json"]["messages"][1]["content"]
+    video_part = next(p for p in content if p["type"] == "video_url")
+    assert video_part["video_url"]["url"] == "https://res.example.com/v.mp4"
+
+
+def test_qwen_missing_oss_returns_failed():
+    client = QwenVideoClient(api_key="k", http_post=lambda *a, **k: None)
+    out = client.analyze({"platform": "douyin"}, {"play_url": "https://douyinvod.com/x"}, {})
+    assert out["final_status"] == "failed"  # 只有 play_url(模型取不到)→ 视作无源