feat(ci): 部署脚本从 base64 内嵌改为调用仓库脚本

- Staging: 从 Watchtower 改为 SSH 脚本主动部署
- Production: 从 base64 内嵌改为调用 scripts/ci-deploy-production.sh
- 部署脚本版本化，可审阅、可回滚

.env.example

@@ -3,6 +3,7 @@
 # ==================== 应用配置 ====================
 APP_NAME=小虾 SaaS
 APP_BASE_URL=http://localhost:3000
+APP_ENV=development
 
 # ==================== 数据库配置 ====================
 DATABASE_URL=postgresql://xiaoxia_user:your_password@localhost:5432/xiaoxia_saas
@@ -35,7 +36,8 @@ ENVIRONMENT=development
 DEBUG=true
 
 # ==================== CORS 配置 ====================
-CORS_ORIGINS=["http://localhost:3000","http://localhost:5173"]
+# 逗号分隔的域名列表（Settings 读取 CORS_ORIGINS_RAW）
+CORS_ORIGINS_RAW=http://localhost:3000,http://localhost:5173
 
 # ==================== 阿里云 OSS 配置 ====================
 OSS_ENDPOINT=oss-cn-hangzhou.aliyuncs.com
@@ -49,6 +51,7 @@ OSS_BUCKET_NAME=xiaoxia-autocut
 #       cosyvoice-v3-plus (高质量，系统音色少)
 #       cosyvoice-v3.5-flash / cosyvoice-v3.5-plus (仅支持克隆/设计音色，无系统音色)
 # 音色: v3系列系统音色带 _v3 后缀，如 longxiaochun_v3, longxiaoxia_v3, longanyang (无后缀)
+# 注意：COSYVOICE_* 变量由 packages/shared/config.py 的 SharedSettings 读取
 COSYVOICE_API_KEY=your-cosyvoice-api-key
 COSYVOICE_BASE_URL=https://dashscope.aliyuncs.com/api/v1
 COSYVOICE_MODEL=cosyvoice-v3-flash

apps/api/app/__init__.py

@@ -1 +0,0 @@
-"""API application package."""

apps/api/app/api/__init__.py

@@ -1 +0,0 @@
-"""API package."""

apps/api/app/api/router.py

@@ -13,6 +13,7 @@ from app.api.routes.generated_videos import router as generated_videos_router
 from app.api.routes.generation_tasks import router as generation_tasks_router
 from app.api.routes.health import router as health_check_router
 from app.api.routes.ingest_jobs import router as ingest_jobs_router
+from app.api.routes.internal_render import router as internal_render_router
 from app.api.routes.jobs import router as jobs_router
 from app.api.routes.projects import router as projects_router
 from app.api.routes.recipes import router as recipes_router
@@ -156,3 +157,7 @@ api_router.include_router(
     feature_flags_router,
     tags=["Internal"],
 )
+api_router.include_router(
+    internal_render_router,
+    tags=["Internal"],
+)

apps/api/app/api/routes/_helpers.py

@@ -0,0 +1,48 @@
+"""路由层共享辅助函数 — 消除跨文件重复定义。"""
+
+from typing import Any
+
+from fastapi import HTTPException, status
+
+from packages.application import GetProjectUseCase
+from packages.ports.user_repository import UserRepository
+
+
+def check_project_access(project_id: str, user_id: str, project_repository) -> None:
+    """检查用户是否有项目访问权限。
+
+    合并自 asset_libraries.py / edit_plans.py 的同名函数。
+    - 空 project_id 直接放行（兼容 edit_plans 中 project_id 可选的场景）
+    - 错误信息使用中文，与项目其他路由保持一致
+    """
+    if not project_id or not project_id.strip():
+        return
+    project = project_repository.find_by_id(project_id)
+    if project is None:
+        raise HTTPException(status_code=404, detail="项目不存在")
+    if not project.can_access(user_id):
+        raise HTTPException(status_code=403, detail="无权访问该项目")
+
+
+def get_user_plan(user_id: str, user_repository: UserRepository) -> str:
+    """获取用户的订阅计划名称。"""
+    user = user_repository.find_by_id(user_id)
+    if user is None:
+        return "free"
+    return getattr(user, "subscription_plan", "free") or "free"
+
+
+def require_project_and_library(
+    project_id: str,
+    library_id: str,
+    project_repository: Any,
+    asset_library_repository: Any,
+) -> None:
+    """Verify project and asset library exist."""
+    project = GetProjectUseCase(project_repository).execute(project_id)
+    if project is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
+
+    libraries = asset_library_repository.find_by_project(project_id)
+    if not any(item.id == library_id for item in libraries):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Asset library not found")

apps/api/app/api/routes/asset_libraries.py

@@ -22,18 +22,11 @@ from packages.application import (
 )
 from packages.domain import AssetLibrary, AssetLibraryKind
 
+from ._helpers import check_project_access
+
 router = APIRouter()
 
 
-def _check_project_access(project_id: str, user_id: str, project_repository) -> None:
-    """检查用户是否有项目访问权限"""
-    project = project_repository.find_by_id(project_id)
-    if project is None:
-        raise HTTPException(status_code=404, detail=f"Project {project_id} not found")
-    if not project.can_access(user_id):
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied to project")
-
-
 def _to_asset_library_response(item) -> AssetLibraryResponse:
     return AssetLibraryResponse(
         id=item.id,
@@ -168,7 +161,7 @@ def delete_asset_library(
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="素材库不存在")
 
     # 权限校验：检查用户是否有项目访问权限
-    _check_project_access(library.project_id, authenticated_user.user.id, project_repository)
+    check_project_access(library.project_id, authenticated_user.user.id, project_repository)
 
     # 删除库内所有素材（无 FK 级联，需手动清理）
     assets_in_library = asset_repository.find_by_library(library_id)

apps/api/app/api/routes/assets.py

@@ -27,6 +27,8 @@ from packages.application import (
 )
 from packages.domain import AssetStatus, ClassificationStatus
 
+from app.api.routes._helpers import check_project_access
+
 logger = logging.getLogger(__name__)
 
 router = APIRouter()
@@ -72,14 +74,6 @@ def _to_asset_response(item, storage_service=None) -> AssetResponse:
     )
 
 
