xiaoxia-saas/packages/adapters/sqlalchemy_impl/template_repository.py

"""SQLAlchemy implementation of TemplateRepository.

模板 segments 数据源已统一为 template_clip_configs 表。
旧 template_segments 表不再读写，保留表结构供历史数据查询。
"""

from __future__ import annotations

import uuid
from typing import List, Optional

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import (
    EditPlanModel,
    TemplateCategoryModel,
    TemplateClipConfigModel,
    TemplateModel,
    TemplateSegmentModel,
)
from packages.domain.template import Template, TemplateCategory, TemplateSegment


class SQLAlchemyTemplateRepository:
    """SQLAlchemy 剪辑计划模板仓储."""

    def __init__(self, session: Session) -> None:
        self.session = session

    # ── Template CRUD ──

    def list_by_user(
        self,
        user_id: str,
        *,
        skip: int = 0,
        limit: int = 50,
        category: Optional[str] = None,
        tag: Optional[str] = None,
        keyword: Optional[str] = None,
        mode: Optional[str] = None,
    ) -> List[Template]:
        query = self.session.query(TemplateModel).filter(
            TemplateModel.user_id == user_id,
            TemplateModel.is_active.is_(True),
        )
        if category:
            query = query.filter(TemplateModel.category == category)
        if mode:
            query = query.filter(TemplateModel.mode == mode)
        if keyword:
            like_pattern = f"%{keyword}%"
            query = query.filter(TemplateModel.name.like(like_pattern))
        if tag:
            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：从 template_clip_configs 读取，映射为 TemplateSegment
        if templates:
            template_ids = [t.id for t in templates]
            clip_models = (
                self.session.query(TemplateClipConfigModel)
                .filter(TemplateClipConfigModel.template_id.in_(template_ids))
                .order_by(TemplateClipConfigModel.order)
                .all()
            )
            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 = clip_map.get(t.id, [])
        return templates

    def get(self, template_id: str, user_id: str) -> Optional[Template]:
        model = (
            self.session.query(TemplateModel)
            .filter(
                TemplateModel.id == template_id,
                TemplateModel.user_id == user_id,
            )
            .first()
        )
        if model is None:
            return None
        template = self._model_to_entity(model)
        template.segments = self.list_segments(template.id)
        return template

    def create(self, template: Template) -> Template:
        model = TemplateModel(
            id=template.id,
            user_id=template.user_id,
            name=template.name,
            mode=template.mode,
            category=template.category,
            tags=template.tags,
            title_config=template.title_config,
            subtitle_config=template.subtitle_config,
            bgm_config=template.bgm_config,
            estimated_duration=template.estimated_duration,
            is_active=template.is_active,
        )
        self.session.add(model)
        self.session.flush()
        self.session.refresh(model)
        result = self._model_to_entity(model)
        result.segments = template.segments
        return result

    def update(self, template: Template) -> Template:
        model = (
            self.session.query(TemplateModel)
            .filter(
                TemplateModel.id == template.id,
                TemplateModel.user_id == template.user_id,
            )
            .first()
        )
        if model is None:
            raise ValueError(f"Template {template.id} not found")
        model.name = template.name
        model.mode = template.mode
        model.category = template.category
        model.tags = template.tags
        model.title_config = template.title_config
        model.subtitle_config = template.subtitle_config
        model.bgm_config = template.bgm_config
        model.estimated_duration = template.estimated_duration
        model.is_active = template.is_active
        self.session.commit()
        self.session.refresh(model)
        result = self._model_to_entity(model)
        result.segments = template.segments
        return result

    def delete(self, template_id: str, user_id: str) -> bool:
        model = (
            self.session.query(TemplateModel)
            .filter(
                TemplateModel.id == template_id,
                TemplateModel.user_id == user_id,
            )
            .first()
        )
        if model is None:
            return False
        model.is_active = False
        # 清理 template_clip_configs（主数据源）
        self.session.query(TemplateClipConfigModel).filter(
            TemplateClipConfigModel.template_id == template_id,
        ).delete(synchronize_session=False)
        # 同时清理旧 template_segments（兼容历史数据）
        self.session.query(TemplateSegmentModel).filter(
            TemplateSegmentModel.template_id == template_id,
        ).delete(synchronize_session=False)
        self.session.commit()
        return True

    def count_by_user(
        self,
        user_id: str,
        *,
        category: Optional[str] = None,
        tag: Optional[str] = None,
        keyword: Optional[str] = None,
        mode: Optional[str] = None,
    ) -> int:
        query = self.session.query(TemplateModel).filter(
            TemplateModel.user_id == user_id,
            TemplateModel.is_active.is_(True),
        )
        if category:
            query = query.filter(TemplateModel.category == category)
        if keyword:
            query = query.filter(TemplateModel.name.like(f"%{keyword}%"))
        if tag:
            query = query.filter(TemplateModel.tags.like(f'%"{tag}"%'))
        return query.count()

    def copy_template(self, template_id: str, user_id: str, new_name: str) -> Template:
        """复制模板（含所有 segments，从 template_clip_configs 读取并写入）。"""
        source = self.get(template_id, user_id)
        if source is None:
            raise ValueError(f"Template {template_id} not found")

        new_id = str(uuid.uuid4())
        new_template = Template(
            id=new_id,
            user_id=user_id,
            name=new_name,
            mode=source.mode,
            category=source.category,
            tags=list(source.tags),
            title_config=dict(source.title_config),
            subtitle_config=dict(source.subtitle_config),
            bgm_config=dict(source.bgm_config),
            estimated_duration=source.estimated_duration,
            is_active=True,
        )
        created = self.create(new_template)

        # 复制 segments → 写入 template_clip_configs
        new_segments: List[TemplateSegment] = []
        for seg in source.segments:
            new_seg = TemplateSegment(
                id=str(uuid.uuid4()),
                template_id=new_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)
            config = {"material_type": seg.material_type} if seg.material_type else {}
            clip_model = TemplateClipConfigModel(
                id=new_seg.id,
                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(clip_model)
        if new_segments:
            self.session.commit()

        created.segments = new_segments
        return created

    # ── Segments（数据源：template_clip_configs）──

    def list_segments(self, template_id: str) -> List[TemplateSegment]:
        """从 template_clip_configs 读取并按 TemplateSegment 格式返回。

        优先读 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 old_models]

    def create_segments(self, segments: List[TemplateSegment]) -> List[TemplateSegment]:
        """将 segments 写入 template_clip_configs 表。

        material_type 信息保存在 config JSON 字段中。
        """
        for seg in segments:
            config = {"material_type": seg.material_type} if seg.material_type else {}
            clip_model = TemplateClipConfigModel(
                id=seg.id,
                template_id=seg.template_id,
                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(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(TemplateClipConfigModel)
            .filter(TemplateClipConfigModel.template_id == template_id)
            .delete()
        )
        # 同时清理旧表（兼容历史数据）
        self.session.query(TemplateSegmentModel).filter(
            TemplateSegmentModel.template_id == template_id,
        ).delete(synchronize_session=False)
        self.session.commit()
        return count

    # ── Categories ──

    def list_categories(self, user_id: str) -> List[TemplateCategory]:
        models = (
            self.session.query(TemplateCategoryModel)
            .filter(TemplateCategoryModel.user_id == user_id)
            .order_by(TemplateCategoryModel.created_at)
            .all()
        )
        return [self._category_model_to_entity(m) for m in models]

    def create_category(self, category: TemplateCategory) -> TemplateCategory:
        model = TemplateCategoryModel(
            id=category.id,
            user_id=category.user_id,
            name=category.name,
        )
        self.session.add(model)
        self.session.commit()
        self.session.refresh(model)
        return self._category_model_to_entity(model)

    def get_category(self, category_id: str, user_id: str) -> Optional[TemplateCategory]:
        model = (
            self.session.query(TemplateCategoryModel)
            .filter(
                TemplateCategoryModel.id == category_id,
                TemplateCategoryModel.user_id == user_id,
            )
            .first()
        )
        if model is None:
            return None
        return self._category_model_to_entity(model)

    def delete_category(self, category_id: str, user_id: str) -> bool:
        model = (
            self.session.query(TemplateCategoryModel)
            .filter(
                TemplateCategoryModel.id == category_id,
                TemplateCategoryModel.user_id == user_id,
            )
            .first()
        )
        if model is None:
            return False
        self.session.delete(model)
        self.session.commit()
        return True

    # ── Tags ──

    def list_tags(self, user_id: str) -> List[str]:
        """获取用户所有模板的标签（去重）。"""
        models = (
            self.session.query(TemplateModel)
            .filter(
                TemplateModel.user_id == user_id,
                TemplateModel.is_active.is_(True),
                TemplateModel.tags.isnot(None),
            )
            .all()
        )
        tags_set: set[str] = set()
        for m in models:
            if m.tags:
                for t in m.tags:
                    if t:
                        tags_set.add(t)
        return sorted(tags_set)

    # ── 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 ──

    @staticmethod
    def _model_to_entity(model: TemplateModel) -> Template:
        return Template(
            id=model.id,
            user_id=model.user_id,
            name=model.name,
            mode=model.mode,
            category=model.category or "",
            tags=model.tags or [],
            title_config=model.title_config or {},
            subtitle_config=model.subtitle_config or {},
            bgm_config=model.bgm_config or {},
            estimated_duration=model.estimated_duration or 0.0,
            is_active=model.is_active,
            created_at=model.created_at,
            updated_at=model.updated_at,
        )

    @staticmethod
    def _clip_config_to_segment(model: TemplateClipConfigModel) -> TemplateSegment:
        """将 TemplateClipConfigModel 映射为 TemplateSegment（前端兼容格式）。"""
        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,
            segment_order=model.segment_order,
            duration_min=model.duration_min,
            duration_max=model.duration_max,
            material_type=model.material_type,
            created_at=model.created_at,
            updated_at=model.updated_at,
        )

    @staticmethod
    def _category_model_to_entity(model: TemplateCategoryModel) -> TemplateCategory:
        return TemplateCategory(
            id=model.id,
            user_id=model.user_id,
            name=model.name,
            created_at=model.created_at,
        )