"""SQLAlchemy implementation of EditPlanClipRepository.""" from __future__ import annotations from typing import List, 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