apps/worker/worker_app/tasks/atom_clip_tagging.py

@@ -25,11 +25,14 @@ logger = get_task_logger(__name__)
 
 
 @celery_app.task(name="worker.tag_atom_clip", bind=True, max_retries=2, default_retry_delay=10)
-def tag_atom_clip_task(self, atom_clip_id: str) -> dict:
+def tag_atom_clip_task(self, atom_clip_id: str, force: bool = False) -> dict:
     """为单个原子片段生成 AI 标签.
 
     Args:
         atom_clip_id: 原子片段 ID。
+        force: True 时允许覆盖只有 inherited_tags 的降级记录
+            （视觉 API 曾失败写入的占位标签，#1970）。
+            已有完整标签（含 has_text）始终跳过，保证幂等。
 
     Returns:
         任务结果 dict：status / clip_id / ai_tags（部分字段）。
@@ -43,9 +46,11 @@ def tag_atom_clip_task(self, atom_clip_id: str) -> dict:
         if clip is None:
             return {"status": "skipped", "reason": "clip not found", "clip_id": atom_clip_id}
 
-        # 已有标签则跳过（幂等）
+        # 已有完整标签则跳过（幂等）；force 仅放行缺失 has_text 的降级记录
         if clip.ai_tags is not None:
-            return {"status": "skipped", "reason": "already tagged", "clip_id": atom_clip_id}
+            has_real_tags = isinstance(clip.ai_tags, dict) and "has_text" in clip.ai_tags
+            if has_real_tags or not force:
+                return {"status": "skipped", "reason": "already tagged", "clip_id": atom_clip_id}
 
         # 获取素材信息
         asset = asset_repo.find_by_id(clip.asset_id)

apps/worker/worker_app/tasks/backfill_atom_clip_tags.py

@@ -30,6 +30,7 @@ def backfill_atom_clip_tags(
     batch_size: int = DEFAULT_BATCH_SIZE,
     batch_interval: int = DEFAULT_BATCH_INTERVAL,
     max_clips: int = 0,
+    force: bool = False,
 ) -> dict:
     """批量回填未打标的 atom_clips.
 
