From 9558969a31edd98544c578df4e3b55f88852973e Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sat, 19 Sep 2026 10:41:59 +0800 Subject: [PATCH] =?UTF-8?q?feat(worker):=20#1970=20AI=20=E6=A0=87=E7=AD=BE?= =?UTF-8?q?=20backfill=20=E6=94=AF=E6=8C=81=20force=20=E9=87=8D=E6=89=93?= =?UTF-8?q?=E9=99=8D=E7=BA=A7=E8=AE=B0=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DOUBAO_VISION_MODEL 修复后，此前视觉 API 失败写入的 {'inherited_tags': [...]} 降级记录（非 NULL）不会被默认 backfill 捞起， tag 任务也按幂等跳过。 - find_untagged 加 include_downgraded：用 ai_tags['has_text'].as_string() IS NULL 同时匹配 NULL 与缺 has_text 键的降级记录（json/jsonb 通用， has_text=true/false 的完整记录排除） - tag_atom_clip_task / backfill_atom_clip_tags 加 force 参数， backfill 透传 kwargs={'force': ...} - 7 个回归测试（任务层测试兼容旧测试 sys.modules 注入 MagicMock 的污染） --- .../worker_app/tasks/atom_clip_tagging.py | 11 +- .../tasks/backfill_atom_clip_tags.py | 6 +- .../asset_atom_clip_repository.py | 24 +- tests/unit/test_1970_force_backfill_tags.py | 347 ++++++++++++++++++ 4 files changed, 375 insertions(+), 13 deletions(-) create mode 100644 tests/unit/test_1970_force_backfill_tags.py diff --git a/apps/worker/worker_app/tasks/atom_clip_tagging.py b/apps/worker/worker_app/tasks/atom_clip_tagging.py index e03ecb5c4..f3dc24ca0 100644 --- a/apps/worker/worker_app/tasks/atom_clip_tagging.py +++ b/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) diff --git a/apps/worker/worker_app/tasks/backfill_atom_clip_tags.py b/apps/worker/worker_app/tasks/backfill_atom_clip_tags.py index 49399ae4b..3d8d76048 100644 --- a/apps/worker/worker_app/tasks/backfill_atom_clip_tags.py +++ b/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: diff --git a/packages/adapters/sqlalchemy_impl/asset_atom_clip_repository.py b/packages/adapters/sqlalchemy_impl/asset_atom_clip_repository.py index 8133149f7..c1f09abdd 100644 --- a/packages/adapters/sqlalchemy_impl/asset_atom_clip_repository.py +++ b/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: diff --git a/tests/unit/test_1970_force_backfill_tags.py b/tests/unit/test_1970_force_backfill_tags.py new file mode 100644 index 000000000..47cdfddc7 --- /dev/null +++ b/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}] -- 2.54.0