Merge branch 'develop' into fix/render-engine-security-p0-p1

# Conflicts:
#	apps/worker/video_processing/path_security.py

.ci-trigger

@@ -1 +1 @@
-# CI trigger Fri Jun 26 09:53:28 PM CST 2026
+trigger: 1784009947

.gitea/workflows/ci-cd.yml

@@ -101,6 +101,62 @@ jobs:
           bandit --version
           pytest --version
 
+      - name: Secret detection (detect-secrets)
+        shell: sh
+        run: |
+          set -eu
+          echo "=== Installing detect-secrets ==="
+          python3 -m pip install -q detect-secrets
+          detect-secrets --version
+          echo ""
+          echo "=== Running secret scan ==="
+          detect-secrets scan \
+            --all-files \
+            --exclude-files '(^|/)(tests|test|e2e|__tests__|spec|docs|node_modules|site-packages|migrations|alembic|.gitea|.git|.pytest_cache|.next|dist|build)/' \
+            --exclude-files '\.(md|rst|txt|lock|example|sample|min\.js|min\.css|spec\.ts|test\.ts|test\.py)$' \
+            --exclude-files '(package-lock|yarn\.lock|poetry\.lock|Pipfile\.lock)$' \
+            --disable-plugin Base64HighEntropyString \
+            --disable-plugin HexHighEntropyString \
+            --disable-plugin BasicAuthDetector \
+            --disable-plugin KeywordDetector \
+            --disable-plugin IPPublicDetector \
+            2>&1 | tee /tmp/secrets-scan.json
+
+          FOUND=$(python3 -c "
+          import json
+          try:
+              with open('/tmp/secrets-scan.json') as f:
+                  data = json.load(f)
+              results = data.get('results', {})
+              total = sum(len(v) for v in results.values())
+              print(total)
+          except Exception:
+              print('error')
+          ")
+          echo ""
+          echo "Secrets detected: $FOUND"
+          if [ "$FOUND" != "0" ] && [ "$FOUND" != "error" ]; then
+            echo ""
+            echo "=== Secret details ==="
+            python3 -c "
+          import json
+          with open('/tmp/secrets-scan.json') as f:
+              data = json.load(f)
+          for fpath, items in data.get('results', {}).items():
+              for item in items:
+                  line = item.get('line_number', '?')
+                  stype = item.get('type', '?')
+                  hashed = item.get('hashed_secret', '')[:16]
+                  print(f'  {fpath}:{line} [{stype}] {hashed}...')
+          "
+            echo ""
+            echo "ERROR: Potential secrets detected in code!"
+            echo "If these are false positives, add exclusions in the CI workflow."
+            exit 1
+          fi
+          echo "Secret scan completed - no secrets detected"
+
+
       - name: Run code quality checks
         shell: sh
         run: |
@@ -110,12 +166,110 @@ jobs:
           python3 -m isort --check-only alembic apps packages tests scripts
           python3 -m flake8 apps packages tests --count --statistics
 
-      - name: Run security scan
+      - name: Ruff lint (advisory mode - 摸底阶段)
+        if: always()
+        shell: sh
+        run: |
+          set +e
+          echo "=== Installing ruff ==="
+          python3 -m pip install -q ruff
+          ruff --version
+          echo ""
+          echo "=== Running ruff lint (advisory mode) ==="
+          echo "告警模式，不阻断CI。用于摸底问题数量，后续分批修复后正式替换flake8。"
+          echo ""
+          ruff check apps packages tests scripts             --statistics             --output-format concise             2>&1 | tail -30
+          EXIT_CODE=$?
+          echo ""
+          if [ "$EXIT_CODE" != "0" ]; then
+            echo "ruff 发现 lint 问题（告警模式，不阻断）"
+            echo "问题分类统计见上方，后续将分批修复"
+          else
+            echo "ruff 检查全部通过 ✅"
+          fi
+          exit 0
+
+      - name: Type check (mypy, advisory mode)
+        if: always()
+        shell: sh
+        run: |
+          set +e
+          echo "=== Installing mypy ==="
+          python3 -m pip install -q mypy
+          mypy --version
+          echo ""
+          echo "=== Running mypy type check (advisory mode) ==="
+          echo "告警模式，不阻断CI"
+          echo ""
+          # 只检查核心业务代码，跳过测试和迁移
+          EXIT_CODE=0
+          mypy apps/api/app packages             --ignore-missing-imports             --no-site-packages             --no-strict-optional             --explicit-package-bases             --exclude 'tests/|test_|migrations/|alembic/'             --no-error-summary             2>&1 | head -60 || EXIT_CODE=$?
+          echo ""
+          if [ "$EXIT_CODE" != "0" ]; then
+            echo "mypy 发现类型问题（告警模式，不阻断）"
+            echo "建议后续逐步修复"
+          else
+            echo "mypy 类型检查通过 ✅"
+          fi
+          exit 0
+
+      - name: Run security scan (bandit)
         shell: sh
         run: |
           set -eu
           bandit -r apps packages -q -ll
 
+      - name: Python dependency vulnerability scan (pip-audit)
+        shell: sh
+        run: |
+          set -eu
+          echo "=== Installing pip-audit ==="
+          python3 -m pip install -q pip-audit
+          pip-audit --version
+          echo ""
+          echo "=== Scanning Python dependencies ==="
+          EXIT_CODE=0
+          for req_file in requirements.txt requirements-base.txt requirements-dev.txt; do
+            if [ -f "$req_file" ]; then
+              echo "--- Scanning $req_file ---"
+              pip-audit -r "$req_file" --desc on 2>&1 | head -40 || EXIT_CODE=$?
+              echo ""
+            fi
+          done
+          echo "pip-audit scan completed (advisory mode - warnings only, not blocking CI)"
+          if [ "$EXIT_CODE" != "0" ]; then
+            echo "WARNING: Potential vulnerabilities found in dependencies."
+          fi
+          exit 0
+
+      - name: Dead code detection (vulture)
+        if: always()
+        shell: sh
+        run: |
+          set +e
+          echo "=== Installing vulture ==="
+          python3 -m pip install -q vulture
+          vulture --version
+          echo ""
+          echo "=== Running vulture dead code scan (confidence >= 70%) ==="
+          echo "告警模式，不阻断CI。置信度>=90%建议尽快确认。"
+          echo ""
+          # 按置信度从高到低输出，便于优先查看高价值条目
+          vulture apps packages scripts \
+            --exclude "tests,test,migrations,.gitea,docs,node_modules,site-packages,*/test_*.py,*/conftest.py" \
+            --min-confidence 70 \
+            2>&1 | sort -t'(' -k2 -rn | head -80
+          EXIT_CODE=$?
+          echo ""
+          echo "=== vulture scan summary ==="
+          if [ "$EXIT_CODE" != "0" ]; then
+            echo "发现潜在死代码（可能包含框架装饰器注册的函数，为误报）"
+            echo "建议：定期人工审查高置信度（>=90%）条目"
+          else
+            echo "未发现明显死代码 ✅"
+          fi
+          exit 0
+
       - name: Validate release scripts syntax
         shell: sh
         run: |
@@ -651,10 +805,10 @@ jobs:
           echo "=== CI 失败通知 ==="
           FAILED_JOB="Frontend Lint" python3 scripts/ci_notify_failure.py
 
-  deploy-staging:
-    name: Build & Push Staging (Watchtower auto-deploy)
-    runs-on: saas
-    timeout-minutes: 30
+  build-staging-api:
+    name: Build Staging API Image
+    runs-on: [saas, build-farm]
+    timeout-minutes: 20
     needs: [validate, frontend-lint]
 
     if: github.event_name == 'push' && (github.ref_name == 'main' || github.ref_name == 'develop')
@@ -704,30 +858,391 @@ jobs:
                 if member.name:
                     tar.extract(member, '.')
         INNERPY
-
-    - name: Build and push all images to Gitea Registry
+    - name: Docker login to Registry
       shell: sh
       env:
         REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
       run: |
         set -eu
-        chmod +x scripts/build_release_images.sh
-        ALLOW_SHARED_PRODUCTION_BUILD_HOST=true REGISTRY_TOKEN="${REGISTRY_TOKEN}" \
-          scripts/build_release_images.sh "${GITHUB_SHA}" staging
+        printf '%s' "${REGISTRY_TOKEN}" | docker login git.xiaoxiajianji.com -u xiaoxia --password-stdin
+        echo "Docker login successful"
+    - name: Setup cache strategy
+      shell: sh
+      run: |
+        set -eu
+        # develop/main 分支写回缓存，其他分支只读
+        if [ "${GITHUB_REF_NAME}" = "develop" ] || [ "${GITHUB_REF_NAME}" = "main" ]; then
+          echo "CACHE_MODE=read-write" >> $GITHUB_ENV
+          echo "Cache mode: read-write (will push cache)"
+        else
+          echo "CACHE_MODE=read-only" >> $GITHUB_ENV
+          echo "Cache mode: read-only"
+        fi
 
-    - name: Tag and push :staging images (Watchtower auto-update)
+    - name: Setup buildx builder (docker-container driver)
+      shell: sh
+      run: |
+        set -eu
+        # 确保使用 docker-container driver 以支持 cache export 功能
+        if ! docker buildx inspect ci-builder > /dev/null 2>&1; then
+          docker buildx create --use --name ci-builder --driver docker-container
+          echo "Created ci-builder (docker-container driver)"
+        else
+          docker buildx use ci-builder
+          echo "Using existing ci-builder"
+        fi
+        docker buildx inspect --bootstrap
+
+    - name: Build and push API image (buildx cache)
       shell: sh
-      env:
-        REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
       run: |
         set -eu
         REGISTRY="git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas"
-        if [ -n "${REGISTRY_TOKEN:-}" ]; then
-          printf '%s' "${REGISTRY_TOKEN}" | docker login git.xiaoxiajianji.com -u xiaoxia --password-stdin 2>/dev/null
+        IMAGE_NAME="xiaoxia-saas-api"
+        CACHE_REF="${REGISTRY}/api-cache:develop"
+
+        CACHE_FROM="type=registry,ref=${CACHE_REF},ignore-error=true"
+
+        if [ "${CACHE_MODE}" = "read-write" ]; then
+          CACHE_TO="type=registry,ref=${CACHE_REF},mode=max"
+          echo "Building API image with read-write cache..."
+          docker buildx build             --build-arg APP_VERSION="${GITHUB_SHA}"             --cache-from "${CACHE_FROM}"             --cache-to "${CACHE_TO}"             -f infra/docker/api.Dockerfile             -t "${REGISTRY}/${IMAGE_NAME}:${GITHUB_SHA}"             --push             .
+        else
+          echo "Building API image with read-only cache..."
+          docker buildx build             --build-arg APP_VERSION="${GITHUB_SHA}"             --cache-from "${CACHE_FROM}"             -f infra/docker/api.Dockerfile             -t "${REGISTRY}/${IMAGE_NAME}:${GITHUB_SHA}"             --push             .
         fi
+        echo "API image pushed: ${REGISTRY}/${IMAGE_NAME}:${GITHUB_SHA}"
+
+    - name: Notify CI failure
+      if: failure()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 失败通知 ==="
+        FAILED_JOB="Build Staging API Image" python3 scripts/ci_notify_failure.py
+
+  build-staging-worker:
+    name: Build Staging Worker Image
+    runs-on: [saas, build-farm]
+    timeout-minutes: 20
+    needs: [validate, frontend-lint]
+
+    if: github.event_name == 'push' && (github.ref_name == 'main' || github.ref_name == 'develop')
+
+    steps:
+    - name: Checkout code
+      shell: sh
+      env:
+        GITHUB_TOKEN: ${{ github.token }}
+      run: |
+        set -eu
+        python3 - <<'INNERPY'
+        import io, os, tarfile, time, urllib.request, urllib.error
+        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+        last_err = None
+        for attempt in range(5):
+            try:
+                with urllib.request.urlopen(request, timeout=120) as response:
+                    archive = response.read()
+                break
+            except urllib.error.HTTPError as e:
+                last_err = e
+                if e.code >= 500 and attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+            except Exception as e:
+                last_err = e
+                if attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+        else:
+            raise last_err
+        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+            for member in tar.getmembers():
+                name = member.name
+                if name == root_prefix[:-1]:
+                    continue
+                if name.startswith(root_prefix):
+                    member.name = name[len(root_prefix):]
+                if member.name:
+                    tar.extract(member, '.')
+        INNERPY
+    - name: Docker login to Registry
+      shell: sh
+      env:
+        REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
+      run: |
+        set -eu
+        printf '%s' "${REGISTRY_TOKEN}" | docker login git.xiaoxiajianji.com -u xiaoxia --password-stdin
+        echo "Docker login successful"
+    - name: Setup cache strategy
+      shell: sh
+      run: |
+        set -eu
+        # develop/main 分支写回缓存，其他分支只读
+        if [ "${GITHUB_REF_NAME}" = "develop" ] || [ "${GITHUB_REF_NAME}" = "main" ]; then
+          echo "CACHE_MODE=read-write" >> $GITHUB_ENV
+          echo "Cache mode: read-write (will push cache)"
+        else
+          echo "CACHE_MODE=read-only" >> $GITHUB_ENV
+          echo "Cache mode: read-only"
+        fi
+
+    - name: Setup buildx builder (docker-container driver)
+      shell: sh
+      run: |
+        set -eu
+        # 确保使用 docker-container driver 以支持 cache export 功能
+        if ! docker buildx inspect ci-builder > /dev/null 2>&1; then
+          docker buildx create --use --name ci-builder --driver docker-container
+          echo "Created ci-builder (docker-container driver)"
+        else
+          docker buildx use ci-builder
+          echo "Using existing ci-builder"
+        fi
+        docker buildx inspect --bootstrap
+
+    - name: Build and push Worker image (buildx cache)
+      shell: sh
+      run: |
+        set -eu
+        REGISTRY="git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas"
+        IMAGE_NAME="xiaoxia-saas-worker"
+        CACHE_REF="${REGISTRY}/worker-cache:develop"
+
+        CACHE_FROM="type=registry,ref=${CACHE_REF},ignore-error=true"
+
+        if [ "${CACHE_MODE}" = "read-write" ]; then
+          CACHE_TO="type=registry,ref=${CACHE_REF},mode=max"
+          echo "Building Worker image with read-write cache..."
+          docker buildx build             --build-arg APP_VERSION="${GITHUB_SHA}"             --cache-from "${CACHE_FROM}"             --cache-to "${CACHE_TO}"             -f infra/docker/worker.Dockerfile             -t "${REGISTRY}/${IMAGE_NAME}:${GITHUB_SHA}"             --push             .
+        else
+          echo "Building Worker image with read-only cache..."
+          docker buildx build             --build-arg APP_VERSION="${GITHUB_SHA}"             --cache-from "${CACHE_FROM}"             -f infra/docker/worker.Dockerfile             -t "${REGISTRY}/${IMAGE_NAME}:${GITHUB_SHA}"             --push             .
+        fi
+        echo "Worker image pushed: ${REGISTRY}/${IMAGE_NAME}:${GITHUB_SHA}"
+
+    - name: Notify CI failure
+      if: failure()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 失败通知 ==="
+        FAILED_JOB="Build Staging Worker Image" python3 scripts/ci_notify_failure.py
+
+  build-staging-web:
+    name: Build Staging Web Image
+    runs-on: [saas, build-farm]
+    timeout-minutes: 20
+    needs: [validate, frontend-lint]
+
+    if: github.event_name == 'push' && (github.ref_name == 'main' || github.ref_name == 'develop')
+
+    steps:
+    - name: Checkout code
+      shell: sh
+      env:
+        GITHUB_TOKEN: ${{ github.token }}
+      run: |
+        set -eu
+        python3 - <<'INNERPY'
+        import io, os, tarfile, time, urllib.request, urllib.error
+        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+        last_err = None
+        for attempt in range(5):
+            try:
+                with urllib.request.urlopen(request, timeout=120) as response:
+                    archive = response.read()
+                break
+            except urllib.error.HTTPError as e:
+                last_err = e
+                if e.code >= 500 and attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+            except Exception as e:
+                last_err = e
+                if attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+        else:
+            raise last_err
+        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+            for member in tar.getmembers():
+                name = member.name
+                if name == root_prefix[:-1]:
+                    continue
+                if name.startswith(root_prefix):
+                    member.name = name[len(root_prefix):]
+                if member.name:
+                    tar.extract(member, '.')
+        INNERPY
+    - name: Docker login to Registry
+      shell: sh
+      env:
+        REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
+      run: |
+        set -eu
+        printf '%s' "${REGISTRY_TOKEN}" | docker login git.xiaoxiajianji.com -u xiaoxia --password-stdin
+        echo "Docker login successful"
+    - name: Setup cache strategy
+      shell: sh
+      run: |
+        set -eu
+        # develop/main 分支写回缓存，其他分支只读
+        if [ "${GITHUB_REF_NAME}" = "develop" ] || [ "${GITHUB_REF_NAME}" = "main" ]; then
+          echo "CACHE_MODE=read-write" >> $GITHUB_ENV
+          echo "Cache mode: read-write (will push cache)"
+        else
+          echo "CACHE_MODE=read-only" >> $GITHUB_ENV
+          echo "Cache mode: read-only"
+        fi
+
+    - name: Build frontend assets (npm build)
+      shell: sh
+      run: |
+        set -eu
+        NPM_CACHE_VOLUME="xiaoxia-npm-cache"
+        if ! docker volume inspect "$NPM_CACHE_VOLUME" >/dev/null 2>&1; then
+          docker volume create "$NPM_CACHE_VOLUME" >/dev/null
+          echo "Created npm cache volume: $NPM_CACHE_VOLUME"
+        fi
+
+        docker run --rm           -v "$PWD:/workspace"           -v "$NPM_CACHE_VOLUME:/workspace/apps/web/node_modules"           -w /workspace/apps/web           docker.m.daocloud.io/library/node:20           sh -lc "npm ci && npm run build"
+
+        test -f apps/web/dist/index.html
+        echo "Frontend build complete: $(ls apps/web/dist/ | head -5)"
+
+    - name: Setup buildx builder (docker-container driver)
+      shell: sh
+      run: |
+        set -eu
+        # 确保使用 docker-container driver 以支持 cache export 功能
+        if ! docker buildx inspect ci-builder > /dev/null 2>&1; then
+          docker buildx create --use --name ci-builder --driver docker-container
+          echo "Created ci-builder (docker-container driver)"
+        else
+          docker buildx use ci-builder
+          echo "Using existing ci-builder"
+        fi
+        docker buildx inspect --bootstrap
+
+    - name: Build and push Web image (buildx cache)
+      shell: sh
+      run: |
+        set -eu
+        REGISTRY="git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas"
+        IMAGE_NAME="xiaoxia-saas-web"
+        CACHE_REF="${REGISTRY}/web-cache:develop"
+        NGINX_CONF="infra/docker/nginx-staging.conf"
+
+        CACHE_FROM="type=registry,ref=${CACHE_REF},ignore-error=true"
+
+        if [ "${CACHE_MODE}" = "read-write" ]; then
+          CACHE_TO="type=registry,ref=${CACHE_REF},mode=max"
+          echo "Building Web image with read-write cache..."
+          docker buildx build             --cache-from "${CACHE_FROM}"             --cache-to "${CACHE_TO}"             -f infra/docker/web-artifact.Dockerfile             --build-arg "NGINX_CONF=${NGINX_CONF}"             -t "${REGISTRY}/${IMAGE_NAME}:${GITHUB_SHA}"             --push             .
+        else
+          echo "Building Web image with read-only cache..."
+          docker buildx build             --cache-from "${CACHE_FROM}"             -f infra/docker/web-artifact.Dockerfile             --build-arg "NGINX_CONF=${NGINX_CONF}"             -t "${REGISTRY}/${IMAGE_NAME}:${GITHUB_SHA}"             --push             .
+        fi
+        echo "Web image pushed: ${REGISTRY}/${IMAGE_NAME}:${GITHUB_SHA}"
+
+    - name: Notify CI failure
+      if: failure()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 失败通知 ==="
+        FAILED_JOB="Build Staging Web Image" python3 scripts/ci_notify_failure.py
+
+  deploy-staging:
+    name: Deploy Staging (Watchtower auto-deploy)
+    runs-on: saas
+    timeout-minutes: 15
+    needs: [build-staging-api, build-staging-worker, build-staging-web]
+
+    if: github.event_name == 'push' && (github.ref_name == 'main' || github.ref_name == 'develop')
+
+    steps:
+    - name: Checkout code
+      shell: sh
+      env:
+        GITHUB_TOKEN: ${{ github.token }}
+      run: |
+        set -eu
+        python3 - <<'INNERPY'
+        import io, os, tarfile, time, urllib.request, urllib.error
+        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+        last_err = None
+        for attempt in range(5):
+            try:
+                with urllib.request.urlopen(request, timeout=120) as response:
+                    archive = response.read()
+                break
+            except urllib.error.HTTPError as e:
+                last_err = e
+                if e.code >= 500 and attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+            except Exception as e:
+                last_err = e
+                if attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+        else:
+            raise last_err
+        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+            for member in tar.getmembers():
+                name = member.name
+                if name == root_prefix[:-1]:
+                    continue
+                if name.startswith(root_prefix):
+                    member.name = name[len(root_prefix):]
+                if member.name:
+                    tar.extract(member, '.')
+        INNERPY
+    - name: Docker login to Registry
+      shell: sh
+      env:
+        REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
+      run: |
+        set -eu
+        printf '%s' "${REGISTRY_TOKEN}" | docker login git.xiaoxiajianji.com -u xiaoxia --password-stdin
+        echo "Docker login successful"
+
+    - name: Tag and push :staging images (Watchtower auto-update)
+      shell: sh
+      run: |
+        set -eu
+        REGISTRY="git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas"
+
         for svc in api worker web; do
+          echo "Pulling ${REGISTRY}/xiaoxia-saas-${svc}:${GITHUB_SHA} ..."
+          docker pull "${REGISTRY}/xiaoxia-saas-${svc}:${GITHUB_SHA}"
           docker tag "${REGISTRY}/xiaoxia-saas-${svc}:${GITHUB_SHA}" "${REGISTRY}/xiaoxia-saas-${svc}:staging"
           docker push "${REGISTRY}/xiaoxia-saas-${svc}:staging"
+          echo "$svc :staging tagged and pushed"
         done
         echo "All :staging images pushed. Watchtower will auto-deploy within 60s."
 
@@ -796,8 +1311,7 @@ jobs:
       run: |
         set +e
         echo "=== CI 失败通知 ==="
-        FAILED_JOB="Build & Push Staging (Watchtower auto-deploy)" python3 scripts/ci_notify_failure.py
-
+        FAILED_JOB="Deploy Staging" python3 scripts/ci_notify_failure.py
 
 
   staging-e2e:
@@ -951,10 +1465,10 @@ jobs:
 
 
 
-  build-production-runtime-images:
-    name: Build Production Runtime Images
-    runs-on: saas
-    timeout-minutes: 30
+  build-production-api:
+    name: Build Production API Image
+    runs-on: [saas, build-farm]
+    timeout-minutes: 20
     needs: [validate, frontend-lint]
 
     if: startsWith(github.ref, 'refs/tags/v')
@@ -966,7 +1480,7 @@ jobs:
         GITHUB_TOKEN: ${{ github.token }}
       run: |
         set -eu
-        python3 - <<'PY'
+        python3 - <<'INNERPY'
         import io, os, tarfile, time, urllib.request, urllib.error
         url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
         request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
@@ -991,7 +1505,6 @@ jobs:
                     print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
                     time.sleep(wait)
                     continue
-                raise
         else:
             raise last_err
         with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
@@ -1004,16 +1517,251 @@ jobs:
                     member.name = name[len(root_prefix):]
                 if member.name:
                     tar.extract(member, '.')
-        PY
-
-    - name: Build and push all images (api + worker + web, with buildx cache)
+        INNERPY
+    - name: Docker login to Registry
       shell: sh
       env:
         REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
       run: |
         set -eu
-        chmod +x scripts/build_release_images.sh
-        REGISTRY_TOKEN="${REGISTRY_TOKEN}" scripts/build_release_images.sh "${GITHUB_REF_NAME}"
+        printf '%s' "${REGISTRY_TOKEN}" | docker login git.xiaoxiajianji.com -u xiaoxia --password-stdin
+        echo "Docker login successful"
+
+    - name: Setup buildx builder (docker-container driver)
+      shell: sh
+      run: |
+        set -eu
+        # 确保使用 docker-container driver 以支持 cache export 功能
+        if ! docker buildx inspect ci-builder > /dev/null 2>&1; then
+          docker buildx create --use --name ci-builder --driver docker-container
+          echo "Created ci-builder (docker-container driver)"
+        else
+          docker buildx use ci-builder
+          echo "Using existing ci-builder"
+        fi
+        docker buildx inspect --bootstrap
+
+    - name: Build and push API image (buildx cache)
+      shell: sh
+      run: |
+        set -eu
+        REGISTRY="git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas"
+        IMAGE_NAME="xiaoxia-saas-api"
+        VERSION="${GITHUB_REF_NAME}"
+        CACHE_REF="${REGISTRY}/api-cache:main"
+
+        echo "Building Production API image: ${VERSION}"
+        docker buildx build           --build-arg APP_VERSION="${VERSION}"           --cache-from "type=registry,ref=${CACHE_REF},ignore-error=true"           --cache-to "type=registry,ref=${CACHE_REF},mode=max"           -f infra/docker/api.Dockerfile           -t "${REGISTRY}/${IMAGE_NAME}:${VERSION}"           --push           .
+        echo "Production API image pushed: ${REGISTRY}/${IMAGE_NAME}:${VERSION}"
+
+    - name: Notify CI failure
+      if: failure()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 失败通知 ==="
+        FAILED_JOB="Build Production API Image" python3 scripts/ci_notify_failure.py
+
+  build-production-worker:
+    name: Build Production Worker Image
+    runs-on: [saas, build-farm]
+    timeout-minutes: 20
+    needs: [validate, frontend-lint]
+
+    if: startsWith(github.ref, 'refs/tags/v')
+
+    steps:
+    - name: Checkout code
+      shell: sh
+      env:
+        GITHUB_TOKEN: ${{ github.token }}
+      run: |
+        set -eu
+        python3 - <<'INNERPY'
+        import io, os, tarfile, time, urllib.request, urllib.error
+        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+        last_err = None
+        for attempt in range(5):
+            try:
+                with urllib.request.urlopen(request, timeout=120) as response:
+                    archive = response.read()
+                break
+            except urllib.error.HTTPError as e:
+                last_err = e
+                if e.code >= 500 and attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+            except Exception as e:
+                last_err = e
+                if attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+        else:
+            raise last_err
+        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+            for member in tar.getmembers():
+                name = member.name
+                if name == root_prefix[:-1]:
+                    continue
+                if name.startswith(root_prefix):
+                    member.name = name[len(root_prefix):]
+                if member.name:
+                    tar.extract(member, '.')
+        INNERPY
+    - name: Docker login to Registry
+      shell: sh
+      env:
+        REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
+      run: |
+        set -eu
+        printf '%s' "${REGISTRY_TOKEN}" | docker login git.xiaoxiajianji.com -u xiaoxia --password-stdin
+        echo "Docker login successful"
+
+    - name: Setup buildx builder (docker-container driver)
+      shell: sh
+      run: |
+        set -eu
+        # 确保使用 docker-container driver 以支持 cache export 功能
+        if ! docker buildx inspect ci-builder > /dev/null 2>&1; then
+          docker buildx create --use --name ci-builder --driver docker-container
+          echo "Created ci-builder (docker-container driver)"
+        else
+          docker buildx use ci-builder
+          echo "Using existing ci-builder"
+        fi
+        docker buildx inspect --bootstrap
+
+    - name: Build and push Worker image (buildx cache)
+      shell: sh
+      run: |
+        set -eu
+        REGISTRY="git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas"
+        IMAGE_NAME="xiaoxia-saas-worker"
+        VERSION="${GITHUB_REF_NAME}"
+        CACHE_REF="${REGISTRY}/worker-cache:main"
+
+        echo "Building Production Worker image: ${VERSION}"
+        docker buildx build           --build-arg APP_VERSION="${VERSION}"           --cache-from "type=registry,ref=${CACHE_REF},ignore-error=true"           --cache-to "type=registry,ref=${CACHE_REF},mode=max"           -f infra/docker/worker.Dockerfile           -t "${REGISTRY}/${IMAGE_NAME}:${VERSION}"           --push           .
+        echo "Production Worker image pushed: ${REGISTRY}/${IMAGE_NAME}:${VERSION}"
+
+    - name: Notify CI failure
+      if: failure()
+      shell: sh
+      run: |
+        set +e
+        echo "=== CI 失败通知 ==="
+        FAILED_JOB="Build Production Worker Image" python3 scripts/ci_notify_failure.py
+
+  build-production-web:
+    name: Build Production Web Image
+    runs-on: [saas, build-farm]
+    timeout-minutes: 20
+    needs: [validate, frontend-lint]
+
+    if: startsWith(github.ref, 'refs/tags/v')
+
+    steps:
+    - name: Checkout code
+      shell: sh
+      env:
+        GITHUB_TOKEN: ${{ github.token }}
+      run: |
+        set -eu
+        python3 - <<'INNERPY'
+        import io, os, tarfile, time, urllib.request, urllib.error
+        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+        last_err = None
+        for attempt in range(5):
+            try:
+                with urllib.request.urlopen(request, timeout=120) as response:
+                    archive = response.read()
+                break
+            except urllib.error.HTTPError as e:
+                last_err = e
+                if e.code >= 500 and attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+            except Exception as e:
+                last_err = e
+                if attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+        else:
+            raise last_err
+        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+            for member in tar.getmembers():
+                name = member.name
+                if name == root_prefix[:-1]:
+                    continue
+                if name.startswith(root_prefix):
+                    member.name = name[len(root_prefix):]
+                if member.name:
+                    tar.extract(member, '.')
+        INNERPY
+    - name: Docker login to Registry
+      shell: sh
+      env:
+        REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
+      run: |
+        set -eu
+        printf '%s' "${REGISTRY_TOKEN}" | docker login git.xiaoxiajianji.com -u xiaoxia --password-stdin
+        echo "Docker login successful"
+
+    - name: Build frontend assets (npm build)
+      shell: sh
+      run: |
+        set -eu
+        NPM_CACHE_VOLUME="xiaoxia-npm-cache"
+        if ! docker volume inspect "$NPM_CACHE_VOLUME" >/dev/null 2>&1; then
+          docker volume create "$NPM_CACHE_VOLUME" >/dev/null
+        fi
+
+        docker run --rm           -v "$PWD:/workspace"           -v "$NPM_CACHE_VOLUME:/workspace/apps/web/node_modules"           -w /workspace/apps/web           docker.m.daocloud.io/library/node:20           sh -lc "npm ci && npm run build"
+
+        test -f apps/web/dist/index.html
+        echo "Frontend build complete"
+
+    - name: Setup buildx builder (docker-container driver)
+      shell: sh
+      run: |
+        set -eu
+        # 确保使用 docker-container driver 以支持 cache export 功能
+        if ! docker buildx inspect ci-builder > /dev/null 2>&1; then
+          docker buildx create --use --name ci-builder --driver docker-container
+          echo "Created ci-builder (docker-container driver)"
+        else
+          docker buildx use ci-builder
+          echo "Using existing ci-builder"
+        fi
+        docker buildx inspect --bootstrap
+
+    - name: Build and push Web image (buildx cache)
+      shell: sh
+      run: |
+        set -eu
+        REGISTRY="git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas"
+        IMAGE_NAME="xiaoxia-saas-web"
+        VERSION="${GITHUB_REF_NAME}"
+        CACHE_REF="${REGISTRY}/web-cache:main"
+        NGINX_CONF="infra/docker/nginx-production.conf"
+
+        echo "Building Production Web image: ${VERSION}"
+        docker buildx build           --cache-from "type=registry,ref=${CACHE_REF},ignore-error=true"           --cache-to "type=registry,ref=${CACHE_REF},mode=max"           -f infra/docker/web-artifact.Dockerfile           --build-arg "NGINX_CONF=${NGINX_CONF}"           -t "${REGISTRY}/${IMAGE_NAME}:${VERSION}"           --push           .
+        echo "Production Web image pushed: ${REGISTRY}/${IMAGE_NAME}:${VERSION}"
 
     - name: Cleanup old Docker images
       if: always()
@@ -1036,17 +1784,60 @@ jobs:
       run: |
         set +e
         echo "=== CI 失败通知 ==="
-        FAILED_JOB="Build Production Runtime Images" python3 scripts/ci_notify_failure.py
-
+        FAILED_JOB="Build Production Web Image" python3 scripts/ci_notify_failure.py
 
   deploy-production:
     name: Deploy Production
     runs-on: saas
     timeout-minutes: 20
     if: startsWith(github.ref, 'refs/tags/v')
-    needs: build-production-runtime-images
+    needs: [build-production-api, build-production-worker, build-production-web]
 
     steps:
+    - name: Checkout code
+      shell: sh
+      env:
+        GITHUB_TOKEN: ${{ github.token }}
+      run: |
+        set -eu
+        python3 - <<'INNERPY'
+        import io, os, tarfile, time, urllib.request, urllib.error
+        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+        last_err = None
+        for attempt in range(5):
+            try:
+                with urllib.request.urlopen(request, timeout=120) as response:
+                    archive = response.read()
+                break
+            except urllib.error.HTTPError as e:
+                last_err = e
+                if e.code >= 500 and attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+            except Exception as e:
+                last_err = e
+                if attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+        else:
+            raise last_err
+        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+            for member in tar.getmembers():
+                name = member.name
+                if name == root_prefix[:-1]:
+                    continue
+                if name.startswith(root_prefix):
+                    member.name = name[len(root_prefix):]
+                if member.name:
+                    tar.extract(member, '.')
+        INNERPY
     - name: Install SSH client
       shell: sh
       run: |
@@ -1104,6 +1895,52 @@ 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: Production smoke test (健康检查门禁)
+      if: success()
+      shell: sh
+      run: |
+        set -eu
+        API_BASE="https://api.xiaoxiajianji.com"
+        WEB_BASE="https://saas.xiaoxiajianji.com"
+
+        echo "=== 生产部署门禁：外部健康检查 ==="
+        echo "等待服务启动稳定（30s）..."
+        sleep 30
+
+        echo "--- Check 1: API health endpoint ---"
+        for i in $(seq 1 20); do
+          HEALTH=$(curl -sf --max-time 10 "${API_BASE}/health") && break
+          echo "  Attempt $i/20: not ready yet, waiting 5s..."
+          sleep 5
+        done
+        if [ -z "$HEALTH" ]; then
+          echo "FAIL: API /health unreachable after 100s"
+          echo "生产环境健康检查未通过，部署失败！"
+          echo "（回滚机制待接入，当前需手动回滚）"
+          exit 1
+        fi
+        echo "API health OK: $HEALTH"
+
+        echo "--- Check 2: API login endpoint (expect 401/422) ---"
+        HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" --max-time 10 -X POST           "${API_BASE}/api/v1/auth/login"           -H "Content-Type: application/json"           -d '{"email":"smoke@test.com","password":"wrong"}')
+        if [ "$HTTP_CODE" != "401" ] && [ "$HTTP_CODE" != "422" ]; then
+          echo "FAIL: login returned HTTP $HTTP_CODE (expected 401 or 422)"
+          exit 1
+        fi
+        echo "Login API OK: HTTP $HTTP_CODE"
+
+        echo "--- Check 3: Web frontend ---"
+        HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" --max-time 10 "${WEB_BASE}/")
+        if [ "$HTTP_CODE" != "200" ]; then
+          echo "FAIL: web frontend returned HTTP $HTTP_CODE (expected 200)"
+          exit 1
+        fi
+        echo "Web frontend OK: HTTP $HTTP_CODE"
+
+        echo ""
+        echo "=== ✅ 生产环境健康检查全部通过 ==="
+        echo "Version: ${GITHUB_REF_NAME}"
+
     - name: Notify CI success
       if: success()
       shell: sh

apps/api/app/api/routes/templates.py

@@ -45,7 +45,6 @@ from packages.application.template.use_cases import (
     CreateTemplateUseCase,
     DeleteCategoryUseCase,
     DeleteTemplateUseCase,
-    GetTemplateUsageUseCase,
     GetTemplateUseCase,
     ListCategoriesUseCase,
     ListTagsUseCase,

apps/web/e2e/core-generation.spec.ts

@@ -251,6 +251,9 @@ test.describe("Core generation flow", () => {
     await expect(page.locator(".xx-products-page")).toBeVisible({
       timeout: 15_000,
     });
+
+    // 清理所有路由，避免页面关闭时飞地API请求导致测试报错
+    await page.unrouteAll({ behavior: "ignoreErrors" });
   });
 
   test("generation task API creates and lists tasks", async ({ request }) => {

apps/worker/video_processing/concat_engine.py

@@ -0,0 +1,697 @@
+"""视频拼接/合并引擎 — 多段视频按顺序拼接成一个成片.
+
+基于 FFmpeg 实现两种拼接模式：
+1. **concat demuxer（stream copy）**：最快，所有视频编码参数必须一致
+2. **concat filter（重新编码）**：更灵活，支持不同分辨率/编码/帧率的视频
+
+使用场景：
+- 多段素材按顺序合并成一个视频
+- 视频分割后重新拼接
+- 片头 + 正片 + 片尾拼接
+
+降级策略：
+- 优先尝试 stream copy（速度快、无质量损失）
+- 参数不一致时自动降级到 concat filter
+- 某段视频失败时跳过，不阻断整体拼接
+"""
+
+from __future__ import annotations
+
+import logging
+import tempfile
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, probe_video_info, run_ffmpeg
+from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path
+
+logger = logging.getLogger(__name__)
+
+
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+MAX_CONCAT_SEGMENTS = 50  # 最大拼接段数（安全上限，防止OOM）
+
+ALLOWED_VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".flv", ".wmv"}
+
+# concat demuxer 要求一致的参数列表
+CONCAT_DEMUXER_REQUIRED_PARAMS = [
+    "codec_name",  # 视频编码
+    "width",  # 宽度
+    "height",  # 高度
+    "r_frame_rate",  # 帧率
+    "pix_fmt",  # 像素格式
+    "sample_rate",  # 音频采样率
+    "channels",  # 音频声道数
+    "audio_codec",  # 音频编码
+]
+
+
+# ── 拼接片段配置 ──────────────────────────────────────────────────────────────
+
+
+@dataclass
+class ConcatSegment:
+    """单个拼接片段."""
+
+    video_path: str  # 视频文件路径
+    start_time: float = 0.0  # 开始时间（秒），从视频的哪个位置开始取
+    duration: float = 0.0  # 持续时长（秒），0表示取到末尾
+    has_audio: bool = True  # 是否包含音频
+
+    @classmethod
+    def from_dict(cls, seg: dict) -> "ConcatSegment":
+        """从字典创建拼接片段，带安全类型转换."""
+        try:
+            start_time = max(0.0, float(seg.get("start_time", 0.0)))
+        except (TypeError, ValueError):
+            start_time = 0.0
+
+        try:
+            duration = max(0.0, float(seg.get("duration", 0.0)))
+        except (TypeError, ValueError):
+            duration = 0.0
+
+        return cls(
+            video_path=str(seg.get("video_path", "")),
+            start_time=start_time,
+            duration=duration,
+            has_audio=bool(seg.get("has_audio", True)),
+        )
+
+
+@dataclass
+class ConcatConfig:
+    """视频拼接配置."""
+
+    segments: list[ConcatSegment] = field(default_factory=list)
+    output_width: int = 0  # 输出宽度（0=自动取第一段）
+    output_height: int = 0  # 输出高度（0=自动取第一段）
+    output_fps: float = 0.0  # 输出帧率（0=自动取第一段）
+    force_reencode: bool = False  # 强制重新编码（不用 stream copy）
+    transition: str = "none"  # 转场效果（none/crossfade）- 预留
+    transition_duration: float = 0.3  # 转场时长
+
+    @classmethod
+    def from_config_dict(cls, config: dict | None) -> "ConcatConfig":
+        """从配置字典创建 ConcatConfig."""
+        if not config or not isinstance(config, dict):
+            return cls()
+
+        segments_raw = config.get("segments", [])
+        segments: list[ConcatSegment] = []
+
+        if isinstance(segments_raw, list):
+            for s in segments_raw:
+                if isinstance(s, dict) and s.get("video_path"):
+                    try:
+                        seg = ConcatSegment.from_dict(s)
+                        if seg.video_path:
+                            segments.append(seg)
+                    except Exception:
+                        logger.warning("[concat] skip invalid segment: %s", s)
+                        continue
+
+        try:
+            output_width = max(0, int(config.get("output_width", 0)))
+        except (TypeError, ValueError):
+            output_width = 0
+
+        try:
+            output_height = max(0, int(config.get("output_height", 0)))
+        except (TypeError, ValueError):
+            output_height = 0
+
+        try:
+            output_fps = max(0.0, float(config.get("output_fps", 0.0)))
+        except (TypeError, ValueError):
+            output_fps = 0.0
+
+        return cls(
+            segments=segments,
+            output_width=output_width,
+            output_height=output_height,
+            output_fps=output_fps,
+            force_reencode=bool(config.get("force_reencode", False)),
+            transition=str(config.get("transition", "none")),
+            transition_duration=max(0.1, float(config.get("transition_duration", 0.3))),
+        )
+
+    @property
+    def has_effect(self) -> bool:
+        """是否有有效片段需要拼接."""
+        return len([s for s in self.segments if s.video_path]) >= 2
+
+    @property
+    def total_segments(self) -> int:
+        """有效片段数量."""
+        return len([s for s in self.segments if s.video_path])
+
+
+# ── 路径安全校验 ────────────────────────────────────────────────────────────
+
+
+def _validate_video_path(video_path: str, work_dir: Path) -> None:
+    """校验视频文件路径安全性.
+
+    规则：
+    - local:// schema → 必须在 work_dir 内
+    - 相对路径 → 必须在 work_dir 内
+    - 绝对路径 → 必须在允许目录白名单内
+    - 扩展名必须是视频格式
+
+    Raises:
+        PathSecurityError: 路径不安全
+    """
+    if not video_path or not isinstance(video_path, str):
+        raise PathSecurityError("视频路径不能为空")
+
+    # 本地路径（local:// 或相对路径 / 绝对路径）
+    if video_path.startswith("local://") or not video_path.startswith(("http://", "https://", "oss://")):
+        is_abs = video_path.startswith("/") and not video_path.startswith("local://")
+        resolved_path = safe_resolve_path(
+            video_path,
+            work_dir,
+            allow_outside=is_abs,
+            allowed_extensions=ALLOWED_VIDEO_EXTENSIONS,
+        )
+        # 绝对路径额外检查白名单目录（用realpath规范化后的真实路径比较，防止 ../ 遍历绕过）
+        if is_abs:
+            resolved_work_dir = work_dir.resolve()
+            try:
+                resolved_path.relative_to(resolved_work_dir)
+            except ValueError:
+                if not is_in_allowed_dirs(resolved_path):
+                    raise PathSecurityError(f"视频路径不在允许目录内: {video_path[:80]}")
+    # URL类型路径不做本地路径校验（由下载阶段的SSRF防护负责）
+    # 但检查扩展名
+    else:
+        path_part = video_path.split("?")[0].split("#")[0]
+        ext = Path(path_part).suffix.lower()
+        if ext and ext not in ALLOWED_VIDEO_EXTENSIONS:
+            raise PathSecurityError(f"不允许的视频文件类型: {ext}")
+
+
+# ── 视频拼接引擎 ──────────────────────────────────────────────────────────────
+
+
+class ConcatEngine:
+    """视频拼接引擎 — 支持 stream copy 和重新编码两种模式."""
+
+    def __init__(self, work_dir: Path):
+        self.work_dir = work_dir
+        self.work_dir.mkdir(parents=True, exist_ok=True)
+
+    # ── 主入口 ────────────────────────────────────────────────────────
+
+    def concat_videos(
+        self,
+        config: ConcatConfig,
+        output_path: Path,
+    ) -> Path:
+        """拼接多段视频.
+
+        自动选择最优拼接策略：
+        1. 所有片段参数一致 → concat demuxer（stream copy，最快）
+        2. 参数不一致或有裁剪 → concat filter（重新编码）
+
+        Args:
+            config: 拼接配置
+            output_path: 输出文件路径
+
+        Returns:
+            输出文件路径
+        """
+        valid_segments = [s for s in config.segments if s.video_path]
+
+        if not valid_segments:
+            raise ValueError("No valid video segments to concat")
+
+        # ── 安全校验：段数上限 ──
+        if len(valid_segments) > MAX_CONCAT_SEGMENTS:
+            raise ValueError(f"Too many concat segments: {len(valid_segments)} > {MAX_CONCAT_SEGMENTS}")
+
+        # ── 安全校验：所有视频路径白名单校验 ──
+        safe_segments = []
+        for seg in valid_segments:
+            try:
+                _validate_video_path(seg.video_path, self.work_dir)
+                safe_segments.append(seg)
+            except PathSecurityError as e:
+                logger.warning("[concat] skip segment: path security check failed: %s", e)
+
+        if len(safe_segments) != len(valid_segments):
+            valid_segments = safe_segments
+            config.segments = safe_segments
+            logger.info("[concat] %d segments passed security check", len(safe_segments))
+
+        if not valid_segments:
+            raise ValueError("No valid video segments after security check")
+
+        if len(valid_segments) == 1:
+            # 只有一段，直接复制
+            import shutil
+
+            logger.info("[concat] single segment, copy directly")
+            shutil.copy2(valid_segments[0].video_path, output_path)
+            return output_path
+
+        # 判断能否用 stream copy
+        can_stream_copy = self._can_use_stream_copy(config)
+
+        if can_stream_copy and not config.force_reencode:
+            logger.info("[concat] using concat demuxer (stream copy)")
+            try:
+                return self._concat_demuxer(config, output_path)
+            except Exception as e:
+                logger.warning("[concat] demuxer failed, fallback to filter: %s", e)
+
+        # 降级到 concat filter
+        logger.info("[concat] using concat filter (re-encode)")
+        return self._concat_filter(config, output_path)
+
+    # ── 模式判断 ──────────────────────────────────────────────────────
+
+    def _can_use_stream_copy(self, config: ConcatConfig) -> bool:
+        """判断是否可以使用 concat demuxer（stream copy）.
+
+        条件：
+        1. 所有视频编码参数一致（分辨率、帧率、编码、像素格式）
+        2. 所有音频参数一致（采样率、声道、编码）
+        3. 没有设置 start_time 裁剪（或可以通过 concat demuxer 的 inpoint/outpoint 实现）
+        4. 没有强制重新编码
+        """
+        if config.force_reencode:
+            return False
+
+        # 如果有转场效果，必须重新编码
+        if config.transition != "none":
+            return False
+
+        # 探测所有视频的参数
+        video_infos = []
+        for seg in config.segments:
+            if not seg.video_path:
+                continue
+            try:
+                info = probe_video_info(seg.video_path)
+                video_infos.append(info)
+            except Exception:
+                logger.warning("[concat] probe failed for %s", seg.video_path[-40:])
+                return False
+
+        if len(video_infos) < 2:
+            return False
+
+        # 检查参数一致性
+        base_info = video_infos[0]
+        for info in video_infos[1:]:
+            for param in CONCAT_DEMUXER_REQUIRED_PARAMS:
+                base_val = base_info.get(param)
+                curr_val = info.get(param)
+                if base_val != curr_val:
+                    logger.debug(
+                        "[concat] param mismatch: %s (%s vs %s)",
+                        param,
+                        base_val,
+                        curr_val,
+                    )
+                    return False
+
+        # 检查是否有裁剪需求
+        # concat demuxer 支持 inpoint/outpoint，所以有裁剪也可以用
+        # 但为了简单和稳定性，有裁剪时也用 filter 模式
+        # （inpoint/outpoint 不是所有格式都支持得好）
+        has_trimming = any(seg.start_time > 0 or seg.duration > 0 for seg in config.segments if seg.video_path)
+        if has_trimming:
+            return False
+
+        return True
+
+    # ── 模式1：concat demuxer（stream copy） ──────────────────────────
+
+    def _concat_demuxer(self, config: ConcatConfig, output_path: Path) -> Path:
+        """使用 concat demuxer 拼接（stream copy）.
+
+        优点：速度极快，无质量损失
+        缺点：要求所有视频参数完全一致
+        """
+        # 生成 concat 文件列表
+        list_file = self.work_dir / "concat_list.txt"
+        lines = []
+        for seg in config.segments:
+            if not seg.video_path:
+                continue
+            # 路径转义：单引号替换为 '\''
+            safe_path = str(seg.video_path).replace("'", "'\\''")
+            lines.append(f"file '{safe_path}'")
+
+        list_file.write_text("\n".join(lines), encoding="utf-8")
+
+        command = [
+            FFMPEG_BIN,
+            "-y",
+            "-f",
+            "concat",
+            "-safe",
+            "0",
+            "-i",
+            str(list_file),
+            "-c",
+            "copy",
+            "-copyts",
+            str(output_path),
+        ]
+
+        logger.info("[concat] demuxer: %d segments", config.total_segments)
+        run_ffmpeg(command)
+        return output_path
+
+    # ── 模式2：concat filter（重新编码） ──────────────────────────────
+
+    def _concat_filter(self, config: ConcatConfig, output_path: Path) -> Path:
+        """使用 concat filter 拼接（重新编码）.
+
+        优点：支持不同参数的视频，支持裁剪
+        缺点：需要重新编码，较慢
+        """
+        valid_segments = [s for s in config.segments if s.video_path]
+        num_segments = len(valid_segments)
+
+        # 构建输入参数
+        input_args: list[str] = []
+        for seg in valid_segments:
+            input_args.extend(["-i", seg.video_path])
+
+        # 确定输出参数
+        output_width, output_height, output_fps = self._get_output_params(config)
+
+        # 构建 filter_complex
+        filter_parts: list[str] = []
+        concat_inputs = ""
+
+        for i, seg in enumerate(valid_segments):
+            vid_label = f"v{i}"
+            aud_label = f"a{i}"
+
+            seg_filters: list[str] = []
+
+            # 1. 裁剪（start_time + duration）
+            if seg.start_time > 0 or seg.duration > 0:
+                start = seg.start_time
+                if seg.duration > 0:
+                    end = start + seg.duration
+                    seg_filters.append(f"trim=start={start:.3f}:end={end:.3f}")
+                else:
+                    seg_filters.append(f"trim=start={start:.3f}")
+                seg_filters.append("setpts=PTS-STARTPTS")
+
+                # 音频同步裁剪
+                if seg.has_audio:
+                    if seg.duration > 0:
+                        filter_parts.append(
+                            f"[{i}:a]atrim=start={start:.3f}:end={end:.3f}," f"asetpts=PTS-STARTPTS[{aud_label}]"
+                        )
+                    else:
+                        filter_parts.append(f"[{i}:a]atrim=start={start:.3f}," f"asetpts=PTS-STARTPTS[{aud_label}]")
+                else:
+                    # 无音频时生成静音轨
+                    filter_parts.append(
+                        f"[{i}:v]trim=start={start:.3f}," f"setpts=PTS-STARTPTS, " f"aevalsrc=0:d={0.1}[{aud_label}]"
+                    )
+            else:
+                # 无裁剪，直接用原始标签
+                if not seg.has_audio:
+                    # 无音频时需要生成静音
+                    try:
+                        dur = probe_duration(seg.video_path)
+                    except Exception:
+                        dur = 10.0
+                    filter_parts.append(f"aevalsrc=0:d={dur:.3f}:s=44100[{aud_label}]")
+
+            # 2. 缩放/帧率统一
+            vf_parts = []
+            if not seg_filters:
+                vf_parts.append(f"[{i}:v]")
+            else:
+                vf_parts.append("")
+
+            # 分辨率统一
+            if output_width and output_height:
+                vf_parts.append(
+                    f"scale={output_width}:{output_height}:force_original_aspect_ratio=decrease,"
+                    f"pad={output_width}:{output_height}:(ow-iw)/2:(oh-ih)/2:black"
+                )
+
+            # 帧率统一
+            if output_fps > 0:
+                vf_parts.append(f"fps={output_fps}")
+
+            # 像素格式统一
+            vf_parts.append("format=yuv420p")
+
+            if len(vf_parts) > 1 or (seg_filters and vf_parts):
+                if seg_filters:
+                    # 先裁剪后缩放
+                    crop_str = "".join(seg_filters)
+                    scale_str = "".join(vf_parts[1:])  # 跳过空字符串
+                    if scale_str:
+                        filter_parts.append(f"[{i}:v]{crop_str},{scale_str}[{vid_label}]")
+                    else:
+                        filter_parts.append(f"[{i}:v]{crop_str}[{vid_label}]")
+                else:
+                    filter_parts.append(f"{vf_parts[0]}{''.join(vf_parts[1:])}[{vid_label}]")
+            else:
+                if seg_filters:
+                    filter_parts.append(f"[{i}:v]{''.join(seg_filters)}[{vid_label}]")
+                else:
+                    # 什么都不需要，直接用输入
+                    pass
+
+            # 拼接 concat 的输入标签
+            if seg_filters or (output_width and output_height) or output_fps > 0:
+                concat_inputs += f"[{vid_label}]"
+            else:
+                concat_inputs += f"[{i}:v]"
+
+            # 音频标签
+            if seg.start_time > 0 or seg.duration > 0:
+                # 已经生成了 aud_label
+                pass
+            elif not seg.has_audio:
+                # 已经生成了静音 aud_label
+                pass
+            else:
+                # 使用原始音频
+                pass
+
+        # 简化处理：用更直接的方式构建 filter
+        # 重新整理一下，确保所有输入都有对应的 v_i 和 a_i 标签
+        filter_parts.clear()
+        concat_inputs = ""  # 按段交织: [v0][a0][v1][a1]...
+
+        for i, seg in enumerate(valid_segments):
+            v_label = f"v{i}_in"
+            a_label = f"a{i}_in"
+
+            # 视频处理链
+            v_steps: list[str] = [f"[{i}:v]"]
+
+            # 裁剪
+            if seg.start_time > 0 or seg.duration > 0:
+                start = seg.start_time
+                if seg.duration > 0:
+                    end = start + seg.duration
+                    v_steps.append(f"trim=start={start:.3f}:end={end:.3f},")
+                else:
+                    v_steps.append(f"trim=start={start:.3f},")
+                v_steps.append("setpts=PTS-STARTPTS,")
+
+            # 缩放
+            if output_width and output_height:
+                v_steps.append(
+                    f"scale={output_width}:{output_height}:force_original_aspect_ratio=decrease,"
+                    f"pad={output_width}:{output_height}:(ow-iw)/2:(oh-ih)/2:black,"
+                )
+
+            # 帧率
+            if output_fps > 0:
+                v_steps.append(f"fps={output_fps},")
+
+            # 像素格式
+            v_steps.append("format=yuv420p")
+
+            v_filter = "".join(v_steps) + f"[{v_label}]"
+            filter_parts.append(v_filter)
+
+            # 音频处理链
+            a_steps: list[str] = []
+            if seg.has_audio:
+                a_steps.append(f"[{i}:a]")
+
+                if seg.start_time > 0 or seg.duration > 0:
+                    start = seg.start_time
+                    if seg.duration > 0:
+                        end = start + seg.duration
+                        a_steps.append(f"atrim=start={start:.3f}:end={end:.3f},")
+                    else:
+                        a_steps.append(f"atrim=start={start:.3f},")
+                    a_steps.append("asetpts=PTS-STARTPTS,")
+
+                a_steps.append("aformat=sample_fmts=fltp:sample_rates=44100:channel_layouts=stereo")
+            else:
+                # 生成静音音频
+                try:
+                    dur = probe_duration(seg.video_path)
+                except Exception:
+                    dur = 10.0
+                # 减去裁剪
+                if seg.start_time > 0:
+                    dur = max(0.1, dur - seg.start_time)
+                if seg.duration > 0 and seg.duration < dur:
+                    dur = seg.duration
+                a_steps.append(f"aevalsrc=0:d={dur:.3f}:s=44100:c=stereo")
+
+            a_filter = "".join(a_steps) + f"[{a_label}]"
+            filter_parts.append(a_filter)
+
+            # 按段交织排列（v_i, a_i），这是 FFmpeg concat filter 要求的顺序
+            concat_inputs += f"[{v_label}][{a_label}]"
+
+        # concat filter: 输入按 [v0][a0][v1][a1]... 顺序
+        filter_parts.append(f"{concat_inputs}" f"concat=n={num_segments}:v=1:a=1[vout][aout]")
+
+        filter_complex = ";".join(filter_parts)
+
+        command = [
+            FFMPEG_BIN,
+            "-y",
+            *input_args,
+            "-filter_complex",
+            filter_complex,
+            "-map",
+            "[vout]",
+            "-map",
+            "[aout]",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "fast",
+            "-crf",
+            "23",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-movflags",
+            "+faststart",
+            str(output_path),
+        ]
+
+        logger.info(
+            "[concat] filter: %d segments, %dx%d, %.2f fps",
+            num_segments,
+            output_width,
+            output_height,
+            output_fps,
+        )
+        run_ffmpeg(command)
+        return output_path
+
+    # ── 辅助方法 ──────────────────────────────────────────────────────
+
+    def _get_output_params(self, config: ConcatConfig) -> tuple[int, int, float]:
+        """获取输出参数（宽、高、帧率）.
+
+        优先级：
+        1. config 中显式指定的
+        2. 第一段视频的参数
+        """
+        valid_segments = [s for s in config.segments if s.video_path]
+
+        width = config.output_width
+        height = config.output_height
+        fps = config.output_fps
+
+        # 如果没有显式指定，用第一段的参数
+        if (width == 0 or height == 0 or fps == 0) and valid_segments:
+            try:
+                info = probe_video_info(valid_segments[0].video_path)
+                if width == 0:
+                    width = int(info.get("width", 1080))
+                if height == 0:
+                    height = int(info.get("height", 1920))
+                if fps == 0:
+                    fps_str = info.get("r_frame_rate", "30/1")
+                    if "/" in str(fps_str):
+                        num, den = str(fps_str).split("/")
+                        try:
+                            fps = float(num) / float(den)
+                        except (ValueError, ZeroDivisionError):
+                            fps = 30.0
+                    else:
+                        fps = float(fps_str) if fps_str else 30.0
+            except Exception:
+                # 探测失败，用默认值
+                if width == 0:
+                    width = 1080
+                if height == 0:
+                    height = 1920
+                if fps == 0:
+                    fps = 30.0
+
+        return width, height, fps
+
+
+# ── 便捷函数 ──────────────────────────────────────────────────────────────────
+
+
+def concat_video_files(
+    video_paths: list[str],
+    output_path: Path,
+    *,
+    work_dir: Path | None = None,
+    force_reencode: bool = False,
+) -> Path:
+    """简单拼接多个视频文件.
+
+    Args:
+        video_paths: 视频文件路径列表
+        output_path: 输出路径
+        work_dir: 工作目录（默认输出文件所在目录）
+        force_reencode: 是否强制重新编码
+
+    Returns:
+        输出文件路径
+    """
+    if work_dir is None:
+        work_dir = output_path.parent
+
+    segments = [ConcatSegment(video_path=p) for p in video_paths if p]
+    config = ConcatConfig(segments=segments, force_reencode=force_reencode)
+
+    engine = ConcatEngine(work_dir)
+    return engine.concat_videos(config, output_path)
+
+
+def concat_videos_from_config(
+    config_dict: dict | None,
+    output_path: Path,
+    *,
+    work_dir: Path,
+) -> Path | None:
+    """从配置字典执行视频拼接.
+
+    降级策略：配置无效或拼接失败时返回 None.
+    """
+    config = ConcatConfig.from_config_dict(config_dict)
+    if not config.has_effect:
+        return None
+
+    try:
+        engine = ConcatEngine(work_dir)
+        return engine.concat_videos(config, output_path)
+    except Exception as e:
+        logger.error("[concat] concat failed: %s", e)
+        return None

apps/worker/video_processing/ffmpeg_utils.py

@@ -119,6 +119,54 @@ def run_ffmpeg(
         raise
 
 
+def run_ffprobe(
+    command: list[str],
+    *,
+    capture_output: bool = True,
+    timeout: int = 30,
+) -> tuple[str, str]:
+    """执行 FFprobe 命令。
+
+    Args:
+        command: 完整的 ffprobe 命令列表（含 "ffprobe" 本身）
+        capture_output: 是否捕获 stdout/stderr
+        timeout: 超时时间（秒），默认 30s；None 表示不设超时
+
+    Returns:
+        (stdout, stderr) 元组
+
+    Raises:
+        subprocess.CalledProcessError: 命令执行失败时抛出
+        subprocess.TimeoutExpired: 超时未完成时抛出
+    """
+    try:
+        result = subprocess.run(  # nosec B603
+            command,
+            check=True,
+            stdout=subprocess.PIPE if capture_output else None,
+            stderr=subprocess.PIPE if capture_output else None,
+            text=True,
+            timeout=timeout,
+        )
+        return (result.stdout or "", result.stderr or "")
+    except subprocess.TimeoutExpired:
+        logger.error(
+            "FFprobe 命令超时 (%ds): command=%s",
+            timeout or -1,
+            " ".join(str(c) for c in command[:20]),
+        )
+        raise
+    except subprocess.CalledProcessError as e:
+        stderr_text = (e.stderr or "").strip()
+        logger.error(
+            "FFprobe 命令失败: exit_code=%d command=%s\nstderr:\n%s",
+            e.returncode,
+            " ".join(str(c) for c in command[:20]),
+            stderr_text[:5000],
+        )
+        raise
+
+
 def probe_has_audio(local_path: str | Path) -> bool:
     """探测文件是否包含音频流。
 

apps/worker/video_processing/multi_track_mixer.py

@@ -0,0 +1,478 @@
+"""多轨道混音引擎 — 支持多路音频独立音量调节与混合.
+
+基于 FFmpeg amix / amerge 实现：
+- 支持任意数量音频轨道（原音、BGM、配音、音效等）
+- 每轨独立音量调节
+- 每轨独立淡入淡出
+- 每轨独立时间偏移（delay）
+- 总输出音量归一化补偿
+
+作为 render_audio.py 的增强模块，在 mix_audio 后处理阶段被调用。
+与 bgm_mixer.py 的关系：
+- bgm_mixer 专注 BGM 单轨道的复杂处理（循环、人声闪避）
+- 本模块专注多路轨道的统一音量调节与混合
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
+from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path
+
+if TYPE_CHECKING:
+    from video_processing.render_audio import RenderContext
+
+logger = logging.getLogger(__name__)
+
+
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+TRACK_TYPE_MAIN = "main"  # 原音（视频原声）
+TRACK_TYPE_BGM = "bgm"  # 背景音乐
+TRACK_TYPE_VOICEOVER = "voiceover"  # 配音（TTS/人声）
+TRACK_TYPE_SFX = "sfx"  # 音效
+TRACK_TYPE_AMBIENT = "ambient"  # 环境音
+
+MAX_AUDIO_TRACKS = 8  # 最大混音轨道数（安全上限，防止资源耗尽）
+
+# 各轨道默认音量（相对主音频）
+DEFAULT_VOLUMES = {
+    TRACK_TYPE_MAIN: 1.0,
+    TRACK_TYPE_BGM: 0.3,
+    TRACK_TYPE_VOICEOVER: 1.0,
+    TRACK_TYPE_SFX: 0.7,
+    TRACK_TYPE_AMBIENT: 0.2,
+}
+
+
+@dataclass
+class AudioTrack:
+    """单条音频轨道配置."""
+
+    track_id: str  # 轨道唯一标识
+    track_type: str  # 轨道类型（main/bgm/voiceover/sfx/ambient）
+    audio_path: str  # 音频文件路径
+    volume: float = 1.0  # 音量 0.0 ~ 2.0
+    fade_in: float = 0.0  # 淡入时长（秒）
+    fade_out: float = 0.0  # 淡出时长（秒）
+    start_time: float = 0.0  # 开始时间（相对于视频起点，秒）
+    duration: float = 0.0  # 持续时长（0表示到文件末尾）
+    enabled: bool = True  # 是否启用
+
+    @classmethod
+    def from_dict(cls, track: dict) -> "AudioTrack":
+        """从字典创建 AudioTrack，带安全类型转换."""
+        track_type = str(track.get("track_type", TRACK_TYPE_SFX))
+        default_vol = DEFAULT_VOLUMES.get(track_type, 1.0)
+
+        try:
+            volume = float(track.get("volume", default_vol))
+        except (TypeError, ValueError):
+            volume = default_vol
+        volume = max(0.0, min(2.0, volume))
+
+        try:
+            fade_in = max(0.0, float(track.get("fade_in", 0.0)))
+        except (TypeError, ValueError):
+            fade_in = 0.0
+
+        try:
+            fade_out = max(0.0, float(track.get("fade_out", 0.0)))
+        except (TypeError, ValueError):
+            fade_out = 0.0
+
+        try:
+            start_time = max(0.0, float(track.get("start_time", 0.0)))
+        except (TypeError, ValueError):
+            start_time = 0.0
+
+        try:
+            duration = max(0.0, float(track.get("duration", 0.0)))
+        except (TypeError, ValueError):
+            duration = 0.0
+
+        return cls(
+            track_id=str(track.get("track_id", "")),
+            track_type=track_type,
+            audio_path=str(track.get("audio_path", "")),
+            volume=volume,
+            fade_in=fade_in,
+            fade_out=fade_out,
+            start_time=start_time,
+            duration=duration,
+            enabled=bool(track.get("enabled", True)),
+        )
+
+
+@dataclass
+class MultiTrackMixConfig:
+    """多轨道混音配置."""
+
+    tracks: list[AudioTrack] = field(default_factory=list)
+    master_volume: float = 1.0  # 主输出音量
+    normalize: bool = True  # 是否自动归一化补偿
+    max_output_volume: float = 1.5  # 最大输出音量（防止爆音）
+
+    @classmethod
+    def from_config_dict(cls, config: dict | None) -> "MultiTrackMixConfig":
+        """从 plan.config.audio_tracks 字典创建配置."""
+        if not config or not isinstance(config, dict):
+            return cls()
+
+        tracks_raw = config.get("tracks", [])
+        tracks: list[AudioTrack] = []
+
+        if isinstance(tracks_raw, list):
+            for t in tracks_raw:
+                if isinstance(t, dict) and t.get("audio_path"):
+                    try:
+                        track = AudioTrack.from_dict(t)
+                        if track.enabled and track.audio_path:
+                            tracks.append(track)
+                    except Exception:
+                        logger.warning("[multi-track] skip invalid track config: %s", t)
+                        continue
+
+        try:
+            master_volume = float(config.get("master_volume", 1.0))
+            master_volume = max(0.0, min(2.0, master_volume))
+        except (TypeError, ValueError):
+            master_volume = 1.0
+
+        return cls(
+            tracks=tracks,
+            master_volume=master_volume,
+            normalize=bool(config.get("normalize", True)),
+            max_output_volume=float(config.get("max_output_volume", 1.5)),
+        )
+
+    @property
+    def has_effect(self) -> bool:
+        """是否有有效轨道需要混音."""
+        return len([t for t in self.tracks if t.enabled and t.audio_path]) > 0
+
+
+# ── 路径安全校验 ────────────────────────────────────────────────────────────
+
+
+ALLOWED_AUDIO_EXTENSIONS = {".mp3", ".wav", ".aac", ".ogg", ".flac", ".m4a", ".wma"}
+
+
+def _validate_audio_path(audio_path: str, work_dir: Path) -> None:
+    """校验音频文件路径安全性.
+
+    规则：
+    - local:// schema → 必须在 work_dir 内
+    - 相对路径 → 必须在 work_dir 内
+    - 绝对路径 → 必须在允许目录白名单内
+    - 扩展名必须是音频格式
+
+    Raises:
+        PathSecurityError: 路径不安全
+    """
+    if not audio_path or not isinstance(audio_path, str):
+        raise PathSecurityError("音频路径不能为空")
+
+    # 本地路径（local:// 或相对路径）
+    if audio_path.startswith("local://") or not audio_path.startswith(("http://", "https://", "oss://")):
+        is_abs = audio_path.startswith("/") and not audio_path.startswith("local://")
+        resolved_path = safe_resolve_path(
+            audio_path,
+            work_dir,
+            allow_outside=is_abs,
+            allowed_extensions=ALLOWED_AUDIO_EXTENSIONS,
+        )
+        # 绝对路径额外检查白名单目录（用realpath规范化后的真实路径比较，防止 ../ 遍历绕过）
+        if is_abs:
+            resolved_work_dir = work_dir.resolve()
+            try:
+                resolved_path.relative_to(resolved_work_dir)
+            except ValueError:
+                if not is_in_allowed_dirs(resolved_path):
+                    raise PathSecurityError(f"音频路径不在允许目录内: {audio_path[:80]}")
+    # URL类型路径不做本地路径校验（由下载阶段的SSRF防护负责）
+    # 但检查扩展名
+    else:
+        # URL路径，检查扩展名白名单（取 ? 之前的部分）
+        path_part = audio_path.split("?")[0].split("#")[0]
+        from pathlib import Path as _P
+
+        ext = _P(path_part).suffix.lower()
+        if ext and ext not in ALLOWED_AUDIO_EXTENSIONS:
+            raise PathSecurityError(f"不允许的音频文件类型: {ext}")
+
+
+# ── 单轨道预处理 ────────────────────────────────────────────────────────────
+
+
+def _prepare_single_track(
+    ctx: "RenderContext",
+    track: AudioTrack,
+    target_duration: float,
+    output_path: Path,
+) -> bool:
+    """预处理单条轨道：音量 + 淡入淡出 + 时间偏移 + 截断.
+
+    生成一个精确对齐时间轴的音频文件，后续统一 amix 混音。
+
+    Returns:
+        True 表示处理成功，False 表示失败（跳过）
+    """
+    try:
+        audio_dur = probe_duration(track.audio_path)
+    except Exception:
+        logger.warning("[multi-track] probe failed, skip track: %s", track.track_id)
+        return False
+
+    if audio_dur <= 0:
+        return False
+
+    # 计算实际有效时长
+    effective_start = track.start_time
+    if track.duration > 0:
+        effective_dur = min(track.duration, audio_dur)
+    else:
+        effective_dur = audio_dur
+
+    # 如果轨道完全在视频时长之外，跳过
+    if effective_start >= target_duration:
+        return False
+    if effective_start + effective_dur <= 0:
+        return False
+
+    # 构建滤镜链
+    filter_parts: list[str] = []
+
+    # 1. 先截断到有效范围
+    trim_start = 0.0  # 从源文件的哪个位置开始取
+    if effective_start < 0:
+        trim_start = -effective_start
+        effective_start = 0.0
+
+    # 实际需要的源时长
+    need_dur = min(effective_dur, target_duration - effective_start)
+    if need_dur <= 0:
+        return False
+
+    filter_parts.append(f"atrim={trim_start:.3f}:{trim_start + need_dur:.3f}")
+    filter_parts.append("asetpts=N/SR/TB")
+
+    # 2. 音量调节
+    if abs(track.volume - 1.0) > 0.001:
+        filter_parts.append(f"volume={track.volume:.3f}")
+
+    # 3. 淡入
+    if track.fade_in > 0 and track.fade_in < need_dur:
+        filter_parts.append(f"afade=t=in:st=0:d={track.fade_in:.3f}")
+
+    # 4. 淡出
+    if track.fade_out > 0 and track.fade_out < need_dur:
+        fade_start = need_dur - track.fade_out
+        if fade_start > 0:
+            filter_parts.append(f"afade=t=out:st={fade_start:.3f}:d={track.fade_out:.3f}")
+
+    # 5. 时间偏移（用 adelay 实现开头静音填充）
+    if effective_start > 0.01:
+        delay_ms = int(effective_start * 1000)
+        filter_parts.append(f"adelay={delay_ms}|{delay_ms}")
+
+    # 6. 最终截断到目标总时长
+    filter_parts.append(f"atrim=0:{target_duration:.3f}")
+    filter_parts.append("asetpts=N/SR/TB")
+
+    filter_str = ",".join(filter_parts)
+
+    command = [
+        FFMPEG_BIN,
+        "-y",
+        "-i",
+        track.audio_path,
+        "-filter:a",
+        filter_str,
+        "-c:a",
+        "aac",
+        "-b:a",
+        "128k",
+        str(output_path),
+    ]
+
+    logger.info(
+        "[multi-track] prepare track: id=%s type=%s vol=%.2f start=%.2f dur=%.2f",
+        track.track_id,
+        track.track_type,
+        track.volume,
+        effective_start,
+        need_dur,
+    )
+
+    try:
+        run_ffmpeg(command)
+        return True
+    except Exception as e:
+        logger.warning("[multi-track] track prepare failed: %s, error=%s", track.track_id, e)
+        return False
+
+
+# ── 多轨道混音主入口 ─────────────────────────────────────────────────────────
+
+
+def mix_multi_track(
+    ctx: "RenderContext",
+    main_audio_path: Path,
+    config: MultiTrackMixConfig,
+    target_duration: float,
+) -> Path:
+    """多轨道混音：主音频 + 多条附加轨道.
+
+    Args:
+        ctx: 渲染上下文
+        main_audio_path: 主音频文件路径（原音）
+        config: 多轨道混音配置
+        target_duration: 目标总时长
+
+    Returns:
+        混音后的音频文件路径
+    """
+    output_path = ctx.work_dir / f"multi_track_mix_{ctx.plan_id}.aac"
+
+    if target_duration <= 0:
+        target_duration = 5.0
+
+    # ── 安全校验：轨道数量上限 ──
+    enabled_tracks = [t for t in config.tracks if t.enabled and t.audio_path]
+    if len(enabled_tracks) > MAX_AUDIO_TRACKS:
+        logger.warning(
+            "[multi-track] too many tracks: %d > %d, truncating to max",
+            len(enabled_tracks),
+            MAX_AUDIO_TRACKS,
+        )
+        enabled_tracks = enabled_tracks[:MAX_AUDIO_TRACKS]
+        # 更新 config.tracks 为截断后的列表
+        config.tracks = enabled_tracks
+
+    # ── 安全校验：所有音频路径白名单校验 ──
+    # 主音频路径
+    try:
+        _validate_audio_path(str(main_audio_path), ctx.work_dir)
+    except PathSecurityError as e:
+        logger.error("[multi-track] main audio path security check failed: %s", e)
+        raise
+
+    # 各轨道音频路径
+    valid_tracks = []
+    for track in enabled_tracks:
+        try:
+            _validate_audio_path(track.audio_path, ctx.work_dir)
+            valid_tracks.append(track)
+        except PathSecurityError as e:
+            logger.warning("[multi-track] skip track %s: path security check failed: %s", track.track_id, e)
+
+    if len(valid_tracks) != len(enabled_tracks):
+        config.tracks = valid_tracks
+        logger.info("[multi-track] %d tracks passed security check", len(valid_tracks))
+
+    # 收集所有有效轨道（已预处理好的）
+    prepared_tracks: list[Path] = []
+
+    # 主音频作为第0轨
+    prepared_tracks.append(main_audio_path)
+
+    # 预处理每条附加轨道
+    for i, track in enumerate(config.tracks):
+        if not track.enabled or not track.audio_path:
+            continue
+
+        track_out = ctx.work_dir / f"track_{i}_{ctx.plan_id}.aac"
+        if _prepare_single_track(ctx, track, target_duration, track_out):
+            prepared_tracks.append(track_out)
+
+    # 如果只有主音频，直接返回（无需混音）
+    if len(prepared_tracks) <= 1:
+        import shutil
+
+        shutil.copy2(main_audio_path, output_path)
+        return output_path
+
+    # 使用 amix 混音
+    num_inputs = len(prepared_tracks)
+
+    # 构建输入参数
+    input_args: list[str] = []
+    for tp in prepared_tracks:
+        input_args.extend(["-i", str(tp)])
+
+    # amix 的 duration=first 以第一个输入（主音频）时长为准
+    # normalize 补偿：amix 会把每路音量除以 N，需要乘回来
+    # 但如果所有轨道都同时有声，可能会爆音，所以用 master_volume 控制
+    if config.normalize:
+        # 经验值：不是所有轨道都同时有声，补偿系数取 N * 0.7
+        compensate = num_inputs * 0.7
+    else:
+        compensate = 1.0
+
+    final_volume = compensate * config.master_volume
+    final_volume = min(final_volume, config.max_output_volume)
+
+    # 构建 filter_complex
+    inputs_label = "".join(f"[{i}:a]" for i in range(num_inputs))
+    filter_complex = (
+        f"{inputs_label}amix=inputs={num_inputs}:duration=first:dropout_transition=0[outa];"
+        f"[outa]volume={final_volume:.3f}[final]"
+    )
+
+    command = [
+        FFMPEG_BIN,
+        "-y",
+        *input_args,
+        "-filter_complex",
+        filter_complex,
+        "-map",
+        "[final]",
+        "-c:a",
+        "aac",
+        "-b:a",
+        "128k",
+        str(output_path),
+    ]
+
+    logger.info(
+        "[multi-track] mix %d tracks, master_vol=%.2f compensate=%.2f final_vol=%.2f",
+        num_inputs,
+        config.master_volume,
+        compensate,
+        final_volume,
+    )
+
+    try:
+        run_ffmpeg(command)
+    except Exception as e:
+        logger.error("[multi-track] mix failed, fallback to main audio only: %s", e)
+        import shutil
+
+        shutil.copy2(main_audio_path, output_path)
+
+    return output_path
+
+
+# ── 便捷函数：从 plan.config 快速混音 ───────────────────────────────────────
+
+
+def mix_audio_tracks_from_config(
+    ctx: "RenderContext",
+    main_audio_path: Path,
+    audio_tracks_config: dict | None,
+    target_duration: float,
+) -> Path:
+    """从 plan.config.audio_tracks 配置执行多轨道混音.
+
+    降级策略：配置无效或混音失败时返回主音频。
+    """
+    config = MultiTrackMixConfig.from_config_dict(audio_tracks_config)
+    if not config.has_effect:
+        return main_audio_path
+
+    return mix_multi_track(ctx, main_audio_path, config, target_duration)

apps/worker/video_processing/oss_helpers.py

@@ -220,25 +220,64 @@ def resolve_asset_path(asset_id: str, work_dir: Path) -> Path | None:
     """从 asset_id 解析到本地文件路径。
 
     策略（按优先级）：
-    1. 如果 asset_id 是本地绝对路径（/var/storage/...）→ 直接返回
+    1. 如果 asset_id 是本地绝对路径（/var/storage/...）→ 安全校验后返回
     2. 如果 work_dir 下已有缓存文件 → 返回缓存路径
     3. 从 OSS 下载到 work_dir/{hash}.mp4 → 返回下载路径
     4. 下载失败 → 返回 None
 
     缓存策略：以 asset_id 的 SHA256 前 16 位为文件名，避免重复下载。
-    """
-    # 1. 本地绝对路径
-    if asset_id.startswith("/") and os.path.exists(asset_id):
-        return Path(asset_id)
 
-    # 2. 缓存命中
+    安全：
+    - 本地绝对路径必须在 ASSET_ALLOWED_DIRS 环境变量指定的目录内
+    - 文件名经过 sanitize，防止路径遍历
+    - 禁止空字节、控制字符
+    """
+    from video_processing.path_security import (
+        PathSecurityError,
+        get_allowed_local_dirs,
+        is_in_allowed_dirs,
+        sanitize_filename,
+    )
+
+    if not asset_id or not isinstance(asset_id, str):
+        return None
+
+    # 空字节检测
+    if "\x00" in asset_id:
+        logger.warning("asset_id 包含空字节，拒绝: %s", asset_id[:50])
+        return None
+
+    # 1. 本地绝对路径 — 必须在允许的目录内
+    if asset_id.startswith("/") and os.path.exists(asset_id):
+        try:
+            resolved = Path(asset_id).resolve()
+            if is_in_allowed_dirs(resolved, get_allowed_local_dirs()):
+                return resolved
+            else:
+                logger.warning(
+                    "本地素材路径不在允许目录内，拒绝: %s (allowed=%s)",
+                    asset_id[:80],
+                    get_allowed_local_dirs(),
+                )
+                return None
+        except (OSError, PathSecurityError):
+            return None
+
+    # 2. 缓存命中（使用 hash 而非原始 ID，防止路径遍历）
     cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16]
-    cached_path = work_dir / f"{cache_hash}.mp4"
+    safe_name = sanitize_filename(cache_hash)
+    cached_path = work_dir / f"{safe_name}.mp4"
     if cached_path.exists() and cached_path.stat().st_size > 0:
         return cached_path
 
-    # 3. 从 OSS 下载
-    if download_asset(asset_id, cached_path):
+    # 3. 从 OSS 下载（先标准化 key，防止路径遍历注入）
+    safe_key = normalize_storage_key(asset_id)
+    # 额外校验：存储键不能包含 ../ 或绝对路径
+    if ".." in safe_key or safe_key.startswith("/"):
+        logger.warning("asset_id 包含路径遍历模式，拒绝下载: %s", asset_id[:80])
+        return None
+
+    if download_asset(safe_key, cached_path):
         return cached_path
 
     return None

apps/worker/video_processing/path_security.py

@@ -0,0 +1,303 @@
+"""路径安全校验工具 — 路径遍历防护.
+
+统一的文件路径安全校验方案，覆盖所有渲染管线中的路径处理场景：
+- 本地素材路径校验
+- local:// 路径 schema 校验
+- 工作目录内路径安全约束
+- 防止路径遍历攻击 (../)
+
+防护要点：
+1. 所有用户可控路径必须在允许的目录内
+2. 解析符号链接后的真实路径仍需在允许目录内
+3. 禁止空路径、相对路径遍历、绝对路径逃逸
+4. 路径字符限制与规范化
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+# 最大路径长度
+MAX_PATH_LENGTH = 4096
+
+# 允许的文件扩展名（渲染相关）
+ALLOWED_MEDIA_EXTENSIONS = {
+    ".mp4",
+    ".mov",
+    ".avi",
+    ".mkv",
+    ".webm",
+    ".flv",
+    ".wmv",  # 视频
+    ".mp3",
+    ".wav",
+    ".aac",
+    ".ogg",
+    ".flac",
+    ".m4a",
+    ".wma",  # 音频
+    ".jpg",
+    ".jpeg",
+    ".png",
+    ".gif",
+    ".bmp",
+    ".webp",
+    ".tiff",  # 图片
+    ".srt",
+    ".ass",
+    ".vtt",
+    ".sub",  # 字幕
+    ".txt",
+    ".json",  # 文本/配置
+}
+
+# local:// schema 前缀
+LOCAL_SCHEMA_PREFIX = "local://"
+
+
+class PathSecurityError(ValueError):
+    """路径安全校验失败."""
+
+    pass
+
+
+def safe_resolve_path(
+    input_path: str | Path,
+    base_dir: str | Path,
+    *,
+    allow_outside: bool = False,
+    allowed_extensions: set[str] | None = None,
+) -> Path:
+    """安全解析路径，确保最终路径在 base_dir 内.
+
+    Args:
+        input_path: 输入路径（相对或绝对）
+        base_dir: 基路径目录，解析后的路径必须在此目录内
+        allow_outside: 是否允许路径在 base_dir 外（默认禁止）
+        allowed_extensions: 允许的文件扩展名集合（None 表示不限制）
+
+    Returns:
+        解析后的绝对路径 Path 对象
+
+    Raises:
+        PathSecurityError: 路径不安全
+    """
+    if input_path is None:
+        raise PathSecurityError("路径不能为空")
+
+    path_str = str(input_path).strip()
+    if not path_str:
+        raise PathSecurityError("路径不能为空")
+
+    if len(path_str) > MAX_PATH_LENGTH:
+        raise PathSecurityError(f"路径过长 ({len(path_str)} > {MAX_PATH_LENGTH})")
+
+    # 空字节检测（必须在 Path() 之前）
+    if "\x00" in path_str:
+        raise PathSecurityError("路径包含空字节")
+
+    # 处理 local:// schema
+    if path_str.startswith(LOCAL_SCHEMA_PREFIX):
+        path_str = path_str[len(LOCAL_SCHEMA_PREFIX) :]
+        # local:// 后必须是相对路径（相对于 base_dir），不能是绝对路径
+        if os.path.isabs(path_str):
+            raise PathSecurityError("local:// 路径不能是绝对路径")
+
+    # 规范化 base_dir
+    base_dir = Path(base_dir).resolve()
+    if not base_dir.is_dir():
+        raise PathSecurityError(f"基路径不是有效目录: {base_dir}")
+
+    # 解析输入路径
+    input_path_obj = Path(path_str)
+
+    # 如果是绝对路径且不允许外部路径
+    if input_path_obj.is_absolute() and not allow_outside:
+        raise PathSecurityError("禁止使用绝对路径（需在工作目录内）")
+
+    # 组合并解析为绝对路径
+    if input_path_obj.is_absolute():
+        full_path = input_path_obj.resolve()
+    else:
+        full_path = (base_dir / input_path_obj).resolve()
+
+    # 检查路径遍历 — 确保最终路径在 base_dir 内
+    if not allow_outside:
+        try:
+            full_path.relative_to(base_dir)
+        except ValueError:
+            raise PathSecurityError(f"路径遍历检测：路径 '{path_str}' 超出基路径 '{base_dir}' 范围")
+
+    # 扩展名校验
+    if allowed_extensions is not None:
+        ext = full_path.suffix.lower()
+        if ext and ext not in allowed_extensions:
+            raise PathSecurityError(f"不允许的文件类型: {ext}")
+
+    # 检查危险路径模式
+    _check_dangerous_patterns(full_path)
+
+    return full_path
+
+
+def _check_dangerous_patterns(path: Path) -> None:
+    """检查危险路径模式."""
+    path_str = str(path)
+
+    # 检查空字节
+    if "\x00" in path_str:
+        raise PathSecurityError("路径包含空字节")
+
+    # 检查特殊设备文件（Linux）
+    dangerous_prefixes = [
+        "/proc/",
+        "/sys/",
+        "/dev/",
+        "/etc/passwd",
+        "/etc/shadow",
+        "/root/",
+        "/boot/",
+        "/var/run/",
+    ]
+    for prefix in dangerous_prefixes:
+        if path_str.startswith(prefix):
+            raise PathSecurityError(f"禁止访问系统路径: {prefix}")
+
+
+def is_path_safe(
+    input_path: str | Path,
+    base_dir: str | Path,
+    *,
+    allow_outside: bool = False,
+) -> bool:
+    """便捷函数：检查路径是否安全，不抛异常."""
+    try:
+        safe_resolve_path(input_path, base_dir, allow_outside=allow_outside)
+        return True
+    except PathSecurityError:
+        return False
+
+
+def validate_local_schema_path(
+    schema_path: str,
+    work_dir: str | Path,
+) -> Path:
+    """校验 local:// schema 路径，返回安全的本地路径.
+
+    local:// 路径规则：
+    - 必须以 local:// 开头
+    - 后面必须是相对路径
+    - 最终解析后必须在 work_dir 内
+    - 不允许 ../ 遍历
+
+    Args:
+        schema_path: local:// 开头的路径
+        work_dir: 工作目录
+
+    Returns:
+        解析后的安全路径
+
+    Raises:
+        PathSecurityError: 路径不安全
+    """
+    if not schema_path.startswith(LOCAL_SCHEMA_PREFIX):
+        raise PathSecurityError(f"路径必须以 {LOCAL_SCHEMA_PREFIX} 开头")
+
+    return safe_resolve_path(schema_path, work_dir, allow_outside=False)
+
+
+def sanitize_filename(filename: str) -> str:
+    """清理文件名，移除危险字符.
+
+    保留：字母、数字、下划线、连字符、点、中文字符
+    移除：路径分隔符、控制字符、特殊符号等
+    """
+    import re
+
+    if not filename:
+        return "unnamed"
+
+    # 移除路径分隔符和危险字符
+    # 保留: 字母数字、中文字符、下划线、连字符、点、空格
+    sanitized = re.sub(r'[\\/\x00-\x1f\x7f<>:"|?*]', "_", filename)
+
+    # 移除开头的点和连续的点（防止隐藏文件和路径遍历）
+    while sanitized.startswith("."):
+        sanitized = sanitized[1:]
+
+    # 限制长度
+    if len(sanitized) > 255:
+        name, ext = os.path.splitext(sanitized)
+        sanitized = name[: 255 - len(ext)] + ext
+
+    # 空文件名兜底
+    if not sanitized or sanitized == ".":
+        sanitized = "unnamed"
+
+    return sanitized
+
+
+# ── 允许目录配置 ──────────────────────────────────────────────────────────────
+
+
+def get_allowed_local_dirs() -> list[Path]:
+    """获取允许的本地素材目录列表（从环境变量读取）.
+
+    环境变量 ASSET_ALLOWED_DIRS，多个目录用冒号分隔（Linux）或分号分隔（Windows）。
+    默认包含 /tmp。
+
+    用于：
+    - resolve_asset_path 本地绝对路径白名单
+    - PiP local_path 类型白名单
+    - 贴纸本地路径白名单
+    """
+    env_dirs = os.environ.get("ASSET_ALLOWED_DIRS", "")
+    dirs: list[Path] = []
+    if env_dirs:
+        import re
+
+        sep = ";" if os.name == "nt" else ":"
+        for d in re.split(f"[{sep}]", env_dirs):
+            d = d.strip()
+            if d:
+                try:
+                    dirs.append(Path(d).resolve())
+                except OSError:
+                    pass
+    # 默认允许 /tmp
+    if not dirs:
+        try:
+            dirs.append(Path("/tmp").resolve())  # nosec B108
+        except OSError:
+            pass
+    return dirs
+
+
+def is_in_allowed_dirs(path: str | Path, allowed_dirs: list[Path] | None = None) -> bool:
+    """检查路径是否在允许的目录列表内.
+
+    Args:
+        path: 待检查的路径
+        allowed_dirs: 允许的目录列表，None 则使用默认配置
+
+    Returns:
+        True 表示在允许目录内
+    """
+    if allowed_dirs is None:
+        allowed_dirs = get_allowed_local_dirs()
+
+    try:
+        resolved = Path(path).resolve()
+        for allowed in allowed_dirs:
+            try:
+                resolved.relative_to(allowed)
+                return True
+            except ValueError:
+                continue
+        return False
+    except OSError:
+        return False

apps/worker/video_processing/pip_engine.py

@@ -465,18 +465,44 @@ class PiPEngine:
         layer: PiPLayerConfig,
         asset_path_map: dict[str, Path],
     ) -> Path | None:
-        """验证图层素材是否可用，返回本地路径或None（降级跳过）."""
+        """验证图层素材是否可用，返回本地路径或None（降级跳过）.
+
+        安全：
+        - local_path 类型：必须在允许的目录内，防止路径遍历
+        - url 类型：必须通过 SSRF 安全校验
+        """
+        from video_processing.path_security import is_in_allowed_dirs
+        from video_processing.url_security import UrlSecurityError, validate_url_safety
+
         try:
             if layer.source_type == "local_path":
-                path = Path(layer.source)
-                if path.exists():
-                    return path
+                if not layer.source:
+                    return None
+                # 路径安全校验：必须在允许目录内
+                src_path = Path(layer.source)
+                if not src_path.exists():
+                    return None
+                if not is_in_allowed_dirs(src_path):
+                    logger.warning(
+                        "PiP local_path 不在允许目录内，拒绝: %s",
+                        layer.source[:80],
+                    )
+                    return None
+                return src_path.resolve()
             elif layer.source_type == "asset_id":
                 if layer.source in asset_path_map:
                     return asset_path_map[layer.source]
+                return None
             elif layer.source_type == "url":
-                # URL类型由调用者负责下载，这里返回标记
-                return None  # 暂时不支持直接URL
+                # URL类型：先做SSRF安全校验，由调用者负责实际下载
+                try:
+                    validate_url_safety(layer.source, purpose="pip_source")
+                    logger.info("PiP URL 安全校验通过: %s", layer.source[:80])
+                except UrlSecurityError as e:
+                    logger.warning("PiP URL 安全校验失败: %s (error=%s)", layer.source[:80], e)
+                    return None
+                # 暂时不支持直接URL下载，返回None表示降级跳过
+                return None
         except Exception as e:
             logger.warning("PiP素材验证失败: %s", e)
 

apps/worker/video_processing/render_audio.py

@@ -77,6 +77,7 @@ def mix_audio(
     *,
     bgm_path: str | None = None,
     bgm_config: dict | None = None,
+    audio_tracks_config: dict | None = None,
 ) -> Path | None:
     """音频后处理混音.
 
@@ -87,6 +88,8 @@ def mix_audio(
     4. 输出时长截断到 video_duration
     5. 无音频流的 clip 会被自动跳过，避免 FFmpeg 引用 [i:a] 失败
     6. 如果提供了 bgm_path，则额外混入 BGM（支持淡入淡出、循环、人声闪避）
+    7. 如果配置了 audio_tracks，则混入多轨道音频（配音、音效等）
+    8. 如果配置了降噪，最后应用降噪
 
     Args:
         ctx: 渲染上下文
@@ -94,6 +97,7 @@ def mix_audio(
         video_duration: 视频总时长（用于截断音频）
         bgm_path: BGM 音频本地路径，为 None 时不混入 BGM
         bgm_config: BGM 配置字典（volume/fade_in/fade_out/sidechain 等）
+        audio_tracks_config: 多轨道音频配置（tracks/master_volume 等）
 
     Returns:
         混音后的音频文件路径，无音频时返回 None
@@ -157,10 +161,21 @@ def mix_audio(
         try:
             # 这里 main_audio 就是 output_path，先有主音频再混 BGM
             final_path = mix_bgm_with_main(ctx, output_path, bgm_cfg, video_duration)
-            return _apply_noise_reduction_if_needed(ctx, final_path)
+            output_path = final_path
         except Exception:
             logger.exception("[bgm] BGM 混音失败，回退到无 BGM 音频: plan_id=%s", ctx.plan_id)
-            return _apply_noise_reduction_if_needed(ctx, output_path)
+
+    # ── 多轨道混音（配音/音效等） ──
+    if audio_tracks_config and audio_tracks_config.get("enabled", False):
+        from video_processing.multi_track_mixer import mix_audio_tracks_from_config
+
+        try:
+            tracks_config = audio_tracks_config.get("tracks_config") or audio_tracks_config
+            multi_output = mix_audio_tracks_from_config(ctx, output_path, tracks_config, video_duration)
+            if multi_output and multi_output != output_path:
+                output_path = multi_output
+        except Exception:
+            logger.exception("[multi-track] 多轨道混音失败，回退: plan_id=%s", ctx.plan_id)
 
     return _apply_noise_reduction_if_needed(ctx, output_path)
 

apps/worker/video_processing/sticker_engine.py

@@ -392,10 +392,48 @@ class StickerEngine:
                     )
                     parsed_stickers.append((z, config))
                 else:
-                    # 图片贴纸
-                    image_path = s.get("image_path", "") or s.get("image_url", "")
-                    if not image_path or not Path(image_path).exists():
-                        logger.warning("贴纸素材不存在，跳过: %s", image_path)
+                    # 图片贴纸 — 安全校验：区分本地路径和URL
+                    image_path = s.get("image_path", "")
+                    image_url = s.get("image_url", "")
+
+                    safe_image_path: Path | None = None
+
+                    if image_path:
+                        # 本地路径：路径遍历防护
+                        from video_processing.path_security import is_in_allowed_dirs
+
+                        try:
+                            p = Path(image_path)
+                            if not p.exists():
+                                logger.warning("贴纸素材不存在，跳过: %s", image_path[:80])
+                                continue
+                            if not is_in_allowed_dirs(p):
+                                logger.warning("贴纸路径不在允许目录内，拒绝: %s", image_path[:80])
+                                continue
+                            safe_image_path = p.resolve()
+                        except Exception as e:
+                            logger.warning("贴纸路径校验失败，跳过: %s error=%s", image_path[:80], e)
+                            continue
+                    elif image_url:
+                        # URL：SSRF 安全校验（暂不自动下载，仅校验安全性）
+                        from video_processing.url_security import (
+                            UrlSecurityError,
+                            validate_url_safety,
+                        )
+
+                        try:
+                            validate_url_safety(image_url, purpose="sticker_image")
+                        except UrlSecurityError as e:
+                            logger.warning("贴纸URL安全校验失败，跳过: %s error=%s", image_url[:80], e)
+                            continue
+                        # URL 类型暂不支持自动下载，跳过
+                        logger.info("贴纸URL类型暂不支持自动下载，跳过: %s", image_url[:80])
+                        continue
+                    else:
+                        logger.warning("贴纸缺少 image_path 和 image_url，跳过")
+                        continue
+
+                    if safe_image_path is None:
                         continue
 
                     config = ImageStickerConfig(
@@ -414,11 +452,11 @@ class StickerEngine:
                         fade_in=float(s.get("fade_in", 0)),
                         fade_out=float(s.get("fade_out", 0)),
                         z_index=z,
-                        image_url=str(s.get("image_url", "")),
+                        image_url=image_url,
                     )
                     parsed_stickers.append((z, config))
                     image_stickers.append(config)
-                    image_paths.append(image_path)
+                    image_paths.append(str(safe_image_path))
 
             except Exception as e:
                 logger.warning("贴纸配置解析失败，跳过: %s", e)

apps/worker/video_processing/subtitle_render_engine.py

@@ -0,0 +1,687 @@
+"""字幕渲染引擎 — 统一管理字幕样式配置与视频烧录.
+
+与现有模块的关系：
+- render_subtitles.py：生成静态整段标题/字幕的 ASS 文件
+- subtitle_generator.py：从 ASR 时间轴生成 ASS 文件
+- 本模块：统一的字幕样式配置 + 烧录滤镜生成 + 多源字幕合并
+
+支持的字幕来源：
+1. 静态标题/字幕（title_config / subtitle_config）
+2. ASR 自动字幕（asr_subtitle_timeline）
+3. 手动字幕（manual_subtitles 时间轴）
+
+支持的样式配置：
+- 字体、字号、颜色
+- 描边（颜色、宽度）
+- 阴影（偏移、模糊、颜色）
+- 背景框（颜色、透明度、圆角、边距）
+- 位置（9宫格 + 自定义坐标）
+- 对齐方式
+- 动画（淡入淡出、滑入滑出、打字机）
+- 多行/换行规则
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path
+from video_processing.render_subtitles import generate_ass_subtitles
+from video_processing.subtitle_generator import generate_ass_from_timeline
+
+logger = logging.getLogger(__name__)
+
+
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+ALLOWED_SUBTITLE_EXTENSIONS = {".srt", ".ass", ".vtt", ".sub"}
+
+# 9宫格位置映射（ASS alignment 编号）
+POSITION_ALIGNMENT = {
+    "top_left": 7,
+    "top_center": 8,
+    "top_right": 9,
+    "middle_left": 4,
+    "center": 5,
+    "middle_right": 6,
+    "bottom_left": 1,
+    "bottom_center": 2,
+    "bottom_right": 3,
+}
+
+# 位置简称兼容
+POSITION_ALIASES = {
+    "top": "top_center",
+    "bottom": "bottom_center",
+    "middle": "center",
+    "left": "middle_left",
+    "right": "middle_right",
+}
+
+DEFAULT_FONT = "思源黑体"
+DEFAULT_FONT_SIZE = 24
+DEFAULT_COLOR = "#FFFFFF"
+DEFAULT_STROKE_COLOR = "#000000"
+DEFAULT_STROKE_WIDTH = 1.5
+DEFAULT_POSITION = "bottom_center"
+DEFAULT_MAX_CHARS_PER_LINE = 20
+
+
+# ── 字幕样式配置 ────────────────────────────────────────────────────────────
+
+
+@dataclass
+class SubtitleStyle:
+    """字幕样式配置."""
+
+    font_name: str = DEFAULT_FONT
+    font_size: int = DEFAULT_FONT_SIZE
+    font_color: str = DEFAULT_COLOR
+    bold: bool = False
+    italic: bool = False
+
+    # 描边
+    stroke_enabled: bool = True
+    stroke_color: str = DEFAULT_STROKE_COLOR
+    stroke_width: float = DEFAULT_STROKE_WIDTH
+
+    # 阴影
+    shadow_enabled: bool = False
+    shadow_color: str = "#000000"
+    shadow_offset_x: int = 2
+    shadow_offset_y: int = 2
+    shadow_blur: float = 0.0
+
+    # 背景框
+    background_enabled: bool = False
+    background_color: str = "#000000"
+    background_opacity: float = 0.5  # 0.0 ~ 1.0
+    background_padding: int = 8
+    background_radius: int = 4
+
+    # 位置
+    position: str = DEFAULT_POSITION  # 9宫格位置名
+    margin_v: int = 60  # 垂直边距
+    margin_l: int = 40  # 左边距
+    margin_r: int = 40  # 右边距
+
+    # 多行
+    max_chars_per_line: int = DEFAULT_MAX_CHARS_PER_LINE
+    line_spacing: int = 0  # 行间距
+
+    # 动画
+    fade_in: float = 0.0  # 淡入时长（秒）
+    fade_out: float = 0.0  # 淡出时长（秒）
+    animation_type: str = "none"  # none/fade/slide/typewriter
+
+    @classmethod
+    def from_dict(cls, config: dict[str, Any] | None) -> "SubtitleStyle":
+        """从字典创建样式配置，带安全类型转换."""
+        if not config or not isinstance(config, dict):
+            return cls()
+
+        def safe_str(key: str, default: str) -> str:
+            val = config.get(key, default)
+            return str(val) if val is not None else default
+
+        def safe_int(key: str, default: int) -> int:
+            try:
+                return int(config.get(key, default))
+            except (TypeError, ValueError):
+                return default
+
+        def safe_float(key: str, default: float) -> float:
+            try:
+                return float(config.get(key, default))
+            except (TypeError, ValueError):
+                return default
+
+        def safe_bool(key: str, default: bool) -> bool:
+            return bool(config.get(key, default))
+
+        position = safe_str("position", DEFAULT_POSITION)
+        position = POSITION_ALIASES.get(position, position)
+        if position not in POSITION_ALIGNMENT:
+            position = DEFAULT_POSITION
+
+        return cls(
+            font_name=safe_str("font", DEFAULT_FONT),
+            font_size=safe_int("size", DEFAULT_FONT_SIZE),
+            font_color=safe_str("color", DEFAULT_COLOR),
+            bold=safe_bool("bold", False),
+            italic=safe_bool("italic", False),
+            stroke_enabled=safe_bool("stroke_enabled", True),
+            stroke_color=safe_str("stroke_color", DEFAULT_STROKE_COLOR),
+            stroke_width=safe_float("stroke_width", DEFAULT_STROKE_WIDTH),
+            shadow_enabled=safe_bool("shadow_enabled", False),
+            shadow_color=safe_str("shadow_color", "#000000"),
+            shadow_offset_x=safe_int("shadow_offset_x", 2),
+            shadow_offset_y=safe_int("shadow_offset_y", 2),
+            shadow_blur=safe_float("shadow_blur", 0.0),
+            background_enabled=safe_bool("background_enabled", False),
+            background_color=safe_str("background_color", "#000000"),
+            background_opacity=max(0.0, min(1.0, safe_float("background_opacity", 0.5))),
+            background_padding=safe_int("background_padding", 8),
+            background_radius=safe_int("background_radius", 4),
+            position=position,
+            margin_v=safe_int("margin_v", 60),
+            margin_l=safe_int("margin_l", 40),
+            margin_r=safe_int("margin_r", 40),
+            max_chars_per_line=safe_int("max_chars_per_line", DEFAULT_MAX_CHARS_PER_LINE),
+            line_spacing=safe_int("line_spacing", 0),
+            fade_in=max(0.0, safe_float("fade_in", 0.0)),
+            fade_out=max(0.0, safe_float("fade_out", 0.0)),
+            animation_type=safe_str("animation_type", "none"),
+        )
+
+    @property
+    def alignment(self) -> int:
+        """获取 ASS alignment 编号."""
+        return POSITION_ALIGNMENT.get(self.position, 2)
+
+    @property
+    def ass_font_color(self) -> str:
+        """ASS 格式颜色 &HAABBGGRR."""
+        return _hex_to_ass_color(self.font_color)
+
+    @property
+    def ass_stroke_color(self) -> str:
+        return _hex_to_ass_color(self.stroke_color)
+
+    @property
+    def ass_shadow_color(self) -> str:
+        return _hex_to_ass_color(self.shadow_color)
+
+    @property
+    def ass_background_color(self) -> str:
+        """背景框颜色（ASS BackColour），带透明度."""
+        alpha_hex = _opacity_to_ass_alpha(self.background_opacity)
+        color_bgr = _hex_to_ass_bgr(self.background_color)
+        return f"&H{alpha_hex}{color_bgr}"
+
+
+# ── 工具函数 ──────────────────────────────────────────────────────────────────
+
+
+def _hex_to_ass_color(hex_color: str) -> str:
+    """HEX → ASS 颜色 &HAABBGGRR（默认不透明）."""
+    hex_color = hex_color.lstrip("#")
+    if len(hex_color) != 6:
+        return "&H00FFFFFF"
+    r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6]
+    return f"&H00{b.upper()}{g.upper()}{r.upper()}"
+
+
+def _hex_to_ass_bgr(hex_color: str) -> str:
+    """HEX → ASS BGR 部分（不含 alpha）."""
+    hex_color = hex_color.lstrip("#")
+    if len(hex_color) != 6:
+        return "FFFFFF"
+    r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6]
+    return f"{b.upper()}{g.upper()}{r.upper()}"
+
+
+def _opacity_to_ass_alpha(opacity: float) -> str:
+    """不透明度 → ASS alpha（00=不透明，FF=完全透明）."""
+    alpha = 255 - int(opacity * 255)
+    return f"{alpha:02X}"
+
+
+def _escape_ass_text(text: str) -> str:
+    """转义 ASS 文本特殊字符."""
+    text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N")
+    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 _wrap_text(text: str, max_chars: int) -> list[str]:
+    """按字数换行，优先标点断开."""
+    if len(text) <= max_chars:
+        return [text]
+
+    lines: list[str] = []
+    remaining = text
+
+    while len(remaining) > max_chars:
+        break_point = max_chars
+        punctuations = "，。！？、；：,.;:!?"
+
+        for i in range(max_chars, max_chars // 2, -1):
+            if i < len(remaining) and remaining[i] in punctuations:
+                break_point = i + 1
+                break
+
+        lines.append(remaining[:break_point])
+        remaining = remaining[break_point:]
+
+    if remaining:
+        lines.append(remaining)
+
+    return lines
+
+
+# ── 字幕片段 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class SubtitleSegment:
+    """单个字幕片段."""
+
+    start: float  # 开始时间（秒）
+    end: float  # 结束时间（秒）
+    text: str  # 字幕文本
+    style_name: str = "Default"  # 使用的样式名
+
+
+# ── 字幕渲染引擎 ──────────────────────────────────────────────────────────────
+
+
+class SubtitleRenderEngine:
+    """字幕渲染引擎 — 统一管理多源字幕的 ASS 文件生成.
+
+    支持合并多个字幕来源到同一个 ASS 文件：
+    - 标题（顶部，单独样式）
+    - 字幕（底部，单独样式）
+    - ASR 时间轴字幕
+    - 手动字幕
+
+    输出一个统一的 ASS 文件，供 FFmpeg subtitles filter 烧录。
+    """
+
+    def __init__(
+        self,
+        video_width: int = 1080,
+        video_height: int = 1920,
+        video_duration: float = 0.0,
+    ):
+        self.video_width = video_width
+        self.video_height = video_height
+        self.video_duration = video_duration
+        self._styles: dict[str, SubtitleStyle] = {}
+        self._segments: list[SubtitleSegment] = []
+        self._style_counter = 0
+
+    # ── 样式管理 ──────────────────────────────────────────────────────
+
+    def add_style(self, name: str, style: SubtitleStyle) -> str:
+        """注册一个样式，返回样式名."""
+        self._styles[name] = style
+        return name
+
+    def get_or_create_style(self, base_name: str, style: SubtitleStyle) -> str:
+        """获取或创建样式（避免重复）."""
+        if base_name in self._styles:
+            return base_name
+        self._styles[base_name] = style
+        return base_name
+
+    # ── 字幕源添加 ────────────────────────────────────────────────────
+
+    def add_title(self, text: str, style: SubtitleStyle | None = None) -> None:
+        """添加整段标题（显示整个视频时长）."""
+        if not text or not text.strip():
+            return
+
+        style = style or SubtitleStyle(
+            position="top_center",
+            font_size=48,
+            bold=True,
+            stroke_enabled=True,
+            stroke_width=2.0,
+        )
+        style_name = self.get_or_create_style("TitleStyle", style)
+
+        self._segments.append(
+            SubtitleSegment(
+                start=0.0,
+                end=self.video_duration if self.video_duration > 0 else 9999.0,
+                text=text.strip(),
+                style_name=style_name,
+            )
+        )
+
+    def add_subtitle_text(self, text: str, style: SubtitleStyle | None = None) -> None:
+        """添加整段字幕（显示整个视频时长）."""
+        if not text or not text.strip():
+            return
+
+        style = style or SubtitleStyle()
+        style_name = self.get_or_create_style("SubtitleStyle", style)
+
+        self._segments.append(
+            SubtitleSegment(
+                start=0.0,
+                end=self.video_duration if self.video_duration > 0 else 9999.0,
+                text=text.strip(),
+                style_name=style_name,
+            )
+        )
+
+    def add_timeline_segments(
+        self,
+        segments: list[dict] | list[SubtitleSegment],
+        style: SubtitleStyle | None = None,
+    ) -> None:
+        """添加时间轴字幕片段（ASR 或手动字幕）.
+
+        segments 可以是：
+        - SubtitleSegment 列表
+        - dict 列表，每个 dict 含 start/end/text 字段
+        """
+        if not segments:
+            return
+
+        style = style or SubtitleStyle()
+        style_name = self.get_or_create_style("Default", style)
+
+        for seg in segments:
+            if isinstance(seg, SubtitleSegment):
+                seg.style_name = style_name
+                self._segments.append(seg)
+            elif isinstance(seg, dict):
+                try:
+                    start = float(seg.get("start", 0))
+                    end = float(seg.get("end", 0))
+                    text = str(seg.get("text", ""))
+                    if end > start and text.strip():
+                        self._segments.append(
+                            SubtitleSegment(
+                                start=start,
+                                end=end,
+                                text=text.strip(),
+                                style_name=style_name,
+                            )
+                        )
+                except (TypeError, ValueError):
+                    continue
+
+    def add_asr_timeline(self, timeline: Any, style: SubtitleStyle | None = None) -> None:
+        """从 SubtitleTimeline 对象添加 ASR 字幕."""
+        if not timeline or not hasattr(timeline, "segments") or not timeline.segments:
+            return
+
+        style = style or SubtitleStyle()
+        style_name = self.get_or_create_style("ASRStyle", style)
+
+        for seg in timeline.segments:
+            if hasattr(seg, "start") and hasattr(seg, "end") and hasattr(seg, "text"):
+                if seg.end > seg.start and seg.text.strip():
+                    self._segments.append(
+                        SubtitleSegment(
+                            start=seg.start,
+                            end=seg.end,
+                            text=seg.text.strip(),
+                            style_name=style_name,
+                        )
+                    )
+
+    # ── ASS 文件生成 ──────────────────────────────────────────────────
+
+    def generate_ass(self, output_path: Path) -> Path:
+        """生成 ASS 字幕文件.
+
+        Returns:
+            生成的文件路径；如果没有字幕内容，返回空文件。
+        """
+        if not self._segments:
+            output_path.parent.mkdir(parents=True, exist_ok=True)
+            output_path.write_text("", encoding="utf-8")
+            return output_path
+
+        # 确保至少有 Default 样式
+        if "Default" not in self._styles:
+            self._styles["Default"] = SubtitleStyle()
+
+        # 生成样式行
+        style_lines = []
+        for name, style in self._styles.items():
+            style_lines.append(self._build_ass_style_line(name, style))
+
+        # 生成事件行（按时间排序）
+        self._segments.sort(key=lambda s: s.start)
+        event_lines = []
+        for seg in self._segments:
+            event_lines.append(self._build_ass_event_line(seg))
+
+        # 组装文件
+        ass_content = f"""[Script Info]
+ScriptType: v4.00+
+PlayResX: {self.video_width}
+PlayResY: {self.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(style_lines)}
+
+[Events]
+Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
+{chr(10).join(event_lines)}
+"""
+
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_text(ass_content, encoding="utf-8")
+        return output_path
+
+    def _build_ass_style_line(self, name: str, style: SubtitleStyle) -> str:
+        """构建一条 ASS Style 行."""
+        bold_val = -1 if style.bold else 0
+        italic_val = -1 if style.italic else 0
+
+        # BorderStyle: 1=outline+shadow, 3=opaque box（背景框）
+        if style.background_enabled:
+            border_style = 3
+            back_color = style.ass_background_color
+        else:
+            border_style = 1
+            back_color = style.ass_shadow_color if style.shadow_enabled else style.ass_font_color
+
+        outline_val = style.stroke_width if style.stroke_enabled else 0.0
+        shadow_val = style.shadow_offset_y if style.shadow_enabled else 0
+
+        return (
+            f"Style: {name},{style.font_name},{style.font_size},{style.ass_font_color},"
+            f"&H000000FF,{style.ass_stroke_color},{back_color},"
+            f"{bold_val},{italic_val},0,0,100,100,0,0,"
+            f"{border_style},{outline_val},{shadow_val},{style.alignment},"
+            f"{style.margin_l},{style.margin_r},{style.margin_v},1"
+        )
+
+    def _build_ass_event_line(self, seg: SubtitleSegment) -> str:
+        """构建一条 ASS Dialogue 事件行."""
+        style = self._styles.get(seg.style_name, SubtitleStyle())
+        max_chars = style.max_chars_per_line
+
+        # 自动换行
+        lines = _wrap_text(seg.text, max_chars)
+        display_text = "\\N".join(lines)
+
+        # 动画效果（淡入淡出）
+        effect_tags = ""
+        if style.fade_in > 0 or style.fade_out > 0:
+            fade_in_ms = int(style.fade_in * 1000)
+            fade_out_ms = int(style.fade_out * 1000)
+            effect_tags = f"{{\\fad({fade_in_ms},{fade_out_ms})}}"
+
+        safe_text = _escape_ass_text(display_text)
+        start_time = _format_ass_time(max(0, seg.start))
+        end_time = _format_ass_time(max(seg.start + 0.1, seg.end))
+
+        return f"Dialogue: 0,{start_time},{end_time},{seg.style_name},,0,0,0,," f"{effect_tags}{safe_text}"
+
+    @property
+    def has_subtitles(self) -> bool:
+        """是否有字幕内容."""
+        return len(self._segments) > 0
+
+
+# ── 便捷函数：从 plan.config 快速生成 ASS ────────────────────────────────────
+
+
+def build_subtitles_from_plan(
+    output_path: Path,
+    plan_config: dict,
+    *,
+    video_width: int,
+    video_height: int,
+    video_duration: float,
+    asr_timeline: Any = None,
+) -> Path | None:
+    """从 plan.config 构建字幕 ASS 文件.
+
+    支持的配置项：
+    - title_config: 标题配置（含 text/style）
+    - subtitle_config: 字幕配置（含 text/style）
+    - asr_subtitles: ASR 字幕开关 + 样式
+    - manual_subtitles: 手动字幕片段列表
+
+    Returns:
+        生成的 ASS 文件路径；如果没有任何字幕，返回 None
+    """
+    engine = SubtitleRenderEngine(
+        video_width=video_width,
+        video_height=video_height,
+        video_duration=video_duration,
+    )
+
+    has_any = False
+
+    # 1. 标题
+    title_cfg = plan_config.get("title_config") or {}
+    if isinstance(title_cfg, dict):
+        title_text = str(title_cfg.get("text", ""))
+        title_enabled = title_cfg.get("enabled", True)
+        if title_enabled and title_text.strip():
+            style_dict = title_cfg.get("style") or {}
+            style = SubtitleStyle.from_dict(style_dict)
+            # 标题默认样式：顶部、大字号、粗体
+            if style.position == DEFAULT_POSITION and style.font_size == DEFAULT_FONT_SIZE:
+                style.position = "top_center"
+                style.font_size = 48
+                style.bold = True
+            engine.add_title(title_text, style)
+            has_any = True
+
+    # 2. 静态字幕
+    sub_cfg = plan_config.get("subtitle_config") or {}
+    if isinstance(sub_cfg, dict):
+        sub_text = str(sub_cfg.get("text", ""))
+        sub_enabled = sub_cfg.get("enabled", True)
+        if sub_enabled and sub_text.strip():
+            style_dict = sub_cfg.get("style") or {}
+            style = SubtitleStyle.from_dict(style_dict)
+            engine.add_subtitle_text(sub_text, style)
+            has_any = True
+
+    # 3. ASR 自动字幕
+    asr_cfg = plan_config.get("asr_subtitles") or {}
+    if isinstance(asr_cfg, dict) and asr_cfg.get("enabled", False):
+        if asr_timeline is not None:
+            style_dict = asr_cfg.get("style") or {}
+            style = SubtitleStyle.from_dict(style_dict)
+            engine.add_asr_timeline(asr_timeline, style)
+            has_any = has_any or engine.has_subtitles
+
+    # 4. 手动字幕
+    manual_segs = plan_config.get("manual_subtitles") or []
+    if isinstance(manual_segs, list) and manual_segs:
+        style_dict = (plan_config.get("manual_subtitle_style") or {}) or {}
+        style = SubtitleStyle.from_dict(style_dict)
+        engine.add_timeline_segments(manual_segs, style)
+        has_any = has_any or engine.has_subtitles
+
+    if not has_any:
+        return None
+
+    return engine.generate_ass(output_path)
+
+
+# ── FFmpeg 烧录滤镜生成 ───────────────────────────────────────────────────────
+
+
+def build_subtitle_filter(
+    ass_path: Path | str,
+    *,
+    video_input_label: str = "0:v",
+    output_label: str = "subtitled",
+    work_dir: Path | str | None = None,
+) -> str:
+    """生成 FFmpeg subtitles 滤镜字符串.
+
+    Args:
+        ass_path: ASS 字幕文件路径
+        video_input_label: 视频输入标签（如 "0:v" 或 "[v_out]"）
+        output_label: 输出标签
+        work_dir: 工作目录（必填，用于路径安全校验，防止路径遍历绕过）
+
+    Returns:
+        filter_complex 片段，如 "[0:v]subtitles=xxx.ass[subtitled]"
+
+    Raises:
+        PathSecurityError: 字幕路径不安全或 work_dir 未提供
+    """
+    # ── 安全校验：字幕文件路径白名单 ──
+    ass_path_str = str(ass_path)
+    if work_dir is None or not str(work_dir).strip():
+        raise PathSecurityError("work_dir 必须提供，不能为 None 或空")
+
+    _validate_subtitle_path(ass_path_str, Path(work_dir))
+
+    # FFmpeg subtitles filter 的路径需要转义：
+    # - Windows 路径的 \ → /
+    # - 冒号 : → \:
+    # - 单引号 ' → '\''
+    safe_path = ass_path_str.replace("\\", "/").replace(":", "\\:").replace("'", "'\\''")
+    return f"{video_input_label}subtitles='{safe_path}'[{output_label}]"
+
+
+def _validate_subtitle_path(subtitle_path: str, work_dir: Path) -> None:
+    """校验字幕文件路径安全性.
+
+    规则：
+    - 必须是本地路径（不支持远程URL字幕）
+    - local:// schema → 必须在 work_dir 内
+    - 相对路径 → 必须在 work_dir 内
+    - 绝对路径 → 必须在允许目录白名单内
+    - 扩展名必须是字幕格式
+
+    Raises:
+        PathSecurityError: 路径不安全
+    """
+    if not subtitle_path or not isinstance(subtitle_path, str):
+        raise PathSecurityError("字幕路径不能为空")
+
+    # 不允许远程URL字幕（subtitles滤镜不支持远程加载，且有SSRF风险）
+    if subtitle_path.startswith(("http://", "https://", "oss://")):
+        raise PathSecurityError("不允许使用远程URL字幕文件")
+
+    is_abs = subtitle_path.startswith("/") and not subtitle_path.startswith("local://")
+
+    resolved_path = safe_resolve_path(
+        subtitle_path,
+        work_dir,
+        allow_outside=is_abs,
+        allowed_extensions=ALLOWED_SUBTITLE_EXTENSIONS,
+    )
+
+    # 绝对路径额外检查白名单目录（用realpath规范化后的真实路径比较，防止 ../ 遍历绕过）
+    if is_abs:
+        resolved_work_dir = work_dir.resolve()
+        try:
+            resolved_path.relative_to(resolved_work_dir)
+        except ValueError:
+            if not is_in_allowed_dirs(resolved_path):
+                raise PathSecurityError(f"字幕路径不在允许目录内: {subtitle_path[:80]}")

apps/worker/video_processing/unified_render_service.py

@@ -28,7 +28,6 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
 
-from video_processing.chroma_key_engine import apply_chroma_key_if_needed
 from video_processing.color_grade_engine import ColorGradeConfig, ColorGradeEngine
 from video_processing.ffmpeg_utils import (
     DEFAULT_FPS,
@@ -46,7 +45,7 @@ from video_processing.render_audio import RenderContext, merge_audio_video, mix_
 from video_processing.render_subtitles import generate_ass_subtitles
 from video_processing.reverse_engine import ReverseConfig, ReverseEngine
 from video_processing.speed_engine import SpeedConfig, SpeedEngine
-from video_processing.sticker_engine import StickerEngine, parse_stickers_from_config
+from video_processing.sticker_engine import StickerEngine
 from video_processing.subtitle_generator import generate_ass_from_timeline
 from video_processing.transition_engine import TransitionEngine
 from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_from_clip_config
@@ -343,6 +342,7 @@ class UnifiedRenderService:
         else:
             config = self.plan.config or {}
             bgm_config = config.get("bgm", {}) or {}
+            audio_tracks_config = config.get("audio_tracks") or {}
             noise_reduction_config = config.get("audio_noise_reduction")
             ctx = RenderContext(
                 work_dir=self.work_dir,
@@ -355,6 +355,7 @@ class UnifiedRenderService:
                 video_duration,
                 bgm_path=self.bgm_path,
                 bgm_config=bgm_config,
+                audio_tracks_config=audio_tracks_config,
             )
             t_audio_end = time.time()
             audio_mix_ms = int((t_audio_end - t_audio_start) * 1000)
@@ -639,8 +640,8 @@ class UnifiedRenderService:
         return timeline
 
     def _extract_audio(self, video_path: Path, output_path: Path) -> None:
-        """从视频中提取音频为16kHz单声道wav（ASR友好格式）。"""
-        import subprocess
+        """从视频中提取音频为16kHz单声道wav（ASR友好格式）."""
+        from video_processing.ffmpeg_utils import run_ffmpeg
 
         cmd = [
             "ffmpeg",
@@ -657,15 +658,10 @@ class UnifiedRenderService:
             str(output_path),
         ]
 
-        result = subprocess.run(
-            cmd,
-            capture_output=True,
-            text=True,
-            timeout=120,
-        )
-
-        if result.returncode != 0:
-            raise RuntimeError(f"音频提取失败: {result.stderr[:200]}")
+        try:
+            run_ffmpeg(cmd, timeout=120)
+        except Exception as e:
+            raise RuntimeError(f"音频提取失败: {str(e)[:200]}") from e
 
     def _maybe_add_voiceover_layer(
         self,

apps/worker/video_processing/url_security.py

@@ -0,0 +1,21 @@
+"""URL 安全校验工具 — SSRF 防护（向后兼容层）.
+
+本模块为向后兼容而保留，实际实现已迁移至 packages.shared.url_security。
+所有符号均从该模块重新导出，请新代码直接 import packages.shared.url_security。
+"""
+
+from packages.shared.url_security import (  # noqa: F401
+    ALLOWED_AUDIO_MIME_TYPES,
+    ALLOWED_IMAGE_MIME_TYPES,
+    ALLOWED_PORTS,
+    ALLOWED_SCHEMES,
+    ALLOWED_VIDEO_MIME_TYPES,
+    DEFAULT_MAX_DOWNLOAD_SIZE,
+    MAX_URL_LENGTH,
+    TRUSTED_DOMAINS,
+    UrlSecurityError,
+    is_url_safe,
+    safe_download_bytes,
+    safe_download_file,
+    validate_url_safety,
+)

apps/worker/worker_app/tasks/asset_analyzer.py

@@ -130,6 +130,8 @@ class AssetAnalyzer:
         info = VideoInfo()
 
         try:
+            from video_processing.ffmpeg_utils import run_ffprobe
+
             cmd = [
                 "ffprobe",
                 "-v",
@@ -140,38 +142,31 @@ class AssetAnalyzer:
                 "-show_streams",
                 self.video_path,
             ]
-            result = subprocess.run(
-                cmd,
-                capture_output=True,
-                text=True,
-                timeout=30,
-            )
+            stdout, _ = run_ffprobe(cmd, timeout=30)
+            data = json.loads(stdout)
+            streams = data.get("streams", [])
+            format_info = data.get("format", {})
 
-            if result.returncode == 0:
-                data = json.loads(result.stdout)
-                streams = data.get("streams", [])
-                format_info = data.get("format", {})
+            for stream in streams:
+                if stream.get("codec_type") == "video":
+                    info.width = int(stream.get("width", 0))
+                    info.height = int(stream.get("height", 0))
+                    info.codec = stream.get("codec_name", "")
 
-                for stream in streams:
-                    if stream.get("codec_type") == "video":
-                        info.width = int(stream.get("width", 0))
-                        info.height = int(stream.get("height", 0))
-                        info.codec = stream.get("codec_name", "")
+                    # 解析帧率
+                    fps_str = stream.get("r_frame_rate", "0/1")
+                    if "/" in fps_str:
+                        num, denom = fps_str.split("/")
+                        info.fps = float(num) / float(denom) if float(denom) != 0 else 0.0
+                    else:
+                        info.fps = float(fps_str)
 
-                        # 解析帧率
-                        fps_str = stream.get("r_frame_rate", "0/1")
-                        if "/" in fps_str:
-                            num, denom = fps_str.split("/")
-                            info.fps = float(num) / float(denom) if float(denom) != 0 else 0.0
-                        else:
-                            info.fps = float(fps_str)
+                elif stream.get("codec_type") == "audio":
+                    info.has_audio = True
 
-                    elif stream.get("codec_type") == "audio":
-                        info.has_audio = True
-
-                info.duration = float(format_info.get("duration", 0))
-                info.bitrate = int(format_info.get("bit_rate", 0))
-                info.file_size = int(format_info.get("size", 0))
+            info.duration = float(format_info.get("duration", 0))
+            info.bitrate = int(format_info.get("bit_rate", 0))
+            info.file_size = int(format_info.get("size", 0))
 
         except Exception as e:
             logger.warning(f"Failed to get video info: {e}")
@@ -179,7 +174,7 @@ class AssetAnalyzer:
         self._video_info = info
         return info
 
-    def extract_frames(self, count: int = 10, max_frames: int = 30) -> list[np.ndarray]:
+    def extract_frames(self, count: int = 10) -> list[np.ndarray]:
         """
         从视频中均匀抽取帧
 
@@ -224,14 +219,14 @@ class AssetAnalyzer:
                     output_path,
                 ]
 
-                result = subprocess.run(
-                    cmd,
-                    capture_output=True,
-                    text=True,
-                    timeout=10,
-                )
+                from video_processing.ffmpeg_utils import run_ffmpeg
 
-                if result.returncode == 0 and os.path.exists(output_path):
+                try:
+                    run_ffmpeg(cmd, timeout=10)
+                except Exception:
+                    continue
+
+                if os.path.exists(output_path):
                     # 读取帧并转换为 numpy 数组
                     img = self._load_image_as_array(output_path)
                     if img is not None:
@@ -397,14 +392,19 @@ class AssetAnalyzer:
                 audio_path,
             ]
 
-            result_audio = subprocess.run(
-                cmd,
-                capture_output=True,
-                text=True,
-                timeout=30,
-            )
+            from video_processing.ffmpeg_utils import run_ffmpeg
 
-            if result_audio.returncode == 0 and os.path.exists(audio_path):
+            try:
+                run_ffmpeg(cmd, timeout=30)
+            except Exception:
+                # 音频提取失败，返回默认分析结果
+                return AudioAnalysis(
+                    has_speech=False,
+                    speech_ratio=0.0,
+                    avg_volume=0.0,
+                )
+
+            if os.path.exists(audio_path):
                 # 读取音频数据
                 import struct
 

apps/worker/worker_app/tasks/batch_download.py

@@ -106,7 +106,16 @@ def _download_video_to_file(url: str, dest_path: str) -> None:
     except Exception:
         pass
 
-    # 回退到 HTTP 下载
-    import urllib.request
+    # 回退到 HTTP 下载（含 SSRF 防护 + 大小限制 + 类型校验）
+    from video_processing.url_security import (
+        ALLOWED_VIDEO_MIME_TYPES,
+        safe_download_file,
+    )
 
-    urllib.request.urlretrieve(url, dest_path)  # nosec B310
+    safe_download_file(
+        url,
+        dest_path,
+        purpose="batch_video_download",
+        allowed_mime_types=ALLOWED_VIDEO_MIME_TYPES | {"application/octet-stream"},
+        timeout=300.0,
+    )

apps/worker/worker_app/tasks/compose_video.py

@@ -6,7 +6,6 @@
 from __future__ import annotations
 
 import os
-import subprocess
 import tempfile
 from pathlib import Path
 
@@ -115,16 +114,12 @@ def _compose_with_legacy_engine(task, job_service, job, plan_id: str, db) -> dic
     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]}")
+        from video_processing.ffmpeg_utils import run_ffmpeg
+
+        run_ffmpeg(compose_cmd.command, timeout=3600)
+    except Exception as e:
+        error_msg = f"FFmpeg 执行失败: {str(e)[:500]}"
+        job_service.fail_job(job_id, error_msg)
         raise
 
     # 上传结果

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -257,7 +257,6 @@ def _render_with_legacy(
 ) -> dict:
     """旧引擎路径（VideoComposeService + FFmpeg filter_complex）。"""
     import os
-    import subprocess
 
     from apps.api.app.services.video_compose_service import VideoComposeService
 
@@ -278,16 +277,11 @@ def _render_with_legacy(
 
     logger.info("执行 FFmpeg (legacy): plan_id=%s", plan_id)
     try:
-        subprocess.run(
-            compose_cmd.command,
-            check=True,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            text=True,
-            timeout=3600,
-        )
-    except subprocess.CalledProcessError as e:
-        error_msg = f"FFmpeg 执行失败: {e.stderr[:500]}"
+        from video_processing.ffmpeg_utils import run_ffmpeg
+
+        run_ffmpeg(compose_cmd.command, timeout=3600)
+    except Exception as e:
+        error_msg = f"FFmpeg 执行失败: {str(e)[:500]}"
         logger.error("FFmpeg 执行失败(legacy): %s — %s", plan_id, error_msg)
         _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, error_msg)
         return {"status": "error", "message": error_msg}

apps/worker/worker_app/tasks/generation.py

@@ -364,10 +364,19 @@ def _prepare_bgm_track(
         try:
             parsed = urlparse(audio_url)
             if parsed.scheme in ("http", "https"):
-                import urllib.request
+                from video_processing.url_security import (
+                    ALLOWED_AUDIO_MIME_TYPES,
+                    safe_download_file,
+                )
 
                 logger.info("[task_id=%s] [BGM] 从URL下载: %s", task_id, audio_url[:80])
-                urllib.request.urlretrieve(audio_url, bgm_file)  # nosec B310
+                safe_download_file(
+                    audio_url,
+                    str(bgm_file),
+                    purpose="bgm_download",
+                    allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
+                    timeout=60.0,
+                )
                 if bgm_file.exists() and bgm_file.stat().st_size > 0:
                     return str(bgm_file)
         except Exception as e:
@@ -401,10 +410,19 @@ def _prepare_bgm_track(
 
             preset = get_preset_bgm(preset_id)
             if preset and preset.audio_url:
-                import urllib.request
+                from video_processing.url_security import (
+                    ALLOWED_AUDIO_MIME_TYPES,
+                    safe_download_file,
+                )
 
                 logger.info("[task_id=%s] [BGM] 从预设库下载: preset_id=%s", task_id, preset_id)
-                urllib.request.urlretrieve(preset.audio_url, bgm_file)  # nosec B310
+                safe_download_file(
+                    preset.audio_url,
+                    str(bgm_file),
+                    purpose="bgm_preset_download",
+                    allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
+                    timeout=60.0,
+                )
                 if bgm_file.exists() and bgm_file.stat().st_size > 0:
                     return str(bgm_file)
         except Exception as e:
@@ -418,17 +436,31 @@ def _prepare_bgm_track(
 def _verify_url_accessible(url: str, timeout: float = 10.0, retries: int = 2) -> bool:
     """HEAD 请求校验 URL 可访问（含重试，防止 OSS 抖动误报）。
 
+    安全：
+    - 请求前先做 SSRF 安全校验（内网IP/回环地址/链路本地地址等）
+    - scheme 仅允许 http/https
+    - 端口仅允许 80/443
+
     Args:
         url: 待校验的 URL
         timeout: 单次请求超时时间（秒）
         retries: 最大重试次数（默认 2 次，首次失败后间隔 1s 重试）
 
     Returns:
-        True 表示 URL 可访问（HTTP 2xx/3xx），False 表示所有尝试均失败。
+        True 表示 URL 可访问（HTTP 2xx/3xx），False 表示所有尝试均失败或安全校验不通过。
     """
     import time
     import urllib.request
 
+    from video_processing.url_security import UrlSecurityError, validate_url_safety
+
+    # P0-1 SSRF 防护：请求前先校验 URL 安全性
+    try:
+        validate_url_safety(url, purpose="url_verify")
+    except UrlSecurityError as e:
+        logger.warning("URL 安全校验失败，拒绝访问: url=%s error=%s", url[:80], e)
+        return False
+
     last_error: Exception | None = None
     for attempt in range(1 + retries):
         try:
@@ -461,7 +493,6 @@ def _download_library_assets(
     asset_library_id: str = "",
     project_id: str = "",
     asset_ids: list[str] | None = None,
-    video_extensions: tuple = (".mp4", ".mov", ".avi", ".mkv", ".webm"),
     strict: bool = True,
     task_id: str = "",
     gen_task=None,
@@ -480,7 +511,6 @@ def _download_library_assets(
         asset_library_id: 素材库 ID（可选，与 project_id 二选一）
         project_id: 项目 ID（可选，与 asset_library_id 二选一）
         asset_ids: 指定素材 ID 列表，为空则下载全部 ready 视频素材
-        video_extensions: 支持的视频扩展名（保留兼容，当前按 file_type 过滤）
         strict: 严格模式（默认 True）。
             True — 任何素材下载失败立即抛 RuntimeError；
             False — 跳过失败素材，返回成功列表（调用方可通过日志感知失败）。

apps/worker/worker_app/tasks/ingest.py

@@ -45,6 +45,8 @@ def extract_media_metadata(file_url: str, media_type: str) -> dict:
     try:
         if media_type == "video":
             # 使用 ffprobe 提取视频元数据
+            from video_processing.ffmpeg_utils import run_ffprobe
+
             cmd = [
                 "ffprobe",
                 "-v",
@@ -55,16 +57,11 @@ def extract_media_metadata(file_url: str, media_type: str) -> dict:
                 "-show_streams",
                 file_url,
             ]
-            result = subprocess.run(
-                cmd,
-                capture_output=True,
-                text=True,
-                timeout=30,
-            )
-            if result.returncode == 0:
+            try:
+                stdout, _ = run_ffprobe(cmd, timeout=30)
                 import json as json_lib
 
-                probe_data = json_lib.loads(result.stdout)
+                probe_data = json_lib.loads(stdout)
 
                 # 提取视频流信息
                 for stream in probe_data.get("streams", []):
@@ -83,6 +80,9 @@ def extract_media_metadata(file_url: str, media_type: str) -> dict:
                 metadata["size_bytes"] = int(format_info.get("size", 0))
                 metadata["bitrate"] = int(format_info.get("bit_rate", 0))
 
+            except Exception as e:
+                logger.warning("视频元数据提取失败: %s", e)
+
         elif media_type == "image":
             # 使用 Pillow 提取图片元数据
             try:

apps/worker/worker_app/tasks/voice_extraction.py

@@ -19,14 +19,12 @@ class VoiceExtractor:
     """Extract voice tracks and background music from videos using FFmpeg."""
 
     @staticmethod
-    def _run_ffmpeg(cmd: list[str]) -> subprocess.CompletedProcess:
-        """Run FFmpeg command and return result."""
-        logger.info(f"Running FFmpeg: {chr(39).join(cmd)}")
-        result = subprocess.run(cmd, capture_output=True, text=True)
-        if result.returncode != 0:
-            logger.error(f"FFmpeg error: {result.stderr}")
-            raise RuntimeError(f"FFmpeg failed: {result.stderr}")
-        return result
+    def _run_ffmpeg(cmd: list[str]) -> None:
+        """Run FFmpeg command using 统一 run_ffmpeg 工具."""
+        from video_processing.ffmpeg_utils import run_ffmpeg
+
+        logger.info("Running FFmpeg: %s", " ".join(cmd[:10]))
+        run_ffmpeg(cmd)
 
     def extract_voice(
         self,

packages/application/tts_job/streaming_service.py

@@ -15,6 +15,7 @@ import httpx
 
 from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
 from packages.application.tts_job.text_splitter import split_text
+from packages.shared.url_security import ALLOWED_AUDIO_MIME_TYPES, safe_download_bytes
 
 logger = logging.getLogger(__name__)
 
@@ -224,10 +225,13 @@ class TTSStreamingService:
     # ── 工具方法 ────────────────────────────────────────────
 
     def _download_audio(self, url: str) -> bytes:
-        """下载音频数据。"""
-        resp = httpx.get(url, timeout=60.0, follow_redirects=True)
-        resp.raise_for_status()
-        return resp.content
+        """下载音频数据（含 SSRF 防护 + 大小限制 + 重定向校验）。"""
+        return safe_download_bytes(
+            url,
+            purpose="tts_streaming_download",
+            allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
+            timeout=60.0,
+        )
 
     async def _stream_audio_chunks(self, websocket: Any, audio_data: bytes) -> int:
         """将音频数据分块通过 WebSocket 推送。

packages/application/tts_job/workflow.py

@@ -19,16 +19,19 @@ from typing import Optional
 
 import httpx
 
-from packages.application.cosyvoice_service import (
-    CosyVoiceAuthError,
-    CosyVoiceError,
-    CosyVoiceService,
-)
+from packages.application.cosyvoice_service import CosyVoiceAuthError, CosyVoiceError, CosyVoiceService
 from packages.application.tts_job.audio_merger import AudioMerger
 from packages.application.tts_job.text_splitter import split_text
 from packages.domain.tts_job import TTSJob, TTSJobStatus
 from packages.ports.tts_job_repository import TTSJobRepository
 from packages.shared.storage import SharedStorageService, get_shared_storage_service
+from packages.shared.url_security import (
+    ALLOWED_AUDIO_MIME_TYPES,
+    UrlSecurityError,
+    safe_download_bytes,
+    safe_download_file,
+    validate_url_safety,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -96,10 +99,13 @@ class TTSWorkflowService:
         content_type = content_type_map.get(audio_format, "application/octet-stream")
 
         try:
-            # 下载临时音频
-            resp = httpx.get(temp_url, timeout=60.0, follow_redirects=True)
-            resp.raise_for_status()
-            audio_data = resp.content
+            # 安全下载临时音频（SSRF 防护 + 大小限制 + 重定向校验）
+            audio_data = safe_download_bytes(
+                temp_url,
+                purpose="tts_audio_download",
+                allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
+                timeout=60.0,
+            )
 
             # 上传到 OSS
             file_obj = io.BytesIO(audio_data)
@@ -463,13 +469,15 @@ class TTSWorkflowService:
 
                 total_duration += result.get("duration", 0.0)
 
-                # 下载分段音频到临时文件
-                resp = httpx.get(audio_url, timeout=60.0, follow_redirects=True)
-                resp.raise_for_status()
-
+                # 安全下载分段音频到临时文件（SSRF 防护 + 大小限制）
                 seg_path = os.path.join(temp_dir, f"seg_{idx:03d}.{job.format}")
-                with open(seg_path, "wb") as f:
-                    f.write(resp.content)
+                safe_download_file(
+                    audio_url,
+                    seg_path,
+                    purpose="tts_segment_download",
+                    allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
+                    timeout=60.0,
+                )
                 audio_paths.append(seg_path)
 
             # 合并

packages/shared/url_security.py

@@ -0,0 +1,402 @@
+"""URL 安全校验工具 — SSRF 防护.
+
+统一的外部 URL 安全校验方案，覆盖所有渲染管线和 TTS 中的外部下载场景。
+放在 packages/shared/ 作为单一来源，worker 和 application 层都可引用。
+
+防护要点：
+1. Scheme 白名单：仅允许 http/https
+2. 主机 SSRF 防护：禁止内网 IP、回环地址、链路本地地址、元数据服务
+3. 端口白名单：仅允许 80/443（标准 HTTP/HTTPS）
+4. 域名校验：禁止 IP 直接访问（除非在白名单中）
+5. 重定向防护：手动跟随重定向，每次跳转前重新校验目标 URL
+6. 文件大小限制：流式下载，超过上限立即中断
+7. MIME 类型白名单：可选的内容类型校验
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import logging
+import os
+import socket
+import urllib.error
+import urllib.request
+from urllib.parse import urljoin, urlparse
+
+logger = logging.getLogger(__name__)
+
+# 允许的 URL scheme
+ALLOWED_SCHEMES = {"http", "https"}
+
+# 允许的端口（标准 HTTP/HTTPS）
+ALLOWED_PORTS = {80, 443}
+
+# 可信域名白名单（可根据实际 OSS/CDN 域名配置）
+# 从环境变量读取，格式："oss-cn-hangzhou.aliyuncs.com,cdn.example.com"
+# 默认空表示所有公网域名都允许，但仍会做 SSRF 检查
+TRUSTED_DOMAINS: set[str] = set()
+_env_trusted = os.environ.get("URL_SECURITY_TRUSTED_DOMAINS", "")
+if _env_trusted:
+    TRUSTED_DOMAINS = {d.strip() for d in _env_trusted.split(",") if d.strip()}
+
+# 是否允许 IP 直接访问（默认禁止，防止绕过 DNS 校验）
+ALLOW_DIRECT_IP = os.environ.get("URL_SECURITY_ALLOW_DIRECT_IP", "false").lower() == "true"
+
+# 最大 URL 长度
+MAX_URL_LENGTH = 2048
+
+# 单次下载最大文件大小（默认 200MB）
+DEFAULT_MAX_DOWNLOAD_SIZE = int(os.environ.get("URL_SECURITY_MAX_DOWNLOAD_MB", "200")) * 1024 * 1024
+
+# 允许的音频 MIME 类型白名单
+ALLOWED_AUDIO_MIME_TYPES = {
+    "audio/mpeg",
+    "audio/mp3",
+    "audio/wav",
+    "audio/x-wav",
+    "audio/pcm",
+    "audio/ogg",
+    "audio/opus",
+    "audio/flac",
+    "audio/aac",
+    "audio/m4a",
+    "audio/x-m4a",
+    "audio/mp4",
+    "application/octet-stream",  # 兼容一些 CDN 返回通用类型
+}
+
+# 允许的视频 MIME 类型白名单
+ALLOWED_VIDEO_MIME_TYPES = {
+    "video/mp4",
+    "video/quicktime",
+    "video/x-matroska",
+    "video/webm",
+    "video/avi",
+    "video/x-msvideo",
+    "video/mpeg",
+    "application/octet-stream",
+}
+
+# 允许的图片 MIME 类型白名单
+ALLOWED_IMAGE_MIME_TYPES = {
+    "image/jpeg",
+    "image/png",
+    "image/gif",
+    "image/webp",
+    "image/bmp",
+}
+
+# 下载块大小
+_DOWNLOAD_CHUNK_SIZE = 8192
+
+# 最大重定向次数
+_MAX_REDIRECTS = 5
+
+
+class UrlSecurityError(ValueError):
+    """URL 安全校验失败."""
+
+    pass
+
+
+class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
+    """禁止自动重定向的 handler，用于手动控制重定向以做安全校验."""
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: N802
+        return None
+
+
+def validate_url_safety(url: str, *, purpose: str = "download") -> str:
+    """校验 URL 安全性，返回标准化后的 URL（供下游使用）.
+
+    Args:
+        url: 待校验的 URL
+        purpose: 用途描述（用于日志），如 "bgm_download"、"tts_download"
+
+    Returns:
+        标准化后的 URL
+
+    Raises:
+        UrlSecurityError: URL 不安全
+    """
+    if not url:
+        raise UrlSecurityError("URL 为空")
+
+    if len(url) > MAX_URL_LENGTH:
+        raise UrlSecurityError(f"URL 过长 ({len(url)} > {MAX_URL_LENGTH})")
+
+    # 解析 URL
+    try:
+        parsed = urlparse(url)
+    except Exception as e:
+        raise UrlSecurityError(f"URL 解析失败: {e}") from e
+
+    # 1. Scheme 校验
+    if not parsed.scheme or parsed.scheme.lower() not in ALLOWED_SCHEMES:
+        raise UrlSecurityError(f"不允许的 URL scheme: {parsed.scheme}")
+
+    # 2. 主机名校验
+    hostname = parsed.hostname
+    if not hostname:
+        raise UrlSecurityError("URL 缺少主机名")
+
+    # 2.1 常见内网主机名前置拦截（防止 DNS rebinding 绕过）
+    _check_internal_hostnames(hostname)
+
+    # 3. 端口校验
+    port = parsed.port
+    if port is not None and port not in ALLOWED_PORTS:
+        raise UrlSecurityError(f"不允许的端口: {port}")
+
+    # 4. SSRF 防护 - 解析 IP 并检查
+    try:
+        # 先判断是否是 IP 地址
+        ip_obj = None
+        try:
+            ip_obj = ipaddress.ip_address(hostname)
+        except ValueError:
+            pass  # 不是 IP，继续走域名解析
+
+        if ip_obj is not None:
+            # 是直接 IP 访问
+            if not ALLOW_DIRECT_IP and not _is_trusted_ip(ip_obj):
+                raise UrlSecurityError(f"禁止直接 IP 访问: {hostname}")
+            _check_ssrf_ip(ip_obj)
+        else:
+            # 域名 — 解析 DNS 检查 SSRF
+            _check_ssrf_domain(hostname)
+    except UrlSecurityError:
+        raise
+    except Exception as e:
+        logger.warning("URL 安全校验异常: url=%s purpose=%s error=%s", url[:80], purpose, e)
+        raise UrlSecurityError(f"URL 安全校验异常: {e}") from e
+
+    # 5. 可信域名校验（如果配置了白名单）
+    if TRUSTED_DOMAINS and not _is_trusted_domain(hostname):
+        raise UrlSecurityError(f"域名不在可信白名单中: {hostname}")
+
+    logger.debug("URL 安全校验通过: url=%s purpose=%s", url[:80], purpose)
+    return url
+
+
+def _check_internal_hostnames(hostname: str) -> None:
+    """前置检查常见内网/敏感主机名，防止 DNS 解析层绕过."""
+    hostname_lower = hostname.lower()
+    internal_hostnames = {
+        "localhost",
+        "localhost.localdomain",
+        "ip6-localhost",
+        "ip6-loopback",
+        "metadata",
+        "metadata.google.internal",
+        "169.254.169.254",  # 云元数据服务
+    }
+    if hostname_lower in internal_hostnames:
+        raise UrlSecurityError(f"禁止访问内部主机名: {hostname}")
+
+    # 检查以 .local / .internal 结尾的主机名
+    if hostname_lower.endswith((".local", ".internal", ".localdomain")):
+        raise UrlSecurityError(f"禁止访问内网域名: {hostname}")
+
+
+def _check_ssrf_ip(ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address) -> None:
+    """检查 IP 是否属于 SSRF 风险范围."""
+    # 回环地址
+    if ip_obj.is_loopback:
+        raise UrlSecurityError(f"禁止访问回环地址: {ip_obj}")
+
+    # 私有地址（内网）
+    if ip_obj.is_private:
+        raise UrlSecurityError(f"禁止访问内网地址: {ip_obj}")
+
+    # 链路本地地址
+    if ip_obj.is_link_local:
+        raise UrlSecurityError(f"禁止访问链路本地地址: {ip_obj}")
+
+    # 组播地址
+    if ip_obj.is_multicast:
+        raise UrlSecurityError(f"禁止访问组播地址: {ip_obj}")
+
+    # 未指定地址（0.0.0.0 / ::）
+    if ip_obj.is_unspecified:
+        raise UrlSecurityError(f"禁止访问未指定地址: {ip_obj}")
+
+    # 保留地址
+    if ip_obj.is_reserved:
+        raise UrlSecurityError(f"禁止访问保留地址: {ip_obj}")
+
+
+def _check_ssrf_domain(hostname: str) -> None:
+    """对域名做 DNS 解析并检查所有解析结果的 IP 是否安全.
+
+    注意：这不能完全防止 DNS rebinding，但能防御大部分 SSRF 场景。
+    """
+    try:
+        # 解析所有地址
+        infos = socket.getaddrinfo(hostname, None)
+        if not infos:
+            raise UrlSecurityError(f"域名解析失败: {hostname}")
+
+        for info in infos:
+            ip_str = info[4][0]
+            try:
+                ip_obj = ipaddress.ip_address(ip_str)
+                _check_ssrf_ip(ip_obj)
+            except ValueError:
+                # 无法解析为 IP，跳过（不应该发生）
+                continue
+    except socket.gaierror as e:
+        raise UrlSecurityError(f"域名解析失败: {hostname} ({e})") from e
+
+
+def _is_trusted_ip(ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
+    """检查 IP 是否在可信列表中（目前通过环境变量配置域名，IP 级信任暂不开放）."""
+    return False
+
+
+def _is_trusted_domain(hostname: str) -> bool:
+    """检查域名是否在可信白名单中（支持子域名匹配）."""
+    hostname_lower = hostname.lower()
+    if hostname_lower in TRUSTED_DOMAINS:
+        return True
+    # 检查子域名
+    for domain in TRUSTED_DOMAINS:
+        if hostname_lower.endswith("." + domain.lower()):
+            return True
+    return False
+
+
+def is_url_safe(url: str, *, purpose: str = "download") -> bool:
+    """便捷函数：检查 URL 是否安全，不抛异常."""
+    try:
+        validate_url_safety(url, purpose=purpose)
+        return True
+    except UrlSecurityError:
+        return False
+
+
+# ── 安全下载 ───────────────────────────────────────────────────────────────────
+
+
+def safe_download_file(
+    url: str,
+    dest_path: str,
+    *,
+    purpose: str = "download",
+    max_size: int = DEFAULT_MAX_DOWNLOAD_SIZE,
+    allowed_mime_types: set[str] | None = None,
+    timeout: float = 60.0,
+) -> int:
+    """安全下载 URL 到本地文件。
+
+    包含防护：
+    - SSRF 校验（初始 URL + 每次重定向后都校验）
+    - 重定向次数限制 + 手动跟随（避免重定向绕过 SSRF）
+    - 文件大小限制（流式读取，超过立即中断）
+    - MIME 类型白名单（可选）
+
+    Args:
+        url: 下载 URL
+        dest_path: 目标文件路径
+        purpose: 用途描述（日志用）
+        max_size: 最大下载字节数，超过则中断并抛出 UrlSecurityError
+        allowed_mime_types: 允许的 Content-Type 集合，None 表示不校验
+        timeout: 单次请求超时（秒）
+
+    Returns:
+        实际下载的字节数
+
+    Raises:
+        UrlSecurityError: 安全校验失败
+    """
+    current_url = url
+    redirect_count = 0
+    total_bytes = 0
+
+    # 使用不自动跟随重定向的 opener
+    no_redirect_opener = urllib.request.build_opener(NoRedirectHandler())
+
+    while True:
+        # 每次请求前都做 SSRF 校验（重定向目标也会校验）
+        validate_url_safety(current_url, purpose=purpose)
+
+        req = urllib.request.Request(current_url, method="GET")
+        req.add_header("User-Agent", "xiaoxia-saas-worker/1.0")
+
+        try:
+            resp = no_redirect_opener.open(req, timeout=timeout)  # nosec B310
+        except urllib.error.HTTPError as e:
+            # 3xx 重定向
+            if 300 <= e.code < 400 and e.headers.get("Location"):
+                if redirect_count >= _MAX_REDIRECTS:
+                    raise UrlSecurityError(f"重定向次数超过限制 ({_MAX_REDIRECTS})") from e
+                redirect_count += 1
+                current_url = urljoin(current_url, e.headers["Location"])
+                continue
+            raise UrlSecurityError(f"HTTP 错误: {e.code} {e.reason}") from e
+        except urllib.error.URLError as e:
+            raise UrlSecurityError(f"URL 错误: {e.reason}") from e
+
+        try:
+            # Content-Type 校验
+            if allowed_mime_types is not None:
+                content_type = resp.headers.get("Content-Type", "").split(";")[0].strip().lower()
+                if content_type and content_type not in allowed_mime_types:
+                    raise UrlSecurityError(
+                        f"不允许的 Content-Type: {content_type}, " f"允许: {sorted(allowed_mime_types)}"
+                    )
+
+            # Content-Length 预检
+            content_length = resp.headers.get("Content-Length")
+            if content_length and int(content_length) > max_size:
+                raise UrlSecurityError(f"文件过大: {content_length} bytes > {max_size} bytes 上限")
+
+            # 流式下载，实时检查大小
+            with open(dest_path, "wb") as f:
+                while True:
+                    chunk = resp.read(_DOWNLOAD_CHUNK_SIZE)
+                    if not chunk:
+                        break
+                    total_bytes += len(chunk)
+                    if total_bytes > max_size:
+                        raise UrlSecurityError(f"下载超过大小限制: {total_bytes} bytes > {max_size} bytes")
+                    f.write(chunk)
+
+            return total_bytes
+        finally:
+            resp.close()
+
+
+def safe_download_bytes(
+    url: str,
+    *,
+    purpose: str = "download",
+    max_size: int = DEFAULT_MAX_DOWNLOAD_SIZE,
+    allowed_mime_types: set[str] | None = None,
+    timeout: float = 60.0,
+) -> bytes:
+    """安全下载 URL 并返回字节内容。
+
+    防护同 safe_download_file，但结果返回在内存中（适合小文件）。
+    """
+    import tempfile
+
+    fd, tmp_path = tempfile.mkstemp()
+    os.close(fd)
+
+    try:
+        safe_download_file(
+            url,
+            tmp_path,
+            purpose=purpose,
+            max_size=max_size,
+            allowed_mime_types=allowed_mime_types,
+            timeout=timeout,
+        )
+        with open(tmp_path, "rb") as f:
+            return f.read()
+    finally:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass

pyproject.toml

@@ -48,23 +48,55 @@ omit = [
 ]
 branch = true
 
-[tool.coverage.report]
-exclude_lines = [
-    "pragma: no cover",
-    "def __repr__",
-    "if __name__ == .__main__.:",
-    "raise NotImplementedError",
-    "pass",
-    "if TYPE_CHECKING:",
-    "class .*Protocol",
-    "@abstractmethod",
-    "raise AssertionError",
-    "raise RuntimeError",
-    "if 0:",
-    "if __debug__:",
+[tool.ruff]
+target-version = "py311"
+line-length = 120
+exclude = [
+  ".git",
+  ".cache",
+  "__pycache__",
+  ".venv",
+  "venv",
+  "node_modules",
+  "alembic",
+  ".gitea",
+  ".next",
+  "dist",
+  "build",
 ]
-show_missing = true
-skip_covered = false
 
-[tool.coverage.xml]
-output = "coverage.xml"
+[tool.ruff.lint]
+# 当前阶段：摸底模式，规则集与原flake8对齐
+# 后续迭代计划：
+#   Phase 1: 修完 bugbear 后正式替换 flake8
+#   Phase 2: 启用 UP(pyupgrade) + SIM(simplify)
+#   Phase 3: 启用 RET(return) + ARG(unused-args)
+select = [
+  "E",   # pycodestyle errors（同flake8）
+  "F",   # pyflakes（同flake8）
+  "W",   # pycodestyle warnings（同flake8）
+  "B",   # flake8-bugbear（新增，摸底用）
+]
+# 与原 setup.cfg flake8 配置对齐，确保不新增阻断
+ignore = [
+  "E203",
+  "W503",
+  "E501",  # line-too-long（black管）
+  "E302",
+  "E402",  # module-import-not-at-top（循环导入多）
+  "E722",  # bare-except
+  "W291",
+  "W293",
+  "F401",  # unused-import
+  "F403",
+  "F405",
+  "F841",  # unused-variable
+  "B008",  # do-not-perform-callback-from-arg（fastapi依赖注入）
+]
+
+[tool.ruff.lint.per-file-ignores]
+"__init__.py" = ["F401", "F403", "F405"]
+"tests/*" = ["E402", "F401", "F841"]
+"packages/ports/*" = ["E301", "E704"]
+"apps/*/migrations/*" = ["ALL"]
+"alembic/*" = ["ALL"]

requirements-base.txt

@@ -13,7 +13,7 @@ uvicorn[standard]==0.32.0
 pydantic==2.9.0
 
 # 认证核心
-pyjwt==2.9.0
+pyjwt==2.13.0
 bcrypt==4.2.0
 
 # Redis

scripts/generate_changelog.py

@@ -0,0 +1,183 @@
+#!/usr/bin/env python3
+"""
+自动生成 CHANGELOG 条目。
+
+用法:
+    python3 scripts/generate_changelog.py v0.1.128 v0.1.129
+    python3 scripts/generate_changelog.py v0.1.128 HEAD
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import sys
+import urllib.error
+import urllib.request
+from datetime import datetime
+
+GITEA_URL = os.environ.get("GITEA_URL", "https://git.xiaoxiajianji.com")
+REPO = os.environ.get("GITEA_REPO", "xiaoxia/xiaoxia-saas")
+TOKEN = os.environ.get("GITEA_TOKEN", "")
+
+
+def gitea_api(path: str) -> dict | list:
+    url = f"{GITEA_URL}/api/v1{path}"
+    req = urllib.request.Request(url)
+    if TOKEN:
+        req.add_header("Authorization", f"token {TOKEN}")
+    try:
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            return json.loads(resp.read().decode())
+    except urllib.error.HTTPError as e:
+        print(f"API Error: {e.code} {e.reason}", file=sys.stderr)
+        raise
+
+
+def get_tag_date(tag: str) -> str:
+    try:
+        info = gitea_api(f"/repos/{REPO}/git/refs/tags/{tag}")
+        if isinstance(info, dict):
+            sha = info.get("object", {}).get("sha", "")
+            if sha:
+                commit = gitea_api(f"/repos/{REPO}/git/commits/{sha}")
+                if isinstance(commit, dict):
+                    return commit.get("committer", {}).get("date", "")[:10]
+    except Exception:
+        pass
+    return ""
+
+
+def get_merged_prs_between(from_tag: str, to_tag: str) -> list[dict]:
+    all_prs: list[dict] = []
+    page = 1
+    while True:
+        prs = gitea_api(f"/repos/{REPO}/pulls?state=closed&sort=merged&direction=desc" f"&per_page=50&page={page}")
+        if not isinstance(prs, list) or not prs:
+            break
+        all_prs.extend(prs)
+        if len(prs) < 50:
+            break
+        page += 1
+        if page > 10:
+            break
+
+    merged = [pr for pr in all_prs if pr.get("merged_at")]
+    from_date = get_tag_date(from_tag)
+    to_date = get_tag_date(to_tag) if not to_tag.startswith("HEAD") else datetime.now().strftime("%Y-%m-%d")
+
+    if not from_date:
+        return merged[:50]
+
+    result = []
+    for pr in merged:
+        merged_at = pr.get("merged_at", "")[:10]
+        if from_date <= merged_at <= to_date:
+            result.append(pr)
+    return result
+
+
+def categorize_pr(title: str) -> tuple[str, str]:
+    title = title.strip()
+    lower = title.lower()
+
+    m = re.match(r"^(feat|fix|chore|perf|docs|refactor|test|ci|style|build|security)\s*[:：]", title)
+    if m:
+        prefix = m.group(1)
+        clean_title = title[m.end() :].strip()
+    else:
+        prefix = ""
+        clean_title = title
+
+    if prefix in ("feat", "feature"):
+        return "✨ 功能", clean_title
+    elif prefix == "fix":
+        return "🐛 Bug 修复", clean_title
+    elif prefix in ("refactor", "chore", "style"):
+        return "🔄 重构与清理", clean_title
+    elif prefix in ("perf", "performance"):
+        return "⚡ 性能优化", clean_title
+    elif prefix == "security":
+        return "🔒 安全修复", clean_title
+    elif prefix == "docs":
+        return "📝 文档", clean_title
+    elif prefix == "test":
+        return "🧪 测试", clean_title
+    elif prefix in ("ci", "build"):
+        return "🚀 CI/CD & 基础设施", clean_title
+    else:
+        if any(k in lower for k in ["安全", "security", "cve", "漏洞"]):
+            return "🔒 安全修复", clean_title
+        elif any(k in lower for k in ["修复", "bug"]):
+            return "🐛 Bug 修复", clean_title
+        elif any(k in lower for k in ["新增", "添加", "feat", "功能"]):
+            return "✨ 功能", clean_title
+        elif any(k in lower for k in ["ci", "构建", "workflow", "pipeline"]):
+            return "🚀 CI/CD & 基础设施", clean_title
+        elif any(k in lower for k in ["测试", "test", "e2e"]):
+            return "🧪 测试", clean_title
+        else:
+            return "📌 其他", clean_title
+
+
+def generate_changelog(from_tag: str, to_tag: str, version: str = "") -> str:
+    if not version:
+        version = to_tag
+
+    prs = get_merged_prs_between(from_tag, to_tag)
+
+    categories: dict[str, list[tuple[int, str]]] = {}
+    for pr in prs:
+        cat, title = categorize_pr(pr["title"])
+        pr_num = pr["number"]
+        categories.setdefault(cat, []).append((pr_num, title))
+
+    order = [
+        "🔒 安全修复",
+        "✨ 功能",
+        "🐛 Bug 修复",
+        "⚡ 性能优化",
+        "🔄 重构与清理",
+        "📝 文档",
+        "🧪 测试",
+        "🚀 CI/CD & 基础设施",
+        "📌 其他",
+    ]
+
+    date_str = get_tag_date(to_tag) if not to_tag.startswith("HEAD") else datetime.now().strftime("%Y-%m-%d")
+    lines = [f"## [{version}] - {date_str}", ""]
+
+    for cat in order:
+        items = categories.get(cat, [])
+        if not items:
+            continue
+        lines.append(f"### {cat}")
+        lines.append("")
+        for num, title in sorted(items, key=lambda x: x[0]):
+            short_title = title.split(" — ")[0].split(" - ")[0]
+            if len(short_title) > 80:
+                short_title = short_title[:77] + "..."
+            lines.append(f"- #{num} {short_title}")
+        lines.append("")
+
+    lines.append("---")
+    lines.append("")
+    return "\n".join(lines)
+
+
+def main():
+    if len(sys.argv) < 3:
+        print(f"用法: {sys.argv[0]} <from_tag> <to_tag> [version]")
+        sys.exit(1)
+
+    from_tag = sys.argv[1]
+    to_tag = sys.argv[2]
+    version = sys.argv[3] if len(sys.argv) > 3 else ""
+
+    changelog = generate_changelog(from_tag, to_tag, version)
+    print(changelog)
+
+
+if __name__ == "__main__":
+    main()

scripts/gray_deploy.sh

@@ -1,9 +1,13 @@
 #!/bin/bash
-# 灰度发布脚本：通过Nginx权重调整流量比例
+# 灰度发布脚本：在生产服务器上启动 canary 版本，通过 Nginx 权重切流
 # 用法: ./scripts/gray_deploy.sh <版本号> <灰度百分比>
 #
-# 需要在目标服务器上执行，或通过SSH执行
-# 前提：服务器上运行两个版本的容器（stable + canary），Nginx做加权轮询
+# 前提：
+#   - 在生产服务器上执行（或通过 SSH 管道执行）
+#   - 当前已有全量运行的 production 容器
+#   - Nginx 配置在 /etc/nginx/sites-enabled/00-xiaoxia-saas
+#
+# 灰度范围：API + Web（Worker 暂时全量升级，队列消费无法按比例切流）
 
 set -euo pipefail
 
@@ -12,106 +16,212 @@ GRAY_PCT="${2:-10}"
 
 if [[ -z "$VERSION" ]]; then
     echo "用法: $0 <版本号> [灰度百分比]"
-    echo "示例: $0 v0.1.129 5"
+    echo "示例: $0 v0.1.130 5"
     exit 1
 fi
 
-STABLE_VERSION="${STABLE_VERSION:-current}"
 REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
+NGINX_CONF="${NGINX_CONF:-/etc/nginx/sites-enabled/00-xiaoxia-saas}"
+ENV_FILE="${ENV_FILE:-/var/lib/xiaoxia-saas-production/.env}"
+GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-production/generated}"
+LEGACY_ASSETS_DIR="${LEGACY_ASSETS_DIR:-/var/lib/xiaoxia-saas-production/legacy-assets}"
 
-echo "=========================================="
+# Canary 端口（与 production 错开）
+CANARY_API_PORT=18001
+CANARY_WEB_PORT=13002
+
+echo "============================================"
 echo "  灰度发布"
 echo "  新版本: $VERSION"
 echo "  灰度比例: ${GRAY_PCT}%"
-echo "  稳定版本: $STABLE_VERSION"
-echo "=========================================="
+echo "  Canary API 端口: $CANARY_API_PORT"
+echo "  Canary Web 端口: $CANARY_WEB_PORT"
+echo "============================================"
 
-# 1. 拉取新版本镜像
+# 1. 检查环境
+if [[ ! -f "$ENV_FILE" ]]; then
+    echo "错误: 环境文件不存在: $ENV_FILE"
+    exit 1
+fi
+
+if [[ ! -f "$NGINX_CONF" ]]; then
+    echo "错误: Nginx 配置不存在: $NGINX_CONF"
+    exit 1
+fi
+
+# 2. 拉取新版本镜像
 echo ""
 echo ">>> 拉取新版本镜像..."
-for component in api worker web; do
+for component in api web worker; do
     echo "  拉取 $component:$VERSION ..."
     docker pull "${REGISTRY}-${component}:${VERSION}" 2>&1 | tail -1
 done
+echo "  ✅ 镜像拉取完成"
 
-# 2. 启动灰度版本容器（canary）
+# 3. 启动 API Canary
 echo ""
-echo ">>> 启动灰度版本容器..."
-
-# API canary
-CANARY_API_NAME="saas-api-canary"
-if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_API_NAME}$"; then
-    echo "  停止旧 canary 容器..."
-    docker stop "$CANARY_API_NAME" 2>/dev/null || true
-    docker rm "$CANARY_API_NAME" 2>/dev/null || true
+echo ">>> 启动 API Canary 容器..."
+CANARY_API="xiaoxia-api-canary"
+if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_API}$"; then
+    echo "  停止旧 canary..."
+    docker rm -f "$CANARY_API" >/dev/null 2>&1
 fi
 
-echo "  启动 api canary..."
 docker run -d \
-    --name "$CANARY_API_NAME" \
-    --network saas-network \
-    -e DATABASE_URL="${DATABASE_URL}" \
-    -e REDIS_URL="${REDIS_URL}" \
-    -e FEATURE_FLAG_PROVIDER=redis \
+    --name "$CANARY_API" \
+    --env-file "$ENV_FILE" \
+    --network xiaoxia-net-production \
+    -p "127.0.0.1:${CANARY_API_PORT}:8000" \
+    -e APP_ENV=production \
+    -e APP_VERSION="$VERSION-canary" \
+    -e GENERATED_FILES_DIR=/app/generated \
+    -e GENERATED_FILES_URL_PREFIX=/generated-files \
+    -e PUBLIC_API_BASE_URL=https://saas-api.xiaoxiajianji.com \
+    -v "${GENERATED_DIR}:/app/generated" \
     --restart unless-stopped \
-    "${REGISTRY}-api:${VERSION}"
+    --cpus 1 \
+    --memory 1g \
+    --health-cmd "python -c \"import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=5)\"" \
+    --health-interval 30s \
+    --health-timeout 10s \
+    --health-retries 3 \
+    --health-start-period 40s \
+    --log-driver json-file \
+    --log-opt max-size=50m \
+    --log-opt max-file=3 \
+    "${REGISTRY}-api:${VERSION}" >/dev/null
 
-# Worker canary
-CANARY_WORKER_NAME="saas-worker-canary"
-if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_WORKER_NAME}$"; then
-    echo "  停止旧 worker canary..."
-    docker stop "$CANARY_WORKER_NAME" 2>/dev/null || true
-    docker rm "$CANARY_WORKER_NAME" 2>/dev/null || true
+echo "  ✅ API Canary 已启动（端口 $CANARY_API_PORT）"
+
+# 4. 启动 Web Canary
+echo ""
+echo ">>> 启动 Web Canary 容器..."
+CANARY_WEB="xiaoxia-web-canary"
+if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_WEB}$"; then
+    echo "  停止旧 canary..."
+    docker rm -f "$CANARY_WEB" >/dev/null 2>&1
 fi
 
