style: black format edit_plan_generation.py

3 个修复：
1. render_edit_plan 素材下载路径错误：直接传 asset_id 给 download_asset，
   改为先查 assets 表拿 file_url(storage_key) 再下载
2. 总时长=0.0s 日志误导：_build_plan_and_clips_from_task 中 _VirtualClip
   的 duration 未赋值导致日志永远显示 0s，改为用 probe_duration 预探测
3. video_processing/__init__.py 导入链副作用：顶层导入 render_adapter 等
   重依赖模块，导致只想用 oss_helpers/ffmpeg_utils 等轻量工具的地方也
   被迫拉进 packages/DB 全量依赖。改为只导出轻量工具模块，渲染组件
   按需从子模块导入

.gitea/workflows/ci-cd.yml

@@ -563,6 +563,15 @@ jobs:
             -w /workspace/apps/web \
             docker.m.daocloud.io/library/node:20 \
             sh -lc 'npx vitest run src/test'
+
+      - name: Notify CI failure
+        if: failure()
+        shell: sh
+        run: |
+          set +e
+          echo "=== CI 失败通知 ==="
+          FAILED_JOB="Frontend Lint" python3 scripts/ci_notify_failure.py
+
   deploy-staging:
     name: Build & Push Staging (Watchtower auto-deploy)
     runs-on: saas
@@ -694,6 +703,23 @@ jobs:
         echo "Branch: ${GITHUB_REF_NAME}"
         echo "Commit: ${GITHUB_SHA}"
 
+    - name: Notify CI success
+      if: success()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 成功通知 ==="
+        SUCCESS_JOB="Staging部署成功" python3 scripts/ci_notify_success.py
+
+    - name: Notify CI failure
+      if: failure()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 失败通知 ==="
+        FAILED_JOB="Build & Push Staging (Watchtower auto-deploy)" python3 scripts/ci_notify_failure.py
+
+
 
   staging-e2e:
     name: Staging E2E Tests
@@ -762,6 +788,15 @@ jobs:
           git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
           sh -lc "npm ci && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts"
 
+    - name: Notify CI failure
+      if: failure()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 失败通知 ==="
+        FAILED_JOB="Staging E2E Tests" python3 scripts/ci_notify_failure.py
+
+
   staging-api-tests:
     name: Staging API Integration Tests
     runs-on: saas
@@ -827,6 +862,15 @@ jobs:
           git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
           sh -lc 'npm ci && npx playwright test --reporter=line e2e/test_auth.spec.ts e2e/test_asset.spec.ts e2e/test_project.spec.ts'
 
+    - name: Notify CI failure
+      if: failure()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 失败通知 ==="
+        FAILED_JOB="Staging API Integration Tests" python3 scripts/ci_notify_failure.py
+
+
 
   build-production-runtime-images:
     name: Build Production Runtime Images
@@ -907,6 +951,15 @@ jobs:
         echo "Disk usage after cleanup:"
         df -h / | tail -1
 
+    - name: Notify CI failure
+      if: failure()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 失败通知 ==="
+        FAILED_JOB="Build Production Runtime Images" python3 scripts/ci_notify_failure.py
+
+
   deploy-production:
     name: Deploy Production
     runs-on: saas
@@ -972,6 +1025,23 @@ jobs:
 
         echo "$DEPLOY_B64" | base64 -d | ssh -p 22222 -i "$key_path" "$production_user@$production_host" "IMAGE_TAG='${GITHUB_REF_NAME}' REGISTRY_TOKEN='${REGISTRY_TOKEN}' sh"
 
+    - name: Notify CI success
+      if: success()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 成功通知 ==="
+        SUCCESS_JOB="生产部署成功" python3 scripts/ci_notify_success.py
+
+    - name: Notify CI failure
+      if: failure()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 失败通知 ==="
+        FAILED_JOB="Deploy Production" python3 scripts/ci_notify_failure.py
+
+
   production-e2e:
     name: Production Browser E2E
     runs-on: saas
@@ -1040,3 +1110,12 @@ jobs:
           -w /workspace/apps/web \
           git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
           sh -lc 'npm ci && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts e2e/core-titles.spec.ts'
+
+    - name: Notify CI failure
+      if: failure()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 失败通知 ==="
+        FAILED_JOB="Production Browser E2E" python3 scripts/ci_notify_failure.py
+

.gitea/workflows/debug-cmd-agent.yml

@@ -1,44 +0,0 @@
-name: Debug CMD Agent
-
-on:
-  push:
-    branches:
-      - 'debug/cmd-agent'
-
-jobs:
-  debug:
-    name: Debug CMD Agent
-    runs-on: host
-    timeout-minutes: 5
-    steps:
-      - name: Diagnose
-        shell: bash
-        run: |
-          set +e
-          echo "=== 1. CMD Agent config ==="
-          cat /opt/xiaoxia-cmd-agent/config.json 2>/dev/null || cat /opt/xiaoxia-cmd-agent/config.yaml 2>/dev/null || echo "no config found"
-          ls -la /opt/xiaoxia-cmd-agent/ 2>/dev/null
-
-          echo ""
-          echo "=== 2. CMD Agent process ==="
-          ps aux | grep cmd-agent | grep -v grep
-
-          echo ""
-          echo "=== 3. Local curl test (127.0.0.1:18888) ==="
-          curl -s -X POST http://127.0.0.1:18888/cmd-agent/exec \
-            -H "Authorization: Bearer xsa-f2778a6953d59948cd1e5be4d99f60f7" \
-            -H "Content-Type: application/json" \
-            -d '{"command":"hostname"}' 2>&1 || echo "FAILED"
-
-          echo ""
-          echo "=== 4. Nginx config for cmd-agent ==="
-          grep -r "cmd-agent" /etc/nginx/sites-enabled/ 2>/dev/null || \
-          grep -r "cmd-agent" /etc/nginx/conf.d/ 2>/dev/null || \
-          echo "no nginx cmd-agent config found"
-
-          echo ""
-          echo "=== 5. Nginx access log (last 5 lines) ==="
-          tail -5 /var/log/nginx/access.log 2>/dev/null | grep cmd || echo "no log"
-
-          echo ""
-          echo "=== DONE ==="

.gitea/workflows/fix-cmd-agent-auth.yml

@@ -1,46 +0,0 @@
-name: Fix CMD Agent Auth
-
-on:
-  push:
-    branches:
-      - 'debug/cmd-agent'
-
-jobs:
-  fix:
-    runs-on: host
-    steps:
-      - name: 验证不带Bearer
-        run: |
-          curl -s -w "\nHTTP_CODE:%{http_code}" http://127.0.0.1:18888/status -H "Authorization: xsa-f2778a6953d59948cd1e5be4d99f60f7"
-      - name: 验证带Bearer（应该失败）
-        run: |
-          curl -s -w "\nHTTP_CODE:%{http_code}" http://127.0.0.1:18888/status -H "Authorization: Bearer xsa-f2778a6953d59948cd1e5be4d99f60f7"
-      - name: 读取当前server.py的check_auth
-        run: |
-          grep -A 5 "def check_auth" /opt/xiaoxia-cmd-agent/server.py
-      - name: 修复check_auth函数
-        run: |
-          cp /opt/xiaoxia-cmd-agent/server.py /opt/xiaoxia-cmd-agent/server.py.bak
-          sed -i '/def check_auth/,/return True/{
-            /def check_auth/a\        t = self.headers.get("Authorization", "")
-            /if t != AUTH_TOKEN/i\        if t.startswith("Bearer "):\n            t = t[7:]
-          }' /opt/xiaoxia-cmd-agent/server.py
-          echo "Done via sed"
-      - name: 验证修复后的check_auth
-        run: |
-          grep -A 8 "def check_auth" /opt/xiaoxia-cmd-agent/server.py
-      - name: 重启服务
-        run: |
-          systemctl restart xiaoxia-cmd-agent
-      - name: 等待服务启动
-        run: |
-          sleep 3
-      - name: 修复后验证-不带Bearer
-        run: |
-          curl -s -w "\nHTTP_CODE:%{http_code}" http://127.0.0.1:18888/status -H "Authorization: xsa-f2778a6953d59948cd1e5be4d99f60f7"
-      - name: 修复后验证-带Bearer
-        run: |
-          curl -s -w "\nHTTP_CODE:%{http_code}" http://127.0.0.1:18888/status -H "Authorization: Bearer xsa-f2778a6953d59948cd1e5be4d99f60f7"
-      - name: 公网路径验证
-        run: |
-          curl -sk -w "\nHTTP_CODE:%{http_code}" https://127.0.0.1/cmd-agent/status -H "Authorization: Bearer xsa-f2778a6953d59948cd1e5be4d99f60f7"

.gitea/workflows/read-auth.yml

@@ -1,38 +0,0 @@
-name: Read Auth Logic
-
-on:
-  push:
-    branches:
-      - 'debug/cmd-agent'
-
-jobs:
-  read:
-    name: Read check_auth logic
-    runs-on: host
-    timeout-minutes: 3
-    steps:
-      - name: Read
-        shell: bash
-        run: |
-          echo "=== Full server.py (lines 1-50) ==="
-          sed -n '1,50p' /opt/xiaoxia-cmd-agent/server.py
-          echo ""
-          echo "=== Lines 120-160 (startup logic) ==="
-          sed -n '120,160p' /opt/xiaoxia-cmd-agent/server.py
-          echo ""
-          echo "=== Test with X-Token header ==="
-          curl -s -X POST http://127.0.0.1:18888/cmd-agent/exec \
-            -H "X-Token: $(cat /etc/xiaoxia-cmd-agent.token)" \
-            -H "Content-Type: application/json" \
-            -d '{"command":"hostname"}'
-          echo ""
-          echo "=== Test with token in query string ==="
-          curl -s -X POST "http://127.0.0.1:18888/cmd-agent/exec?token=$(cat /etc/xiaoxia-cmd-agent.token)" \
-            -H "Content-Type: application/json" \
-            -d '{"command":"hostname"}'
-          echo ""
-          echo "=== Check if path is /exec not /cmd-agent/exec ==="
-          curl -s -X POST http://127.0.0.1:18888/exec \
-            -H "Authorization: Bearer $(cat /etc/xiaoxia-cmd-agent.token)" \
-            -H "Content-Type: application/json" \
-            -d '{"command":"hostname"}'

.gitea/workflows/read-cmd-agent.yml

@@ -1,27 +0,0 @@
-name: Read CMD Agent Source
-
-on:
-  push:
-    branches:
-      - 'debug/cmd-agent'
-
-jobs:
-  read:
-    name: Read CMD Agent server.py
-    runs-on: host
-    timeout-minutes: 3
-    steps:
-      - name: Read source
-        shell: bash
-        run: |
-          echo "=== CMD Agent server.py (first 80 lines) ==="
-          head -80 /opt/xiaoxia-cmd-agent/server.py
-          echo ""
-          echo "=== Token-related lines ==="
-          grep -n -i "token\|auth\|secret\|key" /opt/xiaoxia-cmd-agent/server.py
-          echo ""
-          echo "=== Systemd service config ==="
-          cat /etc/systemd/system/xiaoxia-cmd-agent.service 2>/dev/null || echo "no systemd service"
-          echo ""
-          echo "=== Environment variables from process ==="
-          cat /proc/1034/environ 2>/dev/null | tr '\0' '\n' | grep -i "token\|auth\|secret\|key" || echo "no env vars found"

.gitea/workflows/read-token.yml

@@ -1,30 +0,0 @@
-name: Read CMD Agent Token
-
-on:
-  push:
-    branches:
-      - 'debug/cmd-agent'
-
-jobs:
-  read:
-    name: Read Real Token
-    runs-on: host
-    timeout-minutes: 3
-    steps:
-      - name: Read
-        shell: bash
-        run: |
-          echo "=== Real CMD Agent Token ==="
-          cat /etc/xiaoxia-cmd-agent.token
-          echo ""
-          echo "=== Test with real token ==="
-          curl -s -X POST http://127.0.0.1:18888/cmd-agent/exec \
-            -H "Authorization: Bearer $(cat /etc/xiaoxia-cmd-agent.token)" \
-            -H "Content-Type: application/json" \
-            -d '{"command":"hostname && whoami"}'
-          echo ""
-          echo "=== Nginx config for cmd-agent (full) ==="
-          sed -n '/cmd-agent/,/}/p' /etc/nginx/sites-enabled/00-xiaoxia-saas | head -20
-          echo ""
-          echo "=== All listening ports ==="
-          ss -tlnp | head -20

apps/api/app/api/router.py

@@ -8,6 +8,7 @@ from app.api.routes.dashboard import router as dashboard_router
 from app.api.routes.duplication import router as duplication_router
 from app.api.routes.edit_plans import router as edit_plans_router
 from app.api.routes.edit_templates import router as edit_templates_router
+from app.api.routes.feature_flags import router as feature_flags_router
 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
@@ -151,3 +152,7 @@ api_router.include_router(
     prefix="/tts",
     tags=["TTS"],
 )
+api_router.include_router(
+    feature_flags_router,
+    tags=["Internal"],
+)

apps/api/app/api/routes/feature_flags.py

@@ -0,0 +1,195 @@
+"""Feature Flag 内部管理接口。
+
+通过内部 API Key 鉴权，支持查看和修改 Feature Flag 配置。
+主要用于灰度发布期间的动态开关控制。
+
+API:
+  GET  /api/v1/internal/feature-flags          - 列出所有 flag
+  GET  /api/v1/internal/feature-flags/{name}   - 查看单个 flag
+  PUT  /api/v1/internal/feature-flags/{name}   - 设置 flag 配置
+  DELETE /api/v1/internal/feature-flags/{name} - 删除 flag
+
+鉴权：X-API-Key header，走内部 API Key 验证
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from app.api.routes.auth import _verify_internal_api_key
+from app.config import settings
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from pydantic import BaseModel, Field
+
+from packages.adapters.redis.feature_flag_store import (
+    FEATURE_FLAG_REDIS_PREFIX,
+    FeatureFlagConfig,
+    RedisFeatureFlagStore,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/internal/feature-flags", tags=["Internal"])
+
+# 允许管理的 flag 白名单（防止误操作其他系统 flag）
+ALLOWED_FLAGS = {
+    "render_engine",
+}
+
+
+def _get_feature_flag_store() -> RedisFeatureFlagStore:
+    """获取 Feature Flag 存储实例。"""
+    return RedisFeatureFlagStore(redis_url=settings.REDIS_URL)
+
+
+class FeatureFlagUpdateRequest(BaseModel):
+    """Feature Flag 更新请求体。"""
+
+    enabled: bool = Field(..., description="是否启用")
+    percentage: int = Field(0, ge=0, le=100, description="灰度百分比 (0-100)")
+    whitelist: list[str] = Field(default_factory=list, description="白名单列表（如 user_id）")
+
+
+class FeatureFlagResponse(BaseModel):
+    """Feature Flag 响应。"""
+
+    name: str
+    enabled: bool
+    percentage: int
+    whitelist: list[str]
+
+    @classmethod
+    def from_config(cls, config: FeatureFlagConfig) -> "FeatureFlagResponse":
+        return cls(
+            name=config.name,
+            enabled=config.enabled,
+            percentage=config.percentage,
+            whitelist=sorted(config.whitelist),
+        )
+
+
+class FeatureFlagCheckResponse(BaseModel):
+    """Flag 激活检查响应。"""
+
+    name: str
+    active: bool
+    identifier: Optional[str] = None
+
+
+def _validate_flag_name(name: str) -> None:
+    """校验 flag 名称是否在允许列表中。"""
+    if name not in ALLOWED_FLAGS:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Unsupported flag: {name}. Allowed: {sorted(ALLOWED_FLAGS)}",
+        )
+
+
+@router.get("", response_model=list[FeatureFlagResponse])
+async def list_feature_flags(
+    _: bool = Depends(_verify_internal_api_key),
+    store: RedisFeatureFlagStore = Depends(_get_feature_flag_store),
+):
+    """列出所有 Feature Flag。"""
+    try:
+        flags = store.list_all()
+        # 同时返回预定义的 flag（即使未设置也显示默认值）
+        result = []
+        for name in sorted(ALLOWED_FLAGS):
+            config = flags.get(name) or FeatureFlagConfig(name=name, enabled=False)
+            result.append(FeatureFlagResponse.from_config(config))
+        # 加上已存在但不在白名单中的 flag（只读展示）
+        for name, config in flags.items():
+            if name not in ALLOWED_FLAGS:
+                result.append(FeatureFlagResponse.from_config(config))
+        return sorted(result, key=lambda x: x.name)
+    except Exception as exc:
+        logger.error("Failed to list feature flags: %s", exc)
+        raise HTTPException(status_code=500, detail=f"Failed to list flags: {exc}")
+
+
+@router.get("/{name}", response_model=FeatureFlagResponse)
+async def get_feature_flag(
+    name: str,
+    _: bool = Depends(_verify_internal_api_key),
+    store: RedisFeatureFlagStore = Depends(_get_feature_flag_store),
+):
+    """获取单个 Feature Flag 配置。"""
+    try:
+        config = store.get(name)
+        return FeatureFlagResponse.from_config(config)
+    except Exception as exc:
+        logger.error("Failed to get feature flag %s: %s", name, exc)
+        raise HTTPException(status_code=500, detail=f"Failed to get flag: {exc}")
+
+
+@router.get("/{name}/check", response_model=FeatureFlagCheckResponse)
+async def check_feature_flag(
+    name: str,
+    identifier: Optional[str] = Query(None, description="标识符，如 user_id"),
+    _: bool = Depends(_verify_internal_api_key),
+    store: RedisFeatureFlagStore = Depends(_get_feature_flag_store),
+):
+    """检查某个标识符是否命中 Feature Flag。"""
+    try:
+        active = store.is_active(name, identifier=identifier)
+        return FeatureFlagCheckResponse(name=name, active=active, identifier=identifier)
+    except Exception as exc:
+        logger.error("Failed to check feature flag %s: %s", name, exc)
+        raise HTTPException(status_code=500, detail=f"Failed to check flag: {exc}")
+
+
+@router.put("/{name}", response_model=FeatureFlagResponse)
+async def update_feature_flag(
+    name: str,
+    request: FeatureFlagUpdateRequest,
+    _: bool = Depends(_verify_internal_api_key),
+    store: RedisFeatureFlagStore = Depends(_get_feature_flag_store),
+):
+    """更新 Feature Flag 配置。
+
+    只允许修改 ALLOWED_FLAGS 列表中的 flag。
+    """
+    _validate_flag_name(name)
+
+    try:
+        config = FeatureFlagConfig(
+            name=name,
+            enabled=request.enabled,
+            percentage=request.percentage,
+            whitelist=set(request.whitelist),
+        )
+        store.set(config)
+        logger.info(
+            "Feature flag updated: name=%s enabled=%s percentage=%d whitelist=%d",
+            name,
+            config.enabled,
+            config.percentage,
+            len(config.whitelist),
+        )
+        return FeatureFlagResponse.from_config(config)
+    except Exception as exc:
+        logger.error("Failed to update feature flag %s: %s", name, exc)
+        raise HTTPException(status_code=500, detail=f"Failed to update flag: {exc}")
+
+
+@router.delete("/{name}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_feature_flag(
+    name: str,
+    _: bool = Depends(_verify_internal_api_key),
+    store: RedisFeatureFlagStore = Depends(_get_feature_flag_store),
+):
+    """删除 Feature Flag。
+
+    只允许删除 ALLOWED_FLAGS 列表中的 flag。
+    """
+    _validate_flag_name(name)
+
+    try:
+        deleted = store.delete(name)
+        logger.info("Feature flag deleted: name=%s deleted=%s", name, deleted)
+        return None
+    except Exception as exc:
+        logger.error("Failed to delete feature flag %s: %s", name, exc)
+        raise HTTPException(status_code=500, detail=f"Failed to delete flag: {exc}")

apps/api/app/config.py

@@ -109,6 +109,9 @@ class Settings(BaseSettings):
     LOG_LEVEL: str = "INFO"
     CORS_ORIGINS_RAW: str = "http://localhost:3000,http://localhost:5173,http://localhost:8000"
 
+    # 渲染引擎选择：legacy=旧VideoComposeService，unified=新UnifiedRenderService
+    RENDER_ENGINE: str = "legacy"
+
     model_config = SettingsConfigDict(
         env_file=".env",
         env_file_encoding="utf-8",

apps/worker/video_processing/__init__.py

@@ -1,18 +1,17 @@
 """
 视频处理模块
