xiaoxia-saas/apps/api/app/services/edit_plan_service.py

"""EditPlanService — 剪辑计划管理业务逻辑.

封装 EditPlan 和 EditPlanClip 的 CRUD 操作、状态机流转、
以及渲染生成流程，提供统一的业务接口供 API 路由层调用。
"""

from __future__ import annotations

import logging
from typing import Any, Dict, List, Optional

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl import (
    SQLAlchemyEditPlanClipRepository,
    SQLAlchemyEditPlanRepository,
    SQLAlchemyGenerationTaskRepository,
)
from packages.domain.edit_plan import EditPlan, EditPlanStatus
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus

logger = logging.getLogger(__name__)


class EditPlanService:
    """剪辑计划管理服务

    职责:
    - 剪辑计划 CRUD（创建、查询、更新、删除）
    - 剪辑片段管理（增删改查、分配素材）
    - 状态机流转（draft → editing → rendering → completed/failed）
    - 渲染生成流程（触发 Celery 任务、查询进度）
    """

    def __init__(self, db: Session) -> None:
        self._plan_repo = SQLAlchemyEditPlanRepository(db)
        self._clip_repo = SQLAlchemyEditPlanClipRepository(db)
        self._generation_task_repo = SQLAlchemyGenerationTaskRepository(db)

    # ── 剪辑计划 CRUD ──────────────────────────────────────────────────────

    def get_plan(self, plan_id: str) -> Optional[EditPlan]:
        """获取计划详情"""
        return self._plan_repo.get(plan_id)

    def get_plan_or_raise(self, plan_id: str) -> EditPlan:
        """获取计划，不存在则抛出 ValueError"""
        plan = self._plan_repo.get(plan_id)
        if plan is None:
            raise ValueError(f"剪辑计划不存在: {plan_id}")
        return plan

    def create_plan(
        self,
        template_id: str,
        name: str,
        *,
        config: Optional[dict[str, Any]] = None,
        total_duration: float = 0.0,
        project_id: str = "",
        created_by_user_id: str = "",
    ) -> EditPlan:
        """创建剪辑计划（基础 CRUD，供内部测试与脚本使用）

        Raises:
            ValueError: 参数校验失败
        """
        plan = EditPlan.create(
            template_id=template_id,
            name=name,
            config=config,
            total_duration=total_duration,
            project_id=project_id,
            created_by_user_id=created_by_user_id,
        )
        created = self._plan_repo.create(plan)
        logger.info("创建剪辑计划: id=%s name=%s", created.id, created.name)
        return created

    def _auto_resume_editing(self, plan_id: str) -> None:
        """如果计划处于 completed/failed 状态，自动切回 editing（编辑操作前置）"""
        plan = self._plan_repo.get(plan_id)
        if plan is None:
            return
        if plan.status in (EditPlanStatus.COMPLETED, EditPlanStatus.FAILED):
            try:
                plan.resume_editing()
                self._plan_repo.update(plan)
                logger.info("自动重新编辑: plan_id=%s", plan_id)
            except ValueError:
                pass

    def update_plan(
        self,
        plan_id: str,
        *,
        name: Optional[str] = None,
        config: Optional[dict[str, Any]] = None,
        total_duration: Optional[float] = None,
    ) -> EditPlan:
        """更新计划基础字段

        Raises:
            ValueError: 计划不存在
        """
        existing = self.get_plan_or_raise(plan_id)

        # 自动从 completed/failed 切回 editing
        self._auto_resume_editing(plan_id)
        existing = self.get_plan_or_raise(plan_id)

        updated = EditPlan(
            id=existing.id,
            template_id=existing.template_id,
            name=name.strip() if name is not None else existing.name,
            status=existing.status,
            total_duration=total_duration if total_duration is not None else existing.total_duration,
            source_edit_plan_id=existing.source_edit_plan_id,
            project_id=existing.project_id,
            created_by_user_id=existing.created_by_user_id,
            config=config if config is not None else existing.config,
            created_at=existing.created_at,
            updated_at=existing.updated_at,
        )
        result = self._plan_repo.update(updated)
        logger.info("更新剪辑计划: id=%s", plan_id)
        return result

    # ── 状态机流转 ──────────────────────────────────────────────────────────

    def transition_status(self, plan_id: str, target_status: EditPlanStatus) -> EditPlan:
        """流转计划状态

        状态流转规则:
        - draft → editing (start_editing)
        - editing → rendering (start_rendering)
        - rendering → completed (mark_completed)
        - rendering → failed (mark_failed)
        - failed → draft (reset_to_draft)

        Raises:
            ValueError: 计划不存在或状态流转非法
        """
        plan = self.get_plan_or_raise(plan_id)

        # 如果已是目标状态，直接返回
        if plan.status == target_status:
            return plan

        # 根据目标状态调用对应的状态机方法
        # EDITING 支持从 draft / completed / failed 进入
        if target_status == EditPlanStatus.EDITING:
            if plan.status == EditPlanStatus.DRAFT:
                plan.start_editing()
            elif plan.status in (EditPlanStatus.COMPLETED, EditPlanStatus.FAILED):
                plan.resume_editing()
            else:
                raise ValueError(f"无法从 {plan.status} 切换到 {target_status}")
            result = self._plan_repo.update(plan)
            logger.info(
                "状态流转: plan_id=%s %s → %s",
                plan_id,
                plan.status,
                target_status,
            )
            return result

        transition_map = {
            EditPlanStatus.RENDERING: plan.start_rendering,
            EditPlanStatus.COMPLETED: plan.mark_completed,
            EditPlanStatus.FAILED: plan.mark_failed,
            EditPlanStatus.DRAFT: plan.reset_to_draft,
        }

        transition_fn = transition_map.get(target_status)
        if transition_fn is None:
            raise ValueError(f"无效的目标状态: {target_status}")

        transition_fn()
        result = self._plan_repo.update(plan)
        logger.info(
            "状态流转: plan_id=%s %s → %s",
            plan_id,
            plan.status,
            target_status,
        )
        return result

    # ── 剪辑片段管理 ────────────────────────────────────────────────────────

    def list_clips(
        self,
        plan_id: str,
        *,
        status: Optional[EditPlanClipStatus] = None,
        skip: int = 0,
        limit: int = 100,
    ) -> List[EditPlanClip]:
        """列出计划的片段"""
        # 确保计划存在
        self.get_plan_or_raise(plan_id)
        return self._clip_repo.list_by_plan(plan_id, status=status, skip=skip, limit=limit)

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

    def get_clip(self, clip_id: str) -> Optional[EditPlanClip]:
        """获取片段详情"""
        return self._clip_repo.get(clip_id)

    def get_clip_or_raise(self, clip_id: str) -> EditPlanClip:
        """获取片段，不存在则抛出 ValueError"""
        clip = self._clip_repo.get(clip_id)
        if clip is None:
            raise ValueError(f"片段不存在: {clip_id}")
        return clip

    def create_clip(
        self,
        plan_id: str,
        clip_type: str,
        order: int,
        *,
        template_clip_config_id: str = "",
        asset_id: str = "",
        text_content: str = "",
        start_time: float = 0.0,
        duration: float = 0.0,
        transition_effect: str = "cut",
        transition_duration: float = 0.0,
        playback_speed: float = 1.0,
        config: Optional[dict[str, Any]] = None,
    ) -> EditPlanClip:
        """创建片段

        Raises:
            ValueError: 计划不存在或参数校验失败
        """
        # 确保计划存在
        self.get_plan_or_raise(plan_id)
        # 自动从 completed/failed 切回 editing
        self._auto_resume_editing(plan_id)

        clip = EditPlanClip.create(
            plan_id=plan_id,
            clip_type=clip_type,
            order=order,
            template_clip_config_id=template_clip_config_id,
            asset_id=asset_id,
            text_content=text_content,
            start_time=start_time,
            duration=duration,
            transition_effect=transition_effect,
            transition_duration=transition_duration,
            playback_speed=playback_speed,
            config=config,
        )
        created = self._clip_repo.create(clip)
        logger.info(
            "创建片段: id=%s plan_id=%s clip_type=%s order=%d",
            created.id,
            plan_id,
            created.clip_type,
            created.order,
        )
        return created

    def update_clip(
        self,
        clip_id: str,
        *,
        clip_type: Optional[str] = None,
        order: Optional[int] = None,
        asset_id: Optional[str] = None,
        text_content: Optional[str] = None,
        start_time: Optional[float] = None,
        duration: Optional[float] = None,
        transition_effect: Optional[str] = None,
        transition_duration: Optional[float] = None,
        playback_speed: Optional[float] = None,
        config: Optional[dict[str, Any]] = None,
    ) -> EditPlanClip:
        """更新片段

        Raises:
            ValueError: 片段不存在
        """
        existing = self.get_clip_or_raise(clip_id)

        # 自动从 completed/failed 切回 editing
        self._auto_resume_editing(existing.plan_id)

        # 速度边界钳制
        if playback_speed is not None:
            if playback_speed <= 0:
                playback_speed = 1.0
            elif playback_speed < 0.25:
                playback_speed = 0.25
            elif playback_speed > 4.0:
                playback_speed = 4.0

        updated = EditPlanClip(
            id=existing.id,
            plan_id=existing.plan_id,
            clip_type=clip_type.strip() if clip_type is not None else existing.clip_type,
            order=order if order is not None else existing.order,
            template_clip_config_id=existing.template_clip_config_id,
            asset_id=asset_id.strip() if asset_id is not None else existing.asset_id,
            text_content=text_content.strip() if text_content is not None else existing.text_content,
            start_time=start_time if start_time is not None else existing.start_time,
            duration=duration if duration is not None else existing.duration,
            transition_effect=(
                transition_effect.strip() if transition_effect is not None else existing.transition_effect
            ),
            transition_duration=(
                transition_duration if transition_duration is not None else existing.transition_duration
            ),
            playback_speed=playback_speed if playback_speed is not None else existing.playback_speed,
            status=existing.status,
            config=config if config is not None else existing.config,
            created_at=existing.created_at,
            updated_at=existing.updated_at,
        )
        result = self._clip_repo.update(updated)
        logger.info("更新片段: id=%s", clip_id)
        return result

    def assign_asset(self, clip_id: str, asset_id: str) -> EditPlanClip:
        """为片段分配素材

        Raises:
            ValueError: 片段不存在或 asset_id 为空
        """
        clip = self.get_clip_or_raise(clip_id)
        # 自动从 completed/failed 切回 editing
        self._auto_resume_editing(clip.plan_id)
        clip.assign_asset(asset_id)
        result = self._clip_repo.update(clip)
        logger.info("分配素材: clip_id=%s asset_id=%s", clip_id, asset_id)
        return result

    def delete_clip(self, clip_id: str) -> bool:
        """删除片段

        Returns:
            bool: 是否删除成功
        """
        deleted = self._clip_repo.delete(clip_id)
        if deleted:
            logger.info("删除片段: id=%s", clip_id)
        return deleted

    def delete_all_clips(self, plan_id: str) -> int:
        """删除计划下所有片段

        Returns:
            int: 删除的片段数量
        """
        count = self._clip_repo.delete_by_plan(plan_id)
        logger.info("删除所有片段: plan_id=%s count=%d", plan_id, count)
        return count

    # ── 片段分割与合并 ──────────────────────────────────────────────────────

    def split_clip(self, clip_id: str, split_time: float) -> Dict[str, Any]:
        """将一个片段从指定位置分割为两个片段

        Args:
            clip_id: 要分割的片段 ID
            split_time: 分割点（相对于片段起始的秒数），必须在 (0, duration) 范围内

        Returns:
            dict: {"left_clip": EditPlanClip, "right_clip": EditPlanClip}

        Raises:
            ValueError: 片段不存在、分割时间越界
        """
        clip = self.get_clip_or_raise(clip_id)
        plan_id = clip.plan_id

        if split_time <= 0 or split_time >= clip.duration:
            raise ValueError(f"分割时间必须在 (0, {clip.duration:.3f}) 范围内，当前: {split_time}")

        self._auto_resume_editing(plan_id)

        original_duration = clip.duration
        left_duration = round(split_time, 3)
        right_duration = round(original_duration - split_time, 3)
        original_order = clip.order

        # 更新左半部分（原片段）
        clip.duration = left_duration
        left_clip = self._clip_repo.update(clip)

        # 后面片段的 order 全部 +1（给右半部分腾位置）
        all_clips = self._clip_repo.list_by_plan(plan_id)
        for c in all_clips:
            if c.order > original_order and c.id != clip_id:
                c.order += 1
                self._clip_repo.update(c)

        # 创建右半部分新片段（继承原片段的大部分属性）
        right_config = dict(clip.config) if clip.config else {}
        # 素材裁剪信息
        if clip.asset_id:
            # 右半部分从 split_time 开始播放
            right_config["trim_start"] = left_duration
            # 左半部分在 split_time 处结束
            left_config = dict(left_clip.config) if left_clip.config else {}
            left_config["trim_end"] = right_duration
            left_clip.config = left_config
            left_clip = self._clip_repo.update(left_clip)

        right_clip = EditPlanClip.create(
            plan_id=plan_id,
            clip_type=clip.clip_type,
            order=original_order + 1,
            template_clip_config_id=clip.template_clip_config_id,
            asset_id=clip.asset_id,
            text_content=clip.text_content,
            start_time=clip.start_time + left_duration,
            duration=right_duration,
            transition_effect=clip.transition_effect,
            transition_duration=clip.transition_duration,
            playback_speed=clip.playback_speed,
            config=right_config,
        )
        created_right = self._clip_repo.create(right_clip)

        logger.info(
            "分割片段: clip_id=%s plan_id=%s split_time=%.3fs left_dur=%.3fs right_dur=%.3fs",
            clip_id,
            plan_id,
            split_time,
            left_duration,
            right_duration,
        )

        return {
            "left_clip": left_clip,
            "right_clip": created_right,
        }

    def merge_clips(self, clip_ids: List[str]) -> EditPlanClip:
        """合并多个连续片段为一个片段

        Args:
            clip_ids: 要合并的片段 ID 列表（至少2个），必须属于同一个计划且 order 连续

        Returns:
            EditPlanClip: 合并后的新片段

        Raises:
            ValueError: 数量不足、不属于同一计划、不连续、类型不一致
        """
        if len(clip_ids) < 2:
            raise ValueError("至少需要 2 个片段才能合并")

        # 读取所有片段
        clips = []
        for cid in clip_ids:
            clip = self.get_clip_or_raise(cid)
            clips.append(clip)

        # 校验：同一计划
        plan_id = clips[0].plan_id
        for c in clips[1:]:
            if c.plan_id != plan_id:
                raise ValueError("只能合并同一计划下的片段")

        # 按 order 排序
        clips.sort(key=lambda c: c.order)

        # 校验：order 连续
        for i in range(1, len(clips)):
            if clips[i].order != clips[i - 1].order + 1:
                raise ValueError(f"片段不连续：order {clips[i-1].order} → {clips[i].order}")

        # 校验：类型一致
        clip_type = clips[0].clip_type
        for c in clips[1:]:
            if c.clip_type != clip_type:
                raise ValueError("只能合并相同类型的片段")

        self._auto_resume_editing(plan_id)

        # 计算合并后的属性
        first_clip = clips[0]
        total_duration = round(sum(c.duration for c in clips), 3)
        first_order = first_clip.order

        # 合并文案（用换行连接）
        merged_text = "\n".join(c.text_content for c in clips if c.text_content.strip())

        # 合并 config（后面的覆盖前面的）
        merged_config: Dict[str, Any] = {}
        for c in clips:
            if c.config:
                merged_config.update(c.config)
        # 清理 trim 相关字段（合并后就是完整片段了）
        merged_config.pop("trim_start", None)
        merged_config.pop("trim_end", None)

        # 更新第一个片段（保留它作为合并结果）
        first_clip.duration = total_duration
        first_clip.text_content = merged_text
        first_clip.config = merged_config
        # 转场保留第一个的（合并后的入点转场）
        # playback_speed 取第一个的
        merged_clip = self._clip_repo.update(first_clip)

        # 删除其余片段
        for c in clips[1:]:
            self._clip_repo.delete(c.id)

        # 后面的片段 order 前移 (len - 1) 位
        shift = len(clips) - 1
        all_clips = self._clip_repo.list_by_plan(plan_id)
        for c in all_clips:
            if c.order > first_order and c.id != merged_clip.id:
                c.order -= shift
                self._clip_repo.update(c)

        logger.info(
            "合并片段: plan_id=%s count=%d total_duration=%.3fs",
            plan_id,
            len(clips),
            total_duration,
        )

        return merged_clip

    # ── 渲染生成流程 ────────────────────────────────────────────────────────

    def get_generation_status(self, plan_id: str) -> Dict[str, Any]:
        """获取渲染进度状态

        Returns:
            dict: {
                "plan": EditPlan,
                "clips": List[EditPlanClip],
                "generation_task_id": Optional[str],
                "generation_task_status": Optional[str],
                "progress": float,
                "error_message": str,
            }

        Raises:
            ValueError: 计划不存在
        """
        plan = self.get_plan_or_raise(plan_id)
        clips = self._clip_repo.list_by_plan(plan_id)

        # 从 plan.config 中获取 generation_task_id
        generation_task_id = plan.config.get("generation_task_id")
        generation_task_status = None
        progress = 0.0
        error_message = ""

        if generation_task_id:
            task = self._generation_task_repo.get(generation_task_id)
            if task:
                generation_task_status = task.status.value if hasattr(task.status, "value") else task.status
                progress = getattr(task, "progress", 0.0) or 0.0
                error_message = getattr(task, "error_message", "") or ""

        return {
            "plan": plan,
            "clips": clips,
            "generation_task_id": generation_task_id,
            "generation_task_status": generation_task_status,
            "progress": progress,
            "error_message": error_message,
        }

    def can_generate(self, plan_id: str) -> tuple[bool, str]:
        """检查是否可以触发渲染

        Returns:
            tuple: (can_generate, reason)
        """
        plan = self.get_plan_or_raise(plan_id)

        # 检查状态
        if plan.status != EditPlanStatus.EDITING:
            return False, "请先编辑并保存模板后再生成视频"

        # 检查是否有片段
        clips = self._clip_repo.list_by_plan(plan_id)
        if not clips:
            return False, "请先添加片段后再生成视频"

        return True, ""

    def mark_clips_ready(self, plan_id: str) -> int:
        """将所有 pending 状态的片段标记为 ready

        Returns:
            int: 标记的片段数量
        """
        clips = self._clip_repo.list_by_plan(
            plan_id,
            status=EditPlanClipStatus.PENDING,
        )
        count = 0
        for clip in clips:
            clip.mark_ready()
            self._clip_repo.update(clip)
            count += 1
        logger.info("标记片段就绪: plan_id=%s count=%d", plan_id, count)
        return count

    def update_plan_config(self, plan_id: str, config_updates: Dict[str, Any]) -> EditPlan:
        """更新计划配置（合并更新）

        Args:
            plan_id: 计划 ID
            config_updates: 要合并的配置

        Returns:
            更新后的计划
        """
        plan = self.get_plan_or_raise(plan_id)
        # 自动从 completed/failed 切回 editing
        self._auto_resume_editing(plan_id)
        plan = self.get_plan_or_raise(plan_id)
        new_config = {**plan.config, **config_updates}

        updated = EditPlan(
            id=plan.id,
            template_id=plan.template_id,
            name=plan.name,
            status=plan.status,
            total_duration=plan.total_duration,
            source_edit_plan_id=plan.source_edit_plan_id,
            project_id=plan.project_id,
            created_by_user_id=plan.created_by_user_id,
            config=new_config,
            created_at=plan.created_at,
            updated_at=plan.updated_at,
        )
        return self._plan_repo.update(updated)