alembic/versions/081_atom_clip_ai_tags.py

@@ -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")

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()
 

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:

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]

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",