fix(worker): 修复瘦身后PYTHONPATH缺少packages目录导致shared模块无法导入

2026-07-15 08:30:58 +08:00
96 changed files with 1191 additions and 4801 deletions
@@ -1,53 +0,0 @@
 name: AI Code Review
 on:
  pull_request:
    types:
      - opened
      - synchronize
      - reopened
 # 同一个 PR 只跑一个 review，新的取消旧的
 concurrency:
  group: code-review-${{ gitea.repository }}-${{ gitea.event.pull_request.number }}
  cancel-in-progress: true
 jobs:
  code-review:
    name: AI Code Review
    runs-on: ubuntu-latest
    # 跳过草稿 PR
    if: ${{ !gitea.event.pull_request.draft }}
    steps:
      - name: Checkout
        uses: actions/checkout@v3
        with:
          fetch-depth: 0
      - name: Install dependencies
        run: |
          python3 -m pip install --upgrade pip
          python3 -m pip install requests
      - name: Run AI Code Review
        env:
          # Gitea 配置（自动从运行环境获取）
          GITEA_API_URL: ${{ gitea.server_url }}
          GITEA_TOKEN: ${{ secrets.REVIEW_GITEA_TOKEN }}
          REPO_NAME: ${{ gitea.repository }}
          PR_NUMBER: ${{ gitea.event.pull_request.number }}
          # LLM 提供商: coze (扣子原生Bot) / openai (OpenAI兼容)
          LLM_PROVIDER: "coze"
          # 扣子模式配置（默认国内站 api.coze.cn）
          LLM_BASE_URL: ${{ secrets.LLM_BASE_URL }}
          LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
          COZE_BOT_ID: ${{ secrets.COZE_BOT_ID }}
          LLM_MODEL: ${{ secrets.LLM_MODEL }}
          # 可选参数
          MAX_DIFF_CHARS: "30000"
          LLM_TIMEOUT: "120"
        run: |
          python3 scripts/ci_code_review.py
        # 审查脚本异常不影响 CI 通过
        continue-on-error: true
@@ -1,29 +0,0 @@
 """add result_count to edit_plans
 Revision ID: 041_result_count
 Revises: 040_playback_speed
 Create Date: 2026-07-15 14:05:00.000000
 """
 from __future__ import annotations
 import sqlalchemy as sa
 from alembic import op
 # revision identifiers, used by Alembic.
 revision = "041_result_count"
 down_revision = "040_playback_speed"
 branch_labels = None
 depends_on = None
 def upgrade() -> None:
    op.add_column(
        "edit_plans",
        sa.Column("result_count", sa.Integer(), nullable=False, server_default="0"),
    )
 def downgrade() -> None:
    op.drop_column("edit_plans", "result_count")
@@ -239,9 +239,7 @@ def get_duplication_detail(
    return _to_detail_response(record)
-@router.delete(
+@router.delete("/records/{record_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None, response_class=Response)
    "/records/{record_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None, response_class=Response
 )
 def delete_duplication_record(
    record_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
@@ -289,7 +287,7 @@ def retry_duplication(
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
-        ) from e
+        )
    if updated is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
@@ -24,7 +24,7 @@ from typing import Any, List, Optional
 from app.auth import AuthenticatedUser, get_current_user
 from app.dependencies import get_db_session, get_project_repository
 from app.schemas.generation_task import GenerationTaskResponse
-from app.services import EditPlanService, EditTemplateService
+from app.services import EditPlanService
 from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
@@ -64,13 +64,6 @@ class EditPlanUpdateRequest(BaseModel):
    )
 class CopyPlanRequest(BaseModel):
    """复制剪辑计划请求体"""
    name: Optional[str] = Field(default=None, min_length=1, max_length=200, description="新计划名称，不传则为「原名 - 副本」")
    project_id: Optional[str] = Field(default=None, description="目标项目 ID，不传则复用源计划的项目")
 class EditPlanResponse(BaseModel):
    """剪辑计划响应体"""
@@ -79,7 +72,6 @@ class EditPlanResponse(BaseModel):
    name: str
    status: str
    total_duration: float
    result_count: int = 0
    project_id: str = ""
    created_by_user_id: str = ""
    config: dict[str, Any]
@@ -116,10 +108,6 @@ class EditPlanGenerationStatusResponse(BaseModel):
    plan_id: str
    plan_status: str
    generation_task_id: Optional[str] = None
    generation_task_status: Optional[str] = None
    progress: float = 0.0
    video_url: str = ""
    error_message: str = ""
    clips: List[ClipStatusItem]
@@ -249,7 +237,6 @@ def _to_response(p: EditPlan) -> EditPlanResponse:
        name=p.name,
        status=p.status.value if hasattr(p.status, "value") else p.status,
        total_duration=p.total_duration,
        result_count=getattr(p, "result_count", 0),
        project_id=p.project_id or "",
        created_by_user_id=p.created_by_user_id or "",
        config=p.config,
@@ -288,11 +275,11 @@ def list_plans(
    if status_filter:
        try:
            status_enum = EditPlanStatus(status_filter)
-        except ValueError as _e:
+        except ValueError:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="无效的筛选条件，请选择正确的状态",
-            ) from _e
+            )
    # 项目鉴权：如果指定了 project_id，校验用户是否有权访问
    if project_id:
@@ -335,7 +322,7 @@ def get_plan(
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(exc),
-        ) from exc
+        )
    # 项目鉴权
    if plan.project_id:
        check_project_access(plan.project_id, current_user.user.id, project_repository)
@@ -371,7 +358,7 @@ def create_plan(
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(exc),
-        ) from exc
+        )
    logger.info(
        "创建剪辑计划: id=%s name=%s by user=%s",
        created.id,
@@ -414,11 +401,11 @@ def update_plan(
        if body.status is not None:
            try:
                target_status = EditPlanStatus(body.status)
-            except ValueError as _e:
+            except ValueError:
                raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail="无效的状态值，请选择正确的状态",
-                ) from _e
+                )
            svc.transition_status(plan_id, target_status)
    except ValueError as exc:
        err_msg = str(exc)
@@ -426,11 +413,11 @@ def update_plan(
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=err_msg,
-            ) from exc
+            )
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=err_msg,
-        ) from exc
+        )
    # 返回最新状态
    result = svc.get_plan_or_raise(plan_id)
@@ -464,138 +451,12 @@ def delete_plan(
    )
@router.post("/{plan_id}/copy", response_model=EditPlanResponse, status_code=status.HTTP_201_CREATED)
 def copy_plan(
    plan_id: str,
    body: CopyPlanRequest,
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
 ) -> EditPlanResponse:
    """复制剪辑计划（含所有片段配置）
    新计划状态为 editing，不含生成任务和结果记录。
    """
    svc = EditPlanService(db)
    # 源计划鉴权
    existing = svc.get_plan(plan_id)
    if existing is None:
        raise HTTPException(status_code=404, detail=f"剪辑计划不存在: {plan_id}")
    if existing.project_id:
        check_project_access(existing.project_id, current_user.user.id, project_repository)
    # 目标项目鉴权（如果指定了不同的项目）
    target_project_id = body.project_id if body.project_id is not None else existing.project_id
    if target_project_id and target_project_id != existing.project_id:
        check_project_access(target_project_id, current_user.user.id, project_repository)
    try:
        new_plan = svc.copy_plan(
            plan_id,
            new_name=body.name,
            project_id=target_project_id,
        )
    except ValueError as e:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
    logger.info(
        "复制剪辑计划: source=%s target=%s by user=%s",
        plan_id,
        new_plan.id,
        current_user.user.id,
    )
    return _to_response(new_plan)
 # ── 保存为模板 ────────────────────────────────────────────────────────────────
 class SaveAsTemplateRequest(BaseModel):
    """保存为模板请求体"""
    name: str = Field(..., min_length=1, max_length=200, description="模板名称")
    description: str = Field(default="", max_length=500, description="模板描述")
    template_type: str = Field(default="custom", max_length=50, description="模板类型")
    preview_url: str = Field(default="", max_length=500, description="预览图 URL")
@router.post(
    "/{plan_id}/save-as-template",
    response_model=dict[str, Any],
    summary="将剪辑计划保存为模板",
    status_code=status.HTTP_201_CREATED,
 )
 def save_plan_as_template(
    plan_id: str,
    body: SaveAsTemplateRequest,
    current_user: AuthenticatedUser = Depends(get_current_user),
    db: Session = Depends(get_db_session),
    project_repo=Depends(get_project_repository),
 ) -> dict[str, Any]:
    """将指定剪辑计划的配置和片段结构保存为一个新模板。
    新模板会复制计划的所有片段配置（类型、时长、转场、文案等），
    但不绑定具体素材，可重复用于创建新的剪辑计划。
    """
    # 校验计划存在性和项目权限
    plan_service = EditPlanService(db)
    plan = plan_service.get_plan(plan_id)
    if plan is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"剪辑计划不存在: {plan_id}",
        )
    if plan.project_id:
        check_project_access(project_repo, current_user, plan.project_id)
    template_service = EditTemplateService(db)
    try:
        result = template_service.save_plan_as_template(
            plan_id=plan_id,
            name=body.name,
            description=body.description,
            template_type=body.template_type,
            preview_url=body.preview_url,
        )
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        ) from e
    template = result["template"]
    clip_configs = result["clip_configs"]
    logger.info(
        "保存计划为模板: plan_id=%s template_id=%s name=%s by user=%s",
        plan_id,
        template.id,
        body.name,
        current_user.user.id,
    )
    return {
        "id": template.id,
        "name": template.name,
        "description": template.description,
        "template_type": template.template_type,
        "editing_mode": template.editing_mode,
        "preview_url": template.preview_url,
        "status": template.status.value,
        "clip_count": len(clip_configs),
        "created_at": template.created_at.isoformat(),
    }
 # ── Include sub-routers (拆分模块) ────────────────────────────────────────────
 from .edit_plans_ai import router as ai_router
 from .edit_plans_clips import router as clips_router
 from .edit_plans_clips_batch import router as clips_batch_router
 from .edit_plans_generation import router as generation_router
 from .edit_plans_timeline import router as timeline_router
 router.include_router(generation_router)
 router.include_router(ai_router)
 router.include_router(timeline_router)
 router.include_router(clips_router, prefix="/{plan_id}/clips", tags=["EditPlan Clips"])
 router.include_router(clips_batch_router, prefix="/{plan_id}/clips", tags=["EditPlan Clips"])
