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

"""SQLAlchemy implementation of EditTemplateRepository."""

from __future__ import annotations

from typing import List, Optional

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import EditTemplateModel
from packages.domain.edit_template import EditTemplate, EditTemplateStatus


class SQLAlchemyEditTemplateRepository:
    """SQLAlchemy 剪辑模板仓储"""

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

    def list_active(
        self,
        *,
        template_type: Optional[str] = None,
        skip: int = 0,
        limit: int = 50,
    ) -> List[EditTemplate]:
        """列出所有激活的模板"""
        query = self.session.query(EditTemplateModel).filter(
            EditTemplateModel.status == EditTemplateStatus.ACTIVE,
        )
        if template_type:
            query = query.filter(EditTemplateModel.template_type == template_type)
        query = query.order_by(
            EditTemplateModel.sort_weight.desc(),
            EditTemplateModel.created_at.desc(),
        )
        models = query.offset(skip).limit(limit).all()
        return [self._model_to_entity(m) for m in models]

    def list_all(
        self,
        *,
        template_type: Optional[str] = None,
        status: Optional[EditTemplateStatus] = None,
        skip: int = 0,
        limit: int = 50,
    ) -> List[EditTemplate]:
        """列出所有模板（含非激活）"""
        query = self.session.query(EditTemplateModel)
        if template_type:
            query = query.filter(EditTemplateModel.template_type == template_type)
        if status:
            query = query.filter(EditTemplateModel.status == status)
        query = query.order_by(
            EditTemplateModel.sort_weight.desc(),
            EditTemplateModel.created_at.desc(),
        )
        models = query.offset(skip).limit(limit).all()
        return [self._model_to_entity(m) for m in models]

    def get(self, template_id: str) -> Optional[EditTemplate]:
        """根据 ID 获取模板"""
        model = self.session.query(EditTemplateModel).filter(EditTemplateModel.id == template_id).first()
        if model is None:
            return None
        return self._model_to_entity(model)

    def create(self, template: EditTemplate) -> EditTemplate:
        """创建模板"""
        model = EditTemplateModel(
            id=template.id,
            name=template.name,
            description=template.description,
            template_type=template.template_type,
            editing_mode=template.editing_mode,
            config=template.config,
            preview_url=template.preview_url,
            sort_weight=template.sort_weight,
            status=template.status,
        )
        self.session.add(model)
        self.session.commit()
        self.session.refresh(model)
        return self._model_to_entity(model)

    def update(self, template: EditTemplate) -> EditTemplate:
        """更新模板"""
        model = self.session.query(EditTemplateModel).filter(EditTemplateModel.id == template.id).first()
        if model is None:
            raise ValueError(f"EditTemplate {template.id} not found")
        model.name = template.name
        model.description = template.description
        model.template_type = template.template_type
        model.editing_mode = template.editing_mode
        model.config = template.config
        model.preview_url = template.preview_url
        model.sort_weight = template.sort_weight
        model.status = template.status
        model.updated_at = template.updated_at
        self.session.commit()
        self.session.refresh(model)
        return self._model_to_entity(model)

    def delete(self, template_id: str) -> bool:
        """删除模板"""
        model = self.session.query(EditTemplateModel).filter(EditTemplateModel.id == template_id).first()
        if model is None:
            return False
        self.session.delete(model)
        self.session.commit()
        return True

    def count(
        self,
        *,
        template_type: Optional[str] = None,
        status: Optional[EditTemplateStatus] = None,
    ) -> int:
        """统计模板数量"""
        query = self.session.query(EditTemplateModel)
        if template_type:
            query = query.filter(EditTemplateModel.template_type == template_type)
        if status:
            query = query.filter(EditTemplateModel.status == status)
        return query.count()

    @staticmethod
    def _model_to_entity(model: EditTemplateModel) -> EditTemplate:
        return EditTemplate(
            id=model.id,
            name=model.name,
            description=model.description or "",
            template_type=model.template_type or "default",
            editing_mode=model.editing_mode or "one_take",
            config=model.config or {},
            preview_url=model.preview_url or "",
            sort_weight=model.sort_weight or 0,
            status=EditTemplateStatus(model.status) if model.status else EditTemplateStatus.ACTIVE,
            created_at=model.created_at,
            updated_at=model.updated_at,
        )