-echo "  启动 worker canary..."
-docker run -d \
-    --name "$CANARY_WORKER_NAME" \
-    --network saas-network \
-    -e DATABASE_URL="${DATABASE_URL}" \
-    -e REDIS_URL="${REDIS_URL}" \
-    --restart unless-stopped \
-    "${REGISTRY}-worker:${VERSION}"
+LEGACY_VOLUME=""
+if [[ -d "$LEGACY_ASSETS_DIR" ]] && [[ -n "$(ls -A "$LEGACY_ASSETS_DIR" 2>/dev/null)" ]]; then
+    LEGACY_VOLUME="-v ${LEGACY_ASSETS_DIR}:/usr/share/nginx/html/assets-legacy/assets:ro"
+fi
 
-# 3. 等待容器健康
+docker run -d \
+    --name "$CANARY_WEB" \
+    --network xiaoxia-net-production \
+    -p "127.0.0.1:${CANARY_WEB_PORT}:80" \
+    --restart unless-stopped \
+    --cpus 0.5 \
+    --memory 256m \
+    $LEGACY_VOLUME \
+    --health-cmd "wget --spider -q http://127.0.0.1:80" \
+    --health-interval 30s \
+    --health-timeout 5s \
+    --health-retries 3 \
+    --health-start-period 10s \
+    --log-driver json-file \
+    --log-opt max-size=50m \
+    --log-opt max-file=3 \
+    "${REGISTRY}-web:${VERSION}" >/dev/null
+
+echo "  ✅ Web Canary 已启动（端口 $CANARY_WEB_PORT）"
+
+# 5. 等待健康检查
 echo ""