+
+轻量工具（ffmpeg_utils / oss_helpers / dedup_helpers）顶层直接导出，
+无额外依赖。渲染相关组件（UnifiedRenderService / RenderAdapter /
+VideoProcessor 等）按需从子模块导入，避免 __init__ 阶段引入
+packages / DB 等重依赖。
 """
 
-# 共享工具模块（供 editing_modes / generation / edit_plan_generation 等复用）
+# 共享工具模块（零外部依赖，供 editing_modes / generation / edit_plan_generation 等复用）
 from . import dedup_helpers, ffmpeg_utils, oss_helpers
-from .processor import VideoProcessor, VideoResult
-from .unified_render_service import RenderResult, UnifiedRenderService
 
 __all__ = [
-    "VideoProcessor",
-    "VideoResult",
     "ffmpeg_utils",
     "oss_helpers",
     "dedup_helpers",
-    "UnifiedRenderService",
-    "RenderResult",
 ]

apps/worker/video_processing/editing_modes.py

@@ -1,657 +0,0 @@
-"""
-视频剪辑模式处理器
-支持四种剪辑模式：一镜到底、画中画、口播、口播+画中画
-"""
-
-import logging
-import os
-import sys
-import tempfile
-from dataclasses import dataclass
-
-if sys.version_info >= (3, 11):
-    from enum import StrEnum
-else:
-    from enum import Enum
-
-    class StrEnum(str, Enum):
-        pass
-
-
-from pathlib import Path
-from typing import Optional
-
-from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_video_info, run_ffmpeg
-
-logger = logging.getLogger(__name__)
-
-
-# 从 domain 层导入 EditingMode，避免重复定义
-from packages.domain.editing_mode import EditingMode
-
-
-class PIPPosition(StrEnum):
-    """画中画位置枚举"""
-
-    TOP_LEFT = "top_left"
-    TOP_RIGHT = "top_right"
-    BOTTOM_LEFT = "bottom_left"
-    BOTTOM_RIGHT = "bottom_right"
-
-
-@dataclass
-class EditingModeConfig:
-    """剪辑模式配置"""
-
-    mode: EditingMode
-    output_width: int = 1280
-    output_height: int = 720
-    output_fps: int = 25
-    pip_position: PIPPosition = PIPPosition.TOP_RIGHT
-    pip_scale: float = 0.25  # 画中画占主画面的比例
-    transition_duration: float = 0.5  # 转场时长（秒）
-    output_codec: str = "libx264"
-    output_preset: str = "medium"
-    output_crf: int = 23
-
-
-class EditingModeProcessor:
-    """剪辑模式处理器"""
-
-    def __init__(self, config: EditingModeConfig, work_dir: Optional[str] = None):
-        """
-        初始化剪辑模式处理器
-
-        Args:
-            config: 剪辑模式配置
-            work_dir: 工作目录，默认使用系统临时目录
-        """
-        self.config = config
-        self.work_dir = work_dir or tempfile.gettempdir()
-
-    def process(
-        self,
-        video_paths: list[str],
-        audio_path: Optional[str] = None,
-        output_path: Optional[str] = None,
-    ) -> str:
-        """
-        根据模式处理视频，返回输出文件路径
-
-        Args:
-            video_paths: 视频素材路径列表
-            audio_path: 音频路径（用于口播模式）
-            output_path: 输出文件路径，默认自动生成
-
-        Returns:
-            输出文件路径
-        """
-        if not video_paths:
-            raise ValueError("video_paths cannot be empty")
-
-        self._validate_inputs(video_paths, audio_path)
-
-        if output_path is None:
-            output_path = self._generate_output_path()
-
-        logger.info(f"Processing videos with mode: {self.config.mode}, count: {len(video_paths)}")
-
-        try:
-            if self.config.mode == EditingMode.ONE_TAKE:
-                return self._one_take(video_paths, output_path)
-            elif self.config.mode == EditingMode.PIP:
-                return self._pip(video_paths, output_path)
-            elif self.config.mode == EditingMode.VOICE_OVER:
-                return self._voice_over(video_paths, audio_path, output_path)
-            elif self.config.mode == EditingMode.VOICE_PIP:
-                return self._voice_pip(video_paths, audio_path, output_path)
-            else:
-                raise ValueError(f"Unsupported editing mode: {self.config.mode}")
-        except Exception as e:
-            logger.error(f"Error processing videos: {e}")
-            raise
-
-    def _validate_inputs(self, video_paths: list[str], audio_path: Optional[str]) -> None:
-        """验证输入文件"""
-        for path in video_paths:
-            if not os.path.exists(path):
-                raise FileNotFoundError(f"Video file not found: {path}")
-            if not os.path.getsize(path) > 0:
-                raise ValueError(f"Video file is empty: {path}")
-
-        if audio_path and not os.path.exists(audio_path):
-            raise FileNotFoundError(f"Audio file not found: {audio_path}")
-
-    def _generate_output_path(self) -> str:
-        """生成输出文件路径"""
-        os.makedirs(self.work_dir, exist_ok=True)
-        return os.path.join(self.work_dir, f"output_{self.config.mode}_{os.getpid()}.mp4")
-
-    def _run_ffmpeg(self, command: list[str], capture_output: bool = True) -> tuple:
-        """执行 FFmpeg 命令 — 委托给共享 ffmpeg_utils.run_ffmpeg"""
-        try:
-            return run_ffmpeg(command, capture_output=capture_output)
-        except RuntimeError as e:
-            logger.error(f"FFmpeg error: {e}")
-            raise
-
-    def _get_video_info(self, video_path: str) -> dict:
-        """获取视频信息 — 委托给共享 ffmpeg_utils.probe_video_info，补充 codec/size 字段"""
-        try:
-            info = probe_video_info(video_path)
-            info["codec"] = "unknown"
-            info["size"] = os.path.getsize(video_path) if os.path.exists(video_path) else 0
-            return info
-        except Exception as e:
-            logger.warning(f"Failed to get video info for {video_path}: {e}")
-            return {"width": 0, "height": 0, "fps": 25, "duration": 0, "codec": "unknown", "size": 0}
-
-    def _get_pip_position_offset(
-        self, main_width: int, main_height: int, pip_width: int, pip_height: int
-    ) -> tuple[int, int]:
-        """获取画中画位置偏移量"""
-        margin = 10
-        position_offsets = {
-            PIPPosition.TOP_LEFT: (margin, margin),
-            PIPPosition.TOP_RIGHT: (main_width - pip_width - margin, margin),
-            PIPPosition.BOTTOM_LEFT: (margin, main_height - pip_height - margin),
-            PIPPosition.BOTTOM_RIGHT: (main_width - pip_width - margin, main_height - pip_height - margin),
-        }
-        return position_offsets.get(self.config.pip_position, position_offsets[PIPPosition.TOP_RIGHT])
-
-    def _normalize_video(self, input_path: str, output_path: str) -> dict:
-        """标准化视频格式：先统一帧率，再缩放/填充"""
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-i",
-            input_path,
-            "-r",
-            str(self.config.output_fps),  # 先统一帧率
-            "-vf",
-            f"scale={self.config.output_width}:{self.config.output_height}:force_original_aspect_ratio=decrease,pad={self.config.output_width}:{self.config.output_height}:(ow-iw)/2:(oh-ih)/2,setsar=1",
-            "-r",
-            str(self.config.output_fps),
-            "-c:v",
-            self.config.output_codec,
-            "-preset",
-            self.config.output_preset,
-            "-crf",
-            str(self.config.output_crf),
-            "-pix_fmt",
-            "yuv420p",
-            "-movflags",
-            "+faststart",
-            "-an",
-            output_path,
-        ]
-        run_ffmpeg(command)
-        return self._get_video_info(output_path)
-
-    def _one_take(self, video_paths: list[str], output_path: str) -> str:
-        """一镜到底模式：顺序拼接视频，添加淡入淡出转场"""
-        if len(video_paths) == 1:
-            return self._normalize_video(video_paths[0], output_path)
-
-        normalized_paths = []
-        for i, path in enumerate(video_paths):
-            normalized = os.path.join(self.work_dir, f"normalized_{i}_{os.getpid()}.mp4")
-            self._normalize_video(path, normalized)
-            normalized_paths.append(normalized)
-
-        durations = [self._get_video_info(p)["duration"] for p in normalized_paths]
-
-        if len(normalized_paths) <= 5:
-            output_path = self._one_take_with_xfade(normalized_paths, durations, output_path)
-        else:
-            output_path = self._one_take_simple_concat(normalized_paths, output_path)
-
-        for p in normalized_paths:
-            try:
-                if p != output_path:
-                    os.remove(p)
-            except Exception as e:
-                logger.warning(f"Operation failed in apps/worker/video_processing/editing_modes.py: {e}", exc_info=True)
-
-        return output_path
-
-    def _one_take_with_xfade(self, normalized_paths: list[str], durations: list[float], output_path: str) -> str:
-        """使用 xfade 滤镜实现转场"""
-        if len(normalized_paths) == 2:
-            transition = self.config.transition_duration
-            offset1 = durations[0] - transition / 2
-
-            command = [
-                FFMPEG_BIN,
-                "-y",
-                "-i",
-                normalized_paths[0],
-                "-i",
-                normalized_paths[1],
-                "-filter_complex",
-                f"[0:v][1:v]xfade=transition=fade:duration={transition}:offset={offset1}[v]",
-                "-map",
-                "[v]",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                output_path,
-            ]
-            run_ffmpeg(command)
-            return output_path
-        else:
-            return self._one_take_simple_concat(normalized_paths, output_path)
-
-    def _one_take_simple_concat(self, normalized_paths: list[str], output_path: str) -> str:
-        """使用 concat demuxer 简单拼接"""
-        concat_file = os.path.join(self.work_dir, f"concat_list_{os.getpid()}.txt")
-        with open(concat_file, "w") as f:
-            for path in normalized_paths:
-                f.write(f"file '{os.path.abspath(path)}'\n")
-
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-f",
-            "concat",
-            "-safe",
-            "0",
-            "-i",
-            concat_file,
-            "-c",
-            "copy",
-            output_path,
-        ]
-        run_ffmpeg(command)
-
-        try:
-            os.remove(concat_file)
-        except Exception as e:
-            logger.warning(f"Operation failed in apps/worker/video_processing/editing_modes.py: {e}", exc_info=True)
-
-        return output_path
-
-    def _pip(self, video_paths: list[str], output_path: str) -> str:
-        """画中画模式：主视频全屏，后续视频叠加在角落"""
-        if not video_paths:
-            raise ValueError("No video paths provided")
-
-        main_video = video_paths[0]
-        main_normalized = os.path.join(self.work_dir, f"main_{os.getpid()}.mp4")
-        main_info = self._normalize_video(main_video, main_normalized)
-
-        if len(video_paths) == 1:
-            os.rename(main_normalized, output_path)
-            return output_path
-
-        pip_width = int(self.config.output_width * self.config.pip_scale)
-        pip_height = int(self.config.output_height * self.config.pip_scale)
-        x_offset, y_offset = self._get_pip_position_offset(
-            self.config.output_width, self.config.output_height, pip_width, pip_height
-        )
-
-        pip_normalized = os.path.join(self.work_dir, f"pip_{os.getpid()}.mp4")
-        pip_info = self._get_video_info(video_paths[1])
-
-        if pip_info["duration"] > main_info["duration"]:
-            temp_pip = os.path.join(self.work_dir, f"pip_temp_{os.getpid()}.mp4")
-            command = [
-                FFMPEG_BIN,
-                "-y",
-                "-i",
-                video_paths[1],
-                "-t",
-                str(main_info["duration"]),
-                "-vf",
-                f"scale={pip_width}:{pip_height}",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                temp_pip,
-            ]
-            run_ffmpeg(command)
-            pip_normalized_input = temp_pip
-        else:
-            command = [
-                FFMPEG_BIN,
-                "-y",
-                "-i",
-                video_paths[1],
-                "-vf",
-                f"scale={pip_width}:{pip_height}",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                pip_normalized,
-            ]
-            run_ffmpeg(command)
-            pip_normalized_input = pip_normalized
-
-        if main_info["duration"] > pip_info["duration"]:
-            looped_pip = os.path.join(self.work_dir, f"pip_looped_{os.getpid()}.mp4")
-            command = [
-                FFMPEG_BIN,
-                "-y",
-                "-stream_loop",
-                "-1",
-                "-i",
-                pip_normalized_input,
-                "-t",
-                str(main_info["duration"]),
-                "-vf",
-                f"scale={pip_width}:{pip_height}",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                looped_pip,
-            ]
-            run_ffmpeg(command)
-            pip_normalized_input = looped_pip
-
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-i",
-            main_normalized,
-            "-i",
-            pip_normalized_input,
-            "-filter_complex",
-            f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]",
-            "-map",
-            "[v]",
-            "-c:v",
-            self.config.output_codec,
-            "-preset",
-            self.config.output_preset,
-            "-crf",
-            str(self.config.output_crf),
-            "-pix_fmt",
-            "yuv420p",
-            output_path,
-        ]
-        run_ffmpeg(command)
-
-        for temp_file in [main_normalized, pip_normalized]:
-            if temp_file and temp_file != output_path:
-                try:
-                    os.remove(temp_file)
-                except Exception as e:
-                    logger.warning(
-                        f"Operation failed in apps/worker/video_processing/editing_modes.py: {e}", exc_info=True
-                    )
-
-        return output_path
-
-    def _voice_over(self, video_paths: list[str], audio_path: Optional[str], output_path: str) -> str:
-        """口播模式：背景画面 + 配音"""
-        if not audio_path:
-            raise ValueError("audio_path is required for VOICE_OVER mode")
-
-        if not video_paths:
-            raise ValueError("No background video provided")
-
-        audio_info = self._get_video_info(audio_path)
-        audio_duration = audio_info["duration"]
-
-        bg_normalized = os.path.join(self.work_dir, f"bg_{os.getpid()}.mp4")
-        bg_info = self._normalize_video(video_paths[0], bg_normalized)
-
-        if bg_info["duration"] < audio_duration:
-            looped_bg = os.path.join(self.work_dir, f"bg_looped_{os.getpid()}.mp4")
-            command = [
-                FFMPEG_BIN,
-                "-y",
-                "-stream_loop",
-                "-1",
-                "-i",
-                bg_normalized,
-                "-t",
-                str(audio_duration),
-                "-vf",
-                f"scale={self.config.output_width}:{self.config.output_height}",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                looped_bg,
-            ]
-            run_ffmpeg(command)
-            bg_normalized = looped_bg
-        elif bg_info["duration"] > audio_duration:
-            temp_bg = os.path.join(self.work_dir, f"bg_trimmed_{os.getpid()}.mp4")
-            command = [
-                FFMPEG_BIN,
-                "-y",
-                "-i",
-                bg_normalized,
-                "-t",
-                str(audio_duration),
-                "-c:v",
-                "copy",
-                temp_bg,
-            ]
-            run_ffmpeg(command)
-            bg_normalized = temp_bg
-
-        blurred_bg = os.path.join(self.work_dir, f"bg_blurred_{os.getpid()}.mp4")
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-i",
-            bg_normalized,
-            "-vf",
-            f"boxblur=5:5,scale={self.config.output_width}:{self.config.output_height}",
-            "-c:v",
-            self.config.output_codec,
-            "-preset",
-            self.config.output_preset,
-            "-crf",
-            str(self.config.output_crf),
-            "-pix_fmt",
-            "yuv420p",
-            blurred_bg,
-        ]
-        run_ffmpeg(command)
-
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-i",
-            blurred_bg,
-            "-i",
-            audio_path,
-            "-filter_complex",
-            "[0:v]drawbox=x=0:y=0:w=iw:h=ih:color=black@0.3:t=fill[v]",
-            "-map",
-            "[v]",
-            "-map",
-            "1:a",
-            "-c:v",
-            self.config.output_codec,
-            "-preset",
-            self.config.output_preset,
-            "-crf",
-            str(self.config.output_crf),
-            "-pix_fmt",
-            "yuv420p",
-            "-shortest",
-            output_path,
-        ]
-        run_ffmpeg(command)
-
-        for temp_file in [bg_normalized, blurred_bg]:
-            try:
-                if temp_file != output_path:
-                    os.remove(temp_file)
-            except Exception as e:
-                logger.warning(f"Operation failed in apps/worker/video_processing/editing_modes.py: {e}", exc_info=True)
-
-        return output_path
-
-    def _voice_pip(self, video_paths: list[str], audio_path: Optional[str], output_path: str) -> str:
-        """口播+画中画模式：口播视频在角落，其他视频作为背景"""
-        if not video_paths:
-            raise ValueError("No video paths provided")
-
-        if len(video_paths) == 1:
-            return self._normalize_video(video_paths[0], output_path)
-
-        voice_video = video_paths[0]
-        bg_video = video_paths[1] if len(video_paths) > 1 else video_paths[0]
-
-        voice_normalized = os.path.join(self.work_dir, f"voice_{os.getpid()}.mp4")
-        voice_info = self._normalize_video(voice_video, voice_normalized)
-
-        bg_normalized = os.path.join(self.work_dir, f"bg_{os.getpid()}.mp4")
-        bg_info = self._normalize_video(bg_video, bg_normalized)
-
-        final_duration = min(voice_info["duration"], bg_info["duration"])
-
-        pip_width = int(self.config.output_width * self.config.pip_scale)
-        pip_height = int(self.config.output_height * self.config.pip_scale)
-        x_offset, y_offset = self._get_pip_position_offset(
-            self.config.output_width, self.config.output_height, pip_width, pip_height
-        )
-
-        voice_adjusted = os.path.join(self.work_dir, f"voice_adj_{os.getpid()}.mp4")
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-i",
-            voice_normalized,
-            "-t",
-            str(final_duration),
-            "-vf",
-            f"scale={pip_width}:{pip_height}",
-            "-c:v",
-            self.config.output_codec,
-            "-preset",
-            self.config.output_preset,
-            "-crf",
-            str(self.config.output_crf),
-            "-pix_fmt",
-            "yuv420p",
-            voice_adjusted,
-        ]
-        run_ffmpeg(command)
-
-        bg_adjusted = os.path.join(self.work_dir, f"bg_adj_{os.getpid()}.mp4")
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-i",
-            bg_normalized,
-            "-t",
-            str(final_duration),
-            "-c:v",
-            "copy",
-            bg_adjusted,
-        ]
-        run_ffmpeg(command)
-
-        if audio_path:
-            command = [
-                FFMPEG_BIN,
-                "-y",
-                "-i",
-                bg_adjusted,
-                "-i",
-                voice_adjusted,
-                "-i",
-                audio_path,
-                "-filter_complex",
-                f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]",
-                "-map",
-                "[v]",
-                "-map",
-                "2:a",
-                "-shortest",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                output_path,
-            ]
-        else:
-            command = [
-                FFMPEG_BIN,
-                "-y",
-                "-i",
-                bg_adjusted,
-                "-i",
-                voice_adjusted,
-                "-filter_complex",
-                f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]",
-                "-map",
-                "[v]",
-                "-map",
-                "1:a",
-                "-shortest",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                output_path,
-            ]
-        run_ffmpeg(command)
-
-        for temp_file in [voice_normalized, voice_adjusted, bg_normalized, bg_adjusted]:
-            try:
-                if temp_file != output_path:
-                    os.remove(temp_file)
-            except Exception as e:
-                logger.warning(f"Operation failed in apps/worker/video_processing/editing_modes.py: {e}", exc_info=True)
-
-        return output_path
-
-
-def create_processor(mode: str, work_dir: Optional[str] = None, **kwargs) -> EditingModeProcessor:
-    """便捷工厂函数：创建剪辑模式处理器"""
-    try:
-        editing_mode = EditingMode(mode)
-    except ValueError:
-        raise ValueError(f"Invalid editing mode: {mode}. Valid modes: {[m.value for m in EditingMode]}")
-
-    config = EditingModeConfig(
-        mode=editing_mode,
-        output_width=kwargs.get("output_width", 1280),
-        output_height=kwargs.get("output_height", 720),
-        output_fps=kwargs.get("output_fps", 25),
-        pip_position=PIPPosition(kwargs.get("pip_position", "top_right")),
-        pip_scale=kwargs.get("pip_scale", 0.25),
-        transition_duration=kwargs.get("transition_duration", 0.5),
-    )
-
-    return EditingModeProcessor(config=config, work_dir=work_dir)

apps/worker/video_processing/ffmpeg_utils.py

@@ -1,8 +1,7 @@
-"""FFmpeg 工具函数 — 从 editing_modes.py / video_compose_service.py 提取的共享原语.
+"""FFmpeg 工具函数 — 共享原语.
 
 提供 FFmpeg / FFprobe 调用、视频信息探测、视频标准化、xfade 转场滤镜构建
-等底层能力，供 EditingModeProcessor、VideoComposeService、UnifiedRenderService
-共同复用。
+等底层能力，供 UnifiedRenderService、VideoComposeService 等复用。
 """
 
 from __future__ import annotations
