Merge pull request 'fix: 修复部署v0.1.64 - SmartEditPlanGenerator迁移 + 移除workspaces死路由' (#45) from fix/deployment-v0.1.64-hotfix into develop

apps/api/app/api/router.py

@@ -14,7 +14,6 @@ from app.api.routes.project_titles import router as project_titles_router
 from app.api.routes.projects import router as projects_router
 from app.api.routes.task_center import router as task_center_router
 from app.api.routes.upload import router as upload_router
-from app.api.routes.workspaces import router as workspaces_router
 from fastapi import APIRouter
 
 api_router = APIRouter(prefix="/api/v1")
@@ -25,10 +24,6 @@ api_router.include_router(
     auth_router,
     tags=["Auth"],
 )
-api_router.include_router(
-    workspaces_router,
-    tags=["Workspace"],
-)
 api_router.include_router(
     projects_router,
     prefix="/projects",

packages/domain/__init__.py

@@ -20,7 +20,6 @@ from .entities import (
     IngestJobStatus,
     Project,
     User,
-    Workspace,
 )
 from .generated_video import GeneratedVideo
 from .generation_task import GenerationTask, GenerationTaskStatus
@@ -50,5 +49,4 @@ __all__ = [
     "TaskPriority",
     "TaskStatus",
     "User",
-    "Workspace",
 ]

packages/domain/edit_plan.py

@@ -31,3 +31,349 @@ class EditPlanResult:
     clips: list[EditClipPlan]
     total_duration: float
     summary: str
+
+
+
+
+import json as _json
+import logging as _logging
+from collections import defaultdict as _defaultdict
+from packages.domain.entities import Asset, AssetStatus
+from packages.domain.classification import AssetClassification
+
+_logger = _logging.getLogger(__name__)
+
+def _calculate_start_times(clips):
+    current_time = 0.0
+    for clip in clips:
+        clip.start_time = current_time
+        current_time += clip.duration
+    return clips
+
+
+class SmartEditPlanGenerator:
+    """
+    智能剪辑计划生成器
+
+    根据素材的分类结果和质量评分，自动编排剪辑计划。
+    支持多种剪辑模式：one_take, pip, voice_over, voice_pip
+    """
+
+    def __init__(self, project_id: str, assets: list[Asset]):
+        self.project_id = project_id
+        # 筛选已就绪的视频素材
+        self.assets = [
+            a for a in assets
+            if a.status == AssetStatus.READY and a.mime_type.startswith("video/")
+        ]
+        self.assets_by_classification: dict[str, list[Asset]] = defaultdict(list)
+
+    def _parse_classification(self, asset: Asset) -> str:
+        """解析素材的分类结果"""
+        # 从 metadata 中获取分类
+        classification = asset.metadata.get("classification", "")
+        if not classification:
+            # 尝试从 classification_result 字段获取
+            classification = asset.metadata.get("classification_result", "")
+        
+        # 如果是 JSON 字符串，解析它
+        if classification and isinstance(classification, str):
+            try:
+                parsed = json.loads(classification)
+                if isinstance(parsed, dict):
+                    classification = parsed.get("classification", "other")
+                elif isinstance(parsed, str):
+                    classification = parsed
+            except (json.JSONDecodeError, TypeError):
+                pass
+        
+        # 验证分类值是否有效
+        valid_classifications = [c.value for c in AssetClassification]
+        if classification not in valid_classifications:
+            classification = "other"
+        
+        return classification
+
+    def _group_by_classification(self) -> None:
+        """按分类结果对素材分组"""
+        for asset in self.assets:
+            classification = self._parse_classification(asset)
+            self.assets_by_classification[classification].append(asset)
+
+    def _sort_by_quality(self, assets: list[Asset]) -> list[Asset]:
+        """按质量评分排序，高分在前"""
+        return sorted(
+            assets,
+            key=lambda a: (-(a.quality_score or 0), a.created_at)
+        )
+
+    def _calculate_clip_duration(self, asset: Asset, target_duration: float, clip_count: int) -> float:
+        """计算单个片段的时长"""
+        if asset.duration:
+            # 如果素材时长超过平均时长，取平均时长
+            avg_duration = target_duration / max(1, clip_count)
+            return min(float(asset.duration), avg_duration)
+        return target_duration / max(1, clip_count)
+
+    def _generate_one_take(self, target_duration: float = 30.0) -> EditPlanResult:
+        """
+        One-Take 模式：按分类分组，组内按质量排序，顺序拼接
+        """
+        self._group_by_classification()
+        
+        clips: list[EditClipPlan] = []
+        sequence = 1
+        
+        # 按优先级排序分类：person > scenic > product > other
+        priority_order = ["person", "scenic", "product", "animal", "food", "tech", "sport", "music", "other"]
+        sorted_classifications = sorted(
+            self.assets_by_classification.keys(),
+            key=lambda c: priority_order.index(c) if c in priority_order else len(priority_order)
+        )
+        
+        for classification in sorted_classifications:
+            sorted_assets = self._sort_by_quality(self.assets_by_classification[classification])
+            for asset in sorted_assets:
+                duration = self._calculate_clip_duration(
+                    asset, target_duration, len(self.assets)
+                )
+                clips.append(EditClipPlan(
+                    asset_id=asset.id,
+                    sequence=sequence,
+                    start_time=0,
+                    duration=duration,
+                    layer="main",
+                    reason=f"按分类 [{classification}] 排列，质量评分 {asset.quality_score or 0:.1f}"
+                ))
+                sequence += 1
+        
+        total_duration = sum(c.duration for c in clips)
+        _calculate_start_times(clips)
+        return EditPlanResult(
+            project_id=self.project_id,
+            editing_mode=EditingMode.ONE_TAKE,
+            clips=clips,
+            total_duration=total_duration,
+            summary=f"One-Take 模式：按 {len(sorted_classifications)} 个分类分组，共 {len(clips)} 段素材"
+        )
+
+    def _generate_pip(self, target_duration: float = 30.0) -> EditPlanResult:
+        """
+        PIP 模式：第一个高质量素材为主画面，其余为画中画
+        """
+        sorted_assets = self._sort_by_quality(self.assets)
+        
+        if not sorted_assets:
+            return EditPlanResult(
+                project_id=self.project_id,
+                editing_mode=EditingMode.PIP,
+                clips=[],
+                total_duration=0,
+                summary="无素材可用"
+            )
+        
+        clips: list[EditClipPlan] = []
+        sequence = 1
+        
+        # 第一个高质量素材作为主画面
+        main_asset = sorted_assets[0]
+        main_duration = min(
+            float(main_asset.duration) if main_asset.duration else target_duration,
+            target_duration
+        )
+        clips.append(EditClipPlan(
+            asset_id=main_asset.id,
+            sequence=sequence,
+            start_time=0,
+            duration=main_duration,
+            layer="main",
+            reason=f"高质量主画面 (质量评分: {main_asset.quality_score or 0:.1f})"
+        ))
+        sequence += 1
+        
+        # 其余素材作为画中画
+        for asset in sorted_assets[1:]:
+            duration = self._calculate_clip_duration(asset, target_duration, len(sorted_assets))
+            clips.append(EditClipPlan(
+                asset_id=asset.id,
+                sequence=sequence,
+                start_time=0,
+                duration=duration,
+                layer="pip",
+                reason=f"画中画素材 (质量评分: {asset.quality_score or 0:.1f})"
+            ))
+            sequence += 1
+        
+        total_duration = main_duration
+        _calculate_start_times(clips)
+        return EditPlanResult(
+            project_id=self.project_id,
+            editing_mode=EditingMode.PIP,
+            clips=clips,
+            total_duration=total_duration,
+            summary=f"PIP 模式：1 个主画面 + {len(sorted_assets) - 1} 个画中画"
+        )
+
+    def _generate_voiceover(self, target_duration: float = 30.0) -> EditPlanResult:
+        """
+        Voiceover 模式：person 类素材为主播口播，其余穿插为 B-roll
+        """
+        self._group_by_classification()
+        
+        person_assets = self._sort_by_quality(
+            self.assets_by_classification.get("person", [])
+        )
+        other_assets = self._sort_by_quality([
+            a for assets in self.assets_by_classification.values()
+            for a in assets
+            if self._parse_classification(a) != "person"
+        ])
+        
+        clips: list[EditClipPlan] = []
+        sequence = 1
+        
+        # 合并口播和 B-roll
+        main_assets = person_assets if person_assets else other_assets
+        broll_assets = [a for a in other_assets if a not in person_assets] if person_assets else []
+        
+        # 优先使用 person 素材作为口播
+        for i, asset in enumerate(main_assets):
+            duration = self._calculate_clip_duration(asset, target_duration, len(main_assets))
+            is_person = asset in person_assets
+            clips.append(EditClipPlan(
+                asset_id=asset.id,
+                sequence=sequence,
+                start_time=0,
+                duration=duration,
+                layer="main" if is_person else "broll",
+                reason=f"{'主播口播' if is_person else 'B-roll'} (质量评分: {asset.quality_score or 0:.1f})"
+            ))
+            sequence += 1
+            
+            # 在口播之间穿插 B-roll
+            if is_person and broll_assets and i < len(main_assets) - 1:
+                broll_asset = broll_assets[i % len(broll_assets)]
+                broll_duration = self._calculate_clip_duration(
+                    broll_asset, target_duration, len(main_assets) + len(broll_assets)
+                )
+                clips.append(EditClipPlan(
+                    asset_id=broll_asset.id,
+                    sequence=sequence,
+                    start_time=0,
+                    duration=broll_duration,
+                    layer="broll",
+                    reason=f"B-roll 穿插 (质量评分: {broll_asset.quality_score or 0:.1f})"
+                ))
+                sequence += 1
+        
+        total_duration = sum(c.duration for c in clips)
+        person_count = len(person_assets)
+        _calculate_start_times(clips)
+        return EditPlanResult(
+            project_id=self.project_id,
+            editing_mode=EditingMode.VOICE_OVER,
+            clips=clips,
+            total_duration=total_duration,
+            summary=f"Voiceover 模式：{person_count} 段口播 + {len(clips) - person_count} 段 B-roll"
+        )
+
+    def _generate_voice_pip(self, target_duration: float = 30.0) -> EditPlanResult:
+        """
+        Voice-PIP 模式：结合 voiceover 和 pip
+        第一个高质量 person 素材为主画面，其余为 PIP B-roll
+        """
+        self._group_by_classification()
+        
+        person_assets = self._sort_by_quality(
+            self.assets_by_classification.get("person", [])
+        )
+        other_assets = self._sort_by_quality([
+            a for assets in self.assets_by_classification.values()
+            for a in assets
+            if self._parse_classification(a) != "person"
+        ])
+        
+        clips: list[EditClipPlan] = []
+        sequence = 1
+        
+        # 主画面：优先使用高质量 person 素材
+        main_asset = person_assets[0] if person_assets else (other_assets[0] if other_assets else None)
+        if main_asset:
+            main_duration = min(
+                float(main_asset.duration) if main_asset.duration else target_duration,
+                target_duration
+            )
+            is_person = main_asset in person_assets
+            clips.append(EditClipPlan(
+                asset_id=main_asset.id,
+                sequence=sequence,
+                start_time=0,
+                duration=main_duration,
+                layer="main",
+                reason=f"{'主播口播' if is_person else '主画面'} (质量评分: {main_asset.quality_score or 0:.1f})"
+            ))
+            sequence += 1
+        
+        # PIP 素材
+        pip_assets = [a for a in (person_assets[1:] + other_assets) if a != main_asset]
+        for asset in pip_assets:
+            duration = self._calculate_clip_duration(asset, target_duration, len(pip_assets) + 1)
+            clips.append(EditClipPlan(
+                asset_id=asset.id,
+                sequence=sequence,
+                start_time=0,
+                duration=duration,
+                layer="pip",
+                reason=f"PIP 素材 (质量评分: {asset.quality_score or 0:.1f})"
+            ))
+            sequence += 1
+        
+        total_duration = sum(c.duration for c in clips)
+        _calculate_start_times(clips)
+        return EditPlanResult(
+            project_id=self.project_id,
+            editing_mode=EditingMode.VOICE_PIP,
+            clips=clips,
+            total_duration=total_duration,
+            summary=f"Voice-PIP 模式：1 个主画面 + {len(pip_assets)} 个 PIP 素材"
+        )
+
+    def generate_plan(
+        self,
+        editing_mode: str = "one_take",
+        target_duration: float = 30.0
+    ) -> EditPlanResult:
+        """
+        生成剪辑计划
+
+        Args:
+            editing_mode: 剪辑模式 (one_take/pip/voice_over/voice_pip)
+            target_duration: 目标时长（秒）
+
+        Returns:
+            EditPlanResult: 编排好的剪辑计划
+        """
+        logger.info(f"Generating edit plan for project {self.project_id} with mode {editing_mode}")
+        
+        if not self.assets:
+            logger.warning(f"No ready video assets found for project {self.project_id}")
+            return EditPlanResult(
+                project_id=self.project_id,
+                editing_mode=EditingMode(editing_mode),
+                clips=[],
+                total_duration=0,
+                summary="无素材可用"
+            )
+        
+        mode = EditingMode(editing_mode.lower())
+        
+        if mode == EditingMode.ONE_TAKE:
+            return self._generate_one_take(target_duration)
+        elif mode == EditingMode.PIP:
+            return self._generate_pip(target_duration)
+        elif mode == EditingMode.VOICE_OVER:
+            return self._generate_voiceover(target_duration)
+        elif mode == EditingMode.VOICE_PIP:
+            return self._generate_voice_pip(target_duration)
+        else:
+            raise ValueError(f"Unknown editing mode: {editing_mode}")