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

"""SQLAlchemy implementation of EditPlanClipRepository."""

from __future__ import annotations

from typing import Optional

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import EditPlanClipModel
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus


class SQLAlchemyEditPlanClipRepository:
    """SQLAlchemy 剪辑计划片段仓储"""

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

    def list_by_plan(
        self,
        plan_id: str,
        *,
        status: Optional[EditPlanClipStatus] = None,
        skip: int = 0,
        limit: int = 100,
    ) -> list[EditPlanClip]:
        """按计划列出片段，按 order 排序"""
        query = self.session.query(EditPlanClipModel).filter(
            EditPlanClipModel.plan_id == plan_id,
        )
        if status:
            query = query.filter(EditPlanClipModel.status == status)
        query = query.order_by(EditPlanClipModel.order)
        models = query.offset(skip).limit(limit).all()
        return [self._model_to_entity(m) for m in models]

    def get(self, clip_id: str) -> Optional[EditPlanClip]:
        """根据 ID 获取片段"""
        model = self.session.query(EditPlanClipModel).filter(EditPlanClipModel.id == clip_id).first()
        if model is None:
            return None
        return self._model_to_entity(model)

    def create(self, clip: EditPlanClip) -> EditPlanClip:
        """创建片段"""
        model = EditPlanClipModel(
            id=clip.id,
            plan_id=clip.plan_id,
            clip_type=clip.clip_type,
            order=clip.order,
            template_clip_config_id=clip.template_clip_config_id,
            asset_id=clip.asset_id,
            text_content=clip.text_content,
            start_time=clip.start_time,
            duration=clip.duration,
            transition_effect=clip.transition_effect,
            transition_duration=clip.transition_duration,
            playback_speed=clip.playback_speed,
            status=clip.status,
            config=clip.config,
        )
        self.session.add(model)
        self.session.commit()
        self.session.refresh(model)
        return self._model_to_entity(model)

    def update(self, clip: EditPlanClip) -> EditPlanClip:
        """更新片段"""
        model = self.session.query(EditPlanClipModel).filter(EditPlanClipModel.id == clip.id).first()
        if model is None:
            raise ValueError(f"EditPlanClip {clip.id} not found")
        model.plan_id = clip.plan_id
        model.clip_type = clip.clip_type
        model.order = clip.order
        model.template_clip_config_id = clip.template_clip_config_id
        model.asset_id = clip.asset_id
        model.text_content = clip.text_content
        model.start_time = clip.start_time
        model.duration = clip.duration
        model.transition_effect = clip.transition_effect
        model.transition_duration = clip.transition_duration
        model.playback_speed = clip.playback_speed
        model.status = clip.status
        model.config = clip.config
        model.updated_at = clip.updated_at
        self.session.commit()
        self.session.refresh(model)
        return self._model_to_entity(model)

    def delete(self, clip_id: str) -> bool:
        """删除片段"""
        model = self.session.query(EditPlanClipModel).filter(EditPlanClipModel.id == clip_id).first()
        if model is None:
            return False
        self.session.delete(model)
        self.session.commit()
        return True

    def delete_by_plan(self, plan_id: str) -> int:
        """删除计划下所有片段，返回删除数量"""
        count = self.session.query(EditPlanClipModel).filter(EditPlanClipModel.plan_id == plan_id).delete()
        self.session.commit()
        return count

    def count(self, *, plan_id: Optional[str] = None, status: Optional[EditPlanClipStatus] = None) -> int:
        """统计片段数量"""
        query = self.session.query(EditPlanClipModel)
        if plan_id:
            query = query.filter(EditPlanClipModel.plan_id == plan_id)
        if status:
            query = query.filter(EditPlanClipModel.status == status)
        return query.count()

    @staticmethod
    def _model_to_entity(model: EditPlanClipModel) -> EditPlanClip:
        return EditPlanClip(
            id=model.id,
            plan_id=model.plan_id,
            clip_type=model.clip_type,
            order=model.order,
            template_clip_config_id=model.template_clip_config_id or "",
            asset_id=model.asset_id or "",
            text_content=model.text_content or "",
            start_time=model.start_time or 0.0,
            duration=model.duration or 0.0,
            transition_effect=model.transition_effect or "cut",
            transition_duration=getattr(model, "transition_duration", 0.0) or 0.0,
            playback_speed=model.playback_speed or 1.0,
            status=EditPlanClipStatus(model.status) if model.status else EditPlanClipStatus.PENDING,
            config=model.config or {},
            created_at=model.created_at,
            updated_at=model.updated_at,
        )

    def list_used_segments_by_user(
        self,
        user_id: str,
        *,
        limit_recent: int = 50,
    ) -> dict[str, list[tuple[float, float]]]:
        """查询用户已有视频中已使用的素材区间（跨视频避让）.

        JOIN edit_plans 表，按 created_by_user_id 过滤，只查 status='completed'
        的 plan 下 status='rendered' 且 asset_id 非空的 clips。按 plan 的
        created_at DESC 取最近 limit_recent 个 plan。

        Returns:
            {asset_id: [(start_time, start_time + duration), ...]}
            空结果返回空 dict。
        """
        from packages.adapters.sqlalchemy_impl.models import EditPlanModel

        if not user_id:
            return {}

        # 1. 查出最近 limit_recent 个已完成 plan 的 ID
        recent_plan_ids = [
            row[0]
            for row in self.session.query(EditPlanModel.id)
            .filter(
                EditPlanModel.created_by_user_id == user_id,
                EditPlanModel.status == "completed",
            )
            .order_by(EditPlanModel.created_at.desc())
            .limit(limit_recent)
            .all()
        ]

        if not recent_plan_ids:
            return {}

        # 2. 查这些 plan 下已渲染、有素材的 clips
        clips = (
            self.session.query(
                EditPlanClipModel.asset_id,
                EditPlanClipModel.start_time,
                EditPlanClipModel.duration,
            )
            .filter(
                EditPlanClipModel.plan_id.in_(recent_plan_ids),
                EditPlanClipModel.status == "rendered",
                EditPlanClipModel.asset_id != "",
                EditPlanClipModel.asset_id.isnot(None),
            )
            .all()
        )

        # 3. 聚合为 {asset_id: [(start, start+duration), ...]}
        result: dict[str, list[tuple[float, float]]] = {}
        for asset_id, start_time, duration in clips:
            if asset_id not in result:
                result[asset_id] = []
            result[asset_id].append((start_time or 0.0, (start_time or 0.0) + (duration or 0.0)))

        return result