-def _check_project_access(project_id: str, user_id: str, project_repository) -> None:
-    """检查用户是否有项目访问权限"""
-    project = project_repository.find_by_id(project_id)
-    if project is None:
-        raise HTTPException(status_code=404, detail=f"Project {project_id} not found")
-    if not project.can_access(user_id):
-        raise HTTPException(status_code=403, detail="Access denied to project")
-
 
 @router.get("", response_model=ListAssetsResponse)
 def list_assets(
@@ -136,7 +130,7 @@ def list_assets(
             library = asset_library_repository.get(library_id)
             if library is None:
                 raise HTTPException(status_code=404, detail=f"AssetLibrary {library_id} not found")
-            _check_project_access(library.project_id, user_id, project_repository)
+            check_project_access(library.project_id, user_id, project_repository)
             if ft:
                 items = asset_repository.find_by_library_and_file_type(library_id, ft, skip=skip, limit=limit)
                 total = asset_repository.count_by_project(library.project_id) if not kind else len(items)
@@ -152,7 +146,7 @@ def list_assets(
 
         # 模式2：指定 project_id
         if project_id:
-            _check_project_access(project_id, user_id, project_repository)
+            check_project_access(project_id, user_id, project_repository)
             if ft:
                 # 无直接方法，加载后按 file_type 过滤（仍比全量加载好）
                 all_items = asset_repository.find_by_project(project_id)
@@ -210,13 +204,13 @@ def list_assets(
         library = asset_library_repository.get(library_id)
         if library is None:
             raise HTTPException(status_code=404, detail=f"AssetLibrary {library_id} not found")
-        _check_project_access(library.project_id, user_id, project_repository)
+        check_project_access(library.project_id, user_id, project_repository)
         if kind:
             all_items = asset_repository.find_by_library_and_file_type(library_id, kind_to_file_type[kind])
         else:
             all_items = asset_repository.find_by_library(library_id)
     elif project_id:
-        _check_project_access(project_id, user_id, project_repository)
+        check_project_access(project_id, user_id, project_repository)
         all_items = asset_repository.find_by_project(project_id)
     else:
         try:
@@ -262,7 +256,7 @@ def update_asset_review_status(
     item = asset_repository.get(asset_id)
     if item is None:
         raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
-    _check_project_access(item.project_id, authenticated_user.user.id, project_repository)
+    check_project_access(item.project_id, authenticated_user.user.id, project_repository)
     _apply_asset_review_status(item, request.review_status)
     updated = asset_repository.update(item)
     return _to_asset_response(updated)
@@ -286,7 +280,7 @@ def batch_delete_assets(
             failed_ids.append(asset_id)
             continue
         try:
-            _check_project_access(item.project_id, user_id, project_repository)
+            check_project_access(item.project_id, user_id, project_repository)
             deleted_ids.append(asset_id)
         except HTTPException:
             failed_ids.append(asset_id)
@@ -307,7 +301,7 @@ def get_asset(
     item = asset_repository.find_by_id(asset_id)
     if item is None:
         raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
-    _check_project_access(item.project_id, authenticated_user.user.id, project_repository)
+    check_project_access(item.project_id, authenticated_user.user.id, project_repository)
     return _to_asset_response(item)
 
 
@@ -322,7 +316,7 @@ def update_asset(
     item = asset_repository.find_by_id(asset_id)
     if item is None:
         raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
-    _check_project_access(item.project_id, authenticated_user.user.id, project_repository)
+    check_project_access(item.project_id, authenticated_user.user.id, project_repository)
 
     # 合并可修改字段
     if request.name is not None:
@@ -346,7 +340,7 @@ def delete_asset(
     item = asset_repository.find_by_id(asset_id)
     if item is None:
         raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
-    _check_project_access(item.project_id, authenticated_user.user.id, project_repository)
+    check_project_access(item.project_id, authenticated_user.user.id, project_repository)
     asset_repository.delete(asset_id)
 
 
@@ -363,7 +357,7 @@ def tag_asset(
     item = asset_repository.find_by_id(asset_id)
     if item is None:
         raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
-    _check_project_access(item.project_id, authenticated_user.user.id, project_repository)
+    check_project_access(item.project_id, authenticated_user.user.id, project_repository)
     for tag_id in request.tag_ids:
         tag = tag_repository.get(tag_id)
         if tag is None:
@@ -387,7 +381,7 @@ def untag_asset(
     item = asset_repository.find_by_id(asset_id)
     if item is None:
         raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
-    _check_project_access(item.project_id, authenticated_user.user.id, project_repository)
+    check_project_access(item.project_id, authenticated_user.user.id, project_repository)
     item.remove_tag(tag_id)
     asset_repository.update(item)
 

apps/api/app/api/routes/auth.py

@@ -206,7 +206,7 @@ async def verify_email_post(
     return _verify_email_token(request.token, user_repository)
 
 
-@router.post("/password/forgot", response_model=MessageResponse, status_code=status.HTTP_202_ACCEPTED)
+@router.post("/forgot-password", response_model=MessageResponse, status_code=status.HTTP_202_ACCEPTED)
 async def forgot_password(
     request: PasswordResetRequestModel,
     user_repository: UserRepository = Depends(get_user_repository),
@@ -223,7 +223,7 @@ async def forgot_password(
     return MessageResponse(message="如果账户存在，密码重置邮件已发送")
 
 
-@router.post("/password/reset", response_model=MessageResponse)
+@router.post("/reset-password", response_model=MessageResponse)
 async def reset_password(
     request: ResetPasswordModel,
     user_repository: UserRepository = Depends(get_user_repository),
@@ -243,7 +243,6 @@ async def logout(
     current_user: AuthenticatedUser = Depends(get_current_user),
 ):
     """登出 - 将当前 token 加入黑名单"""
-    from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 
     if credentials:
         try:

apps/api/app/api/routes/chunked_upload.py

@@ -14,7 +14,6 @@ from typing import Any
 from uuid import uuid4
 
 from app.auth import AuthenticatedUser, get_current_user
-from app.config import get_settings
 from app.core.celery_app import celery_app
 from app.core.storage import OSSStorageService, get_storage_service
 from app.dependencies import (
@@ -35,6 +34,8 @@ from fastapi.params import File
 
 from packages.application import GetProjectUseCase, SubmitIngestJobCommand, SubmitIngestJobUseCase
 
+from app.api.routes._helpers import require_project_and_library
+
 router = APIRouter()
 logger = logging.getLogger(__name__)
 
@@ -113,22 +114,6 @@ def _atomic_check_and_record(upload_id: str, chunk_index: int) -> bool:
             fcntl.flock(f.fileno(), fcntl.LOCK_UN)
 
 
-def _require_project_and_library(
-    project_id: str,
-    library_id: str,
-    project_repository: Any,
-    asset_library_repository: Any,
-) -> None:
-    """Verify project and asset library exist"""
-    project = GetProjectUseCase(project_repository).execute(project_id)
-    if project is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
-
-    libraries = asset_library_repository.find_by_project(project_id)
-    if not any(item.id == library_id for item in libraries):
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Asset library not found")
-
-
 def _load_upload_meta(upload_id: str) -> dict[str, Any]:
     """Load upload metadata"""
     meta_path = _get_upload_meta_path(upload_id)
@@ -206,7 +191,6 @@ async def init_chunked_upload(
     asset_library_repository: Any = Depends(get_asset_library_repository),
 ) -> ChunkedUploadInitResponse:
     """Initialize chunked upload"""
-    settings = get_settings()
 
     # Validate file size
     if request.file_size > MAX_FILE_SIZE:
@@ -221,7 +205,7 @@ async def init_chunked_upload(
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
 
     # Verify asset library
-    _require_project_and_library(
+    require_project_and_library(
         request.project_id,
         request.library_id,
         project_repository,

apps/api/app/api/routes/edit_plans.py

@@ -32,12 +32,6 @@ from fastapi import APIRouter, Depends, HTTPException, Query, status
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 
-from packages.adapters.sqlalchemy_impl.asset_library_repository import (
-    SQLAlchemyAssetLibraryRepository,
-)
-from packages.adapters.sqlalchemy_impl.asset_repository import (
-    SQLAlchemyAssetRepository,
-)
 from packages.adapters.sqlalchemy_impl.generation_task_repository import (
     SQLAlchemyGenerationTaskRepository,
 )
@@ -51,6 +45,8 @@ from packages.application.generation_tasks import (
     CreateGenerationTaskCommand,
     CreateGenerationTaskUseCase,
 )
+
+from ._helpers import check_project_access
 from packages.domain.config_schemas import normalize_plan_config
 from packages.domain.edit_plan import EditPlan, EditPlanStatus
 
@@ -247,17 +243,6 @@ class GenerateFromTemplateResponse(BaseModel):
 # ── Helpers ───────────────────────────────────────────────────────────────────
 
 
-def _check_project_access(project_id: str, user_id: str, project_repository: Any) -> None:
-    """校验用户对项目的访问权限（参照 assets.py 的 can_access 模式）"""
-    if not project_id or not project_id.strip():
-        return
-    project = project_repository.find_by_id(project_id)
-    if project is None:
-        raise HTTPException(status_code=404, detail="项目不存在")
-    if not project.can_access(user_id):
-        raise HTTPException(status_code=403, detail="无权访问该项目")
-
-
 def _to_response(p: EditPlan) -> EditPlanResponse:
     return EditPlanResponse(
         id=p.id,
@@ -311,7 +296,7 @@ def list_plans(
 
     # 项目鉴权：如果指定了 project_id，校验用户是否有权访问
     if project_id:
-        _check_project_access(project_id, current_user.user.id, project_repository)
+        check_project_access(project_id, current_user.user.id, project_repository)
 
     skip = (page - 1) * page_size
     plans = svc.list_plans(
@@ -353,7 +338,7 @@ def get_plan(
         )
     # 项目鉴权
     if plan.project_id:
-        _check_project_access(plan.project_id, current_user.user.id, project_repository)
+        check_project_access(plan.project_id, current_user.user.id, project_repository)
     return _to_response(plan)
 
 
@@ -369,7 +354,7 @@ def create_plan(
     project_id = (body.project_id or "").strip()
     # 项目鉴权
     if project_id:
-        _check_project_access(project_id, current_user.user.id, project_repository)
+        check_project_access(project_id, current_user.user.id, project_repository)
     svc = EditPlanService(db)
     # 标准化 config，填充 cover/title/subtitle/bgm 默认值
     normalized_config = normalize_plan_config(body.config)
@@ -411,7 +396,7 @@ def update_plan(
     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)
+        check_project_access(existing.project_id, current_user.user.id, project_repository)
 
     # 基础字段更新
     try:
@@ -465,7 +450,7 @@ def delete_plan(
     # 项目鉴权
     existing = svc.get_plan(plan_id)
     if existing and existing.project_id:
-        _check_project_access(existing.project_id, current_user.user.id, project_repository)
+        check_project_access(existing.project_id, current_user.user.id, project_repository)
     deleted = svc.delete_plan(plan_id)
     if not deleted:
         raise HTTPException(
@@ -507,7 +492,7 @@ def generate_plan(
     if plan_check is None:
         raise HTTPException(status_code=404, detail=f"剪辑计划不存在: {plan_id}")
     if plan_check.project_id:
-        _check_project_access(plan_check.project_id, current_user.user.id, project_repository)
+        check_project_access(plan_check.project_id, current_user.user.id, project_repository)
 
     # ── 自动兜底 1: draft → editing ──────────────────────────────────────
     if plan_check.status == EditPlanStatus.DRAFT:
@@ -710,7 +695,7 @@ def generate_plan(
     except HTTPException:
         # 已处理的 HTTP 异常直接透传
         raise
-    except Exception as exc:
+    except Exception:
         logger.exception("触发剪辑计划生成失败: plan_id=%s", plan_id)
         # 尝试将计划标记为失败（RENDERING → FAILED 是合法的状态流转）
         try:
@@ -749,7 +734,7 @@ def get_generation_status(
     plan = gen_status["plan"]
     # 项目鉴权
     if plan.project_id:
-        _check_project_access(plan.project_id, current_user.user.id, project_repository)
+        check_project_access(plan.project_id, current_user.user.id, project_repository)
     clips = gen_status["clips"]
 
     clip_items = [
@@ -791,7 +776,7 @@ def list_plan_generations(
     # 验证计划存在 + 项目鉴权
     plan = svc.get_plan_or_raise(plan_id)
     if plan.project_id:
-        _check_project_access(plan.project_id, current_user.user.id, project_repository)
+        check_project_access(plan.project_id, current_user.user.id, project_repository)
 
     gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
     tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
@@ -860,7 +845,7 @@ def ai_recommend_clips(
 
     # 项目鉴权
     if plan.project_id:
-        _check_project_access(plan.project_id, current_user.user.id, project_repository)
+        check_project_access(plan.project_id, current_user.user.id, project_repository)
 
     # 验证状态：只允许 draft 或 editing
     plan_status = plan.status.value if hasattr(plan.status, "value") else plan.status
@@ -909,7 +894,7 @@ def ai_recommend_clips(
             config=normalized_config,
             total_duration=result["total_duration"],
         )
-    except Exception as exc:
+    except Exception:
         logger.exception("AI 推荐写入失败，plan_id=%s 数据可能不一致", plan_id)
         # 尝试回滚未提交的变更
         try:
@@ -990,7 +975,7 @@ def generate_cover(
 
     # 项目鉴权
     if plan.project_id:
-        _check_project_access(plan.project_id, current_user.user.id, project_repository)
+        check_project_access(plan.project_id, current_user.user.id, project_repository)
 
     # 调用 AI 封面生成服务
     from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover
@@ -1110,7 +1095,7 @@ def get_plan_timeline(
     plan = svc.get_plan_or_raise(plan_id)
     # 项目鉴权
     if plan.project_id:
-        _check_project_access(plan.project_id, current_user.user.id, project_repository)
+        check_project_access(plan.project_id, current_user.user.id, project_repository)
 
     clips = svc.list_clips(plan_id=plan_id, skip=0, limit=200)
     # 按 order 排序
@@ -1171,7 +1156,7 @@ def generate_from_template(
 
     # 项目鉴权
     if body.project_id:
-        _check_project_access(body.project_id, current_user.user.id, project_repository)
+        check_project_access(body.project_id, current_user.user.id, project_repository)
 
     template_svc = EditTemplateService(db)
 

apps/api/app/api/routes/generation_tasks.py

@@ -32,6 +32,8 @@ from app.schemas.generation_task import (
 )
 from fastapi import APIRouter, Depends, HTTPException
 
+from app.api.routes._helpers import check_project_access
+
 from packages.application import (
     CreateGenerationTaskCommand,
     CreateGenerationTaskUseCase,
@@ -43,15 +45,6 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter()
 
-def _check_project_access(project_id: str, user_id: str, project_repository) -> None:
-    """检查用户是否有项目访问权限"""
-    project = project_repository.find_by_id(project_id)
-    if project is None:
-        raise HTTPException(status_code=404, detail=f"Project {project_id} not found")
-    if not project.can_access(user_id):
-        raise HTTPException(status_code=403, detail="Access denied to project")
-
-
 def _to_generation_task_response(task) -> GenerationTaskResponse:
     return GenerationTaskResponse(
         id=task.id,
@@ -339,7 +332,7 @@ def get_generation_task(
     if task is None:
         raise HTTPException(status_code=404, detail=f"GenerationTask {task_id} not found")
     if task.project_id:
-        _check_project_access(task.project_id, authenticated_user.user.id, project_repository)
+        check_project_access(task.project_id, authenticated_user.user.id, project_repository)
     return _to_generation_task_response(task)
 
 
@@ -356,7 +349,7 @@ def list_generation_results(
     if task is None:
         raise HTTPException(status_code=404, detail=f"GenerationTask {task_id} not found")
     if task.project_id:
-        _check_project_access(task.project_id, authenticated_user.user.id, project_repository)
+        check_project_access(task.project_id, authenticated_user.user.id, project_repository)
     use_case = ListGeneratedVideosByTaskUseCase(generated_video_repository)
     items = use_case.execute(task_id)
     responses = []

apps/api/app/api/routes/internal_render.py

@@ -0,0 +1,120 @@
+"""渲染结果内部下载接口。
+
+通过内部 API Key 鉴权，为灰度对比工具等内部系统提供渲染结果下载能力。
+
+API:
+  GET  /api/v1/internal/render/videos/{video_id}/download-url  - 获取单个视频下载URL
+  GET  /api/v1/internal/render/tasks/{task_id}/videos          - 获取任务下所有视频及下载URL
+
+鉴权：X-API-Key header，走内部 API Key 验证
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.api.routes.auth import _verify_internal_api_key
+from app.core.storage import OSSStorageService, get_storage_service
+from app.dependencies import get_generated_video_repository
+from fastapi import APIRouter, Depends, HTTPException, Query
+from pydantic import BaseModel
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/internal/render", tags=["Internal"])
+
+
+class InternalRenderVideoItem(BaseModel):
+    """内部渲染视频项。"""
+
+    video_id: str
+    generation_task_id: str
+    project_id: str
+    name: str
+    file_url: str
+    file_size: int | None = None
+    duration: float | None = None
+    width: int | None = None
+    height: int | None = None
+    fps: float | None = None
+    status: str
+    download_url: str
+
+
+class InternalRenderTaskVideosResponse(BaseModel):
+    """任务下所有渲染视频响应。"""
+
+    task_id: str
+    count: int
+    videos: list[InternalRenderVideoItem]
+
+
+class InternalRenderDownloadUrlResponse(BaseModel):
+    """单个视频下载URL响应。"""
+
+    video_id: str
+    download_url: str
+
+
+def _video_to_item(video: Any, download_url: str) -> InternalRenderVideoItem:
+    """将 GeneratedVideo 领域对象转为响应项。"""
+    return InternalRenderVideoItem(
+        video_id=video.id,
+        generation_task_id=video.generation_task_id,
+        project_id=video.project_id,
+        name=video.name,
+        file_url=video.file_url,
+        file_size=getattr(video, "file_size", None),
+        duration=getattr(video, "duration", None),
+        width=getattr(video, "width", None),
+        height=getattr(video, "height", None),
+        fps=getattr(video, "fps", None),
+        status=video.status,
+        download_url=download_url,
+    )
+
+
+@router.get("/videos/{video_id}/download-url", response_model=InternalRenderDownloadUrlResponse)
+def get_render_video_download_url(
+    video_id: str,
+    _: bool = Depends(_verify_internal_api_key),
+    generated_video_repository: Any = Depends(get_generated_video_repository),
+    storage_service: OSSStorageService = Depends(get_storage_service),
+) -> InternalRenderDownloadUrlResponse:
+    """获取单个渲染视频的下载URL（预签名）。"""
+    video = generated_video_repository.get(video_id)
+    if video is None:
+        raise HTTPException(status_code=404, detail=f"GeneratedVideo {video_id} not found")
+
+    download_url = storage_service.get_download_url(video.file_url, expires_seconds=86400)
+    logger.info("内部渲染下载URL生成: video_id=%s", video_id)
+    return InternalRenderDownloadUrlResponse(video_id=video_id, download_url=download_url)
+
+
+@router.get("/tasks/{task_id}/videos", response_model=InternalRenderTaskVideosResponse)
+def get_render_task_videos(
+    task_id: str,
+    status: str | None = Query(None, description="按状态筛选，如 completed/failed"),
+    _: bool = Depends(_verify_internal_api_key),
+    generated_video_repository: Any = Depends(get_generated_video_repository),
+    storage_service: OSSStorageService = Depends(get_storage_service),
+) -> InternalRenderTaskVideosResponse:
+    """获取生成任务下所有渲染视频及下载URL。"""
+    videos = generated_video_repository.list_by_generation_task(task_id)
+
+    # 状态筛选
+    if status:
+        videos = [v for v in videos if v.status == status]
+
+    items = []
+    for video in videos:
+        download_url = storage_service.get_download_url(video.file_url, expires_seconds=86400)
+        items.append(_video_to_item(video, download_url))
+
+    logger.info("内部渲染任务视频查询: task_id=%s count=%d", task_id, len(items))
+    return InternalRenderTaskVideosResponse(
+        task_id=task_id,
+        count=len(items),
+        videos=items,
+    )

apps/api/app/api/routes/jobs.py

@@ -20,7 +20,7 @@ from typing import Any
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.celery_app import celery_app
-from app.dependencies import get_db_session, get_job_repository, get_project_repository
+from app.dependencies import get_job_repository, get_project_repository
 from app.schemas.job import (
     CompleteJobRequest,
     CreateJobRequest,
@@ -51,6 +51,8 @@ from packages.application.jobs import (
 )
 from packages.domain.job import JobType
 
+from app.api.routes._helpers import check_project_access
+
 logger = logging.getLogger(__name__)
 
 router = APIRouter()
@@ -66,15 +68,6 @@ _JOB_TYPE_TO_CELERY_TASK: dict[str, str] = {
 }
 
 
-def _check_project_access(project_id: str, user_id: str, project_repository) -> None:
-    """检查用户是否有项目访问权限。"""
-    project = project_repository.find_by_id(project_id)
-    if project is None:
-        raise HTTPException(status_code=404, detail=f"Project {project_id} not found")
-    if not project.can_access(user_id):
-        raise HTTPException(status_code=403, detail="Access denied to project")
-
-
 # ── 创建任务 ──────────────────────────────────────────────────────────────────
 
 
@@ -89,7 +82,7 @@ def create_job(
 
     创建后任务处于 pending 状态，需要调用 /submit 提交执行。
     """
-    _check_project_access(request.project_id, authenticated_user.user.id, project_repository)
+    check_project_access(request.project_id, authenticated_user.user.id, project_repository)
 
     # 校验 job_type
     try:
@@ -182,7 +175,7 @@ def list_project_jobs(
     offset: int = Query(default=0, ge=0),
 ) -> ListJobsResponse:
     """获取项目下的任务列表。"""
-    _check_project_access(project_id, authenticated_user.user.id, project_repository)
+    check_project_access(project_id, authenticated_user.user.id, project_repository)
 
     use_case = ListJobsUseCase(job_repo)
     jobs = use_case.execute(
@@ -204,7 +197,7 @@ def get_job_statistics(
     project_repository: Any = Depends(get_project_repository),
 ) -> JobStatisticsResponse:
     """获取项目任务统计摘要。"""
-    _check_project_access(project_id, authenticated_user.user.id, project_repository)
+    check_project_access(project_id, authenticated_user.user.id, project_repository)
 
     use_case = GetJobStatisticsUseCase(job_repo)
     stats = use_case.execute(project_id)

apps/api/app/api/routes/recipes.py

@@ -33,6 +33,8 @@ from packages.application.recipe.use_cases import (
 )
 from packages.ports.user_repository import UserRepository
 
+from app.api.routes._helpers import get_user_plan
+
 router = APIRouter()
 
 
@@ -40,13 +42,6 @@ def _get_recipe_repository(session: Session = Depends(get_db_session)) -> SQLAlc
     return SQLAlchemyRecipeRepository(session)
 
 
-def _get_user_plan(user_id: str, user_repository: UserRepository) -> str:
-    user = user_repository.find_by_id(user_id)
-    if user is None:
-        return "free"
-    return getattr(user, "subscription_plan", "free") or "free"
-
-
 def _item_to_response(item) -> RecipeItemResponse:
     return RecipeItemResponse(
         id=item.id,
@@ -194,7 +189,7 @@ def use_recipe(
     user_repository: UserRepository = Depends(get_user_repository),
 ) -> UseRecipeResponse:
     user_id = authenticated_user.user.id
-    plan_name = _get_user_plan(user_id, user_repository)
+    plan_name = get_user_plan(user_id, user_repository)
     use_case = UseRecipeUseCase(recipe_repository)
     try:
         result = use_case.execute(recipe_id, user_id, user_plan=plan_name)

apps/api/app/api/routes/subscription.py

@@ -232,7 +232,7 @@ async def payment_callback(
 
         # 创建账单记录
         record_id = uuid.uuid4().hex
-        record = repo.create(
+        repo.create(
             {
                 "id": record_id,
                 "user_id": user_id,

apps/api/app/api/routes/titles.py

@@ -28,6 +28,8 @@ from packages.application.title_library.use_cases import (
 )
 from packages.ports.user_repository import UserRepository
 
+from app.api.routes._helpers import get_user_plan
+
 router = APIRouter()
 
 
@@ -51,13 +53,6 @@ def _to_response(item) -> TitleLibraryItemResponse:
     )
 
 
-def _get_user_plan(user_id: str, user_repository: UserRepository) -> str:
-    user = user_repository.find_by_id(user_id)
-    if user is None:
-        return "free"
-    return getattr(user, "subscription_plan", "free") or "free"
-
-
 @router.get("", response_model=ListTitleLibraryResponse)
 def list_titles(
     category: Optional[str] = Query(None),
@@ -98,7 +93,7 @@ def create_title(
     user_repository: UserRepository = Depends(get_user_repository),
 ) -> TitleLibraryItemResponse:
     user_id = authenticated_user.user.id
-    plan_name = _get_user_plan(user_id, user_repository)
+    plan_name = get_user_plan(user_id, user_repository)
     command = CreateTitleLibraryCommand(
         user_id=user_id,
         name=request.name,

apps/api/app/api/routes/upload.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Annotated, Any
+from typing import Any
 from uuid import uuid4
 
 from app.auth import AuthenticatedUser, get_current_user
@@ -17,12 +17,13 @@ from app.schemas.upload import (
     DirectUploadCompleteResponse,
     DirectUploadPrepareRequest,
     DirectUploadPrepareResponse,
-    UploadAssetRequest,
     UploadAssetResponse,
 )
 from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
 
-from packages.application import GetProjectUseCase, SubmitIngestJobCommand, SubmitIngestJobUseCase
+from packages.application import SubmitIngestJobCommand, SubmitIngestJobUseCase
+
+from app.api.routes._helpers import require_project_and_library
 
 logger = logging.getLogger(__name__)
 
@@ -80,21 +81,6 @@ def _validate_mime_type(content_type: str | None) -> str:
     return base_type
 
 
-def _require_project_and_library(
-    project_id: str,
-    library_id: str,
-    project_repository: Any,
-    asset_library_repository: Any,
-) -> None:
-    project = GetProjectUseCase(project_repository).execute(project_id)
-    if project is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
-
-    libraries = asset_library_repository.find_by_project(project_id)
-    if not any(item.id == library_id for item in libraries):
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Asset library not found")
-
-
 def _submit_ingest_job(
     project_id: str,
     library_id: str,
@@ -135,7 +121,7 @@ async def prepare_direct_upload(
     # P2-5: 服务端验证 MIME 类型
     validated_content_type = _validate_mime_type(request.content_type)
 
-    _require_project_and_library(
+    require_project_and_library(
         request.project_id,
         request.library_id,
         project_repository,
@@ -183,7 +169,7 @@ async def complete_direct_upload(
     storage_service: OSSStorageService = Depends(get_storage_service),
 ) -> DirectUploadCompleteResponse:
     """确认浏览器直传完成并创建导入任务。"""
-    _require_project_and_library(
+    require_project_and_library(
         request.project_id,
         request.library_id,
         project_repository,
@@ -252,7 +238,7 @@ async def upload_asset(
     storage_service: OSSStorageService = Depends(get_storage_service),
 ) -> UploadAssetResponse:
     """上传素材文件并触发导入流水线。"""
-    _require_project_and_library(project_id, library_id, project_repository, asset_library_repository)
+    require_project_and_library(project_id, library_id, project_repository, asset_library_repository)
 
     # ── 素材去重检测：上传前检查同素材库 + 同 file_hash ──
     if file_hash:

apps/api/app/api/routes/voice_clones.py

@@ -28,7 +28,6 @@ from packages.application.voice_clone.use_cases import (
     VoiceCloneNotRetryableError,
 )
 from packages.application.voice_clone.workflow import (
-    VoiceCloneWorkflowError,
     VoiceCloneWorkflowService,
 )
 

apps/api/app/api/routes/voices.py

@@ -39,6 +39,8 @@ from packages.application.voice_library.use_cases import (
 from packages.domain.preset_voices import PRESET_VOICES
 from packages.ports.user_repository import UserRepository
 
+from app.api.routes._helpers import get_user_plan
+
 router = APIRouter()
 
 
@@ -125,13 +127,6 @@ def _preset_to_unified_response(preset) -> UnifiedVoiceItemResponse:
     )
 
 
-def _get_user_plan(user_id: str, user_repository: UserRepository) -> str:
-    user = user_repository.find_by_id(user_id)
-    if user is None:
-        return "free"
-    return getattr(user, "subscription_plan", "free") or "free"
-
-
 # ==================== 统一配音列表（预置 + 克隆）====================
 
 
@@ -271,7 +266,7 @@ def create_voice(
     sign_url=Depends(get_audio_url_signer),
 ) -> VoiceLibraryItemResponse:
     user_id = authenticated_user.user.id
-    plan_name = _get_user_plan(user_id, user_repository)
+    plan_name = get_user_plan(user_id, user_repository)
     command = CreateVoiceLibraryCommand(
         user_id=user_id,
         name=request.name,

apps/api/app/config.py

@@ -25,7 +25,7 @@ class Settings(BaseSettings):
     DATABASE_POOL_SIZE: int = 20
     DATABASE_MAX_OVERFLOW: int = 10  # 调整为合理值：pool_size(20) + max_overflow(10) = 最大30连接
     DATABASE_POOL_TIMEOUT: int = 30
-    DATABASE_POOL_RECYLE: int = 3600
+    DATABASE_POOL_RECYCLE: int = 3600
     USE_IN_MEMORY_DB: bool = False
     AUTO_CREATE_SCHEMA: bool = False
 
@@ -41,6 +41,11 @@ class Settings(BaseSettings):
     # 密钥轮换天数（到达此天数后建议更换密钥）
     SECRET_ROTATION_DAYS: int = 90
 
+    # JWT 算法与过期时间（与 .env.example 对齐）
+    JWT_ALGORITHM: str = "HS256"
+    JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
+    JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 30
+
     @field_validator("JWT_SECRET_KEY", mode="before")
     @classmethod
     def validate_jwt_secret_key(cls, v):

apps/api/app/core/__init__.py

@@ -1 +0,0 @@
-"""Core configuration package."""

apps/api/app/dependencies.py

@@ -50,20 +50,8 @@ from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
 from packages.adapters.sqlalchemy_impl.voice_library_repository import (
     SQLAlchemyVoiceLibraryRepository,
 )
-from packages.ports.asset_library_repository import AssetLibraryRepository
-from packages.ports.asset_repository import AssetRepository
-from packages.ports.classification_job_repository import ClassificationJobRepository
-from packages.ports.duplication_repository import DuplicationRecordRepository
-from packages.ports.generated_video_repository import GeneratedVideoRepository
-from packages.ports.generation_task_repository import GenerationTaskRepository
-from packages.ports.ingest_job_repository import IngestJobRepository
-from packages.ports.job_repository import JobRepository
-from packages.ports.project_repository import ProjectRepository
 from packages.ports.tag_repository import TagRepository
-from packages.ports.title_library_repository import TitleLibraryRepository
 from packages.ports.user_repository import UserRepository
-from packages.ports.voice_clone_profile_repository import VoiceCloneProfileRepository
-from packages.ports.voice_library_repository import VoiceLibraryRepository
 
 _engine, _SessionLocal = build_session_factory(settings.DATABASE_URL)
 

apps/api/app/middleware/auth.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 from app.auth import AuthenticatedUser
 from app.auth import get_current_user as get_authenticated_user
 from app.dependencies import get_user_repository
-from fastapi import Depends
+from fastapi import Depends, HTTPException
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 
 from packages.domain.entities import User

apps/api/app/middleware/monitoring.py

@@ -6,7 +6,7 @@ import logging
 import time
 from typing import Callable
 
-from fastapi import Request, Response
+from fastapi import Request
 from starlette.middleware.base import BaseHTTPMiddleware
 
 logger = logging.getLogger(__name__)

apps/api/app/schemas/subscription.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import List, Optional
+from typing import Optional
 
 from pydantic import BaseModel, Field
 

apps/api/app/services/auto_clip_service.py

@@ -24,7 +24,7 @@ from packages.adapters.sqlalchemy_impl import (
 )
 from packages.domain.asset import AssetType
 from packages.domain.classification import AssetClassification
-from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
+from packages.domain.edit_plan_clip import EditPlanClip
 
 logger = logging.getLogger(__name__)
 

apps/api/app/services/edit_plan_service.py

@@ -18,7 +18,6 @@ from packages.adapters.sqlalchemy_impl import (
 )
 from packages.domain.edit_plan import EditPlan, EditPlanStatus
 from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
-from packages.domain.generation_task import GenerationTaskStatus
 
 logger = logging.getLogger(__name__)
 

apps/api/app/services/job_service.py

@@ -13,7 +13,6 @@ from __future__ import annotations
 import logging
 from typing import Any
 
-from sqlalchemy.orm import Session
 
 from packages.application.jobs import (
     CancelJobUseCase,

apps/api/app/services/plan_generator_service.py

@@ -13,7 +13,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, List, Optional
+from typing import Any, List
 
 from sqlalchemy.orm import Session
 

apps/api/app/services/video_compose_service.py

@@ -16,7 +16,6 @@ FFmpeg 视频合成编排服务：
 from __future__ import annotations
 
 import logging
-import shutil
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -28,7 +27,7 @@ from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import (
 from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
     SQLAlchemyEditPlanRepository,
 )
-from packages.domain.edit_plan import EditPlan, EditPlanStatus
+from packages.domain.edit_plan import EditPlanStatus
 from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
 from packages.domain.template_clip_config import TransitionEffect
 

apps/web/src/api/dashboard.ts

@@ -1,42 +0,0 @@
-/**
- * 仪表盘 API
- * Phase 1 新增：用户仪表盘概览
- */
-import apiClient from "./client";
-
-/** 仪表盘概览数据 */
-export interface DashboardOverview {
-  /** 素材总数 */
-  total_assets: number;
-  /** 已用存储（字节） */
-  used_storage_bytes: number;
-  /** 总标题数 */
-  total_titles: number;
-  /** 总配音数 */
-  total_voices: number;
-  /** 生成任务总数 */
-  total_tasks: number;
-  /** 成品总数 */
-  total_products: number;
-  /** 最近生成任务 */
-  recent_tasks: Array<{
-    id: string;
-    task_type: string;
-    status: string;
-    progress: number;
-    user_message: string;
-    created_at: string;
-  }>;
-  /** 订阅信息 */
-  subscription: {
-    plan: "free" | "pro" | "enterprise";
-    status: "active" | "inactive" | "expired";
-    expires_at?: string;
-  };
-}
-
-/** 获取仪表盘概览数据 */
-export const getDashboardOverview = async (): Promise<DashboardOverview> => {
-  const response = await apiClient.get("/dashboard/overview");
-  return response.data;
-};

apps/web/src/components/business/business.css

@@ -1,365 +0,0 @@
-/* V21 业务组件统一样式 */
-
-/* ==================== 按钮 ==================== */
-.xx-primary-btn {
-  background: var(--gradient-primary) !important;
-  color: var(--text-inverse) !important;
-  border: none !important;
-  border-radius: var(--radius-md) !important;
-  padding: 10px 20px !important;
-  font-weight: var(--font-weight-bold) !important;
-  box-shadow: var(--shadow-primary) !important;
-  transition: var(--transition-all) !important;
-  cursor: pointer;
-  height: auto !important;
-}
-
-.xx-primary-btn:hover {
-  box-shadow: var(--shadow-hover) !important;
-  transform: translateY(-1px);
-}
-
-.xx-ghost-btn {
-  background: transparent !important;
-  color: var(--primary-color) !important;
-  border: 2px solid var(--primary-color) !important;
-  border-radius: var(--radius-md) !important;
-  padding: var(--space-sm) 18px !important;
-  font-weight: var(--font-weight-bold) !important;
-  transition: var(--transition-all) !important;
-  cursor: pointer;
-  height: auto !important;
-}
-
-.xx-ghost-btn:hover {
-  background: var(--primary-soft) !important;
-}
-
-/* ==================== 卡片 ==================== */
-.xx-card {
-  background: var(--bg-elevated);
-  border: 1px solid var(--border-color);
-  border-radius: var(--radius-xl);
-  box-shadow: var(--shadow-card);
-  padding: var(--space-lg);
-  margin-bottom: 20px;
-  transition: all var(--transition-slow);
-}
-
-.xx-card:hover {
-  box-shadow: var(--shadow-md);
-  transform: translateY(-2px);
-}
-
-/* ==================== 页面结构 ==================== */
-.xx-page {
-  max-width: 1200px;
-  margin: 0 auto;
-  padding: var(--space-lg);
-}
-
-.xx-page-head {
-  display: flex;
-  justify-content: space-between;
-  align-items: flex-start;
-  gap: 18px;
-  margin-bottom: 28px;
-}
-
-.xx-page-head h2 {
-  font-size: 26px;
-  font-weight: var(--font-weight-extrabold);
-  color: var(--text-primary);
-  margin: 0 0 var(--space-sm);
-}
-
-.xx-page-head p {
-  font-size: var(--font-size-base);
-  color: var(--text-secondary);
-  margin: 0;
-}
-
-/* ==================== 表格样式 ==================== */
-.xx-table-card {
-  background: var(--bg-elevated);
-  border: 1px solid var(--border-color);
-  border-radius: var(--radius-xl);
-  box-shadow: var(--shadow-card);
-  padding: 20px;
-  overflow: hidden;
-}
-
-/* 表格包装器 */
-.xx-table-wrapper {
-  border-radius: var(--radius-lg);
-  overflow: hidden;
-}
-
-/* ==================== 标签/Tag ==================== */
-.xx-tag {
-  padding: var(--space-xs) 12px;
-  border-radius: var(--radius-xs);
-  font-size: 13px;
-  font-weight: var(--font-weight-medium);
-}
-
-.xx-tag-indigo {
-  background: var(--primary-soft);
-  color: var(--primary-color);
-  border: 1px solid var(--color-primary-200);
-}
-
-.xx-tag-success {
-  background: var(--success-soft);
-  color: var(--color-secondary-500);
-  border: 1px solid var(--success-border);
-}
-
-.xx-tag-warning {
-  background: var(--warning-soft);
-  color: var(--accent-dark);
-  border: 1px solid var(--color-accent-200);
-}
-
-.xx-tag-error {
-  background: var(--error-soft);
-  color: var(--error-color);
-  border: 1px solid var(--error-border);
-}
-
-/* ==================== 搜索栏 ==================== */
-.xx-search-bar {
-  margin-bottom: 20px;
-}
-
-.xx-search-input {
-  width: 100%;
-  padding: 12px 18px;
-  border: 2px solid var(--border-color);
-  border-radius: var(--radius-md);
-  font-size: var(--font-size-base);
-  background: var(--bg-primary);
-  transition: var(--transition-all);
-  outline: none;
-}
-
-.xx-search-input:focus {
-  border-color: var(--primary-color);
-  box-shadow: 0 0 0 4px
-    color-mix(in srgb, var(--primary-color) 10%, transparent);
-}
-
-/* ==================== Modal ==================== */
-.xx-modal .ant-modal-content {
-  border-radius: var(--radius-xl);
-  padding: var(--space-lg);
-}
-
-.xx-modal .ant-modal-header {
-  border-radius: var(--radius-xl) var(--radius-xl) 0 0;
-  padding: 20px var(--space-lg);
-  border-bottom: 1px solid var(--border-color);
-}
-
-.xx-modal .ant-modal-title {
-  font-size: var(--font-size-lg);
-  font-weight: var(--font-weight-bold);
-  color: var(--text-primary);
-}
-
-.xx-modal .ant-modal-footer {
-  border-top: 1px solid var(--border-color);
-  padding: var(--space-md) var(--space-lg);
-}
-
-/* ==================== 空状态 ==================== */
-.xx-empty-state {
-  text-align: center;
-  padding: var(--space-3xl) var(--space-lg);
-  color: var(--text-secondary);
-}
-
-.xx-empty-state-icon {
-  font-size: 48px;
-  margin-bottom: var(--space-md);
-}
-
-/* ==================== 网格布局 ==================== */
-.xx-grid-2 {
-  display: grid;
-  grid-template-columns: repeat(2, 1fr);
-  gap: 20px;
-}
-
-.xx-grid-3 {
-  display: grid;
-  grid-template-columns: repeat(3, 1fr);
-  gap: 20px;
-}
-
-.xx-grid-4 {
-  display: grid;
-  grid-template-columns: repeat(4, 1fr);
-  gap: 20px;
-}
-
-@media (max-width: 768px) {
-  .xx-grid-2,
-  .xx-grid-3,
-  .xx-grid-4 {
-    grid-template-columns: 1fr;
-  }
-}
-
-/* ==================== 配额展示 ==================== */
-.xx-quota-item {
-  padding: 20px;
-  background: var(--bg-primary);
-  border: 1px solid var(--border-color);
-  border-radius: var(--radius-lg);
-  transition: var(--transition-all);
-}
-
-.xx-quota-item:hover {
-  border-color: var(--primary-color);
-  box-shadow: 0 8px 24px
-    color-mix(in srgb, var(--primary-color) 10%, transparent);
-}
-
-/* ==================== 进度条 ==================== */
-.xx-progress {
-  margin-top: 12px;
-}
-
-/* ==================== Ant Design 覆盖样式 ==================== */
-/* Table overrides */
-.ant-table-wrapper .ant-table-thead > tr > th {
-  background: var(--bg-secondary) !important;
-  font-weight: var(--font-weight-bold) !important;
-  color: var(--text-primary) !important;
-  border-bottom: 2px solid var(--border-color) !important;
-  padding: 14px var(--space-md) !important;
-}
-
-.ant-table-wrapper .ant-table-tbody > tr > td {
-  padding: 14px var(--space-md) !important;
-  border-bottom: 1px solid var(--color-gray-100) !important;
-}
-
-.ant-table-wrapper .ant-table-tbody > tr:hover > td {
-  background: var(--color-gray-50) !important;
-}
-
-/* Card overrides */
-.ant-card {
-  border-radius: var(--radius-xl) !important;
-  border: 1px solid var(--border-color) !important;
-}
-
-.ant-card-head {
-  border-bottom: 1px solid var(--border-color) !important;
-  min-height: 52px !important;
-  padding: 0 var(--space-lg) !important;
-}
-
-.ant-card-head-title {
-  font-weight: var(--font-weight-bold) !important;
-  font-size: var(--font-size-md) !important;
-  color: var(--text-primary) !important;
-}
-
-.ant-card-body {
-  padding: 20px var(--space-lg) !important;
-}
-
-/* Modal overrides */
-.ant-modal-content {
-  border-radius: var(--radius-xl) !important;
-  overflow: hidden;
-}
-
-.ant-modal-header {
-  padding: 20px var(--space-lg) !important;
-  background: var(--bg-primary) !important;
-}
-
-.ant-modal-title {
-  font-weight: var(--font-weight-bold) !important;
-  font-size: var(--font-size-lg) !important;
-  color: var(--text-primary) !important;
-}
-
-.ant-modal-body {
-  padding: var(--space-lg) !important;
-}
-
-.ant-modal-footer {
-  padding: var(--space-md) var(--space-lg) !important;
-}
-
-/* Button overrides */
-.ant-btn-primary {
-  background: var(--gradient-primary) !important;
-  border: none !important;
-  border-radius: var(--radius-md) !important;
-  box-shadow: var(--shadow-primary) !important;
-  height: auto !important;
-  padding: 10px 20px !important;
-  font-weight: var(--font-weight-bold) !important;
-}
-
-.ant-btn-primary:hover {
-  background: var(--gradient-primary) !important;
-  box-shadow: var(--shadow-hover) !important;
-  transform: translateY(-1px);
-}
-
-/* Tag overrides */
-.ant-tag {
-  border-radius: var(--radius-xs) !important;
-  padding: var(--space-xs) 12px !important;
-  font-weight: var(--font-weight-medium) !important;
-}
-
-/* Select overrides */
-.ant-select-selector {
-  border-radius: var(--radius-md) !important;
-  border-color: var(--border-color) !important;
-}
-
-.ant-select:not(.ant-select-disabled):hover .ant-select-selector {
-  border-color: var(--primary-color) !important;
-}
-
-.ant-select-focused .ant-select-selector {
-  border-color: var(--primary-color) !important;
-  box-shadow: 0 0 0 3px
-    color-mix(in srgb, var(--primary-color) 10%, transparent) !important;
-}
-
-/* Input overrides */
-.ant-input {
-  border-radius: var(--radius-md) !important;
-  border-color: var(--border-color) !important;
-  padding: 10px 14px !important;
-}
-
-.ant-input:hover {
-  border-color: var(--primary-color) !important;
-}
-
-.ant-input:focus {
-  border-color: var(--primary-color) !important;
-  box-shadow: 0 0 0 3px
-    color-mix(in srgb, var(--primary-color) 10%, transparent) !important;
-}
-
-/* Progress overrides */
-.ant-progress-inner {
-  background: var(--color-gray-100) !important;
-  border-radius: var(--radius-xs) !important;
-}
-
-.ant-progress-bg {
-  border-radius: var(--radius-xs) !important;
-}

apps/web/src/config/navigation.tsx

@@ -1,211 +0,0 @@
-/**
- * 统一导航配置
- * Header 和 Sidebar 共用此数据源
- */
-import React from "react";
-import {
-  DashboardOutlined,
-  VideoCameraOutlined,
-  FileOutlined,
-  AudioOutlined,
-  FileTextOutlined,
-  TrophyOutlined,
-  AppstoreOutlined,
-  HistoryOutlined,
-  ControlOutlined,
-  CrownOutlined,
-  ScanOutlined,
-  EditOutlined,
-  FolderOutlined,
-} from "@ant-design/icons";
-
-/** 导航项定义 */
-export interface NavItem {
-  key: string;
-  label: string;
-  path: string;
-  icon: React.ReactNode;
-}
-
-/** 导航分组定义 */
-export interface NavGroup {
-  title: string;
-  items: NavItem[];
-}
-
-/**
- * 扁平导航列表（Header 使用）
- */
-export const NAV_ITEMS: NavItem[] = [
-  {
-    key: "dashboard",
-    label: "概览",
-    path: "/app/dashboard",
-    icon: <DashboardOutlined />,
-  },
-  {
-    key: "assets",
-    label: "素材库",
-    path: "/app/assets",
-    icon: <FileOutlined />,
-  },
-  {
-    key: "titles",
-    label: "标题库",
-    path: "/app/titles",
-    icon: <FileTextOutlined />,
-  },
-  {
-    key: "voices",
-    label: "配音库",
-    path: "/app/voices",
-    icon: <AudioOutlined />,
-  },
-  {
-    key: "voice-clone",
-    label: "我的音色",
-    path: "/app/voice-clone",
-    icon: <AudioOutlined />,
-  },
-  {
-    key: "voice-materials",
-    label: "配音素材库",
-    path: "/app/voice-materials",
-    icon: <AudioOutlined />,
-  },
-  {
-    key: "templates",
-    label: "模板库",
-    path: "/app/templates",
-    icon: <AppstoreOutlined />,
-  },
-  {
-    key: "editing-planner",
-    label: "剪辑编辑器",
-    path: "/app/editing-planner",
-    icon: <EditOutlined />,
-  },
-  {
-    key: "my-templates",
-    label: "我的模板",
-    path: "/app/my-templates",
-    icon: <FolderOutlined />,
-  },
-  {
-    key: "generate",
-    label: "一键生成",
-    path: "/app/generate",
-    icon: <VideoCameraOutlined />,
-  },
-  {
-    key: "history",
-    label: "任务历史",
-    path: "/app/history",
-    icon: <HistoryOutlined />,
-  },
-  {
-    key: "products",
-    label: "成品库",
-    path: "/app/products",
-    icon: <TrophyOutlined />,
-  },
-  {
-    key: "duplication",
-    label: "查重",
-    path: "/app/duplication",
-    icon: <ScanOutlined />,
-  },
-];
-
-/**
- * 分组导航列表（Sidebar 使用）
- */
-export const NAV_GROUPS: NavGroup[] = [
-  {
-    title: "创作工具",
-    items: [
-      {
-        key: "dashboard",
-        label: "首页",
-        path: "/app/dashboard",
-        icon: <DashboardOutlined />,
-      },
-      {
-        key: "generate",
-        label: "一键生成",
-        path: "/app/generate",
-        icon: <VideoCameraOutlined />,
-      },
-    ],
-  },
-  {
-    title: "资源管理",
-    items: [
-      {
-        key: "assets",
-        label: "素材库",
-        path: "/app/assets",
-        icon: <FileOutlined />,
-      },
-      {
-        key: "voices",
-        label: "配音库",
-        path: "/app/voices",
-        icon: <AudioOutlined />,
-      },
-      {
-        key: "voice-clone",
-        label: "我的音色",
-        path: "/app/voice-clone",
-        icon: <AudioOutlined />,
-      },
-      {
-        key: "voice-materials",
-        label: "配音素材库",
-        path: "/app/voice-materials",
-        icon: <AudioOutlined />,
-      },
-      {
-        key: "titles",
-        label: "标题库",
-        path: "/app/titles",
-        icon: <FileTextOutlined />,
-      },
-      {
-        key: "products",
-        label: "成片库",
-        path: "/app/products",
-        icon: <TrophyOutlined />,
-      },
-      {
-        key: "templates",
-        label: "模板库",
-        path: "/app/templates",
-        icon: <AppstoreOutlined />,
-      },
-    ],
-  },
-  {
-    title: "系统",
-    items: [
-      {
-        key: "history",
-        label: "任务历史",
-        path: "/app/history",
-        icon: <HistoryOutlined />,
-      },
-      {
-        key: "admin",
-        label: "控制台",
-        path: "/app/admin",
-        icon: <ControlOutlined />,
-      },
-      {
-        key: "subscription",
-        label: "订阅管理",
-        path: "/app/subscription",
-        icon: <CrownOutlined />,
-      },
-    ],
-  },
-];

apps/web/src/pages/admin/Admin.css

@@ -75,35 +75,6 @@
   margin: 0;
 }
 
-/* V21 卡片 */
-.xx-card {
-  background: rgba(255, 255, 255, 0.94);
-  border: 1px solid rgba(226, 232, 240, 0.95);
-  border-radius: var(--radius-lg);
-  box-shadow: 0 10px 30px rgba(15, 23, 42, 0.06);
-  padding: 24px;
-  transition: all 0.3s;
-}
-
-.xx-card:hover {
-  box-shadow: 0 16px 40px rgba(15, 23, 42, 0.08);
-}
-
-.xx-card .ant-card-head {
-  border-bottom: 1px solid rgba(226, 232, 240, 0.8);
-  padding: 20px 24px;
-}
-
-.xx-card .ant-card-head-title {
-  font-weight: 800;
-  font-size: 17px;
-  color: var(--slate, #0f172a);
-}
-
-.xx-card .ant-card-body {
-  padding: 24px;
-}
-
 /* 统计卡片网格 - 4列 */
 .xx-grid-4 {
   display: grid;
@@ -302,17 +273,6 @@
   box-shadow: 0 0 0 3px rgba(79, 70, 229, 0.1) !important;
 }
 
-/* V21 Select */
-.xx-select {
-  border-radius: var(--radius-md) !important;
-}
-
-.xx-select:hover,
-.xx-select:focus {
-  border-color: var(--indigo, #4f46e5) !important;
-  box-shadow: 0 0 0 3px rgba(79, 70, 229, 0.1) !important;
-}
-
 /* V21 Tag */
 .xx-tag {
   border-radius: var(--radius-xs) !important;

apps/web/src/pages/generate/generate.css

@@ -612,54 +612,6 @@
   flex: 1;
 }
 
-/* ============================================================
-   按钮（匹配原型 .btn .ghost / .btn .primary）
-   ============================================================ */
-.xx-btn {
-  display: inline-flex;
-  align-items: center;
-  justify-content: center;
-  gap: 6px;
-  height: 42px;
-  padding: 0 20px;
-  border-radius: var(--radius-sm);
-  font-size: 14px;
-  font-weight: 600;
-  cursor: pointer;
-  transition: all 0.15s ease;
-  border: none;
-  outline: none;
-  white-space: nowrap;
-}
-
-.xx-btn:disabled {
-  opacity: 0.5;
-  cursor: not-allowed;
-}
-
-.xx-btn-primary {
-  background: var(--gradient-primary);
-  color: var(--text-inverse);
-  box-shadow: 0 14px 26px rgba(79, 70, 229, 0.22);
-}
-
-.xx-btn-primary:hover:not(:disabled) {
-  transform: translateY(-2px);
-  box-shadow: 0 18px 34px rgba(79, 70, 229, 0.28);
-}
-
-.xx-btn-ghost {
-  background: var(--bg-primary);
-  border: 1px solid var(--border-color);
-  color: var(--text-secondary);
-}
-
-.xx-btn-ghost:hover:not(:disabled) {
-  border-color: var(--info-border);
-  color: var(--primary-dark);
-  background: var(--primary-soft);
-}
-
 /* ============================================================
    右侧预览区 generate-preview
    ============================================================ */

apps/web/src/router/index.tsx

@@ -170,13 +170,6 @@ export const router = createBrowserRouter([
             Component: m.default,
           })),
       },
-      {
-        path: "my-voices",
-        lazy: () =>
-          import("@/pages/my-voices/MyVoices").then((m) => ({
-            Component: m.default,
-          })),
-      },
       {
         path: "accounts",
         lazy: () =>

apps/worker/video_processing/ffmpeg_utils.py

@@ -42,6 +42,10 @@ XFADE_TRANSITION_MAP: dict[str, str] = {
 
 DEFAULT_TRANSITION_DURATION = 0.5
 
+# FFmpeg 执行默认超时（秒），防止 FFmpeg hang 住导致 worker 永久阻塞
+# 默认 30 分钟，足够处理大部分短视频渲染；超长视频可单独传参覆盖
+DEFAULT_FFMPEG_TIMEOUT = 1800
+
 
 # ── FFmpeg 执行 ───────────────────────────────────────────────────────────────
 
@@ -50,12 +54,14 @@ def run_ffmpeg(
     command: list[str],
     *,
     capture_output: bool = True,
+    timeout: int | None = DEFAULT_FFMPEG_TIMEOUT,
 ) -> tuple[str, str]:
     """执行 FFmpeg 命令。
 
     Args:
         command: 完整的 ffmpeg 命令列表（含 "ffmpeg" 本身）
         capture_output: 是否捕获 stdout/stderr
+        timeout: 超时时间（秒），默认 1800s（30分钟）；None 表示不设超时（不推荐）
 
     Returns:
         (stdout, stderr) 元组
@@ -63,6 +69,7 @@ def run_ffmpeg(
     Raises:
         subprocess.CalledProcessError: 命令执行失败时抛出，
             异常信息包含完整 stderr 以便排查。
+        subprocess.TimeoutExpired: 超时未完成时抛出，FFmpeg 进程会被 kill。
     """
     try:
         result = subprocess.run(  # nosec B603
@@ -71,8 +78,16 @@ def run_ffmpeg(
             stdout=subprocess.PIPE if capture_output else None,
             stderr=subprocess.PIPE if capture_output else None,
             text=True,
+            timeout=timeout,
         )
         return (result.stdout or "", result.stderr or "")
+    except subprocess.TimeoutExpired as e:
+        logger.error(
+            "FFmpeg 命令超时 (%ds): command=%s",
+            timeout or -1,
+            " ".join(str(c) for c in command[:20]),
+        )
+        raise
     except subprocess.CalledProcessError as e:
         # 把完整 stderr 打到日志，方便排查 exit code 183 等问题
         stderr_text = (e.stderr or "").strip()
@@ -148,10 +163,14 @@ def probe_duration(local_path: str | Path) -> float:
 
 
 def probe_video_info(video_path: str) -> dict[str, Any]:
-    """获取视频信息（宽、高、时长、fps）。
+    """获取视频信息（宽、高、时长、fps、编码、像素格式）。
 
     Returns:
-        {"width": int, "height": int, "duration": float, "fps": float}
+        {
+            "width": int, "height": int, "duration": float, "fps": float,
+            "video_codec": str, "audio_codec": str, "pix_fmt": str,
+            "has_audio": bool,
+        }
         失败时返回默认值。
     """
     try:
@@ -160,10 +179,8 @@ def probe_video_info(video_path: str) -> dict[str, Any]:
                 FFPROBE_BIN,
                 "-v",
                 "error",
-                "-select_streams",
-                "v:0",
                 "-show_entries",
-                "stream=width,height,r_frame_rate,duration",
+                "stream=width,height,r_frame_rate,duration,codec_name,codec_type,pix_fmt",
                 "-show_entries",
                 "format=duration",
                 "-of",
@@ -174,19 +191,25 @@ def probe_video_info(video_path: str) -> dict[str, Any]:
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             text=True,
+            timeout=15,
         )
 
         import json
 
         info = json.loads(result.stdout)
-        stream = info.get("streams", [{}])[0]
+        streams = info.get("streams", [])
         fmt = info.get("format", {})
 
-        width = int(stream.get("width", DEFAULT_OUTPUT_WIDTH))
-        height = int(stream.get("height", DEFAULT_OUTPUT_HEIGHT))
+        video_stream = next((s for s in streams if s.get("codec_type") == "video"), {})
+        audio_stream = next((s for s in streams if s.get("codec_type") == "audio"), {})
+
+        width = int(video_stream.get("width", DEFAULT_OUTPUT_WIDTH))
+        height = int(video_stream.get("height", DEFAULT_OUTPUT_HEIGHT))
+        video_codec = video_stream.get("codec_name", "") or ""
+        pix_fmt = video_stream.get("pix_fmt", "") or ""
 
         # 解析帧率
-        fps_str = stream.get("r_frame_rate", "25/1")
+        fps_str = video_stream.get("r_frame_rate", "25/1")
         if "/" in fps_str:
             num, den = fps_str.split("/")
             fps = float(num) / float(den) if float(den) > 0 else DEFAULT_FPS
@@ -194,13 +217,20 @@ def probe_video_info(video_path: str) -> dict[str, Any]:
             fps = float(fps_str) if fps_str else DEFAULT_FPS
 
         # 时长
-        duration = float(fmt.get("duration", 0)) or float(stream.get("duration", 0))
+        duration = float(fmt.get("duration", 0)) or float(video_stream.get("duration", 0))
+
+        has_audio = bool(audio_stream)
+        audio_codec = audio_stream.get("codec_name", "") or ""
 
         return {
             "width": width,
             "height": height,
             "duration": duration,
             "fps": round(fps, 2),
+            "video_codec": video_codec,
+            "audio_codec": audio_codec,
+            "pix_fmt": pix_fmt,
+            "has_audio": has_audio,
         }
     except Exception as e:
         logger.warning("获取视频信息失败: %s, error: %s", video_path, e)
@@ -209,6 +239,10 @@ def probe_video_info(video_path: str) -> dict[str, Any]:
             "height": DEFAULT_OUTPUT_HEIGHT,
             "duration": 0.0,
             "fps": DEFAULT_FPS,
+            "video_codec": "",
+            "audio_codec": "",
+            "pix_fmt": "",
+            "has_audio": True,
         }
 
 

apps/worker/video_processing/oss_helpers.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 import hashlib
 import logging
 import os
+import threading
 from pathlib import Path
 from typing import Optional
 from urllib.parse import urlparse
@@ -17,6 +18,13 @@ import oss2
 
 logger = logging.getLogger(__name__)
 
+# OSS 上传配置
+OSS_CONNECT_TIMEOUT = 10  # 连接超时（秒），防止 TCP 握手挂死
+OSS_UPLOAD_TOTAL_TIMEOUT = 300  # 单文件上传总超时（秒），防止网络慢时无限卡住
+OSS_MULTIPART_THRESHOLD = 100 * 1024 * 1024  # 分片上传阈值：100MB 以上走分片
+OSS_PART_SIZE = 8 * 1024 * 1024  # 分片大小：8MB
+OSS_MULTIPART_NUM_THREADS = 3  # 分片上传并发数
+
 
 # ── OSS 配置 ──────────────────────────────────────────────────────────────────
 
@@ -43,6 +51,9 @@ def oss_bucket() -> oss2.Bucket | None:
     P0-2 修复：endpoint 不带 scheme 时自动补 https:// 前缀，
     确保 sign_url 等依赖 scheme 的方法返回 HTTPS URL。
 
+    P0-staging 修复：增加 connect_timeout=10s，防止网络抖动时
+    TCP 握手阶段无限挂死，导致 worker 进程卡死。
+
     Returns:
         oss2.Bucket 实例，配置缺失时返回 None。
     """
@@ -53,7 +64,12 @@ def oss_bucket() -> oss2.Bucket | None:
     # endpoint 无 scheme 时补 https://，与 API 端 storage.py 保持一致
     if not endpoint.startswith(("http://", "https://")):
         endpoint = f"https://{endpoint}"
-    return oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
+    return oss2.Bucket(
+        oss2.Auth(access_key_id, access_key_secret),
+        endpoint,
+        bucket_name,
+        connect_timeout=OSS_CONNECT_TIMEOUT,
+    )
 
 
 def normalize_storage_key(storage_key_or_url: str) -> str:
@@ -96,6 +112,9 @@ def download_asset(asset_storage_key: str, local_path: Path) -> bool:
 def upload_to_oss(local_path: Path, storage_key: str) -> str | None:
     """上传文件到 OSS，返回公开 URL。
 
+    大文件（>100MB）自动走分片上传，降低内存峰值，减少 OOM 风险。
+    上传加总超时保护（默认 300s），防止网络异常时无限挂死。
+
     Args:
         local_path: 本地文件路径
         storage_key: 目标存储键
@@ -106,18 +125,71 @@ def upload_to_oss(local_path: Path, storage_key: str) -> str | None:
     bucket = oss_bucket()
     if bucket is None:
         return None
-    try:
-        bucket.put_object_from_file(storage_key, str(local_path))
-        settings = oss_settings()
-        if settings:
-            _, _, endpoint, bucket_name = settings
-            endpoint_clean = endpoint.replace("https://", "").replace("http://", "")
-            return f"https://{bucket_name}.{endpoint_clean}/{storage_key}"
+
+    result: dict = {"url": None, "error": None, "file_size": 0}
+    done = threading.Event()
+
+    def _do_upload():
+        try:
+            # 尝试获取文件大小，用于分片判断和日志；stat 失败时 fallback 走普通上传
+            try:
+                file_size = local_path.stat().st_size
+                result["file_size"] = file_size
+                use_multipart = file_size >= OSS_MULTIPART_THRESHOLD
+            except OSError:
+                use_multipart = False
+                file_size = 0
+
+            if use_multipart:
+                # 分片上传：降低内存峰值，每片 8MB，3 线程并发
+                logger.info(
+                    "大文件分片上传: storage_key=%s, size=%.1fMB, part_size=%dMB, threads=%d",
+                    storage_key[:80],
+                    file_size / 1024 / 1024,
+                    OSS_PART_SIZE // 1024 // 1024,
+                    OSS_MULTIPART_NUM_THREADS,
+                )
+                oss2.resumable_upload(
+                    bucket,
+                    storage_key,
+                    str(local_path),
+                    multipart_threshold=OSS_MULTIPART_THRESHOLD,
+                    part_size=OSS_PART_SIZE,
+                    num_threads=OSS_MULTIPART_NUM_THREADS,
+                )
+            else:
+                bucket.put_object_from_file(storage_key, str(local_path))
+
+            # 构造返回 URL
+            settings = oss_settings()
+            if settings:
+                _, _, endpoint, bucket_name = settings
+                endpoint_clean = endpoint.replace("https://", "").replace("http://", "")
+                result["url"] = f"https://{bucket_name}.{endpoint_clean}/{storage_key}"
+        except Exception as e:
+            result["error"] = e
+            logger.exception("上传 OSS 失败: %s", storage_key)
+        finally:
+            done.set()
+
+    upload_thread = threading.Thread(target=_do_upload, daemon=True)
+    upload_thread.start()
+    finished = done.wait(timeout=OSS_UPLOAD_TOTAL_TIMEOUT)
+
+    if not finished:
+        logger.error(
+            "OSS 上传超时（%.0fs），强制中止: storage_key=%s, size=%.1fMB",
+            OSS_UPLOAD_TOTAL_TIMEOUT,
+            storage_key[:80],
+            result["file_size"] / 1024 / 1024 if result["file_size"] else 0,
+        )
         return None
-    except Exception:
-        logger.exception("上传 OSS 失败: %s", storage_key)
+
+    if result["error"]:
         return None
 
+    return result["url"]
+
 
 def get_signed_download_url(storage_key_or_url: str, expires_seconds: int = 3600) -> str | None:
     """生成预签名下载 URL（用于私有 bucket 的 URL 校验或临时下载）。

apps/worker/video_processing/unified_render_service.py

@@ -22,7 +22,6 @@
 from __future__ import annotations
 
 import logging
-import os
 import subprocess
 import time
 from dataclasses import dataclass, field
@@ -296,7 +295,7 @@ WrapStyle: 2
 Encoding: UTF-8
 
 [V4+ Styles]
-Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
+Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding  # noqa: E501
 {chr(10).join(styles)}
 
 [Events]
@@ -460,12 +459,25 @@ class UnifiedRenderService:
 
         is_pass_through = self._can_use_pass_through(layers)
         pass_through_has_audio = False
+        used_stream_copy = False
 
         if is_pass_through:
-            # 直通优化：单clip场景一次FFmpeg同时处理视频+音频，省去提取+合并两次调用
-            pass_through_has_audio = self._render_pass_through(
+            # 先尝试 stream copy 优化（无重编码，性能提升 10 倍+）
+            # 条件不满足或失败时回退到带滤镜的直通渲染
+            stream_copy_ok = self._try_render_stream_copy(
                 layers, output_path, ass_path=ass_path, video_duration=video_duration
             )
+            if stream_copy_ok:
+                used_stream_copy = True
+                # stream copy 模式下，直接探测输出是否有音频
+                clip = layers[0].clips[0]
+                info = probe_video_info(str(clip.local_path))
+                pass_through_has_audio = info.get("has_audio", True)
+            else:
+                # 回退到带滤镜的直通渲染
+                pass_through_has_audio = self._render_pass_through(
+                    layers, output_path, ass_path=ass_path, video_duration=video_duration
+                )
         else:
             filter_complex, input_args = self._build_filter_complex(layers, ass_path=ass_path)
             self._execute_ffmpeg(filter_complex, input_args, video_only_path)
@@ -473,10 +485,11 @@ class UnifiedRenderService:
         t_video_end = time.time()
         video_render_ms = int((t_video_end - t_video_start) * 1000)
         logger.info(
-            "[unified-render] video render done: plan_id=%s duration_ms=%d pass_through=%s",
+            "[unified-render] video render done: plan_id=%s duration_ms=%d pass_through=%s stream_copy=%s",
             self.plan.id,
             video_render_ms,
             is_pass_through,
+            used_stream_copy,
         )
 
         # 6. 音频后处理混音（直通场景已合并处理，跳过）
@@ -619,6 +632,176 @@ class UnifiedRenderService:
             return False
         return True
 
+    def _can_use_stream_copy(
+        self,
+        clip: ResolvedClip,
+        *,
+        ass_path: Path | None = None,
+        video_duration: float = 0.0,
+    ) -> tuple[bool, str]:
+        """判断是否可以走 stream copy（流拷贝，不重编码）。
+
+        性能提升：10 倍以上（典型场景从 20s → 1-2s）。
+
+        条件：
+        1. 视频编码为 h264（输出目标也是 h264）
+        2. 像素格式为 yuv420p
+        3. 分辨率与输出一致（不需要 scale/crop）
+        4. 帧率与输出一致（误差 < 0.1fps）
+        5. 无字幕叠加（字幕需要滤镜）
+        6. 无 trim 需求（或 trim 后恰好等于原时长）
+        7. 无转场、无特效（单 clip 直通已保证）
+
+        Returns:
+            (是否可以 copy, 原因说明)
+        """
+        # 有字幕 → 需要滤镜 → 不能 copy
+        if ass_path is not None:
+            return False, "有字幕叠加"
+
+        # 探测输入视频参数
+        info = probe_video_info(str(clip.local_path))
+
+        # 编码必须是 h264
+        if info.get("video_codec", "") != "h264":
+            return False, f"视频编码不是h264: {info.get('video_codec', 'unknown')}"
+
+        # 像素格式必须是 yuv420p
+        if info.get("pix_fmt", "") != "yuv420p":
+            return False, f"像素格式不是yuv420p: {info.get('pix_fmt', 'unknown')}"
+
+        # 分辨率必须一致
+        if info.get("width", 0) != self.output_width or info.get("height", 0) != self.output_height:
+            return False, (
+                f"分辨率不匹配: "
+                f"{info.get('width', 0)}x{info.get('height', 0)} "
+                f"vs {self.output_width}x{self.output_height}"
+            )
+
+        # 帧率必须一致（误差 < 0.1fps）
+        fps_diff = abs(info.get("fps", 0) - self.output_fps)
+        if fps_diff > 0.1:
+            return False, f"帧率不匹配: {info.get('fps', 0)} vs {self.output_fps}"
+
+        # 检查是否需要 trim
+        effective_duration = UnifiedRenderService._clip_effective_duration(clip)
+        if effective_duration > 0:
+            # 有 trim 需求但视频时长足够，可用 -ss/-t 实现 copy trim
+            input_duration = info.get("duration", 0)
+            if input_duration <= 0:
+                return False, "无法探测输入时长"
+            # trim 起始点 + 目标时长 <= 输入时长
+            start_time = getattr(clip, "start_time", 0) or 0
+            if start_time + effective_duration > input_duration + 0.1:
+                return False, "trim 超出输入时长"
+
+        # video_duration 截断
+        if video_duration > 0 and effective_duration > 0:
+            final_duration = min(effective_duration, video_duration)
+            if final_duration != effective_duration:
+                # 也需要截断，但 -t 可以 copy 模式下用
+                pass
+
+        return True, "所有条件满足"
+
+    def _try_render_stream_copy(
+        self,
+        layers: list[RenderLayer],
+        output_path: Path,
+        *,
+        ass_path: Path | None = None,
+        video_duration: float = 0.0,
+    ) -> bool:
+        """尝试 stream copy 渲染，成功返回 True，失败返回 False（调用方回退到重编码）。
+
+        stream copy 模式：不重编码，直接拷贝视频/音频流，性能提升 10 倍+。
+        仅用于单 clip 直通场景且满足 copy 条件。
+        """
+        clip = layers[0].clips[0]
+        role = layers[0].role
+
+        # 判断是否满足 copy 条件
+        can_copy, reason = self._can_use_stream_copy(clip, ass_path=ass_path, video_duration=video_duration)
+        if not can_copy:
+            logger.info(
+                "[unified-render] stream_copy 跳过: plan_id=%s reason=%s",
+                self.plan.id,
+                reason,
+            )
+            return False
+
+        # 构建 copy 命令
+        command = [
+            FFMPEG_BIN,
+            "-y",
+        ]
+
+        # trim 支持（-ss 放在 -i 前 = input seeking，速度更快但精度稍差；
+        # 放在 -i 后 = output seeking，精度高但慢）
+        # 这里用 output seeking 保证精度，反正 copy 模式已经很快了
+        start_time = getattr(clip, "start_time", 0) or 0
+        effective_duration = UnifiedRenderService._clip_effective_duration(clip)
+
+        command.extend(["-i", str(clip.local_path)])
+
+        if start_time > 0:
+            command.extend(["-ss", f"{start_time:.3f}"])
+
+        # 计算最终时长
+        final_duration = effective_duration
+        if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
+            final_duration = video_duration
+        if final_duration > 0:
+            command.extend(["-t", f"{final_duration:.3f}"])
+
+        # 流拷贝
+        command.extend(
+            [
+                "-c:v",
+                "copy",
+                "-c:a",
+                "copy",
+                "-movflags",
+                "+faststart",
+                str(output_path),
+            ]
+        )
+
+        logger.info(
+            "[unified-render] stream_copy 渲染: plan_id=%s clip=%s role=%s duration=%.2fs",
+            self.plan.id,
+            clip.clip_id,
+            role,
+            final_duration,
+        )
+
+        try:
+            run_ffmpeg(command)
+            # 验证输出文件存在且有大小
+            if output_path.exists() and output_path.stat().st_size > 0:
+                logger.info(
+                    "[unified-render] stream_copy 成功: plan_id=%s size=%d",
+                    self.plan.id,
+                    output_path.stat().st_size,
+                )
+                return True
+            else:
+                logger.warning("[unified-render] stream_copy 输出为空: plan_id=%s", self.plan.id)
+                return False
+        except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
+            logger.warning(
+                "[unified-render] stream_copy 失败，回退到重编码: plan_id=%s error=%s",
+                self.plan.id,
+                str(e)[:200],
+            )
+            # 清理可能的损坏输出文件
+            if output_path.exists():
+                try:
+                    output_path.unlink()
+                except OSError:
+                    pass
+            return False
+
     def _render_pass_through(
         self,
         layers: list[RenderLayer],
@@ -797,7 +980,6 @@ class UnifiedRenderService:
 
         # 计算 PiP 位置
         pip_width = int(self.output_width * _PIP_SCALE)
-        pip_height = int(self.output_height * _PIP_SCALE)
         margin = 20  # 边距
 
         if "overlay" in layer_map:

apps/worker/worker_app/core/config.py

@@ -17,6 +17,7 @@ class WorkerSettings(BaseSettings):
     database_pool_recycle: int = 3600
     environment: str = "development"
     auto_create_schema: bool = False
+    redis_url: str = "redis://redis:6379/0"
 
     # 渲染引擎选择：legacy=旧VideoComposeService，unified=新UnifiedRenderService
     render_engine: str = "legacy"

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -1,13 +1,18 @@
-"""剪辑计划渲染任务 — Phase 8 任务 2.05.
+"""剪辑计划渲染任务 — 支持 Feature Flag 灰度.
 
 Celery 任务 worker.render_edit_plan:
   1. 加载 EditPlan + EditPlanClips
-  2. 下载各片段素材
-  3. 使用 UnifiedRenderService 按时间线+图层渲染
+  2. 根据 Feature Flag 选择渲染引擎（legacy / unified）
+  3. 下载各片段素材 + 渲染
   4. 上传渲染结果到 OSS
   5. 创建 GeneratedVideo 记录 + 查重
   6. 更新 EditPlan / EditPlanClip 状态
   7. 更新 GenerationTask 进度
+
+渲染引擎灰度：
+  - 走 Feature Flag (render_engine) 控制
+  - legacy: VideoComposeService + FFmpeg filter_complex
+  - unified: UnifiedRenderService 图层架构
 """
 
 from __future__ import annotations
@@ -63,14 +68,268 @@ def _get_repos():
 # ── Celery Task ───────────────────────────────────────────────────────────────
 
 
+def _resolve_render_engine(user_id: str) -> str:
+    """根据 Feature Flag 决定使用哪个渲染引擎。
+
+    Returns:
+        "legacy" 或 "unified"
+    """
+    try:
+        from video_processing.render_engine_resolver import get_render_engine_resolver
+
+        resolver = get_render_engine_resolver()
+        return resolver.get_engine(user_id=user_id)
+    except Exception as exc:
+        logger.warning("获取渲染引擎配置失败，fallback 到 legacy: %s", exc)
+        return "legacy"
+
+
+def _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, error_msg: str):
+    """统一的计划失败标记工具。"""
+    plan = plan_repo.get(plan_id)
+    if plan and plan.status.value == "rendering":
+        plan.mark_failed()
+        plan_repo.update(plan)
+    if generation_task_id:
+        gen_task = gen_task_repo.get(generation_task_id)
+        if gen_task and gen_task.status.value != "failed":
+            gen_task.status = "failed"
+            gen_task.error_message = error_msg
+            gen_task.completed_at = datetime.now(timezone.utc)
+            gen_task_repo.update(gen_task)
+
+
+def _finalize_render_success(
+    plan,
+    plan_repo,
+    clip_repo,
+    gen_task_repo,
+    db,
+    plan_id: str,
+    output_url: str,
+    storage_key: str,
+    duration: float,
+    file_size: int,
+    width: int,
+    height: int,
+    rendered_clip_ids: list[str],
+    failed_clip_ids: list[str],
+    generation_task_id: str,
+    output_path: Path,
+    engine: str,
+) -> dict:
+    """渲染成功后的统一收尾：查重 + 更新状态 + 返回结果。"""
+    # 创建 GeneratedVideo 记录 + 查重
+    project_id = plan.project_id or ""
+    batch_id = plan.config.get("batch_id", "")
+    mode = plan.config.get("mode", "edit_plan")
+    if generation_task_id and project_id:
+        try:
+            create_video_record_and_dedup(
+                generation_task_id=generation_task_id,
+                project_id=project_id,
+                batch_id=batch_id,
+                file_url=output_url or "",
+                file_size=file_size,
+                duration=duration,
+                video_path=str(output_path),
+                mode=mode,
+                session=db,
+                width=width,
+                height=height,
+                fps=OUTPUT_FPS,
+            )
+        except Exception as dedup_err:
+            logger.warning("查重失败（不影响渲染结果）: %s", dedup_err)
+
+    # 更新片段状态为 rendered
+    for clip_id in rendered_clip_ids:
+        clip = clip_repo.get(clip_id)
+        if clip and clip.status.value == "ready":
+            clip.mark_rendered()
+            clip_repo.update(clip)
+
+    # 更新 EditPlan 状态为 completed
+    plan.config["rendered_url"] = output_url or ""
+    plan.config["rendered_storage_key"] = storage_key
+    plan.mark_completed()
+    plan_repo.update(plan)
+
+    # 更新 GenerationTask 状态为 completed
+    if generation_task_id:
+        gen_task = gen_task_repo.get(generation_task_id)
+        if gen_task:
+            gen_task.status = "completed"
+            gen_task.progress = 100.0
+            gen_task.result_count = len(rendered_clip_ids)
+            gen_task.completed_at = datetime.now(timezone.utc)
+            gen_task_repo.update(gen_task)
+
+    logger.info(
+        "剪辑计划渲染完成: plan_id=%s engine=%s rendered=%d failed=%d duration=%.1fs",
+        plan_id,
+        engine,
+        len(rendered_clip_ids),
+        len(failed_clip_ids),
+        duration,
+    )
+
+    return {
+        "status": "completed",
+        "plan_id": plan_id,
+        "rendered_count": len(rendered_clip_ids),
+        "failed_count": len(failed_clip_ids),
+        "output_url": output_url,
+        "duration": duration,
+    }
+
+
+def _render_with_unified(
+    plan,
+    clips,
+    asset_path_map: dict[str, Path],
+    tmpdir_path: Path,
+    rendered_clip_ids: list[str],
+    plan_id: str,
+    generation_task_id: str,
+    plan_repo,
+    clip_repo,
+    gen_task_repo,
+    db,
+) -> dict:
+    """统一渲染引擎路径（UnifiedRenderService 图层架构）。"""
+    render_service = UnifiedRenderService(
+        plan=plan,
+        clips=clips,
+        asset_path_map=asset_path_map,
+        work_dir=tmpdir_path,
+        output_width=OUTPUT_WIDTH,
+        output_height=OUTPUT_HEIGHT,
+        output_fps=int(OUTPUT_FPS),
+    )
+
+    try:
+        render_result = render_service.render()
+    except Exception as render_err:
+        logger.error("渲染失败(unified): %s — %s", plan_id, render_err)
+        _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, f"渲染失败: {render_err}")
+        return {"status": "error", "message": f"渲染失败: {render_err}"}
+
+    output_path = render_result.output_path
+
+    # 上传到 OSS
+    storage_key = f"rendered/{plan_id}/output.mp4"
+    output_url = upload_to_oss(output_path, storage_key)
+
+    failed_clip_ids: list[str] = []
+    return _finalize_render_success(
+        plan=plan,
+        plan_repo=plan_repo,
+        clip_repo=clip_repo,
+        gen_task_repo=gen_task_repo,
+        db=db,
+        plan_id=plan_id,
+        output_url=output_url or "",
+        storage_key=storage_key,
+        duration=render_result.duration,
+        file_size=render_result.file_size,
+        width=render_result.width,
+        height=render_result.height,
+        rendered_clip_ids=rendered_clip_ids,
+        failed_clip_ids=failed_clip_ids,
+        generation_task_id=generation_task_id,
+        output_path=output_path,
+        engine="unified",
+    )
+
+
+def _render_with_legacy(
+    plan,
+    clips,
+    rendered_clip_ids: list[str],
+    failed_clip_ids: list[str],
+    tmpdir_path: Path,
+    plan_id: str,
+    generation_task_id: str,
+    plan_repo,
+    clip_repo,
+    gen_task_repo,
+    db,
+) -> dict:
+    """旧引擎路径（VideoComposeService + FFmpeg filter_complex）。"""
+    import os
+    import subprocess
+
+    from apps.api.app.services.video_compose_service import VideoComposeService
+
+    compose_svc = VideoComposeService(db)
+
+    # 校验合成条件
+    validation = compose_svc.validate_compose(plan_id)
+    if not validation.valid:
+        error_msg = "; ".join(validation.errors)
+        logger.error("合成校验失败(legacy): %s — %s", plan_id, error_msg)
+        _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, f"合成校验失败: {error_msg}")
+        return {"status": "error", "message": error_msg}
+
+    # 构建 FFmpeg 命令
+    output_dir = os.environ.get("VIDEO_OUTPUT_DIR", str(tmpdir_path))
+    output_path = Path(output_dir) / f"{plan_id}.mp4"
+    compose_cmd = compose_svc.build_compose_command(plan_id, str(output_path))
+
+    logger.info("执行 FFmpeg (legacy): plan_id=%s", plan_id)
+    try:
+        subprocess.run(
+            compose_cmd.command,
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            timeout=3600,
+        )
+    except subprocess.CalledProcessError as e:
+        error_msg = f"FFmpeg 执行失败: {e.stderr[:500]}"
+        logger.error("FFmpeg 执行失败(legacy): %s — %s", plan_id, error_msg)
+        _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, error_msg)
+        return {"status": "error", "message": error_msg}
+
+    # 获取文件大小
+    file_size = output_path.stat().st_size if output_path.exists() else 0
+    duration = compose_cmd.estimated_duration or 0.0
+
+    # 上传到 OSS
+    storage_key = f"rendered/{plan_id}/output.mp4"
+    output_url = upload_to_oss(output_path, storage_key)
+
+    return _finalize_render_success(
+        plan=plan,
+        plan_repo=plan_repo,
+        clip_repo=clip_repo,
+        gen_task_repo=gen_task_repo,
+        db=db,
+        plan_id=plan_id,
+        output_url=output_url or "",
+        storage_key=storage_key,
+        duration=duration,
+        file_size=file_size,
+        width=OUTPUT_WIDTH,
+        height=OUTPUT_HEIGHT,
+        rendered_clip_ids=rendered_clip_ids,
+        failed_clip_ids=failed_clip_ids,
+        generation_task_id=generation_task_id,
+        output_path=output_path,
+        engine="legacy",
+    )
+
+
 @celery_app.task(name="worker.render_edit_plan", bind=True, max_retries=2)
 def render_edit_plan(self, plan_id: str) -> dict:
     """渲染剪辑计划
 
     流程：
     1. 加载 EditPlan + EditPlanClips
-    2. 下载各片段素材到临时目录，构建 asset_path_map
-    3. 使用 UnifiedRenderService 按时间线+图层渲染
+    2. 根据 Feature Flag 选择渲染引擎（legacy / unified）
+    3. 下载素材 + 渲染
     4. 上传渲染结果到 OSS
     5. 创建 GeneratedVideo 记录 + 查重
     6. 更新 EditPlan → completed, EditPlanClips → rendered
@@ -79,6 +338,7 @@ def render_edit_plan(self, plan_id: str) -> dict:
     logger.info("开始渲染剪辑计划: plan_id=%s", plan_id)
 
     generation_task_id = ""
+    engine = "legacy"
 
     for repos in _get_repos():
         plan_repo, clip_repo, gen_task_repo, db = repos
@@ -93,7 +353,12 @@ def render_edit_plan(self, plan_id: str) -> dict:
             # 获取 generation_task_id（提前读取，确保 except 块可用）
             generation_task_id = plan.config.get("generation_task_id", "")
 
-            # 2. 加载片段列表（按 order 排序）
+            # 2. 选择渲染引擎（Feature Flag 灰度控制）
+            user_id = plan.created_by_user_id or ""
+            engine = _resolve_render_engine(user_id)
+            logger.info("剪辑计划渲染引擎: plan_id=%s engine=%s user_id=%s", plan_id, engine, user_id)
+
+            # 3. 加载片段列表（按 order 排序）
             clips = clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
             if not clips:
                 logger.warning("剪辑计划没有片段: %s", plan_id)
@@ -174,100 +439,38 @@ def render_edit_plan(self, plan_id: str) -> dict:
                             gen_task_repo.update(gen_task)
                     return {"status": "error", "message": "所有片段素材下载失败"}
 
-                # 4. 使用 UnifiedRenderService 渲染
-                render_service = UnifiedRenderService(
-                    plan=plan,
-                    clips=clips,
-                    asset_path_map=asset_path_map,
-                    work_dir=tmpdir_path,
-                    output_width=OUTPUT_WIDTH,
-                    output_height=OUTPUT_HEIGHT,
-                    output_fps=int(OUTPUT_FPS),
-                )
+                # 4. 根据引擎选择渲染方式
+                if engine == "unified":
+                    result = _render_with_unified(
+                        plan=plan,
+                        clips=clips,
+                        asset_path_map=asset_path_map,
+                        tmpdir_path=tmpdir_path,
+                        rendered_clip_ids=rendered_clip_ids,
+                        plan_id=plan_id,
+                        generation_task_id=generation_task_id,
+                        plan_repo=plan_repo,
+                        clip_repo=clip_repo,
+                        gen_task_repo=gen_task_repo,
+                        db=db,
+                    )
+                else:
+                    result = _render_with_legacy(
+                        plan=plan,
+                        clips=clips,
+                        rendered_clip_ids=rendered_clip_ids,
+                        failed_clip_ids=failed_clip_ids,
+                        tmpdir_path=tmpdir_path,
+                        plan_id=plan_id,
+                        generation_task_id=generation_task_id,
+                        plan_repo=plan_repo,
+                        clip_repo=clip_repo,
+                        gen_task_repo=gen_task_repo,
+                        db=db,
+                    )
 
-                try:
-                    render_result = render_service.render()
-                except Exception as render_err:
-                    logger.error("渲染失败: %s — %s", plan_id, render_err)
-                    plan.mark_failed()
-                    plan_repo.update(plan)
-                    if generation_task_id:
-                        gen_task = gen_task_repo.get(generation_task_id)
-                        if gen_task:
-                            gen_task.status = "failed"
-                            gen_task.error_message = f"渲染失败: {render_err}"
-                            gen_task.completed_at = datetime.now(timezone.utc)
-                            gen_task_repo.update(gen_task)
-                    return {"status": "error", "message": f"渲染失败: {render_err}"}
-
-                output_path = render_result.output_path
-
-                # 5. 上传到 OSS
-                storage_key = f"rendered/{plan_id}/output.mp4"
-                output_url = upload_to_oss(output_path, storage_key)
-
-                # 6. 创建 GeneratedVideo 记录 + 查重
-                project_id = plan.project_id or ""
-                batch_id = plan.config.get("batch_id", "")
-                mode = plan.config.get("mode", "edit_plan")
-                if generation_task_id and project_id:
-                    try:
-                        create_video_record_and_dedup(
-                            generation_task_id=generation_task_id,
-                            project_id=project_id,
-                            batch_id=batch_id,
-                            file_url=output_url or "",
-                            file_size=render_result.file_size,
-                            duration=render_result.duration,
-                            video_path=str(output_path),
-                            mode=mode,
-                            session=db,
-                            width=render_result.width,
-                            height=render_result.height,
-                            fps=OUTPUT_FPS,
-                        )
-                    except Exception as dedup_err:
-                        logger.warning("查重失败（不影响渲染结果）: %s", dedup_err)
-
-                # 7. 更新片段状态为 rendered
-                for clip_id in rendered_clip_ids:
-                    clip = clip_repo.get(clip_id)
-                    if clip and clip.status.value == "ready":
-                        clip.mark_rendered()
-                        clip_repo.update(clip)
-
-                # 8. 更新 EditPlan 状态为 completed
-                plan.config["rendered_url"] = output_url or ""
-                plan.config["rendered_storage_key"] = storage_key
-                plan.mark_completed()
-                plan_repo.update(plan)
-
-                # 9. 更新 GenerationTask 状态为 completed
-                if generation_task_id:
-                    gen_task = gen_task_repo.get(generation_task_id)
-                    if gen_task:
-                        gen_task.status = "completed"
-                        gen_task.progress = 100.0
-                        gen_task.result_count = len(rendered_clip_ids)
-                        gen_task.completed_at = datetime.now(timezone.utc)
-                        gen_task_repo.update(gen_task)
-
-                logger.info(
-                    "剪辑计划渲染完成: plan_id=%s rendered=%d failed=%d duration=%.1fs",
-                    plan_id,
-                    len(rendered_clip_ids),
-                    len(failed_clip_ids),
-                    render_result.duration,
-                )
-
-                return {
-                    "status": "completed",
-                    "plan_id": plan_id,
-                    "rendered_count": len(rendered_clip_ids),
-                    "failed_count": len(failed_clip_ids),
-                    "output_url": output_url,
-                    "duration": render_result.duration,
-                }
+                result["engine"] = engine
+                return result
 
         except Exception as exc:
             logger.exception("渲染剪辑计划异常: %s", plan_id)

apps/worker/worker_app/tasks/generation.py

@@ -113,6 +113,7 @@ from video_processing.oss_helpers import (
     get_signed_download_url,
     upload_to_oss,
 )
+from video_processing.render_engine_resolver import ENGINE_LEGACY, ENGINE_UNIFIED
 from video_processing.unified_render_service import UnifiedRenderService
 
 # ── 虚拟 Plan / Clip（内存中构建，不写数据库） ────────────────────────────────
@@ -573,6 +574,148 @@ def _validate_template_exists(template_id: str) -> None:
         session.close()
 
 
+# ── 渲染引擎选择 ─────────────────────────────────────────────────────────────
+
+
+def _resolve_render_engine(user_id: str) -> str:
+    """根据 Feature Flag 决定使用哪个渲染引擎。
+
+    Returns:
+        "legacy" 或 "unified"
+    """
+    try:
+        from video_processing.render_engine_resolver import get_render_engine_resolver
+
+        resolver = get_render_engine_resolver()
+        return resolver.get_engine(user_id=user_id)
+    except Exception as exc:
+        logger.warning("获取渲染引擎配置失败，fallback 到 unified: %s", exc)
+        return ENGINE_UNIFIED
+
+
+# ── 旧引擎渲染（FFmpeg filter_complex） ────────────────────────────────────────
+
+
+def _render_with_legacy_engine(
+    task_id: str,
+    virtual_clips: list[_VirtualClip],
+    asset_path_map: dict[str, Path],
+    work_dir: Path,
+    output_path: Path,
+) -> tuple[float, int]:
+    """旧引擎渲染路径：手动构建 FFmpeg filter_complex 命令。
+
+    说明：generate_video 任务使用虚拟 clips（无 EditPlan 数据库记录），
+    因此无法直接复用 VideoComposeService。这里手动构建等价的 filter_complex
+    命令，与旧引擎行为一致（scale → crop → setpts → trim → setpts，
+    无 fps 归一化，保持原帧率）。
+
+    支持模式：one_take / pip / voice_over / voice_pip
+    - 所有模式统一走 concat 滤镜（与旧引擎多片段逻辑一致）
+
+    Returns:
+        (duration_seconds, file_size_bytes)
+    """
+    import subprocess
+
+    main_clips = [
+        c
+        for c in virtual_clips
+        if c.clip_type in ("main", "b_roll", "background")
+        or (c.clip_type == "main" and c.config.get("role") == "b_roll")
+    ]
+    if not main_clips:
+        main_clips = virtual_clips[:1]
+
+    input_args: list[str] = []
+    video_filters: list[str] = []
+    audio_filters: list[str] = []
+
+    for i, clip in enumerate(main_clips):
+        local_path = asset_path_map.get(clip.asset_id)
+        if not local_path:
+            continue
+        input_args.extend(["-i", str(local_path)])
+
+        duration = clip.duration or 0.0
+
+        # 视频滤镜：scale → crop → setpts → trim → setpts（与旧引擎一致）
+        vf = (
+            f"[{i}:v]"
+            f"scale={OUTPUT_WIDTH}:{OUTPUT_HEIGHT}:force_original_aspect_ratio=increase,"
+            f"crop={OUTPUT_WIDTH}:{OUTPUT_HEIGHT},"
+            f"setpts=PTS-STARTPTS,"
+            f"trim=0:{duration:.3f},"
+            f"setpts=PTS-STARTPTS"
+            f"[v{i}]"
+        )
+        video_filters.append(vf)
+
+        # 音频滤镜：atrim → asetpts
+        af = f"[{i}:a]atrim=0:{duration:.3f},asetpts=PTS-STARTPTS[a{i}]"
+        audio_filters.append(af)
+
+    n = len(main_clips)
+
+    if n == 1:
+        video_label = "[v0]"
+        audio_label = "[a0]"
+    else:
+        # concat 视频
+        v_inputs = "".join(f"[v{i}]" for i in range(n))
+        video_filters.append(f"{v_inputs}concat=n={n}:v=1:a=0[outv]")
+        # concat 音频
+        a_inputs = "".join(f"[a{i}]" for i in range(n))
+        audio_filters.append(f"{a_inputs}concat=n={n}:v=0:a=1[outa]")
+        video_label = "[outv]"
+        audio_label = "[outa]"
+
+    # 组装 filter_complex
+    fc_parts = video_filters + audio_filters
+    filter_complex = ";".join(fc_parts)
+
+    command = [
+        FFMPEG_BIN,
+        "-y",
+        *input_args,
+        "-filter_complex",
+        filter_complex,
+        "-map",
+        video_label,
+        "-map",
+        audio_label,
+        "-c:v",
+        "libx264",
+        "-crf",
+        "23",
+        "-preset",
+        "medium",
+        "-c:a",
+        "aac",
+        "-b:a",
+        "192k",
+        "-movflags",
+        "+faststart",
+        str(output_path),
+    ]
+
+    logger.info("[task_id=%s] [渲染] legacy 引擎 FFmpeg 开始: clips=%d", task_id, n)
+    try:
+        run_ffmpeg(command)
+    except subprocess.CalledProcessError as e:
+        logger.error(
+            "[task_id=%s] [渲染] legacy 引擎 FFmpeg 失败: %s\nfilter_complex: %s",
+            task_id,
+            e,
+            filter_complex[:500],
+        )
+        raise
+
+    file_size = output_path.stat().st_size if output_path.exists() else 0
+    duration = probe_duration(output_path)
+    return duration, file_size
+
+
 # ── Celery Task ──────────────────────────────────────────────────────────────
 
 
@@ -730,31 +873,59 @@ def generate_video(self, task_id: str) -> dict:
                 )
                 _flush_logs(task_id, gen_task)
 
-            # 使用 UnifiedRenderService 渲染
-            logger.info("[task_id=%s] [渲染] FFmpeg 渲染开始", task_id)
+            # 3. 根据 Feature Flag 选择渲染引擎
+            user_id = getattr(gen_task, "created_by_user_id", "") if gen_task else ""
+            engine = _resolve_render_engine(user_id) if user_id else ENGINE_UNIFIED
+            logger.info("[task_id=%s] [渲染] 引擎选择: %s (user_id=%s)", task_id, engine, user_id)
+
             render_start = time.monotonic()
-            render_service = UnifiedRenderService(
-                plan=virtual_plan,
-                clips=virtual_clips,
-                asset_path_map=asset_path_map,
-                work_dir=temp_path,
-                output_width=OUTPUT_WIDTH,
-                output_height=OUTPUT_HEIGHT,
-                output_fps=int(OUTPUT_FPS),
-            )
-            render_result = render_service.render()
-            render_elapsed = time.monotonic() - render_start
-            logger.info(
-                "[task_id=%s] [渲染] FFmpeg 渲染完成: 耗时=%.1fs",
-                task_id,
-                render_elapsed,
-            )
+            render_output_path = temp_path / f"rendered-{task_id}.mp4"
+
+            if engine == ENGINE_LEGACY:
+                # 旧引擎：filter_complex + concat（保持原帧率，无 fps 归一化）
+                render_duration, render_file_size = _render_with_legacy_engine(
+                    task_id=task_id,
+                    virtual_clips=virtual_clips,
+                    asset_path_map=asset_path_map,
+                    work_dir=temp_path,
+                    output_path=render_output_path,
+                )
+                render_elapsed = time.monotonic() - render_start
+                logger.info(
+                    "[task_id=%s] [渲染] legacy 引擎完成: 耗时=%.1fs, 时长=%.2fs",
+                    task_id,
+                    render_elapsed,
+                    render_duration,
+                )
+            else:
+                # 新引擎：UnifiedRenderService 图层架构
+                logger.info("[task_id=%s] [渲染] unified 引擎 FFmpeg 渲染开始", task_id)
+                render_service = UnifiedRenderService(
+                    plan=virtual_plan,
+                    clips=virtual_clips,
+                    asset_path_map=asset_path_map,
+                    work_dir=temp_path,
+                    output_width=OUTPUT_WIDTH,
+                    output_height=OUTPUT_HEIGHT,
+                    output_fps=int(OUTPUT_FPS),
+                )
+                render_result = render_service.render()
+                render_output_path = render_result.output_path
+                render_duration = render_result.duration
+                render_file_size = render_result.file_size
+                render_elapsed = time.monotonic() - render_start
+                logger.info(
+                    "[task_id=%s] [渲染] unified 引擎完成: 耗时=%.1fs",
+                    task_id,
+                    render_elapsed,
+                )
 
             if gen_task:
                 gen_task.append_log(
                     "渲染",
-                    f"FFmpeg 渲染完成, 耗时={render_elapsed:.1f}s",
+                    f"引擎={engine}, 耗时={render_elapsed:.1f}s",
                     duration=round(render_elapsed, 2),
+                    engine=engine,
                 )
                 _flush_logs(task_id, gen_task)
 
@@ -762,14 +933,14 @@ def generate_video(self, task_id: str) -> dict:
             if audio_path:
                 final_path = temp_path / f"final-{task_id}.mp4"
                 try:
-                    _mux_audio_track(render_result.output_path, audio_path, final_path)
+                    _mux_audio_track(render_output_path, audio_path, final_path)
                     # 混音成功，使用混音后的文件
                     output_path = final_path
                 except Exception as mux_err:
                     logger.warning("[task_id=%s] [混音] 音频混合失败，使用无音频版本: %s", task_id, mux_err)
-                    output_path = render_result.output_path
+                    output_path = render_output_path
             else:
-                output_path = render_result.output_path
+                output_path = render_output_path
 
             file_size = output_path.stat().st_size
             duration = probe_duration(output_path)

packages/__init__.py

@@ -1 +0,0 @@
-"""Packages root."""

packages/adapters/__init__.py

@@ -1 +0,0 @@
-"""Adapters package for external implementations."""

packages/adapters/sqlalchemy_impl/models.py

@@ -1,6 +1,6 @@
 from datetime import datetime, timezone
 
-from sqlalchemy import JSON, Boolean, Column, DateTime, Float, Integer, String, Text, UniqueConstraint, create_engine
+from sqlalchemy import JSON, Boolean, Column, DateTime, Float, Integer, String, Text, UniqueConstraint
 from sqlalchemy.orm import declarative_base
 
 Base = declarative_base()

packages/application/__init__.py

@@ -27,7 +27,6 @@ from .generated_videos import (
 from .generation_tasks import (
     CreateGenerationTaskCommand,
     CreateGenerationTaskUseCase,
-    GetGenerationTaskUseCase,
 )
 from .ingest_jobs import SubmitIngestJobCommand, SubmitIngestJobUseCase
 from .jobs import (

packages/application/auth/jwt_handler.py

@@ -12,10 +12,9 @@ JWT 处理器委托层
     payload = jwt_handler.verify_access_token(token)
 """
 
-from datetime import datetime, timedelta
 from typing import Any, Dict, Optional
 
-from packages.application.auth.jwt_service import JWTConfig, JWTService, TokenType
+from packages.application.auth.jwt_service import JWTConfig, JWTService
 
 
 class JWTHandler:

packages/application/auth/jwt_service.py

@@ -208,10 +208,10 @@ def _get_jwt_service():
         kw = dict(secret_key=settings.JWT_SECRET_KEY)
         if hasattr(settings, "JWT_ALGORITHM"):
             kw["algorithm"] = settings.JWT_ALGORITHM
-        if hasattr(settings, "ACCESS_TOKEN_EXPIRE_MINUTES"):
-            kw["access_token_expire_minutes"] = settings.ACCESS_TOKEN_EXPIRE_MINUTES
-        if hasattr(settings, "REFRESH_TOKEN_EXPIRE_DAYS"):
-            kw["refresh_token_expire_days"] = settings.REFRESH_TOKEN_EXPIRE_DAYS
+        if hasattr(settings, "JWT_ACCESS_TOKEN_EXPIRE_MINUTES"):
+            kw["access_token_expire_minutes"] = settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES
+        if hasattr(settings, "JWT_REFRESH_TOKEN_EXPIRE_DAYS"):
+            kw["refresh_token_expire_days"] = settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS
         _jwt_service_instance = JWTService(JWTConfig(**kw))
     return _jwt_service_instance
 

packages/application/auth/login_use_case.py

@@ -293,7 +293,7 @@ class LogoutUseCase:
         try:
             if request.logout_all_devices:
                 # 删除所有设备的 session
-                count = self.session_store.delete_all_user_sessions(request.user_id)
+                self.session_store.delete_all_user_sessions(request.user_id)
                 return True, None
             else:
                 # 删除当前 session

packages/application/auth/password_hasher.py

@@ -85,8 +85,6 @@ class PasswordHasher:
             True 如果需要重新哈希
         """
         try:
-            hashed_bytes = hashed_password.encode("utf-8")
-            current_rounds = bcrypt.getsalt(hashed_bytes)
 
             # 提取当前的 cost factor
             # bcrypt hash 格式: $2b$rounds$salt+hash

packages/application/auth/register_user_use_case.py

@@ -3,7 +3,7 @@
 """
 
 import secrets
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timezone
 from typing import Optional
 from uuid import uuid4
 

packages/application/common/pagination.py

@@ -3,7 +3,7 @@
 """
 
 from math import ceil
-from typing import Generic, List, Optional, TypeVar
+from typing import Generic, List, TypeVar
 
 from pydantic import BaseModel, Field
 

packages/application/jobs.py

@@ -7,9 +7,7 @@ from __future__ import annotations
 
 import logging
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
 from typing import Any
-from uuid import uuid4
 
 from packages.domain.job import Job, JobStatus, JobType
 from packages.ports.job_repository import JobRepository

packages/application/recipe/use_cases.py

@@ -9,7 +9,6 @@ from typing import List, Optional
 from packages.adapters.sqlalchemy_impl.recipe_repository import SQLAlchemyRecipeRepository
 from packages.application.recipe.commands import (
     CreateRecipeCommand,
-    RecipeItemCommand,
     UpdateRecipeCommand,
 )
 from packages.domain.recipe import Recipe, RecipeItem

packages/application/tts_job/__init__.py

@@ -1 +0,0 @@
-"""TTS Job application layer."""

packages/application/tts_job/streaming_service.py

@@ -148,7 +148,6 @@ class TTSStreamingService:
 
         # 并发合成所有分段，按顺序流式推送
         queue: asyncio.Queue[tuple[int, Optional[bytes], Optional[str]]] = asyncio.Queue()
-        completed_count = 0
 
         async def _synthesize_one(idx: int, seg_text: str) -> None:
             """合成单个分段并放入队列。"""

packages/application/tts_job/workflow.py

@@ -24,7 +24,7 @@ from packages.application.cosyvoice_service import (
     CosyVoiceError,
     CosyVoiceService,
 )
-from packages.application.tts_job.audio_merger import AudioMergeError, AudioMerger
+from packages.application.tts_job.audio_merger import AudioMerger
 from packages.application.tts_job.text_splitter import split_text
 from packages.domain.tts_job import TTSJob, TTSJobStatus
 from packages.ports.tts_job_repository import TTSJobRepository

packages/application/voice_clone/use_cases.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import uuid
 from typing import List, Optional
 
 from packages.domain.voice_clone_profile import VoiceCloneProfile

packages/application/voice_clone/workflow.py

@@ -10,7 +10,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Optional
+from typing import Optional
 
 from packages.application.cosyvoice_service import (
     CosyVoiceAuthError,
@@ -21,9 +21,8 @@ from packages.application.voice_clone.use_cases import (
     CreateVoiceCloneUseCase,
     RetryVoiceCloneUseCase,
     VoiceCloneNotFoundError,
-    VoiceCloneNotRetryableError,
 )
-from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
+from packages.domain.voice_clone_profile import VoiceCloneProfile
 from packages.ports.voice_clone_profile_repository import VoiceCloneProfileRepository
 
 logger = logging.getLogger(__name__)

packages/domain/classification.py

@@ -13,7 +13,6 @@ else:
         pass
 
 
-from typing import Any
 from uuid import uuid4
 
 

packages/domain/config_schemas.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 
 import copy
 from enum import Enum
-from typing import List, Optional
+from typing import Optional
 
 from pydantic import BaseModel, Field
 

packages/infrastructure/feature_flags.py

@@ -24,7 +24,7 @@ from __future__ import annotations
 
 import logging
 from dataclasses import dataclass, field
-from typing import Any, Dict, Optional, Set
+from typing import Dict, Optional
 
 logger = logging.getLogger(__name__)
 

packages/infrastructure/module_registry.py

@@ -14,7 +14,7 @@ from __future__ import annotations
 import logging
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any, Callable, Dict, List, Optional, Set
+from typing import Any, Callable, Dict, List, Optional
 
 logger = logging.getLogger(__name__)
 

packages/ports/asset_library_repository.py

@@ -2,7 +2,7 @@
 
 from abc import ABC, abstractmethod
 
-from packages.domain import AssetLibrary, AssetLibraryKind
+from packages.domain import AssetLibrary
 
 
 class AssetLibraryRepository(ABC):

scripts/build_release_images.sh

@@ -30,7 +30,8 @@ fi
 # ---- Registry 配置 ----
 REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
 CACHE_REGISTRY="${CACHE_REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
-CACHE_TAG="${CACHE_TAG:-release}"
+# 主缓存 tag：develop 分支构建时写入，所有分支读取
+CACHE_TAG_PRIMARY="${CACHE_TAG:-develop}"
 
 API_IMAGE="xiaoxia-saas-api:$VERSION"
 WORKER_IMAGE="xiaoxia-saas-worker:$VERSION"
@@ -45,6 +46,7 @@ REGISTRY_WEB="${REGISTRY}/xiaoxia-saas-web:$VERSION"
 
 USE_CACHE=0
 USE_PUSH=0
+CACHE_WRITE=0
 
 # 检查 buildx 和 Registry 认证
 if docker buildx version >/dev/null 2>&1; then
@@ -54,7 +56,10 @@ if docker buildx version >/dev/null 2>&1; then
   docker buildx use default 2>/dev/null || true
 fi
 
-echo "=== Building API image ==="
+# ---- 缓存读写策略（按分支隔离）----
+# 默认只读不写，防止 feature 分支污染主缓存
+# 只有 develop/main 分支才写回缓存
+BRANCH_NAME="${GITHUB_REF_NAME:-${CI_COMMIT_BRANCH:-unknown}}"
 if [ "$USE_CACHE" -eq 1 ]; then
   docker buildx build \
     --build-arg APP_VERSION="$VERSION" \
@@ -68,6 +73,52 @@ else
   docker build --pull=false --build-arg APP_VERSION="$VERSION" -f infra/docker/api.Dockerfile -t "$API_IMAGE" -t "$API_LATEST" .
 fi
 
+build_with_cache() {
+  # usage: build_with_cache <image_name> <dockerfile> <extra_args...>
+  IMG_NAME="$1"
+  DOCKERFILE="$2"
+  shift 2
+  EXTRA_ARGS="$*"
+
+  CACHE_FROM="type=registry,ref=${CACHE_REGISTRY}/${IMG_NAME}-cache:${CACHE_TAG_PRIMARY},ignore-error=true"
+
+  if [ "$CACHE_WRITE" -eq 1 ]; then
+    CACHE_TO="type=registry,ref=${CACHE_REGISTRY}/${IMG_NAME}-cache:${CACHE_TAG_PRIMARY},mode=max"
+    echo "  cache: read+write from ${CACHE_REGISTRY}/${IMG_NAME}-cache:${CACHE_TAG_PRIMARY}"
+  else
+    CACHE_TO=""
+    echo "  cache: read-only from ${CACHE_REGISTRY}/${IMG_NAME}-cache:${CACHE_TAG_PRIMARY}"
+  fi
+
+  if [ "$USE_CACHE" -eq 1 ]; then
+    if [ -n "$CACHE_TO" ]; then
+      docker buildx build \
+        $EXTRA_ARGS \
+        --cache-from "$CACHE_FROM" \
+        --cache-to "$CACHE_TO" \
+        -f "$DOCKERFILE" \
+        -t "$IMG_NAME:$VERSION" \
+        --load \
+        .
+    else
+      docker buildx build \
+        $EXTRA_ARGS \
+        --cache-from "$CACHE_FROM" \
+        -f "$DOCKERFILE" \
+        -t "$IMG_NAME:$VERSION" \
+        --load \
+        .
+    fi
+  else
+    docker build --pull=false $EXTRA_ARGS -f "$DOCKERFILE" -t "$IMG_NAME:$VERSION" .
+  fi
+}
+
+echo "=== Building API image ==="
+build_with_cache "api" "infra/docker/api.Dockerfile" \
+  "--build-arg APP_VERSION=$VERSION"
+docker tag "$API_IMAGE" "$API_LATEST"
+
 echo "=== Building Worker image ==="
 if [ "$USE_CACHE" -eq 1 ]; then
   docker buildx build \
@@ -83,9 +134,16 @@ else
 fi
 
 echo "=== Building Web image (with buildx cache) ==="
-# 先构建前端产物
+# 先构建前端产物（使用持久化 npm 缓存卷）
+NPM_CACHE_VOLUME="xiaoxia-npm-cache"
+if ! docker volume inspect "$NPM_CACHE_VOLUME" >/dev/null 2>&1; then
+  docker volume create "$NPM_CACHE_VOLUME" >/dev/null
+  echo "  Created npm cache volume: $NPM_CACHE_VOLUME"
+fi
+
 docker run --rm \
   -v "$PWD:/workspace" \
+  -v "$NPM_CACHE_VOLUME:/workspace/apps/web/node_modules" \
   -w /workspace/apps/web \
   docker.m.daocloud.io/library/node:20 \
   sh -lc "npm ci && npm run build"

scripts/check_migration_safety.py

@@ -247,3 +247,4 @@ def main() -> int:
 
 if __name__ == "__main__":
     sys.exit(main())
+

scripts/ci-deploy-production.sh

@@ -0,0 +1,256 @@
+#!/bin/sh
+# ============================================================
+# Production 部署脚本 - Registry 拉取方式
+# ============================================================
+# 原始位置：原内嵌在 .gitea/workflows/ci-cd.yml 的 deploy-production Job 中
+# 以 base64 编码存储在 DEPLOY_B64 变量中，通过 SSH 管道传送到服务器执行
+#
+# 功能：
+#   1. 登录 Gitea Registry
+#   2. Pull api / worker / web 三个镜像
+#   3. 备份旧前端静态资源（兼容 CDN 缓存，防止 404）
+#   4. 检查基础设施容器（PostgreSQL / Redis）
+#   5. 执行数据库 Migration
+#   6. 停止并重新启动三个业务容器
+#   7. 健康检查等待就绪
+#   8. 清理旧镜像
+#
+# 依赖的环境变量（由 CI 通过 SSH 传入）：
+#   IMAGE_TAG       - 镜像版本标签（如 v0.1.127，对应 git tag）
+#   REGISTRY_TOKEN  - Gitea Registry 访问令牌
+# ============================================================
+
+set -eu
+
+IMAGE_TAG="${IMAGE_TAG:-}"
+REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
+REGISTRY_USER="${REGISTRY_USER:-xiaoxia}"
+REGISTRY_TOKEN="${REGISTRY_TOKEN:-}"
+
+ENV_FILE="${ENV_FILE:-/var/lib/xiaoxia-saas-production/.env}"
+GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-production/generated}"
+LEGACY_ASSETS_DIR="${LEGACY_ASSETS_DIR:-/var/lib/xiaoxia-saas-production/legacy-assets}"
+
+if [ -z "$IMAGE_TAG" ]; then
+  echo "ERROR: IMAGE_TAG is required"
+  exit 1
+fi
+
+test -f "$ENV_FILE"
+mkdir -p "$GENERATED_DIR"
+mkdir -p "$LEGACY_ASSETS_DIR"
+
+# ---- 登录 Registry ----
+if [ -n "$REGISTRY_TOKEN" ]; then
+  echo "Logging in to registry: $REGISTRY"
+  REGISTRY_HOST=$(echo "$REGISTRY" | cut -d/ -f1)
+  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 三个镜像 ----
+REGISTRY_API="${REGISTRY}/xiaoxia-saas-api:${IMAGE_TAG}"
+REGISTRY_WORKER="${REGISTRY}/xiaoxia-saas-worker:${IMAGE_TAG}"
+REGISTRY_WEB="${REGISTRY}/xiaoxia-saas-web:${IMAGE_TAG}"
+
+LOCAL_API="xiaoxia-saas-api:${IMAGE_TAG}"
+LOCAL_WORKER="xiaoxia-saas-worker:${IMAGE_TAG}"
+LOCAL_WEB="xiaoxia-saas-web:${IMAGE_TAG}"
+
+echo "Pulling API image..."
+docker pull "$REGISTRY_API"
+echo "Pulling Worker image..."
+docker pull "$REGISTRY_WORKER"
+echo "Pulling Web image..."
+docker pull "$REGISTRY_WEB"
+
+# ---- Re-tag 成本地镜像名 ----
+docker tag "$REGISTRY_API" "$LOCAL_API"
+docker tag "$REGISTRY_WORKER" "$LOCAL_WORKER"
+docker tag "$REGISTRY_WEB" "$LOCAL_WEB"
+echo "All images pulled and tagged."
+
+# ---- 备份旧前端 assets（生产环境访问量大，防止 CDN 缓存命中 404） ----
+echo "Backing up legacy assets from current web container..."
+if docker inspect xiaoxia-web-production >/dev/null 2>&1; then
+  _tmpdir="/tmp/legacy-assets-$$"
+  rm -rf "$_tmpdir"
+  mkdir -p "$_tmpdir"
+  docker cp xiaoxia-web-production:/usr/share/nginx/html/assets/. "$_tmpdir/" 2>/dev/null || true
+  # 复制到 LEGACY_ASSETS_DIR（下一次部署时作为 fallback 挂载）
+  if [ -d "$_tmpdir" ] && [ "$(ls -A "$_tmpdir" 2>/dev/null)" ]; then
+    cp -an "$_tmpdir"/. "$LEGACY_ASSETS_DIR"/ 2>/dev/null || true
+    echo "Legacy assets backed up: $(ls "$_tmpdir" | wc -l) files"
+  fi
+  rm -rf "$_tmpdir"
+else
+  echo "No existing web container, skipping legacy assets backup"
+fi
+
+# 清理超过 7 天的旧 assets 文件
+if [ -d "$LEGACY_ASSETS_DIR" ]; then
+  find "$LEGACY_ASSETS_DIR" -type f -mtime +7 -delete 2>/dev/null || true
+  echo "Legacy assets cleanup done (retain 7 days)"
+fi
+
+# ---- 检查基础设施容器状态 ----
+echo "Checking infrastructure containers..."
+for c in xiaoxia-postgres-production xiaoxia-redis-production; do
+  if ! docker inspect "$c" >/dev/null 2>&1; then
+    echo "ERROR: Required container not found: $c"
+    exit 1
+  fi
+  state=$(docker inspect -f '{{.State.Status}}' "$c")
+  if [ "$state" != "running" ]; then
+    echo "ERROR: Container not running: $c ($state)"
+    exit 1
+  fi
+done
+
+# ---- 创建生产网络 ----
+docker network create xiaoxia-net-production 2>/dev/null || true
+
+# ---- 执行数据库 Migration ----
+echo "Running database migrations..."
+docker run --rm \
+  --env-file "$ENV_FILE" \
+  --network xiaoxia-net-production \
+  -e APP_ENV=production \
+  "$LOCAL_API" sh -c "cd /app && alembic upgrade head"
+echo "Migrations completed."
+
+# ---- 停止旧容器 ----
+echo "Stopping old containers..."
+docker rm -f xiaoxia-api-production 2>/dev/null || true
+docker rm -f xiaoxia-worker-production 2>/dev/null || true
+docker rm -f xiaoxia-web-production 2>/dev/null || true
+
+# ---- 启动参数（日志配置统一） ----
+LOG_OPTS="--log-driver json-file --log-opt max-size=50m --log-opt max-file=3"
+
+# ---- 启动 API 容器 ----
+echo "Starting API container..."
+docker run -d \
+  --name xiaoxia-api-production \
+  --env-file "$ENV_FILE" \
+  --network xiaoxia-net-production \
+  -p 127.0.0.1:8001:8000 \
+  -e APP_ENV=production \
+  -e APP_VERSION="$IMAGE_TAG" \
+  -e GENERATED_FILES_DIR=/app/generated \
+  -e GENERATED_FILES_URL_PREFIX=/generated-files \
+  -e PUBLIC_API_BASE_URL=https://api.xiaoxiajianji.com \
+  -v "$GENERATED_DIR:/app/generated" \
+  --restart unless-stopped \
+  --cpus 2 \
+  --memory 2g \
+  --health-cmd "python3 -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 \
+  "$LOCAL_API"
+
+# ---- 启动 Worker 容器 ----
+echo "Starting Worker container..."
+docker run -d \
+  --name xiaoxia-worker-production \
+  --env-file "$ENV_FILE" \
+  --network xiaoxia-net-production \
+  -e APP_ENV=production \
+  -e APP_VERSION="$IMAGE_TAG" \
+  -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://api.xiaoxiajianji.com \
+  -v "$GENERATED_DIR:/app/generated" \
+  --restart unless-stopped \
+  --cpus 2 \
+  --memory 2g \
+  --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 \
+  "$LOCAL_WORKER"
+
+# ---- 启动 Web 容器 ----
+# Legacy assets 挂载到 /usr/share/nginx/html/assets-legacy/assets/
+# nginx 配置中 assets location 有 fallback 规则
+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"
+  echo "Web container: legacy assets mounted (fallback)"
+else
+  echo "Web container: no legacy assets to mount"
+fi
+
+echo "Starting Web container..."
+docker run -d \
+  --name xiaoxia-web-production \
+  --network xiaoxia-net-production \
+  -p 127.0.0.1:3002:80 \
+  --restart unless-stopped \
+  --cpus 0.5 \
+  --memory 512m \
+  $LEGACY_VOLUME \
+  --health-cmd "wget --spider -q http://127.0.0.1:80" \
+  --health-interval 30s \
+  --health-timeout 5s \
+  --health-retries 3 \
+  $LOG_OPTS \
+  "$LOCAL_WEB"
+
+# ---- 等待 API 就绪 ----
+echo "Waiting for API to become healthy..."
+i=0
+while [ "$i" -lt 40 ]; do
+  if curl -sf --max-time 5 http://127.0.0.1:8001/health >/dev/null 2>&1; then
+    echo "API is healthy!"
+    break
+  fi
+  i=$((i + 1))
+  echo "  Waiting... ($i/40)"
+  sleep 3
+done
+
+if [ "$i" -ge 40 ]; then
+  echo "ERROR: API did not become healthy within 120s"
+  docker logs --tail 50 xiaoxia-api-production
+  exit 1
+fi
+
+# ---- 等待 Web 就绪 ----
+echo "Waiting for Web to become healthy..."
+i=0
+while [ "$i" -lt 15 ]; do
+  if curl -sf --max-time 5 http://127.0.0.1:3002/ >/dev/null 2>&1; then
+    echo "Web is healthy!"
+    break
+  fi
+  i=$((i + 1))
+  echo "  Waiting... ($i/15)"
+  sleep 2
+done
+
+if [ "$i" -ge 15 ]; then
+  echo "ERROR: Web did not become healthy within 30s"
+  docker logs --tail 30 xiaoxia-web-production
+  exit 1
+fi
+
+# ---- 清理旧镜像 ----
+echo "Cleaning up old images..."
+docker image prune -af --filter "until=168h" 2>/dev/null || true
+docker builder prune -af --filter "until=168h" 2>/dev/null || true
+
+echo ""
+echo "=== Production deployment complete ==="
+echo "API:      http://127.0.0.1:8001"
+echo "Web:      http://127.0.0.1:3002"
+echo "Version:  $IMAGE_TAG"
+docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Image}}" | grep production

scripts/ci-deploy-staging.sh

@@ -0,0 +1,219 @@
+#!/bin/sh
+# ============================================================
+# Staging 部署脚本 - Registry 拉取方式（替代 Watchtower）
+# ============================================================
+# 说明：原 Staging 使用 Watchtower 轮询 :staging 标签自动更新
+# 本脚本替代 Watchtower，由 CI 主动触发部署，优势：
+#   - 部署时机精确可控，无需固定 sleep 等待
+#   - 部署完成立即健康检查，失败可快速回滚
+#   - 部署日志完整记录在 CI 中
+#
+# 功能：
+#   1. 登录 Gitea Registry
+#   2. Pull api / worker / web 三个镜像（用 commit SHA 作为 tag）
+#   3. 执行数据库 Migration
+#   4. 停止并重新启动三个业务容器
+#   5. 健康检查等待就绪
+#
+# 依赖的环境变量（由 CI 通过 SSH 传入）：
+#   IMAGE_TAG       - 镜像标签（一般为 commit SHA）
+#   REGISTRY_TOKEN  - Gitea Registry 访问令牌
+# ============================================================
+
+set -eu
+
+IMAGE_TAG="${IMAGE_TAG:-}"
+REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
+REGISTRY_USER="${REGISTRY_USER:-xiaoxia}"
+REGISTRY_TOKEN="${REGISTRY_TOKEN:-}"
+
+ENV_FILE="${ENV_FILE:-/var/lib/xiaoxia-saas-staging/.env}"
+GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-staging/generated}"
+
+if [ -z "$IMAGE_TAG" ]; then
+  echo "ERROR: IMAGE_TAG is required"
+  exit 1
+fi
+
+test -f "$ENV_FILE"
+mkdir -p "$GENERATED_DIR"
+
+# ---- 登录 Registry ----
+if [ -n "$REGISTRY_TOKEN" ]; then
+  echo "Logging in to registry: $REGISTRY"
+  REGISTRY_HOST=$(echo "$REGISTRY" | cut -d/ -f1)
+  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 三个镜像 ----
+REGISTRY_API="${REGISTRY}/xiaoxia-saas-api:${IMAGE_TAG}"
+REGISTRY_WORKER="${REGISTRY}/xiaoxia-saas-worker:${IMAGE_TAG}"
+REGISTRY_WEB="${REGISTRY}/xiaoxia-saas-web:${IMAGE_TAG}"
+
+LOCAL_API="xiaoxia-saas-api:staging-${IMAGE_TAG}"
+LOCAL_WORKER="xiaoxia-saas-worker:staging-${IMAGE_TAG}"
+LOCAL_WEB="xiaoxia-saas-web:staging-${IMAGE_TAG}"
+
+echo "Pulling API image..."
+docker pull "$REGISTRY_API"
+echo "Pulling Worker image..."
+docker pull "$REGISTRY_WORKER"
+echo "Pulling Web image..."
+docker pull "$REGISTRY_WEB"
+
+# ---- Re-tag 成本地镜像名 ----
+docker tag "$REGISTRY_API" "$LOCAL_API"
+docker tag "$REGISTRY_WORKER" "$LOCAL_WORKER"
+docker tag "$REGISTRY_WEB" "$LOCAL_WEB"
+echo "All images pulled and tagged."
+
+# ---- 检查基础设施容器 ----
+echo "Checking infrastructure containers..."
+for c in xiaoxia-postgres-staging xiaoxia-redis-staging; do
+  if ! docker inspect "$c" >/dev/null 2>&1; then
+    echo "ERROR: Required container not found: $c"
+    exit 1
+  fi
+  state=$(docker inspect -f '{{.State.Status}}' "$c")
+  if [ "$state" != "running" ]; then
+    echo "ERROR: Container not running: $c ($state)"
+    exit 1
+  fi
+done
+
+# ---- 创建 Staging 网络 ----
+docker network create xiaoxia-net-staging 2>/dev/null || true
+
+# ---- 执行数据库 Migration ----
+echo "Running database migrations..."
+docker run --rm \
+  --env-file "$ENV_FILE" \
+  --network xiaoxia-net-staging \
+  -e APP_ENV=staging \
+  "$LOCAL_API" sh -c "cd /app && alembic upgrade head"
+echo "Migrations completed."
+
+# ---- 停止旧容器 ----
+echo "Stopping old 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 容器 ----
+echo "Starting API container..."
+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="$IMAGE_TAG" \
+  -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 \
+  --cpus 1 \
+  --memory 1g \
+  --health-cmd "python3 -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 30s \
+  $LOG_OPTS \
+  "$LOCAL_API"
+
+# ---- 启动 Worker 容器 ----
+echo "Starting Worker container..."
+docker run -d \
+  --name xiaoxia-worker-staging \
+  --env-file "$ENV_FILE" \
+  --network xiaoxia-net-staging \
+  -e APP_ENV=staging \
+  -e APP_VERSION="$IMAGE_TAG" \
+  -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 \
+  --cpus 1 \
+  --memory 1g \
+  --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 \
+  "$LOCAL_WORKER"
+
+# ---- 启动 Web 容器 ----
+echo "Starting Web container..."
+docker run -d \
+  --name xiaoxia-web-staging \
+  --network xiaoxia-net-staging \
+  -p 127.0.0.1:3001:80 \
+  --restart unless-stopped \
+  --cpus 0.5 \
+  --memory 256m \
+  --health-cmd "wget --spider -q http://127.0.0.1:80" \
+  --health-interval 30s \
+  --health-timeout 5s \
+  --health-retries 3 \
+  $LOG_OPTS \
+  "$LOCAL_WEB"
+
+# ---- 等待 API 就绪 ----
+echo "Waiting for API to become healthy..."
+i=0
+while [ "$i" -lt 30 ]; do
+  if curl -sf --max-time 5 http://127.0.0.1:8000/health >/dev/null 2>&1; then
+    echo "API is healthy!"
+    break
+  fi
+  i=$((i + 1))
+  echo "  Waiting... ($i/30)"
+  sleep 2
+done
+
+if [ "$i" -ge 30 ]; then
+  echo "ERROR: API did not become healthy within 60s"
+  docker logs --tail 50 xiaoxia-api-staging
+  exit 1
+fi
+
+# ---- 等待 Web 就绪 ----
+echo "Waiting for Web to become healthy..."
+i=0
+while [ "$i" -lt 15 ]; do
+  if curl -sf --max-time 5 http://127.0.0.1:3001/ >/dev/null 2>&1; then
+    echo "Web is healthy!"
+    break
+  fi
+  i=$((i + 1))
+  echo "  Waiting... ($i/15)"
+  sleep 2
+done
+
+if [ "$i" -ge 15 ]; then
+  echo "ERROR: Web did not become healthy within 30s"
+  docker logs --tail 30 xiaoxia-web-staging
+  exit 1
+fi
+
+# ---- 清理旧镜像 ----
+echo "Cleaning up old images..."
+docker image prune -af --filter "until=72h" 2>/dev/null || true
+
+echo ""
+echo "=== Staging deployment complete ==="
+echo "API:      http://127.0.0.1:8000"
+echo "Web:      http://127.0.0.1:3001"
+echo "Version:  $IMAGE_TAG"
+docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Image}}" | grep staging

tests/unit/test_edit_plan_worker_failure.py

@@ -63,6 +63,8 @@ class StubEditPlan:
     template_id: str = "tmpl-001"
     status: Any = None
     config: dict = field(default_factory=dict)
+    project_id: str = ""
+    created_by_user_id: str = "user-001"
 
     def mark_failed(self):
         self.status = _StubStatus("failed")

tests/unit/test_ffmpeg_timeout_protection.py

@@ -0,0 +1,93 @@
+"""FFmpeg 超时保护测试。
+
+验证 run_ffmpeg / probe_video_info 的超时保护机制，
+防止 FFmpeg hang 住导致 worker 永久阻塞。
+"""
+
+from __future__ import annotations
+
+import subprocess
+from unittest.mock import MagicMock, patch
+
+import pytest
+from video_processing.ffmpeg_utils import (
+    DEFAULT_FFMPEG_TIMEOUT,
+    probe_video_info,
+    run_ffmpeg,
+)
+
+# ── run_ffmpeg 超时保护 ──────────────────────────────────────────────────────
+
+
+class TestRunFFmpegTimeout:
+    """run_ffmpeg 超时保护测试。"""
+
+    def test_default_timeout_is_set(self):
+        """默认超时应为 1800 秒（30分钟）。"""
+        assert DEFAULT_FFMPEG_TIMEOUT == 1800
+
+    def test_timeout_expired_is_raised(self):
+        """超时未完成时 TimeoutExpired 异常被传播。"""
+        with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
+            mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg", "test"], timeout=1)
+            with pytest.raises(subprocess.TimeoutExpired):
+                run_ffmpeg(["ffmpeg", "test"])
+
+    def test_custom_timeout(self):
+        """支持自定义超时时间。"""
+        with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
+            mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg"], timeout=5)
+            with pytest.raises(subprocess.TimeoutExpired):
+                run_ffmpeg(["ffmpeg", "test"], timeout=5)
+
+    def test_none_timeout_disables_protection(self):
+        """timeout=None 可以禁用超时保护（不推荐）。"""
+        with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
+            mock_result = MagicMock()
+            mock_result.stdout = ""
+            mock_result.stderr = ""
+            mock_run.return_value = mock_result
+            run_ffmpeg(["ffmpeg", "test"], timeout=None)
+            # 验证 timeout=None 被传递
+            call_kwargs = mock_run.call_args.kwargs
+            assert call_kwargs["timeout"] is None
+
+    def test_called_process_error_still_raised(self):
+        """超时异常不影响原有 CalledProcessError 的抛出。"""
+        with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
+            mock_run.side_effect = subprocess.CalledProcessError(returncode=1, cmd=["ffmpeg"], stderr="error msg")
+            with pytest.raises(subprocess.CalledProcessError):
+                run_ffmpeg(["ffmpeg", "test"])
+
+
+# ── probe_video_info 超时保护 ────────────────────────────────────────────────
+
+
+class TestProbeVideoInfoTimeout:
+    """probe_video_info 超时保护测试。"""
+
+    def test_probe_uses_timeout(self):
+        """probe_video_info 调用 ffprobe 时应设置 timeout=15。"""
+        with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
+            mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffprobe"], timeout=15)
+            # 超时异常被捕获，返回默认值
+            result = probe_video_info("/tmp/test.mp4")
+            assert result["width"] == 1280  # DEFAULT_OUTPUT_WIDTH
+            assert result["height"] == 720  # DEFAULT_OUTPUT_HEIGHT
+
+    def test_probe_success(self):
+        """正常情况应解析 ffprobe JSON 输出。"""
+        fake_output = """
+        {
+            "streams": [{"width": 1920, "height": 1080, "codec_type": "video", "r_frame_rate": "30/1", "duration": "10.5"}],
+            "format": {"duration": "10.5"}
+        }
+        """
+        with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
+            mock_result = MagicMock()
+            mock_result.stdout = fake_output
+            mock_run.return_value = mock_result
+            result = probe_video_info("/tmp/test.mp4")
+            assert result["width"] == 1920
+            assert result["height"] == 1080
+            assert abs(result["duration"] - 10.5) < 0.01

tests/unit/test_generation_feature_flag.py

@@ -0,0 +1,338 @@
+"""generate_video 任务 Feature Flag 灰度引擎选择单元测试.
+
+覆盖：
+- _resolve_render_engine 正常返回 unified / legacy
+- Feature Flag 不可用时 fallback 到 unified
+- 白名单 / 百分比 / 全局开关各场景
+- _render_with_legacy_engine 命令构建与输出验证
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from datetime import datetime, timezone
+from types import ModuleType
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+from pathlib import Path
+
+import pytest
+
+# ── Mock worker 模块以避免数据库连接 ──────────────────────────────────────────
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "worker"))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+_mock_db_mod = ModuleType("worker_app.db")
+_mock_db_mod.SessionLocal = MagicMock()
+sys.modules.setdefault("worker_app.db", _mock_db_mod)
+
+_mock_celery_mod = ModuleType("worker_app.celery_app")
+_mock_celery_app = MagicMock()
+_mock_celery_app.task = lambda **kwargs: lambda fn: fn
+_mock_celery_mod.celery_app = _mock_celery_app
+sys.modules.setdefault("worker_app.celery_app", _mock_celery_mod)
+
+# Mock worker_app.core.config 避免 settings 加载
+_mock_config_mod = ModuleType("worker_app.core.config")
+_mock_settings = MagicMock()
+_mock_settings.redis_url = None
+_mock_settings.render_engine = "unified"
+_mock_config_mod.get_settings = lambda: _mock_settings
+sys.modules.setdefault("worker_app.core", ModuleType("worker_app.core"))
+sys.modules.setdefault("worker_app.core.config", _mock_config_mod)
+
+
+# ── 测试用数据类 ──────────────────────────────────────────────────────────────
+
+
+class _TestClip:
+    def __init__(self, asset_id, duration=30.0, clip_type="main", config=None, order=0):
+        self.id = f"clip_{asset_id}"
+        self.plan_id = "test-plan"
+        self.clip_type = clip_type
+        self.order = order
+        self.asset_id = asset_id
+        self.duration = duration
+        self.config = config or {}
+        self.start_time = 0.0
+        self.transition_effect = "cut"
+
+
+# ── RenderEngineResolver 基础行为测试 ───────────────────────────────────────
+
+
+def test_resolver_unified_when_enabled_100_percent():
+    """flag 全局开启（percentage=100）时，返回 unified。"""
+    from video_processing.render_engine_resolver import RenderEngineResolver
+
+    from packages.adapters.redis.feature_flag_store import (
+        FeatureFlagConfig,
+        InMemoryFeatureFlagStore,
+    )
+
+    store = InMemoryFeatureFlagStore()
+    store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
+    resolver = RenderEngineResolver(default_engine="legacy", store=store)
+
+    assert resolver.get_engine(user_id="user-123") == "unified"
+
+
+def test_resolver_legacy_when_flag_disabled():
+    """flag 全局关闭时，返回默认引擎 legacy。"""
+    from video_processing.render_engine_resolver import RenderEngineResolver
+
+    from packages.adapters.redis.feature_flag_store import (
+        FeatureFlagConfig,
+        InMemoryFeatureFlagStore,
+    )
+
+    store = InMemoryFeatureFlagStore()
+    store.set(FeatureFlagConfig(name="render_engine", enabled=False, percentage=100))
+    resolver = RenderEngineResolver(default_engine="legacy", store=store)
+
+    assert resolver.get_engine(user_id="user-123") == "legacy"
+
+
+def test_resolver_whitelist_overrides_percentage_0():
+    """白名单用户即使 percentage=0 也走 unified。"""
+    from video_processing.render_engine_resolver import RenderEngineResolver
+
+    from packages.adapters.redis.feature_flag_store import (
+        FeatureFlagConfig,
+        InMemoryFeatureFlagStore,
+    )
+
+    store = InMemoryFeatureFlagStore()
+    store.set(
+        FeatureFlagConfig(
+            name="render_engine",
+            enabled=True,
+            percentage=0,
+            whitelist={"user-vip"},
+        )
+    )
+    resolver = RenderEngineResolver(default_engine="legacy", store=store)
+
+    assert resolver.get_engine(user_id="user-vip") == "unified"
+    assert resolver.get_engine(user_id="user-other") == "legacy"
+
+
+def test_resolver_percentage_0_all_legacy():
+    """percentage=0 且无白名单时，全部走 legacy。"""
+    from video_processing.render_engine_resolver import RenderEngineResolver
+
+    from packages.adapters.redis.feature_flag_store import (
+        FeatureFlagConfig,
+        InMemoryFeatureFlagStore,
+    )
+
+    store = InMemoryFeatureFlagStore()
+    store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=0))
+    resolver = RenderEngineResolver(default_engine="legacy", store=store)
+
+    for i in range(50):
+        assert resolver.get_engine(user_id=f"user-{i}") == "legacy"
+
+
+def test_resolver_default_unified_when_flag_off():
+    """默认引擎设为 unified 且 flag 关闭时，返回 unified。"""
+    from video_processing.render_engine_resolver import RenderEngineResolver
+
+    from packages.adapters.redis.feature_flag_store import (
+        FeatureFlagConfig,
+        InMemoryFeatureFlagStore,
+    )
+
+    store = InMemoryFeatureFlagStore()
+    store.set(FeatureFlagConfig(name="render_engine", enabled=False, percentage=0))
+    resolver = RenderEngineResolver(default_engine="unified", store=store)
+
+    assert resolver.get_engine(user_id="user-123") == "unified"
+
+
+# ── _render_with_legacy_engine 集成测试 ──────────────────────────────────────
+
+
+def test_legacy_engine_single_clip_keeps_original_fps():
+    """单 clip 场景：输出保持原帧率（不做 fps 归一化），分辨率缩放正确。"""
+    import subprocess
+    import tempfile
+
+    from video_processing.ffmpeg_utils import probe_video_info
+
+    from apps.worker.worker_app.tasks.generation import _render_with_legacy_engine
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmp_path = Path(tmpdir)
+        input_path = tmp_path / "input.mp4"
+        output_path = tmp_path / "output.mp4"
+
+        # 生成 1 秒 30fps 测试视频（带音频）
+        subprocess.run(
+            [
+                "ffmpeg",
+                "-y",
+                "-f",
+                "lavfi",
+                "-i",
+                "color=c=red:s=640x360:d=1:r=30",
+                "-f",
+                "lavfi",
+                "-i",
+                "anullsrc=r=44100:cl=stereo:d=1",
+                "-c:v",
+                "libx264",
+                "-pix_fmt",
+                "yuv420p",
+                "-c:a",
+                "aac",
+                "-shortest",
+                str(input_path),
+            ],
+            check=True,
+            capture_output=True,
+        )
+
+        clip = _TestClip(asset_id="asset-1", duration=1.0)
+        asset_path_map = {"asset-1": input_path}
+
+        duration, file_size = _render_with_legacy_engine(
+            task_id="test-task",
+            virtual_clips=[clip],
+            asset_path_map=asset_path_map,
+            work_dir=tmp_path,
+            output_path=output_path,
+        )
+
+        assert output_path.exists()
+        assert file_size > 0
+        assert duration > 0
+
+        # 旧引擎保持原帧率（30fps），不做 fps 归一化
+        info = probe_video_info(str(output_path))
+        assert abs(info.get("fps", 0) - 30.0) < 0.5
+        assert info.get("width") == 1280
+        assert info.get("height") == 720
+
+
+def test_legacy_engine_two_clips_concat_duration():
+    """多 clip 场景：concat 后时长为两片段之和。"""
+    import subprocess
+    import tempfile
+
+    from video_processing.ffmpeg_utils import probe_duration
+
+    from apps.worker.worker_app.tasks.generation import _render_with_legacy_engine
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmp_path = Path(tmpdir)
+        input1 = tmp_path / "input1.mp4"
+        input2 = tmp_path / "input2.mp4"
+        output_path = tmp_path / "output.mp4"
+
+        for idx, inp in enumerate([input1, input2]):
+            color = "red" if idx == 0 else "blue"
+            subprocess.run(
+                [
+                    "ffmpeg",
+                    "-y",
+                    "-f",
+                    "lavfi",
+                    "-i",
+                    f"color=c={color}:s=640x360:d=1:r=30",
+                    "-f",
+                    "lavfi",
+                    "-i",
+                    "anullsrc=r=44100:cl=stereo:d=1",
+                    "-c:v",
+                    "libx264",
+                    "-pix_fmt",
+                    "yuv420p",
+                    "-c:a",
+                    "aac",
+                    "-shortest",
+                    str(inp),
+                ],
+                check=True,
+                capture_output=True,
+            )
+
+        clip1 = _TestClip(asset_id="asset-1", duration=1.0, clip_type="main", order=0)
+        clip2 = _TestClip(asset_id="asset-2", duration=1.0, clip_type="main", order=1)
+        asset_path_map = {"asset-1": input1, "asset-2": input2}
+
+        duration, file_size = _render_with_legacy_engine(
+            task_id="test-task",
+            virtual_clips=[clip1, clip2],
+            asset_path_map=asset_path_map,
+            work_dir=tmp_path,
+            output_path=output_path,
+        )
+
+        assert output_path.exists()
+        assert file_size > 0
+        assert abs(duration - 2.0) < 0.2
+
+
+def test_legacy_engine_broll_mode_supported():
+    """b_roll 类型的 clip 也被正确识别为主图层并渲染。"""
+    import subprocess
+    import tempfile
+
+    from apps.worker.worker_app.tasks.generation import _render_with_legacy_engine
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmp_path = Path(tmpdir)
+        input_path = tmp_path / "input.mp4"
+        output_path = tmp_path / "output.mp4"
+
+        subprocess.run(
+            [
+                "ffmpeg",
+                "-y",
+                "-f",
+                "lavfi",
+                "-i",
+                "color=c=green:s=640x360:d=1:r=30",
+                "-f",
+                "lavfi",
+                "-i",
+                "anullsrc=r=44100:cl=stereo:d=1",
+                "-c:v",
+                "libx264",
+                "-pix_fmt",
+                "yuv420p",
+                "-c:a",
+                "aac",
+                "-shortest",
+                str(input_path),
+            ],
+            check=True,
+            capture_output=True,
+        )
+
+        clip = _TestClip(
+            asset_id="asset-1",
+            duration=1.0,
+            clip_type="main",
+            config={"role": "b_roll"},
+        )
+        asset_path_map = {"asset-1": input_path}
+
+        duration, file_size = _render_with_legacy_engine(
+            task_id="test-task",
+            virtual_clips=[clip],
+            asset_path_map=asset_path_map,
+            work_dir=tmp_path,
+            output_path=output_path,
+        )
+
+        assert output_path.exists()
+        assert file_size > 0
+        assert duration > 0

tests/unit/test_internal_render_download.py

@@ -0,0 +1,190 @@
+"""渲染结果内部下载接口单元测试。
+
+测试 internal_render 路由的核心逻辑，mock 掉 repository 和 storage 依赖。
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+from app.api.routes.internal_render import (
+    InternalRenderDownloadUrlResponse,
+    InternalRenderTaskVideosResponse,
+    _video_to_item,
+    get_render_task_videos,
+    get_render_video_download_url,
+)
+
+# ── Helpers ────────────────────────────────────────────────────────────────
+
+
+class MockVideo:
+    """模拟 GeneratedVideo 领域对象。"""
+
+    def __init__(self, **kwargs):
+        self.id = kwargs.get("id", "video-1")
+        self.generation_task_id = kwargs.get("generation_task_id", "task-1")
+        self.project_id = kwargs.get("project_id", "proj-1")
+        self.name = kwargs.get("name", "test_video.mp4")
+        self.file_url = kwargs.get("file_url", "videos/test/output.mp4")
+        self.file_size = kwargs.get("file_size", 1024000)
+        self.duration = kwargs.get("duration", 30.5)
+        self.width = kwargs.get("width", 1080)
+        self.height = kwargs.get("height", 1920)
+        self.fps = kwargs.get("fps", 30.0)
+        self.status = kwargs.get("status", "completed")
+
+
+# ── _video_to_item 测试 ────────────────────────────────────────────────────
+
+
+class TestVideoToItem:
+    """测试视频对象转响应项。"""
+
+    def test_basic_conversion(self):
+        video = MockVideo(id="v1", generation_task_id="t1", status="completed")
+        item = _video_to_item(video, "https://oss.example.com/download?v1")
+        assert item.video_id == "v1"
+        assert item.generation_task_id == "t1"
+        assert item.status == "completed"
+        assert item.download_url == "https://oss.example.com/download?v1"
+
+    def test_missing_optional_fields(self):
+        """缺可选字段时返回 None。"""
+        video = MockVideo()
+        # 去掉可选字段
+        del video.file_size
+        del video.duration
+        item = _video_to_item(video, "https://example.com/dl")
+        assert item.file_size is None
+        assert item.duration is None
+        assert item.width == 1080  # 还在
+
+
+# ── 路由函数测试 ────────────────────────────────────────────────────────────
+
+
+class TestGetRenderVideoDownloadUrl:
+    """测试单个视频下载URL接口。"""
+
+    def test_video_exists(self):
+        video = MockVideo(id="v-abc", file_url="videos/abc/out.mp4")
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = video
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://oss.test/signed?v=abc"
+
+        result = get_render_video_download_url(
+            video_id="v-abc",
+            _=True,
+            generated_video_repository=mock_repo,
+            storage_service=mock_storage,
+        )
+
+        assert isinstance(result, InternalRenderDownloadUrlResponse)
+        assert result.video_id == "v-abc"
+        assert result.download_url == "https://oss.test/signed?v=abc"
+        mock_repo.get.assert_called_once_with("v-abc")
+        mock_storage.get_download_url.assert_called_once()
+
+    def test_video_not_found_raises_404(self):
+        from fastapi import HTTPException
+
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = None
+        mock_storage = MagicMock()
+
+        with pytest.raises(HTTPException) as exc_info:
+            get_render_video_download_url(
+                video_id="nonexistent",
+                _=True,
+                generated_video_repository=mock_repo,
+                storage_service=mock_storage,
+            )
+        assert exc_info.value.status_code == 404
+
+    def test_download_url_long_expiry(self):
+        """过期时间应为 24 小时（86400s）。"""
+        video = MockVideo(id="v1")
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = video
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://oss.test/signed"
+
+        get_render_video_download_url(
+            video_id="v1",
+            _=True,
+            generated_video_repository=mock_repo,
+            storage_service=mock_storage,
+        )
+
+        # 验证 expires_seconds=86400
+        call_kwargs = mock_storage.get_download_url.call_args
+        assert call_kwargs.kwargs.get("expires_seconds") == 86400 or call_kwargs[1].get("expires_seconds") == 86400
+
+
+class TestGetRenderTaskVideos:
+    """测试任务视频列表接口。"""
+
+    def test_list_multiple_videos(self):
+        videos = [
+            MockVideo(id="v1", status="completed"),
+            MockVideo(id="v2", status="completed"),
+            MockVideo(id="v3", status="failed"),
+        ]
+        mock_repo = MagicMock()
+        mock_repo.list_by_generation_task.return_value = videos
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://oss.test/signed"
+
+        result = get_render_task_videos(
+            task_id="task-1",
+            status=None,
+            _=True,
+            generated_video_repository=mock_repo,
+            storage_service=mock_storage,
+        )
+
+        assert isinstance(result, InternalRenderTaskVideosResponse)
+        assert result.task_id == "task-1"
+        assert result.count == 3
+        assert len(result.videos) == 3
+
+    def test_filter_by_status(self):
+        videos = [
+            MockVideo(id="v1", status="completed"),
+            MockVideo(id="v2", status="completed"),
+            MockVideo(id="v3", status="failed"),
+        ]
+        mock_repo = MagicMock()
+        mock_repo.list_by_generation_task.return_value = videos
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://oss.test/signed"
+
+        result = get_render_task_videos(
+            task_id="task-1",
+            status="completed",
+            _=True,
+            generated_video_repository=mock_repo,
+            storage_service=mock_storage,
+        )
+
+        assert result.count == 2
+        assert all(v.status == "completed" for v in result.videos)
+
+    def test_empty_task(self):
+        mock_repo = MagicMock()
+        mock_repo.list_by_generation_task.return_value = []
+        mock_storage = MagicMock()
+
+        result = get_render_task_videos(
+            task_id="empty-task",
+            status=None,
+            _=True,
+            generated_video_repository=mock_repo,
+            storage_service=mock_storage,
+        )
+
+        assert result.count == 0
+        assert result.videos == []

tests/unit/test_oss_upload_crash_fix.py

@@ -0,0 +1,236 @@
+"""P0-staging：OSS 上传崩溃修复测试.
+
+测试：
+1. oss_bucket() 传递 connect_timeout 参数
+2. upload_to_oss() 小文件走 put_object_from_file，大文件走分片上传
+3. upload_to_oss() 超时保护（超过总超时返回 None）
+4. upload_to_oss() 异常时返回 None
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import time
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# ── oss_bucket connect_timeout 测试 ───────────────────────────────────────────
+
+
+class TestOSSBucketConnectTimeout:
+    """测试 oss_bucket() 传递 connect_timeout 参数."""
+
+    def test_oss_bucket_has_connect_timeout(self):
+        """oss_bucket 应传递 connect_timeout=10s 参数."""
+        from video_processing.oss_helpers import oss_bucket
+
+        mock_bucket_instance = MagicMock()
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
+        ):
+            bucket = oss_bucket()
+
+            assert bucket is mock_bucket_instance
+            # 验证 connect_timeout 关键字参数
+            call_kwargs = mock_bucket_cls.call_args[1]
+            assert "connect_timeout" in call_kwargs, "oss_bucket 应传递 connect_timeout 参数"
+            assert (
+                call_kwargs["connect_timeout"] == 10
+            ), f"connect_timeout 应为 10，实际为 {call_kwargs['connect_timeout']}"
+
+    def test_oss_bucket_no_config_returns_none(self):
+        """OSS 配置缺失时返回 None."""
+        from video_processing.oss_helpers import oss_bucket
+
+        with patch.dict(os.environ, {}, clear=True):
+            bucket = oss_bucket()
+            assert bucket is None
+
+
+# ── upload_to_oss 分片上传测试 ────────────────────────────────────────────────
+
+
+class TestUploadToOSSMultipart:
+    """测试 upload_to_oss() 根据文件大小选择上传方式."""
+
+    def _create_temp_file(self, size_bytes: int) -> Path:
+        """创建指定大小的临时文件."""
+        tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+        tmp.write(b"x" * size_bytes)
+        tmp.close()
+        return Path(tmp.name)
+
+    def test_small_file_uses_put_object(self):
+        """小文件（<100MB）走 put_object_from_file."""
+        from video_processing.oss_helpers import upload_to_oss
+
+        small_file = self._create_temp_file(10 * 1024 * 1024)  # 10MB
+        try:
+            mock_bucket = MagicMock()
+
+            with (
+                patch.dict(
+                    os.environ,
+                    {
+                        "OSS_ACCESS_KEY_ID": "test-key",
+                        "OSS_ACCESS_KEY_SECRET": "test-secret",
+                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                        "OSS_BUCKET_NAME": "test-bucket",
+                    },
+                ),
+                patch("video_processing.oss_helpers.oss2.Auth"),
+                patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
+                patch("video_processing.oss_helpers.oss2.resumable_upload") as mock_resumable,
+            ):
+                url = upload_to_oss(small_file, "test/small.mp4")
+
+                # 验证调用了 put_object_from_file
+                mock_bucket.put_object_from_file.assert_called_once()
+                # 验证没调用分片上传
+                mock_resumable.assert_not_called()
+                # 验证返回 URL
+                assert url == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/test/small.mp4"
+        finally:
+            small_file.unlink()
+
+    def test_large_file_uses_resumable_upload(self):
+        """大文件（>=100MB）走 resumable_upload 分片上传."""
+        from video_processing.oss_helpers import upload_to_oss
+
+        large_file = self._create_temp_file(100 * 1024 * 1024)  # 100MB
+        try:
+            mock_bucket = MagicMock()
+
+            with (
+                patch.dict(
+                    os.environ,
+                    {
+                        "OSS_ACCESS_KEY_ID": "test-key",
+                        "OSS_ACCESS_KEY_SECRET": "test-secret",
+                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                        "OSS_BUCKET_NAME": "test-bucket",
+                    },
+                ),
+                patch("video_processing.oss_helpers.oss2.Auth"),
+                patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
+                patch("video_processing.oss_helpers.oss2.resumable_upload") as mock_resumable,
+            ):
+                url = upload_to_oss(large_file, "test/large.mp4")
+
+                # 验证调用了分片上传
+                mock_resumable.assert_called_once()
+                # 验证没调用 put_object_from_file
+                mock_bucket.put_object_from_file.assert_not_called()
+                # 验证分片参数
+                call_kwargs = mock_resumable.call_args[1]
+                assert call_kwargs["multipart_threshold"] == 100 * 1024 * 1024
+                assert call_kwargs["part_size"] == 8 * 1024 * 1024
+                assert call_kwargs["num_threads"] == 3
+                # 验证返回 URL
+                assert url == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/test/large.mp4"
+        finally:
+            large_file.unlink()
+
+
+# ── upload_to_oss 超时测试 ────────────────────────────────────────────────────
+
+
+class TestUploadToOSSTimeout:
+    """测试 upload_to_oss() 超时保护."""
+
+    def test_upload_timeout_returns_none(self):
+        """上传超过总超时时返回 None."""
+        from video_processing.oss_helpers import upload_to_oss
+
+        small_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+        small_file.write(b"x" * 1024)  # 1KB
+        small_file.close()
+        file_path = Path(small_file.name)
+
+        def slow_upload(*args, **kwargs):
+            """模拟慢速上传，超过超时时间."""
+            time.sleep(2)
+
+        mock_bucket = MagicMock()
+        mock_bucket.put_object_from_file.side_effect = slow_upload
+
+        try:
+            with (
+                patch.dict(
+                    os.environ,
+                    {
+                        "OSS_ACCESS_KEY_ID": "test-key",
+                        "OSS_ACCESS_KEY_SECRET": "test-secret",
+                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                        "OSS_BUCKET_NAME": "test-bucket",
+                    },
+                ),
+                patch("video_processing.oss_helpers.oss2.Auth"),
+                patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
+                patch("video_processing.oss_helpers.OSS_UPLOAD_TOTAL_TIMEOUT", 1),  # 1秒超时
+            ):
+                url = upload_to_oss(file_path, "test/slow.mp4")
+                # 超时应返回 None
+                assert url is None, "上传超时应返回 None"
+        finally:
+            file_path.unlink()
+
+    def test_upload_exception_returns_none(self):
+        """上传异常时返回 None."""
+        from video_processing.oss_helpers import upload_to_oss
+
+        small_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+        small_file.write(b"x" * 1024)
+        small_file.close()
+        file_path = Path(small_file.name)
+
+        mock_bucket = MagicMock()
+        mock_bucket.put_object_from_file.side_effect = RuntimeError("Network error")
+
+        try:
+            with (
+                patch.dict(
+                    os.environ,
+                    {
+                        "OSS_ACCESS_KEY_ID": "test-key",
+                        "OSS_ACCESS_KEY_SECRET": "test-secret",
+                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                        "OSS_BUCKET_NAME": "test-bucket",
+                    },
+                ),
+                patch("video_processing.oss_helpers.oss2.Auth"),
+                patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
+            ):
+                url = upload_to_oss(file_path, "test/error.mp4")
+                assert url is None, "上传异常应返回 None"
+        finally:
+            file_path.unlink()
+
+    def test_upload_no_bucket_returns_none(self):
+        """OSS 未配置时返回 None."""
+        from video_processing.oss_helpers import upload_to_oss
+
+        small_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+        small_file.write(b"x" * 1024)
+        small_file.close()
+        file_path = Path(small_file.name)
+
+        try:
+            with patch.dict(os.environ, {}, clear=True):
+                url = upload_to_oss(file_path, "test/noconfig.mp4")
+                assert url is None
+        finally:
+            file_path.unlink()

tests/unit/test_unified_render_service.py

@@ -1399,3 +1399,239 @@ class TestAudioMixing:
         assert r1 is True and r2 is True and r3 is True
         # 实际只探测了 1 次
         assert mock_probe.call_count == 1
+
+
+# ── 测试 stream copy 流拷贝优化 ───────────────────────────────────────────────
+
+
+class TestStreamCopy:
+    """stream copy 流拷贝优化测试。"""
+
+    def _make_single_clip_service(self):
+        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
+        svc = _make_service(clips)
+        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)
+        return svc, resolved[0], layers
+
+    def test_can_use_stream_copy_all_conditions_met(self):
+        """所有条件满足 → 可以 stream copy。"""
+        svc, clip, layers = self._make_single_clip_service()
+        probe_result = {
+            "width": 1280,
+            "height": 720,
+            "fps": 25.0,
+            "video_codec": "h264",
+            "pix_fmt": "yuv420p",
+            "duration": 5.0,
+            "has_audio": True,
+            "audio_codec": "aac",
+        }
+        with patch(
+            "video_processing.unified_render_service.probe_video_info",
+            return_value=probe_result,
+        ):
+            can_copy, reason = svc._can_use_stream_copy(clip, ass_path=None, video_duration=0)
+        assert can_copy is True
+        assert "所有条件满足" in reason
+
+    def test_cannot_copy_with_subtitles(self):
+        """有字幕 → 不能 stream copy。"""
+        svc, clip, layers = self._make_single_clip_service()
+        can_copy, reason = svc._can_use_stream_copy(clip, ass_path=Path("/tmp/sub.ass"), video_duration=0)
+        assert can_copy is False
+        assert "字幕" in reason
+
+    def test_cannot_copy_wrong_codec(self):
+        """编码不是 h264 → 不能 stream copy。"""
+        svc, clip, layers = self._make_single_clip_service()
+        probe_result = {
+            "width": 1280,
+            "height": 720,
+            "fps": 25.0,
+            "video_codec": "hevc",
+            "pix_fmt": "yuv420p",
+            "duration": 5.0,
+            "has_audio": True,
+            "audio_codec": "aac",
+        }
+        with patch(
+            "video_processing.unified_render_service.probe_video_info",
+            return_value=probe_result,
+        ):
+            can_copy, reason = svc._can_use_stream_copy(clip, ass_path=None, video_duration=0)
+        assert can_copy is False
+        assert "编码" in reason
+
+    def test_cannot_copy_wrong_resolution(self):
+        """分辨率不匹配 → 不能 stream copy。"""
+        svc, clip, layers = self._make_single_clip_service()
+        probe_result = {
+            "width": 1920,
+            "height": 1080,
+            "fps": 25.0,
+            "video_codec": "h264",
+            "pix_fmt": "yuv420p",
+            "duration": 5.0,
+            "has_audio": True,
+            "audio_codec": "aac",
+        }
+        with patch(
+            "video_processing.unified_render_service.probe_video_info",
+            return_value=probe_result,
+        ):
+            can_copy, reason = svc._can_use_stream_copy(clip, ass_path=None, video_duration=0)
+        assert can_copy is False
+        assert "分辨率" in reason
+
+    def test_cannot_copy_wrong_fps(self):
+        """帧率不匹配 → 不能 stream copy。"""
+        svc, clip, layers = self._make_single_clip_service()
+        probe_result = {
+            "width": 1280,
+            "height": 720,
+            "fps": 30.0,
+            "video_codec": "h264",
+            "pix_fmt": "yuv420p",
+            "duration": 5.0,
+            "has_audio": True,
+            "audio_codec": "aac",
+        }
+        with patch(
+            "video_processing.unified_render_service.probe_video_info",
+            return_value=probe_result,
+        ):
+            can_copy, reason = svc._can_use_stream_copy(clip, ass_path=None, video_duration=0)
+        assert can_copy is False
+        assert "帧率" in reason
+
+    def test_cannot_copy_wrong_pix_fmt(self):
+        """像素格式不匹配 → 不能 stream copy。"""
+        svc, clip, layers = self._make_single_clip_service()
+        probe_result = {
+            "width": 1280,
+            "height": 720,
+            "fps": 25.0,
+            "video_codec": "h264",
+            "pix_fmt": "yuv422p",
+            "duration": 5.0,
+            "has_audio": True,
+            "audio_codec": "aac",
+        }
+        with patch(
+            "video_processing.unified_render_service.probe_video_info",
+            return_value=probe_result,
+        ):
+            can_copy, reason = svc._can_use_stream_copy(clip, ass_path=None, video_duration=0)
+        assert can_copy is False
+        assert "像素格式" in reason
+
+    def test_try_render_stream_copy_success(self):
+        """stream copy 渲染成功 → 返回 True。"""
+        svc, clip, layers = self._make_single_clip_service()
+        probe_result = {
+            "width": 1280,
+            "height": 720,
+            "fps": 25.0,
+            "video_codec": "h264",
+            "pix_fmt": "yuv420p",
+            "duration": 5.0,
+            "has_audio": True,
+            "audio_codec": "aac",
+        }
+        output_path = Path("/tmp/test_output.mp4")
+
+        def fake_stat():
+            m = MagicMock()
+            m.st_size = 1024000
+            return m
+
+        with (
+            patch(
+                "video_processing.unified_render_service.probe_video_info",
+                return_value=probe_result,
+            ),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+            patch("pathlib.Path.exists", return_value=True),
+            patch("pathlib.Path.stat", side_effect=fake_stat),
+        ):
+            result = svc._try_render_stream_copy(layers, output_path, ass_path=None, video_duration=0)
+
+        assert result is True
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        assert "-c:v" in cmd
+        assert "copy" in cmd
+        assert "-c:a" in cmd
+
+    def test_try_render_stream_copy_fallback_on_ffmpeg_error(self):
+        """stream copy FFmpeg 失败 → 返回 False（调用方回退到重编码）。"""
+        svc, clip, layers = self._make_single_clip_service()
+        probe_result = {
+            "width": 1280,
+            "height": 720,
+            "fps": 25.0,
+            "video_codec": "h264",
+            "pix_fmt": "yuv420p",
+            "duration": 5.0,
+            "has_audio": True,
+            "audio_codec": "aac",
+        }
+        output_path = Path("/tmp/test_output.mp4")
+
+        import subprocess as sp
+
+        with (
+            patch(
+                "video_processing.unified_render_service.probe_video_info",
+                return_value=probe_result,
+            ),
+            patch(
+                "video_processing.unified_render_service.run_ffmpeg",
+                side_effect=sp.CalledProcessError(1, ["ffmpeg"], stderr="copy failed"),
+            ),
+            patch("pathlib.Path.exists", return_value=False),
+        ):
+            result = svc._try_render_stream_copy(layers, output_path, ass_path=None, video_duration=0)
+
+        assert result is False
+
+    def test_render_uses_stream_copy_when_eligible(self):
+        """完整渲染流程：满足条件时走 stream copy。"""
+        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
+        svc = _make_service(clips)
+
+        probe_result = {
+            "width": 1280,
+            "height": 720,
+            "fps": 25.0,
+            "video_codec": "h264",
+            "pix_fmt": "yuv420p",
+            "duration": 5.0,
+            "has_audio": True,
+            "audio_codec": "aac",
+        }
+
+        def fake_stat():
+            m = MagicMock()
+            m.st_size = 1024000
+            return m
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch(
+                "video_processing.unified_render_service.probe_video_info",
+                return_value=probe_result,
+            ),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+            patch("pathlib.Path.stat", side_effect=fake_stat),
+            patch("shutil.copy2"),
+        ):
+            result = svc.render()
+
+        assert mock_run.call_count == 1
+        cmd = mock_run.call_args[0][0]
+        assert "copy" in cmd
+        assert isinstance(result.output_path, Path)