@@ -32,6 +31,10 @@ XFADE_TRANSITION_MAP: dict[str, str] = {
     "slide_left": "slideleft",
     "slideright": "slideright",
     "slide_right": "slideright",
+    "slideup": "slideup",
+    "slide_up": "slideup",
+    "slidedown": "slidedown",
+    "slide_down": "slidedown",
     "dissolve": "dissolve",
     "wipe": "wipeleft",
     "wipeleft": "wipeleft",
@@ -82,6 +85,41 @@ def run_ffmpeg(
         raise
 
 
+def probe_has_audio(local_path: str | Path) -> bool:
+    """探测文件是否包含音频流。
+
+    Args:
+        local_path: 本地文件路径
+
+    Returns:
+        True 表示有音频流（或探测失败保守返回），False 表示确认无音频流
+    """
+    try:
+        result = subprocess.run(  # nosec B603
+            [
+                FFPROBE_BIN,
+                "-v",
+                "error",
+                "-select_streams",
+                "a:0",
+                "-show_entries",
+                "stream=codec_type",
+                "-of",
+                "default=noprint_wrappers=1:nokey=1",
+                str(local_path),
+            ],
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            timeout=10,
+        )
+        return result.stdout.strip() == "audio"
+    except Exception:
+        # 探测失败保守返回 True，让 FFmpeg 自己处理（避免误删音频）
+        return True
+
+
 def probe_duration(local_path: str | Path) -> float:
     """用 ffprobe 获取视频时长（秒）。
 

apps/worker/video_processing/render_adapter.py

@@ -0,0 +1,299 @@
+"""统一渲染引擎适配层 — Phase 2.
+
+将 EditPlan + EditPlanClips（来自 DB）适配为 UnifiedRenderService 的输入格式，
+封装素材下载、渲染执行、结果上传的完整流程。
+
+职责：
+  1. 从 DB 读取 EditPlan + EditPlanClips
+  2. 下载素材到本地，构建 asset_path_map
+  3. 调用 UnifiedRenderService 执行渲染
+  4. 上传渲染结果到 OSS
+  5. 支持进度回调（对接 JobService）
+"""
+
+from __future__ import annotations
+
+import logging
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable
+
+from sqlalchemy.orm import Session
+from video_processing.oss_helpers import download_asset, upload_to_oss
+from video_processing.unified_render_service import RenderResult, UnifiedRenderService
+
+from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import SQLAlchemyEditPlanClipRepository
+from packages.adapters.sqlalchemy_impl.edit_plan_repository import SQLAlchemyEditPlanRepository
+from packages.domain.edit_plan import EditPlan, EditPlanStatus
+from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
+
+logger = logging.getLogger(__name__)
+
+
+# ── 数据结构 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class RenderAdapterResult:
+    """渲染适配结果。"""
+
+    success: bool
+    output_url: str = ""
+    output_path: Path | None = None
+    duration: float = 0.0
+    file_size: int = 0
+    width: int = 0
+    height: int = 0
+    clip_count: int = 0
+    error_message: str = ""
+
+
+ProgressCallback = Callable[[float, str], None]
+"""进度回调：(progress_0_100, stage_description) → None"""
+
+
+# ── 适配层主体 ────────────────────────────────────────────────────────────────
+
+
+class RenderAdapter:
+    """统一渲染引擎适配层。
+
+    桥接 EditPlan 领域模型与 UnifiedRenderService 图层模型。
+
+    用法::
+
+        adapter = RenderAdapter(db)
+        result = adapter.render_plan(
+            plan_id=plan_id,
+            job_id=job_id,
+            progress_cb=lambda p, s: job_service.update_progress(job_id, p, s),
+        )
+    """
+
+    def __init__(self, db: Session) -> None:
+        self._db = db
+        self._plan_repo = SQLAlchemyEditPlanRepository(db)
+        self._clip_repo = SQLAlchemyEditPlanClipRepository(db)
+
+    # ── 公开方法 ──────────────────────────────────────────────────────────
+
+    def render_plan(
+        self,
+        plan_id: str,
+        *,
+        job_id: str = "",
+        work_dir: Path | None = None,
+        progress_cb: ProgressCallback | None = None,
+    ) -> RenderAdapterResult:
+        """渲染一个 EditPlan。
+
+        完整流程：
+        1. 加载计划与片段
+        2. 下载素材
+        3. 执行统一渲染
+        4. 上传结果
+
+        Args:
+            plan_id: EditPlan ID
+            job_id: 关联的 Job ID（用于结果存储路径）
+            work_dir: 工作目录，不传则使用临时目录
+            progress_cb: 进度回调函数
+
+        Returns:
+            RenderAdapterResult
+        """
+        temp_dir = None
+        try:
+            # 0. 准备工作目录
+            if work_dir is None:
+                temp_dir = tempfile.mkdtemp(prefix="render_")
+                work_dir = Path(temp_dir)
+            work_dir.mkdir(parents=True, exist_ok=True)
+
+            self._report_progress(progress_cb, 5.0, "加载剪辑计划")
+
+            # 1. 加载计划与片段
+            plan = self._plan_repo.get(plan_id)
+            if plan is None:
+                return RenderAdapterResult(
+                    success=False,
+                    error_message=f"剪辑计划不存在: {plan_id}",
+                )
+
+            clips = self._clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
+            ready_clips = [c for c in clips if c.status == EditPlanClipStatus.READY and c.asset_id]
+            ready_clips.sort(key=lambda c: c.order)
+
+            if not ready_clips:
+                return RenderAdapterResult(
+                    success=False,
+                    error_message="没有可渲染的就绪片段",
+                    clip_count=0,
+                )
+
+            logger.info(
+                "开始渲染: plan_id=%s job_id=%s ready_clips=%d engine=unified",
+                plan_id,
+                job_id,
+                len(ready_clips),
+            )
+
+            self._report_progress(progress_cb, 15.0, f"下载素材（{len(ready_clips)} 个）")
+
+            # 2. 下载素材
+            asset_path_map = self._download_assets(ready_clips, work_dir)
+            if not asset_path_map:
+                return RenderAdapterResult(
+                    success=False,
+                    error_message="所有素材下载失败",
+                    clip_count=len(ready_clips),
+                )
+
+            self._report_progress(progress_cb, 40.0, "执行视频渲染")
+
+            # 3. 执行统一渲染
+            render_svc = UnifiedRenderService(
+                plan=plan,
+                clips=ready_clips,
+                asset_path_map=asset_path_map,
+                work_dir=work_dir,
+            )
+            result = render_svc.render()
+
+            self._report_progress(progress_cb, 80.0, "上传渲染结果")
+
+            # 4. 上传结果
+            storage_key = f"rendered/{plan_id}/{job_id or plan_id}.mp4"
+            output_url = upload_to_oss(result.output_path, storage_key)
+
+            self._report_progress(progress_cb, 100.0, "渲染完成")
+
+            logger.info(
+                "[render-adapter] render success: plan_id=%s job_id=%s engine=unified "
+                "duration=%.2fs file_size=%d resolution=%dx%d clip_count=%d",
+                plan_id,
+                job_id,
+                result.duration,
+                result.file_size,
+                result.width,
+                result.height,
+                len(ready_clips),
+            )
+
+            return RenderAdapterResult(
+                success=True,
+                output_url=output_url or "",
+                output_path=result.output_path,
+                duration=result.duration,
+                file_size=result.file_size,
+                width=result.width,
+                height=result.height,
+                clip_count=len(ready_clips),
+            )
+
+        except Exception as exc:
+            logger.exception(
+                "[render-adapter] render failed: plan_id=%s job_id=%s engine=unified error=%s",
+                plan_id,
+                job_id,
+                str(exc)[:200],
+            )
+            return RenderAdapterResult(
+                success=False,
+                error_message=str(exc)[:500],
+            )
+        finally:
+            # 清理临时目录
+            if temp_dir:
+                import shutil
+
+                try:
+                    shutil.rmtree(temp_dir, ignore_errors=True)
+                except Exception:
+                    pass
+
+    def validate_plan(self, plan_id: str) -> tuple[bool, list[str], list[str], int, int]:
+        """校验计划是否可渲染（兼容 VideoComposeService.validate_compose 接口）。
+
+        Returns:
+            (valid, errors, warnings, ready_clip_count, total_clip_count)
+        """
+        errors: list[str] = []
+        warnings: list[str] = []
+
+        plan = self._plan_repo.get(plan_id)
+        if plan is None:
+            return False, [f"剪辑计划不存在: {plan_id}"], [], 0, 0
+
+        if plan.status not in (EditPlanStatus.EDITING, EditPlanStatus.RENDERING):
+            errors.append(f"计划状态不正确，需要 editing 或 rendering，当前: {plan.status}")
+
+        clips = self._clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
+        if not clips:
+            errors.append("计划没有任何片段")
+            return False, errors, warnings, 0, 0
+
+        clips.sort(key=lambda c: c.order)
+
+        ready_count = 0
+        pending_count = 0
+        no_asset_count = 0
+
+        for clip in clips:
+            if clip.status == EditPlanClipStatus.READY:
+                ready_count += 1
+                if not clip.asset_id:
+                    errors.append(f"片段 {clip.id} (order={clip.order}) 没有分配素材")
+                    no_asset_count += 1
+            elif clip.status == EditPlanClipStatus.PENDING:
+                pending_count += 1
+            elif clip.status == EditPlanClipStatus.FAILED:
+                warnings.append(f"片段 {clip.id} (order={clip.order}) 状态为 failed，已跳过")
+
+        if ready_count == 0:
+            errors.append("没有就绪（ready）的片段可以合成")
+
+        if pending_count > 0:
+            warnings.append(f"有 {pending_count} 个片段仍处于 pending 状态")
+
+        return len(errors) == 0, errors, warnings, ready_count, len(clips)
+
+    # ── 内部方法 ──────────────────────────────────────────────────────────
+
+    @staticmethod
+    def _report_progress(progress_cb: ProgressCallback | None, progress: float, stage: str) -> None:
+        """上报进度。"""
+        if progress_cb is not None:
+            try:
+                progress_cb(progress, stage)
+            except Exception:
+                logger.exception("进度回调失败")
+
+    @staticmethod
+    def _download_assets(clips: list[EditPlanClip], work_dir: Path) -> dict[str, Path]:
+        """下载片段素材到本地，返回 asset_id → local_path 映射。
+
+        只保留下载成功的素材。
+        """
+        asset_dir = work_dir / "assets"
+        asset_dir.mkdir(exist_ok=True)
+
+        asset_path_map: dict[str, Path] = {}
+
+        for clip in clips:
+            asset_id = clip.asset_id
+            if not asset_id:
+                continue
+
+            # 生成安全的本地文件名
+            safe_name = f"clip_{clip.order:04d}_{abs(hash(asset_id)) % 100000:05d}.mp4"
+            local_path = asset_dir / safe_name
+
+            if download_asset(asset_id, local_path):
+                asset_path_map[asset_id] = local_path
+                logger.debug("素材下载成功: clip_id=%s asset_id=%s", clip.id, asset_id[:60])
+            else:
+                logger.warning("素材下载失败: clip_id=%s asset_id=%s", clip.id, asset_id[:60])
+
+        return asset_path_map

apps/worker/video_processing/render_engine_resolver.py

@@ -0,0 +1,204 @@
+"""渲染引擎 Feature Flag 解析器。
+
+封装渲染引擎选择逻辑，支持：
+- 环境变量作为默认值（RENDER_ENGINE=legacy/unified）
+- Redis Feature Flag 运行时覆盖（白名单 + 百分比 + 全局开关）
+- 定时刷新，支持热更新不重启 worker
+
+使用方式：
+    resolver = RenderEngineResolver(redis_url="redis://...", default_engine="legacy")
+    engine = resolver.get_engine(user_id="user123")
+    # engine: "legacy" 或 "unified"
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from typing import Optional
+
+from packages.adapters.redis.feature_flag_store import (
+    FeatureFlagConfig,
+    FeatureFlagStore,
+    InMemoryFeatureFlagStore,
+    RedisFeatureFlagStore,
+)
+
+logger = logging.getLogger(__name__)
+
+# Feature Flag 名称常量
+FLAG_RENDER_ENGINE = "render_engine"
+
+# 引擎常量
+ENGINE_LEGACY = "legacy"
+ENGINE_UNIFIED = "unified"
+VALID_ENGINES = {ENGINE_LEGACY, ENGINE_UNIFIED}
+
+
+class RenderEngineResolver:
+    """渲染引擎选择器。
+
+    判定逻辑（从高到低）：
+    1. Redis flag 白名单匹配 → unified
+    2. Redis flag 百分比命中 → unified
+    3. Redis flag 全局开启（100%）→ unified
+    4. 环境变量默认值 → legacy / unified
+
+    当 Redis 不可用时，自动降级到环境变量默认值，不影响业务。
+    """
+
+    def __init__(
+        self,
+        default_engine: str = ENGINE_LEGACY,
+        redis_url: Optional[str] = None,
+        refresh_interval: float = 30.0,
+        store: Optional[FeatureFlagStore] = None,
+    ) -> None:
+        """
+        Args:
+            default_engine: 环境变量默认的引擎名（legacy / unified）
+            redis_url: Redis 连接 URL，传 None 时使用内存实现（测试用）
+            refresh_interval: Redis flag 配置刷新间隔（秒）
+            store: 直接传入 store 实例（测试用，优先级高于 redis_url）
+        """
+        self._default_engine = default_engine.lower() if default_engine else ENGINE_LEGACY
+        if self._default_engine not in VALID_ENGINES:
+            logger.warning(
+                "Invalid default engine '%s', fallback to '%s'",
+                self._default_engine,
+                ENGINE_LEGACY,
+            )
+            self._default_engine = ENGINE_LEGACY
+
+        if store is not None:
+            self._store = store
+        elif redis_url:
+            self._store = RedisFeatureFlagStore(redis_url=redis_url)
+        else:
+            self._store = InMemoryFeatureFlagStore()
+            logger.info("No Redis configured, using in-memory feature flag store")
+
+        self._refresh_interval = refresh_interval
+        self._lock = threading.Lock()
+        self._cached_config: Optional[FeatureFlagConfig] = None
+        self._last_refresh: float = 0.0
+
+    def _maybe_refresh(self) -> None:
+        """惰性刷新配置，超过刷新间隔时从存储重新读取。"""
+        import time
+
+        now = time.time()
+        if now - self._last_refresh < self._refresh_interval:
+            return
+
+        try:
+            config = self._store.get(FLAG_RENDER_ENGINE)
+            with self._lock:
+                self._cached_config = config
+                self._last_refresh = now
+        except Exception as exc:
+            logger.warning("Failed to refresh render engine flag: %s", exc)
+            # 刷新失败时保留旧缓存，不中断业务
+            if self._cached_config is None:
+                # 首次就读失败，设一个默认值
+                with self._lock:
+                    self._cached_config = FeatureFlagConfig(name=FLAG_RENDER_ENGINE)
+                    self._last_refresh = now
+
+    def _get_config(self) -> FeatureFlagConfig:
+        """获取当前 flag 配置（带缓存）。"""
+        if self._cached_config is None:
+            self._maybe_refresh()
+        else:
+            self._maybe_refresh()
+        return self._cached_config or FeatureFlagConfig(name=FLAG_RENDER_ENGINE)
+
+    def get_engine(self, user_id: Optional[str] = None) -> str:
+        """获取当前应该使用的渲染引擎。
+
+        Args:
+            user_id: 用户ID，用于白名单匹配和百分比哈希。
+                    传 None 时只看全局开关。
+
+        Returns:
+            "legacy" 或 "unified"
+        """
+        config = self._get_config()
+
+        # 全局关闭 → 用默认值
+        if not config.enabled:
+            return self._default_engine
+
+        # 白名单匹配 / 百分比命中 → unified
+        if config.is_active(user_id):
+            return ENGINE_UNIFIED
+
+        # 未命中灰度 → 用默认值
+        return self._default_engine
+
+    def should_use_unified(self, user_id: Optional[str] = None) -> bool:
+        """便捷方法：是否应该使用统一渲染引擎。"""
+        return self.get_engine(user_id) == ENGINE_UNIFIED
+
+    def force_refresh(self) -> None:
+        """强制立即刷新配置（用于管理接口修改后立即生效）。"""
+        self._last_refresh = 0.0
+        if isinstance(self._store, RedisFeatureFlagStore):
+            self._store.invalidate_cache(FLAG_RENDER_ENGINE)
+        self._maybe_refresh()
+
+    def get_config_snapshot(self) -> dict:
+        """获取当前配置快照（用于管理接口展示）。"""
+        config = self._get_config()
+        return {
+            "flag_name": FLAG_RENDER_ENGINE,
+            "default_engine": self._default_engine,
+            "enabled": config.enabled,
+            "percentage": config.percentage,
+            "whitelist": sorted(config.whitelist),
+            "refresh_interval": self._refresh_interval,
+            "last_refresh": self._last_refresh,
+        }
+
+    def set_flag(self, config: FeatureFlagConfig) -> None:
+        """设置 flag 配置（管理接口用）。"""
+        config.name = FLAG_RENDER_ENGINE
+        self._store.set(config)
+        self.force_refresh()
+
+
+# 全局单例
+_resolver: Optional[RenderEngineResolver] = None
+_resolver_lock = threading.Lock()
+
+
+def get_render_engine_resolver() -> RenderEngineResolver:
+    """获取全局单例（基于 worker 配置）。"""
+    global _resolver
+    if _resolver is not None:
+        return _resolver
+
+    with _resolver_lock:
+        if _resolver is not None:
+            return _resolver
+
+        try:
+            from worker_app.core.config import get_settings
+
+            settings = get_settings()
+            redis_url = getattr(settings, "redis_url", None) or getattr(settings, "broker_url", None)
+            default = getattr(settings, "render_engine", ENGINE_LEGACY)
+            _resolver = RenderEngineResolver(
+                default_engine=default,
+                redis_url=redis_url,
+            )
+            logger.info(
+                "RenderEngineResolver initialized: default=%s, redis=%s",
+                default,
+                bool(redis_url),
+            )
+        except Exception as exc:
+            logger.warning("Failed to init RenderEngineResolver from settings: %s", exc)
+            _resolver = RenderEngineResolver(default_engine=ENGINE_LEGACY)
+
+    return _resolver

apps/worker/video_processing/unified_render_service.py

@@ -24,6 +24,7 @@ from __future__ import annotations
 import logging
 import os
 import subprocess
+import time
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -43,6 +44,14 @@ from video_processing.ffmpeg_utils import (
 logger = logging.getLogger(__name__)
 
 
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+# Title/Subtitle 默认边距（像素）
+TITLE_MARGIN_TOP = 60
+TITLE_MARGIN_BOTTOM = 60
+TITLE_MARGIN_SIDE = 40
+
+
 # ── 数据结构 ──────────────────────────────────────────────────────────────────
 
 
@@ -89,6 +98,238 @@ class RenderResult:
 # ── clip_type → layer role 映射 ──────────────────────────────────────────────
 
 
+# ── ASS 字幕工具 ─────────────────────────────────────────────────────────────
+
+
+def _hex_to_ass_color(hex_color: str) -> str:
+    """将 HEX 颜色（#RRGGBB）转换为 ASS &HBBGGRR 格式。"""
+    hex_color = hex_color.lstrip("#")
+    if len(hex_color) != 6:
+        return "&H000000"
+    r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6]
+    return f"&H{b.upper()}{g.upper()}{r.upper()}"
+
+
+def _position_to_ass_alignment(position: str) -> int:
+    """将文字位置映射为 ASS \an 对齐编号。
+
+    ASS 对齐编号（数字小键盘布局）：
+      7 8 9
+      4 5 6
+      1 2 3
+    """
+    mapping = {
+        "top": 8,  # 顶部居中
+        "center": 5,  # 居中
+        "bottom": 2,  # 底部居中
+    }
+    return mapping.get(position, 8)
+
+
+def _build_ass_style(
+    style_name: str,
+    *,
+    font_name: str = "思源黑体",
+    font_size: int = 48,
+    primary_color: str = "&H00FFFFFF",
+    outline_color: str = "&H00000000",
+    outline_width: float = 1.0,
+    shadow_blur: float = 0.0,
+    shadow_offset: tuple[int, int] = (0, 0),
+    bold: bool = False,
+    italic: bool = False,
+    alignment: int = 8,
+    margin_v: int = 60,
+    margin_l: int = 40,
+    margin_r: int = 40,
+) -> str:
+    """构建 ASS Style 行。
+
+    Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour,
+            Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle,
+            BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
+    """
+    bold_val = -1 if bold else 0
+    italic_val = -1 if italic else 0
+
+    # BackColour 用于阴影（BorderStyle=1 时 outline + shadow）
+    back_color = primary_color  # 阴影颜色默认同文字色（带透明度由阴影模糊控制）
+
+    # Shadow 值：ASS 中 Shadow 字段是阴影偏移距离（像素），
+    # 我们用 shadow_offset[1] 作为纵向偏移，模糊由 BorderStyle=3 实现
+    # 简化：BorderStyle=1（outline + drop shadow），Shadow 字段表示阴影深度
+    shadow_depth = shadow_offset[1] if shadow_blur > 0 else 0
+
+    return (
+        f"Style: {style_name},{font_name},{font_size},{primary_color},"
+        f"&H000000FF,{outline_color},{back_color},"
+        f"{bold_val},{italic_val},0,0,100,100,0,0,"
+        f"1,{outline_width},{shadow_depth},{alignment},"
+        f"{margin_l},{margin_r},{margin_v},1"
+    )
+
+
+def generate_ass_subtitles(
+    output_path: Path,
+    *,
+    video_width: int,
+    video_height: int,
+    video_duration: float,
+    title_text: str = "",
+    title_config: dict[str, Any] | None = None,
+    subtitle_text: str = "",
+    subtitle_config: dict[str, Any] | None = None,
+) -> Path:
+    """生成 ASS 字幕文件。
+
+    支持 Title（标题）和 Subtitle（字幕）两种字幕类型，
+    各自可独立配置样式、位置和内容。
+
+    Args:
+        output_path: 输出 ASS 文件路径
+        video_width: 视频宽度（用于 ASS PlayResX）
+        video_height: 视频高度（用于 ASS PlayResY）
+        video_duration: 视频总时长（秒），字幕显示整个时长
+        title_text: 标题文本
+        title_config: 标题样式配置（TitleConfig dict）
+        subtitle_text: 字幕文本
+        subtitle_config: 字幕样式配置（SubtitleConfig dict）
+
+    Returns:
+        生成的 ASS 文件路径
+    """
+    title_config = title_config or {}
+    subtitle_config = subtitle_config or {}
+
+    title_enabled = title_config.get("enabled", True) and bool(title_text.strip())
+    subtitle_enabled = subtitle_config.get("enabled", True) and bool(subtitle_text.strip())
+
+    if not title_enabled and not subtitle_enabled:
+        # 没有字幕，生成空文件（仍返回路径，调用方自行判断是否使用）
+        output_path.write_text("", encoding="utf-8")
+        return output_path
+
+    styles: list[str] = []
+    events: list[str] = []
+
+    # ── Title 样式与事件 ──────────────────────────────────────────────────
+    if title_enabled:
+        title_color = _hex_to_ass_color(title_config.get("color", "#ffffff"))
+        title_stroke = title_config.get("stroke", {}) or {}
+        title_shadow = title_config.get("shadow", {}) or {}
+        stroke_color = _hex_to_ass_color(title_stroke.get("color", "#000000"))
+        stroke_width = float(title_stroke.get("width", 1)) if title_stroke.get("enabled", False) else 0.0
+        shadow_blur = float(title_shadow.get("blur", 4)) if title_shadow.get("enabled", False) else 0.0
+        shadow_offset = (
+            title_shadow.get("offset_x", 2) if title_shadow.get("enabled", False) else 0,
+            title_shadow.get("offset_y", 2) if title_shadow.get("enabled", False) else 0,
+        )
+
+        title_alignment = _position_to_ass_alignment(title_config.get("position", "top"))
+
+        styles.append(
+            _build_ass_style(
+                "TitleStyle",
+                font_name=title_config.get("font", "思源黑体"),
+                font_size=int(title_config.get("size", 48)),
+                primary_color=title_color,
+                outline_color=stroke_color,
+                outline_width=stroke_width,
+                shadow_blur=shadow_blur,
+                shadow_offset=shadow_offset,
+                bold=bool(title_config.get("bold", True)),
+                italic=bool(title_config.get("italic", False)),
+                alignment=title_alignment,
+                margin_v=TITLE_MARGIN_TOP,
+                margin_l=TITLE_MARGIN_SIDE,
+                margin_r=TITLE_MARGIN_SIDE,
+            )
+        )
+
+        # 转义 ASS 特殊字符
+        safe_title_text = _escape_ass_text(title_text)
+
+        events.append(
+            "Dialogue: 0,0:00:00.00," f"{_format_ass_time(video_duration)}," "TitleStyle,,0,0,0,," f"{safe_title_text}"
+        )
+
+    # ── Subtitle 样式与事件 ───────────────────────────────────────────────
+    if subtitle_enabled:
+        sub_color = _hex_to_ass_color(subtitle_config.get("color", "#ffffff"))
+        sub_alignment = _position_to_ass_alignment(subtitle_config.get("position", "bottom"))
+
+        styles.append(
+            _build_ass_style(
+                "SubtitleStyle",
+                font_name=subtitle_config.get("font", "思源黑体"),
+                font_size=int(subtitle_config.get("size", 24)),
+                primary_color=sub_color,
+                outline_color="&H00000000",
+                outline_width=1.0,
+                shadow_blur=0.0,
+                shadow_offset=(0, 0),
+                bold=False,
+                italic=False,
+                alignment=sub_alignment,
+                margin_v=TITLE_MARGIN_BOTTOM,
+                margin_l=TITLE_MARGIN_SIDE,
+                margin_r=TITLE_MARGIN_SIDE,
+            )
+        )
+
+        safe_subtitle_text = _escape_ass_text(subtitle_text)
+
+        events.append(
+            "Dialogue: 0,0:00:00.00,"
+            f"{_format_ass_time(video_duration)},"
+            "SubtitleStyle,,0,0,0,,"
+            f"{safe_subtitle_text}"
+        )
+
+    # ── 组装 ASS 文件 ─────────────────────────────────────────────────────
+    ass_content = f"""[Script Info]
+ScriptType: v4.00+
+PlayResX: {video_width}
+PlayResY: {video_height}
+ScaledBorderAndShadow: yes
+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
+{chr(10).join(styles)}
+
+[Events]
+Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
+{chr(10).join(events)}
+"""
+
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(ass_content, encoding="utf-8")
+    return output_path
+
+
+def _escape_ass_text(text: str) -> str:
+    r"""转义 ASS 文本中的特殊字符。
+
+    ASS 中换行用 \N（硬换行）或 \n（软换行），
+    大括号 {} 用于覆盖样式，需要转义。
+    """
+    # 将实际换行转为 ASS 硬换行
+    text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N")
+    # 转义大括号（ASS 用它做样式覆盖标签）
+    text = text.replace("{", "(").replace("}", ")")
+    return text
+
+
+def _format_ass_time(seconds: float) -> str:
+    """将秒数格式化为 ASS 时间格式 H:MM:SS.cc。"""
+    hours = int(seconds // 3600)
+    minutes = int((seconds % 3600) // 60)
+    secs = seconds % 60
+    return f"{hours}:{minutes:02d}:{secs:05.2f}"
+
+
 def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> str:
     """根据 clip_type 和 config.role 确定图层角色。
 
@@ -116,6 +357,8 @@ def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> str:
     # main type
     if role == "b_roll":
         return "broll"
+    if role == "audio":
+        return "audio"
     return "main"
 
 
@@ -165,11 +408,26 @@ class UnifiedRenderService:
         self.transition_duration = transition_duration
 
     def render(self) -> RenderResult:
-        """执行渲染，返回 RenderResult。
+        """执行渲染，返回 RenderResult.
+
+        优化路径：
+        - 单图层单 clip → 直通模式（-vf），性能最优
+        - 其他情况 → 完整 filter_complex 渲染
+
+        字幕渲染流程：
+        1. 视频主渲染（直通或完整链路）
+        2. 如有 title/subtitle，叠加 ASS 字幕
+
+        音频后处理：
+        1. 主图层音频 concat 拼接
+        2. 独立音频轨 amix 混入
+        3. 合并到输出视频
 
         Raises:
             ValueError: 没有可渲染的片段时抛出
         """
+        t_start = time.time()
+
         # 1. 解析 clips → ResolvedClips（跳过无素材的 clip）
         resolved = self._resolve_clips()
         if not resolved:
@@ -178,16 +436,94 @@ class UnifiedRenderService:
         # 2. 分组为 RenderLayers
         layers = self._group_clips_into_layers(resolved)
 
-        # 3. 构建 filter_complex
+        # 3. 计算视频总时长（用于字幕显示时长）
+        video_duration = self._estimate_total_duration(layers)
+
+        # 4. 生成 ASS 字幕文件（如果有 title/subtitle 配置）
+        ass_path = self._maybe_generate_ass(video_duration)
+
+        # 灰度埋点：开始渲染
+        layer_roles = [layer.role for layer in layers]
+        clip_counts = {layer.role: len(layer.clips) for layer in layers}
+        logger.info(
+            "[unified-render] start render: plan_id=%s clip_count=%d layers=%s clip_counts=%s",
+            self.plan.id,
+            len(resolved),
+            layer_roles,
+            clip_counts,
+        )
+
+        # 5. 视频主渲染
+        t_video_start = time.time()
+        video_only_path = self.work_dir / f"rendered_{self.plan.id}_video.mp4"
         output_path = self.work_dir / f"rendered_{self.plan.id}.mp4"
-        filter_complex, input_args = self._build_filter_complex(layers)
 
-        # 4. 执行 FFmpeg
-        self._execute_ffmpeg(filter_complex, input_args, output_path)
+        is_pass_through = self._can_use_pass_through(layers)
+        pass_through_has_audio = False
 
-        # 5. 探测输出
+        if is_pass_through:
+            # 直通优化：单clip场景一次FFmpeg同时处理视频+音频，省去提取+合并两次调用
+            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)
+
+        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",
+            self.plan.id,
+            video_render_ms,
+            is_pass_through,
+        )
+
+        # 6. 音频后处理混音（直通场景已合并处理，跳过）
+        t_audio_start = time.time()
+        audio_mix_ms = 0
+        has_audio = False
+
+        if is_pass_through:
+            # 直通场景已在一次调用中完成视频+音频
+            has_audio = pass_through_has_audio
+        else:
+            audio_path = self._mix_audio(layers, video_duration)
+            t_audio_end = time.time()
+            audio_mix_ms = int((t_audio_end - t_audio_start) * 1000)
+            has_audio = audio_path is not None
+            if has_audio:
+                logger.info(
+                    "[unified-render] audio mix done: plan_id=%s duration_ms=%d",
+                    self.plan.id,
+                    audio_mix_ms,
+                )
+                # 7. 合并音视频
+                self._merge_audio_video(video_only_path, audio_path, output_path)
+            else:
+                # 无音频，直接用无声视频
+                import shutil
+
+                shutil.copy2(video_only_path, output_path)
+
+        # 8. 探测输出
         duration, file_size, width, height = self._probe_output(output_path)
 
+        t_total = int((time.time() - t_start) * 1000)
+        logger.info(
+            "[unified-render] render done: plan_id=%s total_ms=%d video_ms=%d audio_ms=%d "
+            "output_duration=%.2fs output_size=%d resolution=%dx%d has_audio=%s",
+            self.plan.id,
+            t_total,
+            video_render_ms,
+            audio_mix_ms if has_audio else 0,
+            duration,
+            file_size,
+            width,
+            height,
+            has_audio,
+        )
+
         return RenderResult(
             output_path=output_path,
             duration=duration,
@@ -196,6 +532,208 @@ class UnifiedRenderService:
             height=height,
         )
 
+    def _estimate_total_duration(self, layers: list[RenderLayer]) -> float:
+        """估算视频总时长（用于字幕等需要）。
+
+        取主图层（main/broll/background）的总时长，转场重叠按 transition_duration 估算。
+        """
+        # 找主图层（第一个有视频内容的图层）
+        main_layer = None
+        for role in ("main", "broll", "background"):
+            for layer in layers:
+                if layer.role == role:
+                    main_layer = layer
+                    break
+            if main_layer:
+                break
+
+        if not main_layer or not main_layer.clips:
+            return 0.0
+
+        total = sum(UnifiedRenderService._clip_effective_duration(c) for c in main_layer.clips)
+
+        # 减去转场重叠时间（粗略估算）
+        n_clips = len(main_layer.clips)
+        if n_clips > 1:
+            total -= (n_clips - 1) * self.transition_duration
+
+        return max(0.1, total)
+
+    def _maybe_generate_ass(self, video_duration: float) -> Path | None:
+        """根据 plan.config 生成 ASS 字幕文件。
+
+        Returns:
+            ASS 文件路径，没有字幕时返回 None
+        """
+        config = self.plan.config or {}
+        title_cfg = config.get("title", {}) or {}
+        subtitle_cfg = config.get("subtitle", {}) or {}
+
+        title_enabled = title_cfg.get("enabled", True)
+        subtitle_enabled = subtitle_cfg.get("enabled", True)
+        title_text = title_cfg.get("text", "") or ""
+        subtitle_text = subtitle_cfg.get("text", "") or ""
+
+        has_title = title_enabled and bool(title_text.strip())
+        has_subtitle = subtitle_enabled and bool(subtitle_text.strip())
+
+        if not has_title and not has_subtitle:
+            return None
+
+        ass_path = self.work_dir / f"subtitles_{self.plan.id}.ass"
+
+        generate_ass_subtitles(
+            ass_path,
+            video_width=self.output_width,
+            video_height=self.output_height,
+            video_duration=video_duration,
+            title_text=title_text,
+            title_config=title_cfg,
+            subtitle_text=subtitle_text,
+            subtitle_config=subtitle_cfg,
+        )
+
+        logger.info(
+            "生成字幕: plan_id=%s title=%s subtitle=%s ass=%s",
+            self.plan.id,
+            has_title,
+            has_subtitle,
+            ass_path,
+        )
+        return ass_path
+
+    def _can_use_pass_through(self, layers: list[RenderLayer]) -> bool:
+        """判断是否可以走直通优化路径。
+
+        条件：
+        1. 只有 1 个图层
+        2. 该图层是视频图层（main/broll/background），不是 overlay/corner_voice/audio
+        3. 该图层只有 1 个 clip（无转场需求）
+        """
+        if len(layers) != 1:
+            return False
+        layer = layers[0]
+        if layer.role not in ("main", "broll", "background"):
+            return False
+        if len(layer.clips) != 1:
+            return False
+        return True
+
+    def _render_pass_through(
+        self,
+        layers: list[RenderLayer],
+        output_path: Path,
+        *,
+        ass_path: Path | None = None,
+        video_duration: float = 0.0,
+    ) -> bool:
+        """单图层单 clip 直通渲染（使用 -vf 而非 -filter_complex），一次性输出带音频的最终视频。
+
+        性能优化：
+        - 避免 filter_complex 的解析和调度开销，单clip场景性能提升 ~30%
+        - 视频+音频一次FFmpeg调用完成，省去后续音频提取+音视频合并两次调用
+
+        Args:
+            layers: 图层列表（只有1个图层1个clip）
+            output_path: 输出文件路径
+            ass_path: ASS 字幕文件路径，有则叠加字幕
+            video_duration: 视频总时长（用于截断音频，0表示不额外截断）
+
+        Returns:
+            True 表示输出包含音频（近似判断，实际以输出文件为准）
+        """
+        clip = layers[0].clips[0]
+        role = layers[0].role
+
+        # 构建视频滤镜链（与 _build_filter_complex 中预处理逻辑一致）
+        filters: list[str] = []
+
+        # trim
+        effective_duration = UnifiedRenderService._clip_effective_duration(clip)
+
+        if effective_duration > 0:
+            filters.append(f"trim=duration={effective_duration}")
+            filters.append("setpts=PTS-STARTPTS")
+
+        # scale + crop（铺满裁剪）
+        if role in ("overlay", "corner_voice"):
+            pip_w = int(self.output_width * _PIP_SCALE)
+            pip_h = int(self.output_height * _PIP_SCALE)
+            filters.append(f"scale={pip_w}:{pip_h}")
+        else:
+            # main / broll / background: 铺满裁剪
+            filters.append(f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase")
+            filters.append(f"crop={self.output_width}:{self.output_height}")
+
+        filters.append("setpts=PTS-STARTPTS")
+        filters.append(f"fps={self.output_fps}")
+        filters.append("format=yuv420p")
+
+        # 字幕叠加
+        if ass_path is not None:
+            ass_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:")
+            filters.append(f"subtitles='{ass_filter_path}'")
+
+        vf_str = ",".join(filters)
+
+        # 最终输出时长：取 clip 有效时长和 video_duration 的较小值
+        final_duration = effective_duration
+        if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
+            final_duration = video_duration
+
+        command = [
+            FFMPEG_BIN,
+            "-y",
+            "-i",
+            str(clip.local_path),
+            "-vf",
+            vf_str,
+            "-c:v",
+            "libx264",
+            "-crf",
+            "23",
+            "-preset",
+            "medium",
+            "-pix_fmt",
+            "yuv420p",
+            "-movflags",
+            "+faststart",
+        ]
+
+        # 音频处理：background 通常是图片无音频，跳过；其他编码为 aac
+        # background 以外的视频素材，默认带音频
+        has_audio = role != "background"
+        if has_audio:
+            command.extend(["-c:a", "aac", "-b:a", "128k"])
+
+        # 统一截断时长（同时作用于视频和音频）
+        if final_duration > 0:
+            command.extend(["-t", f"{final_duration:.3f}"])
+
+        command.append(str(output_path))
+
+        logger.info(
+            "直通渲染: plan_id=%s clip=%s role=%s duration=%.2fs has_audio=%s",
+            self.plan.id,
+            clip.clip_id,
+            role,
+            effective_duration,
+            has_audio,
+        )
+        try:
+            run_ffmpeg(command)
+        except subprocess.CalledProcessError as e:
+            logger.error(
+                "直通渲染失败: plan_id=%s clip=%s exit_code=%d\nvf=%s",
+                self.plan.id,
+                clip.clip_id,
+                e.returncode,
+                vf_str[:2000],
+            )
+            raise
+
+        return has_audio
+
     # ── 内部方法 ──────────────────────────────────────────────────────────────
 
     def _resolve_clips(self) -> list[ResolvedClip]:
@@ -277,9 +815,15 @@ class UnifiedRenderService:
         layers = sorted(layer_map.values(), key=lambda lyr: lyr.z_index)
         return layers
 
-    def _build_filter_complex(self, layers: list[RenderLayer]) -> tuple[str, list[str]]:
+    def _build_filter_complex(
+        self, layers: list[RenderLayer], *, ass_path: Path | None = None
+    ) -> tuple[str, list[str]]:
         """构建 FFmpeg filter_complex 字符串和输入参数列表。
 
+        Args:
+            layers: 图层列表
+            ass_path: ASS 字幕文件路径，有则在最后叠加字幕
+
         Returns:
             (filter_complex_str, input_args_list)
             input_args_list 是 ["-i", path1, "-i", path2, ...] 格式
@@ -311,14 +855,7 @@ class UnifiedRenderService:
             filters: list[str] = []
 
             # trim — 始终将输出截断到有效时长，防止 xfade offset 与实际时长不匹配
-            # 有效时长 = min(指定时长, 实际时长)；若均未设置则跳过
-            effective_duration = 0.0
-            if clip.duration > 0:
-                effective_duration = (
-                    min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
-                )
-            elif clip.actual_duration > 0:
-                effective_duration = clip.actual_duration
+            effective_duration = UnifiedRenderService._clip_effective_duration(clip)
 
             if effective_duration > 0:
                 filters.append(f"trim=duration={effective_duration}")
@@ -335,11 +872,12 @@ class UnifiedRenderService:
                 )
                 filters.append(f"crop={self.output_width}:{self.output_height}")
             else:
-                # main / broll: scale + pad 保持宽高比
+                # main / broll: 铺满裁剪（scale to cover + center crop）
+                # 对齐链路A编辑器合成行为，与主流短视频平台一致
                 filters.append(
-                    f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=decrease"
+                    f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase"
                 )
-                filters.append(f"pad={self.output_width}:{self.output_height}" ":(ow-iw)/2:(oh-ih)/2:black")
+                filters.append(f"crop={self.output_width}:{self.output_height}")
 
             filters.append("setpts=PTS-STARTPTS")
             filters.append(f"fps={self.output_fps}")
@@ -354,14 +892,7 @@ class UnifiedRenderService:
             layer_clip_indices = [all_clips.index(c) for c in layer.clips]
             layer_labels = [preprocessed_labels[i] for i in layer_clip_indices]
             # 使用 trim 后的有效时长，与 Step 1 的 trim=duration 保持一致
-            layer_durations = []
-            for i in layer_clip_indices:
-                c = all_clips[i]
-                if c.duration > 0:
-                    eff = min(c.duration, c.actual_duration) if c.actual_duration > 0 else c.duration
-                else:
-                    eff = c.actual_duration if c.actual_duration > 0 else 0.0
-                layer_durations.append(eff)
+            layer_durations = [UnifiedRenderService._clip_effective_duration(all_clips[i]) for i in layer_clip_indices]
             layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices]
 
             if len(layer_labels) == 1:
@@ -421,7 +952,12 @@ class UnifiedRenderService:
                 filter_parts.append(f"[{final_video_label}][{overlay_label}]" f"overlay={x}:{y}[{combined_label}]")
                 final_video_label = combined_label
 
-        filter_parts.append(f"[{final_video_label}]format=yuv420p[final_video]")
+        # 叠加字幕（如有）+ 最终像素格式
+        if ass_path is not None:
+            ass_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:")
+            filter_parts.append(f"[{final_video_label}]subtitles='{ass_filter_path}',format=yuv420p[final_video]")
+        else:
+            filter_parts.append(f"[{final_video_label}]format=yuv420p[final_video]")
 
         filter_complex = ";".join(filter_parts)
         return filter_complex, input_args
@@ -476,7 +1012,7 @@ class UnifiedRenderService:
             raise
 
     def _probe_output(self, output_path: Path) -> tuple[float, int, int, int]:
-        """探测输出文件的时长、大小、宽高。
+        """探测输出文件的时长、大小、宽高.
 
         Returns:
             (duration, file_size, width, height)
@@ -489,3 +1025,304 @@ class UnifiedRenderService:
             info["width"],
             info["height"],
         )
+
+    # ── 音频后处理 ────────────────────────────────────────────────────────
+
+    def _mix_audio(self, layers: list[RenderLayer], video_duration: float) -> Path | None:
+        """音频后处理混音.
+
+        处理逻辑：
+        1. 主音频源按优先级查找：main > broll（background 不参与主音频，通常是图片无音轨）
+        2. 主图层音频按顺序 concat 拼接
+        3. 独立音频轨（audio role）用 amix 混入
+        4. 输出时长截断到 video_duration
+        5. 无音频流的 clip 会被自动跳过，避免 FFmpeg 引用 [i:a] 失败
+
+        Args:
+            layers: 图层列表
+            video_duration: 视频总时长（用于截断音频）
+
+        Returns:
+            混音后的音频文件路径，无音频时返回 None
+        """
+        # 按优先级精确查找主音频图层：main > broll
+        # background 不参与主音频（通常是静态图片，无音轨）
+        layer_map = {layer.role: layer for layer in layers}
+        main_layer = None
+        for role in ("main", "broll"):
+            if role in layer_map and layer_map[role].clips:
+                main_layer = layer_map[role]
+                break
+
+        main_clips: list[ResolvedClip] = main_layer.clips if main_layer else []
+
+        # 没有主视频图层时兜底：检查 overlay/corner_voice 层是否有带音频的素材
+        if not main_clips:
+            for role in ("overlay", "corner_voice"):
+                if role in layer_map and layer_map[role].clips:
+                    main_clips = layer_map[role].clips
+                    break
+
+        # 收集独立音频轨
+        audio_clips: list[ResolvedClip] = []
+        if "audio" in layer_map:
+            audio_clips = layer_map["audio"].clips
+
+        # ── 防御：过滤掉无音频流的 clip ──
+        # 源视频可能没有音频流（如静音视频、纯图片转的视频），直接引用 [i:a] 会导致 FFmpeg 失败
+        main_clips = [c for c in main_clips if self._clip_has_audio(c)]
+        audio_clips = [c for c in audio_clips if self._clip_has_audio(c)]
+
+        if not main_clips and not audio_clips:
+            return None
+
+        # 构建音频处理命令
+        output_path = self.work_dir / f"audio_{self.plan.id}.aac"
+
+        # 简单场景：只有主图层 + 无独立音频 → 直接从视频提取音频并拼接
+        if main_clips and not audio_clips:
+            self._concat_main_audio(main_clips, output_path, video_duration)
+            return output_path
+
+        # 有独立音频轨 → amix 混音
+        self._mix_with_independent_audio(main_clips, audio_clips, output_path, video_duration)
+        return output_path
+
+    def _concat_main_audio(self, clips: list[ResolvedClip], output_path: Path, video_duration: float) -> None:
+        """主图层音频 concat 拼接（对齐链路A行为）.
+
+        每个 clip 提取音频 → trim → 按顺序 concat。
+        """
+        if len(clips) == 1:
+            # 单 clip，直接提取音频，截断到 min(clip有效时长, 视频总时长)
+            clip = clips[0]
+            effective_duration = self._clip_effective_duration(clip)
+            # 最终时长：取 clip 有效时长和视频总时长的较小值
+            # （视频总时长由主图层决定，但单 clip 场景下两者应该一致，仍做保护）
+            final_duration = effective_duration
+            if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
+                final_duration = video_duration
+
+            command = [
+                FFMPEG_BIN,
+                "-y",
+                "-i",
+                str(clip.local_path),
+                "-vn",
+                "-acodec",
+                "aac",
+                "-b:a",
+                "128k",
+            ]
+            if final_duration > 0:
+                command.extend(["-t", f"{final_duration:.3f}"])
+            command.append(str(output_path))
+            run_ffmpeg(command)
+            return
+
+        # 多 clip，用 filter_complex concat
+        input_args: list[str] = []
+        filter_parts: list[str] = []
+
+        for i, clip in enumerate(clips):
+            input_args.extend(["-i", str(clip.local_path)])
+            effective_duration = self._clip_effective_duration(clip)
+            if effective_duration > 0:
+                filter_parts.append(f"[{i}:a]atrim=0:{effective_duration:.3f},asetpts=PTS-STARTPTS[a{i}]")
+            else:
+                filter_parts.append(f"[{i}:a]asetpts=PTS-STARTPTS[a{i}]")
+
+        audio_labels = "".join(f"[a{i}]" for i in range(len(clips)))
+        filter_parts.append(f"{audio_labels}concat=n={len(clips)}:v=0:a=1[outa]")
+
+        # 截断到视频总时长
+        if video_duration > 0:
+            filter_parts.append(f"[outa]atrim=0:{video_duration:.3f}[final_audio]")
+            final_label = "final_audio"
+        else:
+            final_label = "outa"
+
+        filter_complex = ";".join(filter_parts)
+
+        command = [
+            FFMPEG_BIN,
+            "-y",
+            *input_args,
+            "-filter_complex",
+            filter_complex,
+            "-map",
+            f"[{final_label}]",
+            "-acodec",
+            "aac",
+            "-b:a",
+            "128k",
+            str(output_path),
+        ]
+        run_ffmpeg(command)
+
+    def _mix_with_independent_audio(
+        self,
+        main_clips: list[ResolvedClip],
+        audio_clips: list[ResolvedClip],
+        output_path: Path,
+        video_duration: float,
+    ) -> None:
+        """主音频 + 独立音频轨 amix 混音.
+
+        Args:
+            main_clips: 主视频 clips（提取音频后 concat）
+            audio_clips: 独立音频轨 clips
+            output_path: 输出路径
+            video_duration: 视频总时长
+        """
+        input_args: list[str] = []
+        filter_parts: list[str] = []
+        mix_labels: list[str] = []
+
+        input_idx = 0
+
+        # 1. 主图层音频 concat
+        if main_clips:
+            for clip in main_clips:
+                input_args.extend(["-i", str(clip.local_path)])
+                effective_duration = self._clip_effective_duration(clip)
+                if effective_duration > 0:
+                    filter_parts.append(
+                        f"[{input_idx}:a]atrim=0:{effective_duration:.3f},asetpts=PTS-STARTPTS[ma{input_idx}]"
+                    )
+                else:
+                    filter_parts.append(f"[{input_idx}:a]asetpts=PTS-STARTPTS[ma{input_idx}]")
+                input_idx += 1
+
+            if len(main_clips) == 1:
+                mix_labels.append("ma0")
+            else:
+                main_labels = "".join(f"[ma{i}]" for i in range(len(main_clips)))
+                filter_parts.append(f"{main_labels}concat=n={len(main_clips)}:v=0:a=1[main_audio]")
+                mix_labels.append("main_audio")
+
+        # 2. 独立音频轨
+        for j, clip in enumerate(audio_clips):
+            input_args.extend(["-i", str(clip.local_path)])
+            effective_duration = self._clip_effective_duration(clip)
+            volume = clip.config.get("volume", 1.0) if clip.config else 1.0
+            label = f"ia{j}"
+            filters = []
+            if effective_duration > 0:
+                filters.append(f"atrim=0:{effective_duration:.3f}")
+            filters.append("asetpts=PTS-STARTPTS")
+            if volume != 1.0:
+                filters.append(f"volume={volume}")
+            filter_parts.append(f"[{input_idx}:a]{','.join(filters)}[{label}]")
+            mix_labels.append(label)
+            input_idx += 1
+
+        # 3. amix 混音
+        mix_inputs = "".join(f"[{label}]" for label in mix_labels)
+        n_inputs = len(mix_labels)
+        # normalized=0 保持音量，duration=shortest 取最短
+        filter_parts.append(f"{mix_inputs}amix=inputs={n_inputs}:duration=longest:normalize=0[mixed_audio]")
+
+        # 4. 截断到视频时长
+        if video_duration > 0:
+            filter_parts.append(f"[mixed_audio]atrim=0:{video_duration:.3f}[final_audio]")
+            final_label = "final_audio"
+        else:
+            final_label = "mixed_audio"
+
+        filter_complex = ";".join(filter_parts)
+
+        command = [
+            FFMPEG_BIN,
+            "-y",
+            *input_args,
+            "-filter_complex",
+            filter_complex,
+            "-map",
+            f"[{final_label}]",
+            "-acodec",
+            "aac",
+            "-b:a",
+            "128k",
+            str(output_path),
+        ]
+
+        logger.info(
+            "音频混音: plan_id=%s main_clips=%d audio_clips=%d",
+            self.plan.id,
+            len(main_clips),
+            len(audio_clips),
+        )
+        try:
+            run_ffmpeg(command)
+        except subprocess.CalledProcessError as e:
+            logger.error(
+                "音频混音失败: plan_id=%s exit_code=%d\nfilter_complex:\n%s",
+                self.plan.id,
+                e.returncode,
+                filter_complex[:3000],
+            )
+            raise
+
+    def _merge_audio_video(self, video_path: Path, audio_path: Path, output_path: Path) -> None:
+        """将音频合并到视频中（视频流拷贝，音频直接复用）.
+
+        Args:
+            video_path: 无声视频路径
+            audio_path: 音频文件路径
+            output_path: 输出文件路径
+        """
+        command = [
+            FFMPEG_BIN,
+            "-y",
+            "-i",
+            str(video_path),
+            "-i",
+            str(audio_path),
+            "-c:v",
+            "copy",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-map",
+            "0:v:0",
+            "-map",
+            "1:a:0",
+            "-shortest",
+            "-movflags",
+            "+faststart",
+            str(output_path),
+        ]
+
+        logger.info("合并音视频: plan_id=%s", self.plan.id)
+        try:
+            run_ffmpeg(command)
+        except subprocess.CalledProcessError as e:
+            logger.error(
+                "合并音视频失败: plan_id=%s exit_code=%d",
+                self.plan.id,
+                e.returncode,
+            )
+            raise
+
+    @staticmethod
+    def _clip_effective_duration(clip: ResolvedClip) -> float:
+        """计算 clip 的有效时长."""
+        if clip.duration > 0:
+            return min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
+        return clip.actual_duration if clip.actual_duration > 0 else 0.0
+
+    def _clip_has_audio(self, clip: ResolvedClip) -> bool:
+        """探测 clip 是否有音频流（带缓存）.
+
+        避免同一个 clip 被多次 ffprobe 探测。
+        """
+        if not hasattr(self, "_audio_cache"):
+            self._audio_cache: dict[str, bool] = {}
+        key = str(clip.local_path)
+        if key not in self._audio_cache:
+            from .ffmpeg_utils import probe_has_audio
+
+            self._audio_cache[key] = probe_has_audio(clip.local_path)
+        return self._audio_cache[key]

apps/worker/video_processing/video_compose_service.py

@@ -1,821 +0,0 @@
-"""
-视频合成服务
-支持多种剪辑模式和转场效果，包含完整的安全校验
-"""
-
-import logging
-import os
-import subprocess
-import tempfile
-from dataclasses import dataclass
-from enum import Enum
-
-try:
-    from enum import StrEnum
-except ImportError:
-
-    class StrEnum(str, Enum):  # type: ignore[no-redef]
-        """Python 3.10 兼容的 StrEnum 回退实现。"""
-
-        pass
-
-
-from pathlib import Path
-from typing import Optional
-
-from packages.domain.editing_mode import EditingMode
-
-logger = logging.getLogger(__name__)
-
-# ========== 安全常量 ==========
-# 允许的输出目录白名单（使用环境变量或系统临时目录，避免硬编码 /tmp）
-_VIDEO_OUTPUT_DIR = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output"))
-ALLOWED_OUTPUT_DIRS = [_VIDEO_OUTPUT_DIR, "/var/app/rendered"]
-
-# 允许的输入路径前缀白名单
-ALLOWED_INPUT_PREFIXES = ("s3://", "oss://", "local://", "/var/storage/")
-
-# 允许的转场效果白名单
-ALLOWED_TRANSITIONS = {
-    "fade",
-    "slideleft",
-    "slideright",
-    "dissolve",
-    "wipeleft",
-    "wiperight",
-    "cut",
-    "slideup",
-    "slidedown",
-}
-
-# 转场效果映射
-_XFADE_TRANSITION_MAP = {
-    "fade": "fade",
-    "slideleft": "slideleft",
-    "slideright": "slideright",
-    "dissolve": "dissolve",
-    "wipeleft": "wipeleft",
-    "wiperight": "wiperight",
-    "cut": "cut",
-    "slideup": "slideup",
-    "slidedown": "slidedown",
-}
-
-
-class VideoComposeError(Exception):
-    """视频合成服务异常"""
-
-    pass
-
-
-class PIPPosition(StrEnum):
-    """画中画位置枚举"""
-
-    TOP_LEFT = "top_left"
-    TOP_RIGHT = "top_right"
-    BOTTOM_LEFT = "bottom_left"
-    BOTTOM_RIGHT = "bottom_right"
-
-
-@dataclass
-class Clip:
-    """视频片段"""
-
-    asset_id: str  # 资源ID，对应输入路径
-    start_time: float = 0.0
-    duration: float = 0.0
-    transition: str = "fade"  # 转场效果
-
-
-@dataclass
-class EditingModeConfig:
-    """剪辑模式配置"""
-
-    mode: EditingMode
-    output_width: int = 1280
-    output_height: int = 720
-    output_fps: int = 25
-    pip_position: PIPPosition = PIPPosition.TOP_RIGHT
-    pip_scale: float = 0.25  # 画中画占主画面的比例
-    transition_duration: float = 0.5  # 转场时长（秒）
-    output_codec: str = "libx264"
-    output_preset: str = "medium"
-    output_crf: int = 23
-
-
-class VideoComposeService:
-    """视频合成服务"""
-
-    def __init__(self, config: EditingModeConfig, work_dir: Optional[str] = None):
-        """
-        初始化视频合成服务
-
-        Args:
-            config: 剪辑模式配置
-            work_dir: 工作目录，默认使用系统临时目录
-        """
-        self.config = config
-        self.work_dir = work_dir or tempfile.gettempdir()
-        self._ffmpeg_bin = "ffmpeg"
-        self._ffprobe_bin = "ffprobe"
-
-    def _validate_output_path(self, path: str) -> str:
-        """
-        校验输出路径是否在允许范围内 (P0 修复)
-
-        防止路径穿越攻击，如 /app/config/../../../etc/passwd
-
-        Args:
-            path: 用户提供的输出路径
-
-        Returns:
-            标准化后的绝对路径
-
-        Raises:
-            ValueError: 路径不在允许范围内
-        """
-        abs_path = os.path.abspath(path)
-        for allowed_dir in ALLOWED_OUTPUT_DIRS:
-            allowed_abs = os.path.abspath(allowed_dir)
-            if abs_path.startswith(allowed_abs):
-                return abs_path
-        raise ValueError(f"输出路径不在允许范围内: {path}")
-
-    def _validate_input_path(self, path: str) -> bool:
-        """
-        校验输入路径格式是否合法 (P1-1 修复)
-
-        Args:
-            path: 输入文件路径
-
-        Returns:
-            是否合法
-        """
-        return any(path.startswith(prefix) for prefix in ALLOWED_INPUT_PREFIXES)
-
-    def _validate_transition(self, transition: str) -> str:
-        """
-        校验转场效果是否在白名单内 (P1-2 修复)
-
-        Args:
-            transition: 转场效果名称
-
-        Returns:
-            安全的转场效果名称
-        """
-        if transition not in ALLOWED_TRANSITIONS:
-            logger.warning(f"未知的转场效果 '{transition}'，使用默认 'fade'")
-            return "fade"
-        return transition
-
-    def _get_validated_transition(self, transition: str) -> str:
-        """获取白名单校验后的转场效果名称"""
-        return _XFADE_TRANSITION_MAP.get(self._validate_transition(transition), "fade")
-
-    def compose(self, clips: list[Clip], output_path: Optional[str] = None) -> str:
-        """
-        合成视频
-
-        Args:
-            clips: 视频片段列表，每个片段包含 asset_id 和转场配置
-            output_path: 输出文件路径
-
-        Returns:
-            输出文件路径
-        """
-        if not clips:
-            raise ValueError("clips 不能为空")
-
-        # P1-1: 校验所有输入路径
-        for clip in clips:
-            if not self._validate_input_path(clip.asset_id):
-                raise ValueError(f"不合法的输入路径: {clip.asset_id}")
-
-        # 生成默认输出路径并校验
-        if output_path is None:
-            output_path = self._generate_output_path()
-
-        # P0: 校验输出路径
-        validated_output = self._validate_output_path(output_path)
-
-        logger.info(f"合成视频，片段数: {len(clips)}, 输出: {validated_output}")
-
-        # 获取输入路径列表
-        input_paths = [clip.asset_id for clip in clips]
-
-        try:
-            if self.config.mode == EditingMode.ONE_TAKE:
-                return self._one_take(input_paths, validated_output, clips)
-            elif self.config.mode == EditingMode.PIP:
-                return self._pip(input_paths, validated_output)
-            elif self.config.mode == EditingMode.VOICE_OVER:
-                return self._voice_over(input_paths, validated_output)
-            elif self.config.mode == EditingMode.VOICE_PIP:
-                return self._voice_pip(input_paths, validated_output)
-            else:
-                raise ValueError(f"不支持的剪辑模式: {self.config.mode}")
-        except Exception as e:
-            logger.error(f"视频合成失败: {e}")
-            raise VideoComposeError(f"视频合成失败: {e}") from e
-
-    def _generate_output_path(self) -> str:
-        """生成输出文件路径"""
-        os.makedirs(self.work_dir, exist_ok=True)
-        return os.path.join(self.work_dir, f"output_{self.config.mode}_{os.getpid()}.mp4")
-
-    def _validate_inputs(self, video_paths: list[str], audio_path: Optional[str] = None) -> None:
-        """验证输入文件存在"""
-        for path in video_paths:
-            if not os.path.exists(path):
-                raise FileNotFoundError(f"视频文件不存在: {path}")
-            if not os.path.getsize(path) > 0:
-                raise ValueError(f"视频文件为空: {path}")
-
-        if audio_path and not os.path.exists(audio_path):
-            raise FileNotFoundError(f"音频文件不存在: {audio_path}")
-
-    def _run_ffmpeg(self, command: list[str], capture_output: bool = True) -> tuple:
-        """执行 FFmpeg 命令"""
-        logger.debug(f"Running FFmpeg: {' '.join(command)}")
-        try:
-            result = subprocess.run(
-                command,
-                check=True,
-                stdout=subprocess.PIPE if capture_output else None,
-                stderr=subprocess.PIPE if capture_output else None,
-                text=capture_output,
-            )
-            return result.stdout or "", result.stderr or ""
-        except subprocess.CalledProcessError as e:
-            stderr = e.stderr.decode() if e.stderr else str(e)
-            logger.error(f"FFmpeg error: {stderr}")
-            raise RuntimeError(f"FFmpeg 执行失败: {stderr}") from e
-
-    def _get_video_info(self, video_path: str) -> dict:
-        """获取视频信息"""
-        try:
-            result = subprocess.run(
-                [
-                    self._ffprobe_bin,
-                    "-v",
-                    "error",
-                    "-show_entries",
-                    "stream=width,height,r_frame_rate,duration,codec_name",
-                    "-show_entries",
-                    "format=duration,size",
-                    "-of",
-                    "json",
-                    video_path,
-                ],
-                check=True,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-                text=True,
-            )
-            import json
-
-            data = json.loads(result.stdout)
-            streams = data.get("streams", [{}])
-            video_stream = next((s for s in streams if s.get("codec_type") == "video"), streams[0] if streams else {})
-            fmt = data.get("format", {})
-
-            fps_str = video_stream.get("r_frame_rate", "25/1")
-            fps_parts = fps_str.split("/")
-            fps = float(fps_parts[0]) / float(fps_parts[1]) if len(fps_parts) == 2 else float(fps_parts[0])
-
-            return {
-                "width": int(video_stream.get("width", 0)),
-                "height": int(video_stream.get("height", 0)),
-                "fps": fps,
-                "duration": float(fmt.get("duration", 0)),
-                "codec": video_stream.get("codec_name", "unknown"),
-                "size": int(fmt.get("size", 0)),
-            }
-        except Exception as e:
-            logger.warning(f"获取视频信息失败 {video_path}: {e}")
-            return {"width": 0, "height": 0, "fps": 25, "duration": 0, "codec": "unknown", "size": 0}
-
-    def _get_pip_position_offset(
-        self, main_width: int, main_height: int, pip_width: int, pip_height: int
-    ) -> tuple[int, int]:
-        """获取画中画位置偏移量"""
-        margin = 10
-        position_offsets = {
-            PIPPosition.TOP_LEFT: (margin, margin),
-            PIPPosition.TOP_RIGHT: (main_width - pip_width - margin, margin),
-            PIPPosition.BOTTOM_LEFT: (margin, main_height - pip_height - margin),
-            PIPPosition.BOTTOM_RIGHT: (main_width - pip_width - margin, main_height - pip_height - margin),
-        }
-        return position_offsets.get(self.config.pip_position, position_offsets[PIPPosition.TOP_RIGHT])
-
-    def _normalize_video(self, input_path: str, output_path: str) -> dict:
-        """标准化视频格式"""
-        command = [
-            self._ffmpeg_bin,
-            "-y",
-            "-i",
-            input_path,
-            "-r",
-            str(self.config.output_fps),
-            "-vf",
-            f"scale={self.config.output_width}:{self.config.output_height}:force_original_aspect_ratio=decrease,pad={self.config.output_width}:{self.config.output_height}:(ow-iw)/2:(oh-ih)/2,setsar=1",
-            "-r",
-            str(self.config.output_fps),
-            "-c:v",
-            self.config.output_codec,
-            "-preset",
-            self.config.output_preset,
-            "-crf",
-            str(self.config.output_crf),
-            "-pix_fmt",
-            "yuv420p",
-            "-movflags",
-            "+faststart",
-            "-an",
-            output_path,
-        ]
-        self._run_ffmpeg(command)
-        return self._get_video_info(output_path)
-
-    def _one_take(self, video_paths: list[str], output_path: str, clips: list[Clip]) -> str:
-        """一镜到底模式"""
-        if len(video_paths) == 1:
-            return self._normalize_video(video_paths[0], output_path)
-
-        normalized_paths = []
-        for i, path in enumerate(video_paths):
-            normalized = os.path.join(self.work_dir, f"normalized_{i}_{os.getpid()}.mp4")
-            self._normalize_video(path, normalized)
-            normalized_paths.append(normalized)
-
-        durations = [self._get_video_info(p)["duration"] for p in normalized_paths]
-
-        if len(normalized_paths) <= 5:
-            output_path = self._one_take_with_xfade(normalized_paths, durations, output_path, clips)
-        else:
-            output_path = self._one_take_simple_concat(normalized_paths, output_path)
-
-        for p in normalized_paths:
-            try:
-                if p != output_path:
-                    os.remove(p)
-            except Exception as e:
-                logger.warning(
-                    f"Operation failed in apps/worker/video_processing/video_compose_service.py: {e}", exc_info=True
-                )
-
-        return output_path
-
-    def _one_take_with_xfade(
-        self, normalized_paths: list[str], durations: list[float], output_path: str, clips: list[Clip]
-    ) -> str:
-        """使用 xfade 滤镜实现转场 (P1-2: 转场参数白名单校验)"""
-        if len(normalized_paths) == 2:
-            # 获取当前片段的转场效果并校验白名单
-            transition = "fade"
-            if len(clips) > 1:
-                transition = self._get_validated_transition(clips[1].transition)
-
-            trans_duration = self.config.transition_duration
-            offset1 = durations[0] - trans_duration / 2
-
-            command = [
-                self._ffmpeg_bin,
-                "-y",
-                "-i",
-                normalized_paths[0],
-                "-i",
-                normalized_paths[1],
-                "-filter_complex",
-                f"[0:v][1:v]xfade=transition={transition}:duration={trans_duration}:offset={offset1}[v]",
-                "-map",
-                "[v]",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                output_path,
-            ]
-            self._run_ffmpeg(command)
-            return output_path
-        else:
-            return self._one_take_simple_concat(normalized_paths, output_path)
-
-    def _one_take_simple_concat(self, normalized_paths: list[str], output_path: str) -> str:
-        """使用 concat demuxer 简单拼接"""
-        concat_file = os.path.join(self.work_dir, f"concat_list_{os.getpid()}.txt")
-        with open(concat_file, "w") as f:
-            for path in normalized_paths:
-                f.write(f"file '{os.path.abspath(path)}'\n")
-
-        command = [
-            self._ffmpeg_bin,
-            "-y",
-            "-f",
-            "concat",
-            "-safe",
-            "0",
-            "-i",
-            concat_file,
-            "-c",
-            "copy",
-            output_path,
-        ]
-        self._run_ffmpeg(command)
-
-        try:
-            os.remove(concat_file)
-        except Exception as e:
-            logger.warning(
-                f"Operation failed in apps/worker/video_processing/video_compose_service.py: {e}", exc_info=True
-            )
-
-        return output_path
-
-    def _pip(self, video_paths: list[str], output_path: str) -> str:
-        """画中画模式"""
-        if not video_paths:
-            raise ValueError("No video paths provided")
-
-        main_video = video_paths[0]
-        main_normalized = os.path.join(self.work_dir, f"main_{os.getpid()}.mp4")
-        main_info = self._normalize_video(main_video, main_normalized)
-
-        if len(video_paths) == 1:
-            os.rename(main_normalized, output_path)
-            return output_path
-
-        pip_width = int(self.config.output_width * self.config.pip_scale)
-        pip_height = int(self.config.output_height * self.config.pip_scale)
-        x_offset, y_offset = self._get_pip_position_offset(
-            self.config.output_width, self.config.output_height, pip_width, pip_height
-        )
-
-        pip_normalized = os.path.join(self.work_dir, f"pip_{os.getpid()}.mp4")
-        pip_info = self._get_video_info(video_paths[1])
-
-        if pip_info["duration"] > main_info["duration"]:
-            temp_pip = os.path.join(self.work_dir, f"pip_temp_{os.getpid()}.mp4")
-            command = [
-                self._ffmpeg_bin,
-                "-y",
-                "-i",
-                video_paths[1],
-                "-t",
-                str(main_info["duration"]),
-                "-vf",
-                f"scale={pip_width}:{pip_height}",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                temp_pip,
-            ]
-            self._run_ffmpeg(command)
-            pip_normalized_input = temp_pip
-        else:
-            command = [
-                self._ffmpeg_bin,
-                "-y",
-                "-i",
-                video_paths[1],
-                "-vf",
-                f"scale={pip_width}:{pip_height}",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                pip_normalized,
-            ]
-            self._run_ffmpeg(command)
-            pip_normalized_input = pip_normalized
-
-        if main_info["duration"] > pip_info["duration"]:
-            looped_pip = os.path.join(self.work_dir, f"pip_looped_{os.getpid()}.mp4")
-            command = [
-                self._ffmpeg_bin,
-                "-y",
-                "-stream_loop",
-                "-1",
-                "-i",
-                pip_normalized_input,
-                "-t",
-                str(main_info["duration"]),
-                "-vf",
-                f"scale={pip_width}:{pip_height}",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                looped_pip,
-            ]
-            self._run_ffmpeg(command)
-            pip_normalized_input = looped_pip
-
-        command = [
-            self._ffmpeg_bin,
-            "-y",
-            "-i",
-            main_normalized,
-            "-i",
-            pip_normalized_input,
-            "-filter_complex",
-            f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]",
-            "-map",
-            "[v]",
-            "-c:v",
-            self.config.output_codec,
-            "-preset",
-            self.config.output_preset,
-            "-crf",
-            str(self.config.output_crf),
-            "-pix_fmt",
-            "yuv420p",
-            output_path,
-        ]
-        self._run_ffmpeg(command)
-
-        for temp_file in [main_normalized, pip_normalized]:
-            if temp_file and temp_file != output_path:
-                try:
-                    os.remove(temp_file)
-                except Exception as e:
-                    logger.warning(
-                        f"Operation failed in apps/worker/video_processing/video_compose_service.py: {e}", exc_info=True
-                    )
-
-        return output_path
-
-    def _voice_over(self, video_paths: list[str], audio_path: str, output_path: str) -> str:
-        """口播模式"""
-        if not audio_path:
-            raise ValueError("audio_path is required for VOICE_OVER mode")
-
-        if not video_paths:
-            raise ValueError("No background video provided")
-
-        audio_info = self._get_video_info(audio_path)
-        audio_duration = audio_info["duration"]
-
-        bg_normalized = os.path.join(self.work_dir, f"bg_{os.getpid()}.mp4")
-        bg_info = self._normalize_video(video_paths[0], bg_normalized)
-
-        if bg_info["duration"] < audio_duration:
-            looped_bg = os.path.join(self.work_dir, f"bg_looped_{os.getpid()}.mp4")
-            command = [
-                self._ffmpeg_bin,
-                "-y",
-                "-stream_loop",
-                "-1",
-                "-i",
-                bg_normalized,
-                "-t",
-                str(audio_duration),
-                "-vf",
-                f"scale={self.config.output_width}:{self.config.output_height}",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                looped_bg,
-            ]
-            self._run_ffmpeg(command)
-            bg_normalized = looped_bg
-        elif bg_info["duration"] > audio_duration:
-            temp_bg = os.path.join(self.work_dir, f"bg_trimmed_{os.getpid()}.mp4")
-            command = [
-                self._ffmpeg_bin,
-                "-y",
-                "-i",
-                bg_normalized,
-                "-t",
-                str(audio_duration),
-                "-c:v",
-                "copy",
-                temp_bg,
-            ]
-            self._run_ffmpeg(command)
-            bg_normalized = temp_bg
-
-        blurred_bg = os.path.join(self.work_dir, f"bg_blurred_{os.getpid()}.mp4")
-        command = [
-            self._ffmpeg_bin,
-            "-y",
-            "-i",
-            bg_normalized,
-            "-vf",
-            f"boxblur=5:5,scale={self.config.output_width}:{self.config.output_height}",
-            "-c:v",
-            self.config.output_codec,
-            "-preset",
-            self.config.output_preset,
-            "-crf",
-            str(self.config.output_crf),
-            "-pix_fmt",
-            "yuv420p",
-            blurred_bg,
-        ]
-        self._run_ffmpeg(command)
-
-        command = [
-            self._ffmpeg_bin,
-            "-y",
-            "-i",
-            blurred_bg,
-            "-i",
-            audio_path,
-            "-filter_complex",
-            "[0:v]drawbox=x=0:y=0:w=iw:h=ih:color=black@0.3:t=fill[v]",
-            "-map",
-            "[v]",
-            "-map",
-            "1:a",
-            "-c:v",
-            self.config.output_codec,
-            "-preset",
-            self.config.output_preset,
-            "-crf",
-            str(self.config.output_crf),
-            "-pix_fmt",
-            "yuv420p",
-            "-shortest",
-            output_path,
-        ]
-        self._run_ffmpeg(command)
-
-        for temp_file in [bg_normalized, blurred_bg]:
-            try:
-                if temp_file != output_path:
-                    os.remove(temp_file)
-            except Exception as e:
-                logger.warning(
-                    f"Operation failed in apps/worker/video_processing/video_compose_service.py: {e}", exc_info=True
-                )
-
-        return output_path
-
-    def _voice_pip(self, video_paths: list[str], audio_path: Optional[str], output_path: str) -> str:
-        """口播+画中画模式"""
-        if not video_paths:
-            raise ValueError("No video paths provided")
-
-        if len(video_paths) == 1:
-            return self._normalize_video(video_paths[0], output_path)
-
-        voice_video = video_paths[0]
-        bg_video = video_paths[1] if len(video_paths) > 1 else video_paths[0]
-
-        voice_normalized = os.path.join(self.work_dir, f"voice_{os.getpid()}.mp4")
-        voice_info = self._normalize_video(voice_video, voice_normalized)
-
-        bg_normalized = os.path.join(self.work_dir, f"bg_{os.getpid()}.mp4")
-        bg_info = self._normalize_video(bg_video, bg_normalized)
-
-        final_duration = min(voice_info["duration"], bg_info["duration"])
-
-        pip_width = int(self.config.output_width * self.config.pip_scale)
-        pip_height = int(self.config.output_height * self.config.pip_scale)
-        x_offset, y_offset = self._get_pip_position_offset(
-            self.config.output_width, self.config.output_height, pip_width, pip_height
-        )
-
-        voice_adjusted = os.path.join(self.work_dir, f"voice_adj_{os.getpid()}.mp4")
-        command = [
-            self._ffmpeg_bin,
-            "-y",
-            "-i",
-            voice_normalized,
-            "-t",
-            str(final_duration),
-            "-vf",
-            f"scale={pip_width}:{pip_height}",
-            "-c:v",
-            self.config.output_codec,
-            "-preset",
-            self.config.output_preset,
-            "-crf",
-            str(self.config.output_crf),
-            "-pix_fmt",
-            "yuv420p",
-            voice_adjusted,
-        ]
-        self._run_ffmpeg(command)
-
-        bg_adjusted = os.path.join(self.work_dir, f"bg_adj_{os.getpid()}.mp4")
-        command = [
-            self._ffmpeg_bin,
-            "-y",
-            "-i",
-            bg_normalized,
-            "-t",
-            str(final_duration),
-            "-c:v",
-            "copy",
-            bg_adjusted,
-        ]
-        self._run_ffmpeg(command)
-
-        if audio_path:
-            command = [
-                self._ffmpeg_bin,
-                "-y",
-                "-i",
-                bg_adjusted,
-                "-i",
-                voice_adjusted,
-                "-i",
-                audio_path,
-                "-filter_complex",
-                f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]",
-                "-map",
-                "[v]",
-                "-map",
-                "2:a",
-                "-shortest",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                output_path,
-            ]
-        else:
-            command = [
-                self._ffmpeg_bin,
-                "-y",
-                "-i",
-                bg_adjusted,
-                "-i",
-                voice_adjusted,
-                "-filter_complex",
-                f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]",
-                "-map",
-                "[v]",
-                "-map",
-                "1:a",
-                "-shortest",
-                "-c:v",
-                self.config.output_codec,
-                "-preset",
-                self.config.output_preset,
-                "-crf",
-                str(self.config.output_crf),
-                "-pix_fmt",
-                "yuv420p",
-                output_path,
-            ]
-        self._run_ffmpeg(command)
-
-        for temp_file in [voice_normalized, voice_adjusted, bg_normalized, bg_adjusted]:
-            try:
-                if temp_file != output_path:
-                    os.remove(temp_file)
-            except Exception as e:
-                logger.warning(
-                    f"Operation failed in apps/worker/video_processing/video_compose_service.py: {e}", exc_info=True
-                )
-
-        return output_path
-
-
-def create_compose_service(mode: str, work_dir: Optional[str] = None, **kwargs) -> VideoComposeService:
-    """便捷工厂函数：创建视频合成服务"""
-    try:
-        editing_mode = EditingMode(mode)
-    except ValueError:
-        raise ValueError(f"无效的剪辑模式: {mode}. 有效模式: {[m.value for m in EditingMode]}")
-
-    config = EditingModeConfig(
-        mode=editing_mode,
-        output_width=kwargs.get("output_width", 1280),
-        output_height=kwargs.get("output_height", 720),
-        output_fps=kwargs.get("output_fps", 25),
-        pip_position=PIPPosition(kwargs.get("pip_position", "top_right")),
-        pip_scale=kwargs.get("pip_scale", 0.25),
-        transition_duration=kwargs.get("transition_duration", 0.5),
-    )
-
-    return VideoComposeService(config=config, work_dir=work_dir)

apps/worker/worker_app/core/config.py

@@ -18,6 +18,9 @@ class WorkerSettings(BaseSettings):
     environment: str = "development"
     auto_create_schema: bool = False
 
+    # 渲染引擎选择：legacy=旧VideoComposeService，unified=新UnifiedRenderService
+    render_engine: str = "legacy"
+
     model_config = SettingsConfigDict(
         env_file=".env",
         env_file_encoding="utf-8",

apps/worker/worker_app/tasks/compose_video.py

@@ -39,6 +39,10 @@ def _get_job_service():
 def compose_video(self, job_id: str, **kwargs):
     """视频合成任务。
 
+    根据 RENDER_ENGINE 配置选择渲染引擎：
+      - legacy: 旧 VideoComposeService（filter_complex 模式）
+      - unified: 新 UnifiedRenderService（图层架构）
+
     Args:
         job_id: JobService 中的任务 ID
         **kwargs: 来自 Job.payload 的额外参数（plan_id, output_path 等）
@@ -56,66 +60,18 @@ def compose_video(self, job_id: str, **kwargs):
             job_service.fail_job(job_id, "Missing plan_id in job payload")
             return {"status": "error", "message": "Missing plan_id"}
 
-        # 标记为 running
-        job_service.update_progress(job_id, progress=10.0, current_stage="初始化合成环境")
+        # 判断使用哪个渲染引擎
+        # 优先级：Redis Feature Flag（白名单 > 百分比） > 环境变量默认
+        from video_processing.render_engine_resolver import get_render_engine_resolver
 
-        # 延迟导入 VideoComposeService
-        from apps.api.app.services.video_compose_service import VideoComposeService
+        resolver = get_render_engine_resolver()
+        user_id = job.created_by_user_id or None
+        engine = resolver.get_engine(user_id=user_id)
 
-        compose_svc = VideoComposeService(db)
-
-        # 校验合成条件
-        job_service.update_progress(job_id, progress=20.0, current_stage="校验合成条件")
-        validation = compose_svc.validate_compose(plan_id)
-        if not validation.valid:
-            error_msg = "; ".join(validation.errors)
-            job_service.fail_job(job_id, f"合成校验失败: {error_msg}")
-            return {"status": "error", "message": error_msg}
-
-        # 构建合成命令
-        job_service.update_progress(job_id, progress=30.0, current_stage="构建 FFmpeg 命令")
-        _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output"))
-        output_path = os.path.join(_output_dir, f"{job_id}.mp4")
-        compose_cmd = compose_svc.build_compose_command(plan_id, output_path)
-
-        # 执行 FFmpeg
-        job_service.update_progress(job_id, progress=50.0, current_stage="正在执行视频合成")
-        logger.info("Executing FFmpeg for job %s, plan %s", job_id, 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:
-            job_service.fail_job(job_id, f"FFmpeg 执行失败: {e.stderr[:500]}")
-            raise
-
-        # 上传结果
-        job_service.update_progress(job_id, progress=80.0, current_stage="上传合成结果")
-        storage_key = f"rendered/{plan_id}/{job_id}.mp4"
-
-        from worker_app.tasks.edit_plan_generation import _upload_to_oss
-
-        output_url = _upload_to_oss(Path(output_path), storage_key)
-
-        # 更新 Job 状态为完成
-        result_data = {
-            "plan_id": plan_id,
-            "output_path": output_path,
-            "storage_key": storage_key,
-            "output_url": output_url or "",
-            "estimated_duration": compose_cmd.estimated_duration,
-            "clip_count": len(compose_cmd.clip_chains),
-        }
-        job_service.complete_job(job_id, result=result_data)
-
-        logger.info("视频合成完成: job_id=%s, plan_id=%s", job_id, plan_id)
-        return {"status": "completed", "job_id": job_id, "result": result_data}
+        if engine == "unified":
+            return _compose_with_unified_engine(self, job_service, job, plan_id, db)
+        else:
+            return _compose_with_legacy_engine(self, job_service, job, plan_id, db)
 
     except self.retry_exc as exc:
         logger.warning("视频合成重试中: job_id=%s, exc=%s", job_id, exc)
@@ -129,11 +85,145 @@ def compose_video(self, job_id: str, **kwargs):
         raise self.retry(exc=exc, countdown=60)
     finally:
         db.close()
-        # 清理临时文件
+
+
+def _compose_with_legacy_engine(task, job_service, job, plan_id: str, db) -> dict:
+    """旧引擎渲染路径（VideoComposeService）。"""
+    job_id = job.id
+
+    # 标记为 running
+    job_service.update_progress(job_id, progress=10.0, current_stage="初始化合成环境")
+
+    # 延迟导入 VideoComposeService
+    from apps.api.app.services.video_compose_service import VideoComposeService
+
+    compose_svc = VideoComposeService(db)
+
+    # 校验合成条件
+    job_service.update_progress(job_id, progress=20.0, current_stage="校验合成条件")
+    validation = compose_svc.validate_compose(plan_id)
+    if not validation.valid:
+        error_msg = "; ".join(validation.errors)
+        job_service.fail_job(job_id, f"合成校验失败: {error_msg}")
+        return {"status": "error", "message": error_msg}
+
+    # 构建合成命令
+    job_service.update_progress(job_id, progress=30.0, current_stage="构建 FFmpeg 命令")
+    _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output"))
+    output_path = os.path.join(_output_dir, f"{job_id}.mp4")
+    compose_cmd = compose_svc.build_compose_command(plan_id, output_path)
+
+    # 执行 FFmpeg
+    job_service.update_progress(job_id, progress=50.0, current_stage="正在执行视频合成")
+    logger.info("Executing FFmpeg for job %s, plan %s", job_id, 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:
+        job_service.fail_job(job_id, f"FFmpeg 执行失败: {e.stderr[:500]}")
+        raise
+
+    # 上传结果
+    job_service.update_progress(job_id, progress=80.0, current_stage="上传合成结果")
+    storage_key = f"rendered/{plan_id}/{job_id}.mp4"
+
+    from worker_app.tasks.edit_plan_generation import _upload_to_oss
+
+    output_url = _upload_to_oss(Path(output_path), storage_key)
+
+    # 更新 Job 状态为完成
+    result_data = {
+        "plan_id": plan_id,
+        "output_path": output_path,
+        "storage_key": storage_key,
+        "output_url": output_url or "",
+        "estimated_duration": compose_cmd.estimated_duration,
+        "clip_count": len(compose_cmd.clip_chains),
+        "engine": "legacy",
+    }
+    job_service.complete_job(job_id, result=result_data)
+
+    logger.info("视频合成完成(legacy): job_id=%s, plan_id=%s", job_id, plan_id)
+    return {"status": "completed", "job_id": job_id, "result": result_data}
+
+
+def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> dict:
+    """新引擎渲染路径（UnifiedRenderService + RenderAdapter）。"""
+    job_id = job.id
+
+    # 标记为 running
+    job_service.update_progress(job_id, progress=10.0, current_stage="初始化统一渲染引擎")
+
+    from video_processing.render_adapter import RenderAdapter
+
+    adapter = RenderAdapter(db)
+
+    # 校验合成条件
+    job_service.update_progress(job_id, progress=15.0, current_stage="校验合成条件")
+    valid, errors, warnings, ready_count, total_count = adapter.validate_plan(plan_id)
+    if not valid:
+        error_msg = "; ".join(errors)
+        job_service.fail_job(job_id, f"合成校验失败: {error_msg}")
+        return {"status": "error", "message": error_msg}
+
+    # 进度回调
+    def progress_cb(progress: float, stage: str) -> None:
         try:
-            _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output"))
-            output_path = os.path.join(_output_dir, f"{job_id}.mp4")
-            if Path(output_path).exists():
-                Path(output_path).unlink()
-        except Exception as e:
-            logger.warning(f"Operation failed in apps/worker/worker_app/tasks/compose_video.py: {e}", exc_info=True)
+            job_service.update_progress(job_id, progress=progress, current_stage=stage)
+        except Exception:
+            logger.exception("更新进度失败")
+
+    # 执行渲染
+    job_service.update_progress(job_id, progress=20.0, current_stage="开始渲染")
+    logger.info("统一渲染引擎开始: job_id=%s plan_id=%s", job_id, plan_id)
+
+    result = adapter.render_plan(
+        plan_id=plan_id,
+        job_id=job_id,
+        progress_cb=progress_cb,
+    )
+
+    if not result.success:
+        job_service.fail_job(job_id, f"渲染失败: {result.error_message}")
+        raise RuntimeError(result.error_message)
+
+    # 更新 Job 状态为完成
+    result_data = {
+        "plan_id": plan_id,
+        "output_path": str(result.output_path) if result.output_path else "",
+        "storage_key": f"rendered/{plan_id}/{job_id}.mp4",
+        "output_url": result.output_url,
+        "estimated_duration": result.duration,
+        "clip_count": result.clip_count,
+        "engine": "unified",
+        "width": result.width,
+        "height": result.height,
+        "file_size": result.file_size,
+    }
+    job_service.complete_job(job_id, result=result_data)
+
+    logger.info(
+        "视频合成完成(unified): job_id=%s plan_id=%s duration=%.2fs",
+        job_id,
+        plan_id,
+        result.duration,
+    )
+    return {"status": "completed", "job_id": job_id, "result": result_data}
+
+
+def _cleanup_output(job_id: str) -> None:
+    """清理临时输出文件。"""
+    try:
+        _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output"))
+        output_path = os.path.join(_output_dir, f"{job_id}.mp4")
+        if Path(output_path).exists():
+            Path(output_path).unlink()
+    except Exception as e:
+        logger.warning(f"清理输出文件失败: {e}", exc_info=True)

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -116,6 +116,15 @@ def render_edit_plan(self, plan_id: str) -> dict:
                 rendered_clip_ids: list[str] = []
                 failed_clip_ids: list[str] = []
 
+                # 预先批量查询所有素材的 storage_key（file_url）
+                from packages.adapters.sqlalchemy_impl.models import AssetModel
+
+                clip_asset_ids = [c.asset_id for c in clips if c.asset_id]
+                asset_storage_map: dict[str, str] = {}
+                if clip_asset_ids:
+                    assets = db.query(AssetModel).filter(AssetModel.id.in_(clip_asset_ids)).all()
+                    asset_storage_map = {a.id: a.file_url for a in assets if a.file_url}
+
                 for clip in clips:
                     if not clip.asset_id:
                         # 没有素材的片段跳过，标记为失败
@@ -129,10 +138,22 @@ def render_edit_plan(self, plan_id: str) -> dict:
                         rendered_clip_ids.append(clip.id)
                         continue
 
+                    storage_key = asset_storage_map.get(clip.asset_id)
+                    if not storage_key:
+                        logger.warning(
+                            "片段素材无 storage_key，跳过: clip_id=%s asset_id=%s",
+                            clip.id,
+                            clip.asset_id,
+                        )
+                        clip.mark_failed()
+                        clip_repo.update(clip)
+                        failed_clip_ids.append(clip.id)
+                        continue
+
                     # 下载素材
-                    ext = Path(clip.asset_id).suffix or ".mp4"
+                    ext = Path(storage_key).suffix or ".mp4"
                     local_path = tmpdir_path / f"clip_{clip.order:04d}{ext}"
-                    if download_asset(clip.asset_id, local_path):
+                    if download_asset(storage_key, local_path):
                         asset_path_map[clip.asset_id] = local_path
                         rendered_clip_ids.append(clip.id)
                     else:

apps/worker/worker_app/tasks/generation.py

@@ -124,6 +124,7 @@ class _VirtualPlan:
 
     id: str
     name: str = ""
+    config: dict[str, Any] = field(default_factory=dict)
 
 
 @dataclass
@@ -161,13 +162,15 @@ def _build_plan_and_clips_from_task(
     """
     plan = _VirtualPlan(id=task_id, name=f"Generated-{task_id[:8]}")
 
-    # 为每个下载路径生成合成 asset_id
+    # 为每个下载路径生成合成 asset_id，并预探测素材时长
     asset_path_map: dict[str, Path] = {}
     path_to_asset_id: dict[Path, str] = {}
+    path_duration: dict[Path, float] = {}
     for i, p in enumerate(downloaded_paths):
         asset_id = f"gen_{task_id[:8]}_{i:03d}{p.suffix or '.mp4'}"
         asset_path_map[asset_id] = p
         path_to_asset_id[p] = asset_id
+        path_duration[p] = probe_duration(p)
 
     clips: list[_VirtualClip] = []
     n = len(downloaded_paths)
@@ -183,6 +186,7 @@ def _build_plan_and_clips_from_task(
                     clip_type=clip_type,
                     order=i,
                     asset_id=path_to_asset_id[p],
+                    duration=path_duration[p],
                 )
             )
     elif mode == "voice_over":
@@ -195,6 +199,7 @@ def _build_plan_and_clips_from_task(
                     clip_type="main",
                     order=i,
                     asset_id=path_to_asset_id[p],
+                    duration=path_duration[p],
                     config={"role": "b_roll"},
                 )
             )
@@ -214,6 +219,7 @@ def _build_plan_and_clips_from_task(
                     clip_type=clip_type,
                     order=i,
                     asset_id=path_to_asset_id[p],
+                    duration=path_duration[p],
                 )
             )
     else:
@@ -226,6 +232,7 @@ def _build_plan_and_clips_from_task(
                     clip_type="main",
                     order=i,
                     asset_id=path_to_asset_id[p],
+                    duration=path_duration[p],
                 )
             )
 
@@ -380,31 +387,37 @@ def _download_library_assets(
 
         session = SessionLocal()
         try:
-            # 构建查询：根据模式选择不同的过滤条件
+            # 构建查询
             query = session.query(AssetModel).filter(
                 AssetModel.status == "ready",
                 AssetModel.file_type.in_(["video", "video/mp4", "video/quicktime"]),
             )
 
-            if asset_library_id:
-                # 素材库模式
-                query = query.filter(AssetModel.asset_library_id == asset_library_id)
+            if asset_ids:
+                # 明确指定了 asset_ids：直接按 ID 查，不预先按 library/project 过滤
+                # 避免项目级素材或跨库素材因为 library_id 不匹配而查不到
+                # 归属安全由后面的归属校验保证
+                query = query.filter(AssetModel.id.in_(asset_ids))
                 logger.info(
-                    "下载素材库视频: asset_library_id=%s asset_ids=%s",
-                    asset_library_id,
-                    asset_ids or "all",
+                    "下载指定素材: asset_ids=%d 个, asset_library_id=%s, project_id=%s",
+                    len(asset_ids),
+                    asset_library_id or "none",
+                    project_id or "none",
                 )
             else:
-                # 项目级模式
-                query = query.filter(AssetModel.project_id == project_id)
-                logger.info(
-                    "下载项目级视频: project_id=%s asset_ids=%s",
-                    project_id,
-                    asset_ids or "all",
-                )
-
-            if asset_ids:
-                query = query.filter(AssetModel.id.in_(asset_ids))
+                # 未指定 asset_ids：按 library 或 project 下载全部 ready 视频
+                if asset_library_id:
+                    query = query.filter(AssetModel.asset_library_id == asset_library_id)
+                    logger.info(
+                        "下载素材库全部视频: asset_library_id=%s",
+                        asset_library_id,
+                    )
+                else:
+                    query = query.filter(AssetModel.project_id == project_id)
+                    logger.info(
+                        "下载项目全部视频: project_id=%s",
+                        project_id,
+                    )
 
             assets = query.order_by(AssetModel.created_at).all()
 
@@ -421,13 +434,15 @@ def _download_library_assets(
                 if missing_ids:
                     raise ValueError(f"素材不存在: asset_ids={sorted(missing_ids)}")
                 for asset in assets:
+                    # 校验素材库归属（只要传了 asset_library_id 就校验）
                     if asset_library_id and asset.asset_library_id != asset_library_id:
                         raise ValueError(
                             f"素材不属于指定素材库: asset_id={asset.id}, "
                             f"expected_asset_library_id={asset_library_id}, "
                             f"actual_asset_library_id={asset.asset_library_id}"
                         )
-                    if not asset_library_id and project_id and asset.project_id != project_id:
+                    # 校验项目归属（只要传了 project_id 就校验）
+                    if project_id and asset.project_id != project_id:
                         raise ValueError(
                             f"素材不属于指定项目: asset_id={asset.id}, "
                             f"expected_project_id={project_id}, "

packages/adapters/redis/__init__.py

@@ -1,3 +1,9 @@
+from packages.adapters.redis.feature_flag_store import (
+    FeatureFlagConfig,
+    FeatureFlagStore,
+    InMemoryFeatureFlagStore,
+    RedisFeatureFlagStore,
+)
 from packages.adapters.redis.session_store import (
     NoopSessionStore,
     RedisConfig,
@@ -5,4 +11,13 @@ from packages.adapters.redis.session_store import (
     get_session_store,
 )
 
-__all__ = ["NoopSessionStore", "RedisConfig", "SessionStore", "get_session_store"]
+__all__ = [
+    "FeatureFlagConfig",
+    "FeatureFlagStore",
+    "InMemoryFeatureFlagStore",
+    "NoopSessionStore",
+    "RedisConfig",
+    "RedisFeatureFlagStore",
+    "SessionStore",
+    "get_session_store",
+]

packages/adapters/redis/feature_flag_store.py

@@ -0,0 +1,259 @@
+"""Feature Flag 存储实现。
+
+支持两种后端：
+- RedisFeatureFlagStore：生产环境使用，支持多实例共享、热更新
+- InMemoryFeatureFlagStore：测试/开发环境使用，纯内存
+
+支持的 Flag 类型：
+- 全局开关（enabled: bool）
+- 白名单（whitelist: Set[str]，如 user_id 列表）
+- 百分比切流（percentage: 0-100，基于标识符哈希取模）
+
+判定优先级：白名单 > 百分比 > 全局开关
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import threading
+import time
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Optional, Set
+
+logger = logging.getLogger(__name__)
+
+# Redis key 前缀
+FEATURE_FLAG_REDIS_PREFIX = "feature_flag:"
+
+
+@dataclass
+class FeatureFlagConfig:
+    """单个 Feature Flag 的配置。"""
+
+    name: str
+    enabled: bool = False
+    percentage: int = 0  # 0-100
+    whitelist: Set[str] = field(default_factory=set)
+
+    def to_dict(self) -> dict:
+        return {
+            "name": self.name,
+            "enabled": self.enabled,
+            "percentage": self.percentage,
+            "whitelist": sorted(self.whitelist),
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "FeatureFlagConfig":
+        return cls(
+            name=data["name"],
+            enabled=bool(data.get("enabled", False)),
+            percentage=int(data.get("percentage", 0)),
+            whitelist=set(data.get("whitelist", [])),
+        )
+
+    def is_active(self, identifier: Optional[str] = None) -> bool:
+        """判断当前 flag 是否激活。
+
+        判定优先级：
+        1. 全局关闭 → False
+        2. 白名单匹配 → True
+        3. 百分比命中 → True
+        4. 其他 → False
+
+        Args:
+            identifier: 用于白名单匹配和百分比哈希的标识符（如 user_id）。
+                       传 None 时只看全局开关 + 百分比（百分比用随机值）。
+        """
+        if not self.enabled:
+            return False
+
+        # 白名单：精确匹配
+        if identifier and identifier in self.whitelist:
+            return True
+
+        # 百分比：0 直接 False，100 直接 True
+        if self.percentage <= 0:
+            # 没有白名单且百分比为0 → 未启用
+            return False
+        if self.percentage >= 100:
+            return True
+
+        # 基于 identifier 做哈希取模，确保同一用户始终落在同一侧
+        if identifier:
+            hash_val = int(
+                hashlib.md5(f"{self.name}:{identifier}".encode("utf-8")).hexdigest(), 16  # nosec B324
+            )  # nosec B324 - 用于哈希取模做百分比切流，非安全用途
+            return (hash_val % 100) < self.percentage
+
+        # 无 identifier 且百分比在 0-100 之间 → 按比例随机（不保证一致性）
+        import random
+
+        return random.randint(0, 99) < self.percentage
+
+
+class FeatureFlagStore(ABC):
+    """Feature Flag 存储抽象接口。"""
+
+    @abstractmethod
+    def get(self, name: str) -> FeatureFlagConfig:
+        """获取指定 flag 的配置，不存在则返回默认配置（关闭状态）。"""
+        ...
+
+    @abstractmethod
+    def set(self, config: FeatureFlagConfig) -> None:
+        """设置 flag 配置。"""
+        ...
+
+    @abstractmethod
+    def delete(self, name: str) -> bool:
+        """删除 flag，返回是否成功删除。"""
+        ...
+
+    @abstractmethod
+    def list_all(self) -> dict[str, FeatureFlagConfig]:
+        """列出所有 flag。"""
+        ...
+
+    def is_active(self, name: str, identifier: Optional[str] = None) -> bool:
+        """便捷方法：判断 flag 是否激活。"""
+        return self.get(name).is_active(identifier)
+
+
+class InMemoryFeatureFlagStore(FeatureFlagStore):
+    """内存实现，用于测试和本地开发。"""
+
+    def __init__(self) -> None:
+        self._flags: dict[str, FeatureFlagConfig] = {}
+        self._lock = threading.Lock()
+
+    def get(self, name: str) -> FeatureFlagConfig:
+        with self._lock:
+            return self._flags.get(name, FeatureFlagConfig(name=name, enabled=False))
+
+    def set(self, config: FeatureFlagConfig) -> None:
+        with self._lock:
+            self._flags[config.name] = config
+
+    def delete(self, name: str) -> bool:
+        with self._lock:
+            if name in self._flags:
+                del self._flags[name]
+                return True
+            return False
+
+    def list_all(self) -> dict[str, FeatureFlagConfig]:
+        with self._lock:
+            return dict(self._flags)
+
+
+class RedisFeatureFlagStore(FeatureFlagStore):
+    """Redis 实现，支持多实例共享配置。
+
+    每个 flag 存在一个独立的 Redis hash key 中：
+      Key: feature_flag:{name}
+      Fields: enabled, percentage, whitelist(JSON array)
+    """
+
+    def __init__(self, redis_url: str, key_prefix: str = FEATURE_FLAG_REDIS_PREFIX) -> None:
+        import redis as redis_lib
+
+        self._redis = redis_lib.from_url(redis_url, decode_responses=True)
+        self._key_prefix = key_prefix
+        # 本地缓存 + TTL，减少 Redis 调用
+        self._cache: dict[str, tuple[FeatureFlagConfig, float]] = {}
+        self._cache_ttl = 5.0  # 秒，默认5秒本地缓存
+        self._lock = threading.Lock()
+
+    def _redis_key(self, name: str) -> str:
+        return f"{self._key_prefix}{name}"
+
+    def _parse_whitelist(self, raw: Optional[str]) -> Set[str]:
+        if not raw:
+            return set()
+        try:
+            data = json.loads(raw)
+            return set(data) if isinstance(data, list) else set()
+        except (json.JSONDecodeError, TypeError):
+            return set()
+
+    def get(self, name: str) -> FeatureFlagConfig:
+        now = time.time()
+
+        # 先查本地缓存
+        with self._lock:
+            cached = self._cache.get(name)
+            if cached and now - cached[1] < self._cache_ttl:
+                return cached[0]
+
+        # 从 Redis 读取
+        try:
+            key = self._redis_key(name)
+            data = self._redis.hgetall(key)
+            if not data:
+                config = FeatureFlagConfig(name=name, enabled=False)
+            else:
+                config = FeatureFlagConfig(
+                    name=name,
+                    enabled=(data.get("enabled", "0") in ("1", "true", "True")),
+                    percentage=int(data.get("percentage", 0)),
+                    whitelist=self._parse_whitelist(data.get("whitelist")),
+                )
+
+            # 写入本地缓存
+            with self._lock:
+                self._cache[name] = (config, now)
+
+            return config
+        except Exception as exc:
+            logger.warning("Failed to get feature flag %s from Redis: %s", name, exc)
+            # Redis 不可用时返回默认值（关闭），不影响业务
+            return FeatureFlagConfig(name=name, enabled=False)
+
+    def set(self, config: FeatureFlagConfig) -> None:
+        key = self._redis_key(config.name)
+        self._redis.hset(
+            key,
+            mapping={
+                "enabled": "1" if config.enabled else "0",
+                "percentage": str(config.percentage),
+                "whitelist": json.dumps(sorted(config.whitelist), ensure_ascii=False),
+            },
+        )
+        # 失效本地缓存
+        with self._lock:
+            self._cache.pop(config.name, None)
+
+    def delete(self, name: str) -> bool:
+        key = self._redis_key(name)
+        result = self._redis.delete(key)
+        with self._lock:
+            self._cache.pop(name, None)
+        return bool(result)
+
+    def list_all(self) -> dict[str, FeatureFlagConfig]:
+        pattern = f"{self._key_prefix}*"
+        result: dict[str, FeatureFlagConfig] = {}
+        try:
+            cursor = 0
+            while True:
+                cursor, keys = self._redis.scan(cursor=cursor, match=pattern, count=100)
+                for key in keys:
+                    name = key[len(self._key_prefix) :]
+                    result[name] = self.get(name)
+                if cursor == 0:
+                    break
+        except Exception as exc:
+            logger.warning("Failed to list feature flags from Redis: %s", exc)
+        return result
+
+    def invalidate_cache(self, name: Optional[str] = None) -> None:
+        """手动失效本地缓存。"""
+        with self._lock:
+            if name:
+                self._cache.pop(name, None)
+            else:
+                self._cache.clear()

packages/domain/entities.py

@@ -134,6 +134,25 @@ class AssetStatus(StrEnum):
     PROCESSING = "processing"
     ERROR = "error"
 
+    @classmethod
+    def _missing_(cls, value: object) -> "AssetStatus":
+        """兼容历史数据，避免枚举转换失败导致500。
+
+        - uploaded → READY（早期版本用 uploaded 表示上传完成）
+        - 其他未知值 → READY（兜底，不阻塞业务）
+        """
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in ("uploaded", "success", "ok", "done", "complete"):
+                return cls.READY
+            if normalized in ("upload", "uploading_start", "upload_start"):
+                return cls.UPLOADING
+            if normalized in ("failed", "fail", "err"):
+                return cls.ERROR
+            if normalized in ("process", "processing", "running", "run"):
+                return cls.PROCESSING
+        return cls.READY
+
 
 class ClassificationStatus(StrEnum):
     PENDING = "pending"
@@ -141,6 +160,23 @@ class ClassificationStatus(StrEnum):
     COMPLETED = "completed"
     FAILED = "failed"
 
+    @classmethod
+    def _missing_(cls, value: object) -> "ClassificationStatus":
+        """兼容历史数据，避免枚举转换失败导致500。
+
+        - done → COMPLETED（早期版本用 done 表示完成）
+        - 其他未知值 → PENDING（兜底，不阻塞业务）
+        """
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in ("done", "success", "finished", "complete"):
+                return cls.COMPLETED
+            if normalized in ("fail", "error", "err"):
+                return cls.FAILED
+            if normalized in ("process", "processing", "running", "run"):
+                return cls.PROCESSING
+        return cls.PENDING
+
 
 @dataclass(slots=True)
 class Asset:

scripts/ci_notify_success.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""发送 CI 成功通知到飞书/项目群 webhook。"""
+
+import json
+import os
+import sys
+import urllib.request
+
+
+def main() -> int:
+    webhook = os.environ.get("CI_NOTIFY_WEBHOOK", "")
+    if not webhook:
+        print("未配置 CI_NOTIFY_WEBHOOK，跳过成功通知")
+        print("如需启用，请在仓库 Settings -> Secrets and variables -> Actions 中添加 CI_NOTIFY_WEBHOOK")
+        return 0
+
+    success_job = os.environ.get("SUCCESS_JOB", "Unknown Job")
+    branch = os.environ.get("GITHUB_REF_NAME", "unknown")
+    commit = os.environ.get("GITHUB_SHA", "unknown")[:8]
+    actor = os.environ.get("GITHUB_ACTOR", "unknown")
+    run_id = os.environ.get("GITHUB_RUN_ID", "unknown")
+    repo = os.environ.get("GITHUB_REPOSITORY", "unknown")
+    run_url = f"https://git.xiaoxiajianji.com/{repo}/actions/runs/{run_id}"
+
+    payload = {
+        "msg_type": "interactive",
+        "card": {
+            "header": {
+                "title": {
+                    "tag": "plain_text",
+                    "content": "✅ CI 构建成功",
+                },
+                "status": "green",
+            },
+            "elements": [
+                {
+                    "tag": "div",
+                    "text": {
+                        "tag": "lark_md",
+                        "content": (
+                            f"**任务**: {success_job}\n"
+                            f"**分支**: {branch}\n"
+                            f"**提交**: {commit}\n"
+                            f"**提交者**: {actor}\n"
+                            f"**Run ID**: {run_id}"
+                        ),
+                    },
+                },
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": "查看构建详情"},
+                            "url": run_url,
+                            "type": "primary",
+                        }
+                    ],
+                },
+            ],
+        },
+    }
+
+    data = json.dumps(payload).encode("utf-8")
+    req = urllib.request.Request(
+        webhook,
+        data=data,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            resp.read()
+        print("成功通知已发送")
+    except Exception as e:
+        print(f"成功通知发送失败: {e}", file=sys.stderr)
+        return 1
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/unit/test_asset_status_enum.py

@@ -0,0 +1,72 @@
+"""AssetStatus 枚举兼容性测试。
+
+验证历史脏数据（如 'uploaded'）不会导致枚举转换失败。
+"""
+
+import pytest
+
+from packages.domain.entities import AssetStatus
+
+
+class TestAssetStatusNormalValues:
+    """正常值应该正确映射。"""
+
+    def test_uploading(self):
+        assert AssetStatus("uploading") == AssetStatus.UPLOADING
+
+    def test_ready(self):
+        assert AssetStatus("ready") == AssetStatus.READY
+
+    def test_processing(self):
+        assert AssetStatus("processing") == AssetStatus.PROCESSING
+
+    def test_error(self):
+        assert AssetStatus("error") == AssetStatus.ERROR
+
+
+class TestAssetStatusHistoricalValues:
+    """历史脏数据应该正确映射到对应状态，不抛异常。"""
+
+    @pytest.mark.parametrize("value", ["uploaded", "Uploaded", "UPLOADED", " uploaded "])
+    def test_uploaded_maps_to_ready(self, value):
+        """生产环境发现的 'uploaded' 历史值应映射为 READY。"""
+        assert AssetStatus(value) == AssetStatus.READY
+
+    @pytest.mark.parametrize("value", ["success", "ok", "done", "complete"])
+    def test_other_ready_like_values_map_to_ready(self, value):
+        assert AssetStatus(value) == AssetStatus.READY
+
+    @pytest.mark.parametrize("value", ["upload", "uploading_start", "upload_start"])
+    def test_upload_like_values_map_to_uploading(self, value):
+        assert AssetStatus(value) == AssetStatus.UPLOADING
+
+    @pytest.mark.parametrize("value", ["failed", "fail", "err"])
+    def test_error_like_values_map_to_error(self, value):
+        assert AssetStatus(value) == AssetStatus.ERROR
+
+    @pytest.mark.parametrize("value", ["process", "running", "run"])
+    def test_processing_like_values_map_to_processing(self, value):
+        assert AssetStatus(value) == AssetStatus.PROCESSING
+
+
+class TestAssetStatusFallback:
+    """完全未知的值兜底为 READY，不抛500。"""
+
+    @pytest.mark.parametrize("value", ["unknown", "foo_bar", ""])
+    def test_unknown_value_falls_back_to_ready(self, value):
+        assert AssetStatus(value) == AssetStatus.READY
+
+    def test_none_value_falls_back_to_ready(self):
+        assert AssetStatus(None) == AssetStatus.READY  # type: ignore[arg-type]
+
+    def test_int_value_falls_back_to_ready(self):
+        assert AssetStatus(123) == AssetStatus.READY  # type: ignore[arg-type]
+
+
+class TestAssetStatusStrValue:
+    """枚举值仍为字符串类型，不影响序列化。"""
+
+    def test_value_unchanged(self):
+        assert AssetStatus.READY.value == "ready"
+        assert AssetStatus.ERROR.value == "error"
+        assert isinstance(AssetStatus.READY, str)

tests/unit/test_classification_status_enum.py

@@ -0,0 +1,68 @@
+"""ClassificationStatus 枚举兼容性测试。
+
+验证历史脏数据（如 'done'）不会导致枚举转换失败。
+"""
+
+import pytest
+
+from packages.domain.entities import ClassificationStatus
+
+
+class TestClassificationStatusNormalValues:
+    """正常值应该正确映射。"""
+
+    def test_pending(self):
+        assert ClassificationStatus("pending") == ClassificationStatus.PENDING
+
+    def test_processing(self):
+        assert ClassificationStatus("processing") == ClassificationStatus.PROCESSING
+
+    def test_completed(self):
+        assert ClassificationStatus("completed") == ClassificationStatus.COMPLETED
+
+    def test_failed(self):
+        assert ClassificationStatus("failed") == ClassificationStatus.FAILED
+
+
+class TestClassificationStatusHistoricalValues:
+    """历史脏数据应该正确映射到对应状态，不抛异常。"""
+
+    @pytest.mark.parametrize("value", ["done", "Done", "DONE", " done "])
+    def test_done_maps_to_completed(self, value):
+        """生产环境发现的 'done' 历史值应映射为 COMPLETED。"""
+        assert ClassificationStatus(value) == ClassificationStatus.COMPLETED
+
+    @pytest.mark.parametrize("value", ["success", "finished", "complete"])
+    def test_other_done_like_values_map_to_completed(self, value):
+        assert ClassificationStatus(value) == ClassificationStatus.COMPLETED
+
+    @pytest.mark.parametrize("value", ["fail", "error", "err"])
+    def test_error_like_values_map_to_failed(self, value):
+        assert ClassificationStatus(value) == ClassificationStatus.FAILED
+
+    @pytest.mark.parametrize("value", ["process", "running", "run"])
+    def test_processing_like_values_map_to_processing(self, value):
+        assert ClassificationStatus(value) == ClassificationStatus.PROCESSING
+
+
+class TestClassificationStatusFallback:
+    """完全未知的值兜底为 PENDING，不抛500。"""
+
+    @pytest.mark.parametrize("value", ["unknown", "foo_bar", ""])
+    def test_unknown_value_falls_back_to_pending(self, value):
+        assert ClassificationStatus(value) == ClassificationStatus.PENDING
+
+    def test_none_value_falls_back_to_pending(self):
+        assert ClassificationStatus(None) == ClassificationStatus.PENDING  # type: ignore[arg-type]
+
+    def test_int_value_falls_back_to_pending(self):
+        assert ClassificationStatus(123) == ClassificationStatus.PENDING  # type: ignore[arg-type]
+
+
+class TestClassificationStatusStrValue:
+    """枚举值仍为字符串类型，不影响序列化。"""
+
+    def test_value_unchanged(self):
+        assert ClassificationStatus.COMPLETED.value == "completed"
+        assert ClassificationStatus.PENDING.value == "pending"
+        assert isinstance(ClassificationStatus.COMPLETED, str)

tests/unit/test_feature_flag.py

@@ -0,0 +1,424 @@
+"""Feature Flag 单元测试。
+
+测试 FeatureFlagConfig、InMemoryFeatureFlagStore、RenderEngineResolver 的核心逻辑。
+"""
+
+from __future__ import annotations
+
+import time
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from packages.adapters.redis.feature_flag_store import (
+    FeatureFlagConfig,
+    InMemoryFeatureFlagStore,
+)
+
+# ── FeatureFlagConfig 测试 ──────────────────────────────────────────────────
+
+
+class TestFeatureFlagConfig:
+    """FeatureFlagConfig 核心逻辑测试。"""
+
+    def test_default_disabled(self):
+        """默认配置为关闭状态。"""
+        config = FeatureFlagConfig(name="test_flag")
+        assert config.enabled is False
+        assert config.percentage == 0
+        assert config.whitelist == set()
+        assert config.is_active() is False
+        assert config.is_active("user1") is False
+
+    def test_global_enabled_100_percent(self):
+        """100% + 启用 = 全部命中。"""
+        config = FeatureFlagConfig(name="test_flag", enabled=True, percentage=100)
+        assert config.is_active() is True
+        assert config.is_active("user1") is True
+        assert config.is_active("any_user") is True
+
+    def test_global_enabled_0_percent_no_whitelist(self):
+        """启用但 0% 且无白名单 = 不命中。"""
+        config = FeatureFlagConfig(name="test_flag", enabled=True, percentage=0)
+        assert config.is_active() is False
+        assert config.is_active("user1") is False
+
+    def test_whitelist_takes_priority(self):
+        """白名单优先级高于百分比。"""
+        config = FeatureFlagConfig(
+            name="test_flag",
+            enabled=True,
+            percentage=0,
+            whitelist={"user1", "user2"},
+        )
+        assert config.is_active("user1") is True
+        assert config.is_active("user2") is True
+        assert config.is_active("user3") is False
+
+    def test_whitelist_with_percentage(self):
+        """白名单用户即使百分比为0也命中，非白名单按百分比。"""
+        config = FeatureFlagConfig(
+            name="test_flag",
+            enabled=True,
+            percentage=100,  # 100% 所有人命中
+            whitelist={"user1"},
+        )
+        assert config.is_active("user1") is True
+        assert config.is_active("user999") is True  # 100% 命中
+
+    def test_percentage_consistency_same_user(self):
+        """同一用户多次调用结果一致（哈希确定性）。"""
+        config = FeatureFlagConfig(name="test_flag", enabled=True, percentage=50)
+        results = [config.is_active("user_fixed") for _ in range(100)]
+        assert all(r == results[0] for r in results)
+
+    def test_percentage_different_users_distributed(self):
+        """不同用户分布大致符合百分比（统计检验，宽松阈值）。"""
+        config = FeatureFlagConfig(name="test_flag", enabled=True, percentage=50)
+        active_count = sum(1 for i in range(1000) if config.is_active(f"user_{i}"))
+        # 50% 上下浮动 10% 都算合理
+        assert 400 <= active_count <= 600, f"Expected ~500, got {active_count}"
+
+    def test_percentage_boundary_0_and_100(self):
+        """0% 和 100% 的边界情况。"""
+        config_0 = FeatureFlagConfig(name="test", enabled=True, percentage=0)
+        config_100 = FeatureFlagConfig(name="test", enabled=True, percentage=100)
+
+        for i in range(100):
+            assert config_0.is_active(f"user_{i}") is False
+            assert config_100.is_active(f"user_{i}") is True
+
+    def test_disabled_ignores_all_other_settings(self):
+        """关闭时忽略白名单和百分比。"""
+        config = FeatureFlagConfig(
+            name="test_flag",
+            enabled=False,
+            percentage=100,
+            whitelist={"user1"},
+        )
+        assert config.is_active("user1") is False
+        assert config.is_active() is False
+
+    def test_none_identifier_with_percentage(self):
+        """无 identifier 时按随机比例（0% 和 100% 是确定的）。"""
+        config_0 = FeatureFlagConfig(name="test", enabled=True, percentage=0)
+        config_100 = FeatureFlagConfig(name="test", enabled=True, percentage=100)
+        assert config_0.is_active(None) is False
+        assert config_100.is_active(None) is True
+
+    def test_to_dict_and_from_dict(self):
+        """序列化和反序列化对称。"""
+        original = FeatureFlagConfig(
+            name="test_flag",
+            enabled=True,
+            percentage=30,
+            whitelist={"user_a", "user_b", "user_c"},
+        )
+        data = original.to_dict()
+        restored = FeatureFlagConfig.from_dict(data)
+        assert restored.name == original.name
+        assert restored.enabled == original.enabled
+        assert restored.percentage == original.percentage
+        assert restored.whitelist == original.whitelist
+
+    def test_from_dict_with_missing_fields(self):
+        """from_dict 缺失字段时使用默认值。"""
+        config = FeatureFlagConfig.from_dict({"name": "minimal"})
+        assert config.name == "minimal"
+        assert config.enabled is False
+        assert config.percentage == 0
+        assert config.whitelist == set()
+
+
+# ── InMemoryFeatureFlagStore 测试 ───────────────────────────────────────────
+
+
+class TestInMemoryFeatureFlagStore:
+    """内存存储实现测试。"""
+
+    def test_get_nonexistent_returns_default(self):
+        """获取不存在的 flag 返回默认配置（关闭）。"""
+        store = InMemoryFeatureFlagStore()
+        config = store.get("nonexistent")
+        assert config.name == "nonexistent"
+        assert config.enabled is False
+
+    def test_set_and_get(self):
+        """设置后可以读取。"""
+        store = InMemoryFeatureFlagStore()
+        config = FeatureFlagConfig(name="test", enabled=True, percentage=50, whitelist={"u1"})
+        store.set(config)
+
+        got = store.get("test")
+        assert got.enabled is True
+        assert got.percentage == 50
+        assert got.whitelist == {"u1"}
+
+    def test_delete_existing(self):
+        """删除存在的 flag 返回 True。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(FeatureFlagConfig(name="test", enabled=True))
+        assert store.delete("test") is True
+        assert store.get("test").enabled is False
+
+    def test_delete_nonexistent(self):
+        """删除不存在的 flag 返回 False。"""
+        store = InMemoryFeatureFlagStore()
+        assert store.delete("nonexistent") is False
+
+    def test_list_all(self):
+        """列出所有 flag。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(FeatureFlagConfig(name="flag_a", enabled=True))
+        store.set(FeatureFlagConfig(name="flag_b", percentage=10))
+
+        all_flags = store.list_all()
+        assert len(all_flags) == 2
+        assert "flag_a" in all_flags
+        assert "flag_b" in all_flags
+        assert all_flags["flag_a"].enabled is True
+
+    def test_is_active_convenience(self):
+        """is_active 便捷方法。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(FeatureFlagConfig(name="render", enabled=True, percentage=0, whitelist={"vip_user"}))
+        assert store.is_active("render", "vip_user") is True
+        assert store.is_active("render", "normal_user") is False
+        assert store.is_active("nonexistent") is False
+
+
+# ── RenderEngineResolver 测试 ───────────────────────────────────────────────
+
+
+class TestRenderEngineResolver:
+    """渲染引擎选择器测试。"""
+
+    def test_default_legacy_when_flag_disabled(self):
+        """flag 关闭时使用默认引擎（legacy）。"""
+        store = InMemoryFeatureFlagStore()
+        resolver = self._make_resolver(store=store, default="legacy")
+        assert resolver.get_engine() == "legacy"
+        assert resolver.get_engine("user1") == "legacy"
+
+    def test_default_unified_when_flag_disabled(self):
+        """flag 关闭但默认值是 unified 时返回 unified。"""
+        store = InMemoryFeatureFlagStore()
+        resolver = self._make_resolver(store=store, default="unified")
+        assert resolver.get_engine() == "unified"
+
+    def test_whitelist_user_uses_unified(self):
+        """白名单用户走新引擎。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(
+            FeatureFlagConfig(
+                name="render_engine",
+                enabled=True,
+                percentage=0,
+                whitelist={"beta_tester"},
+            )
+        )
+        resolver = self._make_resolver(store=store, default="legacy")
+        assert resolver.get_engine("beta_tester") == "unified"
+        assert resolver.get_engine("normal_user") == "legacy"
+
+    def test_100_percent_all_unified(self):
+        """100% 时所有用户走新引擎。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
+        resolver = self._make_resolver(store=store, default="legacy")
+        for i in range(50):
+            assert resolver.get_engine(f"user_{i}") == "unified"
+
+    def test_invalid_default_engine_fallback(self):
+        """无效默认值回退到 legacy。"""
+        store = InMemoryFeatureFlagStore()
+        resolver = self._make_resolver(store=store, default="invalid_value")
+        assert resolver.get_engine() == "legacy"
+
+    def test_should_use_unified_helper(self):
+        """should_use_unified 便捷方法。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(
+            FeatureFlagConfig(
+                name="render_engine",
+                enabled=True,
+                percentage=0,
+                whitelist={"user_a"},
+            )
+        )
+        resolver = self._make_resolver(store=store)
+        assert resolver.should_use_unified("user_a") is True
+        assert resolver.should_use_unified("user_b") is False
+
+    def test_config_snapshot(self):
+        """配置快照。"""
+        store = InMemoryFeatureFlagStore()
+        store.set(
+            FeatureFlagConfig(
+                name="render_engine",
+                enabled=True,
+                percentage=30,
+                whitelist={"u1", "u2"},
+            )
+        )
+        resolver = self._make_resolver(store=store)
+        snapshot = resolver.get_config_snapshot()
+        assert snapshot["flag_name"] == "render_engine"
+        assert snapshot["enabled"] is True
+        assert snapshot["percentage"] == 30
+        assert snapshot["whitelist"] == ["u1", "u2"]
+
+    def test_set_flag_updates_config(self):
+        """通过 set_flag 修改后立即生效。"""
+        store = InMemoryFeatureFlagStore()
+        resolver = self._make_resolver(store=store, default="legacy")
+
+        # 初始：关闭
+        assert resolver.get_engine("user1") == "legacy"
+
+        # 开启 100%
+        resolver.set_flag(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
+        assert resolver.get_engine("user1") == "unified"
+
+        # 关闭
+        resolver.set_flag(FeatureFlagConfig(name="render_engine", enabled=False))
+        assert resolver.get_engine("user1") == "legacy"
+
+    def test_force_refresh(self):
+        """强制刷新不报错。"""
+        store = InMemoryFeatureFlagStore()
+        resolver = self._make_resolver(store=store)
+        resolver.force_refresh()  # 不抛异常即可
+
+    def test_does_not_affect_in_flight_tasks(self):
+        """
+        热更新不影响在途任务验证：
+        任务开始时确定引擎，中途配置变更不改变当前任务的引擎选择。
+        （这是通过"每次调用 get_engine 时读取当前配置"来保证的，
+        任务开始时调用一次拿到结果，之后不再变化）
+        """
+        store = InMemoryFeatureFlagStore()
+        store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
+        resolver = self._make_resolver(store=store, default="legacy")
+
+        # 模拟任务开始时获取引擎
+        engine_at_start = resolver.get_engine("user1")
+        assert engine_at_start == "unified"
+
+        # 任务进行中关闭 flag
+        store.set(FeatureFlagConfig(name="render_engine", enabled=False))
+        resolver.force_refresh()
+
+        # 在途任务持有的 engine_at_start 仍然是 unified（不随配置变化）
+        assert engine_at_start == "unified"
+        # 新任务会拿到 legacy
+        assert resolver.get_engine("user1") == "legacy"
+
+    # ── 辅助方法 ──
+
+    @staticmethod
+    def _make_resolver(store=None, default="legacy"):
+        from apps.worker.video_processing.render_engine_resolver import (
+            RenderEngineResolver,
+        )
+
+        return RenderEngineResolver(
+            default_engine=default,
+            store=store or InMemoryFeatureFlagStore(),
+            refresh_interval=9999,  # 测试时禁用自动刷新
+        )
+
+
+# ── RedisFeatureFlagStore 降级测试（无 Redis 环境） ───────────────────────
+
+
+class TestRedisStoreDegradation:
+    """Redis 不可用时的降级行为测试。"""
+
+    def test_get_returns_default_when_redis_unavailable(self):
+        """Redis 连接失败时返回默认关闭配置，不抛异常。"""
+        import importlib
+
+        from packages.adapters.redis import feature_flag_store as ff_module
+
+        # 模拟 redis 模块不存在的场景不好做，这里直接测试异常捕获逻辑
+        store = ff_module.RedisFeatureFlagStore.__new__(ff_module.RedisFeatureFlagStore)
+        store._redis = MagicMock()
+        store._redis.hgetall.side_effect = ConnectionError("Redis down")
+        store._key_prefix = ff_module.FEATURE_FLAG_REDIS_PREFIX
+        store._cache = {}
+        store._cache_ttl = 5.0
+        import threading
+
+        store._lock = threading.Lock()
+
+        config = store.get("render_engine")
+        assert config.enabled is False
+        assert config.name == "render_engine"
+
+    def test_list_all_returns_empty_on_redis_error(self):
+        """Redis 错误时 list_all 返回空字典。"""
+        import importlib
+
+        from packages.adapters.redis import feature_flag_store as ff_module
+
+        store = ff_module.RedisFeatureFlagStore.__new__(ff_module.RedisFeatureFlagStore)
+        store._redis = MagicMock()
+        store._redis.scan.side_effect = ConnectionError("Redis down")
+        store._key_prefix = ff_module.FEATURE_FLAG_REDIS_PREFIX
+        store._cache = {}
+        store._cache_ttl = 5.0
+        import threading
+
+        store._lock = threading.Lock()
+
+        result = store.list_all()
+        assert result == {}
+
+
+class TestRedisStoreListAll:
+    """RedisFeatureFlagStore list_all 正常路径测试。"""
+
+    def _make_store(self):
+        from packages.adapters.redis import feature_flag_store as ff_module
+
+        store = ff_module.RedisFeatureFlagStore.__new__(ff_module.RedisFeatureFlagStore)
+        store._redis = MagicMock()
+        store._key_prefix = ff_module.FEATURE_FLAG_REDIS_PREFIX
+        store._cache = {}
+        store._cache_ttl = 5.0
+        import threading
+
+        store._lock = threading.Lock()
+        return store
+
+    def test_list_all_scan_with_match_param(self):
+        """list_all 调用 redis.scan 时使用正确的 match 参数名。"""
+        store = self._make_store()
+        prefix = store._key_prefix
+
+        # 模拟 scan 返回 2 个 key，分 2 次游标
+        store._redis.scan.side_effect = [
+            (10, [f"{prefix}render_engine", f"{prefix}other_flag"]),
+            (0, []),
+        ]
+        # 模拟 hgetall 返回配置
+        store._redis.hgetall.return_value = {
+            b"enabled": b"true",
+            b"percentage": b"50",
+            b"whitelist": b'["user1","user2"]',
+        }
+
+        result = store.list_all()
+
+        # 验证 scan 被调用了 2 次（游标遍历）
+        assert store._redis.scan.call_count == 2
+        # 验证参数名是 match（不是 match_pattern）
+        first_call_kwargs = store._redis.scan.call_args_list[0][1]
+        assert "match" in first_call_kwargs
+        assert "match_pattern" not in first_call_kwargs
+        assert first_call_kwargs["match"] == f"{prefix}*"
+        # 验证返回了 2 个 flag
+        assert len(result) == 2
+        assert "render_engine" in result
+        assert "other_flag" in result

tests/unit/test_oneclick_gen_p0_fixes.py

@@ -132,10 +132,8 @@ class TestDownloadLibraryAssets:
         session.query.return_value = query
         filter_result = MagicMock()
         query.filter.return_value = filter_result
-        in_filter = MagicMock()
-        filter_result.filter.return_value = in_filter
         id_filter = MagicMock()
-        in_filter.filter.return_value = id_filter
+        filter_result.filter.return_value = id_filter
         assets = [self._make_asset("a1", "video/a1.mp4")]
         id_filter.order_by.return_value.all.return_value = assets
 

tests/unit/test_render_adapter.py

@@ -0,0 +1,424 @@
+"""RenderAdapter 单元测试 — Phase 2.
+
+测试适配层的计划加载、素材下载、引擎调用、结果上传等逻辑。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+from video_processing.render_adapter import RenderAdapter, RenderAdapterResult
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class FakeClip:
+    """模拟 EditPlanClip。"""
+
+    id: str
+    plan_id: str = "plan_001"
+    clip_type: str = "main"
+    order: int = 0
+    asset_id: str = ""
+    text_content: str = ""
+    start_time: float = 0.0
+    duration: float = 0.0
+    transition_effect: str = "cut"
+    status: str = "ready"
+    config: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class FakePlan:
+    """模拟 EditPlan。"""
+
+    id: str = "plan_001"
+    name: str = "测试计划"
+    status: str = "editing"
+    config: dict[str, Any] = field(default_factory=dict)
+
+
+def _make_clip(
+    clip_id: str,
+    clip_type: str = "main",
+    order: int = 0,
+    asset_id: str | None = None,
+    duration: float = 5.0,
+    status: str = "ready",
+    transition_effect: str = "cut",
+    config: dict[str, Any] | None = None,
+) -> FakeClip:
+    # asset_id 为 None 时生成默认值，为空字符串时保留空串
+    if asset_id is None:
+        asset_id = f"asset_{clip_id}.mp4"
+    return FakeClip(
+        id=clip_id,
+        clip_type=clip_type,
+        order=order,
+        asset_id=asset_id,
+        duration=duration,
+        status=status,
+        transition_effect=transition_effect,
+        config=config or {},
+    )
+
+
+def _make_adapter(
+    plan: FakePlan | None = None,
+    clips: list[FakeClip] | None = None,
+) -> tuple[RenderAdapter, MagicMock, MagicMock]:
+    """创建测试用的 RenderAdapter 及 mock repo。
+
+    Returns:
+        (adapter, mock_plan_repo, mock_clip_repo)
+    """
+    mock_db = MagicMock()
+    adapter = RenderAdapter(mock_db)
+
+    # 替换内部 repo
+    mock_plan_repo = MagicMock()
+    mock_clip_repo = MagicMock()
+    adapter._plan_repo = mock_plan_repo
+    adapter._clip_repo = mock_clip_repo
+
+    # 设置默认返回
+    if plan is not None:
+        mock_plan_repo.get.return_value = plan
+    if clips is not None:
+        mock_clip_repo.list_by_plan.return_value = clips
+
+    return adapter, mock_plan_repo, mock_clip_repo
+
+
+# ── validate_plan 测试 ───────────────────────────────────────────────────────
+
+
+class TestValidatePlan:
+    def test_plan_not_found(self):
+        """计划不存在时校验失败。"""
+        adapter, mock_plan_repo, _ = _make_adapter(plan=None)
+        mock_plan_repo.get.return_value = None
+
+        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")
+
+        assert not valid
+        assert len(errors) == 1
+        assert "不存在" in errors[0]
+        assert ready_count == 0
+        assert total_count == 0
+
+    def test_no_clips(self):
+        """没有任何片段时校验失败。"""
+        plan = FakePlan(id="plan_001", status="editing")
+        adapter, _, mock_clip_repo = _make_adapter(plan=plan, clips=[])
+
+        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")
+
+        assert not valid
+        assert any("没有任何片段" in e for e in errors)
+
+    def test_no_ready_clips(self):
+        """没有 ready 片段时校验失败。"""
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [
+            _make_clip("c1", status="pending"),
+            _make_clip("c2", status="pending"),
+        ]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")
+
+        assert not valid
+        assert any("没有就绪" in e for e in errors)
+        assert ready_count == 0
+        assert total_count == 2
+
+    def test_ready_clip_no_asset(self):
+        """ready 片段没有 asset_id 时报错。"""
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [
+            _make_clip("c1", asset_id=""),
+        ]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")
+
+        assert not valid
+        assert any("没有分配素材" in e for e in errors)
+
+    def test_valid_plan(self):
+        """正常计划校验通过。"""
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [
+            _make_clip("c1", order=0, duration=3.0),
+            _make_clip("c2", order=1, duration=4.0),
+        ]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")
+
+        assert valid
+        assert len(errors) == 0
+        assert ready_count == 2
+        assert total_count == 2
+
+    def test_wrong_status(self):
+        """计划状态不正确时报错。"""
+        plan = FakePlan(id="plan_001", status="draft")
+        clips = [_make_clip("c1")]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        valid, errors, _, _, _ = adapter.validate_plan("plan_001")
+
+        assert not valid
+        assert any("状态不正确" in e for e in errors)
+
+    def test_mixed_status_with_warnings(self):
+        """混合状态时有 pending/failed 警告。"""
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [
+            _make_clip("c1", order=0, status="ready"),
+            _make_clip("c2", order=1, status="pending"),
+            _make_clip("c3", order=2, status="failed"),
+        ]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")
+
+        assert valid
+        assert any("pending" in w for w in warnings)
+        assert any("failed" in w for w in warnings)
+        assert ready_count == 1
+        assert total_count == 3
+
+
+# ── render_plan 测试 ─────────────────────────────────────────────────────────
+
+
+class TestRenderPlan:
+    def test_plan_not_found(self):
+        """计划不存在时返回失败。"""
+        adapter, mock_plan_repo, _ = _make_adapter(plan=None)
+        mock_plan_repo.get.return_value = None
+
+        result = adapter.render_plan("plan_001")
+
+        assert not result.success
+        assert "不存在" in result.error_message
+
+    def test_no_ready_clips(self):
+        """没有 ready 片段时返回失败。"""
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [_make_clip("c1", status="pending")]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        result = adapter.render_plan("plan_001")
+
+        assert not result.success
+        assert "没有可渲染" in result.error_message
+        assert result.clip_count == 0
+
+    @patch("video_processing.render_adapter.download_asset")
+    def test_all_assets_download_fail(self, mock_download):
+        """所有素材下载失败时返回失败。"""
+        mock_download.return_value = False
+
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [_make_clip("c1", order=0, duration=5.0)]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        result = adapter.render_plan("plan_001")
+
+        assert not result.success
+        assert "素材下载失败" in result.error_message
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    @patch("video_processing.render_adapter.download_asset")
+    def test_successful_render(self, mock_download, mock_render_cls, mock_upload, tmp_path):
+        """完整渲染流程成功。"""
+
+        # 素材下载成功
+        def _fake_download(asset_id, local_path):
+            local_path.parent.mkdir(parents=True, exist_ok=True)
+            local_path.write_bytes(b"fake video data")
+            return True
+
+        mock_download.side_effect = _fake_download
+
+        # 渲染成功
+        mock_render = MagicMock()
+        mock_render.render.return_value = MagicMock(
+            output_path=tmp_path / "output.mp4",
+            duration=10.0,
+            file_size=102400,
+            width=1280,
+            height=720,
+        )
+        mock_render_cls.return_value = mock_render
+
+        # 上传成功
+        mock_upload.return_value = "https://oss.example.com/rendered/plan_001/job_001.mp4"
+
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [
+            _make_clip("c1", order=0, duration=5.0),
+            _make_clip("c2", order=1, duration=5.0),
+        ]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        result = adapter.render_plan(
+            "plan_001",
+            job_id="job_001",
+            work_dir=tmp_path / "work",
+        )
+
+        assert result.success
+        assert result.output_url.startswith("https://")
+        assert result.duration == 10.0
+        assert result.width == 1280
+        assert result.height == 720
+        assert result.clip_count == 2
+
+        # 验证 UnifiedRenderService 被正确调用
+        mock_render_cls.assert_called_once()
+        call_kwargs = mock_render_cls.call_args
+        assert call_kwargs.kwargs["plan"] is plan
+        assert len(call_kwargs.kwargs["clips"]) == 2
+        assert len(call_kwargs.kwargs["asset_path_map"]) == 2
+
+    @patch("video_processing.render_adapter.download_asset")
+    def test_progress_callback(self, mock_download, tmp_path):
+        """进度回调被正确触发。"""
+
+        def _fake_download(asset_id, local_path):
+            local_path.parent.mkdir(parents=True, exist_ok=True)
+            local_path.write_bytes(b"fake data")
+            return True
+
+        mock_download.side_effect = _fake_download
+
+        # 模拟渲染异常，避免走到最后
+        with patch("video_processing.render_adapter.UnifiedRenderService") as mock_render_cls:
+            mock_render = MagicMock()
+            mock_render.render.side_effect = RuntimeError("render error")
+            mock_render_cls.return_value = mock_render
+
+            plan = FakePlan(id="plan_001", status="editing")
+            clips = [_make_clip("c1", order=0, duration=5.0)]
+            adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+            progress_values = []
+
+            def progress_cb(progress: float, stage: str) -> None:
+                progress_values.append((progress, stage))
+
+            result = adapter.render_plan(
+                "plan_001",
+                work_dir=tmp_path / "work",
+                progress_cb=progress_cb,
+            )
+
+            # 即使渲染失败，前期进度也应该上报了
+            assert len(progress_values) > 0
+            # 第一个进度应该是加载计划
+            assert progress_values[0][1] == "加载剪辑计划"
+
+    @patch("video_processing.render_adapter.download_asset")
+    def test_partial_asset_download(self, mock_download, tmp_path):
+        """部分素材下载失败时，只使用成功的素材。"""
+        download_results = [True, False, True]  # 3个素材中2个成功
+
+        def _fake_download(asset_id, local_path):
+            idx = hash(asset_id) % 3
+            if download_results[idx]:
+                local_path.parent.mkdir(parents=True, exist_ok=True)
+                local_path.write_bytes(b"fake data")
+                return True
+            return False
+
+        mock_download.side_effect = _fake_download
+
+        with patch("video_processing.render_adapter.UnifiedRenderService") as mock_render_cls:
+            mock_render = MagicMock()
+            mock_render.render.return_value = MagicMock(
+                output_path=tmp_path / "out.mp4",
+                duration=5.0,
+                file_size=1024,
+                width=1280,
+                height=720,
+            )
+            mock_render_cls.return_value = mock_render
+
+            with patch("video_processing.render_adapter.upload_to_oss", return_value="https://example.com/out.mp4"):
+                plan = FakePlan(id="plan_001", status="editing")
+                clips = [
+                    _make_clip("c1", order=0, duration=3.0, asset_id="asset_001.mp4"),
+                    _make_clip("c2", order=1, duration=3.0, asset_id="asset_002.mp4"),
+                    _make_clip("c3", order=2, duration=3.0, asset_id="asset_003.mp4"),
+                ]
+                adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+                result = adapter.render_plan(
+                    "plan_001",
+                    work_dir=tmp_path / "work",
+                )
+
+                # 至少有部分素材成功，渲染应该进行
+                # （具体成功数量取决于 hash 结果，但至少1个成功就能渲染）
+                assert result.success or "素材下载失败" in result.error_message
+
+
+# ── _download_assets 测试 ────────────────────────────────────────────────────
+
+
+class TestDownloadAssets:
+    @patch("video_processing.render_adapter.download_asset")
+    def test_all_download_success(self, mock_download, tmp_path):
+        """全部素材下载成功。"""
+        mock_download.return_value = True
+
+        clips = [
+            _make_clip("c1", order=0, asset_id="key1.mp4"),
+            _make_clip("c2", order=1, asset_id="key2.mp4"),
+        ]
+
+        result = RenderAdapter._download_assets(clips, tmp_path)
+
+        assert len(result) == 2
+        assert "key1.mp4" in result
+        assert "key2.mp4" in result
+        assert mock_download.call_count == 2
+
+    @patch("video_processing.render_adapter.download_asset")
+    def test_empty_asset_id_skipped(self, mock_download, tmp_path):
+        """空 asset_id 的片段被跳过。"""
+        clips = [
+            _make_clip("c1", order=0, asset_id=""),
+            _make_clip("c2", order=1, asset_id="key2.mp4"),
+        ]
+        mock_download.return_value = True
+
+        result = RenderAdapter._download_assets(clips, tmp_path)
+
+        assert len(result) == 1
+        assert "key2.mp4" in result
+        assert mock_download.call_count == 1  # 只调用了一次下载
+
+    @patch("video_processing.render_adapter.download_asset")
+    def test_all_download_fail(self, mock_download, tmp_path):
+        """全部下载失败返回空字典。"""
+        mock_download.return_value = False
+
+        clips = [
+            _make_clip("c1", order=0, asset_id="key1.mp4"),
+        ]
+
+        result = RenderAdapter._download_assets(clips, tmp_path)
+
+        assert len(result) == 0

tests/unit/test_unified_render_service.py

@@ -16,7 +16,10 @@ from video_processing.unified_render_service import (
     RenderResult,
     ResolvedClip,
     UnifiedRenderService,
+    _hex_to_ass_color,
+    _position_to_ass_alignment,
     _resolve_layer_role,
+    generate_ass_subtitles,
 )
 
 # ── Fixtures ──────────────────────────────────────────────────────────────────
@@ -45,6 +48,7 @@ class FakePlan:
 
     id: str = "plan_001"
     name: str = "测试计划"
+    config: dict[str, Any] = field(default_factory=dict)
 
 
 def _make_clip(
@@ -438,6 +442,56 @@ class TestBuildFilterComplex:
             f"单视频: setpts(position={last_setpts}) 应该在 fps(position={first_fps}) 之前。" f"滤镜链: {chain_str}"
         )
 
+    def test_main_clip_uses_fill_crop_strategy(self):
+        """main/broll clip 使用铺满裁剪策略（scale increase + crop），不是等比+黑边。
+
+        对齐链路A编辑器合成行为，与主流短视频平台一致。
+        """
+        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            fc, _ = svc._build_filter_complex(layers)
+
+        # 验证：scale 使用 force_original_aspect_ratio=increase（铺满）
+        assert "force_original_aspect_ratio=increase" in fc
+        # 验证：有 crop（居中裁剪）
+        assert "crop=1280:720" in fc
+        # 验证：没有 pad（不是黑边模式）
+        assert "pad=" not in fc
+
+    def test_broll_clip_uses_fill_crop_strategy(self):
+        """broll clip 同样使用铺满裁剪策略。"""
+        clips = [_make_clip("c1", "b_roll", order=0, duration=5.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            fc, _ = svc._build_filter_complex(layers)
+
+        assert "force_original_aspect_ratio=increase" in fc
+        assert "crop=1280:720" in fc
+        assert "pad=" not in fc
+
+    def test_background_uses_fill_crop_strategy(self):
+        """background 层也使用铺满裁剪（已有的行为，保持一致）。"""
+        clips = [_make_clip("c1", "background", order=0, duration=5.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            fc, _ = svc._build_filter_complex(layers)
+
+        assert "force_original_aspect_ratio=increase" in fc
+        assert "crop=1280:720" in fc
+
     def test_empty_layers_raises(self):
         """空图层列表抛出 ValueError。"""
         svc = _make_service()
@@ -445,6 +499,298 @@ class TestBuildFilterComplex:
             svc._build_filter_complex([])
 
 
+class TestPassThrough:
+    """测试单图层单 clip 直通优化路径。"""
+
+    def test_can_use_pass_through_single_main_clip(self):
+        """1个main图层 + 1个clip → 可以直通。"""
+        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)
+            assert svc._can_use_pass_through(layers) is True
+
+    def test_can_use_pass_through_single_broll_clip(self):
+        """1个broll图层 + 1个clip → 可以直通。"""
+        clips = [_make_clip("c1", "b_roll", 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)
+            assert svc._can_use_pass_through(layers) is True
+
+    def test_can_use_pass_through_single_background_clip(self):
+        """1个background图层 + 1个clip → 可以直通。"""
+        clips = [_make_clip("c1", "background", 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)
+            assert svc._can_use_pass_through(layers) is True
+
+    def test_cannot_pass_through_multi_clips(self):
+        """1个图层 + 多个clips → 不能直通（需要xfade）。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=3.0),
+            _make_clip("c2", "main", order=1, duration=3.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)
+            assert svc._can_use_pass_through(layers) is False
+
+    def test_cannot_pass_through_multi_layers(self):
+        """多个图层 → 不能直通。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=5.0),
+            _make_clip("c2", "overlay", order=1, 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)
+            assert svc._can_use_pass_through(layers) is False
+
+    def test_cannot_pass_through_overlay_only(self):
+        """只有overlay图层 → 不能直通（需要叠加到主层）。"""
+        clips = [_make_clip("c1", "overlay", 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)
+            assert svc._can_use_pass_through(layers) is False
+
+    def test_render_uses_pass_through_for_single_clip(self):
+        """单clip渲染时走直通路径（调用_render_pass_through而非_execute_ffmpeg）。"""
+        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch.object(svc, "_render_pass_through") as mock_pass,
+            patch.object(svc, "_execute_ffmpeg") as mock_exec,
+            patch.object(svc, "_mix_audio", return_value=None),
+            patch("shutil.copy2"),
+            patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)),
+        ):
+            result = svc.render()
+
+        mock_pass.assert_called_once()
+        mock_exec.assert_not_called()
+        assert result.duration == 5.0
+
+    def test_render_uses_filter_complex_for_multi_clips(self):
+        """多clip渲染时走完整filter_complex路径。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=3.0),
+            _make_clip("c2", "main", order=1, duration=3.0),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch.object(svc, "_render_pass_through") as mock_pass,
+            patch.object(svc, "_execute_ffmpeg") as mock_exec,
+            patch.object(svc, "_mix_audio", return_value=None),
+            patch("shutil.copy2"),
+            patch.object(svc, "_probe_output", return_value=(5.5, 2048, 1280, 720)),
+        ):
+            result = svc.render()
+
+        mock_pass.assert_not_called()
+        mock_exec.assert_called_once()
+        assert result.duration == 5.5
+
+
+# ── 测试 ASS 字幕生成 ────────────────────────────────────────────────────────
+
+
+class TestAssSubtitles:
+    """测试 ASS 字幕生成功能。"""
+
+    def test_hex_to_ass_color_white(self):
+        """#ffffff → &HFFFFFF（ASS 是 BGR 顺序）。"""
+        assert _hex_to_ass_color("#ffffff") == "&HFFFFFF"
+
+    def test_hex_to_ass_color_black(self):
+        """#000000 → &H000000。"""
+        assert _hex_to_ass_color("#000000") == "&H000000"
+
+    def test_hex_to_ass_color_red(self):
+        """#ff0000 红 → &H0000FF（B=00, G=00, R=FF）。"""
+        assert _hex_to_ass_color("#ff0000") == "&H0000FF"
+
+    def test_hex_to_ass_color_blue(self):
+        """#0000ff 蓝 → &HFF0000（B=FF, G=00, R=00）。"""
+        assert _hex_to_ass_color("#0000ff") == "&HFF0000"
+
+    def test_hex_to_ass_color_no_hash(self):
+        """不带 # 的颜色值也能解析。"""
+        assert _hex_to_ass_color("ff0000") == "&H0000FF"
+
+    def test_position_to_ass_alignment_top(self):
+        """top → 8（顶部居中）。"""
+        assert _position_to_ass_alignment("top") == 8
+
+    def test_position_to_ass_alignment_center(self):
+        """center → 5（居中）。"""
+        assert _position_to_ass_alignment("center") == 5
+
+    def test_position_to_ass_alignment_bottom(self):
+        """bottom → 2（底部居中）。"""
+        assert _position_to_ass_alignment("bottom") == 2
+
+    def test_generate_ass_with_title_only(self, tmp_path):
+        """只有标题时生成 ASS 文件。"""
+        ass_path = tmp_path / "test.ass"
+        result = generate_ass_subtitles(
+            ass_path,
+            video_width=1280,
+            video_height=720,
+            video_duration=10.0,
+            title_text="测试标题",
+            title_config={
+                "enabled": True,
+                "font": "思源黑体",
+                "size": 48,
+                "color": "#ffffff",
+                "bold": True,
+                "position": "top",
+                "stroke": {"enabled": True, "color": "#000000", "width": 2},
+                "shadow": {"enabled": True, "blur": 4, "offset_x": 2, "offset_y": 2},
+            },
+        )
+
+        assert result == ass_path
+        content = ass_path.read_text(encoding="utf-8")
+        assert "[Script Info]" in content
+        assert "PlayResX: 1280" in content
+        assert "PlayResY: 720" in content
+        assert "[V4+ Styles]" in content
+        assert "TitleStyle" in content
+        assert "测试标题" in content
+        assert "Dialogue:" in content
+
+    def test_generate_ass_with_subtitle_only(self, tmp_path):
+        """只有字幕时生成 ASS 文件。"""
+        ass_path = tmp_path / "test.ass"
+        result = generate_ass_subtitles(
+            ass_path,
+            video_width=1280,
+            video_height=720,
+            video_duration=10.0,
+            subtitle_text="测试字幕内容",
+            subtitle_config={
+                "enabled": True,
+                "font": "思源黑体",
+                "size": 24,
+                "color": "#ffffff",
+                "position": "bottom",
+            },
+        )
+
+        content = ass_path.read_text(encoding="utf-8")
+        assert "SubtitleStyle" in content
+        assert "测试字幕内容" in content
+        assert "Dialogue:" in content
+
+    def test_generate_ass_with_both_title_and_subtitle(self, tmp_path):
+        """同时有标题和字幕。"""
+        ass_path = tmp_path / "test.ass"
+        generate_ass_subtitles(
+            ass_path,
+            video_width=1280,
+            video_height=720,
+            video_duration=10.0,
+            title_text="大标题",
+            title_config={"enabled": True, "position": "top"},
+            subtitle_text="底部字幕",
+            subtitle_config={"enabled": True, "position": "bottom"},
+        )
+
+        content = ass_path.read_text(encoding="utf-8")
+        assert "TitleStyle" in content
+        assert "SubtitleStyle" in content
+        assert "大标题" in content
+        assert "底部字幕" in content
+        # 两条 Dialogue 行
+        assert content.count("Dialogue:") == 2
+
+    def test_generate_ass_disabled_returns_empty(self, tmp_path):
+        """标题和字幕都禁用时返回空文件。"""
+        ass_path = tmp_path / "test.ass"
+        generate_ass_subtitles(
+            ass_path,
+            video_width=1280,
+            video_height=720,
+            video_duration=10.0,
+            title_text="不显示",
+            title_config={"enabled": False},
+            subtitle_text="也不显示",
+            subtitle_config={"enabled": False},
+        )
+
+        content = ass_path.read_text(encoding="utf-8")
+        assert content == ""
+
+    def test_generate_ass_empty_text_returns_empty(self, tmp_path):
+        """文本为空时不生成字幕。"""
+        ass_path = tmp_path / "test.ass"
+        generate_ass_subtitles(
+            ass_path,
+            video_width=1280,
+            video_height=720,
+            video_duration=10.0,
+            title_text="",
+            title_config={"enabled": True},
+            subtitle_text="   ",
+            subtitle_config={"enabled": True},
+        )
+
+        content = ass_path.read_text(encoding="utf-8")
+        assert content == ""
+
+    def test_generate_ass_time_format(self, tmp_path):
+        """验证 ASS 时间格式正确（H:MM:SS.cc）。"""
+        ass_path = tmp_path / "test.ass"
+        generate_ass_subtitles(
+            ass_path,
+            video_width=1280,
+            video_height=720,
+            video_duration=125.5,  # 2分5.5秒
+            title_text="测试",
+            title_config={"enabled": True},
+        )
+
+        content = ass_path.read_text(encoding="utf-8")
+        # 结束时间应该是 0:02:05.50
+        assert "0:02:05.50" in content
+
+    def test_ass_text_escape_newlines(self, tmp_path):
+        """换行符转义为 ASS 的 \\N。"""
+        ass_path = tmp_path / "test.ass"
+        generate_ass_subtitles(
+            ass_path,
+            video_width=1280,
+            video_height=720,
+            video_duration=10.0,
+            title_text="第一行\n第二行",
+            title_config={"enabled": True},
+        )
+
+        content = ass_path.read_text(encoding="utf-8")
+        assert "第一行\\N第二行" in content
+
+
 # ── 测试 render 方法 ─────────────────────────────────────────────────────────
 
 
@@ -464,8 +810,8 @@ class TestRender:
         with pytest.raises(ValueError, match="没有可渲染的片段"):
             svc.render()
 
-    def test_render_success(self):
-        """正常渲染流程。"""
+    def test_render_success_single_clip(self):
+        """单clip正常渲染（走直通路径）。"""
         clips = [_make_clip("c1", "main", order=0)]
         asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
         svc = _make_service(clips, asset_paths)
@@ -473,7 +819,9 @@ class TestRender:
         with (
             _patch_path_exists(),
             patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
-            patch.object(svc, "_execute_ffmpeg") as mock_exec,
+            patch.object(svc, "_render_pass_through") as mock_pass,
+            patch.object(svc, "_mix_audio", return_value=None),
+            patch("shutil.copy2"),
             patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)),
         ):
             result = svc.render()
@@ -483,4 +831,571 @@ class TestRender:
         assert result.file_size == 1024
         assert result.width == 1280
         assert result.height == 720
+        mock_pass.assert_called_once()
+
+    def test_render_success_multi_clips(self):
+        """多clip正常渲染（走完整filter_complex路径）。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=3.0),
+            _make_clip("c2", "main", order=1, duration=3.0),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch.object(svc, "_execute_ffmpeg") as mock_exec,
+            patch.object(svc, "_mix_audio", return_value=None),
+            patch("shutil.copy2"),
+            patch.object(svc, "_probe_output", return_value=(5.5, 2048, 1280, 720)),
+        ):
+            result = svc.render()
+
+        assert isinstance(result, RenderResult)
+        assert result.duration == 5.5
+        assert result.file_size == 2048
+        assert result.width == 1280
+        assert result.height == 720
         mock_exec.assert_called_once()
+
+
+# ── 测试音频后处理 ──────────────────────────────────────────────────────────
+
+
+class TestAudioMixing:
+    """测试音频后处理混音功能。"""
+
+    def test_clip_effective_duration_with_both(self):
+        """指定时长和实际时长都有时取较小值。"""
+        clip = ResolvedClip(
+            clip_id="c1",
+            asset_id="a1",
+            local_path=Path("/tmp/c1.mp4"),
+            clip_type="main",
+            order=0,
+            duration=3.0,
+            actual_duration=5.0,
+        )
+        assert UnifiedRenderService._clip_effective_duration(clip) == 3.0
+
+    def test_clip_effective_duration_only_actual(self):
+        """只有实际时长时用实际时长。"""
+        clip = ResolvedClip(
+            clip_id="c1",
+            asset_id="a1",
+            local_path=Path("/tmp/c1.mp4"),
+            clip_type="main",
+            order=0,
+            duration=0.0,
+            actual_duration=5.0,
+        )
+        assert UnifiedRenderService._clip_effective_duration(clip) == 5.0
+
+    def test_clip_effective_duration_only_specified(self):
+        """只有指定时长时用指定时长。"""
+        clip = ResolvedClip(
+            clip_id="c1",
+            asset_id="a1",
+            local_path=Path("/tmp/c1.mp4"),
+            clip_type="main",
+            order=0,
+            duration=3.0,
+            actual_duration=0.0,
+        )
+        assert UnifiedRenderService._clip_effective_duration(clip) == 3.0
+
+    def test_clip_effective_duration_zero(self):
+        """都没有时返回0。"""
+        clip = ResolvedClip(
+            clip_id="c1",
+            asset_id="a1",
+            local_path=Path("/tmp/c1.mp4"),
+            clip_type="main",
+            order=0,
+            duration=0.0,
+            actual_duration=0.0,
+        )
+        assert UnifiedRenderService._clip_effective_duration(clip) == 0.0
+
+    def test_mix_audio_single_main_clip(self):
+        """单主clip时直接提取音频。"""
+        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            result = svc._mix_audio(layers, 5.0)
+
+        assert result is not None
+        assert result.name == "audio_plan_001.aac"
+        mock_run.assert_called_once()
+        # 验证命令包含 -vn（无视频）和 aac 编码
+        cmd = mock_run.call_args[0][0]
+        assert "-vn" in cmd
+        assert "aac" in cmd
+
+    def test_mix_audio_multi_main_clips(self):
+        """多主clip时用concat拼接音频。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=3.0),
+            _make_clip("c2", "main", order=1, duration=2.0),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            result = svc._mix_audio(layers, 4.5)
+
+        assert result is not None
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        # 验证有 filter_complex 和 concat
+        assert "-filter_complex" in cmd
+        cmd_str = " ".join(cmd)
+        assert "concat=n=2:v=0:a=1" in cmd_str
+
+    def test_mix_audio_with_independent_audio_track(self):
+        """有独立音频轨时用amix混音。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=5.0),
+            _make_clip(
+                "bgm1",
+                "main",
+                order=0,
+                duration=5.0,
+                config={"role": "audio", "volume": 0.5},
+            ),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_bgm1.mp4": Path("/tmp/asset_bgm1.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            result = svc._mix_audio(layers, 5.0)
+
+        assert result is not None
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        cmd_str = " ".join(cmd)
+        assert "amix" in cmd_str
+        assert "volume=0.5" in cmd_str
+
+    def test_mix_audio_no_audio_returns_none(self):
+        """没有音频素材时返回None。"""
+        # 构造一个没有音频的场景（比如纯文字）
+        clips = [_make_clip("t1", "title", order=0, duration=3.0)]
+        clips[0].asset_id = ""  # 无素材
+        asset_paths: dict[str, Path] = {}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=3.0),
+        ):
+            # 没有素材的clip会被跳过，layers为空
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            result = svc._mix_audio(layers, 3.0)
+
+        assert result is None
+
+    def test_mix_audio_background_not_used_as_main(self):
+        """background 图层不参与主音频，main 优先级更高。"""
+        clips = [
+            _make_clip("bg1", "background", order=0, duration=5.0),
+            _make_clip("c1", "main", order=0, duration=5.0),
+        ]
+        asset_paths = {
+            "asset_bg1.mp4": Path("/tmp/asset_bg1.mp4"),
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            result = svc._mix_audio(layers, 5.0)
+
+        assert result is not None
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        # 验证主音频源是 main 的 c1，不是 background 的 bg1
+        # 单 main clip 走直接提取路径，输入文件应该只有 c1
+        cmd_str = " ".join(cmd)
+        assert "asset_c1.mp4" in cmd_str
+        assert "asset_bg1.mp4" not in cmd_str
+
+    def test_mix_audio_main_priority_over_broll(self):
+        """main 图层优先级高于 broll。"""
+        clips = [
+            _make_clip("b1", "b_roll", order=0, duration=5.0),
+            _make_clip("c1", "main", order=0, duration=5.0),
+        ]
+        asset_paths = {
+            "asset_b1.mp4": Path("/tmp/asset_b1.mp4"),
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            result = svc._mix_audio(layers, 5.0)
+
+        assert result is not None
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        cmd_str = " ".join(cmd)
+        # 主音频源应该是 main 的 c1，不是 broll 的 b1
+        assert "asset_c1.mp4" in cmd_str
+        assert "asset_b1.mp4" not in cmd_str
+
+    def test_mix_audio_broll_used_when_no_main(self):
+        """没有 main 时，broll 作为主音频源。"""
+        clips = [_make_clip("b1", "b_roll", order=0, duration=5.0)]
+        asset_paths = {"asset_b1.mp4": Path("/tmp/asset_b1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            result = svc._mix_audio(layers, 5.0)
+
+        assert result is not None
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        assert "-vn" in cmd
+        assert "asset_b1.mp4" in " ".join(cmd)
+
+    def test_mix_audio_single_clip_truncated_to_video_duration(self):
+        """单clip音频截断到 video_duration（video_duration < clip有效时长）。"""
+        clips = [_make_clip("c1", "main", order=0, duration=10.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=10.0),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            # video_duration 只有 3.0，小于 clip 的 10.0
+            result = svc._mix_audio(layers, 3.0)
+
+        assert result is not None
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        # 验证 -t 参数是 3.0 不是 10.0
+        t_index = cmd.index("-t")
+        assert t_index >= 0
+        t_value = float(cmd[t_index + 1])
+        assert t_value == 3.0
+
+    def test_merge_audio_video(self):
+        """合并音视频命令正确。"""
+        svc = _make_service([], {})
+        video_path = Path("/tmp/video.mp4")
+        audio_path = Path("/tmp/audio.aac")
+        output_path = Path("/tmp/output.mp4")
+
+        with patch("video_processing.unified_render_service.run_ffmpeg") as mock_run:
+            svc._merge_audio_video(video_path, audio_path, output_path)
+
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        assert "-c:v" in cmd
+        assert "copy" in cmd
+        assert "-map" in cmd
+        assert "-shortest" in cmd
+
+    def test_render_calls_audio_mixing(self):
+        """多clip完整render流程会调用音频混音（非直通路径）。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=3.0),
+            _make_clip("c2", "main", order=1, duration=2.0),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch.object(svc, "_execute_ffmpeg"),
+            patch.object(svc, "_mix_audio", return_value=Path("/tmp/audio.aac")) as mock_mix,
+            patch.object(svc, "_merge_audio_video") as mock_merge,
+            patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)),
+        ):
+            result = svc.render()
+
+        mock_mix.assert_called_once()
+        mock_merge.assert_called_once()
+        assert result.duration == 5.0
+
+    def test_render_without_audio_copies_video(self):
+        """多clip无音频时走copy路径（非直通路径）。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=3.0),
+            _make_clip("c2", "main", order=1, duration=2.0),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch.object(svc, "_execute_ffmpeg"),
+            patch.object(svc, "_mix_audio", return_value=None),
+            patch("shutil.copy2") as mock_copy,
+            patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)),
+        ):
+            result = svc.render()
+
+        mock_copy.assert_called_once()
+        assert result.duration == 5.0
+
+    def test_render_pass_through_skips_audio_mix(self):
+        """直通场景下视频+音频一次完成，跳过 _mix_audio 和 _merge_audio_video。"""
+        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch.object(svc, "_render_pass_through", return_value=True) as mock_pt,
+            patch.object(svc, "_mix_audio") as mock_mix,
+            patch.object(svc, "_merge_audio_video") as mock_merge,
+            patch("shutil.copy2") as mock_copy,
+            patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)),
+        ):
+            result = svc.render()
+
+        # 直通场景调用了 _render_pass_through，跳过了 _mix_audio / _merge / copy
+        mock_pt.assert_called_once()
+        mock_mix.assert_not_called()
+        mock_merge.assert_not_called()
+        mock_copy.assert_not_called()
+        assert result.duration == 5.0
+
+    def test_pass_through_main_has_aac_audio(self):
+        """直通main/broll场景输出带aac音频。"""
+        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            result = svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=5.0)
+
+        assert result is True  # main 类型返回有音频
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        assert "-an" not in cmd  # 不再是无声
+        assert "aac" in cmd  # 有aac音频编码
+        assert "-b:a" in cmd
+
+    def test_pass_through_background_no_audio(self):
+        """直通background场景不带音频（图片素材）。"""
+        clips = [_make_clip("bg1", "background", order=0, duration=5.0)]
+        asset_paths = {"asset_bg1.mp4": Path("/tmp/asset_bg1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            result = svc._render_pass_through(layers, Path("/tmp/out.mp4"))
+
+        assert result is False  # background 返回无音频
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        assert "aac" not in cmd  # 没有音频编码参数
+
+    # ── 无音轨视频防御测试 ──
+
+    def test_mix_audio_main_no_audio_stream_returns_none(self):
+        """主图层clip无音频流且无独立音频轨时，返回None（不报错）。"""
+        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.ffmpeg_utils.probe_has_audio", return_value=False),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            result = svc._mix_audio(layers, 5.0)
+
+        assert result is None
+        # 没有音频流时不应调用 FFmpeg
+        mock_run.assert_not_called()
+
+    def test_mix_audio_partial_clips_no_audio_filtered(self):
+        """部分主图层clip无音频流时，过滤掉无音轨的，剩余有音频的正常concat。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=3.0),  # 无音频
+            _make_clip("c2", "main", order=1, duration=2.0),  # 有音频
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        # 模拟：c1 无音频，c2 有音频
+        def fake_has_audio(path):
+            return "c2" in str(path)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.ffmpeg_utils.probe_has_audio", side_effect=fake_has_audio),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            result = svc._mix_audio(layers, 5.0)
+
+        assert result is not None
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        cmd_str = " ".join(cmd)
+        # 只剩 1 个有效音频 clip，走单clip路径（-vn），不走 filter_complex concat
+        assert "-vn" in cmd
+        assert "concat=n=2" not in cmd_str
+
+    def test_mix_audio_all_main_no_audio_but_independent_track(self):
+        """主图层全部无音频，但有独立音频轨时，正常走amix混音。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=5.0),  # 无音频
+            _make_clip(
+                "bgm1",
+                "main",
+                order=0,
+                duration=5.0,
+                config={"role": "audio", "volume": 0.5},
+            ),  # 独立音频轨（有音频）
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_bgm1.mp4": Path("/tmp/asset_bgm1.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        def fake_has_audio(path):
+            return "bgm" in str(path)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.ffmpeg_utils.probe_has_audio", side_effect=fake_has_audio),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            result = svc._mix_audio(layers, 5.0)
+
+        assert result is not None
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+        cmd_str = " ".join(cmd)
+        # 只有独立音频轨参与混音，amix 输入数=1
+        assert "amix=inputs=1" in cmd_str
+
+    def test_mix_audio_both_no_audio_returns_none(self):
+        """主图层和独立音频轨都无音频时，返回None。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=5.0),
+            _make_clip(
+                "bgm1",
+                "main",
+                order=0,
+                duration=5.0,
+                config={"role": "audio"},
+            ),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_bgm1.mp4": Path("/tmp/asset_bgm1.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.ffmpeg_utils.probe_has_audio", return_value=False),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            result = svc._mix_audio(layers, 5.0)
+
+        assert result is None
+        mock_run.assert_not_called()
+
+    def test_clip_has_audio_cache(self):
+        """_clip_has_audio 带缓存，同一clip只探测一次。"""
+        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+        ):
+            resolved = svc._resolve_clips()
+            clip = resolved[0]
+
+        with patch("video_processing.ffmpeg_utils.probe_has_audio", return_value=True) as mock_probe:
+            # 调用 3 次
+            r1 = svc._clip_has_audio(clip)
+            r2 = svc._clip_has_audio(clip)
+            r3 = svc._clip_has_audio(clip)
+
+        assert r1 is True and r2 is True and r3 is True
+        # 实际只探测了 1 次
+        assert mock_probe.call_count == 1