-echo ">>> 等待容器健康..."
-sleep 10
-if ! docker ps --format '{{.Names}} {{.Status}}' | grep -q "$CANARY_API_NAME"; then
-    echo "错误: API canary 容器未运行"
-    docker logs "$CANARY_API_NAME" --tail 20
+echo ">>> 等待 Canary 容器健康..."
+for i in $(seq 1 40); do
+    api_healthy=$(docker inspect --format='{{.State.Health.Status}}' "$CANARY_API" 2>/dev/null || echo "starting")
+    web_healthy=$(docker inspect --format='{{.State.Health.Status}}' "$CANARY_WEB" 2>/dev/null || echo "starting")
+    
+    if [[ "$api_healthy" == "healthy" && "$web_healthy" == "healthy" ]]; then
+        echo "  ✅ API + Web Canary 均健康（用时 ${i}s）"
+        break
+    fi
+    
+    if [[ "$api_healthy" == "unhealthy" ]]; then
+        echo "  ❌ API Canary 健康检查失败"
+        docker logs --tail 30 "$CANARY_API"
+        exit 1
+    fi
+    if [[ "$web_healthy" == "unhealthy" ]]; then
+        echo "  ❌ Web Canary 健康检查失败"
+        docker logs --tail 20 "$CANARY_WEB"
+        exit 1
+    fi
+    
+    sleep 3
+done
+
+# 6. 更新 Nginx 配置 - 添加 upstream 权重
+echo ""
+echo ">>> 更新 Nginx 权重（稳定: $((100-GRAY_PCT))% / 灰度: ${GRAY_PCT}%）..."
+
+# 备份
+BAK_FILE="${NGINX_CONF}.bak.gray.$(date +%Y%m%d%H%M%S)"
+cp "$NGINX_CONF" "$BAK_FILE"
+echo "  已备份: $BAK_FILE"
+
+# 生成 upstream 块
+UPSTREAM_BLOCK="
+# Gray release upstreams（自动生成 - gray_deploy.sh）
+upstream saas_api_backend {
+    server 127.0.0.1:8001 weight=$((100-GRAY_PCT));
+    server 127.0.0.1:${CANARY_API_PORT} weight=${GRAY_PCT};
+}
+
+upstream saas_web_backend {
+    server 127.0.0.1:3002 weight=$((100-GRAY_PCT));
+    server 127.0.0.1:${CANARY_WEB_PORT} weight=${GRAY_PCT};
+}
+"
+
+# 在文件最前面插入 upstream 块
+TMP_CONF=$(mktemp)
+{
+    echo "$UPSTREAM_BLOCK"
+    cat "$NGINX_CONF"
+} > "$TMP_CONF"
+
+# 替换 proxy_pass 指向 upstream
+# API: proxy_pass http://127.0.0.1:8001 -> proxy_pass http://saas_api_backend
+sed -i 's|proxy_pass http://127\.0\.0\.1:8001|proxy_pass http://saas_api_backend|g' "$TMP_CONF"
+# Web: proxy_pass http://127.0.0.1:3002/ -> proxy_pass http://saas_web_backend/
+sed -i 's|proxy_pass http://127\.0\.0\.1:3002/|proxy_pass http://saas_web_backend/|g' "$TMP_CONF"
+
+# 测试配置
+mv "$TMP_CONF" "$NGINX_CONF"
+if ! nginx -t 2>&1; then
+    echo "  ❌ Nginx 配置测试失败，回滚..."
+    cp "$BAK_FILE" "$NGINX_CONF"
+    nginx -t
     exit 1
 fi
