Harden video media stabilization and simplify board status UI

.env.example

@@ -53,6 +53,7 @@ CREATION_SEARCH_SORT_TYPE=综合
 # -----------------------------------------------------------------------------
 CRAWLER_OSS_UPLOAD_URL=http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream
 CRAWLER_OSS_UPLOAD_TIMEOUT_SECONDS=60
+CK_VIDEO_OSS_RETRY_DELAYS_SECONDS=5,10
 
 # -----------------------------------------------------------------------------
 # LLM providers

acquisition/media/service.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+import time
 from typing import Callable
 
 from acquisition.media.oss_client import to_oss
@@ -15,6 +16,7 @@ class StabilizedMedia:
     cdn_url: str
     position: int
     status: str = "ready"
+    error_message: str = ""
 
 
 Uploader = Callable[[str, str], str]
@@ -26,12 +28,14 @@ def stabilize_media_urls(
     video_urls: list[str] | None = None,
     settings: Settings | None = None,
     uploader: Uploader | None = None,
+    video_fallback_to_source: bool = True,
+    video_retry_delays_seconds: tuple[float, ...] = (),
+    sleep: Callable[[float], None] = time.sleep,
 ) -> list[StabilizedMedia]:
     """Transfer image/video URLs to OSS/CDN and return DB-ready media rows.
 
-    The uploader falls back to the original source URL when OSS transfer fails,
-    so acquisition can still store the candidate and make the failure visible in
-    later review instead of dropping the item.
+    Images keep the old soft-fallback behavior. Videos can opt into strict mode
+    so expiring source URLs do not leak into classification/decode.
     """
     upload = uploader or (
         lambda url, media_type: to_oss(url, media_type, settings=settings)
@@ -59,9 +63,22 @@ def stabilize_media_urls(
         if not url:
             continue
         position += 1
-        try:
-            cdn = upload(url, "video")
-        except Exception:
+        cdn = ""
+        error_message = ""
+        attempts = len(video_retry_delays_seconds) + 1
+        for attempt in range(attempts):
+            try:
+                cdn = upload(url, "video")
+                if cdn and (video_fallback_to_source or cdn != url):
+                    error_message = ""
+                    break
+                error_message = "oss_upload_returned_source_url"
+                cdn = ""
+            except Exception as exc:
+                error_message = str(exc)
+            if attempt < len(video_retry_delays_seconds):
+                sleep(video_retry_delays_seconds[attempt])
+        if not cdn and video_fallback_to_source:
             cdn = url
         rows.append(
             StabilizedMedia(
@@ -70,6 +87,7 @@ def stabilize_media_urls(
                 cdn_url=cdn,
                 position=position,
                 status="ready" if cdn else "failed",
+                error_message=error_message,
             )
         )
     return rows

acquisition/runner.py

@@ -81,7 +81,7 @@ def _job_id(job: AcquisitionJob) -> UUID:
 
 def _best_video_url(media: list[StabilizedMedia]) -> str:
     for row in media:
-        if row.media_type == "video" and row.cdn_url:
+        if row.media_type == "video" and row.status == "ready" and row.cdn_url:
             return row.cdn_url
     return ""
 
@@ -143,11 +143,32 @@ def _content_mode(platform: str, detail: Any) -> str:
 def _skip_reason_message(reason: str) -> str:
     if reason == "video_url_missing":
         return "Video post has no usable video URL; skipped before classification."
+    if reason == "video_oss_failed":
+        return "Video OSS transfer failed after retries; skipped before classification."
     if reason == "unsupported_content_mode":
         return "Unsupported content mode; skipped before classification."
     return reason or "Skipped before classification."
 
 
+def _requires_strict_video_oss(platform: str, content_mode: str) -> bool:
+    return content_mode == "video_post"
+
+
+def _video_media_errors(media_rows: list[StabilizedMedia]) -> list[dict[str, Any]]:
+    errors: list[dict[str, Any]] = []
+    for row in media_rows:
+        if row.media_type != "video" or row.status == "ready":
+            continue
+        errors.append(
+            {
+                "source_url": row.source_url,
+                "status": row.status,
+                "error_message": row.error_message,
+            }
+        )
+    return errors
+
+
 def _record_candidate(
     repo: AcquisitionRepository,
     *,
@@ -231,10 +252,15 @@ def _record_candidate(
         )
         return RecordCandidateResult(displayed=True, skipped_processing=True)
 
+    strict_video_oss = _requires_strict_video_oss(platform, content_mode)
     media_rows = media_stabilizer(
         image_urls=detail.image_urls,
         video_urls=detail.video_urls,
         settings=settings,
+        video_fallback_to_source=not strict_video_oss,
+        video_retry_delays_seconds=(
+            settings.video_oss_retry_delays_seconds if strict_video_oss else ()
+        ),
     )
 
     for row in media_rows:
@@ -247,8 +273,30 @@ def _record_candidate(
             position=row.position,
             status=row.status,
             source_payload={},
-            metadata={},
+            metadata={"error_message": row.error_message} if row.error_message else {},
+        )
+
+    if strict_video_oss and not _best_video_url(media_rows):
+        repo.add_item_classification(
+            item_id=item.id,
+            is_creation_knowledge=None,
+            label="video_oss_failed",
+            confidence=None,
+            reason=_skip_reason_message("video_oss_failed"),
+            model_name=None,
+            prompt_version="media_stabilization_guard",
+            result_payload={
+                "content_mode": content_mode,
+                "skip_reason": "video_oss_failed",
+                "video_oss_retry_delays_seconds": list(
+                    settings.video_oss_retry_delays_seconds
+                ),
+                "media_errors": _video_media_errors(media_rows),
+            },
+            status="skipped",
+            error_message="video_oss_failed",
         )
+        return RecordCandidateResult(displayed=True, skipped_processing=True)
 
     if classify:
         result = classifier(

app/frontend/dist/index.html

@@ -4,8 +4,8 @@
     <meta charset="UTF-8" />
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <title>创作知识</title>
-    <script type="module" crossorigin src="/app/assets/index-BsOtccz2.js"></script>
-    <link rel="stylesheet" crossorigin href="/app/assets/index-C-saPy1c.css">
+    <script type="module" crossorigin src="/app/assets/index-D2o7sqDa.js"></script>
+    <link rel="stylesheet" crossorigin href="/app/assets/index-qEoc1W1z.css">
   </head>
   <body>
     <div id="root"></div>

app/frontend/src/features/query-board/AxisColumn.jsx

@@ -1,15 +1,12 @@
 import { useState } from 'react'
-import { AXIS_ICON, AXIS_LABEL, axisRowStats, axisValues, hasFinalKnowledge, isQuerySearched, pct } from './model.js'
+import { AXIS_ICON, AXIS_LABEL, axisRowStats, axisValues } from './model.js'
 
-export default function AxisColumn({ data, family, axis, latestByQuery }) {
+export default function AxisColumn({ data, family, axis }) {
   const values = axisValues(data, family, axis)
   const tree = data.axis_trees?.[axis] || []
   const hasTree = tree.length > 0
   const [collapsed, setCollapsed] = useState(() => new Set())
   const familyItems = family.items || []
-  const searched = familyItems.filter((item) => isQuerySearched(latestByQuery.get(item.query))).length
-  const knowledge = familyItems.filter((item) => hasFinalKnowledge(latestByQuery.get(item.query))).length
-  const total = familyItems.length || 1
   const toggle = (path) => {
     setCollapsed((prev) => {
       const next = new Set(prev)
@@ -20,15 +17,13 @@ export default function AxisColumn({ data, family, axis, latestByQuery }) {
   }
 
   const renderFlatRow = (value, index) => {
-    const stats = axisRowStats(familyItems, latestByQuery, axis, value)
-    const isMuted = stats.searched === 0 && index > 8
+    const stats = axisRowStats(familyItems, axis, value)
+    const isMuted = stats.total === 0 && index > 8
     return (
       <div className={`axv ${isMuted ? 'muted' : ''}`} key={value}>
         <span className="tree-caret empty" />
         <span className="tree-label">{value}</span>
-        <span className="tree-dots">
-          {stats.searched > 0 && <i className="dot blue" />}
-          {stats.knowledge > 0 && <i className="dot green" />}
+        <span className="tree-count">
           {stats.total > 0 && <b>{stats.total}</b>}
         </span>
       </div>
@@ -38,13 +33,13 @@ export default function AxisColumn({ data, family, axis, latestByQuery }) {
   const renderTreeNode = (node, level = 3) => {
     const children = node.children || []
     const childNames = children.map((child) => child.name)
-    const stats = axisRowStats(familyItems, latestByQuery, axis, [node.name, ...childNames])
+    const stats = axisRowStats(familyItems, axis, [node.name, ...childNames])
     const isCollapsed = collapsed.has(node.path)
     const canExpand = level === 3 && children.length > 0
     return (
       <div className="tree-group" key={node.path}>
         <button
-          className={`axv tree-node level-${level} ${stats.searched === 0 ? 'muted' : ''}`}
+          className={`axv tree-node level-${level} ${stats.total === 0 ? 'muted' : ''}`}
           type="button"
           onClick={() => canExpand && toggle(node.path)}
         >
@@ -52,21 +47,17 @@ export default function AxisColumn({ data, family, axis, latestByQuery }) {
             {canExpand ? (isCollapsed ? '›' : '⌄') : ''}
           </span>
           <span className="tree-label">{node.name}</span>
-          <span className="tree-dots">
-            {stats.searched > 0 && <i className="dot blue" />}
-            {stats.knowledge > 0 && <i className="dot green" />}
+          <span className="tree-count">
             {stats.total > 0 && <b>{stats.total}</b>}
           </span>
         </button>
         {canExpand && !isCollapsed && children.map((child) => {
-          const childStats = axisRowStats(familyItems, latestByQuery, axis, child.name)
+          const childStats = axisRowStats(familyItems, axis, child.name)
           return (
-            <div className={`axv tree-node level-4 ${childStats.searched === 0 ? 'muted' : ''}`} key={child.path}>
+            <div className={`axv tree-node level-4 ${childStats.total === 0 ? 'muted' : ''}`} key={child.path}>
               <span className="tree-caret empty" />
               <span className="tree-label">{child.name}</span>
-              <span className="tree-dots">
-                {childStats.searched > 0 && <i className="dot blue" />}
-                {childStats.knowledge > 0 && <i className="dot green" />}
+              <span className="tree-count">
                 {childStats.total > 0 && <b>{childStats.total}</b>}
               </span>
             </div>
@@ -83,20 +74,6 @@ export default function AxisColumn({ data, family, axis, latestByQuery }) {
           <span className="axis-icon">{AXIS_ICON[axis] || axis.slice(0, 1)}</span>
           <span>{AXIS_LABEL[axis] || axis}</span>
         </div>
-        <div className="axis-metrics">
-          <div className="metric-pair">
-            <span className="metric-blue">搜索 {pct(searched, total)}%</span>
-            <span className="metric-green">知识 {pct(knowledge, total)}%</span>
-          </div>
-          <div className="mini-bars" aria-hidden="true">
-            <span style={{ width: `${Math.min(100, pct(searched, total))}%` }} />
-            <span style={{ width: `${Math.min(100, pct(knowledge, total))}%` }} />
-          </div>
-          <div className="metric-counts">
-            <span>{searched}/{familyItems.length}</span>
-            <span>{knowledge}/{familyItems.length}</span>
-          </div>
-        </div>
       </div>
       <div className="axlist">
         {hasTree ? tree.map((node) => renderTreeNode(node)) : values.map(renderFlatRow)}

app/frontend/src/features/query-board/QueryBoardPage.jsx

@@ -80,10 +80,6 @@ export default function QueryBoardPage() {
   return (
     <div className="dashboard-page">
       <div className="dashboard-toolbar">
-        <div className="filter-chips">
-          <button className="filter-chip blue active" type="button"><span>✓</span>搜索过的</button>
-          <button className="filter-chip green active" type="button"><span>✓</span>有知识的</button>
-        </div>
         <div className="family-switch">
           {families.map((row) => (
             <button
@@ -109,7 +105,6 @@ export default function QueryBoardPage() {
             axis={axis}
             data={preview}
             family={family}
-            latestByQuery={latestByQuery}
           />
         ))}
         <QueryColumn

app/frontend/src/features/query-board/model.js

@@ -87,13 +87,11 @@ export function axisValues(data, family, axis) {
   return data.axis_values?.[key] || uniqueAxisValues(family.items, key)
 }
 
-export function axisRowStats(familyItems, latestByQuery, axis, value) {
+export function axisRowStats(familyItems, axis, value) {
   const key = axis === '作用/感受/意图' ? '目的池' : axis
   const values = Array.isArray(value) ? new Set(value) : new Set([value])
   const related = familyItems.filter((item) => values.has(item.parts?.[key]))
-  const searched = related.filter((item) => isQuerySearched(latestByQuery.get(item.query))).length
-  const knowledge = related.filter((item) => hasFinalKnowledge(latestByQuery.get(item.query))).length
-  return { total: related.length, searched, knowledge }
+  return { total: related.length }
 }
 
 export function singletonMaps(singleton) {
@@ -133,10 +131,6 @@ export function itemBrief(item) {
   return item.brief || item.raw_summary || item.classification?.reason || ''
 }
 
-export function hasFinalKnowledge(row) {
-  return (row?.decoded_count || 0) > 0 && (row?.payload_count || 0) > 0
-}
-
 export function isQuerySearched(row) {
   return (row?.candidate_count || 0) > 0
 }

app/frontend/src/styles/legacy.css

@@ -815,7 +815,6 @@ h1 {
   padding: 20px 14px;
 }
 
-.filter-chips,
 .family-switch {
   display: flex;
   align-items: center;
@@ -823,55 +822,8 @@ h1 {
   flex-wrap: wrap;
 }
 
-.filter-chip {
-  display: inline-flex;
-  align-items: center;
-  gap: 7px;
-  height: 32px;
-  padding: 0 11px;
-  border: 1px solid var(--line);
-  border-radius: 6px;
-  background: #fff;
-  color: #4e5969;
-  font-size: 13px;
-  font-weight: 600;
-}
-
-.filter-chip span {
-  width: 14px;
-  height: 14px;
-  display: inline-grid;
-  place-items: center;
-  border-radius: 3px;
-  font-size: 10px;
-  color: #fff;
-}
-
-.filter-chip.blue {
-  color: #165dff;
-  border-color: #94bfff;
-  background: #e8f3ff;
-}
-
-.filter-chip.blue span {
-  background: #165dff;
-}
-
-.filter-chip.green {
-  color: #00875a;
-  border-color: #7be0b2;
-  background: #e8fff3;
-}
-
-.filter-chip.green span {
-  background: #00a870;
-}
-
 .family-switch {
   min-height: 32px;
-  margin-left: 8px;
-  padding-left: 12px;
-  border-left: 1px solid var(--line);
 }
 
 .clear-filter {
@@ -1006,50 +958,6 @@ h1 {
   margin-top: 9px;
 }
 
-.metric-pair,
-.metric-counts {
-  display: flex;
-  align-items: center;
-  justify-content: space-between;
-  gap: 8px;
-  font-size: 10px;
-  font-weight: 700;
-}
-
-.metric-blue {
-  color: #165dff;
-}
-
-.metric-green {
-  color: #00a870;
-}
-
-.mini-bars {
-  display: grid;
-  grid-template-columns: 1fr 1fr;
-  gap: 8px;
-  margin: 4px 0 3px;
-}
-
-.mini-bars span {
-  height: 3px;
-  min-width: 8px;
-  border-radius: 999px;
-}
-
-.mini-bars span:first-child {
-  background: #165dff;
-}
-
-.mini-bars span:last-child {
-  background: #00a870;
-}
-
-.metric-counts {
-  color: #86909c;
-  font-weight: 500;
-}
-
 .dashboard-page .qhd {
   display: flex;
   align-items: flex-start;
@@ -1141,32 +1049,16 @@ h1 {
   text-overflow: ellipsis;
 }
 
-.tree-dots {
+.tree-count {
   margin-left: auto;
   display: inline-flex;
   align-items: center;
-  gap: 4px;
   min-width: 32px;
   justify-content: flex-end;
 }
 
-.dot {
-  width: 5px;
-  height: 5px;
-  border-radius: 50%;
-  display: inline-block;
-}
-
-.dot.blue {
-  background: #165dff;
-}
-
-.dot.green {
-  background: #00a870;
-}
-
-.tree-dots b {
-  color: #00a870;
+.tree-count b {
+  color: #86909c;
   font-size: 9px;
   font-weight: 700;
 }

core/config.py

@@ -40,6 +40,21 @@ def env_value(
     return value or ""
 
 
+def env_float_tuple(
+    key: str,
+    file_env: dict[str, str],
+    default: str,
+) -> tuple[float, ...]:
+    raw = env_value(key, file_env, default)
+    values: list[float] = []
+    for part in raw.split(","):
+        text = part.strip()
+        if not text:
+            continue
+        values.append(float(text))
+    return tuple(values)
+
+
 @dataclass
 class PgConfig:
     """过程库（Greenplum open_aigc，schema creation_knowledge）连接配置。"""
@@ -124,6 +139,7 @@ class Settings:
     # query 搜索 + OSS 转存（默认值必填：tests/test_video_extract.py 用关键字构造 Settings 且不传这些）
     oss_upload_url: str = "http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream"
     oss_upload_timeout_seconds: int = 60
+    video_oss_retry_delays_seconds: tuple[float, ...] = (5.0, 10.0)
     search_default_limit: int = 5
     search_content_type: str = "图文"
     search_sort_type: str = "综合"
@@ -206,6 +222,11 @@ class Settings:
                     "60",
                 )
             ),
+            video_oss_retry_delays_seconds=env_float_tuple(
+                "CK_VIDEO_OSS_RETRY_DELAYS_SECONDS",
+                file_env,
+                "5,10",
+            ),
             search_default_limit=int(env_value(
                 "CREATION_SEARCH_DEFAULT_LIMIT", file_env,
                 "5",

decode_content/readers/service.py

@@ -51,6 +51,10 @@ def _media_url(asset: MediaAsset) -> str | None:
     return asset.cdn_url or asset.oss_url or asset.source_url
 
 
+def _stable_media_url(asset: MediaAsset) -> str | None:
+    return asset.cdn_url or asset.oss_url
+
+
 def post_from_candidate_item(item: CandidateItem, media_assets: list[MediaAsset]) -> Post:
     """Build the reader input from formal acquisition DB objects."""
 
@@ -64,7 +68,9 @@ def post_from_candidate_item(item: CandidateItem, media_assets: list[MediaAsset]
             continue
         media_type = asset.media_type.lower()
         if media_type == "video":
-            video_urls.append(url)
+            stable_url = _stable_media_url(asset)
+            if stable_url:
+                video_urls.append(stable_url)
             continue
         if media_type in {"image", "cover", "frame"}:
             image_urls.append(url)
@@ -174,7 +180,11 @@ def read_post(
         else:
             if settings is None:
                 raise ValueError("settings is required when reading video posts")
-            extracted = read_video(post, settings=settings)
+            extracted = read_video(
+                post,
+                settings=settings,
+                oss_video_url=post.video_urls[0],
+            )
     else:
         extracted = image_reader(post) if image_reader is not None else read_imgtext(post, extractor=extractor)
     return read_result_from_extracted(post, extracted)

tests/test_acquisition_runner.py

@@ -583,6 +583,45 @@ def test_run_batch_records_douyin_provider_in_metadata_and_source_payload():
     assert item.source_payload["detail"]["detail_provider"] == "aiddit"
 
 
+def test_run_batch_skips_douyin_video_when_oss_transfer_fails():
+    repo = FakeRepo()
+    adapter = DouyinProviderAdapter()
+
+    def media_stabilizer(**kwargs):
+        assert kwargs["video_fallback_to_source"] is False
+        assert kwargs["video_retry_delays_seconds"] == (5.0, 10.0)
+        return [
+            StabilizedMedia(
+                media_type="video",
+                source_url="https://video.test/761.mp4",
+                cdn_url="",
+                position=1,
+                status="failed",
+                error_message="oss timeout",
+            )
+        ]
+
+    result = run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("douyin",),
+        search_limit=1,
+        display_limit=1,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=media_stabilizer,
+        classifier=lambda **kwargs: (_ for _ in ()).throw(AssertionError("classifier should be skipped")),
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    assert result.done == 1
+    assert repo.media[0].status == "failed"
+    assert repo.media[0].metadata["error_message"] == "oss timeout"
+    assert repo.classifications[0].status == "skipped"
+    assert repo.classifications[0].label == "video_oss_failed"
+    assert repo.classifications[0].result_payload["media_errors"][0]["error_message"] == "oss timeout"
+
+
 def test_run_batch_records_duplicate_search_provider_without_fetching_detail():
     repo = FakeRepo()
     adapter = DouyinProviderAdapter()

tests/test_decode_readers_service.py

@@ -123,6 +123,33 @@ def test_read_item_routes_video_to_video_reader():
     assert result.cards[0].content == "片段知识"
 
 
+def test_video_source_url_without_oss_or_cdn_is_not_decodable():
+    import pytest
+
+    item_id = uuid4()
+    item = CandidateItem(
+        id=item_id,
+        platform="douyin",
+        platform_item_id="v1",
+        content_mode="video_post",
+        title="视频",
+    )
+    media = [
+        MediaAsset(
+            item_id=item_id,
+            media_type="video",
+            source_url="https://video.test/temporary.mp4",
+            position=1,
+        )
+    ]
+
+    post = post_from_candidate_item(item, media)
+
+    assert post.video_urls == []
+    with pytest.raises(UnsupportedPostModeError, match="video_url_missing"):
+        read_item(item, media)
+
+
 def test_read_post_rejects_unsupported_and_missing_video():
     unsupported = Post(
         id="dy_x",

tests/test_media_service.py

@@ -46,3 +46,51 @@ def test_stabilize_media_urls_falls_back_to_source_on_upload_error():
         "https://src.test/c.mp4",
     ]
     assert [row.status for row in rows] == ["ready", "ready"]
+
+
+def test_stabilize_video_retries_then_uses_cdn():
+    calls = []
+    sleeps = []
+
+    def uploader(url: str, media_type: str) -> str:
+        calls.append((url, media_type))
+        if len(calls) < 3:
+            raise RuntimeError("temporary oss failure")
+        return "https://cdn.test/video.mp4"
+
+    rows = stabilize_media_urls(
+        video_urls=["https://src.test/c.mp4"],
+        uploader=uploader,
+        video_fallback_to_source=False,
+        video_retry_delays_seconds=(5, 10),
+        sleep=sleeps.append,
+    )
+
+    assert len(calls) == 3
+    assert sleeps == [5, 10]
+    assert rows[0].cdn_url == "https://cdn.test/video.mp4"
+    assert rows[0].status == "ready"
+
+
+def test_stabilize_video_strict_mode_does_not_fallback_to_source():
+    calls = []
+    sleeps = []
+
+    def uploader(url: str, media_type: str) -> str:
+        calls.append((url, media_type))
+        return url
+
+    rows = stabilize_media_urls(
+        video_urls=["https://src.test/c.mp4"],
+        uploader=uploader,
+        video_fallback_to_source=False,
+        video_retry_delays_seconds=(5, 10),
+        sleep=sleeps.append,
+    )
+
+    assert len(calls) == 3
+    assert sleeps == [5, 10]
+    assert rows[0].source_url == "https://src.test/c.mp4"
+    assert rows[0].cdn_url == ""
+    assert rows[0].status == "failed"
+    assert rows[0].error_message == "oss_upload_returned_source_url"