diff --git a/alembic/versions/081_atom_clip_ai_tags.py b/alembic/versions/081_atom_clip_ai_tags.py deleted file mode 100644 index 09e7a1e69..000000000 --- a/alembic/versions/081_atom_clip_ai_tags.py +++ /dev/null @@ -1,26 +0,0 @@ -"""add ai_tags to asset_atom_clips for #1970 fragment-level AI tagging - -Revision ID: 081_atom_clip_ai_tags -Revises: 080_edit_plan_clips_atom_clip_id -Create Date: 2026-09-18 -""" - -import sqlalchemy as sa - -from alembic import op - -revision = "081_atom_clip_ai_tags" -down_revision = "080_edit_plan_clips_atom_clip_id" -branch_labels = None -depends_on = None - - -def upgrade() -> None: - op.add_column( - "asset_atom_clips", - sa.Column("ai_tags", sa.JSON(), nullable=True), - ) - - -def downgrade() -> None: - op.drop_column("asset_atom_clips", "ai_tags") diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index 874b310d0..e1f2924a5 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -493,6 +493,41 @@ class RenderAdapter: logger.warning("ASR 服务初始化失败，自动字幕将不可用: %s", e) return None + def _resolve_clip_has_text(self, clips: list[Any]) -> list[bool] | None: + """#1970：按源视频片段顺序解析 atom_clip.ai_tags.has_text。 + + 顺序与 UnifiedRenderService 的「非 audio 源片段」口径一致。 + 仅当 atom_clip 存在 ai_tags 字典且 has_text 显式为 False 时标记为 + 无文字（允许 hflip）；atom_clip_id 缺失、ai_tags 未生成、has_text 为 + true/null/非布尔值时一律按有文字处理（保守不翻转）。 + 查询失败时返回 None，渲染层回退到全保守路径。 + """ + video_clips = [c for c in clips if getattr(c, "clip_type", "main") != "audio"] + atom_ids: list[str] = [] + seen: set[str] = set() + for c in video_clips: + atom_id = getattr(c, "atom_clip_id", "") or "" + if atom_id and atom_id not in seen: + seen.add(atom_id) + atom_ids.append(atom_id) + if not atom_ids: + return None + try: + from packages.adapters.sqlalchemy_impl.asset_atom_clip_repository import ( + SQLAlchemyAssetAtomClipRepository, + ) + + atom_clips = SQLAlchemyAssetAtomClipRepository(self._db).find_by_ids(atom_ids) + except Exception as exc: + logger.warning("[render-adapter] atom_clip ai_tags 查询失败，hflip 全量保守处理: %s", exc) + return None + has_text_map: dict[str, bool] = {} + for ac in atom_clips: + ai_tags = getattr(ac, "ai_tags", None) + no_text = isinstance(ai_tags, dict) and ai_tags.get("has_text") is False + has_text_map[ac.id] = not no_text + return [has_text_map.get((getattr(c, "atom_clip_id", "") or ""), True) for c in video_clips] + def _do_render( self, plan: Any, @@ -542,6 +577,7 @@ class RenderAdapter: ) # 4. 执行统一渲染 + clip_has_text = self._resolve_clip_has_text(clips) render_svc = UnifiedRenderService( plan=plan, clips=clips, @@ -552,6 +588,7 @@ class RenderAdapter: bgm_path=bgm_path, asr_service=asr_service, voiceover_audio_path=voiceover_audio_path, + clip_has_text=clip_has_text, ) result = render_svc.render() diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 87880789b..3e85c5727 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -155,6 +155,7 @@ class UnifiedRenderService: asr_service: Any = None, # ASRService 实例，用于自动生成字幕 bgm_path: str | None = None, # BGM 本地文件路径 voiceover_audio_path: str | None = None, # 配音素材库音频本地路径 + clip_has_text: list[bool] | None = None, # 源视频片段是否有文字（来自 atom_clip.ai_tags.has_text） ): self.plan = plan self.clips = clips @@ -167,6 +168,8 @@ class UnifiedRenderService: self.asr_service = asr_service self.bgm_path = bgm_path self.voiceover_audio_path = voiceover_audio_path + # #1970：片段级文字检测（顺序与非 audio 的源视频片段一致）；None 表示无可靠检测，保守不翻转 + self._clip_has_text = clip_has_text self._transition_engine = TransitionEngine(default_duration=transition_duration) self._speed_engine = SpeedEngine() self._asr_timeline_cache: Any = None # ASR 字幕结果缓存，避免重复调用 @@ -186,7 +189,9 @@ class UnifiedRenderService: 种子 hash(generation_task_id + video_index)%10000，同一任务重渲结果一致。 dedup_enabled=False 时返回 None，调用方不注入任何微变换。 - P1 字幕检测：无可靠的片段文字轨道信息，hflip 一律关闭（宁可不翻转）。 + hflip 放开（#1970）：clip_has_text 来自 atom_clip.ai_tags.has_text， + 仅 AI 明确判定无文字的片段可参与 50% 翻转；未打标签 / has_text 为 + true/null 或缺位时一律视为有文字，保持保守不翻转。 """ if self._micro_plan_loaded: return self._micro_plan_cache @@ -200,11 +205,14 @@ class UnifiedRenderService: cfg = self.plan.config or {} task_id = str(cfg.get("generation_task_id", "") or "") video_index = int(cfg.get("video_index", 0) or 0) + # self._clip_has_text 顺序与非 audio 源片段一致； + # None（未提供检测，如内存直渲/旧任务）→ 纯函数层按全有文字保守处理； + # 列表短于片段数时缺位片段同样按有文字处理 self._micro_plan_cache = build_micro_transform_plan( task_id, video_index, clip_count, - clip_has_text=None, # P1 保守策略：全部按有文字处理，不翻转 + clip_has_text=self._clip_has_text, enable_bgm_offset=bool(cfg.get("bgm")), ) except Exception as e: diff --git a/tests/unit/test_1970_hflip_ai_tags.py b/tests/unit/test_1970_hflip_ai_tags.py new file mode 100644 index 000000000..8d19e8dfd --- /dev/null +++ b/tests/unit/test_1970_hflip_ai_tags.py @@ -0,0 +1,185 @@ +"""#1970 hflip 放开（has_text 来自 atom_clip.ai_tags）端到端参数链路测试。 + +覆盖： +1. UnifiedRenderService 传入 clip_has_text 后微变换计划的翻转门控； +2. RenderAdapter._resolve_clip_has_text 按 atom_clip.ai_tags.has_text + 解析布尔列表（显式 False 才可翻转，其余保守），失败回退 None； +3. 纯函数层在「混合有/无文字」列表下的行为（顺序对齐）。 +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest +from video_processing.micro_transform_pure import build_micro_transform_plan + + +def _make_service(plan_config: dict | None = None, clip_has_text=None): + from video_processing.unified_render_service import UnifiedRenderService + + svc = object.__new__(UnifiedRenderService) + svc.plan = MagicMock() + svc.plan.config = plan_config or {} + svc.plan.id = "plan-1" + svc.plan.clips = [] + svc._micro_plan_cache = None + svc._micro_plan_loaded = False + svc._clip_has_text = clip_has_text + return svc + + +def _clip(clip_id: str, atom_clip_id: str = "", clip_type: str = "main"): + return SimpleNamespace(id=clip_id, atom_clip_id=atom_clip_id, clip_type=clip_type) + + +def _atom(clip_id: str, ai_tags): + return SimpleNamespace(id=clip_id, ai_tags=ai_tags) + + +class TestServiceClipHasText: + def test_none_stays_conservative(self): + # 未注入检测列表：所有片段一律不翻转 + svc = _make_service({"generation_task_id": "t1"}, clip_has_text=None) + plan = svc._get_micro_transform_plan(30) + assert plan is not None + assert all(c.has_text for c in plan.clips) + assert all(not c.hflip for c in plan.clips) + + def test_explicit_no_text_allows_hflip(self): + # AI 明确判定无文字：允许参与 50% 翻转（40 段应至少出现一些翻转） + svc = _make_service({"generation_task_id": "t-allow"}, clip_has_text=[False] * 40) + plan = svc._get_micro_transform_plan(40) + assert plan is not None + assert all(not c.has_text for c in plan.clips) + assert any(c.hflip for c in plan.clips) + assert all(not c.hflip or not c.has_text for c in plan.clips) + + def test_all_text_never_flips(self): + svc = _make_service({"generation_task_id": "t-text"}, clip_has_text=[True] * 40) + plan = svc._get_micro_transform_plan(40) + assert all(c.has_text for c in plan.clips) + assert all(not c.hflip for c in plan.clips) + + def test_mixed_order_alignment(self): + # 仅第 0、2 个片段无文字；has_text 标记必须与片段序号严格对齐 + svc = _make_service({"generation_task_id": "t-mix"}, clip_has_text=[False, True, False, True]) + plan = svc._get_micro_transform_plan(4) + assert [c.has_text for c in plan.clips] == [False, True, False, True] + assert all(not plan.clips[i].hflip for i in (1, 3)) + for i in (0, 2): + # 无文字片段的翻转由 50% 种子决定，但允许翻转（不强制一定翻） + assert plan.clips[i].has_text is False + + def test_list_shorter_than_clips_missing_are_conservative(self): + # 列表短于片段数：缺位片段按有文字处理 + svc = _make_service({"generation_task_id": "t-short"}, clip_has_text=[False]) + plan = svc._get_micro_transform_plan(3) + assert [c.has_text for c in plan.clips] == [False, True, True] + assert not plan.clips[1].hflip and not plan.clips[2].hflip + + def test_plan_reproducible_with_real_list(self): + cfg = {"generation_task_id": "task-x", "video_index": 1} + flags = [False, True, False, False, True] + p1 = _make_service(cfg, clip_has_text=flags)._get_micro_transform_plan(5) + p2 = _make_service(dict(cfg), clip_has_text=list(flags))._get_micro_transform_plan(5) + assert [c.hflip for c in p1.clips] == [c.hflip for c in p2.clips] + + +class TestPureMixedFlags: + def test_pure_function_mixed_flags(self): + plan = build_micro_transform_plan("seed-1", 0, 4, clip_has_text=[False, True, False, True]) + assert [c.has_text for c in plan.clips] == [False, True, False, True] + # 有文字片段绝不翻转 + assert not plan.clips[1].hflip and not plan.clips[3].hflip + + +class TestResolveClipHasText: + def _adapter(self): + from video_processing.render_adapter import RenderAdapter + + return RenderAdapter(MagicMock()) + + def test_no_atom_ids_returns_none(self): + adapter = self._adapter() + clips = [_clip("c1", ""), _clip("c2", "")] + assert adapter._resolve_clip_has_text(clips) is None + + def test_explicit_false_only_maps_to_false(self): + adapter = self._adapter() + clips = [ + _clip("c1", "a1"), + _clip("c2", "a2"), + _clip("c3", "a3"), + _clip("c4", "a4"), + _clip("c5", "a5"), + ] + atoms = [ + _atom("a1", {"has_text": False}), # 明确无文字 → False + _atom("a2", {"has_text": True}), # 有文字 + _atom("a3", None), # 标签未生成 + _atom("a4", {"scene": ["工厂"]}), # has_text 缺失（null） + _atom("a5", {"has_text": "false"}), # 非布尔 → 保守 + ] + with patch( + "packages.adapters.sqlalchemy_impl.asset_atom_clip_repository." + "SQLAlchemyAssetAtomClipRepository.find_by_ids", + return_value=atoms, + ): + result = adapter._resolve_clip_has_text(clips) + assert result == [False, True, True, True, True] + + def test_audio_clips_excluded_and_order_kept(self): + adapter = self._adapter() + clips = [ + _clip("c1", "a1", clip_type="main"), + _clip("bgm", "", clip_type="audio"), + _clip("c2", "a2", clip_type="pip"), + ] + atoms = [ + _atom("a1", {"has_text": False}), + _atom("a2", {"has_text": False}), + ] + with patch( + "packages.adapters.sqlalchemy_impl.asset_atom_clip_repository." + "SQLAlchemyAssetAtomClipRepository.find_by_ids", + return_value=atoms, + ) as mock_find: + result = adapter._resolve_clip_has_text(clips) + # 只查非 audio 片段的 atom id，且顺序为 main → pip + assert mock_find.call_args.args[0] == ["a1", "a2"] + assert result == [False, False] + + def test_missing_atom_record_defaults_true(self): + adapter = self._adapter() + clips = [_clip("c1", "a1"), _clip("c2", "a2")] + with patch( + "packages.adapters.sqlalchemy_impl.asset_atom_clip_repository." + "SQLAlchemyAssetAtomClipRepository.find_by_ids", + return_value=[_atom("a1", {"has_text": False})], # a2 查不到 + ): + result = adapter._resolve_clip_has_text(clips) + assert result == [False, True] + + def test_query_failure_returns_none(self): + adapter = self._adapter() + clips = [_clip("c1", "a1")] + with patch( + "packages.adapters.sqlalchemy_impl.asset_atom_clip_repository." + "SQLAlchemyAssetAtomClipRepository.find_by_ids", + side_effect=RuntimeError("db down"), + ): + assert adapter._resolve_clip_has_text(clips) is None + + def test_duplicate_atom_ids_queried_once(self): + adapter = self._adapter() + clips = [_clip("c1", "a1"), _clip("c2", "a1")] + with patch( + "packages.adapters.sqlalchemy_impl.asset_atom_clip_repository." + "SQLAlchemyAssetAtomClipRepository.find_by_ids", + return_value=[_atom("a1", {"has_text": False})], + ) as mock_find: + result = adapter._resolve_clip_has_text(clips) + assert mock_find.call_args.args[0] == ["a1"] + assert result == [False, False] diff --git a/tests/unit/test_1970_micro_transform_render.py b/tests/unit/test_1970_micro_transform_render.py index baaa04039..b27bbfefb 100644 --- a/tests/unit/test_1970_micro_transform_render.py +++ b/tests/unit/test_1970_micro_transform_render.py @@ -22,6 +22,7 @@ def _make_service(plan_config: dict | None = None, clips=None): svc.plan.clips = clips or [] svc._micro_plan_cache = None svc._micro_plan_loaded = False + svc._clip_has_text = None return svc @@ -159,6 +160,7 @@ class TestStreamCopyGate: svc.clips = [source] svc._micro_plan_cache = None svc._micro_plan_loaded = False + svc._clip_has_text = None resolved = ResolvedClip( clip_id="c1", asset_id="a1",