-echo "  ✅ API canary 运行中"
 
-# 4. 更新Nginx权重
+nginx -s reload
+echo "  ✅ Nginx 已 reload，灰度生效"
+
+# 7. 验证灰度流量
 echo ""
-echo ">>> 更新Nginx权重 (稳定: $((100-GRAY_PCT))% / 灰度: ${GRAY_PCT}%)..."
-
-NGINX_CONF="${NGINX_CONF:-/etc/nginx/conf.d/saas-api.conf}"
-if [[ -f "$NGINX_CONF" ]]; then
-    # 备份
-    cp "$NGINX_CONF" "${NGINX_CONF}.bak.$(date +%Y%m%d%H%M%S)"
-    
-    # 更新 upstream 权重（需要根据实际配置调整）
-    echo "  请手动更新 Nginx upstream 配置中的权重"
-    echo "  示例配置:"
-    cat <<EOF
-    upstream saas_api_backend {
-        server saas-api:8000 weight=$((100-GRAY_PCT));
-        server saas-api-canary:8000 weight=${GRAY_PCT};
-    }
-EOF
-    nginx -t && nginx -s reload
-    echo "  ✅ Nginx 已reload"
-else
-    echo "  警告: Nginx 配置文件不存在 ($NGINX_CONF)"
-    echo "  请手动配置灰度流量权重"
-fi
+echo ">>> 验证灰度流量..."
+gray_hits=0
+total_hits=20
+for i in $(seq 1 $total_hits); do
+    resp=$(curl -s -o /dev/null -w "%{http_code}" -H "X-Gray-Test: 1" http://127.0.0.1:${CANARY_API_PORT}/health 2>/dev/null || echo "000")
+    if [[ "$resp" == "200" ]]; then
+        gray_hits=$((gray_hits + 1))
+    fi
+    sleep 0.1
+done
+echo "  Canary 健康验证: $gray_hits/$total_hits 请求成功"
 
 echo ""
-echo "=========================================="
+echo "============================================"
 echo "  ✅ 灰度发布完成"
-echo "  新版本: $VERSION (${GRAY_PCT}%流量)"
-echo "  监控: Grafana / 日志"
-echo "=========================================="
+echo "  版本: $VERSION (${GRAY_PCT}%流量)"
+echo "  API: 127.0.0.1:$CANARY_API_PORT"
+echo "  Web: 127.0.0.1:$CANARY_WEB_PORT"
+echo "  Nginx 备份: $BAK_FILE"
+echo "  回滚: ./scripts/rollback.sh"
+echo "  Worker: 暂不灰度（队列消费无法按比例切流）"
+echo "============================================"

scripts/release.sh

@@ -1,6 +1,11 @@
 #!/bin/bash
-# 一键发布脚本：打tag → 触发生产镜像构建 → 部署到灰度
-# 用法: ./scripts/release.sh v0.1.129 [--gray 5]
+# 一键发布脚本：打 tag → 触发 CI 构建 → 可选灰度发布
+# 用法: ./scripts/release.sh v0.1.130 [--gray 5]
+#
+# 说明：
+#   - 打 tag 后 CI 会自动构建镜像并全量部署到生产
+#   - 加 --gray 参数则在构建完成后执行灰度切流（需 SSH 到生产服务器执行）
+#   - 加 --no-deploy 只打 tag 不触发自动部署
 
 set -euo pipefail
 
@@ -11,13 +16,12 @@ usage() {
     echo "用法: $0 <版本号> [--gray 百分比] [--no-deploy]"
     echo ""
     echo "示例:"
-    echo "  $0 v0.1.129                    # 打tag + 全量发布"
-    echo "  $0 v0.1.129 --gray 5           # 打tag + 5%灰度发布"
-    echo "  $0 v0.1.129 --no-deploy        # 只打tag，不部署"
+    echo "  $0 v0.1.130                  # 打tag + 全量发布（CI自动部署）"
+    echo "  $0 v0.1.130 --gray 5         # 打tag + 5%灰度发布"
+    echo "  $0 v0.1.130 --no-deploy      # 只打tag，不部署"
     exit 1
 }
 
-# 参数解析
 VERSION=""
 GRAY_PCT=0
 DEPLOY=true
@@ -47,80 +51,63 @@ while [[ $# -gt 0 ]]; do
 done
 
 if [[ -z "$VERSION" ]]; then
-    echo "错误: 请指定版本号（如 v0.1.129）"
+    echo "错误: 请指定版本号（如 v0.1.130）"
     usage
 fi
 
-echo "=========================================="
+echo "============================================"
 echo "  发布版本: $VERSION"
 echo "  灰度比例: ${GRAY_PCT}%"
 echo "  自动部署: $DEPLOY"
-echo "=========================================="
+echo "============================================"
 
 cd "$REPO_ROOT"
 
-# 1. 确认在 develop 分支
+# 1. 确认分支
 CURRENT_BRANCH=$(git rev-parse --abbrev-ref HEAD)
 if [[ "$CURRENT_BRANCH" != "develop" ]]; then
-    echo "错误: 请切换到 develop 分支后再发布"
+    echo "错误: 请在 develop 分支上打 tag"
     exit 1
 fi
 
-# 2. 拉取最新代码
+# 2. 拉取最新
 echo ""
 echo ">>> 拉取最新代码..."
 git pull origin develop
 
-# 3. 生成 CHANGELOG
-echo ""
-echo ">>> 生成 CHANGELOG..."
-if [[ -f "scripts/generate_changelog.py" ]]; then
-    PREV_TAG=$(git describe --tags --abbrev=0 HEAD^ 2>/dev/null || echo "")
-    if [[ -n "$PREV_TAG" ]]; then
-        python3 scripts/generate_changelog.py \
-            --from-tag "$PREV_TAG" \
-            --to-tag HEAD \
-            --gitea-token "${GITEA_TOKEN:-}" \
-            --output /tmp/changelog_$$.md
-        echo "CHANGELOG 已生成到 /tmp/changelog_$$.md"
-    fi
-fi
-
-# 4. 打tag
-echo ""
-echo ">>> 打 tag $VERSION ..."
+# 3. 检查 tag 是否已存在
 if git rev-parse "$VERSION" >/dev/null 2>&1; then
-    echo "警告: tag $VERSION 已存在，跳过打tag"
+    echo "警告: tag $VERSION 已存在，跳过打 tag"
 else
+    echo ""
+    echo ">>> 打 tag $VERSION ..."
     git tag -a "$VERSION" -m "Release $VERSION"
     git push origin "$VERSION"
-    echo "Tag $VERSION 已推送，触发生产镜像构建..."
+    echo "  ✅ Tag 已推送，CI 将自动构建生产镜像"
 fi
 
-# 5. 等待镜像构建
+# 4. 部署提示
 if [[ "$DEPLOY" == "true" ]]; then
     echo ""
-    echo ">>> 等待镜像构建完成（约10-15分钟）..."
-    echo "    镜像: git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas-{api,worker,web}:$VERSION"
-    
-    # 这里可以加镜像存在性检查
-    echo "    （镜像构建由CI自动完成，请在Gitea Actions中确认）"
-fi
-
-# 6. 灰度部署
-if [[ "$DEPLOY" == "true" && "$GRAY_PCT" -gt 0 ]]; then
+    echo ">>> 构建 & 部署"
+    echo "  CI 会自动执行:"
+    echo "    1. Build Production Runtime Images（约10-15分钟）"
+    echo "    2. Deploy Production（SSH 到生产服务器部署）"
     echo ""
-    echo ">>> 灰度部署: ${GRAY_PCT}% 流量到 $VERSION"
-    if [[ -f "scripts/gray_deploy.sh" ]]; then
-        ./scripts/gray_deploy.sh "$VERSION" "$GRAY_PCT"
-    else
-        echo "警告: gray_deploy.sh 不存在，跳过灰度部署"
+    echo "  查看进度: Gitea Actions → 对应 tag 的 run"
+    
+    if [[ "$GRAY_PCT" -gt 0 ]]; then
+        echo ""
+        echo ">>> 灰度发布"
+        echo "  构建部署完成后，在生产服务器上执行:"
+        echo "    cd /var/lib/xiaoxia-saas-production"
+        echo "    ./gray_deploy.sh $VERSION $GRAY_PCT"
     fi
 fi
 
 echo ""
-echo "=========================================="
-echo "  ✅ 发布流程完成"
+echo "============================================"
+echo "  ✅ 发布流程触发完成"
 echo "  版本: $VERSION"
 echo "  灰度: ${GRAY_PCT}%"
-echo "=========================================="
+echo "============================================"

scripts/rollback_gray.sh

@@ -0,0 +1,77 @@
+#!/bin/bash
+# 灰度回滚脚本：切回全量稳定版本，停止 canary 容器
+# 用法: ./scripts/rollback_gray.sh
+
+set -euo pipefail
+
+NGINX_CONF="${NGINX_CONF:-/etc/nginx/sites-enabled/00-xiaoxia-saas}"
+CANARY_API="${CANARY_API:-xiaoxia-api-canary}"
+CANARY_WEB="${CANARY_WEB:-xiaoxia-web-canary}"
+
+echo "============================================"
+echo "  灰度回滚"
+echo "  目标: 全量切回稳定版本"
+echo "============================================"
+
+# 1. 找最近的灰度备份
+echo ""
+echo ">>> 查找最近的灰度备份..."
+LATEST_BAK=$(ls -t "${NGINX_CONF}".bak.gray.* 2>/dev/null | head -1 || true)
+
+if [[ -z "$LATEST_BAK" ]]; then
+    echo "  未找到灰度备份，尝试手动移除 upstream 配置..."
+    
+    # 手动回滚：移除 upstream 块，把 proxy_pass 改回 127.0.0.1
+    TMP_CONF=$(mktemp)
+    
+    # 移除 upstream 块（从 "# Gray release upstreams" 到空行结束）
+    awk '
+        /^# Gray release upstreams/ { skip=1; next }
+        skip && /^$/ && !found_first_empty { found_first_empty=1; next }
+        skip && found_first_empty && /^$/ { skip=0; found_first_empty=0; next }
+        skip { next }
+        { print }
+    ' "$NGINX_CONF" > "$TMP_CONF"
+    
+    # 把 upstream 名改回 IP
+    sed -i 's|proxy_pass http://saas_api_backend|proxy_pass http://127.0.0.1:8001|g' "$TMP_CONF"
+    sed -i 's|proxy_pass http://saas_web_backend/|proxy_pass http://127.0.0.1:3002/|g' "$TMP_CONF"
+    
+    mv "$TMP_CONF" "$NGINX_CONF"
+else
+    echo "  从备份恢复: $LATEST_BAK"
+    cp "$LATEST_BAK" "$NGINX_CONF"
+fi
+
+# 2. 测试并 reload nginx
+echo ""
+echo ">>> Nginx 测试 & reload..."
+if ! nginx -t 2>&1; then
+    echo "  ❌ Nginx 配置测试失败！请检查"
+    exit 1
+fi
+nginx -s reload
+echo "  ✅ Nginx 已回滚，全量切回稳定版本"
+
+# 3. 停止 canary 容器（延迟停止，保留30分钟便于排查）
+echo ""
+echo ">>> Canary 容器将在30分钟后停止（便于排查）"
+echo "    立即停止请执行: docker rm -f $CANARY_API $CANARY_WEB"
+
+# 30分钟后停止（后台执行，不阻塞脚本）
+(
+    sleep 1800
+    for c in "$CANARY_API" "$CANARY_WEB"; do
+        if docker ps --format '{{.Names}}' | grep -q "^${c}$"; then
+            docker stop "$c" >/dev/null 2>&1 && docker rm "$c" >/dev/null 2>&1
+            echo "[$(date)] 已停止 canary 容器: $c"
+        fi
+    done
+) &
+
+echo ""
+echo "============================================"
+echo "  ✅ 灰度回滚完成"
+echo "  流量已全部切回稳定版本"
+echo "  Canary 容器: 30分钟后自动清理"
+echo "============================================"

tests/unit/test_path_security.py

@@ -0,0 +1,243 @@
+"""路径安全校验工具单元测试 — 路径遍历防护."""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "worker"))
+
+from video_processing.path_security import (  # noqa: E402
+    PathSecurityError,
+    get_allowed_local_dirs,
+    is_in_allowed_dirs,
+    is_path_safe,
+    safe_resolve_path,
+    sanitize_filename,
+    validate_local_schema_path,
+)
+
+
+class TestSafeResolvePath(unittest.TestCase):
+    """安全路径解析测试."""
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        import shutil
+
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    # ── 正常路径 ─────────────────────────────────────────────────────────
+
+    def test_simple_relative_path(self):
+        """简单相对路径应该正常解析."""
+        result = safe_resolve_path("test.mp4", self.tmpdir)
+        self.assertEqual(result.name, "test.mp4")
+        self.assertTrue(str(result).startswith(self.tmpdir))
+
+    def test_subdirectory_path(self):
+        """子目录路径应该正常解析."""
+        result = safe_resolve_path("sub/dir/file.mp4", self.tmpdir)
+        self.assertTrue(str(result).startswith(self.tmpdir))
+        self.assertIn("sub/dir/file.mp4", str(result).replace("\\", "/"))
+
+    def test_dot_slash_path(self):
+        """./ 开头的路径应该正常解析."""
+        result = safe_resolve_path("./test.mp4", self.tmpdir)
+        self.assertEqual(result.name, "test.mp4")
+
+    # ── 路径遍历防护 ─────────────────────────────────────────────────────
+
+    def test_parent_traversal_rejected(self):
+        """../ 路径遍历应该被拒绝."""
+        with self.assertRaises(PathSecurityError):
+            safe_resolve_path("../etc/passwd", self.tmpdir)
+
+    def test_multiple_parent_traversal_rejected(self):
+        """多级 ../ 遍历应该被拒绝."""
+        with self.assertRaises(PathSecurityError):
+            safe_resolve_path("../../etc/passwd", self.tmpdir)
+
+    def test_mixed_traversal_rejected(self):
+        """混合路径遍历应该被拒绝."""
+        with self.assertRaises(PathSecurityError):
+            safe_resolve_path("./sub/../../etc/shadow", self.tmpdir)
+
+    def test_absolute_path_rejected(self):
+        """绝对路径（超出基目录）应该被拒绝."""
+        with self.assertRaises(PathSecurityError):
+            safe_resolve_path("/etc/passwd", self.tmpdir)
+
+    # ── 空字节注入 ───────────────────────────────────────────────────────
+
+    def test_null_byte_rejected(self):
+        """空字节注入应该被拒绝."""
+        with self.assertRaises(PathSecurityError):
+            safe_resolve_path("test\x00.mp4", self.tmpdir)
+
+    # ── 空路径 ──────────────────────────────────────────────────────────
+
+    def test_empty_path_rejected(self):
+        """空路径应该被拒绝."""
+        with self.assertRaises(PathSecurityError):
+            safe_resolve_path("", self.tmpdir)
+
+    def test_none_path_rejected(self):
+        """None 路径应该被拒绝."""
+        with self.assertRaises(PathSecurityError):
+            safe_resolve_path(None, self.tmpdir)  # type: ignore
+
+    def test_whitespace_path_rejected(self):
+        """空白路径应该被拒绝."""
+        with self.assertRaises(PathSecurityError):
+            safe_resolve_path("   ", self.tmpdir)
+
+    # ── 路径长度 ────────────────────────────────────────────────────────
+
+    def test_too_long_path_rejected(self):
+        """超长路径应该被拒绝."""
+        long_path = "a" * 5000 + ".mp4"
+        with self.assertRaises(PathSecurityError):
+            safe_resolve_path(long_path, self.tmpdir)
+
+    # ── 系统路径防护 ─────────────────────────────────────────────────────
+
+    def test_proc_path_rejected_when_absolute(self):
+        """/proc/ 路径在绝对路径模式下应该被拒绝（因为超出基目录）."""
+        with self.assertRaises(PathSecurityError):
+            safe_resolve_path("/proc/self/environ", self.tmpdir)
+
+    # ── 扩展名校验 ───────────────────────────────────────────────────────
+
+    def test_extension_whitelist_pass(self):
+        """白名单内的扩展名应该通过."""
+        result = safe_resolve_path(
+            "test.mp4",
+            self.tmpdir,
+            allowed_extensions={".mp4", ".mov"},
+        )
+        self.assertEqual(result.suffix.lower(), ".mp4")
+
+    def test_extension_whitelist_reject(self):
+        """白名单外的扩展名应该被拒绝."""
+        with self.assertRaises(PathSecurityError):
+            safe_resolve_path(
+                "test.exe",
+                self.tmpdir,
+                allowed_extensions={".mp4", ".mov"},
+            )
+
+
+class TestLocalSchemaPath(unittest.TestCase):
+    """local:// schema 路径测试."""
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        import shutil
+
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def test_valid_local_schema(self):
+        """有效的 local:// 相对路径应该通过."""
+        # 创建测试文件
+        test_file = Path(self.tmpdir) / "test.mp4"
+        test_file.touch()
+
+        result = validate_local_schema_path("local://test.mp4", self.tmpdir)
+        self.assertTrue(result.exists())
+
+    def test_local_schema_absolute_rejected(self):
+        """local:// + 绝对路径应该被拒绝."""
+        with self.assertRaises(PathSecurityError):
+            validate_local_schema_path("local:///etc/passwd", self.tmpdir)
+
+    def test_local_schema_traversal_rejected(self):
+        """local:// + 路径遍历应该被拒绝."""
+        with self.assertRaises(PathSecurityError):
+            validate_local_schema_path("local://../etc/passwd", self.tmpdir)
+
+    def test_non_local_schema_rejected(self):
+        """非 local:// 开头的路径应该被拒绝."""
+        with self.assertRaises(PathSecurityError):
+            validate_local_schema_path("http://example.com/test", self.tmpdir)
+
+
+class TestSanitizeFilename(unittest.TestCase):
+    """文件名清理测试."""
+
+    def test_normal_filename(self):
+        """正常文件名应该保持不变."""
+        self.assertEqual(sanitize_filename("video.mp4"), "video.mp4")
+
+    def test_path_separators_removed(self):
+        """路径分隔符应该被替换."""
+        self.assertNotIn("/", sanitize_filename("../path/to/file.mp4"))
+        self.assertNotIn("\\", sanitize_filename("..\\path\\file.mp4"))
+
+    def test_leading_dots_removed(self):
+        """开头的点应该被移除."""
+        result = sanitize_filename(".hidden")
+        self.assertFalse(result.startswith("."))
+        self.assertEqual(result, "hidden")
+
+    def test_multiple_leading_dots_removed(self):
+        """多个开头的点应该全部被移除."""
+        result = sanitize_filename("...hidden")
+        self.assertFalse(result.startswith("."))
+
+    def test_empty_filename_default(self):
+        """空文件名应该返回 unnamed."""
+        self.assertEqual(sanitize_filename(""), "unnamed")
+
+    def test_special_chars_removed(self):
+        """特殊字符应该被替换."""
+        result = sanitize_filename('file<name>:"test|?*.mp4')
+        self.assertNotIn("<", result)
+        self.assertNotIn(">", result)
+        self.assertNotIn(":", result)
+        self.assertNotIn('"', result)
+        self.assertNotIn("|", result)
+        self.assertNotIn("?", result)
+        self.assertNotIn("*", result)
+
+    def test_chinese_filename_preserved(self):
+        """中文文件名应该保留."""
+        result = sanitize_filename("视频素材.mp4")
+        self.assertIn("视频素材", result)
+
+    def test_long_filename_truncated(self):
+        """超长文件名应该被截断."""
+        long_name = "a" * 300 + ".mp4"
+        result = sanitize_filename(long_name)
+        self.assertLessEqual(len(result), 255)
+        self.assertTrue(result.endswith(".mp4"))
+
+
+class TestAllowedDirs(unittest.TestCase):
+    """允许目录配置测试."""
+
+    def test_get_allowed_dirs_returns_list(self):
+        """get_allowed_local_dirs 应该返回列表."""
+        dirs = get_allowed_local_dirs()
+        self.assertIsInstance(dirs, list)
+
+    def test_is_in_allowed_dirs_tmp(self):
+        """/tmp 应该在默认允许目录内."""
+        self.assertTrue(is_in_allowed_dirs("/tmp/test.mp4"))
+
+    def test_is_path_safe_convenience(self):
+        """is_path_safe 便捷函数应该正常工作."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            self.assertTrue(is_path_safe("test.mp4", tmpdir))
+            self.assertFalse(is_path_safe("../etc/passwd", tmpdir))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_pr312_security_debt.py

@@ -0,0 +1,484 @@
+"""PR #312 安全债务修复 单元测试.
+
+测试4个P1安全修复：
+1. 多轨道混音：audio_path 路径安全 + 轨道数量上限
+2. 视频拼接：video_path 路径安全 + 段数上限
+3. 字幕渲染：字幕文件路径白名单校验
+"""
+
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+import pytest
+from video_processing.path_security import PathSecurityError
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def work_dir(tmp_path):
+    return tmp_path
+
+
+@pytest.fixture
+def sample_audio(work_dir):
+    """生成一个测试音频文件."""
+    import subprocess
+
+    path = work_dir / "test.aac"
+    subprocess.run(
+        [
+            "ffmpeg",
+            "-y",
+            "-f",
+            "lavfi",
+            "-i",
+            "sine=frequency=440:duration=1:sample_rate=44100",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            str(path),
+        ],
+        capture_output=True,
+        check=True,
+        timeout=30,
+    )
+    return path
+
+
+@pytest.fixture
+def sample_video(work_dir):
+    """生成一个测试视频文件."""
+    import subprocess
+
+    path = work_dir / "test.mp4"
+    subprocess.run(
+        [
+            "ffmpeg",
+            "-y",
+            "-f",
+            "lavfi",
+            "-i",
+            "testsrc=duration=1:size=320x240:rate=30",
+            "-f",
+            "lavfi",
+            "-i",
+            "sine=frequency=440:duration=1:sample_rate=44100",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "ultrafast",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-shortest",
+            str(path),
+        ],
+        capture_output=True,
+        check=True,
+        timeout=60,
+    )
+    return path
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# 1. 多轨道混音安全测试
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestMultiTrackSecurity:
+    """多轨道混音安全测试."""
+
+    def test_track_count_limit_exceeded(self, work_dir, sample_audio):
+        """超过最大轨道数时应截断到上限."""
+        from unittest.mock import MagicMock, patch
+
+        from video_processing.multi_track_mixer import (
+            MAX_AUDIO_TRACKS,
+            AudioTrack,
+            MultiTrackMixConfig,
+            mix_multi_track,
+        )
+
+        # 创建超过上限的轨道数
+        tracks = []
+        for i in range(MAX_AUDIO_TRACKS + 5):
+            tracks.append(
+                AudioTrack(
+                    track_id=f"track_{i}",
+                    track_type="sfx",
+                    audio_path=str(sample_audio),
+                    volume=0.5,
+                )
+            )
+
+        config = MultiTrackMixConfig(tracks=tracks)
+
+        ctx = MagicMock()
+        ctx.work_dir = work_dir
+        ctx.plan_id = "test_plan"
+
+        # mock _prepare_single_track 避免实际跑ffmpeg
+        with patch("video_processing.multi_track_mixer._prepare_single_track", return_value=True):
+            with patch("video_processing.multi_track_mixer.run_ffmpeg"):
+                import shutil
+
+                with patch("shutil.copy2"):
+                    result = mix_multi_track(ctx, sample_audio, config, 10.0)
+
+        # 验证轨道被截断到上限
+        assert len(config.tracks) == MAX_AUDIO_TRACKS
+        assert result is not None
+
+    def test_track_count_within_limit(self, work_dir, sample_audio):
+        """轨道数在限制内时正常处理."""
+        from unittest.mock import MagicMock, patch
+
+        from video_processing.multi_track_mixer import (
+            MAX_AUDIO_TRACKS,
+            AudioTrack,
+            MultiTrackMixConfig,
+            mix_multi_track,
+        )
+
+        tracks = []
+        for i in range(3):
+            tracks.append(
+                AudioTrack(
+                    track_id=f"track_{i}",
+                    track_type="sfx",
+                    audio_path=str(sample_audio),
+                    volume=0.5,
+                )
+            )
+
+        config = MultiTrackMixConfig(tracks=tracks)
+
+        ctx = MagicMock()
+        ctx.work_dir = work_dir
+        ctx.plan_id = "test_plan"
+
+        with patch("video_processing.multi_track_mixer._prepare_single_track", return_value=True):
+            with patch("video_processing.multi_track_mixer.run_ffmpeg"):
+                result = mix_multi_track(ctx, sample_audio, config, 10.0)
+
+        assert len(config.tracks) == 3
+        assert result is not None
+
+    def test_audio_path_traversal_attack(self, work_dir, sample_audio):
+        """路径遍历攻击应被拦截."""
+        from video_processing.multi_track_mixer import _validate_audio_path
+
+        # 路径遍历
+        with pytest.raises(PathSecurityError):
+            _validate_audio_path("../../../etc/passwd", work_dir)
+
+        # local:// 路径遍历
+        with pytest.raises(PathSecurityError):
+            _validate_audio_path("local://../../../etc/passwd", work_dir)
+
+    def test_audio_path_allowed_extension(self, work_dir, sample_audio):
+        """允许的音频扩展名应通过校验."""
+        from video_processing.multi_track_mixer import _validate_audio_path
+
+        # 在work_dir内的音频文件
+        test_file = work_dir / "test.mp3"
+        test_file.touch()
+        _validate_audio_path(str(test_file), work_dir)  # 不应抛异常
+
+        test_file2 = work_dir / "test.wav"
+        test_file2.touch()
+        _validate_audio_path(str(test_file2), work_dir)  # 不应抛异常
+
+    def test_audio_path_disallowed_extension(self, work_dir):
+        """不允许的文件扩展名应被拦截."""
+        from video_processing.multi_track_mixer import _validate_audio_path
+
+        test_file = work_dir / "test.exe"
+        test_file.touch()
+        with pytest.raises(PathSecurityError):
+            _validate_audio_path(str(test_file), work_dir)
+
+        test_file2 = work_dir / "test.php"
+        test_file2.touch()
+        with pytest.raises(PathSecurityError):
+            _validate_audio_path(str(test_file2), work_dir)
+
+    def test_audio_path_empty(self, work_dir):
+        """空路径应被拦截."""
+        from video_processing.multi_track_mixer import _validate_audio_path
+
+        with pytest.raises(PathSecurityError):
+            _validate_audio_path("", work_dir)
+
+        with pytest.raises(PathSecurityError):
+            _validate_audio_path(None, work_dir)
+
+    def test_audio_path_traversal_bypass_startswith(self, work_dir):
+        """【P1绕过】用../构造伪work_dir前缀路径，真实路径逃逸，必须被拦截.
+
+        漏洞：旧代码用 startswith(str(work_dir)) 比原始字符串，
+        /tmp/work/../../opt/secret.aac 会通过 startswith 检查，跳过白名单校验。
+        修复：用 realpath 规范化后再比较。
+        """
+        from video_processing.multi_track_mixer import _validate_audio_path
+
+        evil_path = str(work_dir / "../../../../opt/secret.aac")
+        with pytest.raises(PathSecurityError, match="不在允许目录"):
+            _validate_audio_path(evil_path, work_dir)
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# 2. 视频拼接安全测试
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestConcatSecurity:
+    """视频拼接安全测试."""
+
+    def test_segment_count_limit_exceeded(self, work_dir, sample_video):
+        """超过最大段数时应报错."""
+        from video_processing.concat_engine import (
+            MAX_CONCAT_SEGMENTS,
+            ConcatConfig,
+            ConcatEngine,
+            ConcatSegment,
+        )
+
+        # 创建超过上限的段数
+        segments = []
+        for i in range(MAX_CONCAT_SEGMENTS + 5):
+            segments.append(ConcatSegment(video_path=str(sample_video)))
+
+        config = ConcatConfig(segments=segments)
+        engine = ConcatEngine(work_dir)
+        output_path = work_dir / "output.mp4"
+
+        with pytest.raises(ValueError, match="Too many concat segments"):
+            engine.concat_videos(config, output_path)
+
+    def test_segment_count_within_limit(self, work_dir, sample_video):
+        """段数在限制内时正常处理."""
+        from unittest.mock import patch
+
+        from video_processing.concat_engine import (
+            MAX_CONCAT_SEGMENTS,
+            ConcatConfig,
+            ConcatEngine,
+            ConcatSegment,
+        )
+
+        segments = [
+            ConcatSegment(video_path=str(sample_video)),
+            ConcatSegment(video_path=str(sample_video)),
+            ConcatSegment(video_path=str(sample_video)),
+        ]
+
+        config = ConcatConfig(segments=segments)
+        engine = ConcatEngine(work_dir)
+        output_path = work_dir / "output.mp4"
+
+        # mock ffmpeg执行
+        with patch.object(engine, "_concat_filter", return_value=output_path):
+            with patch.object(engine, "_can_use_stream_copy", return_value=False):
+                result = engine.concat_videos(config, output_path)
+
+        assert result == output_path
+
+    def test_video_path_traversal_attack(self, work_dir):
+        """路径遍历攻击应被拦截."""
+        from video_processing.concat_engine import _validate_video_path
+
+        with pytest.raises(PathSecurityError):
+            _validate_video_path("../../../etc/passwd", work_dir)
+
+        with pytest.raises(PathSecurityError):
+            _validate_video_path("local://../../../etc/passwd", work_dir)
+
+    def test_video_path_allowed_extension(self, work_dir):
+        """允许的视频扩展名应通过校验."""
+        from video_processing.concat_engine import _validate_video_path
+
+        for ext in [".mp4", ".mov", ".avi", ".mkv", ".webm"]:
+            test_file = work_dir / f"test{ext}"
+            test_file.touch()
+            _validate_video_path(str(test_file), work_dir)  # 不应抛异常
+
+    def test_video_path_disallowed_extension(self, work_dir):
+        """不允许的文件扩展名应被拦截."""
+        from video_processing.concat_engine import _validate_video_path
+
+        test_file = work_dir / "test.exe"
+        test_file.touch()
+        with pytest.raises(PathSecurityError):
+            _validate_video_path(str(test_file), work_dir)
+
+        test_file2 = work_dir / "test.js"
+        test_file2.touch()
+        with pytest.raises(PathSecurityError):
+            _validate_video_path(str(test_file2), work_dir)
+
+    def test_video_path_empty(self, work_dir):
+        """空路径应被拦截."""
+        from video_processing.concat_engine import _validate_video_path
+
+        with pytest.raises(PathSecurityError):
+            _validate_video_path("", work_dir)
+
+        with pytest.raises(PathSecurityError):
+            _validate_video_path(None, work_dir)
+
+    def test_video_path_traversal_bypass_startswith(self, work_dir):
+        """【P1绕过】视频路径../遍历绕过startswith检查，必须被拦截.
+
+        漏洞：旧代码用 startswith(str(work_dir)) 比原始字符串，
+        /tmp/work/../../opt/secret.mp4 会通过 startswith 检查，跳过白名单校验。
+        修复：用 realpath 规范化后再比较。
+        """
+        from video_processing.concat_engine import _validate_video_path
+
+        evil_path = str(work_dir / "../../../../opt/secret.mp4")
+        with pytest.raises(PathSecurityError, match="不在允许目录"):
+            _validate_video_path(evil_path, work_dir)
+
+    def test_invalid_segments_skipped(self, work_dir, sample_video):
+        """路径不安全的片段应被跳过."""
+        from unittest.mock import patch
+
+        from video_processing.concat_engine import (
+            ConcatConfig,
+            ConcatEngine,
+            ConcatSegment,
+        )
+
+        segments = [
+            ConcatSegment(video_path=str(sample_video)),
+            ConcatSegment(video_path="../../../etc/passwd"),  # 不安全路径
+            ConcatSegment(video_path=str(sample_video)),
+        ]
+
+        config = ConcatConfig(segments=segments)
+        engine = ConcatEngine(work_dir)
+        output_path = work_dir / "output.mp4"
+
+        with patch.object(engine, "_concat_filter", return_value=output_path):
+            with patch.object(engine, "_can_use_stream_copy", return_value=False):
+                result = engine.concat_videos(config, output_path)
+
+        # 验证只有2个安全片段保留
+        assert len(config.segments) == 2
+        assert result == output_path
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# 3. 字幕渲染安全测试
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestSubtitleSecurity:
+    """字幕渲染安全测试."""
+
+    def test_subtitle_path_traversal_attack(self, work_dir):
+        """路径遍历攻击应被拦截."""
+        from video_processing.subtitle_render_engine import _validate_subtitle_path
+
+        with pytest.raises(PathSecurityError):
+            _validate_subtitle_path("../../../etc/passwd", work_dir)
+
+        with pytest.raises(PathSecurityError):
+            _validate_subtitle_path("local://../../../etc/shadow", work_dir)
+
+    def test_subtitle_path_allowed_extension(self, work_dir):
+        """允许的字幕扩展名应通过校验."""
+        from video_processing.subtitle_render_engine import _validate_subtitle_path
+
+        for ext in [".srt", ".ass", ".vtt", ".sub"]:
+            test_file = work_dir / f"test{ext}"
+            test_file.touch()
+            _validate_subtitle_path(str(test_file), work_dir)  # 不应抛异常
+
+    def test_subtitle_path_disallowed_extension(self, work_dir):
+        """不允许的文件扩展名应被拦截."""
+        from video_processing.subtitle_render_engine import _validate_subtitle_path
+
+        test_file = work_dir / "test.exe"
+        test_file.touch()
+        with pytest.raises(PathSecurityError):
+            _validate_subtitle_path(str(test_file), work_dir)
+
+        test_file2 = work_dir / "test.mp4"
+        test_file2.touch()
+        with pytest.raises(PathSecurityError):
+            _validate_subtitle_path(str(test_file2), work_dir)
+
+    def test_subtitle_remote_url_blocked(self, work_dir):
+        """远程URL字幕应被拦截."""
+        from video_processing.subtitle_render_engine import _validate_subtitle_path
+
+        with pytest.raises(PathSecurityError, match="远程URL"):
+            _validate_subtitle_path("http://evil.com/evil.ass", work_dir)
+
+        with pytest.raises(PathSecurityError, match="远程URL"):
+            _validate_subtitle_path("https://evil.com/evil.srt", work_dir)
+
+    def test_subtitle_path_empty(self, work_dir):
+        """空路径应被拦截."""
+        from video_processing.subtitle_render_engine import _validate_subtitle_path
+
+        with pytest.raises(PathSecurityError):
+            _validate_subtitle_path("", work_dir)
+
+        with pytest.raises(PathSecurityError):
+            _validate_subtitle_path(None, work_dir)
+
+    def test_build_filter_with_safe_path(self, work_dir):
+        """安全路径应正常生成滤镜字符串."""
+        from video_processing.subtitle_render_engine import build_subtitle_filter
+
+        ass_file = work_dir / "subtitle.ass"
+        ass_file.write_text("test", encoding="utf-8")
+
+        result = build_subtitle_filter(ass_file, work_dir=work_dir)
+        assert "subtitles=" in result
+        assert "subtitle.ass" in result
+        assert "[subtitled]" in result
+
+    def test_build_filter_with_unsafe_path_raises(self, work_dir):
+        """不安全路径应抛出异常."""
+        from video_processing.subtitle_render_engine import build_subtitle_filter
+
+        with pytest.raises(PathSecurityError):
+            build_subtitle_filter("../../../etc/passwd", work_dir=work_dir)
+
+    def test_build_filter_work_dir_required(self, work_dir):
+        """不传work_dir时必须报错（防止自证清白绕过）."""
+        from video_processing.subtitle_render_engine import build_subtitle_filter
+
+        ass_file = work_dir / "sub.ass"
+        ass_file.write_text("test", encoding="utf-8")
+
+        # 不传 work_dir 必须报错
+        with pytest.raises(PathSecurityError, match="work_dir"):
+            build_subtitle_filter(ass_file)  # type: ignore[call-arg]
+
+        # 传 None 也必须报错
+        with pytest.raises(PathSecurityError, match="work_dir"):
+            build_subtitle_filter(ass_file, work_dir=None)  # type: ignore[arg-type]
+
+        # 传空字符串也必须报错
+        with pytest.raises(PathSecurityError, match="work_dir"):
+            build_subtitle_filter(ass_file, work_dir="")
+
+    def test_subtitle_path_traversal_bypass_startswith(self, work_dir):
+        """【P1绕过】字幕路径../遍历绕过startswith检查，必须被拦截."""
+        from video_processing.subtitle_render_engine import _validate_subtitle_path
+
+        evil_path = str(work_dir / "../../../../opt/secret.srt")
+        with pytest.raises(PathSecurityError, match="不在允许目录"):
+            _validate_subtitle_path(evil_path, work_dir)

tests/unit/test_tts_oss_transfer.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import unittest
 from datetime import datetime, timezone
 from unittest.mock import MagicMock, patch
 
@@ -67,13 +68,10 @@ def _make_workflow(
 class TestTransferAudioToOSS:
     """测试 _transfer_audio_to_oss 方法。"""
 
-    @patch("packages.application.tts_job.workflow.httpx")
-    def test_success_download_and_upload(self, mock_httpx: MagicMock) -> None:
+    @patch("packages.application.tts_job.workflow.safe_download_bytes")
+    def test_success_download_and_upload(self, mock_download: MagicMock) -> None:
         """成功下载音频并上传到 OSS，返回永久 URL 和 storage_key。"""
-        mock_resp = MagicMock()
-        mock_resp.content = b"fake audio data"
-        mock_resp.raise_for_status.return_value = None
-        mock_httpx.get.return_value = mock_resp
+        mock_download.return_value = b"fake audio data"
 
         storage = MagicMock()
         storage.upload_file.return_value = "https://oss.example.com/tts-outputs/user_001/job_123.mp3"
@@ -89,20 +87,21 @@ class TestTransferAudioToOSS:
         assert url == "https://oss.example.com/tts-outputs/user_001/job_123.mp3"
         assert key == "tts-outputs/user_001/job_123.mp3"
 
-        mock_httpx.get.assert_called_once_with(
+        mock_download.assert_called_once_with(
             "https://cosyvoice-temp.com/audio.mp3",
+            purpose="tts_audio_download",
+            allowed_mime_types=unittest.mock.ANY,
             timeout=60.0,
-            follow_redirects=True,
         )
         storage.upload_file.assert_called_once()
         call_args = storage.upload_file.call_args
         assert call_args[0][1] == "tts-outputs/user_001/job_123.mp3"
         assert call_args[1]["content_type"] == "audio/mpeg"
 
-    @patch("packages.application.tts_job.workflow.httpx")
-    def test_download_failure_fallback(self, mock_httpx: MagicMock) -> None:
+    @patch("packages.application.tts_job.workflow.safe_download_bytes")
+    def test_download_failure_fallback(self, mock_download: MagicMock) -> None:
         """下载失败时回退到原始临时 URL，storage_key 为空。"""
-        mock_httpx.get.side_effect = Exception("Network error")
+        mock_download.side_effect = Exception("Network error")
 
         workflow = _make_workflow()
         url, key = workflow._transfer_audio_to_oss(
@@ -114,13 +113,10 @@ class TestTransferAudioToOSS:
         assert url == "https://cosyvoice-temp.com/audio.mp3"
         assert key == ""
 
-    @patch("packages.application.tts_job.workflow.httpx")
-    def test_upload_failure_fallback(self, mock_httpx: MagicMock) -> None:
+    @patch("packages.application.tts_job.workflow.safe_download_bytes")
+    def test_upload_failure_fallback(self, mock_download: MagicMock) -> None:
         """上传 OSS 失败时回退到原始临时 URL。"""
-        mock_resp = MagicMock()
-        mock_resp.content = b"fake audio data"
-        mock_resp.raise_for_status.return_value = None
-        mock_httpx.get.return_value = mock_resp
+        mock_download.return_value = b"fake audio data"
 
         storage = MagicMock()
         storage.upload_file.side_effect = Exception("OSS bucket error")
@@ -135,13 +131,10 @@ class TestTransferAudioToOSS:
         assert url == "https://cosyvoice-temp.com/audio.mp3"
         assert key == ""
 
-    @patch("packages.application.tts_job.workflow.httpx")
-    def test_wav_content_type(self, mock_httpx: MagicMock) -> None:
+    @patch("packages.application.tts_job.workflow.safe_download_bytes")
+    def test_wav_content_type(self, mock_download: MagicMock) -> None:
         """wav 格式使用正确的 content_type。"""
-        mock_resp = MagicMock()
-        mock_resp.content = b"fake wav data"
-        mock_resp.raise_for_status.return_value = None
-        mock_httpx.get.return_value = mock_resp
+        mock_download.return_value = b"fake wav data"
 
         storage = MagicMock()
         storage.upload_file.return_value = "https://oss.example.com/audio.wav"
@@ -162,13 +155,10 @@ class TestTransferAudioToOSS:
 class TestProcessSynthesisResultWithOSS:
     """测试 process_synthesis_result 集成 OSS 转存。"""
 
-    @patch("packages.application.tts_job.workflow.httpx")
-    def test_stores_permanent_url_and_key(self, mock_httpx: MagicMock) -> None:
+    @patch("packages.application.tts_job.workflow.safe_download_bytes")
+    def test_stores_permanent_url_and_key(self, mock_download: MagicMock) -> None:
         """合成结果存 OSS 永久 URL 和 storage_key。"""
-        mock_resp = MagicMock()
-        mock_resp.content = b"audio bytes"
-        mock_resp.raise_for_status.return_value = None
-        mock_httpx.get.return_value = mock_resp
+        mock_download.return_value = b"audio bytes"
 
         storage = MagicMock()
         storage.upload_file.return_value = "https://oss.example.com/tts-outputs/user_001/test_job_001.mp3"
@@ -192,10 +182,10 @@ class TestProcessSynthesisResultWithOSS:
         assert result.duration == 5.0
         assert result.file_size == 50000
 
-    @patch("packages.application.tts_job.workflow.httpx")
-    def test_fallback_to_temp_url_on_oss_failure(self, mock_httpx: MagicMock) -> None:
+    @patch("packages.application.tts_job.workflow.safe_download_bytes")
+    def test_fallback_to_temp_url_on_oss_failure(self, mock_download: MagicMock) -> None:
         """OSS 转存失败时，使用 CosyVoice 临时 URL（不阻塞合成流程）。"""
-        mock_httpx.get.side_effect = Exception("Download failed")
+        mock_download.side_effect = Exception("Download failed")
 
         repo = MagicMock()
         job = _make_job(status=TTSJobStatus.PROCESSING)
@@ -216,13 +206,10 @@ class TestProcessSynthesisResultWithOSS:
 class TestStartSynthesisSyncWithOSS:
     """测试 start_synthesis 同步路径的 OSS 转存。"""
 
-    @patch("packages.application.tts_job.workflow.httpx")
-    def test_sync_path_transfers_to_oss(self, mock_httpx: MagicMock) -> None:
+    @patch("packages.application.tts_job.workflow.safe_download_bytes")
+    def test_sync_path_transfers_to_oss(self, mock_download: MagicMock) -> None:
         """CosyVoice 同步返回 audio_url 时，也走 OSS 转存。"""
-        mock_resp = MagicMock()
-        mock_resp.content = b"sync audio bytes"
-        mock_resp.raise_for_status.return_value = None
-        mock_httpx.get.return_value = mock_resp
+        mock_download.return_value = b"sync audio bytes"
 
         storage = MagicMock()
         storage.upload_file.return_value = "https://oss.example.com/tts-outputs/user_001/test_job_001.mp3"
@@ -248,10 +235,10 @@ class TestStartSynthesisSyncWithOSS:
         assert job.output_audio_key == "tts-outputs/user_001/test_job_001.mp3"
         assert job.duration == 2.0
 
-    @patch("packages.application.tts_job.workflow.httpx")
-    def test_sync_path_oss_failure_stores_temp_url(self, mock_httpx: MagicMock) -> None:
+    @patch("packages.application.tts_job.workflow.safe_download_bytes")
+    def test_sync_path_oss_failure_stores_temp_url(self, mock_download: MagicMock) -> None:
         """同步路径 OSS 失败时，降级存储临时 URL。"""
-        mock_httpx.get.side_effect = Exception("Network error")
+        mock_download.side_effect = Exception("Network error")
 
         service = MagicMock(spec=CosyVoiceService)
         service.submit_synthesize_task.return_value = {

tests/unit/test_tts_segment_synthesis.py

@@ -233,14 +233,11 @@ class TestStartSegmentSynthesis:
         # 短文本走普通路径，不调用分段
         assert job.status == TTSJobStatus.PROCESSING
 
-    @patch("packages.application.tts_job.workflow.httpx")
-    def test_long_text_sync_segments(self, mock_httpx: MagicMock) -> None:
+    @patch("packages.application.tts_job.workflow.safe_download_file")
+    def test_long_text_sync_segments(self, mock_download: MagicMock) -> None:
         """长文本同步分段：所有段立即返回 audio_url，直接合并。"""
         # Mock 分段音频下载
-        mock_resp = MagicMock()
-        mock_resp.content = b"segment audio"
-        mock_resp.raise_for_status.return_value = None
-        mock_httpx.get.return_value = mock_resp
+        mock_download.return_value = 1024  # 模拟文件大小
 
         service = MagicMock(spec=CosyVoiceService)
         # 每个分段都同步返回 audio_url
@@ -355,14 +352,11 @@ class TestHandleSegmentFailure:
 class TestPollSegmentTasks:
     """测试 _poll_segment_tasks 分段缺失重新合成（适配同步接口）。"""
 
-    @patch("packages.application.tts_job.workflow.httpx")
-    def test_all_segments_done(self, mock_httpx: MagicMock) -> None:
+    @patch("packages.application.tts_job.workflow.safe_download_file")
+    def test_all_segments_done(self, mock_download: MagicMock) -> None:
         """所有分段缺少 audio_url 时重新同步合成，合并后标记完成。"""
         # Mock 下载分段音频
-        mock_resp = MagicMock()
-        mock_resp.content = b"seg audio"
-        mock_resp.raise_for_status.return_value = None
-        mock_httpx.get.return_value = mock_resp
+        mock_download.return_value = 1024  # 模拟文件大小
 
         service = MagicMock(spec=CosyVoiceService)
         service.submit_synthesize_task.side_effect = [
@@ -422,13 +416,10 @@ class TestPollSegmentTasks:
 
         assert result.status == TTSJobStatus.FAILED
 
-    @patch("packages.application.tts_job.workflow.httpx")
-    def test_partial_audio_urls_reuse_existing(self, mock_httpx: MagicMock) -> None:
+    @patch("packages.application.tts_job.workflow.safe_download_file")
+    def test_partial_audio_urls_reuse_existing(self, mock_download: MagicMock) -> None:
         """部分分段已有 audio_url 时直接复用，缺失的重新合成。"""
-        mock_resp = MagicMock()
-        mock_resp.content = b"seg audio"
-        mock_resp.raise_for_status.return_value = None
-        mock_httpx.get.return_value = mock_resp
+        mock_download.return_value = 1024  # 模拟文件大小
 
         service = MagicMock(spec=CosyVoiceService)
         # 只有 1 个分段需要重新合成
@@ -515,11 +506,8 @@ class TestPollAndProcessSynthesisSegmentDetection:
 
         workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage)
 
-        with patch("packages.application.tts_job.workflow.httpx") as mock_httpx:
-            mock_resp = MagicMock()
-            mock_resp.content = b"audio"
-            mock_resp.raise_for_status.return_value = None
-            mock_httpx.get.return_value = mock_resp
+        with patch("packages.application.tts_job.workflow.safe_download_file") as mock_download:
+            mock_download.return_value = 1024  # 模拟文件大小
 
             result = workflow.poll_and_process_synthesis("test_job_seg")
 

tests/unit/test_tts_streaming.py

@@ -221,17 +221,14 @@ class TestTTSStreamingService:
 
     @pytest.mark.asyncio
     async def test_download_audio(self):
-        """下载音频数据。"""
+        """下载音频数据（SSRF防护走safe_download_bytes，mock掉安全层）。"""
         cosyvoice = MagicMock(spec=CosyVoiceService)
         service = TTSStreamingService(cosyvoice)
 
-        with patch("packages.application.tts_job.streaming_service.httpx") as mock_httpx:
-            mock_resp = MagicMock()
-            mock_resp.content = b"audio data"
-            mock_resp.raise_for_status.return_value = None
-            mock_httpx.get.return_value = mock_resp
+        with patch("packages.application.tts_job.streaming_service.safe_download_bytes") as mock_download:
+            mock_download.return_value = b"audio data"
 
             result = service._download_audio("https://example.com/audio.mp3")
 
             assert result == b"audio data"
-            mock_httpx.get.assert_called_once()
+            mock_download.assert_called_once()

tests/unit/test_url_security.py

@@ -0,0 +1,296 @@
+"""URL 安全校验工具单元测试 — SSRF 防护."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "worker"))
+
+import shutil
+import tempfile
+
+from video_processing.url_security import (  # noqa: E402
+    ALLOWED_AUDIO_MIME_TYPES,
+    UrlSecurityError,
+    is_url_safe,
+    safe_download_bytes,
+    safe_download_file,
+    validate_url_safety,
+)
+
+
+class TestUrlSecurityValidation(unittest.TestCase):
+    """URL 安全校验测试."""
+
+    # ── Scheme 白名单 ──────────────────────────────────────────────────────
+
+    def test_http_scheme_allowed(self):
+        """HTTP scheme 应该被允许."""
+        result = validate_url_safety("http://example.com/test", purpose="test")
+        self.assertEqual(result, "http://example.com/test")
+
+    def test_https_scheme_allowed(self):
+        """HTTPS scheme 应该被允许."""
+        result = validate_url_safety("https://example.com/test", purpose="test")
+        self.assertEqual(result, "https://example.com/test")
+
+    def test_file_scheme_rejected(self):
+        """file:// scheme 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("file:///etc/passwd", purpose="test")
+
+    def test_ftp_scheme_rejected(self):
+        """ftp:// scheme 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("ftp://example.com/test", purpose="test")
+
+    def test_empty_scheme_rejected(self):
+        """空 scheme 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("example.com/test", purpose="test")
+
+    # ── 端口白名单 ────────────────────────────────────────────────────────
+
+    def test_port_80_allowed(self):
+        """端口 80 应该被允许."""
+        # 80端口是默认HTTP端口，不显式指定也可以
+        result = validate_url_safety("http://example.com:80/test", purpose="test")
+        self.assertIn("example.com", result)
+
+    def test_port_443_allowed(self):
+        """端口 443 应该被允许."""
+        result = validate_url_safety("https://example.com:443/test", purpose="test")
+        self.assertIn("example.com", result)
+
+    def test_port_8080_rejected(self):
+        """非标准端口 8080 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http://example.com:8080/test", purpose="test")
+
+    def test_port_22_rejected(self):
+        """SSH 端口 22 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http://example.com:22/test", purpose="test")
+
+    # ── SSRF: 直接 IP 访问 ───────────────────────────────────────────────
+
+    def test_loopback_ip_rejected(self):
+        """回环地址 127.0.0.1 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http://127.0.0.1/test", purpose="test")
+
+    def test_private_ip_192_rejected(self):
+        """内网地址 192.168.x.x 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http://192.168.1.1/test", purpose="test")
+
+    def test_private_ip_10_rejected(self):
+        """内网地址 10.x.x.x 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http://10.0.0.1/test", purpose="test")
+
+    def test_private_ip_172_rejected(self):
+        """内网地址 172.16.x.x 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http://172.16.0.1/test", purpose="test")
+
+    def test_unspecified_ip_rejected(self):
+        """未指定地址 0.0.0.0 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http://0.0.0.0/test", purpose="test")
+
+    def test_ipv6_loopback_rejected(self):
+        """IPv6 回环 ::1 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http://[::1]/test", purpose="test")
+
+    def test_ipv6_link_local_rejected(self):
+        """IPv6 链路本地地址应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http://[fe80::1]/test", purpose="test")
+
+    # ── SSRF: 内网主机名 ─────────────────────────────────────────────────
+
+    def test_localhost_rejected(self):
+        """localhost 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http://localhost/test", purpose="test")
+
+    def test_local_domain_rejected(self):
+        """.local 域名应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http://printer.local/test", purpose="test")
+
+    def test_internal_domain_rejected(self):
+        """.internal 域名应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http://db.internal/test", purpose="test")
+
+    # ── URL 格式校验 ─────────────────────────────────────────────────────
+
+    def test_empty_url_rejected(self):
+        """空 URL 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("", purpose="test")
+
+    def test_none_url_rejected(self):
+        """None URL 应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety(None, purpose="test")  # type: ignore
+
+    def test_url_too_long_rejected(self):
+        """超长 URL 应该被拒绝."""
+        long_url = "https://example.com/" + "a" * 3000
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety(long_url, purpose="test")
+
+    def test_no_hostname_rejected(self):
+        """缺少主机名应该被拒绝."""
+        with self.assertRaises(UrlSecurityError):
+            validate_url_safety("http:///test", purpose="test")
+
+    # ── is_url_safe 便捷函数 ─────────────────────────────────────────────
+
+    def test_is_url_safe_true(self):
+        """安全 URL 应该返回 True."""
+        self.assertTrue(is_url_safe("https://example.com/test", purpose="test"))
+
+    def test_is_url_safe_false(self):
+        """不安全 URL 应该返回 False."""
+        self.assertFalse(is_url_safe("http://127.0.0.1/test", purpose="test"))
+
+    def test_is_url_safe_empty(self):
+        """空 URL 应该返回 False."""
+        self.assertFalse(is_url_safe("", purpose="test"))
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class TestSafeDownload(unittest.TestCase):
+    """安全下载函数测试."""
+
+    def setUp(self):
+        self.temp_dir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+    def test_safe_download_file_rejects_ssrf(self):
+        """SSRF 风险 URL 应该被拒绝下载."""
+        dest = os.path.join(self.temp_dir, "test.bin")
+        with self.assertRaises(UrlSecurityError):
+            safe_download_file("http://127.0.0.1/test", dest, purpose="test")
+
+    def test_safe_download_bytes_rejects_ssrf(self):
+        """SSRF 风险 URL 应该被拒绝下载（bytes 版本）."""
+        with self.assertRaises(UrlSecurityError):
+            safe_download_bytes("http://localhost/test", purpose="test")
+
+    def test_safe_download_file_size_limit(self):
+        """超过大小限制应该被拒绝."""
+        # 用 mock server 测试太大的 content-length
+        dest = os.path.join(self.temp_dir, "test.bin")
+        # 直接验证参数：max_size=0 时任何下载都应超限
+        # （这里用一个可访问的 URL 并设置极小的限制）
+        # 为避免依赖外部网络，这里只测试函数参数传递
+        with unittest.mock.patch("urllib.request.build_opener") as mock_opener:
+            mock_resp = unittest.mock.MagicMock()
+            mock_resp.headers = {"Content-Length": "1000"}
+            mock_resp.read.return_value = b""
+            mock_opener.return_value.open.return_value = mock_resp
+            # 设置 max_size=500，content-length=1000 应被拒绝
+            with self.assertRaises(UrlSecurityError):
+                safe_download_file(
+                    "https://example.com/test",
+                    dest,
+                    purpose="test",
+                    max_size=500,
+                )
+
+    def test_safe_download_file_mime_rejected(self):
+        """不允许的 MIME 类型应该被拒绝."""
+        dest = os.path.join(self.temp_dir, "test.bin")
+        with unittest.mock.patch("urllib.request.build_opener") as mock_opener:
+            mock_resp = unittest.mock.MagicMock()
+            mock_resp.headers = {"Content-Type": "text/html"}
+            mock_resp.read.return_value = b""
+            mock_opener.return_value.open.return_value = mock_resp
+            with self.assertRaises(UrlSecurityError):
+                safe_download_file(
+                    "https://example.com/test.mp3",
+                    dest,
+                    purpose="test",
+                    allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
+                )
+
+    def test_safe_download_file_mime_allowed(self):
+        """允许的 MIME 类型应该通过."""
+        dest = os.path.join(self.temp_dir, "test.mp3")
+        with unittest.mock.patch("urllib.request.build_opener") as mock_opener:
+            mock_resp = unittest.mock.MagicMock()
+            mock_resp.headers = {"Content-Type": "audio/mpeg"}
+            mock_resp.read.side_effect = [b"audio_data", b""]
+            mock_resp.geturl.return_value = "https://example.com/test.mp3"
+            mock_opener.return_value.open.return_value = mock_resp
+            size = safe_download_file(
+                "https://example.com/test.mp3",
+                dest,
+                purpose="test",
+                allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
+            )
+            self.assertEqual(size, 10)
+            self.assertTrue(os.path.exists(dest))
+
+    def test_safe_download_file_stream_size_limit(self):
+        """流式下载时超过大小限制应该中断."""
+        dest = os.path.join(self.temp_dir, "test.bin")
+        with unittest.mock.patch("urllib.request.build_opener") as mock_opener:
+            mock_resp = unittest.mock.MagicMock()
+            mock_resp.headers = {}
+            # 每次返回 100 字节，max_size=500，第 6 次读取就超限
+            mock_resp.read.side_effect = lambda n: b"x" * n if n < 1000 else b"x" * 100
+            # 改成返回固定 100 字节，直到第 N 次后返回空
+            call_count = [0]
+
+            def mock_read(size):
+                call_count[0] += 1
+                if call_count[0] > 10:
+                    return b""
+                return b"x" * 100
+
+            mock_resp.read = mock_read
+            mock_opener.return_value.open.return_value = mock_resp
+            with self.assertRaises(UrlSecurityError):
+                safe_download_file(
+                    "https://example.com/test",
+                    dest,
+                    purpose="test",
+                    max_size=500,  # 500 字节上限
+                )
+
+    def test_safe_download_bytes_returns_content(self):
+        """safe_download_bytes 应该返回文件内容."""
+        test_data = b"hello world test audio"
+        with unittest.mock.patch("urllib.request.build_opener") as mock_opener:
+            mock_resp = unittest.mock.MagicMock()
+            mock_resp.headers = {"Content-Type": "audio/mpeg"}
+            call_count = [0]
+
+            def mock_read(size):
+                call_count[0] += 1
+                if call_count[0] > 1:
+                    return b""
+                return test_data
+
+            mock_resp.read = mock_read
+            mock_opener.return_value.open.return_value = mock_resp
+            result = safe_download_bytes(
+                "https://example.com/test.mp3",
+                purpose="test",
+                allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
+            )
+            self.assertEqual(result, test_data)