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 list_plans(
        self,
        *,
        template_id: Optional[str] = None,
        project_id: Optional[str] = None,
        status: Optional[EditPlanStatus] = None,
        skip: int = 0,
        limit: int = 50,
    ) -> List[EditPlan]:
        """列出剪辑计划

        Args:
            template_id: 按模板 ID 筛选
            project_id: 按项目 ID 筛选
            status: 按状态筛选
            skip: 分页偏移
            limit: 每页数量
        """
        if project_id:
            return self._plan_repo.list_by_project(
                project_id,
                status=status,
                skip=skip,
                limit=limit,
            )
        if template_id:
            return self._plan_repo.list_by_template(
                template_id,
                status=status,
                skip=skip,
                limit=limit,
            )
        return self._plan_repo.list_all(status=status, skip=skip, limit=limit)

    def count_plans(
        self,
        *,
        template_id: Optional[str] = None,
        project_id: Optional[str] = None,
        status: Optional[EditPlanStatus] = None,
    ) -> int:
        """统计计划数量

        Note:
            当指定 template_id/project_id 时，通过全量查询计算 total（repo 限制）。
        """
        if project_id:
            all_matching = self._plan_repo.list_by_project(
                project_id,
                status=status,
                skip=0,
                limit=10000,
            )
            return len(all_matching)
        if template_id:
            all_matching = self._plan_repo.list_by_template(
                template_id,
                status=status,
                skip=0,
                limit=10000,
            )
            return len(all_matching)
        return self._plan_repo.count(status=status)

    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:
        """创建剪辑计划

        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 delete_plan(self, plan_id: str) -> bool:
        """删除剪辑计划及其所有片段

        Returns:
            bool: 是否删除成功
        """
        existing = self._plan_repo.get(plan_id)
        if existing is None:
            return False

        # 先删除所有片段
        self._clip_repo.delete_by_plan(plan_id)
        # 再删除计划
        self._plan_repo.delete(plan_id)
        logger.info("删除剪辑计划: id=%s", plan_id)
        return True

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

    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 create_clips_from_assets(
        self,
        plan_id: str,
        asset_ids: list[str],
        *,
        clip_type: str = "main",
    ) -> list[EditPlanClip]:
        """从素材批量创建片段（追加到时间线末尾）。

        Args:
            plan_id: 计划 ID
            asset_ids: 素材 ID 列表（按顺序追加）
            clip_type: 片段类型

        Returns:
            list[EditPlanClip]: 创建的片段列表
        """
        if not asset_ids:
            return []

        # 确保计划存在 + 自动回退状态
        self.get_plan_or_raise(plan_id)
        self._auto_resume_editing(plan_id)

        # 查询素材信息（取 duration）
        from packages.adapters.sqlalchemy_impl.models import AssetModel

        session = self._clip_repo.session  # type: ignore[attr-defined]
        assets = session.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).all()
        asset_map = {a.id: a for a in assets}

        # 从现有片段数量开始追加
        existing_count = self._clip_repo.count(plan_id=plan_id)

        # 批量创建片段
        created: list[EditPlanClip] = []
        for i, asset_id in enumerate(asset_ids):
            asset = asset_map.get(asset_id)
            duration = asset.duration if asset and asset.duration else 0.0

            clip = self.create_clip(
                plan_id=plan_id,
                clip_type=clip_type,
                order=existing_count + i,
                asset_id=asset_id,
                duration=duration,
            )
            created.append(clip)

        logger.info(
            "从素材批量创建片段: plan_id=%s count=%d",
            plan_id,
            len(created),
        )
        return created

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

    def get_plan_with_clips(self, plan_id: str) -> Dict[str, Any]:
        """获取计划及其所有片段

        Returns:
            dict: {"plan": EditPlan, "clips": List[EditPlanClip]}
        """
        plan = self.get_plan_or_raise(plan_id)
        clips = self._clip_repo.list_by_plan(plan_id)
        return {
            "plan": plan,
            "clips": clips,
        }

    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)

    # ── 复制计划 ────────────────────────────────────────────────────────────

    def copy_plan(
        self,
        plan_id: str,
        *,
        new_name: Optional[str] = None,
        project_id: Optional[str] = None,
    ) -> EditPlan:
        """复制一个剪辑计划（含所有片段配置）。

        新计划状态为 editing，不含生成任务和结果记录。

        Args:
            plan_id: 源计划 ID
            new_name: 新计划名称，不传则为「原名 - 副本」
            project_id: 新计划的项目 ID，不传则复用源计划

        Returns:
            EditPlan: 新创建的计划

        Raises:
            ValueError: 源计划不存在
        """
        source = self.get_plan_or_raise(plan_id)
        source_clips = self._clip_repo.list_by_plan(plan_id)

        # 新计划名称
        name = new_name or f"{source.name} - 副本"
        new_project_id = project_id if project_id is not None else source.project_id

        # 复制 plan 配置（去除渲染结果相关字段）
        new_config = dict(source.config)
        new_config.pop("rendered_url", None)
        new_config.pop("rendered_storage_key", None)
        new_config.pop("generation_task_id", None)

        # 创建新计划
        new_plan = EditPlan.create(
            template_id=source.template_id,
            name=name,
            config=new_config,
            total_duration=source.total_duration,
            project_id=new_project_id,
            created_by_user_id=source.created_by_user_id,
            source_edit_plan_id=plan_id,
        )
        # 强制切到 editing 状态
        if new_plan.status != EditPlanStatus.EDITING:
            try:
                new_plan.start_editing()
            except ValueError:
                pass

        created_plan = self._plan_repo.create(new_plan)
        logger.info(
            "复制剪辑计划: source=%s target=%s name=%s clips=%d",
            plan_id,
            created_plan.id,
            name,
            len(source_clips),
        )

        # 复制所有片段
        for clip in source_clips:
            new_clip = self.create_clip(
                plan_id=created_plan.id,
                clip_type=clip.clip_type,
                order=clip.order,
                asset_id=clip.asset_id or "",
                text_content=clip.text_content or "",
                start_time=clip.start_time,
                duration=clip.duration,
                transition_effect=clip.transition_effect or "cut",
                transition_duration=clip.transition_duration or 0.0,
                playback_speed=clip.playback_speed or 1.0,
                config=dict(clip.config) if clip.config else None,
            )
            logger.debug("复制片段: source=%s target=%s order=%d", clip.id, new_clip.id, clip.order)

        return self.get_plan_or_raise(created_plan.id)