@@ -58,7 +58,7 @@ def ai_recommend_clips(
    try:
        plan = svc.get_plan_or_raise(plan_id)
    except ValueError as exc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
    if plan.project_id:
        check_project_access(plan.project_id, current_user.user.id, project_repository)
@@ -103,7 +103,7 @@ def ai_recommend_clips(
            config=normalized_config,
            total_duration=result["total_duration"],
        )
-    except Exception as _e:
+    except Exception:
        logger.exception("AI 推荐写入失败，plan_id=%s 数据可能不一致", plan_id)
        try:
            db.rollback()
@@ -116,7 +116,7 @@ def ai_recommend_clips(
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="AI推荐结果保存失败，请稍后重试",
-        ) from _e
+        )
    logger.info(
        "AI 推荐片段方案: plan_id=%s clips=%d duration=%.1f by user=%s",
@@ -167,7 +167,7 @@ def generate_cover(
    try:
        plan = svc.get_plan_or_raise(plan_id)
    except ValueError as exc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
    if plan.project_id:
        check_project_access(plan.project_id, current_user.user.id, project_repository)
@@ -1,278 +0,0 @@
 """剪辑计划片段（Clip）CRUD 路由。"""
 from __future__ import annotations
 import logging
 from typing import Any, List, Optional
 from app.auth import AuthenticatedUser, get_current_user
 from app.dependencies import get_db_session, get_project_repository
 from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
 from sqlalchemy.orm import Session
 from packages.domain.edit_plan_clip import EditPlanClipStatus
 logger = logging.getLogger(__name__)
 router = APIRouter()
 # ── Schemas ──────────────────────────────────────────────────────────────────
 from pydantic import BaseModel, Field
 class EditPlanClipResponse(BaseModel):
    """剪辑片段响应体"""
    id: str
    plan_id: str
    clip_type: str
    order: int
    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
    status: str
    config: dict[str, Any] = Field(default_factory=dict)
    created_at: Optional[str] = None
    updated_at: Optional[str] = None
 class EditPlanClipListResponse(BaseModel):
    """剪辑片段列表响应体"""
    items: List[EditPlanClipResponse]
    total: int
 class EditPlanClipCreateRequest(BaseModel):
    """创建剪辑片段请求体"""
    clip_type: str = Field(
        ..., min_length=1, max_length=50, description="片段类型: main/intro/outro/overlay/background/b_roll 等"
    )
    order: int = Field(..., ge=0, description="排序序号")
    asset_id: str = Field(default="", max_length=64, description="关联素材 ID")
    text_content: str = Field(default="", max_length=5000, description="文本内容（字幕/配音等）")
    start_time: float = Field(default=0.0, ge=0.0, description="起始时间 (秒)")
    duration: float = Field(default=0.0, ge=0.0, description="时长 (秒)")
    transition_effect: str = Field(default="cut", max_length=50, description="转场效果")
    transition_duration: float = Field(default=0.0, ge=0.0, description="转场时长 (秒)")
    playback_speed: float = Field(default=1.0, gt=0.0, le=10.0, description="播放速度倍率")
    config: dict[str, Any] = Field(default_factory=dict, description="扩展配置 (JSON)")
 class EditPlanClipUpdateRequest(BaseModel):
    """更新剪辑片段请求体"""
    clip_type: Optional[str] = Field(default=None, min_length=1, max_length=50, description="片段类型")
    order: Optional[int] = Field(default=None, ge=0, description="排序序号")
    asset_id: Optional[str] = Field(default=None, max_length=64, description="关联素材 ID")
    text_content: Optional[str] = Field(default=None, max_length=5000, description="文本内容")
    start_time: Optional[float] = Field(default=None, ge=0.0, description="起始时间 (秒)")
    duration: Optional[float] = Field(default=None, ge=0.0, description="时长 (秒)")
    transition_effect: Optional[str] = Field(default=None, max_length=50, description="转场效果")
    transition_duration: Optional[float] = Field(default=None, ge=0.0, description="转场时长 (秒)")
    playback_speed: Optional[float] = Field(default=None, gt=0.0, le=10.0, description="播放速度倍率")
    config: Optional[dict[str, Any]] = Field(default=None, description="扩展配置 (JSON)")
 # ── Helpers ──────────────────────────────────────────────────────────────────
 def _check_plan_access(plan_id: str, user_id: str, project_repository: Any, db: Session) -> Any:
    """验证用户是否有权限访问该剪辑计划（通过项目关联）。
    返回 plan 对象供后续使用，避免重复查询。
    """
    from app.services.edit_plan_service import EditPlanService
    from ._helpers import check_project_access
    svc = EditPlanService(db)
    plan = svc.get_plan(plan_id)
    if plan is None:
        raise HTTPException(status_code=404, detail=f"剪辑计划不存在: {plan_id}")
    if plan.project_id:
        check_project_access(plan.project_id, user_id, project_repository)
    return plan
 def _clip_to_response(clip) -> EditPlanClipResponse:
    """将领域对象转换为响应体"""
    return EditPlanClipResponse(
        id=clip.id,
        plan_id=clip.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,
        status=clip.status.value if hasattr(clip.status, "value") else str(clip.status),
        config=clip.config or {},
        created_at=clip.created_at.isoformat() if clip.created_at else None,
        updated_at=clip.updated_at.isoformat() if clip.updated_at else None,
    )
 def _get_svc(db: Session):
    """获取 EditPlanService 实例"""
    from app.services.edit_plan_service import EditPlanService
    return EditPlanService(db)
 # ── Routes ───────────────────────────────────────────────────────────────────
@router.get("", response_model=EditPlanClipListResponse)
 def list_clips(
    plan_id: str,
    status_filter: Optional[str] = Query(None, alias="status", description="按状态过滤"),
    skip: int = Query(0, ge=0, description="分页偏移"),
    limit: int = Query(100, ge=1, le=500, description="每页数量"),
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
 ) -> EditPlanClipListResponse:
    """获取剪辑计划的片段列表"""
    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
    svc = _get_svc(db)
    status_enum = EditPlanClipStatus(status_filter) if status_filter else None
    clips = svc.list_clips(plan_id, status=status_enum, skip=skip, limit=limit)
    total = svc.count_clips(plan_id, status=status_enum)
    return EditPlanClipListResponse(
        items=[_clip_to_response(c) for c in clips],
        total=total,
    )
@router.post("", response_model=EditPlanClipResponse, status_code=status.HTTP_201_CREATED)
 def create_clip(
    plan_id: str,
    body: EditPlanClipCreateRequest,
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
 ) -> EditPlanClipResponse:
    """创建剪辑片段"""
    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
    svc = _get_svc(db)
    try:
        clip = svc.create_clip(
            plan_id=plan_id,
            clip_type=body.clip_type,
            order=body.order,
            asset_id=body.asset_id,
            text_content=body.text_content,
            start_time=body.start_time,
            duration=body.duration,
            transition_effect=body.transition_effect,
            transition_duration=body.transition_duration,
            playback_speed=body.playback_speed,
            config=body.config,
        )
    except ValueError as e:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
    logger.info("创建剪辑片段: plan_id=%s clip_id=%s by user=%s", plan_id, clip.id, current_user.user.id)
    return _clip_to_response(clip)
@router.get("/{clip_id}", response_model=EditPlanClipResponse)
 def get_clip(
    plan_id: str,
    clip_id: str,
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
 ) -> EditPlanClipResponse:
    """获取剪辑片段详情"""
    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
    svc = _get_svc(db)
    clip = svc.get_clip(clip_id)
    if clip is None:
        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
    if clip.plan_id != plan_id:
        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
    return _clip_to_response(clip)
@router.put("/{clip_id}", response_model=EditPlanClipResponse)
 def update_clip(
    plan_id: str,
    clip_id: str,
    body: EditPlanClipUpdateRequest,
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
 ) -> EditPlanClipResponse:
    """更新剪辑片段"""
    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
    svc = _get_svc(db)
    # 验证 clip 属于该 plan
    clip = svc.get_clip(clip_id)
    if clip is None:
        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
    if clip.plan_id != plan_id:
        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
    try:
        updated = svc.update_clip(
            clip_id,
            clip_type=body.clip_type,
            order=body.order,
            asset_id=body.asset_id,
            text_content=body.text_content,
            start_time=body.start_time,
            duration=body.duration,
            transition_effect=body.transition_effect,
            transition_duration=body.transition_duration,
            playback_speed=body.playback_speed,
            config=body.config,
        )
    except ValueError as e:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
    logger.info("更新剪辑片段: plan_id=%s clip_id=%s by user=%s", plan_id, clip_id, current_user.user.id)
    return _clip_to_response(updated)
@router.delete("/{clip_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None, response_class=Response)
 def delete_clip(
    plan_id: str,
    clip_id: str,
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
 ) -> None:
    """删除剪辑片段"""
    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
    svc = _get_svc(db)
    # 验证 clip 属于该 plan
    clip = svc.get_clip(clip_id)
    if clip is None:
        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
    if clip.plan_id != plan_id:
        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
    deleted = svc.delete_clip(clip_id)
    if not deleted:
        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
    logger.info("删除剪辑片段: plan_id=%s clip_id=%s by user=%s", plan_id, clip_id, current_user.user.id)
    return None
@@ -1,241 +0,0 @@
 """剪辑计划片段批量操作 API。"""
 from __future__ import annotations
 import logging
 from typing import Any, List, Optional
 from app.auth import AuthenticatedUser, get_current_user
 from app.dependencies import get_db_session, get_project_repository
 from fastapi import APIRouter, Depends, HTTPException, status
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 logger = logging.getLogger(__name__)
 router = APIRouter()
 # ── Schemas ──────────────────────────────────────────────────────────────────
 class ClipReorderItem(BaseModel):
    """重排序条目"""
    clip_id: str
    new_order: int = Field(..., ge=0, description="新的排序序号")
 class ClipReorderRequest(BaseModel):
    """片段重排序请求"""
    items: List[ClipReorderItem] = Field(..., min_length=1, max_length=500, description="重排序条目列表")
 class ClipReorderResponse(BaseModel):
    """片段重排序响应"""
    success: bool
    updated_count: int
    message: str = ""
 class ClipBatchDeleteRequest(BaseModel):
    """批量删除片段请求"""
    clip_ids: List[str] = Field(..., min_length=1, max_length=500, description="要删除的片段ID列表")
 class ClipBatchDeleteResponse(BaseModel):
    """批量删除片段响应"""
    success: bool
    deleted_count: int
    message: str = ""
 class ClipsFromAssetsRequest(BaseModel):
    """从素材批量创建片段请求"""
    asset_ids: List[str] = Field(..., min_length=1, max_length=200, description="素材 ID 列表，按顺序追加到时间线末尾")
    clip_type: str = Field(default="main", description="片段类型，默认 main")
 class ClipsFromAssetsResponse(BaseModel):
    """从素材批量创建片段响应"""
    success: bool
    created_count: int
    message: str = ""
    clip_ids: List[str] = Field(default_factory=list, description="创建的片段ID列表")
 # ── Helpers ──────────────────────────────────────────────────────────────────
 def _check_plan_access(plan_id: str, user_id: str, project_repository: Any, db: Session) -> Any:
    """验证用户是否有权限访问该剪辑计划，返回 plan 对象。"""
    from app.services.edit_plan_service import EditPlanService
    from ._helpers import check_project_access
    svc = EditPlanService(db)
    plan = svc.get_plan(plan_id)
    if plan is None:
        raise HTTPException(status_code=404, detail=f"剪辑计划不存在: {plan_id}")
    if plan.project_id:
        check_project_access(plan.project_id, user_id, project_repository)
    return plan
 def _get_svc(db: Session):
    """获取 EditPlanService 实例"""
    from app.services.edit_plan_service import EditPlanService
    return EditPlanService(db)
 # ── Routes ───────────────────────────────────────────────────────────────────
@router.post("/reorder", response_model=ClipReorderResponse)
 def reorder_clips(
    plan_id: str,
    body: ClipReorderRequest,
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
 ) -> ClipReorderResponse:
    """批量重排序片段
    前端拖拽调整顺序后，一次性提交所有变更的 order。
    自动触发编辑状态回退（从 completed/failed 切回 editing）。
    """
    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
    svc = _get_svc(db)
    # 验证所有 clip 都属于该 plan
    clip_ids = [item.clip_id for item in body.items]
    existing_clips = svc.list_clips(plan_id, skip=0, limit=10000)
    existing_ids = {c.id for c in existing_clips}
    invalid_ids = [cid for cid in clip_ids if cid not in existing_ids]
    if invalid_ids:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"以下片段不属于该计划: {', '.join(invalid_ids[:5])}",
        )
    # 执行重排序
    updated_count = 0
    for item in body.items:
        try:
            svc.update_clip(item.clip_id, order=item.new_order)
            updated_count += 1
        except ValueError as e:
            logger.warning("重排序片段失败: clip_id=%s error=%s", item.clip_id, e)
    logger.info(
        "批量重排序片段: plan_id=%s count=%d by user=%s",
        plan_id,
        updated_count,
        current_user.user.id,
    )
    return ClipReorderResponse(
        success=True,
        updated_count=updated_count,
        message=f"成功更新 {updated_count} 个片段的顺序",
    )
@router.post("/batch-delete", response_model=ClipBatchDeleteResponse)
 def batch_delete_clips(
    plan_id: str,
    body: ClipBatchDeleteRequest,
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
 ) -> ClipBatchDeleteResponse:
    """批量删除片段
    自动触发编辑状态回退（从 completed/failed 切回 editing）。
    """
    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
    svc = _get_svc(db)
    # 验证所有 clip 都属于该 plan
    existing_clips = svc.list_clips(plan_id, skip=0, limit=10000)
    existing_ids = {c.id for c in existing_clips}
    valid_ids = [cid for cid in body.clip_ids if cid in existing_ids]
    skipped = len(body.clip_ids) - len(valid_ids)
    # 执行删除
    deleted_count = 0
    for clip_id in valid_ids:
        if svc.delete_clip(clip_id):
            deleted_count += 1
    message = f"成功删除 {deleted_count} 个片段"
    if skipped > 0:
        message += f"，跳过 {skipped} 个不存在的片段"
    logger.info(
        "批量删除片段: plan_id=%s deleted=%d skipped=%d by user=%s",
        plan_id,
        deleted_count,
        skipped,
        current_user.user.id,
    )
    return ClipBatchDeleteResponse(
        success=True,
        deleted_count=deleted_count,
        message=message,
    )
@router.post("/from-assets", response_model=ClipsFromAssetsResponse)
 def create_clips_from_assets(
    plan_id: str,
    body: ClipsFromAssetsRequest,
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
 ) -> ClipsFromAssetsResponse:
    """从素材批量创建片段（追加到时间线末尾）
    一次性将多个素材作为片段添加到剪辑计划，自动读取素材时长。
    自动触发编辑状态回退（completed/failed → editing）。
    """
    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
    svc = _get_svc(db)
    try:
        clips = svc.create_clips_from_assets(
            plan_id=plan_id,
            asset_ids=body.asset_ids,
            clip_type=body.clip_type,
        )
    except ValueError as e:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
    clip_ids = [c.id for c in clips]
    logger.info(
        "从素材批量创建片段: plan_id=%s count=%d by user=%s",
        plan_id,
        len(clips),
        current_user.user.id,
    )
    return ClipsFromAssetsResponse(
        success=True,
        created_count=len(clips),
        message=f"成功创建 {len(clips)} 个片段",
        clip_ids=clip_ids,
    )
@@ -183,7 +183,9 @@ def _auto_fallback_auto_material_mode(
 def _check_queue_limits(gen_task_repo, user_id: str) -> None:
    """队列限流预检查"""
    try:
-        has_count = hasattr(gen_task_repo, "count_pending_by_user") and hasattr(gen_task_repo, "count_pending_total")
+        has_count = hasattr(gen_task_repo, "count_pending_by_user") and hasattr(
            gen_task_repo, "count_pending_total"
        )
        if has_count:
            user_pending = gen_task_repo.count_pending_by_user(user_id)
            global_pending = gen_task_repo.count_pending_total()
@@ -239,7 +241,7 @@ def generate_plan(
    try:
        can_gen, reason = svc.can_generate(plan_id)
    except ValueError as exc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
    if not can_gen:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=reason)
@@ -253,15 +255,12 @@ def generate_plan(
        gen_task_use_case = CreateGenerationTaskUseCase(gen_task_repo)
        plan = svc.get_plan_or_raise(plan_id)
        # 从 plan.config 中读取 asset_ids 并传递给 GenerationTask
        config_asset_ids = (plan.config or {}).get("asset_ids", [])
        gen_task = gen_task_use_case.execute(
            CreateGenerationTaskCommand(
-                project_id=plan.project_id or "",
+                project_id="",
                template_id=plan.template_id,
                created_by_user_id=current_user.user.id,
                source_edit_plan_id=plan_id,
                asset_ids=list(config_asset_ids) if config_asset_ids else [],
            )
        )
@@ -287,7 +286,7 @@ def generate_plan(
        )
    except HTTPException:
        raise
-    except Exception as _e:
+    except Exception:
        logger.exception("触发剪辑计划生成失败: plan_id=%s", plan_id)
        try:
            svc.transition_status(plan_id, EditPlanStatus.FAILED)
@@ -296,7 +295,7 @@ def generate_plan(
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="生成失败，请稍后重试",
-        ) from _e
+        )
@router.get(
@@ -314,7 +313,7 @@ def get_generation_status(
    try:
        gen_status = svc.get_generation_status(plan_id)
    except ValueError as exc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
    plan = gen_status["plan"]
    if plan.project_id:
@@ -334,25 +333,10 @@ def get_generation_status(
        for c in clips
    ]
    # 从 plan.config 中取渲染结果 URL
    video_url = (plan.config or {}).get("rendered_url", "")
    # 从 gen_status 中取进度、错误信息、任务状态
    progress = gen_status.get("progress", 0.0)
    error_message = gen_status.get("error_message", "")
    gen_task_status = gen_status.get("generation_task_status")
    # 如果计划已完成但进度还是0，补100
    plan_status_val = plan.status.value if hasattr(plan.status, "value") else plan.status
    if plan_status_val == "completed" and progress < 100:
        progress = 100.0
    return EditPlanGenerationStatusResponse(
        plan_id=plan_id,
-        plan_status=plan_status_val,
+        plan_status=plan.status.value if hasattr(plan.status, "value") else plan.status,
        generation_task_id=gen_status["generation_task_id"],
        generation_task_status=gen_task_status,
        progress=progress,
        video_url=video_url,
        error_message=error_message,
        clips=clip_items,
    )
@@ -173,7 +173,7 @@ def generate_from_template(
    try:
        template = template_svc.get_template_or_raise(body.template_id)
    except ValueError as exc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
    clip_configs = template_svc.list_clip_configs(body.template_id, skip=0, limit=200)
@@ -105,7 +105,7 @@ async def list_feature_flags(
        return sorted(result, key=lambda x: x.name)
    except Exception as exc:
        logger.error("Failed to list feature flags: %s", exc)
-        raise HTTPException(status_code=500, detail=f"Failed to list flags: {exc}") from exc
+        raise HTTPException(status_code=500, detail=f"Failed to list flags: {exc}")
@router.get("/{name}", response_model=FeatureFlagResponse)
@@ -120,7 +120,7 @@ async def get_feature_flag(
        return FeatureFlagResponse.from_config(config)
    except Exception as exc:
        logger.error("Failed to get feature flag %s: %s", name, exc)
-        raise HTTPException(status_code=500, detail=f"Failed to get flag: {exc}") from exc
+        raise HTTPException(status_code=500, detail=f"Failed to get flag: {exc}")
@router.get("/{name}/check", response_model=FeatureFlagCheckResponse)
@@ -136,7 +136,7 @@ async def check_feature_flag(
        return FeatureFlagCheckResponse(name=name, active=active, identifier=identifier)
    except Exception as exc:
        logger.error("Failed to check feature flag %s: %s", name, exc)
-        raise HTTPException(status_code=500, detail=f"Failed to check flag: {exc}") from exc
+        raise HTTPException(status_code=500, detail=f"Failed to check flag: {exc}")
@router.put("/{name}", response_model=FeatureFlagResponse)
@@ -170,7 +170,7 @@ async def update_feature_flag(
        return FeatureFlagResponse.from_config(config)
    except Exception as exc:
        logger.error("Failed to update feature flag %s: %s", name, exc)
-        raise HTTPException(status_code=500, detail=f"Failed to update flag: {exc}") from exc
+        raise HTTPException(status_code=500, detail=f"Failed to update flag: {exc}")
@router.delete("/{name}", status_code=status.HTTP_204_NO_CONTENT, response_model=None, response_class=Response)
@@ -178,7 +178,7 @@ async def delete_feature_flag(
    name: str,
    _: bool = Depends(_verify_internal_api_key),
    store: RedisFeatureFlagStore = Depends(_get_feature_flag_store),
-):
+) :
    """删除 Feature Flag。
    只允许删除 ALLOWED_FLAGS 列表中的 flag。
@@ -191,4 +191,4 @@ async def delete_feature_flag(
        pass
    except Exception as exc:
        logger.error("Failed to delete feature flag %s: %s", name, exc)
-        raise HTTPException(status_code=500, detail=f"Failed to delete flag: {exc}") from exc
+        raise HTTPException(status_code=500, detail=f"Failed to delete flag: {exc}")
@@ -43,7 +43,6 @@ logger = logging.getLogger(__name__)
 router = APIRouter()
 def _to_generation_task_response(task) -> GenerationTaskResponse:
    return GenerationTaskResponse(
        id=task.id,
@@ -283,28 +282,28 @@ def create_generation_task(
                    created_tasks.append(task)
                else:
                    failed_tasks.append(task)
-            except UserPendingLimitExceeded as _e:
+            except UserPendingLimitExceeded:
                # 兜底：如果预检查后又并发提交了，在这里也拦住
                failed_tasks.append(task)
                if not created_tasks:
                    raise HTTPException(
                        status_code=429,
                        detail="您的待处理任务过多，请等待完成后再提交",
-                    ) from _e
+                    )
                break
-            except GlobalQueueFull as _e:
+            except GlobalQueueFull:
                failed_tasks.append(task)
                if not created_tasks:
                    raise HTTPException(
                        status_code=503,
                        detail="系统繁忙，请稍后再试",
-                    ) from _e
+                    )
                break
    except HTTPException:
        raise
    except Exception as e:
        logger.error("[生成任务] 创建失败: %s", e, exc_info=True)
-        raise HTTPException(status_code=500, detail="创建生成任务失败，请稍后重试或查看任务日志") from e
+        raise HTTPException(status_code=500, detail="创建生成任务失败，请稍后重试或查看任务日志")
    items = [_to_generation_task_response(t) for t in created_tasks + failed_tasks]
    return BatchGenerationTaskResponse(items=items, total=len(items))
@@ -81,11 +81,11 @@ def delete_project(
    use_case = DeleteProjectUseCase(project_repository)
    try:
        deleted = use_case.execute(project_id, authenticated_user.user.id)
-    except PermissionError as _e:
+    except PermissionError:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Only the project owner can delete this project",
-        ) from _e
+        )
    if not deleted:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
-    return  # type: ignore[return-value]
+    return
@@ -2,7 +2,6 @@
 from __future__ import annotations
 import logging
 from dataclasses import replace
 from datetime import datetime, timezone
 from typing import List
@@ -21,8 +20,6 @@ from fastapi import APIRouter, Depends, HTTPException, status
 from packages.ports.user_repository import UserRepository
 logger = logging.getLogger(__name__)
 router = APIRouter()
@@ -257,9 +254,7 @@ async def payment_callback(
        return {"success": True, "message": "支付成功", "record_id": record_id}
    except Exception as e:
        session.rollback()
-        logger.error(f"支付回调处理失败: user_id={user_id}, plan={plan}, error={e}")
+        raise HTTPException(status_code=500, detail=f"支付处理失败: {str(e)}")
        # 不返回原始异常信息，避免泄漏内部实现细节
        raise HTTPException(status_code=500, detail="支付处理失败，请稍后重试") from e
    finally:
        session.close()
@@ -368,9 +368,9 @@ def retry_project_task(
            raise HTTPException(status_code=404, detail="Ingest job not found")
        if _status_value(job.status) != "failed":
            raise HTTPException(status_code=409, detail="Only failed tasks can be retried")
-        use_case = SubmitIngestJobUseCase(ingest_job_repository)  # type: ignore[assignment]
+        use_case = SubmitIngestJobUseCase(ingest_job_repository)
        retried = use_case.execute(
-            SubmitIngestJobCommand(  # type: ignore[arg-type]
+            SubmitIngestJobCommand(
                project_id=job.project_id,
                library_id=job.library_id,
                storage_key=job.storage_key,
@@ -386,6 +386,6 @@ def retry_project_task(
            current_step=_ingest_step(retried),
            source_id=retried.id,
            created_at=retried.created_at,
-            updated_at=retried.updated_at,  # type: ignore[attr-defined]
+            updated_at=retried.updated_at,
        )
    raise HTTPException(status_code=400, detail="Unsupported task type")
@@ -147,9 +147,9 @@ def get_template(
        use_case = GetTemplateUseCase(template_repository)
        template = use_case.execute(template_id, user_id)
        usage = template_repository.get_usage_count(template_id)
-    except Exception as _e:
+    except Exception:
        logger.exception("get_template 查询失败: template_id=%s", template_id)
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="模板查询失败") from _e
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="模板查询失败")
    if template is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found")
    return _to_response(template, usage_count=usage)
@@ -186,7 +186,7 @@ def create_template(
    try:
        template = use_case.execute(command)
    except ValidationError as exc:
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
    return _to_response(template)
@@ -226,10 +226,10 @@ def update_template(
    use_case = UpdateTemplateUseCase(template_repository)
    try:
        template = use_case.execute(command)
-    except NotFoundError as _e:
+    except NotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found") from _e
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found")
    except ValidationError as exc:
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
    return _to_response(template)
@@ -264,10 +264,10 @@ def copy_template(
    use_case = CopyTemplateUseCase(template_repository)
    try:
        template = use_case.execute(command)
-    except NotFoundError as _e:
+    except NotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found") from _e
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found")
    except ValidationError as exc:
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
    return _to_response(template)
@@ -299,9 +299,9 @@ def toggle_favorite(
    use_case = GetTemplateUseCase(template_repository)
    try:
        template = use_case.execute(template_id, user_id)
-    except Exception as _e:
+    except Exception:
        logger.exception("toggle_favorite 查询失败: template_id=%s", template_id)
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found") from _e
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found")
    if template is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found")
    return ToggleFavoriteResponse(id=template_id, is_favorite=False)
@@ -326,10 +326,10 @@ def validate_template(
    use_case = ValidateTemplateUseCase(template_repository)
    try:
        result = use_case.execute(command)
-    except NotFoundError as _e:
+    except NotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found") from _e
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found")
    except ValidationError as exc:
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
    return ValidateTemplateResponse(
        template=_to_response(result.template),
@@ -375,9 +375,7 @@ def create_category(
    )
-@router.delete(
+@router.delete("/categories/{category_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None, response_class=Response)
    "/categories/{category_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None, response_class=Response
 )
 def delete_category(
    category_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
@@ -148,7 +148,7 @@ def create_title(
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail=f"标题库配额已满（{exc.used}/{exc.limit}），请升级套餐",
-        ) from exc
+        )
    return _to_response(item)
@@ -172,8 +172,8 @@ def update_title(
    use_case = UpdateTitleLibraryUseCase(title_repository)
    try:
        item = use_case.execute(command)
-    except NotFoundError as _e:
+    except NotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Title not found") from _e
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Title not found")
    return _to_response(item)
@@ -236,8 +236,8 @@ def get_tts_job(
    use_case = GetTTSJobUseCase(repository)
    try:
        job = use_case.execute(job_id, user_id)
-    except TTSJobNotFoundError as _e:
+    except TTSJobNotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found") from _e
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found")
    return _to_response(job, sign_url)
@@ -253,8 +253,8 @@ def get_tts_job_status(
    use_case = GetTTSJobStatusUseCase(repository)
    try:
        job = use_case.execute(job_id, user_id)
-    except TTSJobNotFoundError as _e:
+    except TTSJobNotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found") from _e
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found")
    output_url = job.output_audio_url
    if output_url:
        output_url = sign_url(output_url)
@@ -309,8 +309,8 @@ def save_tts_job_to_library(
    get_use_case = GetTTSJobUseCase(tts_repository)
    try:
        job = get_use_case.execute(job_id, user_id)
-    except TTSJobNotFoundError as _e:
+    except TTSJobNotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found") from _e
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found")
    # 校验已完成
    if not job.is_completed:
@@ -363,7 +363,7 @@ def save_tts_job_to_library(
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail=f"配音库配额已满（{exc.used}/{exc.limit}），请升级套餐",
-        ) from exc
+        )
    return SaveToLibraryResponse(
        id=item.id,
@@ -141,8 +141,8 @@ def get_voice_clone(
    use_case = GetVoiceCloneUseCase(repository)
    try:
        profile = use_case.execute(clone_id, user_id)
-    except VoiceCloneNotFoundError as _e:
+    except VoiceCloneNotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found") from _e
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found")
    return _to_response(profile)
@@ -157,8 +157,8 @@ def get_voice_clone_status(
    use_case = GetVoiceCloneStatusUseCase(repository)
    try:
        profile = use_case.execute(clone_id, user_id)
-    except VoiceCloneNotFoundError as _e:
+    except VoiceCloneNotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found") from _e
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found")
    return VoiceCloneStatusResponse(
        id=profile.id,
        status=profile.status,
@@ -201,13 +201,13 @@ def retry_voice_clone(
    user_id = authenticated_user.user.id
    try:
        profile = workflow.retry_clone(clone_id, user_id)
-    except VoiceCloneNotFoundError as _e:
+    except VoiceCloneNotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found") from _e
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found")
-    except VoiceCloneNotRetryableError as _e:
+    except VoiceCloneNotRetryableError:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Voice clone is not retryable (only failed clones can be retried)",
-        ) from _e
+        )
    # 如果 profile 处于 processing 且有 task_id，触发 Celery 异步轮询
    task_id = (profile.metadata or {}).get("cosyvoice_task_id", "")
@@ -287,7 +287,7 @@ def create_voice(
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail=f"配音库配额已满（{exc.used}/{exc.limit}），请升级套餐",
-        ) from exc
+        )
    return _to_response(item, sign_url)
@@ -317,8 +317,8 @@ def update_voice(
    use_case = UpdateVoiceLibraryUseCase(voice_repository)
    try:
        item = use_case.execute(command)
-    except NotFoundError as _e:
+    except NotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice not found") from _e
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice not found")
    return _to_response(item, sign_url)
@@ -141,7 +141,7 @@ def safe_enqueue_generation_task(
            global_pending_limit,
            user_id or "unknown",
        )
-        exc: Exception = GlobalQueueFull(pending_count=global_pending, limit=global_pending_limit)
+        exc = GlobalQueueFull(pending_count=global_pending, limit=global_pending_limit)
        _mark_task_failed_safely(task, generation_task_repository, log_prefix, str(exc))
        raise exc
@@ -194,7 +194,7 @@ def safe_enqueue_generation_task(
    if global_over or user_over:
        if global_over:
            reason = f"全局 pending 超限(入队后): {global_after}/{global_pending_limit}"
-            exc = GlobalQueueFull(pending_count=global_after, limit=global_pending_limit)
+            exc: Exception = GlobalQueueFull(pending_count=global_after, limit=global_pending_limit)
        else:
            reason = f"用户 pending 超限(入队后): {user_after}/{user_pending_limit}"
            exc = UserPendingLimitExceeded(user_id=user_id, pending_count=user_after, limit=user_pending_limit)
@@ -132,7 +132,7 @@ def get_tag_repository(
    session: Session = Depends(get_db_session),
 ) -> TagRepository:
    """Provide the SQLAlchemy tag repository implementation."""
-    return SQLAlchemyTagRepository(session)  # type: ignore[return-value]
+    return SQLAlchemyTagRepository(session)
 def get_user_repository(
@@ -105,7 +105,7 @@ class RateLimitMiddleware(BaseHTTPMiddleware):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.paths = set(paths) if paths else None
-        self.requests: dict[str, list[float]] = {}
+        self.requests = {}  # {ip: [timestamps]}
    async def dispatch(self, request: Request, call_next):
        # 如果配置了路径过滤，只对指定路径限流
@@ -12,7 +12,6 @@
 from __future__ import annotations
 import logging
 from collections.abc import Mapping
 from dataclasses import dataclass
 from sqlalchemy.orm import Session
@@ -156,7 +155,7 @@ class AutoClipService:
        self,
        clip: EditPlanClip,
        project_id: str,
-        config_map: Mapping[str, object],
+        config_map: dict[str, object],
    ) -> ClipAssignDetail:
        """为单个片段分配素材。"""
        config = config_map.get(clip.template_clip_config_id) if clip.template_clip_config_id else None
@@ -141,19 +141,6 @@ class EditPlanService:
        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,
@@ -169,10 +156,6 @@ class EditPlanService:
        """
        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,
@@ -229,24 +212,8 @@ class EditPlanService:
            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.EDITING: plan.start_editing,
            EditPlanStatus.RENDERING: plan.start_rendering,
            EditPlanStatus.COMPLETED: plan.mark_completed,
            EditPlanStatus.FAILED: plan.mark_failed,
@@ -325,8 +292,6 @@ class EditPlanService:
        """
        # 确保计划存在
        self.get_plan_or_raise(plan_id)
        # 自动从 completed/failed 切回 editing
        self._auto_resume_editing(plan_id)
        clip = EditPlanClip.create(
            plan_id=plan_id,
@@ -374,9 +339,6 @@ class EditPlanService:
        """
        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:
@@ -419,8 +381,6 @@ class EditPlanService:
            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)
@@ -447,62 +407,6 @@ class EditPlanService:
        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]:
@@ -527,8 +431,6 @@ class EditPlanService:
                "clips": List[EditPlanClip],
                "generation_task_id": Optional[str],
                "generation_task_status": Optional[str],
                "progress": float,
                "error_message": str,
            }
        Raises:
@@ -540,23 +442,17 @@ class EditPlanService:
        # 从 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]:
@@ -607,9 +503,6 @@ class EditPlanService:
            更新后的计划
        """
        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(
@@ -626,85 +519,3 @@ class EditPlanService:
            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)
@@ -12,13 +12,9 @@ from typing import Any, List, Optional
 from sqlalchemy.orm import Session
 from packages.adapters.sqlalchemy_impl import (
    SQLAlchemyEditPlanClipRepository,
    SQLAlchemyEditPlanRepository,
    SQLAlchemyEditTemplateRepository,
    SQLAlchemyTemplateClipConfigRepository,
 )
 from packages.domain.edit_plan import EditPlan
 from packages.domain.edit_plan_clip import EditPlanClip
 from packages.domain.edit_template import EditTemplate, EditTemplateStatus
 from packages.domain.template_clip_config import (
    ClipType,
@@ -41,9 +37,6 @@ class EditTemplateService:
    def __init__(self, db: Session) -> None:
        self._template_repo = SQLAlchemyEditTemplateRepository(db)
        self._clip_config_repo = SQLAlchemyTemplateClipConfigRepository(db)
        self._plan_repo = SQLAlchemyEditPlanRepository(db)
        self._plan_clip_repo = SQLAlchemyEditPlanClipRepository(db)
        self._db = db
    # ── 模板 CRUD ──────────────────────────────────────────────────────────
@@ -401,137 +394,3 @@ class EditTemplateService:
            "template": template,
            "clip_configs": clip_configs,
        }
    # ── 从剪辑计划保存为模板 ──────────────────────────────────────────────
    def save_plan_as_template(
        self,
        plan_id: str,
        name: str,
        *,
        description: str = "",
        template_type: str = "custom",
        preview_url: str = "",
    ) -> dict[str, Any]:
        """将剪辑计划保存为模板
        将指定剪辑计划的配置和片段结构另存为一个新模板，
        方便后续基于该模板快速创建新的剪辑计划。
        转换规则:
        - 计划名称 → 模板名称（调用方传入，支持自定义）
        - 计划 config → 模板 config（整体迁移）
        - 计划 editing_mode 从 config 中提取，默认 one_take
        - 每个片段转换为模板片段配置:
          - clip_type 直接映射
          - order 保持不变
          - duration → min_duration = max_duration = duration（固定时长）
          - text_content → text_template
          - transition_effect 直接映射
          - playback_speed 等播放参数存入 config
          - 不保留 asset_id（模板不绑定具体素材）
        Args:
            plan_id: 源剪辑计划 ID
            name: 新模板名称
            description: 模板描述
            template_type: 模板类型，默认 custom（用户自定义）
            preview_url: 预览图 URL
        Returns:
            dict: {"template": EditTemplate, "clip_configs": List[TemplateClipConfig]}
        Raises:
            ValueError: 计划不存在或名称为空/重复
        """
        # 1. 读取源计划
        plan = self._plan_repo.get(plan_id)
        if plan is None:
            raise ValueError(f"剪辑计划不存在: {plan_id}")
        # 2. 读取所有片段（按 order 排序）
        clips = self._plan_clip_repo.list_by_plan(plan_id)
        clips.sort(key=lambda c: c.order)
        # 3. 提取 editing_mode
        editing_mode = plan.config.get("editing_mode", "one_take") if plan.config else "one_take"
        # 4. 创建模板（复用 create_template 的校验逻辑，但手动构建避免重复查询）
        clean_name = name.strip()
        if not clean_name:
            raise ValueError("模板名称不能为空")
        # 名称重复检查
        existing = self._template_repo.list_all(skip=0, limit=1000)
        for t in existing:
            if t.name == clean_name and t.status == EditTemplateStatus.ACTIVE:
                raise ValueError(f"模板名称已存在: {clean_name}")
        # 从计划 config 中提取模板级配置，去掉运行时/素材相关字段
        plan_config = plan.config or {}
        template_config: dict[str, Any] = {}
        for key, value in plan_config.items():
            # 跳过明显的运行时/实例字段，保留风格/模式类配置
            if key not in {"asset_ids", "source_edit_plan_id", "generation_task_id"}:
                template_config[key] = value
        template = EditTemplate.create(
            name=clean_name,
            description=description,
            template_type=template_type,
            editing_mode=editing_mode,
            config=template_config,
            preview_url=preview_url,
        )
        created_template = self._template_repo.create(template)
        logger.info(
            "从剪辑计划创建模板: plan_id=%s template_id=%s name=%s clip_count=%d",
            plan_id,
            created_template.id,
            clean_name,
            len(clips),
        )
        # 5. 转换每个片段为模板片段配置
        created_configs: List[TemplateClipConfig] = []
        for clip in clips:
            clip_config: dict[str, Any] = {}
            # 播放速度存入 config
            if clip.playback_speed and clip.playback_speed != 1.0:
                clip_config["playback_speed"] = clip.playback_speed
            # 片段自有 config 合并（优先级：clip.config 覆盖上面的）
            if clip.config:
                clip_config.update(clip.config)
            # 去掉素材相关字段
            clip_config.pop("asset_info", None)
            clip_config.pop("source_asset_id", None)
            # 转场效果兼容校验
            try:
                transition = TransitionEffect(clip.transition_effect)
            except ValueError:
                transition = TransitionEffect.CUT
            # 片段类型兼容校验
            try:
                clip_type = ClipType(clip.clip_type)
            except ValueError:
                clip_type = ClipType.MAIN
            clip_config_obj = TemplateClipConfig.create(
                template_id=created_template.id,
                clip_type=clip_type,
                order=clip.order,
                min_duration=clip.duration,
                max_duration=clip.duration,
                text_template=clip.text_content or "",
                transition_effect=transition,
                config=clip_config,
            )
            created = self._clip_config_repo.create(clip_config_obj)
            created_configs.append(created)
        return {
            "template": created_template,
            "clip_configs": created_configs,
        }
@@ -224,7 +224,7 @@ class PlanGeneratorService:
            )
            order += 1
            # 剩余为 overlay
-            for _ in range(1, n):
+            for i in range(1, n):
                clips.append(
                    EditPlanClip.create(
                        plan_id=plan_id,
@@ -237,7 +237,7 @@ class PlanGeneratorService:
        elif editing_mode == EditingMode.VOICE_OVER.value:
            # N 个 main clips（B-roll）
-            for _ in range(n):
+            for i in range(n):
                clips.append(
                    EditPlanClip.create(
                        plan_id=plan_id,
@@ -271,7 +271,7 @@ class PlanGeneratorService:
            )
            order += 1
            # 剩余为 b_roll
-            for _ in range(2, n):
+            for i in range(2, n):
                clips.append(
                    EditPlanClip.create(
                        plan_id=plan_id,
@@ -284,7 +284,7 @@ class PlanGeneratorService:
        else:
            # ONE_TAKE: N 个 main clips
-            for _ in range(n):
+            for i in range(n):
                clips.append(
                    EditPlanClip.create(
                        plan_id=plan_id,
@@ -33,7 +33,6 @@
        "eslint-plugin-react-hooks": "^4.6.2",
        "eslint-plugin-react-refresh": "^0.4.7",
        "jsdom": "^24.1.0",
        "prettier": "^3.0.0",
        "typescript": "^5.5.3",
        "vite": "^5.3.1",
        "vitest": "^1.6.0"
@@ -4829,22 +4828,6 @@
        "node": ">= 0.8.0"
      }
    },
    "node_modules/prettier": {
      "version": "3.9.5",
      "resolved": "https://registry.npmmirror.com/prettier/-/prettier-3.9.5.tgz",
      "integrity": "sha512-/FVl766LpUfB5vXgCYOYa0MeV/441Ia99AeICQIQFTY/Nw0roZwULcXpku5i1/m5kt/baz+s4Zogspd839HSMg==",
      "dev": true,
      "license": "MIT",
      "bin": {
        "prettier": "bin/prettier.cjs"
      },
      "engines": {
        "node": ">=14"
      },
      "funding": {
        "url": "https://github.com/prettier/prettier?sponsor=1"
      }
    },
    "node_modules/pretty-format": {
      "version": "27.5.1",
      "resolved": "https://registry.npmjs.org/pretty-format/-/pretty-format-27.5.1.tgz",
@@ -42,7 +42,6 @@
    "eslint-plugin-react-hooks": "^4.6.2",
    "eslint-plugin-react-refresh": "^0.4.7",
    "jsdom": "^24.1.0",
    "prettier": "^3.0.0",
    "typescript": "^5.5.3",
    "vite": "^5.3.1",
    "vitest": "^1.6.0"
@@ -130,8 +130,6 @@ export interface EditPlan {
  name: string;
  status: EditPlanStatus;
  total_duration: number;
  /** 生成视频数量（后端 EditPlanResponse.result_count） */
  result_count: number;
  config: EditPlanConfig;
  created_at: string;
  updated_at: string;
@@ -163,21 +161,14 @@ export interface GenerateResponse {
  clip_count: number;
 }
-/** 剪辑计划关联的生成记录（实际是 GenerationTask 对象） */
+/** 剪辑计划关联的生成记录 */
 export interface EditPlanGeneration {
-  id: string; // 即 generation_task_id
+  id: string;
-  source_edit_plan_id: string;
+  edit_plan_id: string;
-  template_id: string;
+  generation_task_id: string;
  asset_ids: string[];
  status: EditPlanStatus;
-  progress: number;
+  created_at: string;
-  result_count: number;
+  updated_at: string;
  error_message: string;
  error_info: Record<string, unknown>;
  logs: Array<Record<string, unknown>>;
  retry_count: number;
  created_at?: string;
  updated_at?: string;
 }
 /** 片段生成状态 */
@@ -1,6 +1,5 @@
 /**
 * 成品 / 视频相关 API
 * 包含：列表查询、复核状态、批量下载
 * 后端无 /products 路由，实际从 /generation/tasks 端点获取数据
 */
 import apiClient from "./client";
@@ -16,22 +16,17 @@ import type { EditPlanConfig } from "./editPlans";
 /** 模板条目（后端 TemplateResponse） */
 export interface TemplateItem {
  id: string;
  user_id?: string;
  name: string;
-  description?: string;
+  description: string;
  mode?: string;
  category: string;
  tags?: string[];
-  /** 预估时长（后端字段名 estimated_duration） */
+  target_duration: number;
-  estimated_duration?: number;
+  clip_count: number;
  /** @deprecated 后端已改名为 estimated_duration，保留兼容 */
  target_duration?: number;
  clip_count?: number;
  /** 使用次数 */
  usage_count?: number;
  thumbnail_url?: string;
  preview_url?: string;
-  is_active?: boolean;
+  is_active: boolean;
  is_favorite?: boolean;
  /** 素材规则（片段配置） */
  segments?: TemplateSegment[];
@@ -6,7 +6,7 @@ import React from "react";
 import ReactDOM from "react-dom/client";
 import { RouterProvider } from "react-router-dom";
 import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
-import { ConfigProvider, App as AntApp } from "antd";
+import { ConfigProvider } from "antd";
 import zhCN from "antd/locale/zh_CN";
 import router from "./router";
 import "./index.css";
@@ -91,9 +91,7 @@ ReactDOM.createRoot(document.getElementById("root")!).render(
  <React.StrictMode>
    <QueryClientProvider client={queryClient}>
      <ConfigProvider locale={zhCN} theme={theme}>
-        <AntApp>
+        <RouterProvider router={router} />
          <RouterProvider router={router} />
        </AntApp>
      </ConfigProvider>
    </QueryClientProvider>
  </React.StrictMode>,
@@ -33,9 +33,8 @@ const formatSize = (bytes: number) => {
 /** 格式化时长 */
 const formatDuration = (seconds?: number) => {
  if (!seconds) return "-";
-  const totalSec = Math.round(seconds);
+  const m = Math.floor(seconds / 60);
-  const m = Math.floor(totalSec / 60);
+  const s = seconds % 60;
  const s = totalSec % 60;
  return m > 0 ? `${m}分${s}秒` : `${s}秒`;
 };
@@ -59,9 +59,8 @@ const formatSize = (bytes: number) => {
 /** 格式化时长 */
 const formatDuration = (seconds?: number) => {
  if (!seconds) return "-";
-  const totalSec = Math.round(seconds);
+  const m = Math.floor(seconds / 60);
-  const m = Math.floor(totalSec / 60);
+  const s = seconds % 60;
  const s = totalSec % 60;
  return m > 0 ? `${m}分${s}秒` : `${s}秒`;
 };
@@ -86,9 +86,8 @@ const STATUS_CONFIG: Record<
 /** 格式化时长 */
 const formatDuration = (seconds: number): string => {
  if (seconds <= 0) return "-";
-  const totalSec = Math.round(seconds);
+  const m = Math.floor(seconds / 60);
-  const m = Math.floor(totalSec / 60);
+  const s = seconds % 60;
  const s = totalSec % 60;
  if (m === 0) return `${s}秒`;
  return `${m}分${s > 0 ? `${s}秒` : ""}`;
 };
@@ -254,16 +253,6 @@ export default function EditPlans() {
        <span className="plan-duration">{formatDuration(seconds)}</span>
      ),
    },
    {
      title: "视频数",
      dataIndex: "result_count",
      key: "result_count",
      width: 80,
      align: "center",
      render: (count: number) => (
        <span className="plan-result-count">{count > 0 ? count : "—"}</span>
      ),
    },
    {
      title: "创建时间",
      dataIndex: "created_at",
@@ -1812,27 +1812,7 @@
   ═══════════════════════════════════════ */
 .ep-status-bar {
-  display: flex;
+  display: none;
  align-items: center;
  justify-content: space-between;
  padding: 6px 16px;
  background: var(--ep-bg-card, #fff);
  border-bottom: 1px solid var(--ep-border, #e8e8e8);
  font-size: 12px;
  color: var(--text-secondary, #666);
  flex-shrink: 0;
 }
 .ep-status-left,
 .ep-status-right {
  display: flex;
  align-items: center;
  gap: 4px;
 }
 .ep-status-sep {
  margin: 0 4px;
  opacity: 0.35;
 }
 /* ═══════════════════════════════════════
@@ -3,8 +3,8 @@
 * 四行布局：顶栏(42px) → 模式栏(48px) → 三栏主体 → 底栏(40px)
 */
 import React, { useState, useCallback, useEffect, useRef } from "react";
-import { useSearchParams } from "react-router-dom";
+import { useSearchParams, useNavigate } from "react-router-dom";
-import { message, Modal, Progress, Button } from "antd";
+import { message } from "antd";
 import { useQuery } from "@tanstack/react-query";
 import type {
  EditingTemplate,
@@ -20,23 +20,11 @@ import {
  getTemplateCategories,
  MODE_LABELS,
 } from "@/api/editingPlanner";
-import type {
+import type { EditPlanGeneration, MediaAsset } from "@/api/editPlans";
  EditPlanGeneration,
  EditPlanConfig,
  GeneratedVideo,
  MediaAsset,
  TransitionEffect,
 } from "@/api/editPlans";
 import {
  getMediaAssets,
  getEditPlanGenerations,
  generateCover,
  getEditPlan,
  createEditPlan,
  updateEditPlan,
  generateEditPlan,
  getGenerationStatus,
  getGenerationTaskResults,
 } from "@/api/editPlans";
 import { useUndoRedo } from "./hooks/useUndoRedo";
 import type {
@@ -45,7 +33,6 @@ import type {
  TransitionConfig,
  SpeedConfig,
  TtsConfig,
  TtsMode,
  TrimConfig,
  WatermarkConfig,
  IntroOutroConfig,
@@ -120,8 +107,8 @@ const FILTER_CATEGORIES = ["全部", "种草", "知识", "日常", "推荐"];
 const EditingPlanner: React.FC = () => {
  const [searchParams] = useSearchParams();
  const navigate = useNavigate();
  const urlTemplateId = searchParams.get("templateId") || "";
  const urlPlanId = searchParams.get("planId") || "";
  /* ── 模板列表 ── */
  const [templates, setTemplates] = useState<EditingTemplate[]>([]);
@@ -257,21 +244,6 @@ const EditingPlanner: React.FC = () => {
  const [genHistory, setGenHistory] = useState<EditPlanGeneration[]>([]);
  const [genHistoryLoading, setGenHistoryLoading] = useState(false);
  /* ── 剪辑计划（从列表页编辑进入时） ── */
  const [loadedPlanId, setLoadedPlanId] = useState<string | null>(
    urlPlanId || null,
  );
  /* ── 生成进度 ── */
  const [generating, setGenerating] = useState(false);
  const [genProgress, setGenProgress] = useState(0);
  const [genTotalClips, setGenTotalClips] = useState(0);
  const [genDoneClips, setGenDoneClips] = useState(0);
  const [generated, setGenerated] = useState(false);
  const [generatedVideos, setGeneratedVideos] = useState<GeneratedVideo[]>([]);
  const [genError, setGenError] = useState<string | null>(null);
  const genTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
  /* ── 播放 ── */
  const [isPlaying, setIsPlaying] = useState(false);
  const [currentTime, setCurrentTime] = useState(0);
@@ -402,85 +374,6 @@ const EditingPlanner: React.FC = () => {
      .catch(() => message.error("加载模板详情失败"));
  }, [loadedTemplateId, resetClips]);
  /**
   * 加载已有剪辑计划数据到编辑器
   * 从列表页"编辑"按钮进入时，URL 带 planId，需要还原计划配置
   */
  useEffect(() => {
    if (!loadedPlanId) return;
    getEditPlan(loadedPlanId)
      .then((plan) => {
        // 设置关联的模板（触发模板加载 effect）
        setLoadedTemplateId(plan.template_id);
        // 还原基本信息
        setDraftName(plan.name);
        // 还原 config 中的编辑器状态
        const cfg = plan.config;
        if (cfg.title_config) {
          setTitleSettings((prev) => ({
            ...prev,
            aiAutoSelect: cfg.title_config!.ai_auto_select,
            title: cfg.title_config!.content,
            position: cfg.title_config!.position,
            font: cfg.title_config!.font_preset,
            size: cfg.title_config!.font_size,
            color: cfg.title_config!.font_color || "#ffffff",
          }));
        }
        if (cfg.subtitle_config) {
          setSubtitleSettings((prev) => ({
            ...prev,
            enabled: cfg.subtitle_config!.enabled,
            position: (cfg.subtitle_config!.position ||
              "bottom") as SubtitleStyleConfig["position"],
            font: cfg.subtitle_config!.font,
            fontSize: cfg.subtitle_config!.size,
            fontColor: cfg.subtitle_config!.color || "#ffffff",
            animation: cfg.subtitle_config!.animation,
          }));
        }
        if (cfg.bgm_config) {
          setBgmSettings((prev) => ({
            ...prev,
            enabled: cfg.bgm_config!.enabled,
            music_id: cfg.bgm_config!.music_id || "",
          }));
        }
        // 还原片段 — 延迟设置，等模板加载 effect 先执行 resetClips
        if (cfg.segments && cfg.segments.length > 0) {
          const mapped: ClipData[] = cfg.segments.map((seg, idx) => ({
            id: `seg-${idx}`,
            template_segment_id: `seg-${idx}`,
            type: (seg.material_type === "voiceover"
              ? "voice"
              : "pip") as ClipType,
            duration: (seg.duration_min + seg.duration_max) / 2,
            startOffset: 0,
            script_text: "",
            order: seg.segment_order,
            transition: seg.transition
              ? {
                  type: seg.transition.type as TransitionEffect["type"],
                  duration: seg.transition.duration,
                }
              : undefined,
            speed: seg.playback_speed
              ? { rate: seg.playback_speed, pitchCorrection: true }
              : undefined,
            tts_config: seg.tts_config
              ? { ...seg.tts_config, mode: seg.tts_config.mode as TtsMode }
              : undefined,
            trim_config: seg.trim_config || undefined,
          }));
          setTimeout(() => resetClips(mapped), 100);
        }
      })
      .catch(() => message.error("加载剪辑计划失败"));
  }, [loadedPlanId, resetClips]);
  /* ──────────── 计算 ──────────── */
  const currentTemplate = templates.find((t) => t.id === loadedTemplateId);
@@ -778,65 +671,6 @@ const EditingPlanner: React.FC = () => {
    }
  };
  /** 构建剪辑计划 config（编辑器状态 → API config） */
  const buildPlanConfig = (): EditPlanConfig => ({
    title_config: {
      ai_auto_select: titleSettings.aiAutoSelect,
      content: titleSettings.title,
      position: titleSettings.position,
      font_preset: titleSettings.font,
      font_color: titleSettings.color,
      font_size: titleSettings.size,
    },
    subtitle_config: {
      enabled: subtitleSettings.enabled,
      position: subtitleSettings.position,
      font: subtitleSettings.font,
      color: subtitleSettings.fontColor,
      size: subtitleSettings.fontSize,
      animation: subtitleSettings.animation,
    },
    bgm_config: {
      enabled: bgmSettings.enabled,
      music_id: bgmSettings.music_id,
    },
    estimated_duration: totalDuration,
    segments: clips.map((c, i) => ({
      segment_order: i,
      duration_min: Math.max(1, c.duration - 2),
      duration_max: c.duration + 2,
      material_type: c.type === "voice" ? "voiceover" : "video",
      transition: c.transition
        ? { type: c.transition.type, duration: c.transition.duration }
        : undefined,
      playback_speed: c.speed ? c.speed.rate : undefined,
      tts_config: c.tts_config
        ? {
            mode: c.tts_config.mode,
            text: c.tts_config.text,
            voice_id: c.tts_config.voice_id,
            speed: c.tts_config.speed,
            pitch: c.tts_config.pitch,
            volume: c.tts_config.volume,
            subtitle_sync: c.tts_config.subtitle_sync,
          }
        : undefined,
      trim_config: c.trim_config
        ? {
            start_time: c.trim_config.start_time,
            end_time: c.trim_config.end_time,
          }
        : undefined,
    })),
    watermark_config: { ...watermarkSettings },
    intro_outro_config: { ...introOutroSettings },
    pip_config: { ...pipSettings },
    filter_config: { ...filterSettings },
    green_screen_config: { ...chromaKeySettings },
    sticker_config: { ...stickerSettings },
    cover_config: { ...coverSettings },
  });
  /* 保存 — 无论是否已加载模板，都打开保存弹窗；未加载时创建新模板 */
  const handleOpenSaveModal = () => {
    setSaveModalOpen(true);
@@ -929,148 +763,94 @@ const EditingPlanner: React.FC = () => {
  };
  /**
-   * 剪辑计划生成
+   * 跳转到一键生成页面
-   * 1. 有 planId → 更新计划配置 + 触发生成
+   * 通过 URL SearchParams 传递 edit_plan_id 和完整 planConfig（JSON 序列化）
-   * 2. 无 planId（从模板库直接进入）→ 先创建计划 + 触发生成
+   * 一键生成页面从 params 解析配置，无需重复请求接口
   * 3. 触发生成后轮询状态，完成后获取视频结果
   */
-  const handleGoToGenerate = async () => {
+  const handleGoToGenerate = () => {
-    if (!loadedTemplateId) {
+    const planConfig = {
-      message.warning("请先选择一个模板");
+      title_config: {
-      return;
+        ai_auto_select: titleSettings.aiAutoSelect,
-    }
+        content: titleSettings.title,
-    if (clips.length === 0) {
+        position: titleSettings.position,
-      message.warning("请先添加片段");
+        font_preset: titleSettings.font,
-      return;
+        font_color: titleSettings.color,
-    }
+        font_size: titleSettings.size,
-
+        bold: titleSettings.bold,
-    setGenerating(true);
+        italic: titleSettings.italic,
-    setGenerated(false);
+        stroke: titleSettings.stroke,
-    setGeneratedVideos([]);
+        shadow: titleSettings.shadow,
-    setGenError(null);
+      },
-    setGenProgress(0);
+      subtitle_config: {
-
+        enabled: subtitleSettings.enabled,
-    try {
+        position: subtitleSettings.position,
-      const config = buildPlanConfig();
+        font: subtitleSettings.font,
-      let planId = loadedPlanId;
+        color: subtitleSettings.fontColor,
-
+        size: subtitleSettings.fontSize,
-      if (planId) {
+        animation: subtitleSettings.animation,
-        // 已有计划 → 先重置状态为 draft（failed/editing 等非 draft 状态会被后端拒绝更新和生成）
+      },
-        try {
+      bgm_config: {
-          await updateEditPlan(planId, { status: "draft" });
+        enabled: bgmSettings.enabled,
-        } catch (resetErr) {
+        music_id: bgmSettings.music_id,
-          console.warn("[状态重置跳过]", resetErr);
+      },
-        }
+      mode: currentMode,
-        // 再更新配置
+      total_duration: totalDuration,
-        try {
+      segments: clips.map((c, i) => ({
-          await updateEditPlan(planId, {
+        order: i,
-            config,
+        material_type: c.type === "voice" ? "voiceover" : "video",
-            total_duration: totalDuration,
+        duration: c.duration,
-          });
+        template_segment_id: c.template_segment_id,
-        } catch (updateErr) {
+        script_text: c.script_text,
-          console.warn("[计划更新跳过]", updateErr);
+        voice_asset_id: c.voice_asset_id,
-        }
+        voice_file_url: c.voice_file_url,
-      } else {
+        transition: c.transition
-        // 无计划 → 创建新计划
+          ? { type: c.transition.type, duration: c.transition.duration }
-        const plan = await createEditPlan({
+          : undefined,
-          template_id: loadedTemplateId,
+        playback_speed: c.speed ? c.speed.rate : undefined,
-          name: draftName || "未命名计划",
+        tts_config: c.tts_config
-          config,
+          ? {
-          total_duration: totalDuration,
+              mode: c.tts_config.mode,
-        });
+              text: c.tts_config.text,
-        planId = plan.id;
+              voice_id: c.tts_config.voice_id,
-        setLoadedPlanId(planId);
+              speed: c.tts_config.speed,
-        // 更新 URL 参数（不刷新页面）
+              pitch: c.tts_config.pitch,
-        const params = new URLSearchParams(window.location.search);
+              volume: c.tts_config.volume,
-        params.set("planId", planId);
+              subtitle_sync: c.tts_config.subtitle_sync,
        window.history.replaceState(null, "", `?${params.toString()}`);
      }
      // 触发生成
      const genRes = await generateEditPlan(planId);
      setGenTotalClips(genRes.clip_count);
      message.info("已提交生成，等待处理...");
      // 开始轮询
      startPolling(planId);
    } catch (err) {
      console.error("[生成失败]", err);
      setGenError("生成提交失败，请重试");
      setGenerating(false);
    }
  };
  /** 轮询生成状态，每 2 秒一次 */
  const startPolling = (planId: string) => {
    const poll = async () => {
      try {
        const status = await getGenerationStatus(planId);
        // 计算进度
        const total = status.clips.length || genTotalClips;
        const done = status.clips.filter(
          (c) => c.status === "completed" || c.status === "failed",
        ).length;
        setGenDoneClips(done);
        setGenTotalClips(total);
        setGenProgress(total > 0 ? Math.round((done / total) * 100) : 5);
        if (status.plan_status === "completed") {
          setGenProgress(100);
          setGenerating(false);
          setGenerated(true);
          // 获取视频结果
          if (status.generation_task_id) {
            try {
              const videos = await getGenerationTaskResults(
                status.generation_task_id,
              );
              setGeneratedVideos(videos);
            } catch (e) {
              console.error("[获取视频结果失败]", e);
            }
-          }
+          : undefined,
-          message.success("视频生成完成！");
+        trim_config: c.trim_config
-          return; // 停止轮询
+          ? {
-        }
+              start_time: c.trim_config.start_time,
-
+              end_time: c.trim_config.end_time,
-        if (status.plan_status === "failed") {
+            }
-          setGenerating(false);
+          : undefined,
-          setGenError("生成失败，请重试");
+      })),
-          return; // 停止轮询
+      watermark_config: { ...watermarkSettings },
-        }
+      intro_outro_config: { ...introOutroSettings },
-
+      pip_config: { ...pipSettings },
-        // 继续轮询
+      filter_config: { ...filterSettings },
-        genTimerRef.current = setTimeout(poll, 2000);
+      green_screen_config: { ...chromaKeySettings },
-      } catch (err) {
+      sticker_config: { ...stickerSettings },
-        console.error("[轮询状态失败]", err);
+      cover_config: { ...coverSettings },
        genTimerRef.current = setTimeout(poll, 5000); // 出错后 5 秒重试
      }
    };
-
+    const params = new URLSearchParams();
-    // 首次延迟 2 秒后开始
+    if (loadedTemplateId) {
-    genTimerRef.current = setTimeout(poll, 2000);
+      params.set("edit_plan_id", loadedTemplateId);
    }
    params.set("plan_config", JSON.stringify(planConfig));
    navigate(`/app/generate?${params.toString()}`);
  };
  /** 清理轮询定时器 */
  useEffect(() => {
    return () => {
      if (genTimerRef.current) clearTimeout(genTimerRef.current);
    };
  }, []);
  /* 查看生成历史 */
  const handleViewGenHistory = async () => {
-    const targetId = loadedPlanId || loadedTemplateId;
+    if (!loadedTemplateId) {
-    if (!targetId) {
+      message.warning("请先加载一个模板");
      message.warning("请先加载一个模板或计划");
      return;
    }
    setGenHistoryOpen(true);
    setGenHistoryLoading(true);
    try {
-      const items = await getEditPlanGenerations(targetId);
+      const items = await getEditPlanGenerations(loadedTemplateId);
      setGenHistory(items);
    } catch {
      message.error("加载生成历史失败");
@@ -1119,9 +899,8 @@ const EditingPlanner: React.FC = () => {
          <button
            className="ep-btn ep-btn-primary"
            onClick={handleGoToGenerate}
            disabled={generating}
          >
-            {loadedPlanId ? "🎬 生成视频" : "🎬 创建计划并生成"}
+            🎬 使用此模板生成
          </button>
        </div>
      </div>
@@ -1286,111 +1065,6 @@ const EditingPlanner: React.FC = () => {
        onClose={() => setGenHistoryOpen(false)}
      />
      {/* ═══ 生成进度弹窗 ═══ */}
      <Modal
        title={genError ? "生成失败" : generated ? "生成完成" : "正在生成视频"}
        open={generating || generated || !!genError}
        footer={
          generated
            ? [
                <Button
                  key="close"
                  onClick={() => {
                    setGenerated(false);
                    setGenerating(false);
                  }}
                >
                  关闭
                </Button>,
                generatedVideos.length > 0 && (
                  <Button
                    key="download"
                    type="primary"
                    onClick={() => {
                      const v = generatedVideos[0];
                      const url = v.download_url || v.file_url;
                      if (url) {
                        const a = document.createElement("a");
                        a.href = url;
                        a.download = v.name || "video.mp4";
                        a.target = "_blank";
                        document.body.appendChild(a);
                        a.click();
                        document.body.removeChild(a);
                      }
                    }}
                  >
                    下载视频
                  </Button>
                ),
              ]
            : null
        }
        closable={!generating}
        maskClosable={false}
        width={520}
      >
        {generating && (
          <div style={{ padding: "16px 0" }}>
            <Progress percent={genProgress} status="active" />
            <p style={{ marginTop: 8, color: "var(--text-secondary)" }}>
              已处理 {genDoneClips}/{genTotalClips} 个片段
            </p>
            <p style={{ color: "var(--text-secondary)", fontSize: 12 }}>
              请耐心等待，生成过程中请勿关闭页面
            </p>
          </div>
        )}
        {generated && generatedVideos.length > 0 && (
          <div style={{ padding: "8px 0" }}>
            <video
              src={
                generatedVideos[0].file_url || generatedVideos[0].download_url
              }
              controls
              preload="metadata"
              style={{ width: "100%", maxHeight: 320, borderRadius: 8 }}
            />
            <p
              style={{
                marginTop: 8,
                textAlign: "center",
                color: "var(--text-secondary)",
              }}
            >
              {generatedVideos[0].name}
            </p>
          </div>
        )}
        {generated && !generatedVideos.length && (
          <div style={{ padding: "24px 0", textAlign: "center" }}>
            <p>生成完成，但暂未获取到视频结果</p>
            <p style={{ color: "var(--text-secondary)", fontSize: 12 }}>
              请稍后在剪辑计划列表中查看
            </p>
          </div>
        )}
        {genError && (
          <div
            style={{
              padding: "16px 0",
              textAlign: "center",
              color: "#ff4d4f",
            }}
          >
            <p>{genError}</p>
            <Button
              onClick={() => {
                setGenError(null);
                setGenerating(false);
              }}
            >
              关闭
            </Button>
          </div>
        )}
      </Modal>
      {/* ═══ BGM 选择器 Drawer ═══ */}
      <BgmSelector
        open={bgmDrawerOpen}
@@ -65,7 +65,7 @@ const GenerationHistoryModal: React.FC<GenerationHistoryModalProps> = ({
                  return (
                    <tr key={gen.id} className="ep-gh-table-row">
                      <td className="ep-gh-td ep-gh-td-id">
-                        {gen.id ? `${gen.id.slice(0, 8)}...` : "—"}
+                        {gen.generation_task_id.slice(0, 8)}...
                      </td>
                      <td className="ep-gh-td">
                        <span className={`ep-gh-status-tag ${statusClass}`}>
@@ -23,6 +23,7 @@ import {
  getTemplate,
  toggleFavoriteTemplate,
  copyTemplate,
  generateFromTemplate,
  type TemplateItem,
  type TemplateListParams,
  type TemplateSegment,
@@ -90,11 +91,10 @@ const gradientForCategory = (category: string): string => {
 };
 /** 格式化时长 */
-const formatDuration = (seconds: number | undefined | null): string => {
+const formatDuration = (seconds: number): string => {
-  if (!seconds || seconds <= 0) return "0秒";
+  if (seconds <= 0) return "0秒";
-  const totalSec = Math.round(seconds);
+  const m = Math.floor(seconds / 60);
-  const m = Math.floor(totalSec / 60);
+  const s = seconds % 60;
  const s = totalSec % 60;
  if (m === 0) return `${s}秒`;
  return `${m}分${s > 0 ? `${s}秒` : ""}`;
 };
@@ -237,9 +237,7 @@ const TemplateDetailModal: React.FC<TemplateDetailModalProps> = ({
              {
                key: "duration",
                label: "目标时长",
-                children: formatDuration(
+                children: formatDuration(template.target_duration),
                  template.estimated_duration ?? template.target_duration,
                ),
              },
              {
                key: "clips",
@@ -411,9 +409,7 @@ const TemplateCard: React.FC<TemplateCardProps> = ({
        <div className="xx-template-thumb-name">{template.name}</div>
        <div className="xx-template-thumb-meta">
          <span className="xx-template-thumb-duration">
-            {formatDuration(
+            {formatDuration(template.target_duration)}
              template.estimated_duration ?? template.target_duration,
            )}
          </span>
        </div>
        <div className="xx-template-preview-hint">点击查看详情</div>
@@ -541,6 +537,19 @@ const TemplateLibrary: React.FC = () => {
    },
  });
  // ── 从模板生成剪辑计划 mutation ──
  const generateMutation = useMutation({
    mutationFn: ({ templateId, name }: { templateId: string; name: string }) =>
      generateFromTemplate(templateId, { name }),
    onSuccess: (data) => {
      message.success(`剪辑计划「${data.name}」已创建`);
      navigate("/app/edit-plans");
    },
    onError: () => {
      message.error("生成剪辑计划失败，请稍后重试");
    },
  });
  /** 切换收藏 */
  const toggleFavorite = useCallback(
    (id: string, e?: React.MouseEvent) => {
@@ -573,12 +582,15 @@ const TemplateLibrary: React.FC = () => {
    [copyMutation],
  );
-  /** 使用模板 → 进入剪辑编辑器配置 */
+  /** 使用模板 → 生成剪辑计划 */
  const handleUse = useCallback(
    (template: TemplateItem) => {
-      navigate(`/app/editing-planner?templateId=${template.id}`);
+      generateMutation.mutate({
        templateId: template.id,
        name: `基于「${template.name}」的剪辑计划`,
      });
    },
-    [navigate],
+    [generateMutation, navigate],
  );
  /** 搜索防抖处理 */
@@ -37,7 +37,6 @@ export default defineConfig({
    },
  },
  build: {
    cache: true,
    rollupOptions: {
      output: {
        manualChunks: {
@@ -181,9 +181,9 @@ def _validate_video_path(video_path: str, work_dir: Path) -> None:
            resolved_work_dir = work_dir.resolve()
            try:
                resolved_path.relative_to(resolved_work_dir)
-            except ValueError as _e:
+            except ValueError:
                if not is_in_allowed_dirs(resolved_path):
-                    raise PathSecurityError(f"视频路径不在允许目录内: {video_path[:80]}") from _e
+                    raise PathSecurityError(f"视频路径不在允许目录内: {video_path[:80]}")
    # URL类型路径不做本地路径校验（由下载阶段的SSRF防护负责）
    # 但检查扩展名
    else:
@@ -356,7 +356,7 @@ def check_duplicate_task(self: Task, generated_video_id: str) -> dict:
    except Exception as e:
        logger.error(f"Duplicate check failed for {generated_video_id}: {str(e)}")
        session.rollback()
-        raise self.retry(exc=e, countdown=60) from e
+        raise self.retry(exc=e, countdown=60)
    finally:
        session.close()
        import shutil
@@ -191,9 +191,9 @@ def _validate_audio_path(audio_path: str, work_dir: Path) -> None:
            resolved_work_dir = work_dir.resolve()
            try:
                resolved_path.relative_to(resolved_work_dir)
-            except ValueError as _e:
+            except ValueError:
                if not is_in_allowed_dirs(resolved_path):
-                    raise PathSecurityError(f"音频路径不在允许目录内: {audio_path[:80]}") from _e
+                    raise PathSecurityError(f"音频路径不在允许目录内: {audio_path[:80]}")
    # URL类型路径不做本地路径校验（由下载阶段的SSRF防护负责）
    # 但检查扩展名
    else:
@@ -129,8 +129,8 @@ def safe_resolve_path(
    if not allow_outside:
        try:
            full_path.relative_to(base_dir)
-        except ValueError as _e:
+        except ValueError:
-            raise PathSecurityError(f"路径遍历检测：路径 '{path_str}' 超出基路径 '{base_dir}' 范围") from _e
+            raise PathSecurityError(f"路径遍历检测：路径 '{path_str}' 超出基路径 '{base_dir}' 范围")
    # 扩展名校验
    if allowed_extensions is not None:
@@ -403,7 +403,7 @@ class PiPEngine:
        input_args: list[str] = []
        current_label = base_label
-        for i, (_input_label, layer, path) in enumerate(pip_sources):
+        for i, (input_label, layer, path) in enumerate(pip_sources):
            # 添加输入
            input_args.extend(["-i", str(path)])
@@ -359,7 +359,7 @@ class StickerEngine:
        image_stickers: list[ImageStickerConfig] = []
        image_paths: list[str] = []
-        for _, s in enumerate(stickers):
+        for i, s in enumerate(stickers):
            try:
                sticker_type = s.get("type", "image")
                z = int(s.get("z_index", 10))
@@ -682,6 +682,6 @@ def _validate_subtitle_path(subtitle_path: str, work_dir: Path) -> None:
        resolved_work_dir = work_dir.resolve()
        try:
            resolved_path.relative_to(resolved_work_dir)
-        except ValueError as _e:
+        except ValueError:
            if not is_in_allowed_dirs(resolved_path):
-                raise PathSecurityError(f"字幕路径不在允许目录内: {subtitle_path[:80]}") from _e
+                raise PathSecurityError(f"字幕路径不在允许目录内: {subtitle_path[:80]}")
@@ -1363,37 +1363,23 @@ class UnifiedRenderService:
                # 单 clip 层，直接使用预处理标签
                layer_output_labels[layer.role] = layer_labels[0]
            else:
                # 多 clip 层，用 TransitionEngine 构建转场链
                out_label = f"{layer.role}_merged"
-                # 判断是否全部为硬切：是则用 concat filter，否则用 xfade 转场链
+                # 计算该层使用的转场时长（取首个非零值，否则用默认）
-                all_cut = all(
+                layer_dur = 0.0
-                    t is None or t == "" or str(t).lower() == "cut"
+                for d in layer_transition_durations:
-                    for t in layer_transitions[1:]  # 第一个 clip 的转场忽略
+                    if d > 0:
                        layer_dur = d
                        break
                xfade_filter, _ = self._transition_engine.build_xfade_chain(
                    clip_durations=layer_durations,
                    clip_video_labels=layer_labels,
                    transitions=layer_transitions,
                    transition_duration=layer_dur if layer_dur > 0 else None,
                    output_label=out_label,
                )
-                if all_cut:
+                if xfade_filter:
-                    # 全硬切：用 concat filter，性能远优于 xfade
+                    filter_parts.append(xfade_filter)
                    concat_inputs = "".join(f"[{label}]" for label in layer_labels)
                    filter_parts.append(f"{concat_inputs}concat=n={len(layer_labels)}:v=1:a=0[{out_label}]")
                    logger.info(
                        "[unified-render] layer=%s clips=%d using concat (all hard-cut)",
                        layer.role,
                        len(layer_labels),
                    )
                else:
                    # 有转场效果：用 TransitionEngine 构建 xfade 链
                    layer_dur = 0.0
                    for d in layer_transition_durations:
                        if d > 0:
                            layer_dur = d
                            break
                    xfade_filter, _ = self._transition_engine.build_xfade_chain(
                        clip_durations=layer_durations,
                        clip_video_labels=layer_labels,
                        transitions=layer_transitions,
                        transition_duration=layer_dur if layer_dur > 0 else None,
                        output_label=out_label,
                    )
                    if xfade_filter:
                        filter_parts.append(xfade_filter)
                layer_output_labels[layer.role] = out_label
        # Step 3: 合成各层
@@ -187,7 +187,7 @@ class AssetAnalyzer:
        if self._frames is not None:
            return self._frames
-        frames: list[np.ndarray] = []
+        frames = []
        info = self.get_video_info()
        if info.duration <= 0:
@@ -398,7 +398,7 @@ class AssetAnalyzer:
                run_ffmpeg(cmd, timeout=30)
            except Exception:
                # 音频提取失败，返回默认分析结果
-                return AudioAnalysis(  # type: ignore[call-arg]
+                return AudioAnalysis(
                    has_speech=False,
                    speech_ratio=0.0,
                    avg_volume=0.0,
@@ -63,18 +63,6 @@ def compose_video(self, job_id: str, **kwargs):
        resolver = get_render_engine_resolver()
        user_id = job.created_by_user_id or None
        engine = resolver.get_engine(user_id=user_id)
        # 灰度期间打印详细 flag 配置，便于排查
        config = resolver.get_config_snapshot()
        logger.info(
            "compose_video 引擎选择: job_id=%s engine=%s user_id=%s enabled=%s percentage=%s whitelist=%d default=%s",
            job_id,
            engine,
            user_id,
            config.get("enabled"),
            config.get("percentage"),
            len(config.get("whitelist", [])),
            config.get("default_engine"),
        )
        if engine == "unified":
            return _compose_with_unified_engine(self, job_service, job, plan_id, db)
@@ -90,7 +78,7 @@ def compose_video(self, job_id: str, **kwargs):
            job_service.fail_job(job_id, str(exc)[:500])
        except Exception:
            logger.exception("更新 Job 失败状态时出错")
-        raise self.retry(exc=exc, countdown=60) from exc
+        raise self.retry(exc=exc, countdown=60)
    finally:
        db.close()
@@ -77,21 +77,9 @@ def _resolve_render_engine(user_id: str) -> str:
        from video_processing.render_engine_resolver import get_render_engine_resolver
        resolver = get_render_engine_resolver()
-        engine = resolver.get_engine(user_id=user_id)
+        return resolver.get_engine(user_id=user_id)
        # 灰度期间打印详细 flag 配置，便于排查
        config = resolver.get_config_snapshot()
        logger.info(
            "edit_plan 引擎选择: user_id=%s engine=%s enabled=%s percentage=%s whitelist=%d default=%s",
            user_id,
            engine,
            config.get("enabled"),
            config.get("percentage"),
            len(config.get("whitelist", [])),
            config.get("default_engine"),
        )
        return engine
    except Exception as exc:
-        logger.warning("获取渲染引擎配置失败，fallback 到 legacy: %s", exc, exc_info=True)
+        logger.warning("获取渲染引擎配置失败，fallback 到 legacy: %s", exc)
        return "legacy"
@@ -107,14 +95,6 @@ def _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, err
            gen_task.status = "failed"
            gen_task.error_message = error_msg
            gen_task.completed_at = datetime.now(timezone.utc)
            try:
                gen_task.append_log(
                    stage="render_failed",
                    message=error_msg[:500],
                    level="ERROR",
                )
            except Exception:
                pass
            gen_task_repo.update(gen_task)
@@ -168,13 +148,9 @@ def _finalize_render_success(
            clip.mark_rendered()
            clip_repo.update(clip)
-    # 更新 EditPlan 状态为 completed + 回写实际渲染时长 + 结果数
+    # 更新 EditPlan 状态为 completed
    plan.config["rendered_url"] = output_url or ""
    plan.config["rendered_storage_key"] = storage_key
    if hasattr(plan, "total_duration") and duration > 0:
        plan.total_duration = duration
    if hasattr(plan, "result_count"):
        plan.result_count = 1
    plan.mark_completed()
    plan_repo.update(plan)
@@ -184,15 +160,7 @@ def _finalize_render_success(
        if gen_task:
            gen_task.status = "completed"
            gen_task.progress = 100.0
-            # 剪辑计划是多片段合成 1 个成片，result_count = 1
+            gen_task.result_count = len(rendered_clip_ids)
            gen_task.result_count = 1
            gen_task.append_log(
                stage="render_complete",
                message=f"渲染完成，输出时长 {duration:.1f}s",
                level="INFO",
                engine=engine,
                clip_count=len(rendered_clip_ids),
            )
            gen_task.completed_at = datetime.now(timezone.utc)
            gen_task_repo.update(gen_task)
@@ -397,13 +365,6 @@ def render_edit_plan(self, plan_id: str) -> dict:
                if gen_task:
                    gen_task.status = "running"
                    gen_task.started_at = datetime.now(timezone.utc)
                    gen_task.append_log(
                        stage="render_start",
                        message=f"开始渲染，引擎 {engine}，片段数 {len(clips)}",
                        level="INFO",
                        engine=engine,
                        clip_count=len(clips),
                    )
                    gen_task_repo.update(gen_task)
            # 3. 下载素材并构建 asset_path_map
@@ -468,28 +429,9 @@ def render_edit_plan(self, plan_id: str) -> dict:
                            gen_task.status = "failed"
                            gen_task.error_message = "所有片段素材下载失败"
                            gen_task.completed_at = datetime.now(timezone.utc)
                            gen_task.append_log(
                                stage="download_failed",
                                message="所有片段素材下载失败",
                                level="ERROR",
                            )
                            gen_task_repo.update(gen_task)
                    return {"status": "error", "message": "所有片段素材下载失败"}
                # 素材下载完成，记录日志
                if generation_task_id:
                    gen_task = gen_task_repo.get(generation_task_id)
                    if gen_task:
                        gen_task.append_log(
                            stage="download_done",
                            message=f"素材下载完成，成功 {len(asset_path_map)} 个，失败 {len(failed_clip_ids)} 个",
                            level="INFO",
                            success_count=len(asset_path_map),
                            failed_count=len(failed_clip_ids),
                        )
                        gen_task.progress = 30.0
                        gen_task_repo.update(gen_task)
                # 4. 根据引擎选择渲染方式
                if engine == "unified":
                    result = _render_with_unified(
@@ -541,15 +483,6 @@ def render_edit_plan(self, plan_id: str) -> dict:
                        gen_task.status = "failed"
                        gen_task.error_message = f"渲染异常: {type(exc).__name__}: {exc}"
                        gen_task.completed_at = datetime.now(timezone.utc)
                        try:
                            gen_task.append_log(
                                stage="render_failed",
                                message=f"渲染异常: {type(exc).__name__}: {str(exc)[:500]}",
                                level="ERROR",
                                exception_type=type(exc).__name__,
                            )
                        except Exception:
                            pass
                        gen_task_repo.update(gen_task)
                        logger.info(
                            "GenerationTask 已标记为 failed: task_id=%s plan_id=%s",
@@ -560,6 +493,6 @@ def render_edit_plan(self, plan_id: str) -> dict:
                logger.warning(
                    "更新 GenerationTask 失败状态时异常: task_id=%s error=%s", generation_task_id, e, exc_info=True
                )
-            raise self.retry(exc=exc, countdown=60) from exc
+            raise self.retry(exc=exc, countdown=60)
    return {"status": "error", "message": "数据库连接失败"}
@@ -787,23 +787,10 @@ def _resolve_render_engine(user_id: str) -> str:
        from video_processing.render_engine_resolver import get_render_engine_resolver
        resolver = get_render_engine_resolver()
-        engine = resolver.get_engine(user_id=user_id)
+        return resolver.get_engine(user_id=user_id)
        # 灰度期间打印详细 flag 配置，便于排查
        config = resolver.get_config_snapshot()
        logger.info(
            "[渲染引擎] flag 解析: user_id=%s engine=%s enabled=%s percentage=%s whitelist=%d default=%s",
            user_id,
            engine,
            config.get("enabled"),
            config.get("percentage"),
            len(config.get("whitelist", [])),
            config.get("default_engine"),
        )
        return engine
    except Exception as exc:
-        # 异常时 fallback 到 legacy（保守策略，与 edit_plan_generation 一致）
+        logger.warning("获取渲染引擎配置失败，fallback 到 unified: %s", exc)
-        logger.warning("获取渲染引擎配置失败，fallback 到 legacy: %s", exc, exc_info=True)
+        return ENGINE_UNIFIED
        return ENGINE_LEGACY
 # ── 旧引擎渲染（FFmpeg filter_complex） ────────────────────────────────────────
@@ -1418,7 +1405,7 @@ def generate_video(self, task_id: str) -> dict:
                _repo = SQLAlchemyGenerationTaskRepository(_session)
                gen_task = _repo.get(task_id)
                if gen_task:
-                    gen_task.append_log(  # type: ignore[misc]
+                    gen_task.append_log(
                        "任务失败",
                        str(error),
                        level="ERROR",
@@ -70,13 +70,13 @@ def extract_media_metadata(file_url: str, media_type: str) -> dict:
                        metadata["height"] = int(stream.get("height", 0))
                        metadata["codec"] = stream.get("codec_name", "")
                        metadata["fps"] = (
-                            _safe_parse_fps(stream.get("r_frame_rate", "0/1")) if stream.get("r_frame_rate") else 0  # type: ignore[assignment]
+                            _safe_parse_fps(stream.get("r_frame_rate", "0/1")) if stream.get("r_frame_rate") else 0
                        )
                        break
                # 提取格式信息
                format_info = probe_data.get("format", {})
-                metadata["duration"] = float(format_info.get("duration", 0))  # type: ignore[assignment]
+                metadata["duration"] = float(format_info.get("duration", 0))
                metadata["size_bytes"] = int(format_info.get("size", 0))
                metadata["bitrate"] = int(format_info.get("bit_rate", 0))
@@ -96,7 +96,7 @@ def extract_media_metadata(file_url: str, media_type: str) -> dict:
                    if hasattr(img, "_getexif") and img._getexif():
                        exif = img._getexif()
                        if exif:
-                            metadata["exif"] = {k: str(v) for k, v in exif.items() if isinstance(v, (str, int, float))}  # type: ignore[assignment]
+                            metadata["exif"] = {k: str(v) for k, v in exif.items() if isinstance(v, (str, int, float))}
            except ImportError:
                logger.warning("Pillow not available for image metadata extraction")
            except Exception as e:
@@ -63,7 +63,7 @@ def process_tts_synthesis(self: Task, job_id: str) -> dict:
        if session is not None:
            session.rollback()
        # 超时重试，指数退避
-        raise self.retry(exc=e, countdown=30) from e
+        raise self.retry(exc=e, countdown=30)
    except CosyVoiceError as e:
        logger.error(f"TTS synthesis failed for {job_id}: {e}")
@@ -139,7 +139,7 @@ def process_tts_segment_synthesis(self: Task, job_id: str) -> dict:
        logger.warning(f"TTS segment synthesis timeout for {job_id}: {e}")
        if session is not None:
            session.rollback()
-        raise self.retry(exc=e, countdown=60) from e
+        raise self.retry(exc=e, countdown=60)
    except CosyVoiceError as e:
        logger.error(f"TTS segment synthesis failed for {job_id}: {e}")
@@ -73,7 +73,7 @@ def process_voice_clone(self: Task, profile_id: str) -> dict:
        if session is not None:
            session.rollback()
        # 超时属于临时性故障，延迟 30 秒后重试
-        raise self.retry(exc=e, countdown=30) from e
+        raise self.retry(exc=e, countdown=30)
    except CosyVoiceError as e:
        logger.error(f"Voice clone failed for {profile_id}: {e}")
@@ -105,7 +105,7 @@ def extract_voice_task(self: Task, asset_id: str) -> dict:
    except Exception as e:
        logger.error(f"Voice extraction failed for {asset_id}: {str(e)}")
        session.rollback()
-        raise self.retry(exc=e, countdown=60) from e
+        raise self.retry(exc=e, countdown=60)
    finally:
        session.close()
        import shutil
@@ -141,7 +141,7 @@ def extract_background_task(self: Task, asset_id: str) -> dict:
    except Exception as e:
        logger.error(f"Background extraction failed for {asset_id}: {str(e)}")
        session.rollback()
-        raise self.retry(exc=e, countdown=60) from e
+        raise self.retry(exc=e, countdown=60)
    finally:
        session.close()
        import shutil
@@ -1,104 +0,0 @@
 # CI 大量失败根因排查报告
 **排查时间：** 2026-07-13
 **排查人：** 构建服务器运维Agent
 **范围：** 最近15次 CI run（PR #258~#265 + develop 分支多次 push）
 ## 一、整体概况
 最近 20 次 CI run 中 16 次失败，失败率 **80%**。失败集中在 3 个 Job：
 | Job | 失败率 | 根因类型 |
 |-----|--------|----------|
 | Validate Code Quality | 100% | black 代码格式检查失败 |
 | Unit Tests | 100% | 测试断言未同步国际化改动 |
 | Integration Tests | 100% | 密码重置接口变更未同步测试 |
 | Frontend Lint | 20% | 各 PR 代码质量问题 |
 **结论：3 个全局性失败点导致所有 PR CI 全红，不是代码本身问题，是基础设施/测试用例滞后。**
 ---
 ## 二、详细根因分析
 ### 1. Validate — black 格式检查失败
 **现象：**
 ```
 would reformat scripts/check_migration_safety.py
 1 file would be reformatted, 369 files would be left unchanged.
 Oh no! 💥 💔 💥
 ```
 **根因：**
 `scripts/check_migration_safety.py` 文件不符合 black 格式化规范。该文件是最近新增的迁移安全检查脚本，提交前未本地跑 black 格式化。
 **影响范围：** 所有 PR 及 develop 分支，全量失败。
 **修复方案：**
 ```bash
 black scripts/check_migration_safety.py
 ```
 ---
 ### 2. Unit Tests — 1 个用例失败
 **现象：**
 ```
 FAILED tests/unit/test_asset_library_delete.py::TestDeleteAssetLibrary::test_delete_library_access_denied
  AssertionError: assert 'Access denied' in '无权访问该项目'
 ```
 **统计：** 1442 passed, 1 failed
 **根因：**
 项目之前做了国际化（i18n）改造，错误信息从英文改成了中文，但对应的单元测试断言仍然检查英文 "Access denied"，导致断言失败。
 **影响范围：** 所有 PR 及 develop 分支，全量失败。
 **修复方案：**
 修改 `tests/unit/test_asset_library_delete.py` 中的断言，将 `'Access denied'` 改为 `'无权访问该项目'`，或改为断言 HTTP 状态码（403）而不是错误消息文本。
 ---
 ### 3. Integration Tests — 1 个用例失败
 **现象：**
 ```
 FAILED tests/integration/test_auth.py::TestPasswordReset::test_request_password_reset_success
  assert 404 in (200, 202)
 ```
 **统计：** 45 passed, 1 failed, 13 deselected, 2 rerun
 **根因：**
 密码重置请求接口（`POST /auth/password-reset/request` 或类似路由）返回 404，说明该接口已被移除、路由变更，或对应的功能模块暂时被注释/下线。
 **影响范围：** 所有 PR 及 develop 分支，全量失败。
 **修复方案：**
 - 如果接口确实下线了：删除或 skip 这个测试用例
 - 如果是路由改了：更新测试中的 API 路径
 - 如果是功能待开发：标记为 `@pytest.mark.skip` 并加上 TODO
 ---
 ## 三、修复优先级
 | 优先级 | 问题 | 修复难度 | 预估时间 |
 |--------|------|----------|----------|
 | P0 | black 格式检查失败 | ⭐ | 5分钟 |
 | P0 | 单元测试国际化断言失败 | ⭐ | 10分钟 |
 | P1 | 集成测试密码重置接口404 | ⭐⭐ | 30分钟（需确认接口状态） |
 **建议：** 先修前两个 P0（能让 2/3 的 job 变绿），再处理密码重置那个。
 ---
 ## 四、Runner 执行情况观察
 - 当前 9 个 Runner 全部在线（构建服务器 4 个 + 新服务器 5 个）
 - 失败的 Job 都是在构建服务器的 Runner 上执行的（xiaoxia-ci-runner-2/3 等）
 - 新服务器 5 个 Runner 目前全部空闲（标签修复后首次接任务可能需要时间）
 - 并发能力充足，瓶颈在代码/测试本身，不在 Runner 资源
@@ -1,136 +0,0 @@
 # 三台服务器 Runner 分工规划
 **制定日期：** 2026-07-13
 **状态：** 规划中
 ---
 ## 一、现状总览
 当前共 9 个 Gitea Actions Runner，分布在 3 台服务器上：
 | 服务器 | IP | 配置 | Runner 数量 | 当前状态 |
 |--------|-----|------|-------------|----------|
 | 构建服务器 | 114.55.236.178 | 4核 / 7.1G RAM / 49G NVMe | 4个（ID: 8, 42, 46, 47） | ✅ 在线 |
 | 新CI服务器 | 116.62.226.203 | 8核 / 14G RAM | 5个（ID: 58-62） | ✅ 在线 |
 | 业务服务器 | 47.98.113.167 | - | 0个（旧3个已下线） | ⚠️ 待规划 |
 **所有 Runner 共用标签：** `saas`, `runtime-builder`, `host`, `ubuntu-latest`
 ---
 ## 二、问题分析
 ### 2.1 标签无区分
 所有 Runner 标签完全一致，CI 任务随机分配到任意 Runner，导致：
 - 构建任务（Build）可能跑到配置低的机器上，构建慢
 - 代码检查任务占着构建服务器，影响构建速度
 - 业务服务器跑 CI 影响线上服务稳定性
 ### 2.2 资源浪费
 - 新服务器 8核14G 跑 validate/lint 有点大材小用
 - 构建服务器 4核7G 跑 Docker 构建偏紧张
 ---
 ## 三、规划方案
 ### 3.1 分工原则
 | 服务器 | 角色 | 主要任务类型 | 标签策略 |
 |--------|------|-------------|----------|
 | **构建服务器** (114.55.236.178) | 构建专机 | Build Staging / Build Production / Docker 镜像构建 | 保留 `saas` + `host`，新增 `build-only` |
 | **新CI服务器** (116.62.226.203) | 代码检查专机 | Validate / Unit Tests / Integration Tests / Frontend Lint | 保留 `saas` + `host`，新增 `ci-check` |
 | **业务服务器** (47.98.113.167) | 部署专机 | Deploy Staging / Deploy Production / E2E Tests | 保留 `saas` + `host`，新增 `deploy-only` |
 ### 3.2 具体配置
 #### 构建服务器（4个 Runner）
 - **数量：** 3个（从4个缩减，释放资源给构建缓存）
 - **标签：** `saas`, `host`, `build-only`, `ubuntu-latest`
 - **负责 Job：**
  - `build-staging`
  - `build-production-runtime-images`
  - 其他需要 Docker buildx 的任务
 #### 新CI服务器（5个 Runner）
 - **数量：** 5个（保持不变）
 - **标签：** `saas`, `host`, `ci-check`, `ubuntu-latest`
 - **负责 Job：**
  - `validate`
  - `unit-tests`
  - `integration-tests`
  - `frontend-lint`
  - 安全扫描（gitleaks / pip-audit / vulture 等）
 #### 业务服务器（1-2个 Runner）
 - **数量：** 1-2个（逐步替换旧的3个）
 - **标签：** `saas`, `host`, `deploy-only`, `ubuntu-latest`
 - **负责 Job：**
  - `deploy-staging`
  - `deploy-production`
  - `staging-e2e` / `production-e2e`
  - `staging-api-tests`
 ---
 ## 四、实施步骤
 ### Phase 1: 标签打标（低风险，立即做）
 1. 新服务器 5 个 Runner 添加 `ci-check` 标签
 2. 构建服务器保留 3 个 Runner，添加 `build-only` 标签
 3. 业务服务器部署 1 个新 Runner，标签 `deploy-only`
 ### Phase 2: Job 路由调整（中风险，逐步来）
 1. validate / unit-tests / integration-tests / frontend-lint 改为 `runs-on: ci-check`
 2. build-staging / build-production 改为 `runs-on: build-only`
 3. deploy-* / e2e 改为 `runs-on: deploy-only`
 ### Phase 3: 旧 Runner 下线
 - 业务服务器旧的 3 个 Runner 确认无任务后下线
 - 构建服务器多余的 1 个 Runner 迁移到新服务器
 ---
 ## 五、并发配置优化建议
 ### 5.1 当前并发情况
 - 首发并行 Job：validate + unit-tests + frontend-lint（3个并行）
 - integration-tests 依赖 validate（串行，浪费资源）
 - 无 concurrency 限制，同一分支多次 push 会重复跑
 ### 5.2 优化建议
 **1. integration-tests 改为与 unit-tests 并行**
 ```yaml
 # 当前
 integration-tests:
  needs: validate  # 没必要等validate
 # 优化后
 integration-tests:
  needs: []  # 直接和unit-tests并行跑
 ```
 **2. 增加分支级 concurrency，取消重复构建**
 ```yaml
 concurrency:
  group: ${{ github.workflow }}-${{ github.ref }}
  cancel-in-progress: true
 ```
 同一 PR 多次 push 时，取消旧的构建，只跑最新的。
 **3. Build Staging 移出 PR 门禁**
 - 已在阶段二优化中完成（PR #245）
 - Build Staging 只在 develop/main 上异步构建
 ---
 ## 六、预期收益
 | 指标 | 当前 | 优化后 | 提升 |
 |------|------|--------|------|
 | PR CI 总时长 | ~8-12分钟 | ~4-6分钟 | ⏱️ 缩短 40-50% |
 | 构建速度 | 可能抢到慢机器 | 固定高配构建机 | 🚀 更稳定更快 |
 | 线上稳定性 | CI和业务抢资源 | 部署独立Runner | 🛡️ 隔离保障 |
 | Runner 利用率 | 随机分配 | 按任务类型调度 | 📈 更合理 |
@@ -990,14 +990,6 @@
          "type": "FLOAT",
          "unique": false
        },
        {
          "index": false,
          "name": "result_count",
          "nullable": false,
          "primary_key": false,
          "type": "INTEGER",
          "unique": false
        },
        {
          "index": false,
          "name": "config",
@@ -100,7 +100,6 @@ class SQLAlchemyEditPlanRepository:
            name=plan.name,
            status=plan.status,
            total_duration=plan.total_duration,
            result_count=plan.result_count,
            source_edit_plan_id=plan.source_edit_plan_id or None,
            project_id=plan.project_id or "",
            created_by_user_id=plan.created_by_user_id or "",
@@ -120,7 +119,6 @@ class SQLAlchemyEditPlanRepository:
        model.name = plan.name
        model.status = plan.status
        model.total_duration = plan.total_duration
        model.result_count = plan.result_count
        model.source_edit_plan_id = plan.source_edit_plan_id or None
        model.project_id = plan.project_id or ""
        model.created_by_user_id = plan.created_by_user_id or ""
@@ -154,7 +152,6 @@ class SQLAlchemyEditPlanRepository:
            name=model.name,
            status=EditPlanStatus(model.status) if model.status else EditPlanStatus.DRAFT,
            total_duration=model.total_duration or 0.0,
            result_count=int(model.result_count or 0),
            source_edit_plan_id=model.source_edit_plan_id or "",
            project_id=model.project_id or "",
            created_by_user_id=model.created_by_user_id or "",
@@ -1,10 +1,9 @@
 from datetime import datetime, timezone
 from typing import Any
 from sqlalchemy import JSON, Boolean, Column, DateTime, Float, Integer, String, Text, UniqueConstraint
 from sqlalchemy.orm import declarative_base
-Base: Any = declarative_base()
+Base = declarative_base()
 class UserModel(Base):
@@ -149,7 +148,6 @@ class EditPlanModel(Base):
    name = Column(String(200), nullable=False)
    status = Column(String(20), nullable=False, default="draft", index=True)
    total_duration = Column(Float, nullable=False, default=0.0)
    result_count = Column(Integer, nullable=False, default=0)
    config = Column(JSON, nullable=False, default=dict)
    source_edit_plan_id = Column(String(36), nullable=True, index=True)
    project_id = Column(String(36), nullable=False, default="", index=True)
@@ -20,7 +20,7 @@ class ListAssetLibrariesUseCase:
    def execute(self, project_id: str) -> list[AssetLibrary]:
        if not project_id.strip():
            raise ValueError("project_id 不能为空")
-        return self.asset_library_repository.find_by_project(project_id.strip())  # type: ignore[return-value]
+        return self.asset_library_repository.find_by_project(project_id.strip())
 class CreateAssetLibraryUseCase:
@@ -33,4 +33,4 @@ class CreateAssetLibraryUseCase:
            name=command.name,
            kind=command.kind,
        )
-        return self.asset_library_repository.create(library)  # type: ignore[return-value]
+        return self.asset_library_repository.create(library)
@@ -147,10 +147,10 @@ class JWTService:
                algorithms=[self.config.ALGORITHM],
            )
            return payload
-        except ExpiredSignatureError as _e:
+        except ExpiredSignatureError:
-            raise ExpiredSignatureError("Token has expired") from _e
+            raise ExpiredSignatureError("Token has expired")
        except InvalidTokenError as e:
-            raise InvalidTokenError(f"Invalid token: {str(e)}") from e
+            raise InvalidTokenError(f"Invalid token: {str(e)}")
    def verify_access_token(self, token: str) -> Dict[str, Any]:
        """
@@ -22,7 +22,7 @@ class SubmitClassificationJobUseCase:
            id=uuid4().hex,
            project_id=command.project_id,
            asset_id=command.asset_id,
-            status="pending",  # type: ignore[arg-type]
+            status="pending",
            classification="",
            confidence=0.0,
            error_message="",
@@ -15,7 +15,7 @@ from __future__ import annotations
 import logging
 import time
 from dataclasses import dataclass
-from typing import Any, Callable, Optional
+from typing import Any, Optional
 import httpx
@@ -102,7 +102,7 @@ class CosyVoiceService:
        model: str = "",
        clone_model: str = "",
        http_client: Optional[httpx.Client] = None,
-        audio_url_signer: Optional[Callable[[str], str]] = None,
+        audio_url_signer: Optional[callable] = None,
    ) -> None:
        """初始化 CosyVoice 服务.
@@ -656,8 +656,8 @@ class CosyVoiceService:
                        code = body.get("code", "")
                        message = body.get("message", "")
                        raise CosyVoiceError(f"CosyVoice API 参数错误: HTTP 400, " f"code={code}, message={message}")
-                    except ValueError as _e:
+                    except ValueError:
-                        raise CosyVoiceError(f"CosyVoice API 调用失败: HTTP 400, body={body_text}") from _e
+                        raise CosyVoiceError(f"CosyVoice API 调用失败: HTTP 400, body={body_text}")
                elif response.status_code >= 500:
                    # 服务端错误，可重试
                    last_error = CosyVoiceError(f"CosyVoice API 服务端错误: HTTP {response.status_code}")
@@ -40,7 +40,7 @@ class CreateGenerationTaskUseCase:
            asset_ids=command.asset_ids,
            title_ids=command.title_ids,
            voice_ids=command.voice_ids,
-            status="pending",  # type: ignore[arg-type]
+            status="pending",
            progress=0.0,
            result_count=0,
            error_message="",
@@ -78,16 +78,16 @@ class AudioMerger:
                run_ffmpeg(cmd, timeout=120)
            except CalledProcessError as e:
                logger.error(f"FFmpeg 合并失败: stderr={e.stderr}")
-                raise AudioMergeError(f"FFmpeg 合并失败: {str(e)[:500]}") from e
+                raise AudioMergeError(f"FFmpeg 合并失败: {str(e)[:500]}")
            with open(output_path, "rb") as f:
                return f.read()
-        except TimeoutExpired as _e:
+        except TimeoutExpired:
-            raise AudioMergeError("FFmpeg 合并超时（120 秒）") from _e
+            raise AudioMergeError("FFmpeg 合并超时（120 秒）")
        except AudioMergeError:
            raise
        except Exception as e:
-            raise AudioMergeError(f"音频合并失败: {e}") from e
+            raise AudioMergeError(f"音频合并失败: {e}")
        finally:
            shutil.rmtree(temp_dir, ignore_errors=True)
@@ -24,7 +24,6 @@ from packages.application.voice_clone.use_cases import (
 )
 from packages.domain.voice_clone_profile import VoiceCloneProfile
 from packages.ports.voice_clone_profile_repository import VoiceCloneProfileRepository
 from packages.shared.url_security import UrlSecurityError, validate_url_safety
 logger = logging.getLogger(__name__)
@@ -104,15 +103,6 @@ class VoiceCloneWorkflowService:
        # 2. 提交 CosyVoice 克隆任务（仅有音频 URL 时才标记 processing）
        if source_audio_url:
            # SSRF 防护：校验音频 URL 安全性
            try:
                source_audio_url = validate_url_safety(source_audio_url, purpose="download")
            except UrlSecurityError as e:
                profile.mark_failed(f"音频URL安全校验失败: {e}")
                profile = self.repository.update(profile)
                logger.warning(f"音色克隆音频URL安全校验失败: profile_id={profile.id}, error={e}")
                return profile
            # 标记为 processing
            profile.mark_processing()
            profile = self.repository.update(profile)
@@ -253,15 +243,6 @@ class VoiceCloneWorkflowService:
        # 3. 重新提交 CosyVoice
        if profile.source_audio_url:
            # SSRF 防护：重新校验音频 URL 安全性
            try:
                validate_url_safety(profile.source_audio_url, purpose="download")
            except UrlSecurityError as e:
                profile.mark_failed(f"音频URL安全校验失败: {e}")
                profile = self.repository.update(profile)
                logger.warning(f"音色克隆重试音频URL安全校验失败: profile_id={clone_id}, error={e}")
                return profile
            try:
                submit_result = self.cosyvoice_service.submit_clone_task(
                    audio_url=profile.source_audio_url,
@@ -42,7 +42,6 @@ class EditPlan:
    name: str
    status: EditPlanStatus = EditPlanStatus.DRAFT
    total_duration: float = 0.0
    result_count: int = 0
    source_edit_plan_id: str = ""
    project_id: str = ""
    created_by_user_id: str = ""
@@ -58,7 +57,6 @@ class EditPlan:
        *,
        config: dict[str, Any] | None = None,
        total_duration: float = 0.0,
        result_count: int = 0,
        source_edit_plan_id: str = "",
        project_id: str = "",
        created_by_user_id: str = "",
@@ -75,7 +73,6 @@ class EditPlan:
            name=clean_name,
            status=EditPlanStatus.DRAFT,
            total_duration=total_duration,
            result_count=result_count,
            source_edit_plan_id=source_edit_plan_id.strip(),
            project_id=project_id.strip(),
            created_by_user_id=created_by_user_id.strip(),
@@ -110,13 +107,6 @@ class EditPlan:
        self.status = EditPlanStatus.FAILED
        self.updated_at = datetime.now(timezone.utc)
    def resume_editing(self) -> None:
        """重新进入编辑状态（完成/失败后重新编辑）"""
        if self.status not in (EditPlanStatus.COMPLETED, EditPlanStatus.FAILED):
            raise ValueError(f"只有 completed/failed 状态的计划可以重新编辑，当前状态: {self.status}")
        self.status = EditPlanStatus.EDITING
        self.updated_at = datetime.now(timezone.utc)
    def reset_to_draft(self) -> None:
        """重置为草稿状态（仅从 failed 状态可重置）"""
        if self.status != EditPlanStatus.FAILED:
@@ -170,8 +170,8 @@ class GenerationTask:
        if isinstance(new_status, str):
            try:
                new_status = GenerationTaskStatus(new_status)
-            except ValueError as _e:
+            except ValueError:
-                raise ValueError(f"无效状态: {new_status}") from _e
+                raise ValueError(f"无效状态: {new_status}")
        allowed = _VALID_TRANSITIONS.get(self.status, set())
        if new_status not in allowed:
@@ -147,8 +147,8 @@ class Job:
        if isinstance(job_type, str):
            try:
                job_type = JobType(job_type)
-            except ValueError as _e:
+            except ValueError:
-                raise ValueError(f"不支持的任务类型: {job_type}") from _e
+                raise ValueError(f"不支持的任务类型: {job_type}")
        return cls(
            id=uuid4().hex,
@@ -182,8 +182,8 @@ class Job:
        if isinstance(new_status, str):
            try:
                new_status = JobStatus(new_status)
-            except ValueError as _e:
+            except ValueError:
-                raise ValueError(f"无效状态: {new_status}") from _e
+                raise ValueError(f"无效状态: {new_status}")
        allowed = _VALID_TRANSITIONS.get(self.status, set())
        if new_status not in allowed:
@@ -193,8 +193,8 @@ class TTSJob:
        if isinstance(new_status, str):
            try:
                new_status = TTSJobStatus(new_status)
-            except ValueError as _e:
+            except ValueError:
-                raise ValueError(f"无效状态: {new_status}") from _e
+                raise ValueError(f"无效状态: {new_status}")
        allowed = _VALID_TRANSITIONS.get(self.status, set())
        if new_status not in allowed:
@@ -177,8 +177,8 @@ class VoiceCloneProfile:
        if isinstance(new_status, str):
            try:
                new_status = VoiceCloneStatus(new_status)
-            except ValueError as _e:
+            except ValueError:
-                raise ValueError(f"无效状态: {new_status}") from _e
+                raise ValueError(f"无效状态: {new_status}")
        allowed = _VALID_TRANSITIONS.get(self.status, set())
        if new_status not in allowed:
@@ -114,7 +114,7 @@ class SharedStorageService:
                self.bucket.put_object(storage_key, file_or_path, headers={"Content-Type": content_type})
            return f"{self.public_url}/{storage_key}"
        except Exception as e:
-            raise Exception(f"Failed to upload file to OSS: {e}") from e
+            raise Exception(f"Failed to upload file to OSS: {e}")
    def get_url(self, storage_key: str) -> str:
        """Get public URL for a file."""
@@ -129,7 +129,7 @@ class SharedStorageService:
            os.makedirs(os.path.dirname(local_path), exist_ok=True)
            self.bucket.get_object_to_file(storage_key, local_path)
        except Exception as e:
-            raise Exception(f"Failed to download file from OSS: {e}") from e
+            raise Exception(f"Failed to download file from OSS: {e}")
    def get_download_url(self, storage_key_or_url: str, expires_seconds: int = 3600) -> str:
        """Get signed download URL."""
@@ -63,23 +63,21 @@ exclude = [
  ".next",
  "dist",
  "build",
  "hostexecutor",
 ]
 [tool.ruff.lint]
-# 正式替换 flake8：规则集与原 flake8 完全对齐
+# 当前阶段：摸底模式，规则集与原flake8对齐
 # 后续迭代计划：
-#   Phase 2: 加入 B (flake8-bugbear)，修完后升级为阻断级
+#   Phase 1: 修完 bugbear 后正式替换 flake8
-#   Phase 3: 启用 UP(pyupgrade) + SIM(simplify)
+#   Phase 2: 启用 UP(pyupgrade) + SIM(simplify)
-#   Phase 4: 启用 RET(return) + ARG(unused-args)
+#   Phase 3: 启用 RET(return) + ARG(unused-args)
 select = [
-  "E",   # pycodestyle errors（同 flake8）
+  "E",   # pycodestyle errors（同flake8）
-  "F",   # pyflakes（同 flake8）
+  "F",   # pyflakes（同flake8）
-  "W",   # pycodestyle warnings（同 flake8）
+  "W",   # pycodestyle warnings（同flake8）
-  "B",   # flake8-bugbear（P0-5 Step 2 已完成修复）
+  "B",   # flake8-bugbear（新增，摸底用）
 ]
-# 与原 setup.cfg + .flake8 的 flake8 配置完全对齐
+# 与原 setup.cfg flake8 配置对齐，确保不新增阻断
 # 注意：W503 在 ruff≥0.14 中已被移除（行为变默认），故不列入
 ignore = [
  "E203",
  "E501",  # line-too-long（black管）
@@ -88,19 +86,16 @@ ignore = [
  "E722",  # bare-except
  "W291",
  "W293",
  "B008",  # function-call-in-default-argument（FastAPI 依赖注入模式，大量使用）
  "F401",  # unused-import
  "F403",
  "F405",
  "F841",  # unused-variable
  "B008",  # do-not-perform-callback-from-arg（fastapi依赖注入）
 ]
 [tool.ruff.lint.per-file-ignores]
 "__init__.py" = ["F401", "F403", "F405"]
-"tests/*" = ["E402", "F401", "F821", "F841"]
+"tests/*" = ["E402", "F401", "F841"]
-"packages/ports/*" = ["E301"]  # E704 在 ruff≥0.14 已移除
+"packages/ports/*" = ["E301"]
 "apps/api/app/api/routes/auth.py" = ["ALL"]
 "apps/api/app/api/routes/workspaces.py" = ["ALL"]
 "apps/api/app/middleware/auth.py" = ["ALL"]
 "apps/*/migrations/*" = ["ALL"]
 "alembic/*" = ["ALL"]
@@ -3,7 +3,7 @@
 # 代码质量
 black==26.5.1
 isort==8.0.1
-ruff==0.14.0
+flake8==7.3.0
 bandit==1.9.4
 # 测试
@@ -1,108 +1,43 @@
 #!/bin/bash
 # 自动合并通过 CI 检查的 PR
 # 用法: ./scripts/auto_merge_prs.sh [target_branch]
 #
 # 合并前必须验证的 CI 检查项：
 #   - CI/CD Pipeline / Validate Code Quality And Tests (push)
 #   - CI/CD Pipeline / Frontend Lint (push)
 # 只有两个检查项均为 success 状态才允许合并
-GITEA_API="${GITEA_API_URL:-https://git.xiaoxiajianji.com/api/v1}"
+GITEA_API="https://git.xiaoxiajianji.com/api/v1"
 TOKEN="${GITEA_API_TOKEN:?Please set GITEA_API_TOKEN environment variable}"
 REPO="xiaoxia/xiaoxia-saas"
 TARGET_BRANCH="${1:-develop}"
 # 必需的 CI 检查项（context 名称前缀匹配，避免 pipeline 名称变化导致匹配失败）
 REQUIRED_CHECKS=(
  "Validate Code Quality And Tests"
  "Frontend Lint"
 )
 echo "=== Checking open PRs targeting $TARGET_BRANCH ==="
 # 获取所有 open PR
 PRS=$(curl -s -H "Authorization: token $TOKEN" \
-  "$GITEA_API/repos/$REPO/pulls?state=open&sort=updated&direction=desc" | python3 -c "
+  "$GITEA_API/repos/$REPO/pulls?state=open&labels=0" | python3 -c "
 import json, sys
 data = json.load(sys.stdin)
 for pr in data:
    if pr.get('base', {}).get('ref') == '$TARGET_BRANCH':
-        head_sha = pr.get('head', {}).get('sha', '')
+        if pr.get('mergeable', False):
-        print(f\"{pr['number']}|{pr['title']}|{head_sha}\")
+            print(f\"{pr['number']}|{pr['title']}|{pr.get('mergeable', 'unknown')}\")
 ")
 if [ -z "$PRS" ]; then
-    echo "No open PRs found for $TARGET_BRANCH"
+    echo "No mergeable PRs found for $TARGET_BRANCH"
    exit 0
 fi
-merge_count=0
+echo "$PRS" | while IFS='|' read -r number title mergeable; do
-skip_count=0
+    echo "Merging PR #$number: $title"
 echo "$PRS" | while IFS='|' read -r number title head_sha; do
    echo ""
    echo "--- PR #$number: $title ---"
    echo "  Head SHA: $head_sha"
    # 获取该 commit 的 combined CI 状态
    STATUS_JSON=$(curl -s -H "Authorization: token $TOKEN" \
      "$GITEA_API/repos/$REPO/commits/$head_sha/status")
    # 检查每个必需的 CI 项是否通过
    all_passed=true
    failed_checks=""
    for check_pattern in "${REQUIRED_CHECKS[@]}"; do
        state=$(echo "$STATUS_JSON" | python3 -c "
 import json, sys
 d = json.load(sys.stdin)
 pattern = '$check_pattern'
 # 在 statuses 中找到匹配的最新状态
 target = None
 for s in d.get('statuses', []):
    if pattern in s.get('context', ''):
        target = s
        break  # status 接口返回的是每个 context 的最新状态，取第一个匹配即可
 if target:
    print(target.get('state', 'unknown'))
 else:
    print('not_found')
 ")
        if [ "$state" = "success" ]; then
            echo "  ✅ $check_pattern: $state"
        else
            echo "  ❌ $check_pattern: $state"
            all_passed=false
            failed_checks="$failed_checks $check_pattern($state)"
        fi
    done
    if [ "$all_passed" != "true" ]; then
        echo "  ⏭️  Skipping - CI not passed:$failed_checks"
        skip_count=$((skip_count + 1))
        continue
    fi
    # CI 全部通过，执行合并
    echo "  🚀 All CI checks passed, merging..."
    RESULT=$(curl -s -X POST \
      -H "Authorization: token $TOKEN" \
      -H "Content-Type: application/json" \
      "$GITEA_API/repos/$REPO/pulls/$number/merge" \
-      -d '{"Do": "merge"}')
+      -d '{\"merge_method\": \"merge\"}')
-
+    
-    if echo "$RESULT" | python3 -c "import json,sys; d=json.load(sys.stdin); sys.exit(0 if d.get('merged', False) or 'id' in d else 1)" 2>/dev/null; then
+    if echo "$RESULT" | python3 -c "import json,sys; d=json.load(sys.stdin); sys.exit(0 if 'id' in d else 1)"; then
        echo "  ✅ PR #$number merged successfully"
        merge_count=$((merge_count + 1))
    else
-        echo "  ❌ PR #$number merge failed"
+        echo "  ❌ PR #$number failed: $RESULT"
        # 提取错误信息
        err_msg=$(echo "$RESULT" | python3 -c "import json,sys; d=json.load(sys.stdin); print(d.get('message', str(d)[:200]))" 2>/dev/null)
        echo "     Error: $err_msg"
    fi
 done
 echo ""
 echo "=== Done ==="
 echo "Merged: $merge_count | Skipped: $skip_count"
@@ -60,9 +60,6 @@ fi
 # 默认只读不写，防止 feature 分支污染主缓存
 # 只有 develop/main 分支才写回缓存
 BRANCH_NAME="${GITHUB_REF_NAME:-${CI_COMMIT_BRANCH:-unknown}}"
 # 清理本地旧镜像
 docker rmi -f "$API_IMAGE" "$API_LATEST" 2>/dev/null || true
 if [ "$USE_CACHE" -eq 1 ]; then
  docker buildx build \
    --build-arg APP_VERSION="$VERSION" \
@@ -92,9 +89,6 @@ build_with_cache() {
    echo "  cache: read-only from ${CACHE_REGISTRY}/${IMG_NAME}-cache:${CACHE_TAG_PRIMARY}"
  fi
  # 清理本地旧镜像，避免 buildx --load 报 already exists 错误
  docker rmi -f "$IMG_NAME:$VERSION" 2>/dev/null || true
  if [ "$USE_CACHE" -eq 1 ]; then
    if [ -n "$CACHE_TO" ]; then
      docker buildx build \
@@ -125,9 +119,6 @@ build_with_cache "api" "infra/docker/api.Dockerfile" \
 docker tag "$API_IMAGE" "$API_LATEST"
 echo "=== Building Worker image ==="
 # 清理本地旧镜像
 docker rmi -f "$WORKER_IMAGE" "$WORKER_LATEST" 2>/dev/null || true
 if [ "$USE_CACHE" -eq 1 ]; then
  docker buildx build \
    --build-arg APP_VERSION="$VERSION" \
@@ -157,9 +148,6 @@ docker run --rm \
 test -f apps/web/dist/index.html
 # 清理本地旧镜像
 docker rmi -f "$WEB_IMAGE" 2>/dev/null || true
 if [ "$USE_CACHE" -eq 1 ]; then
  docker buildx build \
    --cache-from "type=registry,ref=${CACHE_REGISTRY}/web-cache:${CACHE_TAG_PRIMARY},ignore-error=true" \
@@ -30,14 +30,10 @@
 from __future__ import annotations
 import argparse
 import json
 import os
 import re
 import subprocess
 import sys
 import time
 import urllib.error
 import urllib.request
 from pathlib import Path
 from typing import List, Tuple
@@ -79,15 +75,6 @@ SAFE_PATTERNS = [
 ]
 def _get_env(*names: str, default: str = "") -> str:
    """按优先级尝试多个环境变量名，返回第一个非空值。"""
    for name in names:
        val = os.environ.get(name, "")
        if val:
            return val
    return default
 def extract_upgrade_content(content: str) -> str:
    """
    从迁移文件中提取 upgrade 函数的内容。
@@ -110,149 +97,34 @@ def extract_upgrade_content(content: str) -> str:
    return content[upgrade_start:upgrade_end]
 def _api_get_with_retry(url: str, token: str, max_retries: int = 3) -> dict | list:
    """
    带重试的 API 调用。
    指数退避：1s, 2s, 4s
    """
    last_error = None
    for attempt in range(max_retries):
        try:
            req = urllib.request.Request(url, headers={"Authorization": f"token {token}"})
            with urllib.request.urlopen(req, timeout=15) as resp:
                return json.loads(resp.read().decode())
        except urllib.error.HTTPError as e:
            # 404 说明目录不存在或分支不存在，直接抛
            if e.code == 404:
                raise
            last_error = e
            if attempt < max_retries - 1:
                wait = 2**attempt
                print(f"   （API 请求失败，{wait}s 后重试 {attempt + 1}/{max_retries}：{e}）")
                time.sleep(wait)
        except Exception as e:
            last_error = e
            if attempt < max_retries - 1:
                wait = 2**attempt
                print(f"   （API 请求失败，{wait}s 后重试 {attempt + 1}/{max_retries}：{e}）")
                time.sleep(wait)
    raise last_error  # type: ignore[misc]
 def get_new_migrations_via_api(diff_target: str) -> List[Path] | None:
    """
    通过 Gitea/GitHub Contents API 对比目标分支，找出 alembic/versions/ 下新增的迁移文件。
    返回 None 表示 API 方式不可用，调用方应尝试其他方式。
    """
    # 同时支持 Gitea 和 GitHub 的环境变量命名
    api_url = _get_env("GITEA_API_URL", "GITHUB_API_URL", "CI_API_V4_URL")
    repo = _get_env("GITEA_REPOSITORY", "GITHUB_REPOSITORY", "CI_PROJECT_PATH")
    token = _get_env("GITEA_TOKEN", "GITHUB_TOKEN", "CI_JOB_TOKEN")
    branch = diff_target.replace("origin/", "")
    if not api_url or not repo or not token:
        print(
            f"   （API 环境变量不完整：api_url={'✓' if api_url else '✗'} repo={'✓' if repo else '✗'} token={'✓' if token else '✗'}）"
        )
        return None
    try:
        url = f"{api_url}/repos/{repo}/contents/alembic/versions?ref={branch}"
        data = _api_get_with_retry(url, token)
        if isinstance(data, dict):
            # Gitea 目录不存在时返回 404，不会到这里；如果返回 dict 可能是错误信息
            print(f"   （API 返回异常：{str(data)[:100]}）")
            return None
        remote_files = {item["name"] for item in data if item["name"].endswith(".py")}
        local_files = {f.name for f in ALEMBIC_VERSIONS_DIR.glob("*.py")}
        new_file_names = sorted(local_files - remote_files)
        if new_file_names:
            result = [ALEMBIC_VERSIONS_DIR / f for f in new_file_names]
            print(f"   （API 对比 {branch} 分支，发现 {len(result)} 个新增迁移）")
            return result
        else:
            print(f"   （API 对比 {branch} 分支，无新增迁移）")
            return []
    except Exception as e:
        print(f"   （API 获取迁移列表失败：{e}）")
        return None
 def get_new_migrations_via_git(diff_target: str) -> List[Path] | None:
    """
    Fallback：通过本地 git diff 找出新增的迁移文件。
    CI 环境中 git 可用时作为 API 失败后的兜底方案。
    """
    try:
        # 确保目标分支存在
        subprocess.run(
            ["git", "fetch", "origin", diff_target.replace("origin/", ""), "--depth=50"],
            capture_output=True,
            cwd=str(REPO_ROOT),
            timeout=30,
        )
        # 优先使用三点diff（找合并基线），失败时回退到两点diff（兼容tar.gz checkout + git init的CI环境）
        diff_args = ["git", "diff", "--name-only", "--diff-filter=A", f"{diff_target}...HEAD"]
        result = subprocess.run(
            diff_args,
            capture_output=True,
            text=True,
            cwd=str(REPO_ROOT),
            timeout=10,
        )
        if result.returncode != 0:
            # fallback: 两点diff（无需共同祖先）
            diff_args_2 = ["git", "diff", "--name-only", "--diff-filter=A", diff_target, "HEAD"]
            result = subprocess.run(
                diff_args_2,
                capture_output=True,
                text=True,
                cwd=str(REPO_ROOT),
                timeout=10,
            )
            if result.returncode != 0:
                print(f"   （git diff 失败：{result.stderr.strip()}）")
                return None
        new_migrations = []
        for line in result.stdout.strip().split("\n"):
            line = line.strip()
            if line.startswith("alembic/versions/") and line.endswith(".py"):
                new_migrations.append(REPO_ROOT / line)
        new_migrations.sort()
        print(f"   （git diff 对比 {diff_target}，发现 {len(new_migrations)} 个新增迁移）")
        return new_migrations
    except Exception as e:
        print(f"   （git diff 方式失败：{e}）")
        return None
 def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
    """
-    找出相对目标分支新增的迁移文件，按优先级尝试多种方式：
+    通过 git diff 对比目标分支/commit，找出 alembic/versions/ 下新增的迁移文件。
-    1. Gitea/GitHub Contents API（最可靠，不受本地 checkout 深度影响）
+    只包含新增文件（A状态），不包含修改或删除的文件。
    2. git diff（API 失败时的兜底）
    3. 全量扫描（以上都失败时的最后兜底，会输出警告）
    """
-    print("🔍 尝试通过 API 获取新增迁移列表...")
+    try:
-    result = get_new_migrations_via_api(diff_target)
+        result = subprocess.run(
-    if result is not None:
+            [
-        return result
+                "git",
-
+                "diff",
-    print("🔍 API 不可用，尝试 git diff 方式...")
+                "--name-only",
-    result = get_new_migrations_via_git(diff_target)
+                "--diff-filter=A",
-    if result is not None:
+                diff_target,
-        return result
+                "HEAD",
-
+                "--",
-    print("⚠️  所有增量方式均失败，降级为检查所有迁移文件")
+                "alembic/versions/",
-    print("   这可能导致历史迁移中的破坏性操作被误报")
+            ],
-    print("   建议检查 CI 环境变量配置（GITHUB_API_URL / GITHUB_REPOSITORY / GITHUB_TOKEN）")
+            cwd=str(REPO_ROOT),
-    return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
+            capture_output=True,
            text=True,
            check=True,
        )
        files = [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]
        return [REPO_ROOT / f for f in files]
    except subprocess.CalledProcessError as e:
        print(f"⚠️  git diff 失败（{diff_target}）：{e.stderr.strip()}")
        print("   降级为检查所有迁移文件")
        return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
 def find_new_migrations(since_revision: str | None = None, diff_against: str | None = None) -> List[Path]:
@@ -1,91 +0,0 @@
 #!/bin/bash
 # 通用Docker镜像构建+推送脚本（local cache为主 + registry cache兜底）
 # M-2优化：解决registry缓存导入慢(247s)和推送不稳定问题
 # 用法: docker_build_push.sh <Dockerfile> <image_tag> <cache_ref> [build_arg...]
 set -eu
 DOCKERFILE="$1"
 IMAGE_TAG="$2"
 CACHE_REF="$3"
 shift 3
 BUILD_ARGS=""
 for arg in "$@"; do
  BUILD_ARGS="$BUILD_ARGS --build-arg $arg"
 done
 if ! docker buildx inspect ci-builder > /dev/null 2>&1; then
  docker buildx create --use --name ci-builder --driver docker-container
  echo "Created ci-builder"
 else
  docker buildx use ci-builder
  echo "Using existing ci-builder"
 fi
 docker buildx inspect --bootstrap
 # 从cache_ref中提取缓存名称（如 api-cache:develop -> api-cache-develop）
 CACHE_NAME=$(echo "$CACHE_REF" | tr '/' '_' | tr ':' '-')
 LOCAL_CACHE_DIR="/tmp/buildx-cache/${CACHE_NAME}"
 mkdir -p "$LOCAL_CACHE_DIR"
 # 缓存源：local优先，registry兜底
 CACHE_FROM_LOCAL="type=local,src=${LOCAL_CACHE_DIR}"
 CACHE_FROM_REGISTRY="type=registry,ref=${CACHE_REF},ignore-error=true"
 # 本地缓存目标（必选，mode=max最大化命中率）
 CACHE_TO_LOCAL="type=local,dest=${LOCAL_CACHE_DIR},mode=max"
 echo "=== Step 1: Build & push image (local cache read-write + registry read) ==="
 echo "Local cache: ${LOCAL_CACHE_DIR}"
 echo "Registry cache: ${CACHE_REF}"
 echo ""
 docker buildx build \
  $BUILD_ARGS \
  --cache-from "${CACHE_FROM_LOCAL}" \
  --cache-from "${CACHE_FROM_REGISTRY}" \
  --cache-to "${CACHE_TO_LOCAL}" \
  -f "${DOCKERFILE}" \
  -t "${IMAGE_TAG}" \
  --push \
  .
 echo ""
 echo "Image pushed: ${IMAGE_TAG}"
 echo "Local cache updated"
 # DISABLED: registry cache too slow echo ""
 # DISABLED: registry cache too slow echo "=== Step 2: Sync registry cache (best effort, retries 3x) ==="
 # DISABLED: registry cache too slow CACHE_TO_REGISTRY="type=registry,ref=${CACHE_REF},mode=max,compression=zstd"
 # DISABLED: registry cache too slow 
 # DISABLED: registry cache too slow MAX_RETRIES=3
 # DISABLED: registry cache too slow SUCCESS=0
 # DISABLED: registry cache too slow for attempt in $(seq 1 $MAX_RETRIES); do
 # DISABLED: registry cache too slow   echo "Registry cache sync attempt $attempt/$MAX_RETRIES"
 # DISABLED: registry cache too slow   if docker buildx build \
 # DISABLED: registry cache too slow     $BUILD_ARGS \
 # DISABLED: registry cache too slow     --cache-from "${CACHE_FROM_LOCAL}" \
 # DISABLED: registry cache too slow     --cache-to "${CACHE_TO_REGISTRY}" \
 # DISABLED: registry cache too slow     -f "${DOCKERFILE}" \
 # DISABLED: registry cache too slow     -t "${IMAGE_TAG}" \
 # DISABLED: registry cache too slow     --push \
 # DISABLED: registry cache too slow     .; then
 # DISABLED: registry cache too slow     echo "Registry cache synced (attempt $attempt)"
 # DISABLED: registry cache too slow     SUCCESS=1
 # DISABLED: registry cache too slow     break
 # DISABLED: registry cache too slow   else
 # DISABLED: registry cache too slow     echo "Registry cache sync failed (attempt $attempt)"
 # DISABLED: registry cache too slow     if [ $attempt -lt $MAX_RETRIES ]; then
 # DISABLED: registry cache too slow       WAIT=$((attempt * 5))
 # DISABLED: registry cache too slow       echo "Retrying in ${WAIT}s..."
 # DISABLED: registry cache too slow       sleep $WAIT
 # DISABLED: registry cache too slow     fi
 # DISABLED: registry cache too slow   fi
 # DISABLED: registry cache too slow done
 # DISABLED: registry cache too slow 
 # DISABLED: registry cache too slow if [ $SUCCESS -eq 0 ]; then
 # DISABLED: registry cache too slow   echo "WARNING: Registry cache sync failed after $MAX_RETRIES attempts (non-fatal, local cache still works)"
 # DISABLED: registry cache too slow fi
 echo ""
 echo "Build completed: ${IMAGE_TAG}"
@@ -1,48 +0,0 @@
 #!/bin/bash
 # mypyå¢žé‡�æ‰«æ��è„šæœ¬ - CIä¸è°ƒç”¨
 # çŽ¯å¢ƒå�˜é‡�: SCAN_MODE, CHANGED_PY_FILES
 set -e
 echo "=== Installing mypy ==="
 python3 -m pip install -q mypy
 mypy --version
 echo ""
 echo "=== Running mypy type check (hard gate mode) ==="
 echo "å‘Šè¦æ¨¡å¼�ï¼Œä¸Íé˜»æ–CI"
 echo ""
 MYPY_COMMON_ARGS="--ignore-missing-imports --no-site-packages --no-strict-optional --explicit-package-bases --exclude tests/|test_|migrations/|alembic/ --no-error-summary --incremental --cache-dir .mypy_cache"
 EXIT_CODE=0
 if [ "$SCAN_MODE" = "incremental" ] && [ -n "$CHANGED_PY_FILES" ]; then
  echo "=== Incremental mypy scan (PR mode) ==="
  echo "Changed files: $(echo $CHANGED_PY_FILES | wc -w) files"
  MYPY_FILES=""
  for f in $CHANGED_PY_FILES; do
    case "$f" in
      apps/*|packages/*)
        MYPY_FILES="$MYPY_FILES $f"
        ;;
    esac
  done
  if [ -n "$MYPY_FILES" ]; then
    echo "Checking: $MYPY_FILES"
    mypy $MYPY_FILES $MYPY_COMMON_ARGS 2>&1 | head -80 || EXIT_CODE=$?
  else
    echo "No mypy-checkable files changed, skipping"
  fi
 else
  echo "=== Full mypy scan ==="
  mypy apps/api/app packages $MYPY_COMMON_ARGS 2>&1 | head -60 || EXIT_CODE=$?
 fi
 echo ""
 if [ "$EXIT_CODE" != "0" ]; then
  echo "mypy å�‘çŽ°ç±»åž‹é—®é¢˜ï¼ˆå‘Šè¦æ¨¡å¼�ï¼Œä¸Íé˜»æ–ï¼‰"
  echo "å»ºè®®å�Žç»é€�æ¥ä¿®å¤�"
 else
  echo "mypy ç±»åž‹æ£€æŸ¥é€šè¿‡"
 fi
@@ -1,627 +0,0 @@
 #!/usr/bin/env python3
 """
 CI Code Review Script
 - 从 Gitea 获取 PR diff
 - 调用 LLM 进行代码审查
 - 将审查结果写回 PR 评论
 """
 import argparse
 import json
 import logging
 import os
 import sys
 from typing import Optional, Tuple
 import requests
 # ============== 日志配置 ==============
 logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s] [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
 )
 logger = logging.getLogger("ci_code_review")
 # ============== 常量配置 ==============
 # diff 最大字符数（超过则截断）
 MAX_DIFF_CHARS = int(os.getenv("MAX_DIFF_CHARS", "30000"))
 # LLM 调用超时时间（秒）
 LLM_TIMEOUT = int(os.getenv("LLM_TIMEOUT", "120"))
 # Gitea API 超时时间（秒）
 GITEA_TIMEOUT = int(os.getenv("GITEA_TIMEOUT", "30"))
 # 最大重试次数
 MAX_RETRIES = int(os.getenv("MAX_RETRIES", "2"))
 # LLM 提供商: openai (OpenAI兼容) / coze (扣子原生Bot API)
 LLM_PROVIDER = os.getenv("LLM_PROVIDER", "coze").lower()
 # ============== 工具函数 ==============
 def truncate_diff(diff_text: str, max_chars: int) -> Tuple[str, bool]:
    """
    截断过大的 diff 内容，避免超出 LLM 上下文限制。
    优先保留文件头和前面的变更，末尾加提示。
    """
    if len(diff_text) <= max_chars:
        return diff_text, False
    # 找到一个合适的截断位置（尽量在文件边界）
    truncated = diff_text[:max_chars]
    # 尝试在最后一个 "diff --git" 处截断，避免截断到一半
    last_file_boundary = truncated.rfind("\ndiff --git ")
    if last_file_boundary > max_chars // 2:
        truncated = truncated[:last_file_boundary]
    truncated += (
        f"\n\n... [DIFF TRUNCATED] 原始 diff 共 {len(diff_text)} 字符，"
        f"已截断至 {len(truncated)} 字符，仅审查前半部分。\n"
    )
    return truncated, True
 def get_env_or_fail(name: str) -> str:
    """从环境变量获取值，不存在则报错退出。"""
    value = os.getenv(name)
    if not value:
        logger.error(f"环境变量 {name} 未设置")
        sys.exit(1)
    return value
 # ============== Gitea API 相关 ==============
 class GiteaClient:
    """Gitea API 客户端"""
    def __init__(self, base_url: str, token: str, repo: str):
        # 确保 base_url 以 / 结尾
        self.base_url = base_url.rstrip("/") + "/"
        self.token = token
        self.repo = repo  # 格式: owner/repo
        self.session = requests.Session()
        self.session.headers.update(
            {
                "Authorization": f"token {token}",
                "Accept": "application/json",
                "Content-Type": "application/json",
            }
        )
    def _api_url(self, path: str) -> str:
        """拼接 API 路径"""
        return f"{self.base_url}api/v1/repos/{self.repo}/{path.lstrip('/')}"
    def get_pr_diff(self, pr_number: int) -> str:
        """
        获取 PR 的 diff 内容。
        Gitea API: GET /repos/{owner}/{repo}/pulls/{index}.diff
        """
        url = self._api_url(f"pulls/{pr_number}.diff")
        logger.info(f"获取 PR #{pr_number} diff: {url}")
        resp = self.session.get(
            url,
            timeout=GITEA_TIMEOUT,
            headers={
                "Accept": "text/plain",
            },
        )
        if resp.status_code != 200:
            logger.error(f"获取 diff 失败: HTTP {resp.status_code} - {resp.text[:200]}")
            raise RuntimeError(f"Failed to get PR diff: HTTP {resp.status_code}")
        diff_text = resp.text
        logger.info(f"获取到 diff，共 {len(diff_text)} 字符")
        return diff_text
    def get_pr_files(self, pr_number: int) -> list:
        """
        获取 PR 修改的文件列表。
        Gitea API: GET /repos/{owner}/{repo}/pulls/{index}/files
        """
        url = self._api_url(f"pulls/{pr_number}/files")
        logger.info(f"获取 PR #{pr_number} 文件列表")
        resp = self.session.get(url, timeout=GITEA_TIMEOUT)
        if resp.status_code != 200:
            logger.warning(f"获取文件列表失败: HTTP {resp.status_code}")
            return []
        files = resp.json()
        logger.info(f"PR 修改了 {len(files)} 个文件")
        return files
    def post_pr_comment(self, pr_number: int, body: str) -> bool:
        """
        在 PR 上发布评论。
        Gitea API: POST /repos/{owner}/{repo}/issues/{index}/comments
        （Gitea 中 PR 评论走 issues 接口）
        """
        url = self._api_url(f"issues/{pr_number}/comments")
        logger.info(f"发布 PR 评论: {url}")
        payload = {"body": body}
        resp = self.session.post(
            url,
            data=json.dumps(payload),
            timeout=GITEA_TIMEOUT,
        )
        if resp.status_code not in (200, 201):
            logger.error(f"发布评论失败: HTTP {resp.status_code} - {resp.text[:200]}")
            return False
        logger.info(f"评论发布成功，评论 ID: {resp.json().get('id', 'unknown')}")
        return True
    def get_existing_review_comments(self, pr_number: int, marker: str) -> list:
        """
        获取 PR 上已有的 AI 审查评论 ID 列表（带标识 marker）。
        """
        url = self._api_url(f"issues/{pr_number}/comments")
        resp = self.session.get(url, timeout=GITEA_TIMEOUT)
        if resp.status_code != 200:
            logger.warning(f"获取评论列表失败: HTTP {resp.status_code}")
            return []
        comments = resp.json()
        review_comment_ids = []
        for c in comments:
            body = c.get("body", "")
            if marker in body:
                review_comment_ids.append(c.get("id"))
        logger.info(f"找到 {len(review_comment_ids)} 条旧的 AI 审查评论")
        return review_comment_ids
    def delete_pr_comment(self, pr_number: int, comment_id: int) -> bool:
        """
        删除 PR 上的指定评论。
        """
        url = self._api_url(f"issues/comments/{comment_id}")
        resp = self.session.delete(url, timeout=GITEA_TIMEOUT)
        if resp.status_code not in (200, 204):
            logger.warning(f"删除评论 {comment_id} 失败: HTTP {resp.status_code}")
            return False
        return True
 def call_llm_openai(
    prompt: str,
    llm_base_url: str,
    llm_api_key: str,
    llm_model: str,
 ) -> Optional[str]:
    """OpenAI 兼容模式调用"""
    base_url = llm_base_url.rstrip("/") + "/"
    api_url = f"{base_url}chat/completions"
    headers = {
        "Authorization": f"Bearer {llm_api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": llm_model,
        "messages": [
            {
                "role": "system",
                "content": "你是一位严谨的资深代码审查专家，擅长发现代码中的逻辑错误、安全隐患和性能问题。",
            },
            {
                "role": "user",
                "content": prompt,
            },
        ],
        "temperature": 0.3,
        "max_tokens": 2048,
    }
    logger.info(f"调用 LLM (OpenAI兼容): {api_url}, model={llm_model}")
    last_error = None
    for attempt in range(MAX_RETRIES + 1):
        try:
            resp = requests.post(
                api_url,
                headers=headers,
                json=payload,
                timeout=LLM_TIMEOUT,
            )
            if resp.status_code != 200:
                logger.warning(f"LLM 调用失败 (第 {attempt + 1} 次): " f"HTTP {resp.status_code} - {resp.text[:200]}")
                last_error = f"HTTP {resp.status_code}"
                continue
            data = resp.json()
            choices = data.get("choices", [])
            if not choices:
                logger.warning(f"LLM 返回空结果 (第 {attempt + 1} 次)")
                last_error = "empty choices"
                continue
            content = choices[0].get("message", {}).get("content", "")
            if not content.strip():
                logger.warning(f"LLM 返回空内容 (第 {attempt + 1} 次)")
                last_error = "empty content"
                continue
            logger.info(f"LLM 审查完成，结果长度: {len(content)} 字符")
            return content
        except requests.Timeout:
            logger.warning(f"LLM 调用超时 (第 {attempt + 1} 次)")
            last_error = "timeout"
        except requests.RequestException as e:
            logger.warning(f"LLM 调用异常 (第 {attempt + 1} 次): {e}")
            last_error = str(e)
    logger.error(f"LLM 调用最终失败: {last_error}")
    return None
 def call_llm_coze(
    prompt: str,
    llm_base_url: str,
    llm_api_key: str,
    llm_model: str,
    coze_bot_id: str,
 ) -> Optional[str]:
    """扣子(Coze)原生 Bot API 调用（支持异步轮询）"""
    import time
    base_url = llm_base_url.rstrip("/") + "/"
    api_url = f"{base_url}v3/chat"
    headers = {
        "Authorization": f"Bearer {llm_api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "bot_id": coze_bot_id,
        "user_id": "ci-code-review-bot",
        "stream": False,
        "additional_messages": [
            {
                "role": "user",
                "content": prompt,
                "content_type": "text",
            }
        ],
    }
    logger.info(f"调用 LLM (Coze): {api_url}, bot_id={coze_bot_id}")
    last_error = None
    for attempt in range(MAX_RETRIES + 1):
        try:
            resp = requests.post(
                api_url,
                headers=headers,
                json=payload,
                timeout=LLM_TIMEOUT,
            )
            if resp.status_code != 200:
                logger.warning(f"Coze 调用失败 (第 {attempt + 1} 次): " f"HTTP {resp.status_code} - {resp.text[:300]}")
                last_error = f"HTTP {resp.status_code}"
                continue
            data = resp.json()
            chat_data = data.get("data", {})
            chat_id = chat_data.get("id", "")
            conversation_id = chat_data.get("conversation_id", "")
            status = chat_data.get("status", "")
            # Coze v3 API 异步：先返回 in_progress，需要轮询
            if status == "in_progress" and conversation_id and chat_id:
                logger.info(f"Coze 异步处理中，开始轮询... (chat_id={chat_id[:12]}...)")
                # 轮询 message 列表接口（GET + query参数），最多等 LLM_TIMEOUT 秒
                poll_url = f"{base_url}v3/chat/message/list"
                poll_start = time.time()
                poll_interval = 3  # 每3秒轮询一次
                while time.time() - poll_start < LLM_TIMEOUT:
                    time.sleep(poll_interval)
                    poll_params = {
                        "chat_id": chat_id,
                        "conversation_id": conversation_id,
                    }
                    poll_resp = requests.get(
                        poll_url,
                        headers=headers,
                        params=poll_params,
                        timeout=GITEA_TIMEOUT,
                    )
                    if poll_resp.status_code != 200:
                        logger.debug(f"轮询返回 HTTP {poll_resp.status_code}: {poll_resp.text[:100]}")
                        continue
                    poll_data = poll_resp.json()
                    if poll_data.get("code", 0) != 0:
                        logger.debug(f"轮询返回错误: {poll_data.get('msg', '')}")
                        continue
                    messages = poll_data.get("data", []) or []
                    # 找assistant的answer消息
                    content = None
                    for msg in messages:
                        if msg.get("role") == "assistant" and msg.get("type") == "answer":
                            content = msg.get("content", "")
                            break
                    if content and content.strip():
                        logger.info(f"Coze 审查完成，结果长度: {len(content)} 字符")
                        return content
                logger.warning(f"Coze 轮询超时 ({LLM_TIMEOUT}s)，未拿到结果")
                last_error = "poll timeout"
                continue
            # 同步返回的情况（兼容）
            content = None
            messages = chat_data.get("messages", []) or data.get("messages", [])
            for msg in messages:
                if msg.get("role") == "assistant" and msg.get("type") == "answer":
                    content = msg.get("content", "")
                    break
            if not content:
                content = chat_data.get("content") or data.get("content")
            if not content:
                choices = data.get("choices", [])
                if choices:
                    content = choices[0].get("message", {}).get("content", "")
            if not content or not content.strip():
                logger.warning(f"Coze 返回空内容 (第 {attempt + 1} 次): {str(data)[:200]}")
                last_error = "empty content"
                continue
            logger.info(f"Coze 审查完成，结果长度: {len(content)} 字符")
            return content
        except requests.Timeout:
            logger.warning(f"Coze 调用超时 (第 {attempt + 1} 次)")
            last_error = "timeout"
        except requests.RequestException as e:
            logger.warning(f"Coze 调用异常 (第 {attempt + 1} 次): {e}")
            last_error = str(e)
    logger.error(f"Coze 调用最终失败: {last_error}")
    return None
 def call_llm_for_review(
    diff_text: str,
    pr_number: int,
    file_list: list,
    llm_base_url: str,
    llm_api_key: str,
    llm_model: str,
    coze_bot_id: str = "",
 ) -> Optional[str]:
    """
    调用 LLM 进行代码审查，返回审查结果文本。
    失败时返回 None。
    根据 LLM_PROVIDER 环境变量选择调用方式。
    """
    prompt = build_review_prompt(diff_text, pr_number, file_list)
    logger.info(f"Prompt 长度: {len(prompt)} 字符")
    provider = LLM_PROVIDER
    if provider == "coze":
        return call_llm_coze(prompt, llm_base_url, llm_api_key, llm_model, coze_bot_id)
    else:
        # 默认 OpenAI 兼容
        return call_llm_openai(prompt, llm_base_url, llm_api_key, llm_model)
 # ============== 主流程 ==============
 def main():
    parser = argparse.ArgumentParser(description="CI AI 代码审查脚本")
    parser.add_argument("--pr", type=int, help="PR 编号（也可通过 PR_NUMBER 环境变量）")
    parser.add_argument("--repo", type=str, help="仓库名 owner/repo（也可通过 REPO_NAME 环境变量）")
    parser.add_argument("--gitea-url", type=str, help="Gitea 地址（也可通过 GITEA_API_URL 环境变量）")
    parser.add_argument("--gitea-token", type=str, help="Gitea Token（也可通过 GITEA_TOKEN 环境变量）")
    parser.add_argument("--dry-run", action="store_true", help="只输出审查结果，不发表评论")
    args = parser.parse_args()
    # 读取配置
    gitea_url = args.gitea_url or os.getenv("GITEA_API_URL") or os.getenv("GITEA_SERVER_URL")
    gitea_token = args.gitea_token or os.getenv("GITEA_TOKEN")
    repo_name = args.repo or os.getenv("REPO_NAME") or os.getenv("GITEA_REPO")
    pr_number = args.pr or int(os.getenv("PR_NUMBER") or os.getenv("GITEA_PR_NUMBER") or 0)
    llm_base_url = os.getenv("LLM_BASE_URL")
    llm_api_key = os.getenv("LLM_API_KEY")
    llm_model = os.getenv("LLM_MODEL", "")
    coze_bot_id = os.getenv("COZE_BOT_ID", os.getenv("COZE_BOTID", ""))
    # 根据 provider 设置默认值
    provider = LLM_PROVIDER
    if provider == "coze":
        # 扣子模式：默认国内站，key 兼容多种环境变量名
        if not llm_base_url:
            llm_base_url = "https://api.coze.cn"
        if not llm_api_key:
            llm_api_key = os.getenv("COZE_API_KEY", "") or os.getenv("COZE_PAT", "")
    else:
        # OpenAI兼容模式：默认模型
        if not llm_model:
            llm_model = "gpt-4o-mini"
    # 必要参数校验
    missing = []
    if not gitea_url:
        missing.append("GITEA_API_URL")
    if not gitea_token:
        missing.append("GITEA_TOKEN")
    if not repo_name:
        missing.append("REPO_NAME")
    if not pr_number:
        missing.append("PR_NUMBER")
    if not llm_base_url:
        missing.append("LLM_BASE_URL")
    if not llm_api_key:
        missing.append("LLM_API_KEY")
    if provider == "coze" and not coze_bot_id:
        missing.append("COZE_BOT_ID (扣子模式需要)")
    if missing:
        logger.error(f"缺少必要配置: {', '.join(missing)}")
        sys.exit(1)
    logger.info(f"开始审查 PR #{pr_number}，仓库: {repo_name}")
    logger.info(f"Gitea: {gitea_url}")
    logger.info(f"LLM: {llm_base_url} (model={llm_model})")
    try:
        # 1. 初始化 Gitea 客户端
        gitea = GiteaClient(gitea_url, gitea_token, repo_name)
        # 2. 获取 PR diff 和文件列表
        try:
            diff_text = gitea.get_pr_diff(pr_number)
            file_list = gitea.get_pr_files(pr_number)
        except Exception as e:
            logger.error(f"获取 PR 信息失败: {e}")
            sys.exit(1)
        # 3. 过滤掉不需要审查的文件（如 lock 文件、生成的文件、二进制文件等）
        skip_extensions = (
            ".lock",
            ".sum",
            ".min.js",
            ".min.css",
            ".map",
            ".png",
            ".jpg",
            ".jpeg",
            ".gif",
            ".svg",
            ".ico",
            ".woff",
            ".woff2",
            ".ttf",
            ".eot",
        )
        skipped_files = []
        if file_list:
            skipped_files = [
                f.get("filename")
                for f in file_list
                if f.get("filename", "").endswith(skip_extensions) or f.get("status") == "removed"
            ]
            if skipped_files:
                logger.info(f"跳过 {len(skipped_files)} 个非文本/已删除文件: {', '.join(skipped_files[:5])}...")
        # 实际从 diff 中移除跳过的文件（按文件边界切割）
        if skipped_files:
            diff_lines = diff_text.split("\n")
            filtered_lines = []
            current_file = None
            skip_current = False
            i = 0
            while i < len(diff_lines):
                line = diff_lines[i]
                # 检测新文件开始: diff --git a/xxx b/xxx
                if line.startswith("diff --git "):
                    # 提取文件名
                    parts = line.split(" ")
                    if len(parts) >= 4:
                        # b/ 后面的是目标文件名
                        current_file = parts[3][2:] if parts[3].startswith("b/") else parts[3]
                        skip_current = any(current_file == sf for sf in skipped_files) or any(
                            current_file.endswith(ext) for ext in skip_extensions
                        )
                    else:
                        skip_current = False
                if not skip_current:
                    filtered_lines.append(line)
                i += 1
            original_len = len(diff_text)
            diff_text = "\n".join(filtered_lines)
            logger.info(f"Diff 过滤后: {original_len} -> {len(diff_text)} 字符 (减少 {original_len - len(diff_text)})")
        # 4. 截断过大的 diff
        diff_text, was_truncated = truncate_diff(diff_text, MAX_DIFF_CHARS)
        if was_truncated:
            logger.warning(f"Diff 过大，已截断至 {len(diff_text)} 字符")
        # 5. 如果 diff 为空，直接跳过
        if not diff_text.strip():
            logger.info("Diff 为空，无需审查")
            sys.exit(0)
        # 6. 调用 LLM 审查
        review_result = call_llm_for_review(
            diff_text=diff_text,
            pr_number=pr_number,
            file_list=file_list,
            llm_base_url=llm_base_url,
            llm_api_key=llm_api_key,
            llm_model=llm_model,
            coze_bot_id=coze_bot_id,
        )
        if not review_result:
            logger.error("LLM 审查失败")
            sys.exit(1)
        # 7. 加上审查时间和标识（便于识别是自动审查）
        from datetime import datetime
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        marker = "<!-- AI_CODE_REVIEW_AUTO_COMMENT -->"
        full_comment = f"""{review_result}
 ---
 <sub>🤖 由 AI 代码审查机器人自动生成 | {timestamp} | 模型: {llm_model}</sub>
 {marker}
 """
        # 8. 输出审查结果到日志
        logger.info("=" * 60)
        logger.info("审查结果:")
        for line in review_result.split("\n")[:30]:
            logger.info(line)
        if len(review_result.split("\n")) > 30:
            logger.info(f"... 共 {len(review_result.split(chr(10)))} 行")
        logger.info("=" * 60)
        # 9. 发布评论（先删除旧的审查评论，避免刷屏）
        if args.dry_run:
            logger.info("--dry-run 模式，跳过发布评论")
            print(full_comment)
        else:
            # 去重：删除之前的 AI 审查评论
            old_comments = gitea.get_existing_review_comments(pr_number, marker)
            if old_comments:
                logger.info(f"找到 {len(old_comments)} 条旧的 AI 审查评论，先删除")
                for cid in old_comments:
                    gitea.delete_pr_comment(pr_number, cid)
            # 发布新评论
            success = gitea.post_pr_comment(pr_number, full_comment)
            if not success:
                logger.error("评论发布失败")
                sys.exit(1)
        # 10. 判断是否有严重问题（可选阻断）
        # 目前只做建议，不阻断合并，始终返回 0
        has_critical = "问题" in review_result and ("❌" in review_result or "需修改" in review_result)
        if has_critical:
            logger.warning("检测到需修改的问题，但当前配置为仅建议，不阻断合并")
        logger.info("代码审查完成")
        sys.exit(0)
    except Exception as e:
        logger.exception(f"审查脚本发生未预期的异常: {e}")
        sys.exit(1)
 if __name__ == "__main__":
    main()
@@ -1,66 +1,30 @@
 #!/bin/sh
 # ===========================================
-# Staging 部署脚本（SSH 模式，支持自动回滚）
+# Staging 部署脚本（SSH 模式）
 # ===========================================
 # 通过 SSH 在 staging 服务器上执行
 #
 # 环境变量:
 #   IMAGE_TAG       - 镜像版本 tag（如 commit SHA 或分支名）
 #   REGISTRY_TOKEN  - Registry 访问令牌
-#   REGISTRY        - Registry 地址（默认 xiaoxia-registry.cn-hangzhou.cr.aliyuncs.com/xiaoxiakeji）
+#   REGISTRY        - Registry 地址（默认 git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas）
 #   REGISTRY_USER   - Registry 用户名（默认 xiaoxia）
 #   ENV_FILE        - 环境变量文件路径
 #   GENERATED_DIR   - 生成文件目录
 #   SKIP_MIGRATION  - 跳过数据库迁移（true/false，默认 false）
 #   SKIP_ROLLBACK   - 失败时跳过自动回滚（true/false，默认 false）
 set -eu
 # ---- 重试工具函数 ----
 retry_cmd() {
  local max_attempts=$1
  local backoff=$2
  shift 2
  local attempt=1
  while [ $attempt -le $max_attempts ]; do
    if "$@"; then
      return 0
    fi
    echo "  attempt $attempt/$max_attempts failed, retrying in ${backoff}s..."
    sleep $backoff
    backoff=$((backoff * 2))
    attempt=$((attempt + 1))
  done
  echo "  ERROR: failed after $max_attempts retries"
  return 1
 }
 retry_docker_login() {
  echo "Logging in to registry (up to 3 retries)"
  if retry_cmd 3 5 sh -c "printf %s "$REGISTRY_TOKEN" | docker login "$REGISTRY_HOST" -u "$REGISTRY_USER" --password-stdin"; then
    return 0
  fi
  echo "WARN: docker login failed after retries, will try pull anyway"
  return 0
 }
 retry_docker_pull() {
  local image=$1
  echo "Pulling $image (up to 3 retries)"
  retry_cmd 3 10 docker pull "$image"
 }
 IMAGE_TAG="${IMAGE_TAG:-}"
-REGISTRY="${REGISTRY:-xiaoxia-registry.cn-hangzhou.cr.aliyuncs.com/xiaoxiakeji}"
+REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
-REGISTRY_USER="${ACR_USERNAME:-${REGISTRY_USER:-nick0415343655}}"
+REGISTRY_USER="${REGISTRY_USER:-xiaoxia}"
-REGISTRY_TOKEN="${ACR_PASSWORD:-${REGISTRY_TOKEN:-}}"
+REGISTRY_TOKEN="${REGISTRY_TOKEN:-}"
 ENV_FILE="${ENV_FILE:-/var/lib/xiaoxia-saas-staging/.env}"
 GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-staging/generated}"
 LEGACY_ASSETS_DIR="${LEGACY_ASSETS_DIR:-/var/lib/xiaoxia-saas-staging/legacy-assets}"
 SKIP_MIGRATION="${SKIP_MIGRATION:-false}"
 SKIP_ROLLBACK="${SKIP_ROLLBACK:-false}"
 if [ -z "$IMAGE_TAG" ]; then
  echo "ERROR: IMAGE_TAG is required"
@@ -71,165 +35,17 @@ test -f "$ENV_FILE"
 mkdir -p "$GENERATED_DIR"
 mkdir -p "$LEGACY_ASSETS_DIR"
-echo "==========================================="
+echo "=========================================="
 echo "  Staging 部署 - $IMAGE_TAG"
-echo "==========================================="
+echo "=========================================="
 # ---- 记录当前运行的镜像版本（用于回滚） ----
 echo "Recording current image versions for rollback..."
 PREV_API_IMAGE=""
 PREV_WORKER_IMAGE=""
 PREV_WEB_IMAGE=""
 for c in xiaoxia-api-staging xiaoxia-worker-staging xiaoxia-web-staging; do
  if docker inspect "$c" >/dev/null 2>&1; then
    img=$(docker inspect -f '{{.Config.Image}}' "$c")
    case "$c" in
      xiaoxia-api-staging) PREV_API_IMAGE="$img" ;;
      xiaoxia-worker-staging) PREV_WORKER_IMAGE="$img" ;;
      xiaoxia-web-staging) PREV_WEB_IMAGE="$img" ;;
    esac
    echo "  $c -> $img"
  else
    echo "  $c -> (not running)"
  fi
 done
 # ---- 回滚函数 ----
 rollback() {
  echo ""
  echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"
  echo "  部署失败，正在自动回滚到上一版本..."
  echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"
  echo ""
  if [ "$SKIP_ROLLBACK" = "true" ]; then
    echo "SKIP_ROLLBACK=true，跳过自动回滚"
    exit 1
  fi
  # 停止当前（失败的）新容器
  echo "Stopping new containers..."
  docker rm -f xiaoxia-api-staging 2>/dev/null || true
  docker rm -f xiaoxia-worker-staging 2>/dev/null || true
  docker rm -f xiaoxia-web-staging 2>/dev/null || true
  LOG_OPTS="--log-driver json-file --log-opt max-size=50m --log-opt max-file=3"
  # 恢复 API
  if [ -n "$PREV_API_IMAGE" ]; then
    echo "Rolling back API to: $PREV_API_IMAGE"
    docker run -d \
      --name xiaoxia-api-staging \
      --env-file "$ENV_FILE" \
      --network xiaoxia-net-staging \
      -p 127.0.0.1:8000:8000 \
      -e APP_ENV=staging \
      -e APP_VERSION="$(echo $PREV_API_IMAGE | grep -oE '[^:]+$')" \
      -e GENERATED_FILES_DIR=/app/generated \
      -e GENERATED_FILES_URL_PREFIX=/generated-files \
      -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
      -v "$GENERATED_DIR:/app/generated" \
      --restart unless-stopped \
      --health-cmd "python -c \"import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=5)\"" \
      --health-interval 30s \
      --health-timeout 10s \
      --health-retries 3 \
      --health-start-period 40s \
      $LOG_OPTS \
      "$PREV_API_IMAGE"
  else
    echo "No previous API image to roll back to"
  fi
  # 恢复 Worker
  if [ -n "$PREV_WORKER_IMAGE" ]; then
    echo "Rolling back Worker to: $PREV_WORKER_IMAGE"
    docker run -d \
      --name xiaoxia-worker-staging \
      --env-file "$ENV_FILE" \
      --network xiaoxia-net-staging \
      -e APP_ENV=staging \
      -e APP_VERSION="$(echo $PREV_WORKER_IMAGE | grep -oE '[^:]+$')" \
      -e WORKER_CONCURRENCY=1 \
      -e WORKER_MAX_TASKS_PER_CHILD=100 \
      -e GENERATED_FILES_DIR=/app/generated \
      -e GENERATED_FILES_URL_PREFIX=/generated-files \
      -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
      -v "$GENERATED_DIR:/app/generated" \
      --restart unless-stopped \
      --health-cmd "sh -c \"grep -q celery /proc/1/cmdline || exit 1\"" \
      --health-interval 30s \
      --health-timeout 10s \
      --health-retries 3 \
      --health-start-period 30s \
      $LOG_OPTS \
      "$PREV_WORKER_IMAGE"
  else
    echo "No previous Worker image to roll back to"
  fi
  # 恢复 Web
  if [ -n "$PREV_WEB_IMAGE" ]; then
    echo "Rolling back Web to: $PREV_WEB_IMAGE"
    LEGACY_VOLUME=""
    if [ -d "$LEGACY_ASSETS_DIR" ] && [ "$(ls -A "$LEGACY_ASSETS_DIR" 2>/dev/null)" ]; then
      LEGACY_VOLUME="-v ${LEGACY_ASSETS_DIR}:/usr/share/nginx/html/assets-legacy/assets:ro"
    fi
    docker run -d \
      --name xiaoxia-web-staging \
      --network xiaoxia-net-staging \
      -p 127.0.0.1:3001:80 \
      --restart unless-stopped \
      $LEGACY_VOLUME \
      --health-cmd "wget --spider -q http://127.0.0.1:80" \
      --health-interval 30s \
      --health-timeout 5s \
      --health-retries 3 \
      $LOG_OPTS \
      "$PREV_WEB_IMAGE"
  else
    echo "No previous Web image to roll back to"
  fi
  # 等待 API 回滚后恢复健康
  if [ -n "$PREV_API_IMAGE" ]; then
    echo "Waiting for rolled-back API to become healthy..."
    i=0
    while [ "$i" -lt 40 ]; do
      if curl -sf --max-time 5 http://127.0.0.1:8000/health >/dev/null 2>&1; then
        echo "Rolled-back API is healthy!"
        break
      fi
      i=$((i + 1))
      echo "  Waiting... ($i/40)"
      sleep 3
    done
    if [ "$i" -ge 40 ]; then
      echo "WARN: Rolled-back API did not become healthy within 120s"
      docker logs --tail 30 xiaoxia-api-staging
    fi
  fi
  echo ""
  echo "==========================================="
  echo "  回滚完成"
  echo "==========================================="
  echo "Previous API:    ${PREV_API_IMAGE:-none}"
  echo "Previous Worker: ${PREV_WORKER_IMAGE:-none}"
  echo "Previous Web:    ${PREV_WEB_IMAGE:-none}"
  echo ""
  echo "部署失败，已自动回滚到上一版本"
  docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Image}}" | grep staging
  exit 1
 }
 # ---- 登录 Registry ----
 if [ -n "$REGISTRY_TOKEN" ]; then
-  echo "=========================================="
+  echo "Logging in to registry: $REGISTRY"
  echo "  Login to Registry (with retries)"
  echo "=========================================="
  REGISTRY_HOST=$(echo "$REGISTRY" | cut -d/ -f1)
-  retry_docker_login
+  printf %s "$REGISTRY_TOKEN" | docker login "$REGISTRY_HOST" -u "$REGISTRY_USER" --password-stdin 2>/dev/null || {
    echo "WARN: docker login failed, will try to pull anyway"
  }
 fi
 # ---- Pull 新版本镜像 ----
@@ -241,12 +57,12 @@ LOCAL_API="xiaoxia-saas-api:${IMAGE_TAG}"
 LOCAL_WORKER="xiaoxia-saas-worker:${IMAGE_TAG}"
 LOCAL_WEB="xiaoxia-saas-web:${IMAGE_TAG}"
-echo "=========================================="
+echo "Pulling API image..."
-echo "  Pull images (with retries)"
+docker pull "$REGISTRY_API"
-echo "=========================================="
+echo "Pulling Worker image..."
-retry_docker_pull "$REGISTRY_API"
+docker pull "$REGISTRY_WORKER"
-retry_docker_pull "$REGISTRY_WORKER"
+echo "Pulling Web image..."
-retry_docker_pull "$REGISTRY_WEB"
+docker pull "$REGISTRY_WEB"
 # Re-tag 成本地名
 docker tag "$REGISTRY_API" "$LOCAL_API"
@@ -301,12 +117,7 @@ if [ "$SKIP_MIGRATION" != "true" ]; then
    --env-file "$ENV_FILE" \
    --network xiaoxia-net-staging \
    -e APP_ENV=staging \
-    "$LOCAL_API" sh -c "cd /app && alembic upgrade head" || {
+    "$LOCAL_API" sh -c "cd /app && alembic upgrade head"
    echo "ERROR: Database migration failed"
    echo "Note: Migration failures are NOT automatically rolled back (data safety)"
    echo "Please manually check and fix the migration, then redeploy"
    exit 1
  }
  echo "Migrations completed."
 else
  echo "Skipping migrations (SKIP_MIGRATION=true)"
@@ -340,7 +151,7 @@ docker run -d \
  --health-retries 3 \
  --health-start-period 40s \
  $LOG_OPTS \
-  "$LOCAL_API" || rollback
+  "$LOCAL_API"
 # ---- 启动 Worker ----
 echo "Starting Worker container..."
@@ -363,7 +174,7 @@ docker run -d \
  --health-retries 3 \
  --health-start-period 30s \
  $LOG_OPTS \
-  "$LOCAL_WORKER" || rollback
+  "$LOCAL_WORKER"
 # ---- 启动 Web ----
 LEGACY_VOLUME=""
@@ -386,7 +197,7 @@ docker run -d \
  --health-timeout 5s \
  --health-retries 3 \
  $LOG_OPTS \
-  "$LOCAL_WEB" || rollback
+  "$LOCAL_WEB"
 # ---- 等待 API 健康 ----
 echo "Waiting for API to become healthy..."
@@ -404,7 +215,7 @@ done
 if [ "$i" -ge 40 ]; then
  echo "ERROR: API did not become healthy within 120s"
  docker logs --tail 50 xiaoxia-api-staging
-  rollback
+  exit 1
 fi
 # ---- 等待 Web 健康 ----
@@ -423,7 +234,7 @@ done
 if [ "$i" -ge 15 ]; then
  echo "ERROR: Web did not become healthy within 30s"
  docker logs --tail 30 xiaoxia-web-staging
-  rollback
+  exit 1
 fi
 # ---- 清理旧镜像 ----
@@ -137,7 +137,7 @@ class PerfAssert:
            result = PerfResult(name=name or threshold_level, threshold_ms=threshold_ms)
            last_response = None
-            for _ in range(num_samples):
+            for i in range(num_samples):
                start = time.perf_counter()
                last_response = func()
                elapsed = (time.perf_counter() - start) * 1000
@@ -261,7 +261,7 @@ def run_perf_test(
    result = PerfResult(name=name, threshold_ms=threshold_ms)
    last_response = None
-    for _ in range(samples):
+    for i in range(samples):
        start = time.perf_counter()
        last_response = func()
        elapsed = (time.perf_counter() - start) * 1000
@@ -530,7 +530,7 @@ class TestLargeDataRequests:
    def test_rapid_sequential_requests(self, auth_headers):
        """快速连续请求不应触发限流导致 500。"""
        statuses = []
-        for _ in range(20):
+        for i in range(20):
            resp = client.get("/api/v1/projects", headers=auth_headers)
            statuses.append(resp.status_code)
@@ -665,16 +665,7 @@ class TestResponseSchema:
        resp = client.get(f"/api/v1/edit-plans/{plan.id}/generation-status")
        assert resp.status_code == 200
        data = resp.json()
-        expected_keys = {
+        expected_keys = {"plan_id", "plan_status", "generation_task_id", "clips"}
            "plan_id",
            "plan_status",
            "generation_task_id",
            "generation_task_status",
            "progress",
            "video_url",
            "error_message",
            "clips",
        }
        assert set(data.keys()) == expected_keys
@@ -636,94 +636,3 @@ class TestGenerationWorkflow:
        p = svc.create_plan("tpl-001", "测试", config={"key1": "val1"})
        updated = svc.update_plan_config(p.id, {"key1": "new_val"})
        assert updated.config["key1"] == "new_val"
 # ── 重新编辑 & 再生成 ────────────────────────────────────────────────────────
 class TestResumeEditingAndRegenerate:
    """完成/失败后重新编辑 → 再生成的状态流转测试"""
    def test_update_plan_from_completed_returns_to_editing(self):
        """更新计划配置：completed → 自动切回 editing"""
        svc = _make_service()
        p = svc.create_plan("tpl-001", "测试")
        svc.transition_status(p.id, EditPlanStatus.EDITING)
        svc.transition_status(p.id, EditPlanStatus.RENDERING)
        svc.transition_status(p.id, EditPlanStatus.COMPLETED)
        updated = svc.update_plan(p.id, name="新名字")
        assert updated.status == EditPlanStatus.EDITING
        assert updated.name == "新名字"
    def test_update_plan_config_from_completed_returns_to_editing(self):
        """update_plan_config: completed → 自动切回 editing"""
        svc = _make_service()
        p = svc.create_plan("tpl-001", "测试")
        svc.transition_status(p.id, EditPlanStatus.EDITING)
        svc.transition_status(p.id, EditPlanStatus.RENDERING)
        svc.transition_status(p.id, EditPlanStatus.COMPLETED)
        updated = svc.update_plan_config(p.id, {"foo": "bar"})
        assert updated.status == EditPlanStatus.EDITING
    def test_create_clip_from_completed_returns_to_editing(self):
        """创建片段：completed → 自动切回 editing"""
        svc = _make_service()
        p = svc.create_plan("tpl-001", "测试")
        svc.transition_status(p.id, EditPlanStatus.EDITING)
        svc.transition_status(p.id, EditPlanStatus.RENDERING)
        svc.transition_status(p.id, EditPlanStatus.COMPLETED)
        svc.create_clip(p.id, "main", 0)
        plan_after = svc.get_plan(p.id)
        assert plan_after.status == EditPlanStatus.EDITING
    def test_assign_asset_from_failed_returns_to_editing(self):
        """分配素材：failed → 自动切回 editing"""
        svc = _make_service()
        p = svc.create_plan("tpl-001", "测试")
        clip = svc.create_clip(p.id, "main", 0)
        svc.transition_status(p.id, EditPlanStatus.EDITING)
        svc.transition_status(p.id, EditPlanStatus.RENDERING)
        svc.transition_status(p.id, EditPlanStatus.FAILED)
        svc.assign_asset(clip.id, "asset-001")
        plan_after = svc.get_plan(p.id)
        assert plan_after.status == EditPlanStatus.EDITING
    def test_completed_can_regenerate_after_edit(self):
        """完成后编辑 → can_generate 返回 True，可再生成"""
        svc = _make_service()
        p = svc.create_plan("tpl-001", "测试")
        svc.create_clip(p.id, "main", 0)
        svc.transition_status(p.id, EditPlanStatus.EDITING)
        svc.transition_status(p.id, EditPlanStatus.RENDERING)
        svc.transition_status(p.id, EditPlanStatus.COMPLETED)
        # 完成后不能直接生成
        can, reason = svc.can_generate(p.id)
        assert not can
        assert "编辑" in reason
        # 编辑后自动切回 editing，可以生成
        svc.update_plan_config(p.id, {"edited": True})
        can, reason = svc.can_generate(p.id)
        assert can, f"期望可生成，实际: {reason}"
    def test_transition_completed_to_editing_via_service(self):
        """通过 transition_status 从 completed 切到 editing"""
        svc = _make_service()
        p = svc.create_plan("tpl-001", "测试")
        svc.transition_status(p.id, EditPlanStatus.EDITING)
        svc.transition_status(p.id, EditPlanStatus.RENDERING)
        svc.transition_status(p.id, EditPlanStatus.COMPLETED)
        result = svc.transition_status(p.id, EditPlanStatus.EDITING)
        assert result.status == EditPlanStatus.EDITING
    def test_resume_editing_from_draft_raises(self):
        """从 draft 直接 resume_editing 应该报错"""
        p = EditPlan.create("tpl-001", "测试")
        with pytest.raises(ValueError):
            p.resume_editing()
@@ -463,291 +463,3 @@ class TestCompositeQueries:
        t = svc.create_template(name="空模板")
        result = svc.get_template_with_configs(t.id)
        assert len(result["clip_configs"]) == 0
 # ===========================================================================
 # Stub Repositories for EditPlan (save_as_template 测试用)
 # ===========================================================================
 class StubEditPlanRepository:
    """内存中的 EditPlan 仓储 stub"""
    def __init__(self) -> None:
        self._plans: dict[str, EditPlan] = {}
        self._counter = 0
    def _next_id(self) -> str:
        self._counter += 1
        return f"plan-{self._counter:03d}"
    def get(self, plan_id: str) -> Optional[EditPlan]:
        return self._plans.get(plan_id)
    def create(self, plan: EditPlan) -> EditPlan:
        if not plan.id:
            plan.id = self._next_id()
        self._plans[plan.id] = plan
        return plan
 class StubEditPlanClipRepository:
    """内存中的 EditPlanClip 仓储 stub"""
    def __init__(self) -> None:
        self._clips: dict[str, EditPlanClip] = {}
        self._counter = 0
    def _next_id(self) -> str:
        self._counter += 1
        return f"clip-{self._counter:03d}"
    def list_by_plan(
        self,
        plan_id: str,
        *,
        status: Optional[str] = None,
        skip: int = 0,
        limit: int = 100,
    ) -> List[EditPlanClip]:
        items = [c for c in self._clips.values() if c.plan_id == plan_id]
        if status:
            items = [c for c in items if c.status.value == status]
        items.sort(key=lambda c: c.order)
        return items[skip : skip + limit]
    def create(self, clip: EditPlanClip) -> EditPlanClip:
        if not clip.id:
            clip.id = self._next_id()
        self._clips[clip.id] = clip
        return clip
 def _make_service_with_plan_stubs():
    """创建使用 stub 仓储的 EditTemplateService（含 plan 相关 stub）"""
    from app.services.edit_template_service import EditTemplateService
    db = MagicMock()
    svc = EditTemplateService(db)
    svc._template_repo = StubEditTemplateRepository()
    svc._clip_config_repo = StubTemplateClipConfigRepository()
    svc._plan_repo = StubEditPlanRepository()
    svc._plan_clip_repo = StubEditPlanClipRepository()
    return svc
 def _make_test_plan_with_clips(svc, *, clip_count: int = 3, plan_config=None):
    """辅助方法：创建一个带片段的测试计划，返回 plan 对象"""
    from packages.domain.edit_plan import EditPlan
    from packages.domain.edit_plan_clip import EditPlanClip
    plan = EditPlan.create(
        template_id="tpl-source",
        name="我的剪辑计划",
        config=plan_config or {"editing_mode": "one_take", "theme": "minimal"},
        project_id="proj-001",
        created_by_user_id="user-001",
    )
    plan.id = "plan-test-001"
    svc._plan_repo.create(plan)
    for i in range(clip_count):
        clip = EditPlanClip.create(
            plan_id=plan.id,
            clip_type=ClipType.MAIN.value,
            order=i,
            asset_id=f"asset-{i:03d}",
            text_content=f"片段{i}的文案",
            duration=10.0 + i * 5,
            transition_effect="cut" if i == 0 else "fade",
            playback_speed=1.0 if i == 0 else 1.5,
            config={"filter": "vivid"} if i == 1 else {},
        )
        svc._plan_clip_repo.create(clip)
    return plan
 # ===========================================================================
 # 保存为模板测试
 # ===========================================================================
 class TestSavePlanAsTemplate:
    """从剪辑计划保存为模板测试"""
    def test_basic_save_as_template(self):
        """基础场景：将有3个片段的计划保存为模板"""
        svc = _make_service_with_plan_stubs()
        plan = _make_test_plan_with_clips(svc, clip_count=3)
        result = svc.save_plan_as_template(plan.id, name="我的自定义模板")
        assert result["template"].name == "我的自定义模板"
        assert result["template"].template_type == "custom"
        assert result["template"].editing_mode == "one_take"
        assert result["template"].status == EditTemplateStatus.ACTIVE
        assert len(result["clip_configs"]) == 3
    def test_clip_configs_correctly_converted(self):
        """片段正确转换为模板片段配置"""
        svc = _make_service_with_plan_stubs()
        plan = _make_test_plan_with_clips(svc, clip_count=2)
        result = svc.save_plan_as_template(plan.id, name="转换测试模板")
        configs = result["clip_configs"]
        configs.sort(key=lambda c: c.order)
        # 第0个片段
        assert configs[0].clip_type == ClipType.MAIN
        assert configs[0].order == 0
        assert configs[0].min_duration == 10.0
        assert configs[0].max_duration == 10.0
        assert configs[0].text_template == "片段0的文案"
        assert configs[0].transition_effect.value == "cut"
        # playback_speed=1.0 不存
        assert "playback_speed" not in configs[0].config
        # 第1个片段
        assert configs[1].order == 1
        assert configs[1].min_duration == 15.0
        assert configs[1].max_duration == 15.0
        assert configs[1].transition_effect.value == "fade"
        # playback_speed=1.5 存入config
        assert configs[1].config.get("playback_speed") == 1.5
        # config 中的 filter 保留
        assert configs[1].config.get("filter") == "vivid"
    def test_no_asset_id_in_template(self):
        """模板不保留具体素材ID"""
        svc = _make_service_with_plan_stubs()
        plan = _make_test_plan_with_clips(svc, clip_count=2)
        result = svc.save_plan_as_template(plan.id, name="素材剥离测试")
        for cfg in result["clip_configs"]:
            # 模板片段配置没有 asset_id 字段
            assert not hasattr(cfg, "asset_id") or not getattr(cfg, "asset_id", "")
            # config 中也不应有素材相关字段
            assert "asset_info" not in cfg.config
            assert "source_asset_id" not in cfg.config
    def test_template_config_stripped_of_runtime_fields(self):
        """模板config剥离运行时字段"""
        svc = _make_service_with_plan_stubs()
        plan_config = {
            "editing_mode": "one_take",
            "theme": "cinematic",
            "asset_ids": ["a1", "a2"],
            "source_edit_plan_id": "old-plan",
            "generation_task_id": "task-123",
        }
        plan = _make_test_plan_with_clips(svc, clip_count=1, plan_config=plan_config)
        result = svc.save_plan_as_template(plan.id, name="配置剥离测试")
        tpl_config = result["template"].config
        assert tpl_config.get("theme") == "cinematic"
        assert "asset_ids" not in tpl_config
        assert "source_edit_plan_id" not in tpl_config
        assert "generation_task_id" not in tpl_config
    def test_plan_not_found_raises_error(self):
        """计划不存在时报错"""
        svc = _make_service_with_plan_stubs()
        with pytest.raises(ValueError, match="剪辑计划不存在"):
            svc.save_plan_as_template("nonexistent-plan", name="不存在的计划")
    def test_empty_name_raises_error(self):
        """模板名称为空时报错"""
        svc = _make_service_with_plan_stubs()
        plan = _make_test_plan_with_clips(svc, clip_count=1)
        with pytest.raises(ValueError, match="模板名称不能为空"):
            svc.save_plan_as_template(plan.id, name="   ")
    def test_duplicate_name_raises_error(self):
        """模板名称重复时报错"""
        svc = _make_service_with_plan_stubs()
        svc.create_template(name="重名模板")
        plan = _make_test_plan_with_clips(svc, clip_count=1)
        with pytest.raises(ValueError, match="模板名称已存在"):
            svc.save_plan_as_template(plan.id, name="重名模板")
    def test_save_zero_clip_plan(self):
        """零片段计划也能保存为模板"""
        svc = _make_service_with_plan_stubs()
        from packages.domain.edit_plan import EditPlan
        plan = EditPlan.create(
            template_id="tpl-source",
            name="空计划",
            config={"editing_mode": "one_take"},
        )
        plan.id = "plan-empty"
        svc._plan_repo.create(plan)
        result = svc.save_plan_as_template(plan.id, name="空模板")
        assert result["template"].name == "空模板"
        assert len(result["clip_configs"]) == 0
    def test_custom_description_and_type(self):
        """自定义描述和模板类型"""
        svc = _make_service_with_plan_stubs()
        plan = _make_test_plan_with_clips(svc, clip_count=1)
        result = svc.save_plan_as_template(
            plan.id,
            name="自定义模板",
            description="这是一个测试模板",
            template_type="vlog",
        )
        assert result["template"].description == "这是一个测试模板"
        assert result["template"].template_type == "vlog"
    def test_unknown_transition_effect_falls_back_to_cut(self):
        """未知转场效果回退到cut"""
        svc = _make_service_with_plan_stubs()
        from packages.domain.edit_plan import EditPlan
        from packages.domain.edit_plan_clip import EditPlanClip
        plan = EditPlan.create(template_id="tpl-src", name="转场测试计划")
        plan.id = "plan-transition-test"
        svc._plan_repo.create(plan)
        clip = EditPlanClip.create(
            plan_id=plan.id,
            clip_type="main",
            order=0,
            duration=10.0,
            transition_effect="weird_effect_that_does_not_exist",
        )
        svc._plan_clip_repo.create(clip)
        result = svc.save_plan_as_template(plan.id, name="转场兼容模板")
        assert result["clip_configs"][0].transition_effect.value == "cut"
    def test_unknown_clip_type_falls_back_to_main(self):
        """未知片段类型回退到main"""
        svc = _make_service_with_plan_stubs()
        from packages.domain.edit_plan import EditPlan
        from packages.domain.edit_plan_clip import EditPlanClip
        plan = EditPlan.create(template_id="tpl-src", name="类型测试计划")
        plan.id = "plan-type-test"
        svc._plan_repo.create(plan)
        clip = EditPlanClip.create(
            plan_id=plan.id,
            clip_type="unknown_clip_type",
            order=0,
            duration=10.0,
        )
        svc._plan_clip_repo.create(clip)
        result = svc.save_plan_as_template(plan.id, name="类型兼容模板")
        assert result["clip_configs"][0].clip_type == ClipType.MAIN
@@ -44,7 +44,7 @@ class TestEditTemplate:
    def test_create_empty_name_raises(self):
        try:
            EditTemplate.create("   ")
-            raise AssertionError("应该抛出 ValueError")
+            assert False, "应该抛出 ValueError"
        except ValueError as e:
            assert "模板名称不能为空" in str(e)
@@ -74,14 +74,14 @@ class TestEditPlan:
    def test_create_empty_name_raises(self):
        try:
            EditPlan.create("tpl-1", "  ")
-            raise AssertionError("应该抛出 ValueError")
+            assert False, "应该抛出 ValueError"
        except ValueError as e:
            assert "计划名称不能为空" in str(e)
    def test_create_empty_template_id_raises(self):
        try:
            EditPlan.create("  ", "test")
-            raise AssertionError("应该抛出 ValueError")
+            assert False, "应该抛出 ValueError"
        except ValueError as e:
            assert "template_id 不能为空" in str(e)
@@ -112,7 +112,7 @@ class TestEditPlan:
        p = EditPlan.create("tpl-1", "test")
        try:
            p.start_rendering()  # draft → rendering 不合法
-            raise AssertionError("应该抛出 ValueError")
+            assert False, "应该抛出 ValueError"
        except ValueError as e:
            logger.warning(f"Operation failed in tests/unit/test_phase8_edit_models.py: {e}", exc_info=True)
@@ -120,7 +120,7 @@ class TestEditPlan:
        p = EditPlan.create("tpl-1", "test")
        try:
            p.mark_completed()  # draft → completed 不合法
-            raise AssertionError("应该抛出 ValueError")
+            assert False, "应该抛出 ValueError"
        except ValueError as e:
            logger.warning(f"Operation failed in tests/unit/test_phase8_edit_models.py: {e}", exc_info=True)
@@ -128,7 +128,7 @@ class TestEditPlan:
        p = EditPlan.create("tpl-1", "test")
        try:
            p.reset_to_draft()  # draft → draft 不合法
-            raise AssertionError("应该抛出 ValueError")
+            assert False, "应该抛出 ValueError"
        except ValueError as e:
            logger.warning(f"Operation failed in tests/unit/test_phase8_edit_models.py: {e}", exc_info=True)
@@ -253,7 +253,7 @@ class TestConcatSecurity:
        # 创建超过上限的段数
        segments = []
-        for _ in range(MAX_CONCAT_SEGMENTS + 5):
+        for i in range(MAX_CONCAT_SEGMENTS + 5):
            segments.append(ConcatSegment(video_path=str(sample_video)))
        config = ConcatConfig(segments=segments)
@@ -367,7 +367,7 @@ class TestVerifyUrlRedirectValidation:
                    except urllib.error.HTTPError as e:
                        if 300 <= e.code < 400 and e.headers.get("Location"):
                            if redirect_count >= max_redirects:
-                                raise Exception(f"重定向次数超过上限 ({max_redirects})") from e
+                                raise Exception(f"重定向次数超过上限 ({max_redirects})")
                            location = e.headers["Location"]
                            current = urljoin(safe_url, location)
                            redirect_count += 1
@@ -115,7 +115,7 @@ class TestAudioMerger:
        # 创建临时文件
        paths = []
-        for _ in range(3):
+        for i in range(3):
            with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
                f.write(b"audio")
                paths.append(f.name)
@@ -151,7 +151,7 @@ class TestAudioMerger:
        mock_run_ffmpeg.side_effect = CalledProcessError(returncode=1, cmd=["ffmpeg"], stderr="error details")
        paths = []
-        for _ in range(2):
+        for i in range(2):
            with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
                f.write(b"audio")
                paths.append(f.name)
@@ -45,7 +45,6 @@ class FakeClip:
    start_time: float = 0.0
    duration: float = 0.0
    transition_effect: str = "cut"
    transition_duration: float = 0.0
    status: str = "ready"
    config: dict[str, Any] = field(default_factory=dict)
@@ -66,7 +65,6 @@ def _make_clip(
    asset_id: str = "",
    duration: float = 0.0,
    transition_effect: str = "cut",
    transition_duration: float = 0.0,
    config: dict[str, Any] | None = None,
 ) -> FakeClip:
    return FakeClip(
@@ -76,7 +74,6 @@ def _make_clip(
        asset_id=asset_id or f"asset_{clip_id}.mp4",
        duration=duration,
        transition_effect=transition_effect,
        transition_duration=transition_duration,
        config=config or {},
    )
@@ -335,7 +332,7 @@ class TestBuildFilterComplex:
        assert "[final_video]" in fc
    def test_single_layer_multi_clips(self):
-        """多个 main clips（默认硬切）→ concat 串联。"""
+        """多个 main clips → xfade 串联。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("c2", "main", order=1, duration=3.0),
@@ -346,35 +343,6 @@ class TestBuildFilterComplex:
        }
        svc = _make_service(clips, asset_paths)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, input_args = svc._build_filter_complex(layers)
        assert input_args.count("-i") == 2
        # 全硬切场景走 concat filter（性能远优于 xfade）
        assert "concat=n=2:v=1:a=0" in fc
        assert "[final_video]" in fc
    def test_single_layer_multi_clips_with_transition(self):
        """多个 main clips 带转场效果 → xfade 串联。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip(
                "c2",
                "main",
                order=1,
                duration=3.0,
                transition_effect="fade",
                transition_duration=0.5,
            ),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
        }
        svc = _make_service(clips, asset_paths)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
@@ -412,7 +380,7 @@ class TestBuildFilterComplex:
        """
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
-            _make_clip("c2", "main", order=1, duration=5.0, transition_effect="fade", transition_duration=0.5),
+            _make_clip("c2", "main", order=1, duration=5.0),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
@@ -169,69 +169,6 @@ class TestStartClone:
        assert profile.status == VoiceCloneStatus.PENDING
        mock_cosyvoice.submit_clone_task.assert_not_called()
    def test_start_clone_ssrf_internal_url_rejected(self) -> None:
        """SSRF 防护：内网 URL 应该被拒绝，profile 标记为 failed。"""
        mock_repo = MagicMock()
        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
        mock_repo.create.side_effect = lambda p: p
        mock_repo.update.side_effect = lambda p: p
        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
        profile = service.start_clone(
            user_id="user-123",
            name="测试音色",
            source_audio_url="http://127.0.0.1/audio.wav",
        )
        # 内网 IP 应该被拒绝，标记为 failed
        assert profile.status == VoiceCloneStatus.FAILED
        assert "安全校验失败" in profile.error_message
        mock_cosyvoice.submit_clone_task.assert_not_called()
    def test_start_clone_ssrf_private_ip_rejected(self) -> None:
        """SSRF 防护：私有网段 IP 应该被拒绝，profile 标记为 failed。"""
        mock_repo = MagicMock()
        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
        mock_repo.create.side_effect = lambda p: p
        mock_repo.update.side_effect = lambda p: p
        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
        profile = service.start_clone(
            user_id="user-123",
            name="测试音色",
            source_audio_url="http://192.168.1.100/audio.wav",
        )
        assert profile.status == VoiceCloneStatus.FAILED
        assert "安全校验失败" in profile.error_message
        mock_cosyvoice.submit_clone_task.assert_not_called()
    def test_start_clone_ssrf_public_url_passes(self) -> None:
        """SSRF 防护：正常公网 URL 应该通过校验。"""
        mock_repo = MagicMock()
        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
        mock_cosyvoice.submit_clone_task.return_value = {
            "voice_id": "voice-ssrf-test",
            "status": "DEPLOYING",
            "request_id": "req-ssrf",
        }
        mock_repo.create.side_effect = lambda p: p
        mock_repo.update.side_effect = lambda p: p
        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
        profile = service.start_clone(
            user_id="user-123",
            name="测试音色",
            source_audio_url="https://example.com/audio.wav",
        )
        # 公网 URL 应该正常通过
        assert profile.status == VoiceCloneStatus.PROCESSING
        mock_cosyvoice.submit_clone_task.assert_called_once()
 # ── process_clone_result ─────────────────────────────────
@@ -376,27 +313,6 @@ class TestRetryClone:
        assert result.status == VoiceCloneStatus.FAILED
        assert "重试失败" in result.error_message
    def test_retry_clone_ssrf_internal_url_rejected(self) -> None:
        """重试时 SSRF 防护：内网 URL 应该被拒绝，profile 标记为 failed。"""
        mock_repo = MagicMock()
        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
        profile = _make_profile(
            status=VoiceCloneStatus.FAILED,
            source_audio_url="http://10.0.0.1/secret.wav",
            retry_count=0,
            max_retries=3,
        )
        mock_repo.get.return_value = profile
        mock_repo.update.side_effect = lambda p: p
        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
        result = service.retry_clone(profile.id, "user-123")
        assert result.status == VoiceCloneStatus.FAILED
        assert "安全校验失败" in result.error_message
        mock_cosyvoice.submit_clone_task.assert_not_called()
 # ── poll_and_process_clone ───────────────────────────────