From 4db830b247bb80822c3049533cfa70ec1196ec2a Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sat, 27 Jun 2026 12:09:16 +0800 Subject: [PATCH 1/2] fix: remove Workspace from domain __init__ (class removed but import remained) --- packages/domain/__init__.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/packages/domain/__init__.py b/packages/domain/__init__.py index cf6e71499..c068bb87f 100644 --- a/packages/domain/__init__.py +++ b/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", ] -- 2.54.0 From 4068a9141305f840e90b49cdee84c3c6f9b4f8ab Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sat, 27 Jun 2026 12:44:04 +0800 Subject: [PATCH 2/2] fix: move SmartEditPlanGenerator to domain layer, remove dead workspaces route --- apps/api/app/api/router.py | 5 - packages/domain/edit_plan.py | 346 +++++++++++++++++++++++++++++++++++ 2 files changed, 346 insertions(+), 5 deletions(-) diff --git a/apps/api/app/api/router.py b/apps/api/app/api/router.py index a7c808825..0b7decf45 100644 --- a/apps/api/app/api/router.py +++ b/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", diff --git a/packages/domain/edit_plan.py b/packages/domain/edit_plan.py index c87564b64..7faf5d8db 100644 --- a/packages/domain/edit_plan.py +++ b/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}") -- 2.54.0