fix: add platform video url fallback

content_agent/business_modules/content_discovery/content_discovery_builder.py

@@ -122,6 +122,8 @@ def _build_content_media_record(
     if not record["play_url"]:
         record["failure_reason"] = result.get("media_failure_reason", "no_play_url_returned")
     record = with_raw_payload(record)
+    if result.get("video_url_candidates"):
+        record["raw_payload"]["video_url_candidates"] = result["video_url_candidates"]
     platform_raw_payload = result.get("platform_raw_payload") or {}
     if isinstance(platform_raw_payload, dict):
         for field in [
@@ -137,6 +139,10 @@ def _build_content_media_record(
             "kuaishou_detail_fallback_status",
             "kuaishou_detail_fallback_exception_type",
             "kuaishou_detail_fallback_error",
+            "douyin_detail_fallback_attempted",
+            "douyin_detail_fallback_status",
+            "douyin_detail_fallback_exception_type",
+            "douyin_detail_fallback_error",
             "shipinhao_research_source",
             "shipinhao_research_status",
             "shipinhao_research_exception_type",

content_agent/integrations/douyin.py

@@ -19,6 +19,7 @@ from content_agent.integrations.crawapi_http import (
     score_from_statistics as _score_from_statistics,
     search_previous_discovery_step as _search_previous_discovery_step,
 )
+from content_agent.integrations import platform_video_url
 
 RAW_CONTENT_ID_KEY = "_".join(["aweme", "id"])
 RAW_AUTHOR_ID_KEY = "_".join(["sec", "uid"])
@@ -54,6 +55,7 @@ class CrawapiDouyinClient:
         max_results_per_query: int | None = 5,
         http_client: Any | None = None,
         rate_limiter: RateLimiter | None = None,
+        video_url_probe_fn: platform_video_url.ProbeFn | None = None,
     ) -> None:
         self.base_url = base_url.rstrip("/") + "/"
         self.keyword_path = keyword_path.lstrip("/")
@@ -70,6 +72,7 @@ class CrawapiDouyinClient:
         self.max_results_per_query = max_results_per_query
         self.http_client = http_client or httpx.Client(timeout=timeout_seconds)
         self.rate_limiter = rate_limiter
+        self.video_url_probe_fn = video_url_probe_fn
 
     @classmethod
     def from_env(cls, env_path: str | Path = ".env") -> "CrawapiDouyinClient":
@@ -137,7 +140,10 @@ class CrawapiDouyinClient:
         results: list[dict[str, Any]] = []
         selected_items = items[:max_results_per_query] if max_results_per_query else items
         for index, item in enumerate(selected_items, start=1):
-            results.append(self._normalize_content_item(query, item, index, has_more, next_cursor))
+            selection = self._select_video_url_with_detail_fallback(item)
+            results.append(
+                self._normalize_content_item(query, item, index, has_more, next_cursor, selection)
+            )
         return results
 
     def fetch_author_works(self, query: dict[str, Any]) -> list[dict[str, Any]]:
@@ -158,7 +164,8 @@ class CrawapiDouyinClient:
         selected_items = items[: self.max_results_per_query] if self.max_results_per_query else items
         results: list[dict[str, Any]] = []
         for index, item in enumerate(selected_items, start=1):
-            normalized = self._normalize_content_item(query, item, index, has_more, next_cursor)
+            selection = self._select_video_url_with_detail_fallback(item)
+            normalized = self._normalize_content_item(query, item, index, has_more, next_cursor, selection)
             normalized["previous_discovery_step"] = "author_works"
             normalized["content_metadata_source"] = "douyin_blogger"
             results.append(normalized)
@@ -189,18 +196,20 @@ class CrawapiDouyinClient:
         index: int,
         has_more: bool,
         next_cursor: str,
+        video_url_selection: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
+        video_url_selection = video_url_selection or self._select_video_url([("search", item)])
         author = item.get("author", {}) if isinstance(item.get("author"), dict) else {}
         statistics = item.get("statistics", {}) if isinstance(item.get("statistics"), dict) else {}
         platform_content_id = str(item.get(RAW_CONTENT_ID_KEY) or "")
         platform_author_id = str(author.get(RAW_AUTHOR_ID_KEY) or "")
-        return {
+        result = {
             "content_discovery_id": f"{query['search_query_id']}_content_{index:03d}",
             "search_query_id": query["search_query_id"],
             "platform": "douyin",
             "platform_content_id": platform_content_id,
             "platform_content_format": _content_format(self.default_content_type),
-            "play_url": _extract_play_url(item),
+            "play_url": video_url_selection.get("play_url"),
             "description": item.get("desc") or item.get("item_title") or "",
             "platform_author_id": platform_author_id,
             "author_display_name": author.get("nickname") or "",
@@ -226,18 +235,17 @@ class CrawapiDouyinClient:
             "platform_raw_payload": {
                 RAW_CONTENT_ID_KEY: platform_content_id,
                 "author": {RAW_AUTHOR_ID_KEY: platform_author_id},
+                **dict(video_url_selection.get("platform_raw_payload") or {}),
             },
         }
+        if video_url_selection.get("media_failure_reason"):
+            result["media_failure_reason"] = video_url_selection["media_failure_reason"]
+        if video_url_selection.get("video_url_candidates"):
+            result["video_url_candidates"] = video_url_selection["video_url_candidates"]
+        return result
 
     def fetch_detail(self, content_id: str) -> dict[str, Any]:
-        data = self._post_json(
-            self.detail_path,
-            {"content_id": str(content_id)},
-            operation="detail",
-            rate_limit_bucket=SEARCH_RATE_LIMIT_BUCKET,
-        )
-        block = data.get("data", {}) if isinstance(data.get("data"), dict) else {}
-        detail = block.get("data", {}) if isinstance(block.get("data"), dict) else {}
+        detail = self._fetch_detail_item(content_id)
         statistics = {
             "digg_count": int(detail.get("like_count") or 0),
             "comment_count": int(detail.get("comment_count") or 0),
@@ -247,10 +255,9 @@ class CrawapiDouyinClient:
         }
         topic_list = detail.get("topic_list") or []
         tags = [t if str(t).startswith("#") else f"#{t}" for t in topic_list if t]
-        video_list = detail.get("video_url_list") or []
-        play_url = video_list[0].get("video_url") if video_list else None
+        selection = self._select_video_url([("detail", detail)])
         publish_ms = detail.get("publish_timestamp")
-        return {
+        result = {
             "platform": "douyin",
             "platform_content_id": str(detail.get("channel_content_id") or content_id),
             "platform_content_url": detail.get("content_link"),
@@ -259,10 +266,79 @@ class CrawapiDouyinClient:
             "author_display_name": detail.get("channel_account_name") or "",
             "statistics": statistics,
             "tags": tags,
-            "play_url": play_url,
+            "play_url": selection.get("play_url"),
             "create_time": int(publish_ms) // 1000 if publish_ms else None,
             "content_metadata_source": "douyin_detail",
+            "platform_raw_payload": dict(selection.get("platform_raw_payload") or {}),
         }
+        if selection.get("media_failure_reason"):
+            result["media_failure_reason"] = selection["media_failure_reason"]
+        if selection.get("video_url_candidates"):
+            result["video_url_candidates"] = selection["video_url_candidates"]
+        return result
+
+    def _select_video_url(self, sources: list[tuple[str, dict[str, Any]]]) -> dict[str, Any]:
+        return platform_video_url.select_video_url(
+            "douyin",
+            sources,
+            probe_fn=self.video_url_probe_fn or self._probe_video_url,
+        )
+
+    def _select_video_url_with_detail_fallback(self, item: dict[str, Any]) -> dict[str, Any]:
+        search_selection = self._select_video_url([("search", item)])
+        content_id = str(item.get(RAW_CONTENT_ID_KEY) or "")
+        if not self._should_fetch_detail_for_video_url(search_selection) or not self.detail_path or not content_id:
+            return search_selection
+        try:
+            detail = self._fetch_detail_item(content_id)
+        except Exception as exc:  # noqa: BLE001 - keep search diagnostics if detail is unavailable.
+            payload = dict(search_selection.get("platform_raw_payload") or {})
+            payload.update(
+                {
+                    "douyin_detail_fallback_attempted": True,
+                    "douyin_detail_fallback_status": "failed",
+                    "douyin_detail_fallback_exception_type": type(exc).__name__,
+                    "douyin_detail_fallback_error": str(exc)[:300],
+                }
+            )
+            return {**search_selection, "platform_raw_payload": payload}
+        detail_selection = self._select_video_url([("search", item), ("detail", detail)])
+        payload = dict(detail_selection.get("platform_raw_payload") or {})
+        payload.update(
+            {
+                "douyin_detail_fallback_attempted": True,
+                "douyin_detail_fallback_status": (
+                    "used" if detail_selection.get("play_url") else "no_valid_play_url"
+                ),
+            }
+        )
+        return {**detail_selection, "platform_raw_payload": payload}
+
+    def _should_fetch_detail_for_video_url(self, selection: dict[str, Any]) -> bool:
+        payload = selection.get("platform_raw_payload") if isinstance(selection.get("platform_raw_payload"), dict) else {}
+        if not selection.get("play_url"):
+            return True
+        host = str(payload.get("selected_video_url_host") or "").lower()
+        if host in {"v11-weba.douyinvod.com", "v96-hcc.douyinvod.com"}:
+            return True
+        return str(payload.get("selected_video_url_probe_status") or "") in {"failed", "failed_fallback"}
+
+    def _fetch_detail_item(self, content_id: str) -> dict[str, Any]:
+        data = self._post_json(
+            self.detail_path,
+            {"content_id": str(content_id)},
+            operation="detail",
+            rate_limit_bucket=SEARCH_RATE_LIMIT_BUCKET,
+        )
+        block = data.get("data", {}) if isinstance(data.get("data"), dict) else {}
+        return block.get("data", {}) if isinstance(block.get("data"), dict) else {}
+
+    def _probe_video_url(self, url: str, platform: str) -> dict[str, Any]:
+        return platform_video_url.probe_url_with_httpx(
+            url,
+            platform,
+            http_client=self.http_client,
+        )
 
     def _post_json(
         self,

content_agent/integrations/kuaishou.py

@@ -115,6 +115,8 @@ def _normalize_kuaishou_item(
     }
     if video_url_selection.get("media_failure_reason"):
         result["media_failure_reason"] = video_url_selection["media_failure_reason"]
+    if video_url_selection.get("video_url_candidates"):
+        result["video_url_candidates"] = video_url_selection["video_url_candidates"]
     return result
 
 
@@ -215,6 +217,8 @@ class CrawapiKuaishouClient:
             "play_url": hydrated.get("play_url"),
             "platform_raw_payload": hydrated.get("platform_raw_payload", {}),
         }
+        if hydrated.get("video_url_candidates"):
+            merged["video_url_candidates"] = hydrated["video_url_candidates"]
         if selection.get("media_failure_reason"):
             merged["media_failure_reason"] = selection["media_failure_reason"]
         else:
@@ -283,6 +287,8 @@ class CrawapiKuaishouClient:
         }
         if selection.get("media_failure_reason"):
             result["media_failure_reason"] = selection["media_failure_reason"]
+        if selection.get("video_url_candidates"):
+            result["video_url_candidates"] = selection["video_url_candidates"]
         return result
 
     def _select_video_url(self, sources: list[tuple[str, dict[str, Any]]]) -> dict[str, Any]:

content_agent/integrations/oss_archive.py

@@ -5,6 +5,7 @@ import time
 from datetime import datetime, timedelta, timezone
 from threading import Lock
 from typing import Any, Callable
+from urllib.parse import urlparse
 
 from content_agent.integrations import oss_upload, timeout_config
 from content_agent.integrations.bounded_pool import DaemonThreadPoolExecutor, run_bounded
@@ -18,6 +19,8 @@ OSS_WORKER_RESULT_TIMEOUT_SECONDS = timeout_config.total_timeout("oss") + 60.0
 DEFAULT_OSS_ARCHIVE_WINDOW_SECONDS = 24 * 60 * 60.0
 DEFAULT_OSS_RETRY_DELAY_SECONDS = 15 * 60.0
 DEFAULT_OSS_ARCHIVE_MAX_WORKERS = 5
+DEFAULT_OSS_URL_FALLBACK_MAX_ATTEMPTS = 3
+DEFAULT_OSS_MEDIA_UPLOAD_BUDGET_SECONDS = 360.0
 
 
 class AsyncArchiveDispatcher:
@@ -271,21 +274,62 @@ def _archive_one(
     retry_delay_seconds: float,
 ) -> dict[str, Any]:
     raw_payload = dict(record.get("raw_payload") or {})
-    attempt_count = int(raw_payload.get("oss_archive_attempt_count") or 0) + 1
+    previous_attempt_count = int(raw_payload.get("oss_archive_attempt_count") or 0)
     deadline = _parse_time(raw_payload.get("oss_archive_deadline_at"))
     if deadline and now >= deadline:
-        return _with_failed_archive(record, raw_payload, now, attempt_count, "oss_upload_failed")
-    attempt_started_at = time.monotonic()
-    upload_result = upload_fn(
-        record.get("play_url") or "",
-        timeout_seconds=attempt_timeout_seconds,
-    )
+        return _with_failed_archive(record, raw_payload, now, previous_attempt_count + 1, "oss_upload_failed")
+    media_started_at = time.monotonic()
+    candidates = _upload_candidates(record, raw_payload)
+    max_attempts = _resolve_url_fallback_max_attempts()
+    media_budget_seconds = _resolve_media_upload_budget_seconds()
+    upload_result: dict[str, Any] = {"status": "failed", "failure_type": "missing_src_url"}
+    selected_candidate: dict[str, Any] | None = None
+    oss_url_attempts: list[dict[str, Any]] = []
+    for candidate in candidates[:max_attempts]:
+        remaining_seconds = media_budget_seconds - (time.monotonic() - media_started_at)
+        if remaining_seconds <= 0:
+            upload_result = {
+                "status": "failed",
+                "failure_type": "oss_upload_budget_exceeded",
+                "attempt_timeout_seconds": attempt_timeout_seconds,
+                "response_absent": True,
+            }
+            break
+        candidate_started_at = time.monotonic()
+        upload_result = upload_fn(
+            candidate.get("url") or "",
+            timeout_seconds=min(attempt_timeout_seconds, remaining_seconds),
+        )
+        oss_url_attempts.append(
+            _oss_url_attempt_summary(
+                candidate,
+                upload_result,
+                duration_ms=_elapsed_ms(candidate_started_at, time.monotonic()),
+            )
+        )
+        if upload_result.get("status") == "ok":
+            selected_candidate = candidate
+            break
+    if not oss_url_attempts and upload_result.get("failure_type") == "missing_src_url":
+        oss_url_attempts.append(
+            {
+                "candidate_index": 0,
+                "host": "",
+                "path": "",
+                "status": "failed",
+                "failure_type": "missing_src_url",
+                "duration_ms": 0,
+            }
+        )
+    attempt_count = previous_attempt_count + len(oss_url_attempts)
     timing_metrics = {
-        "oss_upload_duration_ms": _elapsed_ms(attempt_started_at, time.monotonic()),
+        "oss_upload_duration_ms": _elapsed_ms(media_started_at, time.monotonic()),
         "oss_upload_timeout_seconds": attempt_timeout_seconds,
+        "oss_media_upload_budget_seconds": media_budget_seconds,
         "oss_archive_attempt_started_at": _iso(now),
     }
     if upload_result.get("status") == "ok":
+        selected_candidate = selected_candidate or {}
         updated_raw = {
             **raw_payload,
             "content_media_status": "oss_uploaded",
@@ -299,6 +343,11 @@ def _archive_one(
             "save_oss_timestamp": upload_result.get("save_oss_timestamp"),
             "oss_timing_metrics": timing_metrics,
             "oss_payload_mode": upload_result.get("oss_payload_mode"),
+            "oss_url_attempts": oss_url_attempts,
+            "oss_archive_attempted_candidate_count": len(oss_url_attempts),
+            "oss_archive_selected_video_url_host": selected_candidate.get("host"),
+            "oss_archive_selected_video_url_path": selected_candidate.get("path"),
+            "oss_archive_selected_video_url_index": selected_candidate.get("candidate_index"),
         }
         return {
             **record,
@@ -332,6 +381,8 @@ def _archive_one(
         "oss_archive_updated_at": _iso(now),
         "upload_failure_reason": failure,
         "oss_timing_metrics": timing_metrics,
+        "oss_url_attempts": oss_url_attempts,
+        "oss_archive_attempted_candidate_count": len(oss_url_attempts),
     }
     return {
         **record,
@@ -380,6 +431,64 @@ def _with_failed_archive(
     }
 
 
+def _upload_candidates(record: dict[str, Any], raw_payload: dict[str, Any]) -> list[dict[str, Any]]:
+    candidates = raw_payload.get("video_url_candidates")
+    if not isinstance(candidates, list):
+        candidates = []
+    result: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    play_url = str(record.get("play_url") or "")
+    if play_url:
+        matched = next(
+            (
+                candidate
+                for candidate in candidates
+                if isinstance(candidate, dict) and str(candidate.get("url") or "") == play_url
+            ),
+            None,
+        )
+        first = dict(matched or {})
+        first.setdefault("url", play_url)
+        first.setdefault("host", urlparse(play_url).netloc)
+        first.setdefault("path", "play_url")
+        first.setdefault("source", "play_url")
+        first.setdefault("candidate_index", 0)
+        result.append(first)
+        seen.add(play_url)
+    for candidate in candidates:
+        if not isinstance(candidate, dict):
+            continue
+        url = str(candidate.get("url") or "")
+        if not url or url in seen:
+            continue
+        seen.add(url)
+        item = dict(candidate)
+        item.setdefault("host", urlparse(url).netloc)
+        item.setdefault("path", "")
+        item.setdefault("candidate_index", len(result))
+        result.append(item)
+    return result
+
+
+def _oss_url_attempt_summary(
+    candidate: dict[str, Any],
+    upload_result: dict[str, Any],
+    *,
+    duration_ms: int,
+) -> dict[str, Any]:
+    return _compact(
+        {
+            "candidate_index": candidate.get("candidate_index"),
+            "host": candidate.get("host"),
+            "path": candidate.get("path"),
+            "status": upload_result.get("status"),
+            "failure_type": upload_result.get("failure_type"),
+            "exception_type": upload_result.get("exception_type"),
+            "duration_ms": duration_ms,
+        }
+    )
+
+
 def _is_due(record: dict[str, Any], now: datetime) -> bool:
     if record.get("content_media_status") == "oss_uploaded":
         return False
@@ -435,3 +544,21 @@ def _resolve_max_workers(value: int | None = None) -> int:
     except ValueError:
         return DEFAULT_OSS_ARCHIVE_MAX_WORKERS
     return parsed if parsed > 0 else DEFAULT_OSS_ARCHIVE_MAX_WORKERS
+
+
+def _resolve_url_fallback_max_attempts() -> int:
+    raw = os.environ.get("CONTENT_AGENT_OSS_URL_FALLBACK_MAX_ATTEMPTS")
+    try:
+        value = int(raw) if raw else DEFAULT_OSS_URL_FALLBACK_MAX_ATTEMPTS
+    except (TypeError, ValueError):
+        value = DEFAULT_OSS_URL_FALLBACK_MAX_ATTEMPTS
+    return max(1, min(value, DEFAULT_OSS_URL_FALLBACK_MAX_ATTEMPTS))
+
+
+def _resolve_media_upload_budget_seconds() -> float:
+    raw = os.environ.get("CONTENT_AGENT_OSS_MEDIA_UPLOAD_BUDGET_SECONDS")
+    try:
+        value = float(raw) if raw else DEFAULT_OSS_MEDIA_UPLOAD_BUDGET_SECONDS
+    except (TypeError, ValueError):
+        value = DEFAULT_OSS_MEDIA_UPLOAD_BUDGET_SECONDS
+    return max(1.0, min(value, DEFAULT_OSS_MEDIA_UPLOAD_BUDGET_SECONDS))

content_agent/integrations/oss_upload.py

@@ -10,7 +10,7 @@ from content_agent.integrations import timeout_config
 
 
 DEFAULT_OSS_UPLOAD_URL = "http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream"
-DEFAULT_OSS_TIMEOUT_SECONDS = 300.0  # 5min(原 3600);read 相另设短,防慢吐字节永久卡 do_poll
+DEFAULT_OSS_TIMEOUT_SECONDS = 300.0  # 5min(原 3600);read 相与单次尝试同为 300s
 
 
 def upload_video_from_url(

content_agent/integrations/platform_video_url.py

@@ -13,6 +13,7 @@ from typing import Any, Callable
 from urllib.parse import urlparse
 
 ProbeFn = Callable[[str, str], dict[str, Any]]
+MAX_VIDEO_URL_CANDIDATES = 3
 
 VIDEO_HINT_RE = re.compile(r"(video|play|mp4|m3u8|media|download|src)", re.I)
 IMAGE_HINT_RE = re.compile(r"(image|cover|poster|thumbnail|thumb|heif|jpg|jpeg|png|kvif)", re.I)
@@ -68,12 +69,46 @@ def select_video_url(
         "play_url": selected.url if selected else None,
         "media_failure_reason": None if selected else "no_valid_play_url",
         "platform_raw_payload": build_diagnostic_payload(candidates, selected),
+        "video_url_candidates": serialize_video_url_candidates(
+            candidates,
+            selected,
+            platform,
+            limit=MAX_VIDEO_URL_CANDIDATES,
+        ),
     }
     if result["media_failure_reason"]:
         result["platform_raw_payload"]["no_valid_play_url"] = True
     return result
 
 
+def serialize_video_url_candidates(
+    candidates: list[VideoUrlCandidate],
+    selected: VideoUrlCandidate | None,
+    platform: str,
+    *,
+    limit: int = MAX_VIDEO_URL_CANDIDATES,
+) -> list[dict[str, Any]]:
+    ordered = _ordered_video_candidates(candidates, selected, platform)[: max(0, limit)]
+    result: list[dict[str, Any]] = []
+    for index, candidate in enumerate(ordered):
+        result.append(
+            _compact_candidate(
+                {
+                    "url": candidate.url,
+                    "host": candidate.host,
+                    "path": candidate.path,
+                    "source": _source_from_path(candidate.path),
+                    "probe_status": candidate.probe_status,
+                    "content_type": candidate.content_type,
+                    "content_range": candidate.content_range,
+                    "selected": bool(selected and candidate.url == selected.url),
+                    "candidate_index": index,
+                }
+            )
+        )
+    return result
+
+
 def find_url_candidates(value: Any, *, prefix: str = "$") -> list[VideoUrlCandidate]:
     candidates: list[VideoUrlCandidate] = []
     if isinstance(value, dict):
@@ -180,6 +215,32 @@ def _select_best(candidates: list[VideoUrlCandidate], platform: str) -> VideoUrl
     return selected
 
 
+def _ordered_video_candidates(
+    candidates: list[VideoUrlCandidate],
+    selected: VideoUrlCandidate | None,
+    platform: str,
+) -> list[VideoUrlCandidate]:
+    video_candidates = [candidate for candidate in candidates if candidate.kind == "video"]
+    ordered: list[VideoUrlCandidate] = []
+    seen: set[str] = set()
+    if selected:
+        ordered.append(selected)
+        seen.add(selected.url)
+    for candidate in sorted(
+        video_candidates,
+        key=lambda item: _selection_score(
+            item,
+            platform,
+            fallback=item.probe_status not in {"verified"},
+        ),
+    ):
+        if candidate.url in seen:
+            continue
+        ordered.append(candidate)
+        seen.add(candidate.url)
+    return ordered
+
+
 def _selection_score(candidate: VideoUrlCandidate, platform: str, *, fallback: bool) -> tuple[Any, ...]:
     detail_bonus = 0 if platform == "kuaishou" and ".detail." in candidate.path else 1
     preferred_host = 0
@@ -187,6 +248,14 @@ def _selection_score(candidate: VideoUrlCandidate, platform: str, *, fallback: b
         preferred_host = 0 if "findermp.video.qq.com" in candidate.host else 1
     elif platform == "kuaishou":
         preferred_host = 0 if ("kwai" in candidate.host or "kwimgs" in candidate.host) else 1
+    elif platform == "douyin":
+        host = candidate.host.lower()
+        if "v11-weba.douyinvod.com" in host:
+            preferred_host = 3
+        elif "douyinvod.com" in host:
+            preferred_host = 0
+        else:
+            preferred_host = 1
     return (
         detail_bonus,
         preferred_host,
@@ -227,6 +296,8 @@ def _download_headers(platform: str) -> dict[str, str]:
         headers["Referer"] = "https://channels.weixin.qq.com/"
     elif platform == "kuaishou":
         headers["Referer"] = "https://www.kuaishou.com/"
+    elif platform == "douyin":
+        headers["Referer"] = "https://www.douyin.com/"
     return headers
 
 
@@ -240,3 +311,17 @@ def _int_or_none(value: Any) -> int | None:
         return int(value)
     except (TypeError, ValueError):
         return None
+
+
+def _source_from_path(path: str) -> str:
+    if not path.startswith("$."):
+        return ""
+    remainder = path[2:]
+    for separator in (".", "["):
+        if separator in remainder:
+            return remainder.split(separator, 1)[0]
+    return remainder
+
+
+def _compact_candidate(candidate: dict[str, Any]) -> dict[str, Any]:
+    return {key: value for key, value in candidate.items() if value not in (None, "", {}, [])}

content_agent/integrations/shipinhao.py

@@ -110,6 +110,8 @@ def _normalize_shipinhao_item(
     }
     if video_url_selection.get("media_failure_reason"):
         result["media_failure_reason"] = video_url_selection["media_failure_reason"]
+    if video_url_selection.get("video_url_candidates"):
+        result["video_url_candidates"] = video_url_selection["video_url_candidates"]
     return result
 
 
@@ -201,6 +203,8 @@ class CrawapiShipinhaoClient:
             "play_url": hydrated.get("play_url"),
             "platform_raw_payload": hydrated.get("platform_raw_payload", {}),
         }
+        if hydrated.get("video_url_candidates"):
+            merged["video_url_candidates"] = hydrated["video_url_candidates"]
         if selection.get("media_failure_reason"):
             merged["media_failure_reason"] = selection["media_failure_reason"]
         else:

content_agent/integrations/timeout_config.py

@@ -35,9 +35,9 @@ _HARD_CEILING: dict[str, float] = {
     "query_llm": 300.0,
     "pg": 60.0,
 }
-# 单次 read(两次收到数据之间)上限——短,杜绝 do_poll 永久阻塞。
+# 单次 read(两次收到数据之间)上限。OSS 按业务要求放宽到 300s。
 _READ: dict[str, float] = {
-    "oss": 60.0,
+    "oss": 300.0,
     "video_download": 120.0,
     "video_llm": 120.0,
     "crawapi": 60.0,

tests/test_content_discovery_media_diagnostics.py

@@ -29,6 +29,14 @@ def test_media_record_keeps_no_valid_play_url_diagnostics(tmp_path):
                 "content_metadata_source": "shipinhao_keyword_search",
                 "play_url": None,
                 "media_failure_reason": "no_valid_play_url",
+                "video_url_candidates": [
+                    {
+                        "url": "https://findermp.video.qq.com/video.mp4",
+                        "host": "findermp.video.qq.com",
+                        "path": "$.search.video_url_list[0].video_url",
+                        "source": "search",
+                    }
+                ],
                 "platform_raw_payload": {
                     "channel_content_id": "finder_001",
                     "selected_video_url_path": "$.search.video_url_list[0].video_url",
@@ -58,6 +66,7 @@ def test_media_record_keeps_no_valid_play_url_diagnostics(tmp_path):
     assert media["raw_payload"]["no_valid_play_url"] is True
     assert media["raw_payload"]["selected_video_url_host"] == "findermp.video.qq.com"
     assert media["raw_payload"]["video_url_candidate_counts"] == {"video": 1, "image": 1}
+    assert media["raw_payload"]["video_url_candidates"][0]["url"] == "https://findermp.video.qq.com/video.mp4"
     assert media["raw_payload"]["shipinhao_research_status"] == "not_matched"
     assert media["raw_payload"]["shipinhao_research_item_count"] == 8
     assert media["raw_payload"]["kuaishou_detail_fallback_status"] == "no_valid_play_url"

tests/test_douyin_detail.py

@@ -50,7 +50,10 @@ def test_fetch_detail_maps_canonical_fields():
                 "content_link": "https://www.douyin.com/video/7522164415848893735",
                 "body_text": "原来彩虹真的是圆形的 #治愈系风景 #彩虹",
                 "topic_list": ["治愈系风景", "彩虹", "旅行"],
-                "video_url_list": [{"video_url": "https://www.douyin.com/aweme/v1/play/?video_id=x"}],
+                "video_url_list": [
+                    {"video_url": "https://v11-weba.douyinvod.com/x.mp4"},
+                    {"video_url": "https://v26-web.douyinvod.com/x.mp4"},
+                ],
                 "channel_account_id": "MS4wLjABAAAA",
                 "channel_account_name": "源Dream",
                 "like_count": 5034215,
@@ -66,7 +69,56 @@ def test_fetch_detail_maps_canonical_fields():
     assert result["platform_content_id"] == "7522164415848893735"
     assert result["platform_author_id"] == "MS4wLjABAAAA"
     assert result["tags"] == ["#治愈系风景", "#彩虹", "#旅行"]
-    assert result["play_url"] == "https://www.douyin.com/aweme/v1/play/?video_id=x"
+    assert result["play_url"] == "https://v26-web.douyinvod.com/x.mp4"
+    assert [candidate["host"] for candidate in result["video_url_candidates"]] == [
+        "v26-web.douyinvod.com",
+        "v11-weba.douyinvod.com",
+    ]
     assert result["statistics"]["digg_count"] == 5034215
     assert result["create_time"] == 1751515440  # ms -> s
     assert result["content_metadata_source"] == "douyin_detail"
+
+
+def test_douyin_search_uses_detail_fallback_for_high_risk_host():
+    search_payload = {
+        "code": 0,
+        "data": {
+            "data": [
+                {
+                    "aweme_id": "7522164415848893735",
+                    "desc": "高风险 host",
+                    "author": {"sec_uid": "MS4wLjABAAAA", "nickname": "源Dream"},
+                    "statistics": {},
+                    "video": {
+                        "play_addr": {
+                            "url_list": ["https://v11-weba.douyinvod.com/search.mp4"]
+                        }
+                    },
+                }
+            ],
+            "has_more": False,
+        },
+    }
+    detail_payload = {
+        "code": 0,
+        "data": {
+            "data": {
+                "channel_content_id": "7522164415848893735",
+                "video_url_list": [
+                    {"video_url": "https://v26-web.douyinvod.com/detail.mp4"}
+                ],
+            }
+        },
+    }
+
+    result = _client([_response(search_payload), _response(detail_payload)]).search(
+        {
+            "search_query_id": "q_001",
+            "search_query": "高风险 host",
+            "discovery_start_source": "pattern_itemset",
+        }
+    )[0]
+
+    assert result["play_url"] == "https://v26-web.douyinvod.com/detail.mp4"
+    assert result["platform_raw_payload"]["douyin_detail_fallback_status"] == "used"
+    assert result["video_url_candidates"][0]["source"] == "detail"

tests/test_kuaishou_client.py

@@ -93,6 +93,8 @@ def test_kuaishou_search_maps_canonical_fields():
     assert result["platform_author_id"] == "3xfkwajatdh7p7i"
     assert result["author_display_name"] == "祝福账号"
     assert result["play_url"] == "https://v.kwaicdn.test/a.mp4"
+    assert result["video_url_candidates"][0]["url"] == "https://v.kwaicdn.test/a.mp4"
+    assert result["video_url_candidates"][0]["source"] == "search"
     assert result["statistics"] == {
         "digg_count": 234,
         "comment_count": 34,
@@ -122,6 +124,7 @@ def test_kuaishou_search_falls_back_to_detail_video_url():
     assert result["play_url"] == "https://v.kwaicdn.test/detail.mp4"
     assert result["platform_raw_payload"]["selected_video_url_path"] == "$.detail.video_url_list[0].video_url"
     assert result["platform_raw_payload"]["kuaishou_detail_fallback_status"] == "used"
+    assert result["video_url_candidates"][0]["url"] == "https://v.kwaicdn.test/detail.mp4"
     assert "media_failure_reason" not in result
 
 
@@ -164,6 +167,7 @@ def test_kuaishou_hydrate_search_result_media_uses_detail_fallback():
     assert client.http_client.requests[1]["url"].endswith("/crawler/kuai_shou/detail")
     assert hydrated["play_url"] == "https://v.kwaicdn.test/detail.mp4"
     assert hydrated["platform_raw_payload"]["kuaishou_detail_fallback_status"] == "used"
+    assert hydrated["video_url_candidates"][0]["url"] == "https://v.kwaicdn.test/detail.mp4"
     assert "_platform_search_item" not in hydrated
 
 

tests/test_oss_archive.py

@@ -218,7 +218,7 @@ def test_archive_due_records_keeps_failed_attempt_pending_before_deadline():
             "error_message": "proxy timed out",
             "oss_payload_mode": "no_referer",
             "endpoint_host": "crawler-upload-v2.aiddit.com",
-            "read_timeout_seconds": 60.0,
+            "read_timeout_seconds": 300.0,
             "attempt_timeout_seconds": 300.0,
             "response_absent": True,
         }
@@ -232,13 +232,73 @@ def test_archive_due_records_keeps_failed_attempt_pending_before_deadline():
     assert row["raw_payload"]["oss_archive_last_error"] == "oss_upload_http_error"
     assert row["raw_payload"]["oss_archive_last_error_message"] == "proxy timed out"
     assert row["raw_payload"]["oss_endpoint_host"] == "crawler-upload-v2.aiddit.com"
-    assert row["raw_payload"]["oss_read_timeout_seconds"] == 60.0
+    assert row["raw_payload"]["oss_read_timeout_seconds"] == 300.0
     assert row["raw_payload"]["oss_attempt_timeout_seconds"] == 300.0
     assert row["raw_payload"]["oss_response_absent"] is True
     assert row["raw_payload"]["oss_archive_next_retry_at"] == (NOW + timedelta(minutes=15)).isoformat()
     assert row["raw_payload"]["oss_timing_metrics"]["oss_upload_duration_ms"] >= 0
 
 
+def test_archive_due_records_tries_fallback_candidate_after_first_url_fails():
+    [pending] = mark_archive_pending(
+        [
+            _record(
+                raw_payload={
+                    "run_id": "run_001",
+                    "platform_content_id": "content_001",
+                    "video_url_candidates": [
+                        {
+                            "url": "https://source.example/video.mp4",
+                            "host": "source.example",
+                            "path": "$.search.video_url_list[0].video_url",
+                            "source": "search",
+                            "candidate_index": 0,
+                        },
+                        {
+                            "url": "https://backup.example/video.mp4",
+                            "host": "backup.example",
+                            "path": "$.detail.video_url_list[0].video_url",
+                            "source": "detail",
+                            "candidate_index": 1,
+                        },
+                    ],
+                },
+            )
+        ],
+        now_fn=lambda: NOW,
+    )
+    calls: list[str] = []
+
+    def upload(src_url, **kwargs):
+        calls.append(src_url)
+        if src_url == "https://source.example/video.mp4":
+            return {
+                "status": "failed",
+                "failure_type": "oss_upload_http_error",
+                "exception_type": "ReadTimeout",
+                "read_timeout_seconds": 300.0,
+                "attempt_timeout_seconds": kwargs["timeout_seconds"],
+                "response_absent": True,
+            }
+        return {
+            "status": "ok",
+            "oss_url": "https://res.example/video.mp4",
+            "oss_object_key": "crawler/video/content_001.mp4",
+        }
+
+    [row] = archive_due_records([pending], upload_fn=upload, now_fn=lambda: NOW)
+
+    assert calls == ["https://source.example/video.mp4", "https://backup.example/video.mp4"]
+    assert row["content_media_status"] == "oss_uploaded"
+    assert row["oss_url"] == "https://res.example/video.mp4"
+    assert row["raw_payload"]["oss_archive_attempt_count"] == 2
+    assert row["raw_payload"]["oss_archive_attempted_candidate_count"] == 2
+    assert row["raw_payload"]["oss_archive_selected_video_url_host"] == "backup.example"
+    assert row["raw_payload"]["oss_archive_selected_video_url_path"] == "$.detail.video_url_list[0].video_url"
+    assert row["raw_payload"]["oss_url_attempts"][0]["failure_type"] == "oss_upload_http_error"
+    assert row["raw_payload"]["oss_url_attempts"][1]["status"] == "ok"
+
+
 def test_archive_due_records_keeps_invalid_response_summary():
     [pending] = mark_archive_pending([_record()], now_fn=lambda: NOW)
 

tests/test_oss_upload.py

@@ -85,7 +85,7 @@ def test_upload_video_from_url_read_timeout_keeps_diagnostics():
     assert result["exception_type"] == "ReadTimeout"
     assert result["response_absent"] is True
     assert result["endpoint_host"] == "oss.test"
-    assert result["read_timeout_seconds"] == 60.0
+    assert result["read_timeout_seconds"] == 300.0
     assert result["attempt_timeout_seconds"] == 123
 
 

tests/test_pattern_recall_archive.py

@@ -240,6 +240,58 @@ def test_pattern_recall_skips_qwen_when_oss_archive_is_pending(tmp_path):
     assert evidence["evidence_summary"]["failure_type"] == "oss_upload_http_error"
 
 
+def test_pattern_recall_enqueues_qwen_after_oss_fallback_candidate_succeeds(tmp_path):
+    run_id = "run_001"
+    policy_run_id = "policy_run_001"
+    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
+    runtime.prepare_run(run_id)
+    item = _item(run_id, policy_run_id, "content_pool", "d_001")
+    media = _media(run_id, policy_run_id, item["platform_content_id"])
+    media["raw_payload"]["video_url_candidates"] = [
+        {
+            "url": "https://source.example/content_pool.mp4",
+            "host": "source.example",
+            "path": "$.search.video_url_list[0].video_url",
+            "source": "search",
+            "candidate_index": 0,
+        },
+        {
+            "url": "https://backup.example/content_pool.mp4",
+            "host": "backup.example",
+            "path": "$.detail.video_url_list[0].video_url",
+            "source": "detail",
+            "candidate_index": 1,
+        },
+    ]
+    bundle = _bundle(item)
+    dispatcher = FallbackUploadDispatcher()
+    gemini = OssAwareGeminiVideoClient()
+
+    result = recall_decision.run(
+        run_id,
+        policy_run_id,
+        [item],
+        [media],
+        [bundle],
+        {"ext_data": {"evidence_pack": {"seed_terms": ["养生"]}}},
+        runtime,
+        gemini,
+        archive_dispatcher=dispatcher,
+    )
+
+    assert dispatcher.calls == [
+        "https://source.example/content_pool.mp4",
+        "https://backup.example/content_pool.mp4",
+    ]
+    assert gemini.calls
+    [row] = result["content_media_records"]
+    assert row["content_media_status"] == "oss_uploaded"
+    assert row["raw_payload"]["oss_archive_selected_video_url_host"] == "backup.example"
+    assert row["raw_payload"]["oss_url_attempts"][0]["failure_type"] == "oss_upload_http_error"
+    tasks = runtime.read_media_pipeline_tasks(run_id)
+    assert {task["task_type"] for task in tasks} == {"oss_upload", "qwen_judgment"}
+
+
 class RecordingArchiveDispatcher:
     def __init__(self) -> None:
         self.submitted_content_ids: list[str] = []
@@ -315,6 +367,26 @@ class PendingArchiveDispatcher:
         return archived
 
 
+class FallbackUploadDispatcher:
+    def __init__(self) -> None:
+        self.calls: list[str] = []
+
+    def upload_fn(self, src_url: str, **kwargs) -> dict:
+        self.calls.append(src_url)
+        if src_url.startswith("https://source.example/"):
+            return {
+                "status": "failed",
+                "failure_type": "oss_upload_http_error",
+                "exception_type": "ReadTimeout",
+                "response_absent": True,
+            }
+        return {
+            "status": "ok",
+            "oss_url": "https://res.example/content_pool.mp4",
+            "oss_object_key": "crawler/video/content_pool.mp4",
+        }
+
+
 class DelayedArchiveDispatcher:
     def __init__(self, events: list[str], lock: threading.Lock) -> None:
         self.events = events

tests/test_platform_video_url.py

@@ -62,6 +62,11 @@ def test_kuaishou_prefers_verified_detail_video_over_search():
     assert result["play_url"] == "https://tymov2.a.kwimgs.com/detail.mp4"
     assert result["platform_raw_payload"]["selected_video_url_path"] == "$.detail.video_url_list[0].video_url"
     assert result["platform_raw_payload"]["selected_video_url_probe_status"] == "verified"
+    assert [candidate["url"] for candidate in result["video_url_candidates"]] == [
+        "https://tymov2.a.kwimgs.com/detail.mp4",
+        "https://v23-3.kwaicdn.com/search.mp4",
+    ]
+    assert "video_url_candidates" not in result["platform_raw_payload"]
 
 
 def test_shipinhao_prefers_findermp_video_and_filters_non_video():
@@ -98,6 +103,56 @@ def test_probe_failure_falls_back_to_strong_video_candidate():
     assert result["media_failure_reason"] is None
 
 
+def test_douyin_demotes_v11_weba_when_better_candidate_exists():
+    result = select_video_url(
+        "douyin",
+        [
+            (
+                "search",
+                {
+                    "video": {
+                        "play_addr": {
+                            "url_list": [
+                                "https://v11-weba.douyinvod.com/video.mp4",
+                                "https://v26-web.douyinvod.com/video.mp4",
+                            ]
+                        }
+                    }
+                },
+            )
+        ],
+        probe_fn=_probe(),
+    )
+
+    assert result["play_url"] == "https://v26-web.douyinvod.com/video.mp4"
+    assert result["platform_raw_payload"]["selected_video_url_host"] == "v26-web.douyinvod.com"
+    assert [candidate["host"] for candidate in result["video_url_candidates"]] == [
+        "v26-web.douyinvod.com",
+        "v11-weba.douyinvod.com",
+    ]
+
+
+def test_video_url_candidates_are_capped_to_selected_plus_two():
+    result = select_video_url(
+        "kuaishou",
+        [
+            (
+                "search",
+                {
+                    "video_url_list": [
+                        {"video_url": f"https://v{index}.kwaicdn.com/{index}.mp4"}
+                        for index in range(5)
+                    ]
+                },
+            )
+        ],
+        probe_fn=_probe(),
+    )
+
+    assert len(result["video_url_candidates"]) == 3
+    assert result["video_url_candidates"][0]["url"] == result["play_url"]
+
+
 def test_no_video_candidate_returns_no_valid_play_url():
     result = select_video_url(
         "kuaishou",
@@ -108,3 +163,4 @@ def test_no_video_candidate_returns_no_valid_play_url():
     assert result["play_url"] is None
     assert result["media_failure_reason"] == "no_valid_play_url"
     assert result["platform_raw_payload"]["no_valid_play_url"] is True
+    assert result["video_url_candidates"] == []

tests/test_timeout_hardening.py

@@ -171,7 +171,7 @@ def test_oss_upload_passes_segmented_timeout():
 
     oss_upload.upload_video_from_url("http://v/1.mp4", http_post=fake_post)
     assert isinstance(captured["timeout"], httpx.Timeout)
-    assert captured["timeout"].read == timeout_config.read_timeout("oss")   # 60
+    assert captured["timeout"].read == timeout_config.read_timeout("oss")   # 300
     assert captured["timeout"].write == 300.0
 
 

web2/features/RunListPage.tsx

@@ -198,6 +198,7 @@ export function RunListPage() {
                           <Link className="rbd-run" href={`/runs/${encodeURIComponent(run.run_id)}`} key={run.run_id}>
                             <span className="rbd-run-time">{formatBeijingTime(run.started_at)}</span>
                             <span className={`chip ${statusChipClass(run.status)}`}>{statusLabel(run.status)}</span>
+                            {run.strategy_version ? <span className="chip">策略 {run.strategy_version}</span> : null}
                             <span className="rbd-run-id muted">{run.run_id}</span>
                           </Link>
                         ))}