From f5bc1b8947c1ca2419b99b0bec17abc97f5fe6cb Mon Sep 17 00:00:00 2001
From: CI Test <ci-test@xiaoxia.com>
Date: Fri, 26 Jun 2026 21:21:39 +0800
Subject: [PATCH] chore: squash merge feature/smart-edit-plan into develop
 (resolve conflicts)

- Resolve CI/CD and deploy workflow conflicts with develop version
- Resolve database and dependencies conflicts with develop version
- Keep new edit plan generator and API functionality from feature branch
---
 apps/api/app/api/routes/edit_plans.py         | 108 ++++-
 apps/api/app/schemas/edit_plan.py             |  18 +
 .../worker_app/tasks/edit_plan_generator.py   | 386 ++++++++++++++++++
 packages/adapters/sqlalchemy_impl/models.py   |   5 +
 4 files changed, 515 insertions(+), 2 deletions(-)
 create mode 100644 apps/worker/worker_app/tasks/edit_plan_generator.py

diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py
index 9488f5125..41092a33a 100644
--- a/apps/api/app/api/routes/edit_plans.py
+++ b/apps/api/app/api/routes/edit_plans.py
@@ -9,7 +9,17 @@ from app.dependencies import (
     get_project_repository,
     get_workspace_member_repository,
 )
-from app.schemas.edit_plan import CreateEditPlanRequest, EditPlanClipResponse, EditPlanResponse, EditTemplateResponse
+from app.schemas.edit_plan import (
+    AutoGenerateEditPlanRequest,
+    CreateEditPlanRequest,
+    EditPlanClipResponse,
+    EditPlanResponse,
+    EditTemplateResponse,
+)
+from apps.worker.worker_app.tasks.edit_plan_generator import (
+    EditingMode,
+    SmartEditPlanGenerator,
+)
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
 
