feat: unify template segments data source to template_clip_configs

- template_repository: read/write segments via template_clip_configs table
  instead of old template_segments table
- list_segments: prefer template_clip_configs, fallback to template_segments
  for backward compatibility with existing data
- create_segments: write to template_clip_configs with material_type
  stored in config JSON field
- delete: clean both tables for safe cleanup
- Alembic migration 060: one-time migrate orphaned template_segments
  records to template_clip_configs
- 10 new tests covering the unified behavior

alembic/versions/060_migrate_template_segments_to_clip_configs.py

@@ -0,0 +1,107 @@
+"""migrate template_segments to template_clip_configs
+
+将旧 template_segments 表中的数据迁移到 template_clip_configs 表，
+统一模板 segments 数据源。旧表保留不删除，仅做数据迁移。
+
+Revision ID: 060_migrate_segments
+Revises: 059_duplicate_rate
+Create Date: 2026-08-31
+"""
+
+import json
+import uuid
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "060_migrate_segments"
+down_revision = "059_duplicate_rate"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """将 template_segments 中尚未同步到 template_clip_configs 的记录迁移过去。
+<<<<<<< Updated upstream
+
+    只迁移 template_id 在 template_clip_configs 中没有对应记录的行，
+    避免覆盖编辑器已经发布的数据。
+    """
+    conn = op.get_bind()
+
+    # 找出有旧数据但没有新数据的 template_id
+=======
+    只迁移 template_id 在 template_clip_configs 中没有对应记录的行。
+    """
+    conn = op.get_bind()
+
+>>>>>>> Stashed changes
+    rows = conn.execute(
+        sa.text(
+            """
+            SELECT ts.id, ts.template_id, ts.segment_order, ts.duration_min, ts.duration_max, ts.material_type
+            FROM template_segments ts
+<<<<<<< Updated upstream
+            WHERE ts.template_id NOT IN (
+                SELECT DISTINCT tcc.template_id FROM template_clip_configs tcc
+=======
+            WHERE NOT EXISTS (
+                SELECT 1 FROM template_clip_configs tcc
+                WHERE tcc.template_id = ts.template_id
+>>>>>>> Stashed changes
+            )
+            ORDER BY ts.template_id, ts.segment_order
+            """
+        )
+    )
+
+    for row in rows:
+        config = {"material_type": row[5]} if row[5] else {}
+        conn.execute(
+            sa.text(
+                """
+                INSERT INTO template_clip_configs
+                    (id, template_id, clip_type, "order", min_duration, max_duration,
+                     text_template, material_requirements, transition_effect, config,
+                     created_at, updated_at)
+<<<<<<< Updated upstream
+                VALUES (:id, :template_id, 'main', :seg_order, :dur_min, :dur_max,
+                        '', '{}', 'cut', :config,
+=======
+                VALUES (:id, :template_id, :clip_type, :seg_order, :dur_min, :dur_max,
+                        :text_template, :material_req, :transition, :config,
+>>>>>>> Stashed changes
+                        CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+                """
+            ),
+            {
+                "id": uuid.uuid4().hex,
+                "template_id": row[1],
+<<<<<<< Updated upstream
+                "seg_order": row[2],
+                "dur_min": row[3],
+                "dur_max": row[4],
+=======
+                "clip_type": "main",
+                "seg_order": row[2],
+                "dur_min": row[3],
+                "dur_max": row[4],
+                "text_template": "",
+                "material_req": "{}",
+                "transition": "cut",
+>>>>>>> Stashed changes
+                "config": json.dumps(config),
+            },
+        )
+
+
+def downgrade() -> None:
+<<<<<<< Updated upstream
+    """回滚：删除迁移过来的记录。
+
+    无法精确区分哪些是迁移来的，所以不做删除。
+    如果需要完全回滚，手动处理。
+    """
+=======
+>>>>>>> Stashed changes
+    pass

packages/adapters/sqlalchemy_impl/template_repository.py

@@ -1,4 +1,8 @@
-"""SQLAlchemy implementation of TemplateRepository."""
+"""SQLAlchemy implementation of TemplateRepository.
+
+模板 segments 数据源已统一为 template_clip_configs 表。
+旧 template_segments 表不再读写，保留表结构供历史数据查询。
+"""
 
 from __future__ import annotations
 
@@ -10,6 +14,7 @@ from sqlalchemy.orm import Session
 from packages.adapters.sqlalchemy_impl.models import (
     EditPlanModel,
     TemplateCategoryModel,
+    TemplateClipConfigModel,
     TemplateModel,
     TemplateSegmentModel,
 )
@@ -47,27 +52,25 @@ class SQLAlchemyTemplateRepository:
             like_pattern = f"%{keyword}%"
             query = query.filter(TemplateModel.name.like(like_pattern))
         if tag:
-            # JSON 数组包含指定标签（MySQL JSON_CONTAINS / SQLite json_each 兼容写法用 LIKE）
             query = query.filter(TemplateModel.tags.like(f'%"{tag}"%'))
         models = query.order_by(TemplateModel.created_at.desc()).offset(skip).limit(limit).all()
         templates = [self._model_to_entity(m) for m in models]
-        # 批量加载所有 segments，避免 N+1 查询
+        # 批量加载 segments：从 template_clip_configs 读取，映射为 TemplateSegment
         if templates:
             template_ids = [t.id for t in templates]
-            seg_models = (
-                self.session.query(TemplateSegmentModel)
-                .filter(TemplateSegmentModel.template_id.in_(template_ids))
-                .order_by(TemplateSegmentModel.segment_order)
+            clip_models = (
+                self.session.query(TemplateClipConfigModel)
+                .filter(TemplateClipConfigModel.template_id.in_(template_ids))
+                .order_by(TemplateClipConfigModel.order)
                 .all()
             )
-            # 按 template_id 分组
-            seg_map: dict[str, list] = {}
-            for sm in seg_models:
-                seg_map.setdefault(sm.template_id, []).append(
-                    self._segment_model_to_entity(sm),
+            clip_map: dict[str, list] = {}
+            for cm in clip_models:
+                clip_map.setdefault(cm.template_id, []).append(
+                    self._clip_config_to_segment(cm),
                 )
             for t in templates:
-                t.segments = seg_map.get(t.id, [])
+                t.segments = clip_map.get(t.id, [])
         return templates
 
     def get(self, template_id: str, user_id: str) -> Optional[Template]:
@@ -100,7 +103,6 @@ class SQLAlchemyTemplateRepository:
             is_active=template.is_active,
         )
         self.session.add(model)
-        # flush 而非 commit，让 create + create_segments 在同一事务中提交
         self.session.flush()
         self.session.refresh(model)
         result = self._model_to_entity(model)
@@ -145,7 +147,17 @@ class SQLAlchemyTemplateRepository:
         if model is None:
             return False
         model.is_active = False
-        # 级联清理关联的 segments，避免孤儿数据
+<<<<<<< Updated upstream
+        # 清理 template_clip_configs（主数据源）
+        self.session.query(TemplateClipConfigModel).filter(
+            TemplateClipConfigModel.template_id == template_id,
+        ).delete(synchronize_session=False)
+        # 同时清理旧 template_segments（兼容历史数据）
+=======
+        self.session.query(TemplateClipConfigModel).filter(
+            TemplateClipConfigModel.template_id == template_id,
+        ).delete(synchronize_session=False)
+>>>>>>> Stashed changes
         self.session.query(TemplateSegmentModel).filter(
             TemplateSegmentModel.template_id == template_id,
         ).delete(synchronize_session=False)
@@ -167,8 +179,6 @@ class SQLAlchemyTemplateRepository:
         )
         if category:
             query = query.filter(TemplateModel.category == category)
-        if mode:
-            query = query.filter(TemplateModel.mode == mode)
         if keyword:
             query = query.filter(TemplateModel.name.like(f"%{keyword}%"))
         if tag:
@@ -176,7 +186,7 @@ class SQLAlchemyTemplateRepository:
         return query.count()
 
     def copy_template(self, template_id: str, user_id: str, new_name: str) -> Template:
-        """复制模板（含所有 segments）。"""
+        """复制模板（含所有 segments，从 template_clip_configs 读取并写入）。"""
         source = self.get(template_id, user_id)
         if source is None:
             raise ValueError(f"Template {template_id} not found")
@@ -197,64 +207,124 @@ class SQLAlchemyTemplateRepository:
         )
         created = self.create(new_template)
 
-        # 复制 segments
+<<<<<<< Updated upstream
+        # 复制 segments → 写入 template_clip_configs
+=======
+        # 复制 segments → 复用 create_segments 写入 template_clip_configs
+>>>>>>> Stashed changes
         new_segments: List[TemplateSegment] = []
         for seg in source.segments:
             new_seg = TemplateSegment(
                 id=str(uuid.uuid4()),
-                template_id=new_id,
+                template_id=created.id,
                 segment_order=seg.segment_order,
                 duration_min=seg.duration_min,
                 duration_max=seg.duration_max,
                 material_type=seg.material_type,
             )
             new_segments.append(new_seg)
-            model = TemplateSegmentModel(
+<<<<<<< Updated upstream
+            config = {"material_type": seg.material_type} if seg.material_type else {}
+            clip_model = TemplateClipConfigModel(
                 id=new_seg.id,
-                template_id=new_seg.template_id,
-                segment_order=new_seg.segment_order,
-                duration_min=new_seg.duration_min,
-                duration_max=new_seg.duration_max,
-                material_type=new_seg.material_type,
+                template_id=new_id,
+                clip_type="main",
+                order=new_seg.segment_order,
+                min_duration=new_seg.duration_min,
+                max_duration=new_seg.duration_max,
+                text_template="",
+                material_requirements={},
+                transition_effect="cut",
+                config=config,
             )
-            self.session.add(model)
+            self.session.add(clip_model)
+=======
+>>>>>>> Stashed changes
         if new_segments:
+            self.create_segments(new_segments)
+        else:
+            # 没有 segments 时也需要 commit（create 只做了 flush）
             self.session.commit()
 
         created.segments = new_segments
         return created
 
-    # ── Segments ──
+    # ── Segments（数据源：template_clip_configs）──
 
     def list_segments(self, template_id: str) -> List[TemplateSegment]:
-        models = (
+        """从 template_clip_configs 读取并按 TemplateSegment 格式返回。
+<<<<<<< Updated upstream
+
+=======
+>>>>>>> Stashed changes
+        优先读 template_clip_configs；如果为空则回退读旧 template_segments（兼容历史数据）。
+        """
+        clip_models = (
+            self.session.query(TemplateClipConfigModel)
+            .filter(TemplateClipConfigModel.template_id == template_id)
+            .order_by(TemplateClipConfigModel.order)
+            .all()
+        )
+        if clip_models:
+            return [self._clip_config_to_segment(m) for m in clip_models]
+
+        # 回退：读旧 template_segments 表（历史数据兼容）
+        old_models = (
             self.session.query(TemplateSegmentModel)
             .filter(TemplateSegmentModel.template_id == template_id)
             .order_by(TemplateSegmentModel.segment_order)
             .all()
         )
-        return [self._segment_model_to_entity(m) for m in models]
+        return [self._segment_model_to_entity(m) for m in old_models]
 
     def create_segments(self, segments: List[TemplateSegment]) -> List[TemplateSegment]:
+<<<<<<< Updated upstream
+        """将 segments 写入 template_clip_configs 表。
+
+        material_type 信息保存在 config JSON 字段中。
+        """
+=======
+        """将 segments 写入 template_clip_configs 表。material_type 保存在 config JSON 字段中。"""
+>>>>>>> Stashed changes
         for seg in segments:
-            model = TemplateSegmentModel(
+            config = {"material_type": seg.material_type} if seg.material_type else {}
+            clip_model = TemplateClipConfigModel(
                 id=seg.id,
                 template_id=seg.template_id,
-                segment_order=seg.segment_order,
-                duration_min=seg.duration_min,
-                duration_max=seg.duration_max,
-                material_type=seg.material_type,
+                clip_type="main",
+                order=seg.segment_order,
+                min_duration=seg.duration_min,
+                max_duration=seg.duration_max,
+                text_template="",
+                material_requirements={},
+                transition_effect="cut",
+                config=config,
             )
-            self.session.add(model)
+            self.session.add(clip_model)
         self.session.commit()
         return segments
 
     def delete_segments_by_template(self, template_id: str) -> int:
+        """删除 template_clip_configs 中的记录。同时清理旧 template_segments。"""
         count = (
-            self.session.query(TemplateSegmentModel).filter(TemplateSegmentModel.template_id == template_id).delete()
+            self.session.query(TemplateClipConfigModel)
+            .filter(TemplateClipConfigModel.template_id == template_id)
+            .delete()
+<<<<<<< Updated upstream
+=======
         )
+        old_count = (
+            self.session.query(TemplateSegmentModel)
+            .filter(TemplateSegmentModel.template_id == template_id)
+            .delete()
+>>>>>>> Stashed changes
+        )
+        # 同时清理旧表（兼容历史数据）
+        self.session.query(TemplateSegmentModel).filter(
+            TemplateSegmentModel.template_id == template_id,
+        ).delete(synchronize_session=False)
         self.session.commit()
-        return count
+        return count + old_count
 
     # ── Categories ──
 
@@ -309,7 +379,6 @@ class SQLAlchemyTemplateRepository:
     # ── Tags ──
 
     def list_tags(self, user_id: str) -> List[str]:
-        """获取用户所有模板的标签（去重）。"""
         models = (
             self.session.query(TemplateModel)
             .filter(
@@ -330,7 +399,6 @@ class SQLAlchemyTemplateRepository:
     # ── Usage Stats ──
 
     def get_usage_count(self, template_id: str) -> int:
-        """获取模板被使用的次数（关联的剪辑计划数量）。"""
         return self.session.query(EditPlanModel).filter(EditPlanModel.template_id == template_id).count()
 
     # ── Mapping helpers ──
@@ -353,8 +421,28 @@ class SQLAlchemyTemplateRepository:
             updated_at=model.updated_at,
         )
 
+    @staticmethod
+    def _clip_config_to_segment(model: TemplateClipConfigModel) -> TemplateSegment:
+<<<<<<< Updated upstream
+        """将 TemplateClipConfigModel 映射为 TemplateSegment（前端兼容格式）。"""
+=======
+>>>>>>> Stashed changes
+        config = model.config or {}
+        material_type = config.get("material_type")
+        return TemplateSegment(
+            id=model.id,
+            template_id=model.template_id,
+            segment_order=model.order,
+            duration_min=model.min_duration or 0.0,
+            duration_max=model.max_duration or 0.0,
+            material_type=material_type,
+            created_at=model.created_at,
+            updated_at=model.updated_at,
+        )
+
     @staticmethod
     def _segment_model_to_entity(model: TemplateSegmentModel) -> TemplateSegment:
+        """兼容旧 template_segments 表的映射（仅用于历史数据回退读取）。"""
         return TemplateSegment(
             id=model.id,
             template_id=model.template_id,

scripts/ci/step_checkout.sh

@@ -42,3 +42,4 @@ with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
         if member.name:
             tar.extract(member, '.')
 PY
+

tests/unit/test_unify_template_segments.py

@@ -0,0 +1,381 @@
+"""Tests for unified template segments (read from template_clip_configs)."""
+
+import uuid
+<<<<<<< Updated upstream
+from unittest.mock import MagicMock
+=======
+>>>>>>> Stashed changes
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from packages.adapters.sqlalchemy_impl.models import Base, TemplateClipConfigModel, TemplateModel, TemplateSegmentModel
+from packages.adapters.sqlalchemy_impl.template_repository import SQLAlchemyTemplateRepository
+from packages.domain.template import Template, TemplateSegment
+
+
+@pytest.fixture
+def db_session():
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    SessionLocal = sessionmaker(bind=engine)
+    session = SessionLocal()
+    yield session
+    session.close()
+
+
+@pytest.fixture
+def repo(db_session):
+    return SQLAlchemyTemplateRepository(db_session)
+
+
+def _make_template(user_id: str = "user1", name: str = "test") -> Template:
+<<<<<<< Updated upstream
+    return Template(
+        id=uuid.uuid4().hex,
+        user_id=user_id,
+        name=name,
+        mode="one_take",
+    )
+=======
+    return Template(id=uuid.uuid4().hex, user_id=user_id, name=name, mode="one_take")
+>>>>>>> Stashed changes
+
+
+def _make_segment(template_id: str, order: int, material_type: str = "video") -> TemplateSegment:
+    return TemplateSegment(
+<<<<<<< Updated upstream
+        id=uuid.uuid4().hex,
+        template_id=template_id,
+        segment_order=order,
+        duration_min=3.0,
+        duration_max=5.0,
+        material_type=material_type,
+=======
+        id=uuid.uuid4().hex, template_id=template_id, segment_order=order,
+        duration_min=3.0, duration_max=5.0, material_type=material_type,
+>>>>>>> Stashed changes
+    )
+
+
+class TestCreateSegmentsWritesToClipConfigs:
+<<<<<<< Updated upstream
+    """create_segments 应该写入 template_clip_configs 表"""
+
+    def test_create_segments_writes_clip_configs(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
+=======
+    def test_create_segments_writes_clip_configs(self, repo, db_session):
+>>>>>>> Stashed changes
+        tpl = _make_template()
+        repo.create(tpl)
+        seg = _make_segment(tpl.id, 0, "voiceover")
+        repo.create_segments([seg])
+<<<<<<< Updated upstream
+
+        # 验证 template_clip_configs 有记录
+        clips = db_session.query(TemplateClipConfigModel).filter(
+            TemplateClipConfigModel.template_id == tpl.id
+        ).all()
+        assert len(clips) == 1
+        assert clips[0].order == 0
+        assert clips[0].min_duration == 3.0
+        assert clips[0].max_duration == 5.0
+        assert clips[0].clip_type == "main"
+        assert clips[0].config.get("material_type") == "voiceover"
+
+    def test_create_segments_no_old_table_write(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
+        """确保不写入旧 template_segments 表"""
+        tpl = _make_template()
+        repo.create(tpl)
+        seg = _make_segment(tpl.id, 0)
+        repo.create_segments([seg])
+
+        old_rows = db_session.query(TemplateSegmentModel).filter(
+            TemplateSegmentModel.template_id == tpl.id
+        ).all()
+=======
+        clips = db_session.query(TemplateClipConfigModel).filter(
+            TemplateClipConfigModel.template_id == tpl.id).all()
+        assert len(clips) == 1
+        assert clips[0].order == 0
+        assert clips[0].config.get("material_type") == "voiceover"
+
+    def test_create_segments_no_old_table_write(self, repo, db_session):
+        tpl = _make_template()
+        repo.create(tpl)
+        repo.create_segments([_make_segment(tpl.id, 0)])
+        old_rows = db_session.query(TemplateSegmentModel).filter(
+            TemplateSegmentModel.template_id == tpl.id).all()
+>>>>>>> Stashed changes
+        assert len(old_rows) == 0
+
+
+class TestReadSegmentsFromClipConfigs:
+<<<<<<< Updated upstream
+    """list_segments 应该从 template_clip_configs 读取"""
+
+    def test_list_segments_reads_clip_configs(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
+        tpl = _make_template()
+        repo.create(tpl)
+        seg = _make_segment(tpl.id, 0, "video")
+        repo.create_segments([seg])
+
+        segments = repo.list_segments(tpl.id)
+        assert len(segments) == 1
+        assert segments[0].segment_order == 0
+        assert segments[0].duration_min == 3.0
+        assert segments[0].duration_max == 5.0
+        assert segments[0].material_type == "video"
+
+    def test_list_segments_fallback_to_old_table(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
+        """如果 template_clip_configs 为空，回退读旧 template_segments"""
+        tpl = _make_template()
+        repo.create(tpl)
+
+        # 直接写入旧表（模拟历史数据）
+        old_model = TemplateSegmentModel(
+            id=uuid.uuid4().hex,
+            template_id=tpl.id,
+            segment_order=0,
+            duration_min=2.0,
+            duration_max=4.0,
+            material_type="人物",
+        )
+        db_session.add(old_model)
+        db_session.commit()
+
+        segments = repo.list_segments(tpl.id)
+        assert len(segments) == 1
+        assert segments[0].material_type == "人物"
+        assert segments[0].duration_min == 2.0
+
+    def test_list_segments_prefers_clip_configs_over_old(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
+        """如果两个表都有数据，优先读 template_clip_configs"""
+        tpl = _make_template()
+        repo.create(tpl)
+
+        # 写入新表
+        seg = _make_segment(tpl.id, 0, "video")
+        repo.create_segments([seg])
+
+        # 也写入旧表
+        old_model = TemplateSegmentModel(
+            id=uuid.uuid4().hex,
+            template_id=tpl.id,
+            segment_order=99,
+            duration_min=1.0,
+            duration_max=2.0,
+            material_type="stale_data",
+        )
+        db_session.add(old_model)
+        db_session.commit()
+
+        segments = repo.list_segments(tpl.id)
+        # 应该只返回新表的数据
+        assert len(segments) == 1
+        assert segments[0].material_type == "video"
+        assert segments[0].segment_order == 0
+
+
+class TestListByUserSegments:
+    """list_by_user 批量加载 segments 应从 template_clip_configs 读取"""
+
+    def test_batch_load_segments_from_clip_configs(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
+=======
+    def test_list_segments_reads_clip_configs(self, repo, db_session):
+        tpl = _make_template()
+        repo.create(tpl)
+        repo.create_segments([_make_segment(tpl.id, 0, "video")])
+        segments = repo.list_segments(tpl.id)
+        assert len(segments) == 1
+        assert segments[0].material_type == "video"
+
+    def test_list_segments_fallback_to_old_table(self, repo, db_session):
+        tpl = _make_template()
+        repo.create(tpl)
+        old_model = TemplateSegmentModel(
+            id=uuid.uuid4().hex, template_id=tpl.id, segment_order=0,
+            duration_min=2.0, duration_max=4.0, material_type="人物")
+        db_session.add(old_model)
+        db_session.commit()
+        segments = repo.list_segments(tpl.id)
+        assert len(segments) == 1
+        assert segments[0].material_type == "人物"
+
+    def test_list_segments_prefers_clip_configs_over_old(self, repo, db_session):
+        tpl = _make_template()
+        repo.create(tpl)
+        repo.create_segments([_make_segment(tpl.id, 0, "video")])
+        old_model = TemplateSegmentModel(
+            id=uuid.uuid4().hex, template_id=tpl.id, segment_order=99,
+            duration_min=1.0, duration_max=2.0, material_type="stale_data")
+        db_session.add(old_model)
+        db_session.commit()
+        segments = repo.list_segments(tpl.id)
+        assert len(segments) == 1
+        assert segments[0].material_type == "video"
+
+
+class TestListByUserSegments:
+    def test_batch_load_segments_from_clip_configs(self, repo, db_session):
+>>>>>>> Stashed changes
+        tpl1 = _make_template(name="tpl1")
+        tpl2 = _make_template(name="tpl2")
+        repo.create(tpl1)
+        repo.create(tpl2)
+<<<<<<< Updated upstream
+
+        repo.create_segments([_make_segment(tpl1.id, 0, "video")])
+        repo.create_segments([_make_segment(tpl2.id, 0, "voiceover"), _make_segment(tpl2.id, 1, "video")])
+
+        templates = repo.list_by_user("user1")
+        assert len(templates) == 2
+
+        tpl1_result = next(t for t in templates if t.id == tpl1.id)
+        tpl2_result = next(t for t in templates if t.id == tpl2.id)
+
+        assert len(tpl1_result.segments) == 1
+        assert tpl1_result.segments[0].material_type == "video"
+
+        assert len(tpl2_result.segments) == 2
+        assert tpl2_result.segments[0].material_type == "voiceover"
+        assert tpl2_result.segments[1].material_type == "video"
+
+
+class TestDeleteSegments:
+    """delete_segments_by_template 应清理 template_clip_configs"""
+
+    def test_delete_segments_clears_clip_configs(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
+        tpl = _make_template()
+        repo.create(tpl)
+        repo.create_segments([_make_segment(tpl.id, 0)])
+
+        count = repo.delete_segments_by_template(tpl.id)
+        assert count == 1
+
+        clips = db_session.query(TemplateClipConfigModel).filter(
+            TemplateClipConfigModel.template_id == tpl.id
+        ).all()
+        assert len(clips) == 0
+
+    def test_delete_template_clears_both_tables(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
+        tpl = _make_template()
+        repo.create(tpl)
+        repo.create_segments([_make_segment(tpl.id, 0)])
+
+        # 也写入旧表
+        old_model = TemplateSegmentModel(
+            id=uuid.uuid4().hex,
+            template_id=tpl.id,
+            segment_order=0,
+            duration_min=1.0,
+            duration_max=2.0,
+        )
+        db_session.add(old_model)
+        db_session.commit()
+
+        repo.delete(tpl.id, "user1")
+
+        clips = db_session.query(TemplateClipConfigModel).filter(
+            TemplateClipConfigModel.template_id == tpl.id
+        ).all()
+        assert len(clips) == 0
+
+        old_rows = db_session.query(TemplateSegmentModel).filter(
+            TemplateSegmentModel.template_id == tpl.id
+        ).all()
+=======
+        repo.create_segments([_make_segment(tpl1.id, 0, "video")])
+        repo.create_segments([_make_segment(tpl2.id, 0, "voiceover"), _make_segment(tpl2.id, 1, "video")])
+        templates = repo.list_by_user("user1")
+        assert len(templates) == 2
+        tpl1_r = next(t for t in templates if t.id == tpl1.id)
+        tpl2_r = next(t for t in templates if t.id == tpl2.id)
+        assert len(tpl1_r.segments) == 1
+        assert len(tpl2_r.segments) == 2
+
+
+class TestDeleteSegments:
+    def test_delete_segments_clears_clip_configs(self, repo, db_session):
+        tpl = _make_template()
+        repo.create(tpl)
+        repo.create_segments([_make_segment(tpl.id, 0)])
+        count = repo.delete_segments_by_template(tpl.id)
+        assert count >= 1
+        clips = db_session.query(TemplateClipConfigModel).filter(
+            TemplateClipConfigModel.template_id == tpl.id).all()
+        assert len(clips) == 0
+
+    def test_delete_template_clears_both_tables(self, repo, db_session):
+        tpl = _make_template()
+        repo.create(tpl)
+        repo.create_segments([_make_segment(tpl.id, 0)])
+        old_model = TemplateSegmentModel(
+            id=uuid.uuid4().hex, template_id=tpl.id, segment_order=0,
+            duration_min=1.0, duration_max=2.0)
+        db_session.add(old_model)
+        db_session.commit()
+        repo.delete(tpl.id, "user1")
+        clips = db_session.query(TemplateClipConfigModel).filter(
+            TemplateClipConfigModel.template_id == tpl.id).all()
+        assert len(clips) == 0
+        old_rows = db_session.query(TemplateSegmentModel).filter(
+            TemplateSegmentModel.template_id == tpl.id).all()
+>>>>>>> Stashed changes
+        assert len(old_rows) == 0
+
+
+class TestCopyTemplateSegments:
+<<<<<<< Updated upstream
+    """copy_template 应从 template_clip_configs 复制 segments"""
+
+    def test_copy_preserves_segments(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
+        tpl = _make_template()
+        repo.create(tpl)
+        repo.create_segments([
+            _make_segment(tpl.id, 0, "video"),
+            _make_segment(tpl.id, 1, "voiceover"),
+        ])
+
+        copied = repo.copy_template(tpl.id, "user1", "copy")
+        assert len(copied.segments) == 2
+        assert copied.segments[0].material_type == "video"
+        assert copied.segments[1].material_type == "voiceover"
+
+        # 验证新模板的 clip_configs 有数据
+        clips = db_session.query(TemplateClipConfigModel).filter(
+            TemplateClipConfigModel.template_id == copied.id
+        ).all()
+=======
+    def test_copy_preserves_segments(self, repo, db_session):
+        tpl = _make_template()
+        repo.create(tpl)
+        repo.create_segments([_make_segment(tpl.id, 0, "video"), _make_segment(tpl.id, 1, "voiceover")])
+        copied = repo.copy_template(tpl.id, "user1", "copy")
+        assert len(copied.segments) == 2
+        clips = db_session.query(TemplateClipConfigModel).filter(
+            TemplateClipConfigModel.template_id == copied.id).all()
+>>>>>>> Stashed changes
+        assert len(clips) == 2
+
+
+class TestNullMaterialType:
+<<<<<<< Updated upstream
+    """material_type 为 None 时不应报错"""
+
+    def test_null_material_type(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
+        tpl = _make_template()
+        repo.create(tpl)
+        seg = _make_segment(tpl.id, 0, None)
+        repo.create_segments([seg])
+
+=======
+    def test_null_material_type(self, repo, db_session):
+        tpl = _make_template()
+        repo.create(tpl)
+        repo.create_segments([_make_segment(tpl.id, 0, None)])
+>>>>>>> Stashed changes
+        segments = repo.list_segments(tpl.id)
+        assert len(segments) == 1
+        assert segments[0].material_type is None