tests/unit/test_video_compose_security.py

@@ -1,247 +0,0 @@
-"""
-视频合成服务安全校验单元测试
-针对 PR #159 安全审计发现的问题进行测试
-"""
-
-import os
-import tempfile
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-# 导入被测试的模块
-from apps.worker.video_processing.video_compose_service import (
-    ALLOWED_INPUT_PREFIXES,
-    ALLOWED_OUTPUT_DIRS,
-    ALLOWED_TRANSITIONS,
-    Clip,
-    EditingMode,
-    EditingModeConfig,
-    VideoComposeService,
-)
-
-
-class TestOutputPathValidation:
-    """P0: 输出路径穿越校验测试"""
-
-    def setup_method(self):
-        """测试前设置"""
-        self.config = EditingModeConfig(mode=EditingMode.ONE_TAKE)
-        self.service = VideoComposeService(self.config)
-
-    def test_valid_output_path_in_allowed_dir(self):
-        """测试合法的输出路径"""
-        valid_path = "/tmp/video_output/test.mp4"
-        result = self.service._validate_output_path(valid_path)
-        assert result == os.path.abspath(valid_path)
-
-    def test_valid_output_path_with_relative_components(self):
-        """测试带相对路径成分但最终在允许目录内的路径"""
-        valid_path = "/tmp/video_output/subdir/../test.mp4"
-        result = self.service._validate_output_path(valid_path)
-        assert result == os.path.abspath(valid_path)
-
-    def test_path_traversal_attack_blocked(self):
-        """测试路径穿越攻击被阻止"""
-        # 尝试穿越到 /etc/passwd
-        malicious_path = "/tmp/video_output/../../../etc/passwd"
-        with pytest.raises(ValueError, match="输出路径不在允许范围内"):
-            self.service._validate_output_path(malicious_path)
-
-    def test_path_traversal_attack_blocked_var_app(self):
-        """测试针对 /var/app 的路径穿越攻击被阻止"""
-        malicious_path = "/var/app/rendered/../../config/../../../etc/passwd"
-        with pytest.raises(ValueError, match="输出路径不在允许范围内"):
-            self.service._validate_output_path(malicious_path)
-
-    def test_absolute_path_to_forbidden_location(self):
-        """测试直接访问禁止位置"""
-        forbidden_path = "/etc/shadow"
-        with pytest.raises(ValueError, match="输出路径不在允许范围内"):
-            self.service._validate_output_path(forbidden_path)
-
-    def test_root_path_blocked(self):
-        """测试根目录被阻止"""
-        root_path = "/"
-        with pytest.raises(ValueError, match="输出路径不在允许范围内"):
-            self.service._validate_output_path(root_path)
-
-    def test_absolute_path_to_tmp_not_allowed(self):
-        """测试 /tmp 不在白名单中时应被阻止"""
-        # /tmp 不在 ALLOWED_OUTPUT_DIRS 中
-        tmp_path = "/tmp/test.mp4"
-        with pytest.raises(ValueError, match="输出路径不在允许范围内"):
-            self.service._validate_output_path(tmp_path)
-
-
-class TestInputPathValidation:
-    """P1-1: 输入路径格式校验测试"""
-
-    def setup_method(self):
-        """测试前设置"""
-        self.config = EditingModeConfig(mode=EditingMode.ONE_TAKE)
-        self.service = VideoComposeService(self.config)
-
-    def test_valid_s3_path(self):
-        """测试 S3 路径"""
-        assert self.service._validate_input_path("s3://bucket/key.mp4") is True
-
-    def test_valid_oss_path(self):
-        """测试 OSS 路径"""
-        assert self.service._validate_input_path("oss://bucket/key.mp4") is True
-
-    def test_valid_local_path(self):
-        """测试 local:// 路径"""
-        assert self.service._validate_input_path("local://asset/123.mp4") is True
-
-    def test_valid_var_storage_path(self):
-        """测试 /var/storage/ 路径"""
-        assert self.service._validate_input_path("/var/storage/assets/123.mp4") is True
-
-    def test_path_traversal_in_input_rejected(self):
-        """测试输入路径中的路径穿越尝试被拒绝"""
-        malicious_path = "s3://bucket/../../etc/passwd"
-        # 这会通过前缀检查，但实际使用时文件系统访问会失败
-        # 安全设计：只校验格式前缀
-        assert self.service._validate_input_path(malicious_path) is True
-
-    def test_malicious_input_path_blocked(self):
-        """测试恶意输入路径被阻止"""
-        assert self.service._validate_input_path("/etc/passwd") is False
-        assert self.service._validate_input_path("file:///etc/passwd") is False
-        assert self.service._validate_input_path("http://evil.com/shell.sh") is False
-
-    def test_empty_path_rejected(self):
-        """测试空路径被拒绝"""
-        assert self.service._validate_input_path("") is False
-
-    def test_random_string_rejected(self):
-        """测试随机字符串被拒绝"""
-        assert self.service._validate_input_path("random123") is False
-        assert self.service._validate_input_path("abc../../../etc") is False
-
-
-class TestTransitionValidation:
-    """P1-2: 转场参数白名单校验测试"""
-
-    def setup_method(self):
-        """测试前设置"""
-        self.config = EditingModeConfig(mode=EditingMode.ONE_TAKE)
-        self.service = VideoComposeService(self.config)
-
-    @pytest.mark.parametrize("transition", list(ALLOWED_TRANSITIONS))
-    def test_valid_transitions(self, transition):
-        """测试所有合法的转场效果"""
-        result = self.service._validate_transition(transition)
-        assert result == transition
-
-    def test_invalid_transition_defaults_to_fade(self):
-        """测试非法转场效果默认为 fade"""
-        result = self.service._validate_transition("random_transition")
-        assert result == "fade"
-
-    def test_sql_injection_in_transition_blocked(self):
-        """测试 SQL 注入尝试被阻止"""
-        result = self.service._validate_transition("fade; DROP TABLE videos;--")
-        assert result == "fade"
-
-    def test_shell_injection_in_transition_blocked(self):
-        """测试 Shell 注入尝试被阻止"""
-        result = self.service._validate_transition("fade$(whoami)")
-        assert result == "fade"
-
-    def test_empty_transition_handled(self):
-        """测试空转场名称"""
-        result = self.service._validate_transition("")
-        assert result == "fade"
-
-    def test_none_transition_handled(self):
-        """测试 None 转场名称"""
-        result = self.service._validate_transition(None)
-        assert result == "fade"
-
-    def test_get_validated_transition_returns_mapped(self):
-        """测试 _get_validated_transition 返回映射后的值"""
-        # "fade" 应该映射为 "fade"
-        result = self.service._get_validated_transition("fade")
-        assert result == "fade"
-
-
-class TestComposeSecurityIntegration:
-    """安全集成测试"""
-
-    def setup_method(self):
-        """测试前设置"""
-        self.config = EditingModeConfig(mode=EditingMode.ONE_TAKE)
-        self.service = VideoComposeService(self.config)
-
-    def test_compose_rejects_malicious_output_path(self):
-        """测试 compose 方法拒绝恶意输出路径"""
-        clips = [
-            Clip(asset_id="s3://bucket/video1.mp4"),
-            Clip(asset_id="s3://bucket/video2.mp4"),
-        ]
-
-        with pytest.raises(ValueError, match="输出路径不在允许范围内"):
-            self.service.compose(clips, output_path="/etc/passwd")
-
-    def test_compose_rejects_invalid_input_path(self):
-        """测试 compose 方法拒绝非法输入路径"""
-        clips = [
-            Clip(asset_id="/etc/shadow"),  # 非法路径
-        ]
-
-        with pytest.raises(ValueError, match="不合法的输入路径"):
-            self.service.compose(clips)
-
-    def test_compose_with_valid_paths(self):
-        """测试合法路径可以正常处理"""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            # 创建临时视频文件
-            video_path = os.path.join(tmpdir, "input.mp4")
-            output_path = os.path.join("/tmp/video_output", "output.mp4")
-
-            # 创建空的测试文件（实际测试需要真实视频）
-            with open(video_path, "wb") as f:
-                f.write(b"fake video data")
-
-            clips = [
-                Clip(asset_id=f"local://{video_path}"),
-            ]
-
-            # 验证输入校验通过
-            assert self.service._validate_input_path(f"local://{video_path}") is True
-
-    def test_compose_empty_clips_rejected(self):
-        """测试空片段列表被拒绝"""
-        with pytest.raises(ValueError, match="clips 不能为空"):
-            self.service.compose([])
-
-
-class TestWhiteListConstants:
-    """白名单常量测试"""
-
-    def test_allowed_output_dirs_not_empty(self):
-        """测试输出目录白名单不为空"""
-        assert len(ALLOWED_OUTPUT_DIRS) > 0
-        assert "/tmp/video_output" in ALLOWED_OUTPUT_DIRS
-        assert "/var/app/rendered" in ALLOWED_OUTPUT_DIRS
-
-    def test_allowed_input_prefixes_not_empty(self):
-        """测试输入路径前缀白名单不为空"""
-        assert len(ALLOWED_INPUT_PREFIXES) > 0
-        assert "s3://" in ALLOWED_INPUT_PREFIXES
-        assert "oss://" in ALLOWED_INPUT_PREFIXES
-        assert "local://" in ALLOWED_INPUT_PREFIXES
-        assert "/var/storage/" in ALLOWED_INPUT_PREFIXES
-
-    def test_allowed_transitions_not_empty(self):
-        """测试转场效果白名单不为空"""
-        assert len(ALLOWED_TRANSITIONS) > 0
-        assert "fade" in ALLOWED_TRANSITIONS
-        assert "dissolve" in ALLOWED_TRANSITIONS
-        assert "slideleft" in ALLOWED_TRANSITIONS
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])