style: isort fix imports

- Bug1: 过滤逻辑实际从diff中移除跳过的文件（lock/图片等），节省token
- Bug2: 评论去重接上，每次新审查先删除旧的AI评论，避免刷屏
- Bug3: 异常情况sys.exit(1)替代sys.exit(0)，配合continue-on-error也能看到失败

.gitea/workflows/ci-build.yml

@@ -4,17 +4,6 @@ on:
     branches:
     - main
     - develop
-    - feature/**
-    - feat/**
-    - bugfix/**
-    - fix/**
-    - hotfix/**
-    - release/**
-    - refactor/**
-    - perf/**
-    - docs/**
-    - chore/**
-    - ci/**
     tags:
     - v*
   pull_request:
@@ -35,7 +24,7 @@ concurrency:
 jobs:
   build-staging:
     name: Build Staging ${{ matrix.service_display }} Image
-    runs-on: host
+    runs-on: ci-l2
     timeout-minutes: ${{ matrix.timeout }}
     if: github.event_name == 'push' && (github.ref_name == 'main' || github.ref_name == 'develop')
     strategy:
@@ -157,8 +146,11 @@ jobs:
         '
   deploy-staging:
     name: Deploy Staging (Watchtower auto-deploy)
-    runs-on: saas
+    runs-on: ci-l3
     timeout-minutes: 15
+    concurrency:
+      group: deploy-staging-${{ gitea.ref }}
+      cancel-in-progress: false
     needs:
     - build-staging
     if: github.event_name == 'push' && (github.ref_name == 'main' || github.ref_name == 'develop')
@@ -308,7 +300,7 @@ jobs:
         '
   staging-e2e:
     name: Staging E2E Tests
-    runs-on: saas
+    runs-on: ci-l3
     timeout-minutes: 15
     if: github.ref_name == 'develop' || github.ref_name == 'main'
     needs: deploy-staging
@@ -388,7 +380,7 @@ jobs:
         '
   staging-api-tests:
     name: Staging API Integration Tests
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 10
     if: github.ref_name == 'develop' || github.ref_name == 'main'
     needs: deploy-staging
@@ -468,7 +460,7 @@ jobs:
         '
   build-production:
     name: Build Production ${{ matrix.service_display }} Image
-    runs-on: host
+    runs-on: ci-l2
     timeout-minutes: ${{ matrix.timeout }}
     needs:
     if: startsWith(github.ref, 'refs/tags/v')
@@ -592,8 +584,11 @@ jobs:
         '
   deploy-production:
     name: Deploy Production
-    runs-on: saas
+    runs-on: ci-l3
     timeout-minutes: 30
+    concurrency:
+      group: deploy-production-${{ gitea.ref }}
+      cancel-in-progress: false
     if: startsWith(github.ref, 'refs/tags/v')
     needs:
     - build-production
@@ -735,7 +730,7 @@ jobs:
         '
   production-e2e:
     name: Production Browser E2E
-    runs-on: saas
+    runs-on: ci-l3
     timeout-minutes: 15
     if: startsWith(github.ref, 'refs/tags/v')
     needs: deploy-production

.gitea/workflows/ci-cd.yml

@@ -4,17 +4,6 @@ on:
     branches:
     - main
     - develop
-    - feature/**
-    - feat/**
-    - bugfix/**
-    - fix/**
-    - hotfix/**
-    - release/**
-    - refactor/**
-    - perf/**
-    - docs/**
-    - chore/**
-    - ci/**
     tags:
     - v*
   pull_request:
@@ -30,12 +19,48 @@ on:
 permissions:
   contents: read
 concurrency:
-  group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.ref }}
+  group: ci-cd-${{ gitea.event_name }}-${{ gitea.ref }}
   cancel-in-progress: true
 jobs:
+  check-frontend-only:
+    name: Check if frontend-only change
+    runs-on: ci-l1
+    if: github.event_name == 'pull_request'
+    outputs:
+      skip_backend: ${{ steps.check.outputs.skip_backend }}
+    steps:
+    - name: Checkout code
+      shell: sh
+      env:
+        GITHUB_TOKEN: ${{ github.token }}
+      run: "set -eu\npython3 - <<'PY'\nimport io, os, tarfile, time, urllib.request, urllib.error\nurl = f\"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz\"\nrequest = urllib.request.Request(url, headers={\"Authorization\": f\"token {os.environ['GITHUB_TOKEN']}\"})\nlast_err = None\nfor attempt in range(5):\n    try:\n        with urllib.request.urlopen(request, timeout=120) as response:\n            archive = response.read()\n        break\n    except urllib.error.HTTPError as e:\n        last_err = e\n        if e.code >= 500 and attempt < 4:\n            wait = 2 ** attempt\n            print(f\"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...\")\n            time.sleep(wait)\n            continue\n        raise\n    except Exception as e:\n        last_err = e\n        if attempt < 4:\n            wait = 2 ** attempt\n            print(f\"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...\")\n            time.sleep(wait)\n            continue\n        raise\nelse:\n    raise last_err\nwith tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:\n    root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'\n    for member in tar.getmembers():\n        name = member.name\n        if name == root_prefix[:-1]:\n            continue\n        if name.startswith(root_prefix):\n            member.name = name[len(root_prefix):]\n        if member.name:\n            tar.extract(member, '.')\nPY\n"
+    - name: Check changed files
+      id: check
+      shell: bash
+      env:
+        GITHUB_TOKEN: ${{ github.token }}
+      run: |
+        set -eu
+        PR_NUMBER=$(echo "$GITHUB_REF" | sed 's|refs/pull/||; s|/.*||')
+        API_URL="${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}/files?limit=300"
+        FILES=$(curl -s -H "Authorization: token ${GITHUB_TOKEN}" "$API_URL"           | python3 -c "import sys,json; [print(f['filename']) for f in json.load(sys.stdin)]")
+        FRONTEND_COUNT=$(echo "$FILES" | grep -c '^apps/web/' || true)
+        BACKEND_COUNT=$(echo "$FILES" | grep -cv '^apps/web/' || true)
+        TOTAL=$(echo "$FILES" | grep -cv '^$' || true)
+        echo "变更文件: ${TOTAL} 个 (前端: ${FRONTEND_COUNT}, 后端/公共: ${BACKEND_COUNT})"
+        if [ "$BACKEND_COUNT" = "0" ] && [ "$FRONTEND_COUNT" -gt "0" ]; then
+          echo "skip_backend=true" >> $GITHUB_OUTPUT
+          echo "✅ 纯前端改动，跳过后端检查"
+        else
+          echo "skip_backend=false" >> $GITHUB_OUTPUT
+          echo "🔧 包含后端/公共变更，运行完整CI"
+        fi
+
   validate:
+    needs: check-frontend-only
+    if: always() && needs.check-frontend-only.outputs.skip_backend != 'true'
     name: Validate Code Quality And Tests
-    runs-on: host
+    runs-on: ci-l1
     timeout-minutes: 10
     env:
       DATABASE_URL: postgresql+psycopg://postgres:postgres@127.0.0.1:5432/xiaoxia_saas
@@ -77,6 +102,10 @@ jobs:
 
         python3 -m pip install -q -r requirements-dev.txt
 
+        # Force source install of black/isort to ensure consistent formatting
+        # across compiled/source installations on different machines
+        python3 -m pip install --no-binary :all: black==26.5.1 isort==8.0.1
+
         python3 -m black --version
 
         python3 -m isort --version-number
@@ -90,7 +119,7 @@ jobs:
         '
     - name: Secret detection (detect-secrets)
       shell: sh
-      run: "set -eu\necho \"=== Installing detect-secrets ===\"\npython3 -m pip install -q detect-secrets\ndetect-secrets --version\necho \"\"\necho \"=== Running secret scan ===\"\ndetect-secrets scan \\\n  --all-files \\\n  --exclude-files '(^|/)(tests|test|e2e|__tests__|spec|docs|node_modules|site-packages|migrations|alembic|.gitea|.git|.pytest_cache|.next|dist|build)/' \\\n  --exclude-files '\\.(md|rst|txt|lock|example|sample|min\\.js|min\\.css|spec\\.ts|test\\.ts|test\\.py)$' \\\n  --exclude-files '(package-lock|yarn\\.lock|poetry\\.lock|Pipfile\\.lock)$' \\\n  --disable-plugin Base64HighEntropyString \\\n  --disable-plugin HexHighEntropyString \\\n  --disable-plugin BasicAuthDetector \\\n  --disable-plugin KeywordDetector \\\n  --disable-plugin IPPublicDetector \\\n  2>&1 | tee /tmp/secrets-scan.json\n\nFOUND=$(python3 -c \"\nimport json\ntry:\n    with open('/tmp/secrets-scan.json') as f:\n        data = json.load(f)\n    results = data.get('results', {})\n    total = sum(len(v) for\
+      run: "set -eu\necho \"=== Installing detect-secrets ===\"\npython3 -m pip install -q detect-secrets\ndetect-secrets --version\necho \"\"\necho \"=== Running secret scan ===\"\ndetect-secrets scan \\\n  --all-files \\\n  --exclude-files '(^|/)(tests|test|e2e|__tests__|spec|docs|node_modules|site-packages|migrations|alembic|.gitea|.git|.pytest_cache|.next|dist|build)/' \\\n  --exclude-files '\\.(md|rst|txt|lock|example|sample|min\\.js|min\\.css|spec\\.ts|test\\.ts|test\\.py)$' \\\n  --exclude-files '(package-lock|yarn\\.lock|poetry\\.lock|Pipfile\\.lock)$' \\\n  --disable-plugin Base64HighEntropyString \\\n  --disable-plugin HexHighEntropyString \\\n  --disable-plugin BasicAuthDetector \\\n  --disable-plugin KeywordDetector \\\n  --disable-plugin IPPublicDetector \\\n  > /tmp/secrets-scan.json 2>&1\n\nFOUND=$(python3 -c \"\nimport json\ntry:\n    with open('/tmp/secrets-scan.json') as f:\n        data = json.load(f)\n    results = data.get('results', {})\n    total = sum(len(v) for\
         \ v in results.values())\n    print(total)\nexcept Exception:\n    print('error')\n\")\necho \"\"\necho \"Secrets detected: $FOUND\"\nif [ \"$FOUND\" != \"0\" ] && [ \"$FOUND\" != \"error\" ]; then\n  echo \"\"\n  echo \"=== Secret details ===\"\n  python3 -c \"\nimport json\nwith open('/tmp/secrets-scan.json') as f:\n    data = json.load(f)\nfor fpath, items in data.get('results', {}).items():\n    for item in items:\n        line = item.get('line_number', '?')\n        stype = item.get('type', '?')\n        hashed = item.get('hashed_secret', '')[:16]\n        print(f'  {fpath}:{line} [{stype}] {hashed}...')\n\"\n  echo \"\"\n  echo \"ERROR: Potential secrets detected in code!\"\n  echo \"If these are false positives, add exclusions in the CI workflow.\"\n  exit 1\nfi\necho \"Secret scan completed - no secrets detected\"\n"
     - name: Calculate changed Python files (incremental scan)
       shell: sh
@@ -129,24 +158,51 @@ jobs:
         bash -n scripts/init_production_env.sh
 
         '
-    - name: Validate Alembic migrations
-      shell: sh
-      run: 'set -eu
-
-        python3 -m alembic upgrade head --sql > /tmp/alembic-upgrade.sql
-
-        test -s /tmp/alembic-upgrade.sql
-
-        grep -q "Running upgrade" /tmp/alembic-upgrade.sql
-
-        python3 scripts/check_schema_metadata.py
-
-        '
-    - name: Check migration safety
+    - name: Validate Alembic migrations (with isolated PG)
       shell: sh
       env:
         GITHUB_TOKEN: ${{ github.token }}
-      run: "set -eu\npython3 scripts/check_migration_safety.py --allow-medium-risk --diff-against origin/develop\n"
+      run: |
+        set -eu
+        PG_CONTAINER=ci-pg-validate-${GITHUB_RUN_ID:-$$}
+        docker rm -f "$PG_CONTAINER" 2>/dev/null || true
+        docker run -d --name "$PG_CONTAINER" \
+          --shm-size=256m \
+          -e POSTGRES_USER=postgres \
+          -e POSTGRES_PASSWORD=postgres \
+          -e POSTGRES_DB=xiaoxia_saas \
+          -P \
+          --health-cmd "pg_isready -U postgres" \
+          --health-interval 3s \
+          --health-timeout 3s \
+          --health-retries 20 \
+          postgres:16-alpine
+        PG_PORT=$(docker port "$PG_CONTAINER" 5432/tcp | cut -d: -f2)
+        echo "PostgreSQL port: $PG_PORT"
+        export DATABASE_URL=postgresql+psycopg://postgres:postgres@127.0.0.1:$PG_PORT/xiaoxia_saas
+        for i in $(seq 1 30); do
+          if docker inspect --format='{{.State.Health.Status}}' "$PG_CONTAINER" 2>/dev/null | grep -q healthy; then
+            echo "PostgreSQL is ready on port $PG_PORT"
+            break
+          fi
+            echo "Waiting for PostgreSQL... ($i/30)"
+          sleep 2
+        done
+        docker inspect --format='{{.State.Health.Status}}' "$PG_CONTAINER" | grep -q healthy
+        python3 -m alembic upgrade head --sql > /tmp/alembic-upgrade.sql
+        test -s /tmp/alembic-upgrade.sql
+        grep -q "Running upgrade" /tmp/alembic-upgrade.sql
+        python3 scripts/check_schema_metadata.py
+        # Initialize git for migration safety diff (CI checkout is tar.gz without .git)
+        git init > /dev/null 2>&1
+        git remote add origin https://git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas.git > /dev/null 2>&1
+        git fetch origin develop:refs/remotes/origin/develop --depth=100 > /dev/null 2>&1
+        git add -A > /dev/null 2>&1
+        git -c user.email=ci@local -c user.name=CI commit -m "ci-tmp" > /dev/null 2>&1
+        python3 scripts/check_migration_safety.py --allow-medium-risk --diff-against origin/develop
+        docker rm -f "$PG_CONTAINER" 2>/dev/null || true
+        echo "PostgreSQL container cleaned up"
+
     - name: Job duration summary
       if: always()
       shell: sh
@@ -163,8 +219,10 @@ jobs:
 
         '
   unit-tests:
+    needs: check-frontend-only
+    if: always() && needs.check-frontend-only.outputs.skip_backend != 'true'
     name: Unit Tests
-    runs-on: host
+    runs-on: ci-l2
     timeout-minutes: 8
     env:
       USE_IN_MEMORY_DB: 'true'
@@ -235,10 +293,12 @@ jobs:
         '
   integration-tests:
     name: Integration Tests
-    runs-on: host
+    runs-on: ci-l2
     timeout-minutes: 30
-    if: always()
-    needs: validate
+    if: always() && needs.check-frontend-only.outputs.skip_backend != 'true'
+    needs:
+      - check-frontend-only
+      - validate
     env:
       DATABASE_URL: postgresql+psycopg://postgres:postgres@127.0.0.1:5432/xiaoxia_saas
       USE_IN_MEMORY_DB: 'false'
@@ -352,7 +412,7 @@ jobs:
         '
   frontend-lint:
     name: Frontend Lint
-    runs-on: host
+    runs-on: ci-l1
     timeout-minutes: 10
     steps:
     - name: Checkout code

.gitea/workflows/code-review.yml

@@ -0,0 +1,53 @@
+name: AI Code Review
+
+on:
+  pull_request:
+    types:
+      - opened
+      - synchronize
+      - reopened
+
+# 同一个 PR 只跑一个 review，新的取消旧的
+concurrency:
+  group: code-review-${{ gitea.repository }}-${{ gitea.event.pull_request.number }}
+  cancel-in-progress: true
+
+jobs:
+  code-review:
+    name: AI Code Review
+    runs-on: ubuntu-latest
+    # 跳过草稿 PR
+    if: ${{ !gitea.event.pull_request.draft }}
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+        with:
+          fetch-depth: 0
+
+      - name: Install dependencies
+        run: |
+          python3 -m pip install --upgrade pip
+          python3 -m pip install requests
+
+      - name: Run AI Code Review
+        env:
+          # Gitea 配置（自动从运行环境获取）
+          GITEA_API_URL: ${{ gitea.server_url }}
+          GITEA_TOKEN: ${{ secrets.REVIEW_GITEA_TOKEN }}
+          REPO_NAME: ${{ gitea.repository }}
+          PR_NUMBER: ${{ gitea.event.pull_request.number }}
+          # LLM 提供商: coze (扣子原生Bot) / openai (OpenAI兼容)
+          LLM_PROVIDER: "coze"
+          # 扣子模式配置（默认国内站 api.coze.cn）
+          LLM_BASE_URL: ${{ secrets.LLM_BASE_URL }}
+          LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
+          COZE_BOT_ID: ${{ secrets.COZE_BOT_ID }}
+          LLM_MODEL: ${{ secrets.LLM_MODEL }}
+          # 可选参数
+          MAX_DIFF_CHARS: "30000"
+          LLM_TIMEOUT: "120"
+        run: |
+          python3 scripts/ci_code_review.py
+        # 审查脚本异常不影响 CI 通过
+        continue-on-error: true

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

@@ -24,7 +24,7 @@ from typing import Any, List, Optional
 from app.auth import AuthenticatedUser, get_current_user
 from app.dependencies import get_db_session, get_project_repository
 from app.schemas.generation_task import GenerationTaskResponse
-from app.services import EditPlanService
+from app.services import EditPlanService, EditTemplateService
 from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
@@ -506,6 +506,84 @@ def copy_plan(
         current_user.user.id,
     )
     return _to_response(new_plan)
+# ── 保存为模板 ────────────────────────────────────────────────────────────────
+
+
+class SaveAsTemplateRequest(BaseModel):
+    """保存为模板请求体"""
+
+    name: str = Field(..., min_length=1, max_length=200, description="模板名称")
+    description: str = Field(default="", max_length=500, description="模板描述")
+    template_type: str = Field(default="custom", max_length=50, description="模板类型")
+    preview_url: str = Field(default="", max_length=500, description="预览图 URL")
+
+
+@router.post(
+    "/{plan_id}/save-as-template",
+    response_model=dict[str, Any],
+    summary="将剪辑计划保存为模板",
+    status_code=status.HTTP_201_CREATED,
+)
+def save_plan_as_template(
+    plan_id: str,
+    body: SaveAsTemplateRequest,
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
+    project_repo=Depends(get_project_repository),
+) -> dict[str, Any]:
+    """将指定剪辑计划的配置和片段结构保存为一个新模板。
+
+    新模板会复制计划的所有片段配置（类型、时长、转场、文案等），
+    但不绑定具体素材，可重复用于创建新的剪辑计划。
+    """
+    # 校验计划存在性和项目权限
+    plan_service = EditPlanService(db)
+    plan = plan_service.get_plan(plan_id)
+    if plan is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"剪辑计划不存在: {plan_id}",
+        )
+    if plan.project_id:
+        check_project_access(project_repo, current_user, plan.project_id)
+
+    template_service = EditTemplateService(db)
+    try:
+        result = template_service.save_plan_as_template(
+            plan_id=plan_id,
+            name=body.name,
+            description=body.description,
+            template_type=body.template_type,
+            preview_url=body.preview_url,
+        )
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(e),
+        ) from e
+
+    template = result["template"]
+    clip_configs = result["clip_configs"]
+
+    logger.info(
+        "保存计划为模板: plan_id=%s template_id=%s name=%s by user=%s",
+        plan_id,
+        template.id,
+        body.name,
+        current_user.user.id,
+    )
+
+    return {
+        "id": template.id,
+        "name": template.name,
+        "description": template.description,
+        "template_type": template.template_type,
+        "editing_mode": template.editing_mode,
+        "preview_url": template.preview_url,
+        "status": template.status.value,
+        "clip_count": len(clip_configs),
+        "created_at": template.created_at.isoformat(),
+    }
 
 
 # ── Include sub-routers (拆分模块) ────────────────────────────────────────────

apps/api/app/api/routes/edit_plans_clips.py

@@ -52,7 +52,9 @@ class EditPlanClipListResponse(BaseModel):
 class EditPlanClipCreateRequest(BaseModel):
     """创建剪辑片段请求体"""
 
-    clip_type: str = Field(..., min_length=1, max_length=50, description="片段类型: main/intro/outro/overlay/background/b_roll 等")
+    clip_type: str = Field(
+        ..., min_length=1, max_length=50, description="片段类型: main/intro/outro/overlay/background/b_roll 等"
+    )
     order: int = Field(..., ge=0, description="排序序号")
     asset_id: str = Field(default="", max_length=64, description="关联素材 ID")
     text_content: str = Field(default="", max_length=5000, description="文本内容（字幕/配音等）")
@@ -86,9 +88,10 @@ def _check_plan_access(plan_id: str, user_id: str, project_repository: Any, db:
     """验证用户是否有权限访问该剪辑计划（通过项目关联）。
     返回 plan 对象供后续使用，避免重复查询。
     """
-    from ._helpers import check_project_access
     from app.services.edit_plan_service import EditPlanService
 
+    from ._helpers import check_project_access
+
     svc = EditPlanService(db)
     plan = svc.get_plan(plan_id)
     if plan is None:

apps/api/app/api/routes/edit_plans_clips_batch.py

@@ -75,9 +75,10 @@ class ClipsFromAssetsResponse(BaseModel):
 
 def _check_plan_access(plan_id: str, user_id: str, project_repository: Any, db: Session) -> Any:
     """验证用户是否有权限访问该剪辑计划，返回 plan 对象。"""
-    from ._helpers import check_project_access
     from app.services.edit_plan_service import EditPlanService
 
+    from ._helpers import check_project_access
+
     svc = EditPlanService(db)
     plan = svc.get_plan(plan_id)
     if plan is None:

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

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 from dataclasses import replace
 from datetime import datetime, timezone
 from typing import List
@@ -20,6 +21,8 @@ from fastapi import APIRouter, Depends, HTTPException, status
 
 from packages.ports.user_repository import UserRepository
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter()
 
 
@@ -254,7 +257,9 @@ async def payment_callback(
         return {"success": True, "message": "支付成功", "record_id": record_id}
     except Exception as e:
         session.rollback()
-        raise HTTPException(status_code=500, detail=f"支付处理失败: {str(e)}") from e
+        logger.error(f"支付回调处理失败: user_id={user_id}, plan={plan}, error={e}")
+        # 不返回原始异常信息，避免泄漏内部实现细节
+        raise HTTPException(status_code=500, detail="支付处理失败，请稍后重试") from e
     finally:
         session.close()
 

apps/api/app/services/edit_template_service.py

@@ -12,9 +12,13 @@ from typing import Any, List, Optional
 from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl import (
+    SQLAlchemyEditPlanClipRepository,
+    SQLAlchemyEditPlanRepository,
     SQLAlchemyEditTemplateRepository,
     SQLAlchemyTemplateClipConfigRepository,
 )
+from packages.domain.edit_plan import EditPlan
+from packages.domain.edit_plan_clip import EditPlanClip
 from packages.domain.edit_template import EditTemplate, EditTemplateStatus
 from packages.domain.template_clip_config import (
     ClipType,
@@ -37,6 +41,9 @@ class EditTemplateService:
     def __init__(self, db: Session) -> None:
         self._template_repo = SQLAlchemyEditTemplateRepository(db)
         self._clip_config_repo = SQLAlchemyTemplateClipConfigRepository(db)
+        self._plan_repo = SQLAlchemyEditPlanRepository(db)
+        self._plan_clip_repo = SQLAlchemyEditPlanClipRepository(db)
+        self._db = db
 
     # ── 模板 CRUD ──────────────────────────────────────────────────────────
 
@@ -394,3 +401,137 @@ class EditTemplateService:
             "template": template,
             "clip_configs": clip_configs,
         }
+
+    # ── 从剪辑计划保存为模板 ──────────────────────────────────────────────
+
+    def save_plan_as_template(
+        self,
+        plan_id: str,
+        name: str,
+        *,
+        description: str = "",
+        template_type: str = "custom",
+        preview_url: str = "",
+    ) -> dict[str, Any]:
+        """将剪辑计划保存为模板
+
+        将指定剪辑计划的配置和片段结构另存为一个新模板，
+        方便后续基于该模板快速创建新的剪辑计划。
+
+        转换规则:
+        - 计划名称 → 模板名称（调用方传入，支持自定义）
+        - 计划 config → 模板 config（整体迁移）
+        - 计划 editing_mode 从 config 中提取，默认 one_take
+        - 每个片段转换为模板片段配置:
+          - clip_type 直接映射
+          - order 保持不变
+          - duration → min_duration = max_duration = duration（固定时长）
+          - text_content → text_template
+          - transition_effect 直接映射
+          - playback_speed 等播放参数存入 config
+          - 不保留 asset_id（模板不绑定具体素材）
+
+        Args:
+            plan_id: 源剪辑计划 ID
+            name: 新模板名称
+            description: 模板描述
+            template_type: 模板类型，默认 custom（用户自定义）
+            preview_url: 预览图 URL
+
+        Returns:
+            dict: {"template": EditTemplate, "clip_configs": List[TemplateClipConfig]}
+
+        Raises:
+            ValueError: 计划不存在或名称为空/重复
+        """
+        # 1. 读取源计划
+        plan = self._plan_repo.get(plan_id)
+        if plan is None:
+            raise ValueError(f"剪辑计划不存在: {plan_id}")
+
+        # 2. 读取所有片段（按 order 排序）
+        clips = self._plan_clip_repo.list_by_plan(plan_id)
+        clips.sort(key=lambda c: c.order)
+
+        # 3. 提取 editing_mode
+        editing_mode = plan.config.get("editing_mode", "one_take") if plan.config else "one_take"
+
+        # 4. 创建模板（复用 create_template 的校验逻辑，但手动构建避免重复查询）
+        clean_name = name.strip()
+        if not clean_name:
+            raise ValueError("模板名称不能为空")
+
+        # 名称重复检查
+        existing = self._template_repo.list_all(skip=0, limit=1000)
+        for t in existing:
+            if t.name == clean_name and t.status == EditTemplateStatus.ACTIVE:
+                raise ValueError(f"模板名称已存在: {clean_name}")
+
+        # 从计划 config 中提取模板级配置，去掉运行时/素材相关字段
+        plan_config = plan.config or {}
+        template_config: dict[str, Any] = {}
+        for key, value in plan_config.items():
+            # 跳过明显的运行时/实例字段，保留风格/模式类配置
+            if key not in {"asset_ids", "source_edit_plan_id", "generation_task_id"}:
+                template_config[key] = value
+
+        template = EditTemplate.create(
+            name=clean_name,
+            description=description,
+            template_type=template_type,
+            editing_mode=editing_mode,
+            config=template_config,
+            preview_url=preview_url,
+        )
+        created_template = self._template_repo.create(template)
+        logger.info(
+            "从剪辑计划创建模板: plan_id=%s template_id=%s name=%s clip_count=%d",
+            plan_id,
+            created_template.id,
+            clean_name,
+            len(clips),
+        )
+
+        # 5. 转换每个片段为模板片段配置
+        created_configs: List[TemplateClipConfig] = []
+        for clip in clips:
+            clip_config: dict[str, Any] = {}
+            # 播放速度存入 config
+            if clip.playback_speed and clip.playback_speed != 1.0:
+                clip_config["playback_speed"] = clip.playback_speed
+            # 片段自有 config 合并（优先级：clip.config 覆盖上面的）
+            if clip.config:
+                clip_config.update(clip.config)
+            # 去掉素材相关字段
+            clip_config.pop("asset_info", None)
+            clip_config.pop("source_asset_id", None)
+
+            # 转场效果兼容校验
+            try:
+                transition = TransitionEffect(clip.transition_effect)
+            except ValueError:
+                transition = TransitionEffect.CUT
+
+            # 片段类型兼容校验
+            try:
+                clip_type = ClipType(clip.clip_type)
+            except ValueError:
+                clip_type = ClipType.MAIN
+
+            clip_config_obj = TemplateClipConfig.create(
+                template_id=created_template.id,
+                clip_type=clip_type,
+                order=clip.order,
+                min_duration=clip.duration,
+                max_duration=clip.duration,
+                text_template=clip.text_content or "",
+                transition_effect=transition,
+                config=clip_config,
+            )
+            created = self._clip_config_repo.create(clip_config_obj)
+            created_configs.append(created)
+
+        return {
+            "template": created_template,
+            "clip_configs": created_configs,
+        }

packages/application/voice_clone/workflow.py

@@ -24,6 +24,7 @@ from packages.application.voice_clone.use_cases import (
 )
 from packages.domain.voice_clone_profile import VoiceCloneProfile
 from packages.ports.voice_clone_profile_repository import VoiceCloneProfileRepository
+from packages.shared.url_security import UrlSecurityError, validate_url_safety
 
 logger = logging.getLogger(__name__)
 
@@ -103,6 +104,15 @@ class VoiceCloneWorkflowService:
 
         # 2. 提交 CosyVoice 克隆任务（仅有音频 URL 时才标记 processing）
         if source_audio_url:
+            # SSRF 防护：校验音频 URL 安全性
+            try:
+                source_audio_url = validate_url_safety(source_audio_url, purpose="download")
+            except UrlSecurityError as e:
+                profile.mark_failed(f"音频URL安全校验失败: {e}")
+                profile = self.repository.update(profile)
+                logger.warning(f"音色克隆音频URL安全校验失败: profile_id={profile.id}, error={e}")
+                return profile
+
             # 标记为 processing
             profile.mark_processing()
             profile = self.repository.update(profile)
@@ -243,6 +253,15 @@ class VoiceCloneWorkflowService:
 
         # 3. 重新提交 CosyVoice
         if profile.source_audio_url:
+            # SSRF 防护：重新校验音频 URL 安全性
+            try:
+                validate_url_safety(profile.source_audio_url, purpose="download")
+            except UrlSecurityError as e:
+                profile.mark_failed(f"音频URL安全校验失败: {e}")
+                profile = self.repository.update(profile)
+                logger.warning(f"音色克隆重试音频URL安全校验失败: profile_id={clone_id}, error={e}")
+                return profile
+
             try:
                 submit_result = self.cosyvoice_service.submit_clone_task(
                     audio_url=profile.source_audio_url,

scripts/check_migration_safety.py

@@ -195,16 +195,28 @@ def get_new_migrations_via_git(diff_target: str) -> List[Path] | None:
             timeout=30,
         )
 
+        # 优先使用三点diff（找合并基线），失败时回退到两点diff（兼容tar.gz checkout + git init的CI环境）
+        diff_args = ["git", "diff", "--name-only", "--diff-filter=A", f"{diff_target}...HEAD"]
         result = subprocess.run(
-            ["git", "diff", "--name-only", "--diff-filter=A", f"{diff_target}...HEAD"],
+            diff_args,
             capture_output=True,
             text=True,
             cwd=str(REPO_ROOT),
             timeout=10,
         )
         if result.returncode != 0:
-            print(f"   （git diff 失败：{result.stderr.strip()}）")
-            return None
+            # fallback: 两点diff（无需共同祖先）
+            diff_args_2 = ["git", "diff", "--name-only", "--diff-filter=A", diff_target, "HEAD"]
+            result = subprocess.run(
+                diff_args_2,
+                capture_output=True,
+                text=True,
+                cwd=str(REPO_ROOT),
+                timeout=10,
+            )
+            if result.returncode != 0:
+                print(f"   （git diff 失败：{result.stderr.strip()}）")
+                return None
 
         new_migrations = []
         for line in result.stdout.strip().split("\n"):

scripts/ci_code_review.py

@@ -0,0 +1,627 @@
+#!/usr/bin/env python3
+"""
+CI Code Review Script
+- 从 Gitea 获取 PR diff
+- 调用 LLM 进行代码审查
+- 将审查结果写回 PR 评论
+"""
+
+import argparse
+import json
+import logging
+import os
+import sys
+from typing import Optional, Tuple
+
+import requests
+
+# ============== 日志配置 ==============
+logging.basicConfig(
+    level=logging.INFO,
+    format="[%(asctime)s] [%(levelname)s] %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+logger = logging.getLogger("ci_code_review")
+
+
+# ============== 常量配置 ==============
+# diff 最大字符数（超过则截断）
+MAX_DIFF_CHARS = int(os.getenv("MAX_DIFF_CHARS", "30000"))
+# LLM 调用超时时间（秒）
+LLM_TIMEOUT = int(os.getenv("LLM_TIMEOUT", "120"))
+# Gitea API 超时时间（秒）
+GITEA_TIMEOUT = int(os.getenv("GITEA_TIMEOUT", "30"))
+# 最大重试次数
+MAX_RETRIES = int(os.getenv("MAX_RETRIES", "2"))
+# LLM 提供商: openai (OpenAI兼容) / coze (扣子原生Bot API)
+LLM_PROVIDER = os.getenv("LLM_PROVIDER", "coze").lower()
+
+
+# ============== 工具函数 ==============
+def truncate_diff(diff_text: str, max_chars: int) -> Tuple[str, bool]:
+    """
+    截断过大的 diff 内容，避免超出 LLM 上下文限制。
+    优先保留文件头和前面的变更，末尾加提示。
+    """
+    if len(diff_text) <= max_chars:
+        return diff_text, False
+
+    # 找到一个合适的截断位置（尽量在文件边界）
+    truncated = diff_text[:max_chars]
+    # 尝试在最后一个 "diff --git" 处截断，避免截断到一半
+    last_file_boundary = truncated.rfind("\ndiff --git ")
+    if last_file_boundary > max_chars // 2:
+        truncated = truncated[:last_file_boundary]
+
+    truncated += (
+        f"\n\n... [DIFF TRUNCATED] 原始 diff 共 {len(diff_text)} 字符，"
+        f"已截断至 {len(truncated)} 字符，仅审查前半部分。\n"
+    )
+    return truncated, True
+
+
+def get_env_or_fail(name: str) -> str:
+    """从环境变量获取值，不存在则报错退出。"""
+    value = os.getenv(name)
+    if not value:
+        logger.error(f"环境变量 {name} 未设置")
+        sys.exit(1)
+    return value
+
+
+# ============== Gitea API 相关 ==============
+class GiteaClient:
+    """Gitea API 客户端"""
+
+    def __init__(self, base_url: str, token: str, repo: str):
+        # 确保 base_url 以 / 结尾
+        self.base_url = base_url.rstrip("/") + "/"
+        self.token = token
+        self.repo = repo  # 格式: owner/repo
+        self.session = requests.Session()
+        self.session.headers.update(
+            {
+                "Authorization": f"token {token}",
+                "Accept": "application/json",
+                "Content-Type": "application/json",
+            }
+        )
+
+    def _api_url(self, path: str) -> str:
+        """拼接 API 路径"""
+        return f"{self.base_url}api/v1/repos/{self.repo}/{path.lstrip('/')}"
+
+    def get_pr_diff(self, pr_number: int) -> str:
+        """
+        获取 PR 的 diff 内容。
+        Gitea API: GET /repos/{owner}/{repo}/pulls/{index}.diff
+        """
+        url = self._api_url(f"pulls/{pr_number}.diff")
+        logger.info(f"获取 PR #{pr_number} diff: {url}")
+
+        resp = self.session.get(
+            url,
+            timeout=GITEA_TIMEOUT,
+            headers={
+                "Accept": "text/plain",
+            },
+        )
+        if resp.status_code != 200:
+            logger.error(f"获取 diff 失败: HTTP {resp.status_code} - {resp.text[:200]}")
+            raise RuntimeError(f"Failed to get PR diff: HTTP {resp.status_code}")
+
+        diff_text = resp.text
+        logger.info(f"获取到 diff，共 {len(diff_text)} 字符")
+        return diff_text
+
+    def get_pr_files(self, pr_number: int) -> list:
+        """
+        获取 PR 修改的文件列表。
+        Gitea API: GET /repos/{owner}/{repo}/pulls/{index}/files
+        """
+        url = self._api_url(f"pulls/{pr_number}/files")
+        logger.info(f"获取 PR #{pr_number} 文件列表")
+
+        resp = self.session.get(url, timeout=GITEA_TIMEOUT)
+        if resp.status_code != 200:
+            logger.warning(f"获取文件列表失败: HTTP {resp.status_code}")
+            return []
+
+        files = resp.json()
+        logger.info(f"PR 修改了 {len(files)} 个文件")
+        return files
+
+    def post_pr_comment(self, pr_number: int, body: str) -> bool:
+        """
+        在 PR 上发布评论。
+        Gitea API: POST /repos/{owner}/{repo}/issues/{index}/comments
+        （Gitea 中 PR 评论走 issues 接口）
+        """
+        url = self._api_url(f"issues/{pr_number}/comments")
+        logger.info(f"发布 PR 评论: {url}")
+
+        payload = {"body": body}
+        resp = self.session.post(
+            url,
+            data=json.dumps(payload),
+            timeout=GITEA_TIMEOUT,
+        )
+        if resp.status_code not in (200, 201):
+            logger.error(f"发布评论失败: HTTP {resp.status_code} - {resp.text[:200]}")
+            return False
+
+        logger.info(f"评论发布成功，评论 ID: {resp.json().get('id', 'unknown')}")
+        return True
+
+    def get_existing_review_comments(self, pr_number: int, marker: str) -> list:
+        """
+        获取 PR 上已有的 AI 审查评论 ID 列表（带标识 marker）。
+        """
+        url = self._api_url(f"issues/{pr_number}/comments")
+        resp = self.session.get(url, timeout=GITEA_TIMEOUT)
+        if resp.status_code != 200:
+            logger.warning(f"获取评论列表失败: HTTP {resp.status_code}")
+            return []
+
+        comments = resp.json()
+        review_comment_ids = []
+        for c in comments:
+            body = c.get("body", "")
+            if marker in body:
+                review_comment_ids.append(c.get("id"))
+        logger.info(f"找到 {len(review_comment_ids)} 条旧的 AI 审查评论")
+        return review_comment_ids
+
+    def delete_pr_comment(self, pr_number: int, comment_id: int) -> bool:
+        """
+        删除 PR 上的指定评论。
+        """
+        url = self._api_url(f"issues/comments/{comment_id}")
+        resp = self.session.delete(url, timeout=GITEA_TIMEOUT)
+        if resp.status_code not in (200, 204):
+            logger.warning(f"删除评论 {comment_id} 失败: HTTP {resp.status_code}")
+            return False
+        return True
+
+
+def call_llm_openai(
+    prompt: str,
+    llm_base_url: str,
+    llm_api_key: str,
+    llm_model: str,
+) -> Optional[str]:
+    """OpenAI 兼容模式调用"""
+    base_url = llm_base_url.rstrip("/") + "/"
+    api_url = f"{base_url}chat/completions"
+
+    headers = {
+        "Authorization": f"Bearer {llm_api_key}",
+        "Content-Type": "application/json",
+    }
+
+    payload = {
+        "model": llm_model,
+        "messages": [
+            {
+                "role": "system",
+                "content": "你是一位严谨的资深代码审查专家，擅长发现代码中的逻辑错误、安全隐患和性能问题。",
+            },
+            {
+                "role": "user",
+                "content": prompt,
+            },
+        ],
+        "temperature": 0.3,
+        "max_tokens": 2048,
+    }
+
+    logger.info(f"调用 LLM (OpenAI兼容): {api_url}, model={llm_model}")
+
+    last_error = None
+    for attempt in range(MAX_RETRIES + 1):
+        try:
+            resp = requests.post(
+                api_url,
+                headers=headers,
+                json=payload,
+                timeout=LLM_TIMEOUT,
+            )
+            if resp.status_code != 200:
+                logger.warning(f"LLM 调用失败 (第 {attempt + 1} 次): " f"HTTP {resp.status_code} - {resp.text[:200]}")
+                last_error = f"HTTP {resp.status_code}"
+                continue
+
+            data = resp.json()
+            choices = data.get("choices", [])
+            if not choices:
+                logger.warning(f"LLM 返回空结果 (第 {attempt + 1} 次)")
+                last_error = "empty choices"
+                continue
+
+            content = choices[0].get("message", {}).get("content", "")
+            if not content.strip():
+                logger.warning(f"LLM 返回空内容 (第 {attempt + 1} 次)")
+                last_error = "empty content"
+                continue
+
+            logger.info(f"LLM 审查完成，结果长度: {len(content)} 字符")
+            return content
+
+        except requests.Timeout:
+            logger.warning(f"LLM 调用超时 (第 {attempt + 1} 次)")
+            last_error = "timeout"
+        except requests.RequestException as e:
+            logger.warning(f"LLM 调用异常 (第 {attempt + 1} 次): {e}")
+            last_error = str(e)
+
+    logger.error(f"LLM 调用最终失败: {last_error}")
+    return None
+
+
+def call_llm_coze(
+    prompt: str,
+    llm_base_url: str,
+    llm_api_key: str,
+    llm_model: str,
+    coze_bot_id: str,
+) -> Optional[str]:
+    """扣子(Coze)原生 Bot API 调用（支持异步轮询）"""
+    import time
+
+    base_url = llm_base_url.rstrip("/") + "/"
+    api_url = f"{base_url}v3/chat"
+
+    headers = {
+        "Authorization": f"Bearer {llm_api_key}",
+        "Content-Type": "application/json",
+    }
+
+    payload = {
+        "bot_id": coze_bot_id,
+        "user_id": "ci-code-review-bot",
+        "stream": False,
+        "additional_messages": [
+            {
+                "role": "user",
+                "content": prompt,
+                "content_type": "text",
+            }
+        ],
+    }
+
+    logger.info(f"调用 LLM (Coze): {api_url}, bot_id={coze_bot_id}")
+
+    last_error = None
+    for attempt in range(MAX_RETRIES + 1):
+        try:
+            resp = requests.post(
+                api_url,
+                headers=headers,
+                json=payload,
+                timeout=LLM_TIMEOUT,
+            )
+            if resp.status_code != 200:
+                logger.warning(f"Coze 调用失败 (第 {attempt + 1} 次): " f"HTTP {resp.status_code} - {resp.text[:300]}")
+                last_error = f"HTTP {resp.status_code}"
+                continue
+
+            data = resp.json()
+            chat_data = data.get("data", {})
+            chat_id = chat_data.get("id", "")
+            conversation_id = chat_data.get("conversation_id", "")
+            status = chat_data.get("status", "")
+
+            # Coze v3 API 异步：先返回 in_progress，需要轮询
+            if status == "in_progress" and conversation_id and chat_id:
+                logger.info(f"Coze 异步处理中，开始轮询... (chat_id={chat_id[:12]}...)")
+                # 轮询 message 列表接口（GET + query参数），最多等 LLM_TIMEOUT 秒
+                poll_url = f"{base_url}v3/chat/message/list"
+                poll_start = time.time()
+                poll_interval = 3  # 每3秒轮询一次
+
+                while time.time() - poll_start < LLM_TIMEOUT:
+                    time.sleep(poll_interval)
+                    poll_params = {
+                        "chat_id": chat_id,
+                        "conversation_id": conversation_id,
+                    }
+                    poll_resp = requests.get(
+                        poll_url,
+                        headers=headers,
+                        params=poll_params,
+                        timeout=GITEA_TIMEOUT,
+                    )
+                    if poll_resp.status_code != 200:
+                        logger.debug(f"轮询返回 HTTP {poll_resp.status_code}: {poll_resp.text[:100]}")
+                        continue
+
+                    poll_data = poll_resp.json()
+                    if poll_data.get("code", 0) != 0:
+                        logger.debug(f"轮询返回错误: {poll_data.get('msg', '')}")
+                        continue
+
+                    messages = poll_data.get("data", []) or []
+
+                    # 找assistant的answer消息
+                    content = None
+                    for msg in messages:
+                        if msg.get("role") == "assistant" and msg.get("type") == "answer":
+                            content = msg.get("content", "")
+                            break
+
+                    if content and content.strip():
+                        logger.info(f"Coze 审查完成，结果长度: {len(content)} 字符")
+                        return content
+
+                logger.warning(f"Coze 轮询超时 ({LLM_TIMEOUT}s)，未拿到结果")
+                last_error = "poll timeout"
+                continue
+
+            # 同步返回的情况（兼容）
+            content = None
+            messages = chat_data.get("messages", []) or data.get("messages", [])
+            for msg in messages:
+                if msg.get("role") == "assistant" and msg.get("type") == "answer":
+                    content = msg.get("content", "")
+                    break
+
+            if not content:
+                content = chat_data.get("content") or data.get("content")
+
+            if not content:
+                choices = data.get("choices", [])
+                if choices:
+                    content = choices[0].get("message", {}).get("content", "")
+
+            if not content or not content.strip():
+                logger.warning(f"Coze 返回空内容 (第 {attempt + 1} 次): {str(data)[:200]}")
+                last_error = "empty content"
+                continue
+
+            logger.info(f"Coze 审查完成，结果长度: {len(content)} 字符")
+            return content
+
+        except requests.Timeout:
+            logger.warning(f"Coze 调用超时 (第 {attempt + 1} 次)")
+            last_error = "timeout"
+        except requests.RequestException as e:
+            logger.warning(f"Coze 调用异常 (第 {attempt + 1} 次): {e}")
+            last_error = str(e)
+
+    logger.error(f"Coze 调用最终失败: {last_error}")
+    return None
+
+
+def call_llm_for_review(
+    diff_text: str,
+    pr_number: int,
+    file_list: list,
+    llm_base_url: str,
+    llm_api_key: str,
+    llm_model: str,
+    coze_bot_id: str = "",
+) -> Optional[str]:
+    """
+    调用 LLM 进行代码审查，返回审查结果文本。
+    失败时返回 None。
+    根据 LLM_PROVIDER 环境变量选择调用方式。
+    """
+    prompt = build_review_prompt(diff_text, pr_number, file_list)
+    logger.info(f"Prompt 长度: {len(prompt)} 字符")
+
+    provider = LLM_PROVIDER
+
+    if provider == "coze":
+        return call_llm_coze(prompt, llm_base_url, llm_api_key, llm_model, coze_bot_id)
+    else:
+        # 默认 OpenAI 兼容
+        return call_llm_openai(prompt, llm_base_url, llm_api_key, llm_model)
+
+
+# ============== 主流程 ==============
+def main():
+    parser = argparse.ArgumentParser(description="CI AI 代码审查脚本")
+    parser.add_argument("--pr", type=int, help="PR 编号（也可通过 PR_NUMBER 环境变量）")
+    parser.add_argument("--repo", type=str, help="仓库名 owner/repo（也可通过 REPO_NAME 环境变量）")
+    parser.add_argument("--gitea-url", type=str, help="Gitea 地址（也可通过 GITEA_API_URL 环境变量）")
+    parser.add_argument("--gitea-token", type=str, help="Gitea Token（也可通过 GITEA_TOKEN 环境变量）")
+    parser.add_argument("--dry-run", action="store_true", help="只输出审查结果，不发表评论")
+    args = parser.parse_args()
+
+    # 读取配置
+    gitea_url = args.gitea_url or os.getenv("GITEA_API_URL") or os.getenv("GITEA_SERVER_URL")
+    gitea_token = args.gitea_token or os.getenv("GITEA_TOKEN")
+    repo_name = args.repo or os.getenv("REPO_NAME") or os.getenv("GITEA_REPO")
+    pr_number = args.pr or int(os.getenv("PR_NUMBER") or os.getenv("GITEA_PR_NUMBER") or 0)
+
+    llm_base_url = os.getenv("LLM_BASE_URL")
+    llm_api_key = os.getenv("LLM_API_KEY")
+    llm_model = os.getenv("LLM_MODEL", "")
+    coze_bot_id = os.getenv("COZE_BOT_ID", os.getenv("COZE_BOTID", ""))
+
+    # 根据 provider 设置默认值
+    provider = LLM_PROVIDER
+    if provider == "coze":
+        # 扣子模式：默认国内站，key 兼容多种环境变量名
+        if not llm_base_url:
+            llm_base_url = "https://api.coze.cn"
+        if not llm_api_key:
+            llm_api_key = os.getenv("COZE_API_KEY", "") or os.getenv("COZE_PAT", "")
+    else:
+        # OpenAI兼容模式：默认模型
+        if not llm_model:
+            llm_model = "gpt-4o-mini"
+
+    # 必要参数校验
+    missing = []
+    if not gitea_url:
+        missing.append("GITEA_API_URL")
+    if not gitea_token:
+        missing.append("GITEA_TOKEN")
+    if not repo_name:
+        missing.append("REPO_NAME")
+    if not pr_number:
+        missing.append("PR_NUMBER")
+    if not llm_base_url:
+        missing.append("LLM_BASE_URL")
+    if not llm_api_key:
+        missing.append("LLM_API_KEY")
+    if provider == "coze" and not coze_bot_id:
+        missing.append("COZE_BOT_ID (扣子模式需要)")
+
+    if missing:
+        logger.error(f"缺少必要配置: {', '.join(missing)}")
+        sys.exit(1)
+
+    logger.info(f"开始审查 PR #{pr_number}，仓库: {repo_name}")
+    logger.info(f"Gitea: {gitea_url}")
+    logger.info(f"LLM: {llm_base_url} (model={llm_model})")
+
+    try:
+        # 1. 初始化 Gitea 客户端
+        gitea = GiteaClient(gitea_url, gitea_token, repo_name)
+
+        # 2. 获取 PR diff 和文件列表
+        try:
+            diff_text = gitea.get_pr_diff(pr_number)
+            file_list = gitea.get_pr_files(pr_number)
+        except Exception as e:
+            logger.error(f"获取 PR 信息失败: {e}")
+            sys.exit(1)
+
+        # 3. 过滤掉不需要审查的文件（如 lock 文件、生成的文件、二进制文件等）
+        skip_extensions = (
+            ".lock",
+            ".sum",
+            ".min.js",
+            ".min.css",
+            ".map",
+            ".png",
+            ".jpg",
+            ".jpeg",
+            ".gif",
+            ".svg",
+            ".ico",
+            ".woff",
+            ".woff2",
+            ".ttf",
+            ".eot",
+        )
+        skipped_files = []
+        if file_list:
+            skipped_files = [
+                f.get("filename")
+                for f in file_list
+                if f.get("filename", "").endswith(skip_extensions) or f.get("status") == "removed"
+            ]
+            if skipped_files:
+                logger.info(f"跳过 {len(skipped_files)} 个非文本/已删除文件: {', '.join(skipped_files[:5])}...")
+
+        # 实际从 diff 中移除跳过的文件（按文件边界切割）
+        if skipped_files:
+            diff_lines = diff_text.split("\n")
+            filtered_lines = []
+            current_file = None
+            skip_current = False
+            i = 0
+            while i < len(diff_lines):
+                line = diff_lines[i]
+                # 检测新文件开始: diff --git a/xxx b/xxx
+                if line.startswith("diff --git "):
+                    # 提取文件名
+                    parts = line.split(" ")
+                    if len(parts) >= 4:
+                        # b/ 后面的是目标文件名
+                        current_file = parts[3][2:] if parts[3].startswith("b/") else parts[3]
+                        skip_current = any(current_file == sf for sf in skipped_files) or any(
+                            current_file.endswith(ext) for ext in skip_extensions
+                        )
+                    else:
+                        skip_current = False
+                if not skip_current:
+                    filtered_lines.append(line)
+                i += 1
+            original_len = len(diff_text)
+            diff_text = "\n".join(filtered_lines)
+            logger.info(f"Diff 过滤后: {original_len} -> {len(diff_text)} 字符 (减少 {original_len - len(diff_text)})")
+
+        # 4. 截断过大的 diff
+        diff_text, was_truncated = truncate_diff(diff_text, MAX_DIFF_CHARS)
+        if was_truncated:
+            logger.warning(f"Diff 过大，已截断至 {len(diff_text)} 字符")
+
+        # 5. 如果 diff 为空，直接跳过
+        if not diff_text.strip():
+            logger.info("Diff 为空，无需审查")
+            sys.exit(0)
+
+        # 6. 调用 LLM 审查
+        review_result = call_llm_for_review(
+            diff_text=diff_text,
+            pr_number=pr_number,
+            file_list=file_list,
+            llm_base_url=llm_base_url,
+            llm_api_key=llm_api_key,
+            llm_model=llm_model,
+            coze_bot_id=coze_bot_id,
+        )
+
+        if not review_result:
+            logger.error("LLM 审查失败")
+            sys.exit(1)
+
+        # 7. 加上审查时间和标识（便于识别是自动审查）
+        from datetime import datetime
+
+        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        marker = "<!-- AI_CODE_REVIEW_AUTO_COMMENT -->"
+        full_comment = f"""{review_result}
+
+---
+<sub>🤖 由 AI 代码审查机器人自动生成 | {timestamp} | 模型: {llm_model}</sub>
+
+{marker}
+"""
+
+        # 8. 输出审查结果到日志
+        logger.info("=" * 60)
+        logger.info("审查结果:")
+        for line in review_result.split("\n")[:30]:
+            logger.info(line)
+        if len(review_result.split("\n")) > 30:
+            logger.info(f"... 共 {len(review_result.split(chr(10)))} 行")
+        logger.info("=" * 60)
+
+        # 9. 发布评论（先删除旧的审查评论，避免刷屏）
+        if args.dry_run:
+            logger.info("--dry-run 模式，跳过发布评论")
+            print(full_comment)
+        else:
+            # 去重：删除之前的 AI 审查评论
+            old_comments = gitea.get_existing_review_comments(pr_number, marker)
+            if old_comments:
+                logger.info(f"找到 {len(old_comments)} 条旧的 AI 审查评论，先删除")
+                for cid in old_comments:
+                    gitea.delete_pr_comment(pr_number, cid)
+            # 发布新评论
+            success = gitea.post_pr_comment(pr_number, full_comment)
+            if not success:
+                logger.error("评论发布失败")
+                sys.exit(1)
+
+        # 10. 判断是否有严重问题（可选阻断）
+        # 目前只做建议，不阻断合并，始终返回 0
+        has_critical = "问题" in review_result and ("❌" in review_result or "需修改" in review_result)
+        if has_critical:
+            logger.warning("检测到需修改的问题，但当前配置为仅建议，不阻断合并")
+
+        logger.info("代码审查完成")
+        sys.exit(0)
+
+    except Exception as e:
+        logger.exception(f"审查脚本发生未预期的异常: {e}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

scripts/ci_staging_deploy.sh

@@ -1,6 +1,6 @@
 #!/bin/sh
 # ===========================================
-# Staging 部署脚本（SSH 模式）
+# Staging 部署脚本（SSH 模式，支持自动回滚）
 # ===========================================
 # 通过 SSH 在 staging 服务器上执行
 #
@@ -12,9 +12,44 @@
 #   ENV_FILE        - 环境变量文件路径
 #   GENERATED_DIR   - 生成文件目录
 #   SKIP_MIGRATION  - 跳过数据库迁移（true/false，默认 false）
+#   SKIP_ROLLBACK   - 失败时跳过自动回滚（true/false，默认 false）
 
 set -eu
 
+# ---- 重试工具函数 ----
+retry_cmd() {
+  local max_attempts=$1
+  local backoff=$2
+  shift 2
+  local attempt=1
+  while [ $attempt -le $max_attempts ]; do
+    if "$@"; then
+      return 0
+    fi
+    echo "  attempt $attempt/$max_attempts failed, retrying in ${backoff}s..."
+    sleep $backoff
+    backoff=$((backoff * 2))
+    attempt=$((attempt + 1))
+  done
+  echo "  ERROR: failed after $max_attempts retries"
+  return 1
+}
+
+retry_docker_login() {
+  echo "Logging in to registry (up to 3 retries)"
+  if retry_cmd 3 5 sh -c "printf %s "$REGISTRY_TOKEN" | docker login "$REGISTRY_HOST" -u "$REGISTRY_USER" --password-stdin"; then
+    return 0
+  fi
+  echo "WARN: docker login failed after retries, will try pull anyway"
+  return 0
+}
+
+retry_docker_pull() {
+  local image=$1
+  echo "Pulling $image (up to 3 retries)"
+  retry_cmd 3 10 docker pull "$image"
+}
+
 IMAGE_TAG="${IMAGE_TAG:-}"
 REGISTRY="${REGISTRY:-xiaoxia-registry.cn-hangzhou.cr.aliyuncs.com/xiaoxiakeji}"
 REGISTRY_USER="${ACR_USERNAME:-${REGISTRY_USER:-nick0415343655}}"
@@ -25,6 +60,7 @@ GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-staging/generated}"
 LEGACY_ASSETS_DIR="${LEGACY_ASSETS_DIR:-/var/lib/xiaoxia-saas-staging/legacy-assets}"
 
 SKIP_MIGRATION="${SKIP_MIGRATION:-false}"
+SKIP_ROLLBACK="${SKIP_ROLLBACK:-false}"
 
 if [ -z "$IMAGE_TAG" ]; then
   echo "ERROR: IMAGE_TAG is required"
@@ -35,17 +71,165 @@ test -f "$ENV_FILE"
 mkdir -p "$GENERATED_DIR"
 mkdir -p "$LEGACY_ASSETS_DIR"
 
-echo "=========================================="
+echo "==========================================="
 echo "  Staging 部署 - $IMAGE_TAG"
-echo "=========================================="
+echo "==========================================="
+
+# ---- 记录当前运行的镜像版本（用于回滚） ----
+echo "Recording current image versions for rollback..."
+PREV_API_IMAGE=""
+PREV_WORKER_IMAGE=""
+PREV_WEB_IMAGE=""
+for c in xiaoxia-api-staging xiaoxia-worker-staging xiaoxia-web-staging; do
+  if docker inspect "$c" >/dev/null 2>&1; then
+    img=$(docker inspect -f '{{.Config.Image}}' "$c")
+    case "$c" in
+      xiaoxia-api-staging) PREV_API_IMAGE="$img" ;;
+      xiaoxia-worker-staging) PREV_WORKER_IMAGE="$img" ;;
+      xiaoxia-web-staging) PREV_WEB_IMAGE="$img" ;;
+    esac
+    echo "  $c -> $img"
+  else
+    echo "  $c -> (not running)"
+  fi
+done
+
+# ---- 回滚函数 ----
+rollback() {
+  echo ""
+  echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"
+  echo "  部署失败，正在自动回滚到上一版本..."
+  echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"
+  echo ""
+
+  if [ "$SKIP_ROLLBACK" = "true" ]; then
+    echo "SKIP_ROLLBACK=true，跳过自动回滚"
+    exit 1
+  fi
+
+  # 停止当前（失败的）新容器
+  echo "Stopping new containers..."
+  docker rm -f xiaoxia-api-staging 2>/dev/null || true
+  docker rm -f xiaoxia-worker-staging 2>/dev/null || true
+  docker rm -f xiaoxia-web-staging 2>/dev/null || true
+
+  LOG_OPTS="--log-driver json-file --log-opt max-size=50m --log-opt max-file=3"
+
+  # 恢复 API
+  if [ -n "$PREV_API_IMAGE" ]; then
+    echo "Rolling back API to: $PREV_API_IMAGE"
+    docker run -d \
+      --name xiaoxia-api-staging \
+      --env-file "$ENV_FILE" \
+      --network xiaoxia-net-staging \
+      -p 127.0.0.1:8000:8000 \
+      -e APP_ENV=staging \
+      -e APP_VERSION="$(echo $PREV_API_IMAGE | grep -oE '[^:]+$')" \
+      -e GENERATED_FILES_DIR=/app/generated \
+      -e GENERATED_FILES_URL_PREFIX=/generated-files \
+      -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
+      -v "$GENERATED_DIR:/app/generated" \
+      --restart unless-stopped \
+      --health-cmd "python -c \"import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=5)\"" \
+      --health-interval 30s \
+      --health-timeout 10s \
+      --health-retries 3 \
+      --health-start-period 40s \
+      $LOG_OPTS \
+      "$PREV_API_IMAGE"
+  else
+    echo "No previous API image to roll back to"
+  fi
+
+  # 恢复 Worker
+  if [ -n "$PREV_WORKER_IMAGE" ]; then
+    echo "Rolling back Worker to: $PREV_WORKER_IMAGE"
+    docker run -d \
+      --name xiaoxia-worker-staging \
+      --env-file "$ENV_FILE" \
+      --network xiaoxia-net-staging \
+      -e APP_ENV=staging \
+      -e APP_VERSION="$(echo $PREV_WORKER_IMAGE | grep -oE '[^:]+$')" \
+      -e WORKER_CONCURRENCY=1 \
+      -e WORKER_MAX_TASKS_PER_CHILD=100 \
+      -e GENERATED_FILES_DIR=/app/generated \
+      -e GENERATED_FILES_URL_PREFIX=/generated-files \
+      -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
+      -v "$GENERATED_DIR:/app/generated" \
+      --restart unless-stopped \
+      --health-cmd "sh -c \"grep -q celery /proc/1/cmdline || exit 1\"" \
+      --health-interval 30s \
+      --health-timeout 10s \
+      --health-retries 3 \
+      --health-start-period 30s \
+      $LOG_OPTS \
+      "$PREV_WORKER_IMAGE"
+  else
+    echo "No previous Worker image to roll back to"
+  fi
+
+  # 恢复 Web
+  if [ -n "$PREV_WEB_IMAGE" ]; then
+    echo "Rolling back Web to: $PREV_WEB_IMAGE"
+    LEGACY_VOLUME=""
+    if [ -d "$LEGACY_ASSETS_DIR" ] && [ "$(ls -A "$LEGACY_ASSETS_DIR" 2>/dev/null)" ]; then
+      LEGACY_VOLUME="-v ${LEGACY_ASSETS_DIR}:/usr/share/nginx/html/assets-legacy/assets:ro"
+    fi
+    docker run -d \
+      --name xiaoxia-web-staging \
+      --network xiaoxia-net-staging \
+      -p 127.0.0.1:3001:80 \
+      --restart unless-stopped \
+      $LEGACY_VOLUME \
+      --health-cmd "wget --spider -q http://127.0.0.1:80" \
+      --health-interval 30s \
+      --health-timeout 5s \
+      --health-retries 3 \
+      $LOG_OPTS \
+      "$PREV_WEB_IMAGE"
+  else
+    echo "No previous Web image to roll back to"
+  fi
+
+  # 等待 API 回滚后恢复健康
+  if [ -n "$PREV_API_IMAGE" ]; then
+    echo "Waiting for rolled-back API to become healthy..."
+    i=0
+    while [ "$i" -lt 40 ]; do
+      if curl -sf --max-time 5 http://127.0.0.1:8000/health >/dev/null 2>&1; then
+        echo "Rolled-back API is healthy!"
+        break
+      fi
+      i=$((i + 1))
+      echo "  Waiting... ($i/40)"
+      sleep 3
+    done
+    if [ "$i" -ge 40 ]; then
+      echo "WARN: Rolled-back API did not become healthy within 120s"
+      docker logs --tail 30 xiaoxia-api-staging
+    fi
+  fi
+
+  echo ""
+  echo "==========================================="
+  echo "  回滚完成"
+  echo "==========================================="
+  echo "Previous API:    ${PREV_API_IMAGE:-none}"
+  echo "Previous Worker: ${PREV_WORKER_IMAGE:-none}"
+  echo "Previous Web:    ${PREV_WEB_IMAGE:-none}"
+  echo ""
+  echo "部署失败，已自动回滚到上一版本"
+  docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Image}}" | grep staging
+  exit 1
+}
 
 # ---- 登录 Registry ----
 if [ -n "$REGISTRY_TOKEN" ]; then
-  echo "Logging in to registry: $REGISTRY"
+  echo "=========================================="
+  echo "  Login to Registry (with retries)"
+  echo "=========================================="
   REGISTRY_HOST=$(echo "$REGISTRY" | cut -d/ -f1)
-  printf %s "$REGISTRY_TOKEN" | docker login "$REGISTRY_HOST" -u "$REGISTRY_USER" --password-stdin 2>/dev/null || {
-    echo "WARN: docker login failed, will try to pull anyway"
-  }
+  retry_docker_login
 fi
 
 # ---- Pull 新版本镜像 ----
@@ -57,12 +241,12 @@ LOCAL_API="xiaoxia-saas-api:${IMAGE_TAG}"
 LOCAL_WORKER="xiaoxia-saas-worker:${IMAGE_TAG}"
 LOCAL_WEB="xiaoxia-saas-web:${IMAGE_TAG}"
 
-echo "Pulling API image..."
-docker pull "$REGISTRY_API"
-echo "Pulling Worker image..."
-docker pull "$REGISTRY_WORKER"
-echo "Pulling Web image..."
-docker pull "$REGISTRY_WEB"
+echo "=========================================="
+echo "  Pull images (with retries)"
+echo "=========================================="
+retry_docker_pull "$REGISTRY_API"
+retry_docker_pull "$REGISTRY_WORKER"
+retry_docker_pull "$REGISTRY_WEB"
 
 # Re-tag 成本地名
 docker tag "$REGISTRY_API" "$LOCAL_API"
@@ -117,7 +301,12 @@ if [ "$SKIP_MIGRATION" != "true" ]; then
     --env-file "$ENV_FILE" \
     --network xiaoxia-net-staging \
     -e APP_ENV=staging \
-    "$LOCAL_API" sh -c "cd /app && alembic upgrade head"
+    "$LOCAL_API" sh -c "cd /app && alembic upgrade head" || {
+    echo "ERROR: Database migration failed"
+    echo "Note: Migration failures are NOT automatically rolled back (data safety)"
+    echo "Please manually check and fix the migration, then redeploy"
+    exit 1
+  }
   echo "Migrations completed."
 else
   echo "Skipping migrations (SKIP_MIGRATION=true)"
@@ -151,7 +340,7 @@ docker run -d \
   --health-retries 3 \
   --health-start-period 40s \
   $LOG_OPTS \
-  "$LOCAL_API"
+  "$LOCAL_API" || rollback
 
 # ---- 启动 Worker ----
 echo "Starting Worker container..."
@@ -174,7 +363,7 @@ docker run -d \
   --health-retries 3 \
   --health-start-period 30s \
   $LOG_OPTS \
-  "$LOCAL_WORKER"
+  "$LOCAL_WORKER" || rollback
 
 # ---- 启动 Web ----
 LEGACY_VOLUME=""
@@ -197,7 +386,7 @@ docker run -d \
   --health-timeout 5s \
   --health-retries 3 \
   $LOG_OPTS \
-  "$LOCAL_WEB"
+  "$LOCAL_WEB" || rollback
 
 # ---- 等待 API 健康 ----
 echo "Waiting for API to become healthy..."
@@ -215,7 +404,7 @@ done
 if [ "$i" -ge 40 ]; then
   echo "ERROR: API did not become healthy within 120s"
   docker logs --tail 50 xiaoxia-api-staging
-  exit 1
+  rollback
 fi
 
 # ---- 等待 Web 健康 ----
@@ -234,7 +423,7 @@ done
 if [ "$i" -ge 15 ]; then
   echo "ERROR: Web did not become healthy within 30s"
   docker logs --tail 30 xiaoxia-web-staging
-  exit 1
+  rollback
 fi
 
 # ---- 清理旧镜像 ----

tests/unit/test_edit_template_service.py

@@ -463,3 +463,291 @@ class TestCompositeQueries:
         t = svc.create_template(name="空模板")
         result = svc.get_template_with_configs(t.id)
         assert len(result["clip_configs"]) == 0
+
+
+# ===========================================================================
+# Stub Repositories for EditPlan (save_as_template 测试用)
+# ===========================================================================
+
+
+class StubEditPlanRepository:
+    """内存中的 EditPlan 仓储 stub"""
+
+    def __init__(self) -> None:
+        self._plans: dict[str, EditPlan] = {}
+        self._counter = 0
+
+    def _next_id(self) -> str:
+        self._counter += 1
+        return f"plan-{self._counter:03d}"
+
+    def get(self, plan_id: str) -> Optional[EditPlan]:
+        return self._plans.get(plan_id)
+
+    def create(self, plan: EditPlan) -> EditPlan:
+        if not plan.id:
+            plan.id = self._next_id()
+        self._plans[plan.id] = plan
+        return plan
+
+
+class StubEditPlanClipRepository:
+    """内存中的 EditPlanClip 仓储 stub"""
+
+    def __init__(self) -> None:
+        self._clips: dict[str, EditPlanClip] = {}
+        self._counter = 0
+
+    def _next_id(self) -> str:
+        self._counter += 1
+        return f"clip-{self._counter:03d}"
+
+    def list_by_plan(
+        self,
+        plan_id: str,
+        *,
+        status: Optional[str] = None,
+        skip: int = 0,
+        limit: int = 100,
+    ) -> List[EditPlanClip]:
+        items = [c for c in self._clips.values() if c.plan_id == plan_id]
+        if status:
+            items = [c for c in items if c.status.value == status]
+        items.sort(key=lambda c: c.order)
+        return items[skip : skip + limit]
+
+    def create(self, clip: EditPlanClip) -> EditPlanClip:
+        if not clip.id:
+            clip.id = self._next_id()
+        self._clips[clip.id] = clip
+        return clip
+
+
+def _make_service_with_plan_stubs():
+    """创建使用 stub 仓储的 EditTemplateService（含 plan 相关 stub）"""
+    from app.services.edit_template_service import EditTemplateService
+
+    db = MagicMock()
+    svc = EditTemplateService(db)
+    svc._template_repo = StubEditTemplateRepository()
+    svc._clip_config_repo = StubTemplateClipConfigRepository()
+    svc._plan_repo = StubEditPlanRepository()
+    svc._plan_clip_repo = StubEditPlanClipRepository()
+    return svc
+
+
+def _make_test_plan_with_clips(svc, *, clip_count: int = 3, plan_config=None):
+    """辅助方法：创建一个带片段的测试计划，返回 plan 对象"""
+    from packages.domain.edit_plan import EditPlan
+    from packages.domain.edit_plan_clip import EditPlanClip
+
+    plan = EditPlan.create(
+        template_id="tpl-source",
+        name="我的剪辑计划",
+        config=plan_config or {"editing_mode": "one_take", "theme": "minimal"},
+        project_id="proj-001",
+        created_by_user_id="user-001",
+    )
+    plan.id = "plan-test-001"
+    svc._plan_repo.create(plan)
+
+    for i in range(clip_count):
+        clip = EditPlanClip.create(
+            plan_id=plan.id,
+            clip_type=ClipType.MAIN.value,
+            order=i,
+            asset_id=f"asset-{i:03d}",
+            text_content=f"片段{i}的文案",
+            duration=10.0 + i * 5,
+            transition_effect="cut" if i == 0 else "fade",
+            playback_speed=1.0 if i == 0 else 1.5,
+            config={"filter": "vivid"} if i == 1 else {},
+        )
+        svc._plan_clip_repo.create(clip)
+
+    return plan
+
+
+# ===========================================================================
+# 保存为模板测试
+# ===========================================================================
+
+
+class TestSavePlanAsTemplate:
+    """从剪辑计划保存为模板测试"""
+
+    def test_basic_save_as_template(self):
+        """基础场景：将有3个片段的计划保存为模板"""
+        svc = _make_service_with_plan_stubs()
+        plan = _make_test_plan_with_clips(svc, clip_count=3)
+
+        result = svc.save_plan_as_template(plan.id, name="我的自定义模板")
+
+        assert result["template"].name == "我的自定义模板"
+        assert result["template"].template_type == "custom"
+        assert result["template"].editing_mode == "one_take"
+        assert result["template"].status == EditTemplateStatus.ACTIVE
+        assert len(result["clip_configs"]) == 3
+
+    def test_clip_configs_correctly_converted(self):
+        """片段正确转换为模板片段配置"""
+        svc = _make_service_with_plan_stubs()
+        plan = _make_test_plan_with_clips(svc, clip_count=2)
+
+        result = svc.save_plan_as_template(plan.id, name="转换测试模板")
+        configs = result["clip_configs"]
+        configs.sort(key=lambda c: c.order)
+
+        # 第0个片段
+        assert configs[0].clip_type == ClipType.MAIN
+        assert configs[0].order == 0
+        assert configs[0].min_duration == 10.0
+        assert configs[0].max_duration == 10.0
+        assert configs[0].text_template == "片段0的文案"
+        assert configs[0].transition_effect.value == "cut"
+        # playback_speed=1.0 不存
+        assert "playback_speed" not in configs[0].config
+
+        # 第1个片段
+        assert configs[1].order == 1
+        assert configs[1].min_duration == 15.0
+        assert configs[1].max_duration == 15.0
+        assert configs[1].transition_effect.value == "fade"
+        # playback_speed=1.5 存入config
+        assert configs[1].config.get("playback_speed") == 1.5
+        # config 中的 filter 保留
+        assert configs[1].config.get("filter") == "vivid"
+
+    def test_no_asset_id_in_template(self):
+        """模板不保留具体素材ID"""
+        svc = _make_service_with_plan_stubs()
+        plan = _make_test_plan_with_clips(svc, clip_count=2)
+
+        result = svc.save_plan_as_template(plan.id, name="素材剥离测试")
+
+        for cfg in result["clip_configs"]:
+            # 模板片段配置没有 asset_id 字段
+            assert not hasattr(cfg, "asset_id") or not getattr(cfg, "asset_id", "")
+            # config 中也不应有素材相关字段
+            assert "asset_info" not in cfg.config
+            assert "source_asset_id" not in cfg.config
+
+    def test_template_config_stripped_of_runtime_fields(self):
+        """模板config剥离运行时字段"""
+        svc = _make_service_with_plan_stubs()
+        plan_config = {
+            "editing_mode": "one_take",
+            "theme": "cinematic",
+            "asset_ids": ["a1", "a2"],
+            "source_edit_plan_id": "old-plan",
+            "generation_task_id": "task-123",
+        }
+        plan = _make_test_plan_with_clips(svc, clip_count=1, plan_config=plan_config)
+
+        result = svc.save_plan_as_template(plan.id, name="配置剥离测试")
+
+        tpl_config = result["template"].config
+        assert tpl_config.get("theme") == "cinematic"
+        assert "asset_ids" not in tpl_config
+        assert "source_edit_plan_id" not in tpl_config
+        assert "generation_task_id" not in tpl_config
+
+    def test_plan_not_found_raises_error(self):
+        """计划不存在时报错"""
+        svc = _make_service_with_plan_stubs()
+
+        with pytest.raises(ValueError, match="剪辑计划不存在"):
+            svc.save_plan_as_template("nonexistent-plan", name="不存在的计划")
+
+    def test_empty_name_raises_error(self):
+        """模板名称为空时报错"""
+        svc = _make_service_with_plan_stubs()
+        plan = _make_test_plan_with_clips(svc, clip_count=1)
+
+        with pytest.raises(ValueError, match="模板名称不能为空"):
+            svc.save_plan_as_template(plan.id, name="   ")
+
+    def test_duplicate_name_raises_error(self):
+        """模板名称重复时报错"""
+        svc = _make_service_with_plan_stubs()
+        svc.create_template(name="重名模板")
+        plan = _make_test_plan_with_clips(svc, clip_count=1)
+
+        with pytest.raises(ValueError, match="模板名称已存在"):
+            svc.save_plan_as_template(plan.id, name="重名模板")
+
+    def test_save_zero_clip_plan(self):
+        """零片段计划也能保存为模板"""
+        svc = _make_service_with_plan_stubs()
+        from packages.domain.edit_plan import EditPlan
+
+        plan = EditPlan.create(
+            template_id="tpl-source",
+            name="空计划",
+            config={"editing_mode": "one_take"},
+        )
+        plan.id = "plan-empty"
+        svc._plan_repo.create(plan)
+
+        result = svc.save_plan_as_template(plan.id, name="空模板")
+
+        assert result["template"].name == "空模板"
+        assert len(result["clip_configs"]) == 0
+
+    def test_custom_description_and_type(self):
+        """自定义描述和模板类型"""
+        svc = _make_service_with_plan_stubs()
+        plan = _make_test_plan_with_clips(svc, clip_count=1)
+
+        result = svc.save_plan_as_template(
+            plan.id,
+            name="自定义模板",
+            description="这是一个测试模板",
+            template_type="vlog",
+        )
+
+        assert result["template"].description == "这是一个测试模板"
+        assert result["template"].template_type == "vlog"
+
+    def test_unknown_transition_effect_falls_back_to_cut(self):
+        """未知转场效果回退到cut"""
+        svc = _make_service_with_plan_stubs()
+        from packages.domain.edit_plan import EditPlan
+        from packages.domain.edit_plan_clip import EditPlanClip
+
+        plan = EditPlan.create(template_id="tpl-src", name="转场测试计划")
+        plan.id = "plan-transition-test"
+        svc._plan_repo.create(plan)
+
+        clip = EditPlanClip.create(
+            plan_id=plan.id,
+            clip_type="main",
+            order=0,
+            duration=10.0,
+            transition_effect="weird_effect_that_does_not_exist",
+        )
+        svc._plan_clip_repo.create(clip)
+
+        result = svc.save_plan_as_template(plan.id, name="转场兼容模板")
+        assert result["clip_configs"][0].transition_effect.value == "cut"
+
+    def test_unknown_clip_type_falls_back_to_main(self):
+        """未知片段类型回退到main"""
+        svc = _make_service_with_plan_stubs()
+        from packages.domain.edit_plan import EditPlan
+        from packages.domain.edit_plan_clip import EditPlanClip
+
+        plan = EditPlan.create(template_id="tpl-src", name="类型测试计划")
+        plan.id = "plan-type-test"
+        svc._plan_repo.create(plan)
+
+        clip = EditPlanClip.create(
+            plan_id=plan.id,
+            clip_type="unknown_clip_type",
+            order=0,
+            duration=10.0,
+        )
+        svc._plan_clip_repo.create(clip)
+
+        result = svc.save_plan_as_template(plan.id, name="类型兼容模板")
+        assert result["clip_configs"][0].clip_type == ClipType.MAIN

tests/unit/test_voice_clone_workflow.py

@@ -169,6 +169,69 @@ class TestStartClone:
         assert profile.status == VoiceCloneStatus.PENDING
         mock_cosyvoice.submit_clone_task.assert_not_called()
 
+    def test_start_clone_ssrf_internal_url_rejected(self) -> None:
+        """SSRF 防护：内网 URL 应该被拒绝，profile 标记为 failed。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        mock_repo.create.side_effect = lambda p: p
+        mock_repo.update.side_effect = lambda p: p
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        profile = service.start_clone(
+            user_id="user-123",
+            name="测试音色",
+            source_audio_url="http://127.0.0.1/audio.wav",
+        )
+
+        # 内网 IP 应该被拒绝，标记为 failed
+        assert profile.status == VoiceCloneStatus.FAILED
+        assert "安全校验失败" in profile.error_message
+        mock_cosyvoice.submit_clone_task.assert_not_called()
+
+    def test_start_clone_ssrf_private_ip_rejected(self) -> None:
+        """SSRF 防护：私有网段 IP 应该被拒绝，profile 标记为 failed。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        mock_repo.create.side_effect = lambda p: p
+        mock_repo.update.side_effect = lambda p: p
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        profile = service.start_clone(
+            user_id="user-123",
+            name="测试音色",
+            source_audio_url="http://192.168.1.100/audio.wav",
+        )
+
+        assert profile.status == VoiceCloneStatus.FAILED
+        assert "安全校验失败" in profile.error_message
+        mock_cosyvoice.submit_clone_task.assert_not_called()
+
+    def test_start_clone_ssrf_public_url_passes(self) -> None:
+        """SSRF 防护：正常公网 URL 应该通过校验。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        mock_cosyvoice.submit_clone_task.return_value = {
+            "voice_id": "voice-ssrf-test",
+            "status": "DEPLOYING",
+            "request_id": "req-ssrf",
+        }
+        mock_repo.create.side_effect = lambda p: p
+        mock_repo.update.side_effect = lambda p: p
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        profile = service.start_clone(
+            user_id="user-123",
+            name="测试音色",
+            source_audio_url="https://example.com/audio.wav",
+        )
+
+        # 公网 URL 应该正常通过
+        assert profile.status == VoiceCloneStatus.PROCESSING
+        mock_cosyvoice.submit_clone_task.assert_called_once()
+
 
 # ── process_clone_result ─────────────────────────────────
 
@@ -313,6 +376,27 @@ class TestRetryClone:
         assert result.status == VoiceCloneStatus.FAILED
         assert "重试失败" in result.error_message
 
+    def test_retry_clone_ssrf_internal_url_rejected(self) -> None:
+        """重试时 SSRF 防护：内网 URL 应该被拒绝，profile 标记为 failed。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        profile = _make_profile(
+            status=VoiceCloneStatus.FAILED,
+            source_audio_url="http://10.0.0.1/secret.wav",
+            retry_count=0,
+            max_retries=3,
+        )
+        mock_repo.get.return_value = profile
+        mock_repo.update.side_effect = lambda p: p
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        result = service.retry_clone(profile.id, "user-123")
+
+        assert result.status == VoiceCloneStatus.FAILED
+        assert "安全校验失败" in result.error_message
+        mock_cosyvoice.submit_clone_task.assert_not_called()
+
 
 # ── poll_and_process_clone ───────────────────────────────