@@ -37,6 +38,8 @@ def backfill_atom_clip_tags(
         batch_size: 每批处理数量，默认 10。
         batch_interval: 每批间隔秒数，默认 5。
         max_clips: 最大处理总数，0 表示不限。
+        force: True 时连同只有 inherited_tags 的降级记录一起强制重打
+            （视觉 API 曾失败、DOUBAO_VISION_MODEL 修复后重跑用，#1970）。
 
     Returns:
         任务结果 dict：total_submitted / batches。
@@ -52,7 +55,7 @@ def backfill_atom_clip_tags(
             remaining = max_clips - total_submitted if max_clips > 0 else batch_size
             fetch_limit = min(batch_size, remaining) if max_clips > 0 else batch_size
 
-            untagged = atom_repo.find_untagged(limit=fetch_limit)
+            untagged = atom_repo.find_untagged(limit=fetch_limit, include_downgraded=force)
             if not untagged:
                 break
 
@@ -62,6 +65,7 @@ def backfill_atom_clip_tags(
                     celery_app.send_task(
                         "worker.tag_atom_clip",
                         args=[clip.id],
+                        kwargs={"force": force},
                     )
                     total_submitted += 1
                 except Exception as e:

packages/adapters/sqlalchemy_impl/asset_atom_clip_repository.py

@@ -91,15 +91,21 @@ class SQLAlchemyAssetAtomClipRepository:
         self.session.commit()
         return count > 0
 
-    def find_untagged(self, limit: int = 100) -> list[AssetAtomClip]:
-        """查找 ai_tags IS NULL 的片段，用于回填."""
-        models = (
-            self.session.query(AssetAtomClipModel)
-            .filter(AssetAtomClipModel.ai_tags.is_(None))
-            .order_by(AssetAtomClipModel.created_at.asc())
-            .limit(limit)
-            .all()
-        )
+    def find_untagged(self, limit: int = 100, include_downgraded: bool = False) -> list[AssetAtomClip]:
+        """查找未完成 AI 打标的片段，用于回填.
+
+        默认仅匹配 ai_tags IS NULL；include_downgraded=True 时额外包含
+        只有 inherited_tags 的降级记录（视觉 API 失败时写入，无 has_text 字段），
+        供强制回填（#1970 force backfill）使用。
+        """
+        query = self.session.query(AssetAtomClipModel)
+        if include_downgraded:
+            # as_string() → JSON/JSONB ->> 取值；NULL 记录或缺 has_text 键
+            # （降级记录）均为 NULL，has_text 为 true/false 的完整记录被排除
+            query = query.filter(AssetAtomClipModel.ai_tags["has_text"].as_string().is_(None))
+        else:
+            query = query.filter(AssetAtomClipModel.ai_tags.is_(None))
+        models = query.order_by(AssetAtomClipModel.created_at.asc()).limit(limit).all()
         return [self._to_domain(m) for m in models]
 
     def _to_model(self, clip: AssetAtomClip) -> AssetAtomClipModel:

tests/unit/test_1970_force_backfill_tags.py

@@ -0,0 +1,347 @@
+"""#1970 force 回填降级 AI 标签记录的回归测试。
+
+背景：DOUBAO_VISION_MODEL 未配置时，tagger 降级写入
+{"inherited_tags": [...]}（非 NULL），默认 backfill 只捞 ai_tags IS NULL，
+这批记录永远不会重打。force=True 时应纳入降级记录，并在打标成功后覆盖。
+
+覆盖：
+- find_untagged(include_downgraded) 的 SQL 过滤（SQLite 验证跨库 JSON 取值）
+- tag_atom_clip_task 的 force 跳过/放行/覆盖逻辑
+- backfill_atom_clip_tags(force=True) 给 tag 任务传 kwargs={"force": True}
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from types import SimpleNamespace
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from packages.adapters.sqlalchemy_impl.asset_atom_clip_repository import (
+    SQLAlchemyAssetAtomClipRepository,
+)
+from packages.adapters.sqlalchemy_impl.models import AssetAtomClipModel
+
+# ── 仓储层：find_untagged 过滤 ─────────────────────────────────────────────
+
+
+@pytest.fixture
+def repo_session():
+    engine = create_engine("sqlite:///:memory:")
+    AssetAtomClipModel.__table__.create(engine)
+    SessionTest = sessionmaker(bind=engine)
+    session = SessionTest()
+    now = datetime.now(UTC)
+    session.add_all(
+        [
+            AssetAtomClipModel(
+                id="c-null",
+                asset_id="a1",
+                start_time=0,
+                end_time=1,
+                duration=1,
+                clip_index=0,
+                tags=[],
+                ai_tags=None,
+                created_at=now,
+            ),
+            AssetAtomClipModel(
+                id="c-downgraded-empty",
+                asset_id="a1",
+                start_time=1,
+                end_time=2,
+                duration=1,
+                clip_index=1,
+                tags=[],
+                ai_tags={"inherited_tags": []},
+                created_at=now,
+            ),
+            AssetAtomClipModel(
+                id="c-downgraded-tags",
+                asset_id="a1",
+                start_time=2,
+                end_time=3,
+                duration=1,
+                clip_index=2,
+                tags=[],
+                ai_tags={"inherited_tags": ["口播"]},
+                created_at=now,
+            ),
+            AssetAtomClipModel(
+                id="c-tagged-true",
+                asset_id="a1",
+                start_time=3,
+                end_time=4,
+                duration=1,
+                clip_index=3,
+                tags=[],
+                ai_tags={"has_text": True, "scene": ["室内"], "inherited_tags": []},
+                created_at=now,
+            ),
+            AssetAtomClipModel(
+                id="c-tagged-false",
+                asset_id="a1",
+                start_time=4,
+                end_time=5,
+                duration=1,
+                clip_index=4,
+                tags=[],
+                ai_tags={"has_text": False, "inherited_tags": ["风景"]},
+                created_at=now,
+            ),
+        ]
+    )
+    session.commit()
+    # SQLAlchemy JSON 在 SQLite 下把 None 序列化为 'null' 字符串，
+    # 而生产 PostgreSQL 存的是真 SQL NULL；用原生 SQL 对齐生产语义。
+    from sqlalchemy import text
+
+    session.execute(text("UPDATE asset_atom_clips SET ai_tags = NULL WHERE id = 'c-null'"))
+    session.commit()
+    yield session
+    session.close()
+
+
+def test_find_untagged_default_only_null(repo_session):
+    repo = SQLAlchemyAssetAtomClipRepository(repo_session)
+    ids = {c.id for c in repo.find_untagged(limit=100)}
+    assert ids == {"c-null"}
+
+
+def test_find_untagged_include_downgraded(repo_session):
+    repo = SQLAlchemyAssetAtomClipRepository(repo_session)
+    ids = {c.id for c in repo.find_untagged(limit=100, include_downgraded=True)}
+    # NULL + 两条降级记录；含 has_text=true/false 的完整记录都排除
+    assert ids == {"c-null", "c-downgraded-empty", "c-downgraded-tags"}
+
+
+# ── 任务层：tag_atom_clip_task 的 force 语义 ───────────────────────────────
+
+
+def _import_tag_task_module():
+    from worker_app.tasks import atom_clip_tagging as mod
+
+    return mod
+
+
+def _call_tag_task(mod, clip_id, force):
+    """直接调用任务，兼容两种环境。
+
+    全量收集时旧测试向 sys.modules 注入 celery_app MagicMock（其 task
+    装饰器原样返回裸函数），此时是普通函数需显式传 self=None；
+    正常 Celery 环境下属性是 Task 代理对象（非普通 function），
+    已绑定 self，按业务签名直接调用即可。
+    """
+    import inspect
+
+    obj = mod.tag_atom_clip_task
+    if inspect.isfunction(obj):
+        return obj(None, clip_id, force=force)
+    return obj(clip_id, force=force)
+
+
+def test_tag_task_skips_downgraded_without_force(monkeypatch):
+    mod = _import_tag_task_module()
+    monkeypatch.setattr(
+        mod,
+        "SessionLocal",
+        lambda: SimpleNamespace(
+            rollback=lambda: None,
+            close=lambda: None,
+        ),
+    )
+
+    class _Repo:
+        def __init__(self, db):
+            pass
+
+        def find_by_id(self, clip_id):
+            return SimpleNamespace(
+                id=clip_id,
+                ai_tags={"inherited_tags": []},
+            )
+
+    monkeypatch.setattr(mod, "SQLAlchemyAssetAtomClipRepository", _Repo)
+
+    result = _call_tag_task(mod, "clip-downgraded", force=False)
+    assert result["status"] == "skipped"
+    assert result["reason"] == "already tagged"
+
+
+def test_tag_task_force_retags_downgraded_and_overwrites(monkeypatch):
+    mod = _import_tag_task_module()
+    updated: dict[str, dict] = {}
+
+    class _FakeSession:
+        def rollback(self):
+            pass
+
+        def close(self):
+            pass
+
+    monkeypatch.setattr(mod, "SessionLocal", _FakeSession)
+
+    class _AtomRepo:
+        def __init__(self, db):
+            pass
+
+        def find_by_id(self, clip_id):
+            return SimpleNamespace(
+                id=clip_id,
+                asset_id="asset-1",
+                start_time=0.0,
+                end_time=2.0,
+                tags=["旧标签"],
+                ai_tags={"inherited_tags": []},
+            )
+
+        def update_ai_tags(self, clip_id, ai_tags):
+            updated[clip_id] = ai_tags
+
+    class _AssetRepo:
+        def __init__(self, db):
+            pass
+
+        def find_by_id(self, asset_id):
+            return SimpleNamespace(id=asset_id, storage_key="k/video.mp4")
+
+    monkeypatch.setattr(mod, "SQLAlchemyAssetAtomClipRepository", _AtomRepo)
+    monkeypatch.setattr(mod, "SQLAlchemyAssetRepository", _AssetRepo)
+
+    class _Storage:
+        def get_download_url(self, key, expires_seconds=3600):
+            return "https://example.com/signed.mp4"
+
+    monkeypatch.setattr(mod, "get_shared_storage_service", lambda: _Storage())
+    monkeypatch.setattr(mod, "get_doubao_client", lambda: object())
+    monkeypatch.setattr(mod, "get_mediakit_client", lambda: None)
+
+    new_tags = {
+        "scene": ["室内"],
+        "objects": ["人物"],
+        "action": ["说话"],
+        "shot": "中景",
+        "has_text": True,
+        "inherited_tags": ["旧标签"],
+    }
+    monkeypatch.setattr(mod, "tag_atom_clip", lambda **kw: new_tags)
+
+    result = _call_tag_task(mod, "clip-downgraded", force=True)
+    assert result["status"] == "completed"
+    assert result["has_ai_tags"] is True
+    assert updated["clip-downgraded"] == new_tags
+
+
+def test_tag_task_force_still_skips_complete_tags(monkeypatch):
+    mod = _import_tag_task_module()
+    monkeypatch.setattr(
+        mod,
+        "SessionLocal",
+        lambda: SimpleNamespace(rollback=lambda: None, close=lambda: None),
+    )
+
+    class _Repo:
+        def __init__(self, db):
+            pass
+
+        def find_by_id(self, clip_id):
+            return SimpleNamespace(
+                id=clip_id,
+                ai_tags={"has_text": False, "inherited_tags": []},
+            )
+
+    monkeypatch.setattr(mod, "SQLAlchemyAssetAtomClipRepository", _Repo)
+
+    result = _call_tag_task(mod, "clip-complete", force=True)
+    assert result["status"] == "skipped"
+    assert result["reason"] == "already tagged"
+
+
+# ── backfill 任务：force 透传到 send_task ──────────────────────────────────
+
+
+def test_backfill_force_passes_kwarg(monkeypatch):
+    from worker_app.tasks import backfill_atom_clip_tags as bmod
+
+    sent: list[tuple] = []
+
+    class _FakeSession:
+        def close(self):
+            pass
+
+    monkeypatch.setattr(bmod, "SessionLocal", _FakeSession)
+
+    class _AtomRepo:
+        def __init__(self, db):
+            self.calls: list[bool] = []
+
+        def find_untagged(self, limit, include_downgraded=False):
+            self.calls.append(include_downgraded)
+            # 第一批返回一条降级记录，第二批返回空结束循环
+            if len(self.calls) == 1:
+                return [SimpleNamespace(id="clip-1")]
+            return []
+
+    repo_holder = {}
+
+    def _repo_factory(db):
+        repo = _AtomRepo(db)
+        repo_holder["repo"] = repo
+        return repo
+
+    monkeypatch.setattr(bmod, "SQLAlchemyAssetAtomClipRepository", _repo_factory)
+
+    def _send_task(name, args=None, kwargs=None):
+        sent.append((name, args, kwargs))
+
+    monkeypatch.setattr(bmod.celery_app, "send_task", _send_task)
+
+    result = bmod.backfill_atom_clip_tags(batch_size=10, batch_interval=0, force=True)
+
+    assert result["status"] == "completed"
+    assert result["total_submitted"] == 1
+    assert repo_holder["repo"].calls == [True, True]
+    assert sent == [
+        ("worker.tag_atom_clip", ["clip-1"], {"force": True}),
+    ]
+
+
+def test_backfill_default_does_not_force(monkeypatch):
+    from worker_app.tasks import backfill_atom_clip_tags as bmod
+
+    sent_kwargs: list[dict | None] = []
+
+    class _FakeSession:
+        def close(self):
+            pass
+
+    monkeypatch.setattr(bmod, "SessionLocal", _FakeSession)
+
+    class _AtomRepo:
+        def __init__(self, db):
+            self.calls: list[bool] = []
+
+        def find_untagged(self, limit, include_downgraded=False):
+            self.calls.append(include_downgraded)
+            return [SimpleNamespace(id="clip-null")] if self.calls == [False] else []
+
+    holder = {}
+
+    def _repo_factory(db):
+        holder["repo"] = _AtomRepo(db)
+        return holder["repo"]
+
+    monkeypatch.setattr(bmod, "SQLAlchemyAssetAtomClipRepository", _repo_factory)
+    monkeypatch.setattr(
+        bmod.celery_app,
+        "send_task",
+        lambda name, args=None, kwargs=None: sent_kwargs.append(kwargs),
+    )
+
+    result = bmod.backfill_atom_clip_tags(batch_size=10, batch_interval=0)
+
+    assert result["total_submitted"] == 1
+    assert holder["repo"].calls == [False, False]
+    assert sent_kwargs == [{"force": False}]