@@ -81,6 +91,7 @@ def _to_plan_response(
         title_id=plan.title_id,
         status=plan.status,
         summary=plan.summary,
+        editing_mode=plan.editing_mode,
         clips=[
             EditPlanClipResponse(
                 id=clip.id,
@@ -90,6 +101,7 @@ def _to_plan_response(
                 start_time=float(clip.start_time or 0),
                 duration=float(clip.duration or 0),
                 reason=clip.reason,
+                layer=clip.layer,
             )
             for clip in clips
         ],
@@ -98,7 +110,7 @@ def _to_plan_response(
     )
 
 
-@router.get("/templates", response_model=list[EditTemplateResponse])
+@router.get("/templates/", response_model=list[EditTemplateResponse])
 def list_edit_templates(
     project_id: str,
     workspace_id: str,
@@ -178,6 +190,98 @@ def create_edit_plan(
     return _to_plan_response(plan, clips, {asset.id: asset.name for asset in selected})
 
 
+@router.post("/auto-generate", response_model=EditPlanResponse)
+def auto_generate_edit_plan(
+    project_id: str,
+    request: AutoGenerateEditPlanRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    project_repository=Depends(get_project_repository),
+    asset_repository=Depends(get_asset_repository),
+    workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository),
+    session: Session = Depends(get_db_session),
+) -> EditPlanResponse:
+    """
+    智能生成剪辑计划
+
+    根据素材的分类结果和质量评分，自动编排剪辑计划。
+    支持多种剪辑模式：
+    - one-take: 按分类分组，组内按质量排序，顺序拼接
+    - pip: 第一个高质量素材为主画面，其余为画中画
+    - voiceover: person 类素材为主播口播，其余穿插为 B-roll
+    - voice_pip: 结合 voiceover 和 pip，第一个高质量 person 素材为主画面
+    """
+    _ensure_project(
+        project_id, request.workspace_id, authenticated_user, project_repository, workspace_member_repository
+    )
+
+    # 获取素材库中的所有素材
+    assets = asset_repository.list_by_library(request.asset_library_id)
+    
+    if not assets:
+        raise HTTPException(status_code=422, detail="素材库中暂无素材")
+
+    # 使用智能生成器
+    generator = SmartEditPlanGenerator(project_id, assets)
+    
+    try:
+        plan_result = generator.generate_plan(
+            editing_mode=request.editing_mode,
+            target_duration=request.target_duration
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+    if not plan_result.clips:
+        raise HTTPException(status_code=422, detail="无符合条件的视频素材")
+
+    # 获取模板
+    template = (
+        session.query(EditTemplateModel).filter(EditTemplateModel.id == request.template_id).first()
+        if request.template_id
+        else None
+    )
+    if template is None:
+        template = _default_template(session, request.workspace_id, project_id, authenticated_user.user.id)
+
+    # 创建剪辑计划
+    plan = EditPlanModel(
+        id=uuid4().hex,
+        workspace_id=request.workspace_id,
+        project_id=project_id,
+        template_id=template.id,
+        asset_library_id=request.asset_library_id,
+        title_id=request.title_id or "",
+        status="draft",
+        editing_mode=request.editing_mode,
+        summary=plan_result.summary,
+        created_by_user_id=authenticated_user.user.id,
+        updated_at=datetime.now(timezone.utc),
+    )
+    session.add(plan)
+
+    # 创建剪辑片段
+    clips: list[EditPlanClipModel] = []
+    asset_name_map = {asset.id: asset.name for asset in assets}
+    
+    for clip_plan in plan_result.clips:
+        clip = EditPlanClipModel(
+            id=uuid4().hex,
+            edit_plan_id=plan.id,
+            asset_id=clip_plan.asset_id,
+            sequence=clip_plan.sequence,
+            start_time=clip_plan.start_time,
+            duration=clip_plan.duration,
+            reason=clip_plan.reason,
+            layer=clip_plan.layer,
+        )
+        session.add(clip)
+        clips.append(clip)
+
+    session.commit()
+
+    return _to_plan_response(plan, clips, asset_name_map)
+
+
 @router.get("/{plan_id}", response_model=EditPlanResponse)
 def get_edit_plan(
     project_id: str,
diff --git a/apps/api/app/schemas/edit_plan.py b/apps/api/app/schemas/edit_plan.py
index 445659fd2..1d3c07571 100644
--- a/apps/api/app/schemas/edit_plan.py
+++ b/apps/api/app/schemas/edit_plan.py
@@ -23,6 +23,7 @@ class EditPlanClipResponse(BaseModel):
     start_time: float
     duration: float
     reason: str
+    layer: str = "main"  # main, pip, broll
 
 
 class EditPlanResponse(BaseModel):
@@ -34,6 +35,7 @@ class EditPlanResponse(BaseModel):
     title_id: str = ""
     status: str
     summary: str
+    editing_mode: str | None = None  # one-take, pip, voiceover, voice_pip
     clips: list[EditPlanClipResponse] = Field(default_factory=list)
     created_at: datetime | None = None
     updated_at: datetime | None = None
@@ -44,3 +46,19 @@ class CreateEditPlanRequest(BaseModel):
     asset_library_id: str
     template_id: str = ""
     title_id: str = ""
+
+
+class AutoGenerateEditPlanRequest(BaseModel):
+    """智能生成剪辑计划请求"""
+    workspace_id: str
+    asset_library_id: str
+    editing_mode: str = Field(
+        default="one-take",
+        description="剪辑模式: one-take, pip, voiceover, voice_pip"
+    )
+    target_duration: float = Field(
+        default=30.0,
+        description="目标时长（秒）"
+    )
+    template_id: str = ""
+    title_id: str = ""
diff --git a/apps/worker/worker_app/tasks/edit_plan_generator.py b/apps/worker/worker_app/tasks/edit_plan_generator.py
new file mode 100644
index 000000000..3634d8fd3
--- /dev/null
+++ b/apps/worker/worker_app/tasks/edit_plan_generator.py
@@ -0,0 +1,386 @@
+"""Smart Edit Plan Generator - 根据分类和质量评分智能编排剪辑计划"""
+
+import json
+import logging
+from collections import defaultdict
+from dataclasses import dataclass
+from enum import StrEnum
+from typing import Any
+
+from packages.domain import Asset, AssetClassification, AssetStatus
+
+logger = logging.getLogger(__name__)
+
+
+class EditingMode(StrEnum):
+    """剪辑模式枚举"""
+    ONE_TAKE = "one-take"   # 顺序拼接模式
+    PIP = "pip"             # 画中画模式
+    VOICEOVER = "voiceover" # 口播+B-roll模式
+    VOICE_PIP = "voice_pip" # 口播+画中画组合模式
+
+
+@dataclass
+class EditClipPlan:
+    """单个剪辑片段的编排计划"""
+    asset_id: str
+    sequence: int
+    start_time: float = 0.0
+    duration: float = 0.0
+    layer: str = "main"  # main, pip, broll
+    reason: str = ""
+
+
+@dataclass
+class EditPlanResult:
+    """完整剪辑计划结果"""
+    project_id: str
+    editing_mode: EditingMode
+    clips: list[EditClipPlan]
+    total_duration: float
+    summary: str
+
+
+def _calculate_start_times(clips: list[EditClipPlan]) -> list[EditClipPlan]:
+    """
+    计算时间轴，根据前面的片段时长累加 start_time
+    
+    Args:
+        clips: 已按 sequence 排序的片段列表
+        
+    Returns:
+        修正后的片段列表
+    """
+    current_time = 0.0
+    for clip in clips:
+        clip.start_time = current_time
+        current_time += clip.duration
+    return clips
+
+
+class SmartEditPlanGenerator:
+    """
+    智能剪辑计划生成器
+
+    根据素材的分类结果和质量评分，自动编排剪辑计划。
+    支持多种剪辑模式：one-take, pip, voiceover, 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.VOICEOVER,
+            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/voiceover/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.VOICEOVER:
+            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}")
diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py
index a41065746..fc9fb98f5 100755
--- a/packages/adapters/sqlalchemy_impl/models.py
+++ b/packages/adapters/sqlalchemy_impl/models.py
@@ -156,6 +156,7 @@ class EditPlanModel(Base):
     template_id = Column(String(32), nullable=False, index=True)
     asset_library_id = Column(String(32), nullable=False, index=True)
     title_id = Column(String(32), nullable=False, default="")
+    editing_mode = Column(String(20), nullable=True, default=None, index=True)
     status = Column(String(20), nullable=False, default="draft", index=True)
     summary = Column(Text, nullable=False, default="")
     created_by_user_id = Column(String(32), nullable=False, default="")
@@ -173,6 +174,7 @@ class EditPlanClipModel(Base):
     start_time = Column(Float, nullable=False, default=0)
     duration = Column(Float, nullable=False, default=0)
     reason = Column(Text, nullable=False, default="")
+    layer = Column(String(20), nullable=False, default="main")
 
 
 class IngestJobModel(Base):
@@ -247,6 +249,9 @@ class GeneratedVideoModel(Base):
     generated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
     updated_at = Column(DateTime, nullable=True)
+    video_fingerprint = Column(Text, nullable=True)
+    is_duplicate = Column(Boolean, nullable=False, default=False)
+    duplicate_of = Column(String(32), nullable=True)
 
 
 class TaskModel(Base):