Merge pull request 'fix: 确认生成 API 改为复用 worker.generate_video 渲染路径' (#1308) from fix/confirm-generation-api into main

- 新增 ConfirmGenerationRequest schema 和 POST /tasks/{task_id}/confirm 端点
- 确认生成复用 worker.generate_video 完整渲染路径（不再走 edit_plan_generation.py）
- 默认分辨率 1080x1920（竖屏），支持自定义封面和标题
- 新增字段: is_preview, source_task_id, output_width, output_height, cover_url, custom_title
- 贯穿 Schema → Domain → Application → DB Model → Repository → Worker 全链路
- 更新 task_center.py 和 generation_tasks.py 的重试逻辑传递新字段
- Alembic 迁移 034: 为 generation_tasks 表添加 6 个新列
- 7 个单元测试覆盖确认生成的正常/异常流程
- 使用 getattr 带默认值确保向后兼容

.gitea/workflows/ci-pipeline.yml

@@ -1153,7 +1153,7 @@ jobs:
       cancel-in-progress: false
     needs:
     - build-staging
-    if: github.event_name == 'push' && (github.ref_name == 'main' || github.ref_name == 'develop')
+    if: github.event_name == 'push' && github.ref_name == 'develop'  # main分支不自动部署staging（数据库版本超前19个迁移）
     steps:
     - name: Checkout code
       shell: sh

alembic/versions/034_add_confirm_generation_fields.py

@@ -0,0 +1,60 @@
+"""Add confirm generation fields
+
+Revision ID: 034
+Revises: 033
+Create Date: 2026-07-08
+
+确认生成 API 改造：为 generation_tasks 表添加 is_preview、source_task_id、
+output_width、output_height、cover_url、custom_title 字段。
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "034"
+down_revision = "033"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "generation_tasks",
+        sa.Column("is_preview", sa.Boolean(), nullable=False, server_default=sa.true()),
+    )
+    op.add_column(
+        "generation_tasks",
+        sa.Column("source_task_id", sa.String(32), nullable=False, server_default=""),
+    )
+    op.add_column(
+        "generation_tasks",
+        sa.Column("output_width", sa.Integer(), nullable=False, server_default=sa.text("1280")),
+    )
+    op.add_column(
+        "generation_tasks",
+        sa.Column("output_height", sa.Integer(), nullable=False, server_default=sa.text("720")),
+    )
+    op.add_column(
+        "generation_tasks",
+        sa.Column("cover_url", sa.String(1000), nullable=False, server_default=""),
+    )
+    op.add_column(
+        "generation_tasks",
+        sa.Column("custom_title", sa.String(500), nullable=False, server_default=""),
+    )
+    op.create_index(
+        op.f("ix_generation_tasks_source_task_id"),
+        "generation_tasks",
+        ["source_task_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_generation_tasks_source_task_id"), table_name="generation_tasks")
+    op.drop_column("generation_tasks", "custom_title")
+    op.drop_column("generation_tasks", "cover_url")
+    op.drop_column("generation_tasks", "output_height")
+    op.drop_column("generation_tasks", "output_width")
+    op.drop_column("generation_tasks", "source_task_id")
+    op.drop_column("generation_tasks", "is_preview")

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

@@ -17,6 +17,7 @@ from app.schemas.generated_video import (
 )
 from app.schemas.generation_task import (
     BatchGenerationTaskResponse,
+    ConfirmGenerationRequest,
     CreateGenerationTaskRequest,
     GenerationTaskResponse,
     ListGenerationTasksResponse,
@@ -56,6 +57,12 @@ def _to_generation_task_response(task) -> GenerationTaskResponse:
         source_edit_plan_id=task.source_edit_plan_id or "",
         asset_select_mode=getattr(task, "asset_select_mode", ""),
         batch_id=getattr(task, "batch_id", ""),
+        is_preview=getattr(task, "is_preview", True),
+        source_task_id=getattr(task, "source_task_id", ""),
+        output_width=getattr(task, "output_width", 1280),
+        output_height=getattr(task, "output_height", 720),
+        cover_url=getattr(task, "cover_url", ""),
+        custom_title=getattr(task, "custom_title", ""),
         status=task.status,
         progress=task.progress,
         result_count=task.result_count,
@@ -222,6 +229,12 @@ def create_generation_task(
                 source_edit_plan_id=request.source_edit_plan_id,
                 asset_select_mode=request.asset_select_mode,
                 batch_id=batch_id,
+                is_preview=request.is_preview,
+                source_task_id=request.source_task_id,
+                output_width=request.output_width,
+                output_height=request.output_height,
+                cover_url=request.cover_url,
+                custom_title=request.custom_title,
             )
         )
         celery_app.send_task("worker.generate_video", args=[task.id])
@@ -231,6 +244,63 @@ def create_generation_task(
     return BatchGenerationTaskResponse(items=items, total=len(items))
 
 
+@router.post("/tasks/{task_id}/confirm", response_model=BatchGenerationTaskResponse)
+def confirm_generation(
+    task_id: str,
+    request: ConfirmGenerationRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    generation_task_repository: Any = Depends(get_generation_task_repository),
+    project_repository: Any = Depends(get_project_repository),
+) -> BatchGenerationTaskResponse:
+    """确认生成 — 基于预览任务创建正式生成任务。
+
+    查找预览任务，复制其配置，创建新的正式生成任务（is_preview=False），
+    使用高分辨率，复用 worker.generate_video 渲染路径。
+    """
+    # 1. 查找源预览任务
+    source_task = generation_task_repository.get(task_id)
+    if source_task is None:
+        raise HTTPException(status_code=404, detail=f"Preview task {task_id} not found")
+
+    # 2. 权限检查
+    if source_task.created_by_user_id and source_task.created_by_user_id != authenticated_user.user.id:
+        raise HTTPException(status_code=403, detail="Access denied to this task")
+    if source_task.project_id:
+        _check_project_access(source_task.project_id, authenticated_user.user.id, project_repository)
+
+    # 3. 创建正式生成任务，复制预览任务的配置
+    use_case = CreateGenerationTaskUseCase(generation_task_repository)
+    new_task = use_case.execute(
+        CreateGenerationTaskCommand(
+            project_id=source_task.project_id,
+            asset_library_id=source_task.asset_library_id,
+            strategy_id=source_task.strategy_id,
+            voice_library_id=source_task.voice_library_id,
+            template_id=source_task.template_id,
+            asset_ids=source_task.asset_ids,
+            title_ids=source_task.title_ids,
+            voice_ids=source_task.voice_ids,
+            created_by_user_id=authenticated_user.user.id,
+            source_edit_plan_id=source_task.source_edit_plan_id or "",
+            asset_select_mode=source_task.asset_select_mode,
+            is_preview=False,
+            source_task_id=task_id,
+            output_width=request.output_width,
+            output_height=request.output_height,
+            cover_url=request.cover_url,
+            custom_title=request.custom_title,
+        )
+    )
+
+    # 4. 调度 worker.generate_video（同一条渲染路径）
+    celery_app.send_task("worker.generate_video", args=[new_task.id])
+
+    return BatchGenerationTaskResponse(
+        items=[_to_generation_task_response(new_task)],
+        total=1,
+    )
+
+
 @router.get("/tasks", response_model=ListGenerationTasksResponse)
 def list_generation_tasks(
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
@@ -306,6 +376,12 @@ def retry_generation_task(
             created_by_user_id=authenticated_user.user.id,
             source_edit_plan_id=task.source_edit_plan_id or "",
             asset_select_mode=getattr(task, "asset_select_mode", ""),
+            is_preview=getattr(task, "is_preview", True),
+            source_task_id=getattr(task, "source_task_id", ""),
+            output_width=getattr(task, "output_width", 1280),
+            output_height=getattr(task, "output_height", 720),
+            cover_url=getattr(task, "cover_url", ""),
+            custom_title=getattr(task, "custom_title", ""),
         )
     )
     celery_app.send_task("worker.generate_video", args=[retried.id])

apps/api/app/api/routes/task_center.py

@@ -151,6 +151,12 @@ def retry_task_by_id(
             title_ids=task.title_ids,
             voice_ids=task.voice_ids,
             created_by_user_id=authenticated_user.user.id,
+            is_preview=getattr(task, "is_preview", True),
+            source_task_id=getattr(task, "source_task_id", ""),
+            output_width=getattr(task, "output_width", 1280),
+            output_height=getattr(task, "output_height", 720),
+            cover_url=getattr(task, "cover_url", ""),
+            custom_title=getattr(task, "custom_title", ""),
         )
     )
     celery_app.send_task("worker.generate_video", args=[retried.id])
@@ -233,6 +239,12 @@ def retry_project_task(
                 title_ids=task.title_ids,
                 voice_ids=task.voice_ids,
                 created_by_user_id=authenticated_user.user.id,
+                is_preview=getattr(task, "is_preview", True),
+                source_task_id=getattr(task, "source_task_id", ""),
+                output_width=getattr(task, "output_width", 1280),
+                output_height=getattr(task, "output_height", 720),
+                cover_url=getattr(task, "cover_url", ""),
+                custom_title=getattr(task, "custom_title", ""),
             )
         )
         celery_app.send_task("worker.generate_video", args=[retried.id])

apps/api/app/schemas/generation_task.py

@@ -1,6 +1,15 @@
 from pydantic import BaseModel, Field, model_validator
 
 
+class ConfirmGenerationRequest(BaseModel):
+    """确认生成请求体 — 基于预览任务创建正式生成任务"""
+
+    output_width: int = Field(default=1080, description="输出视频宽度")
+    output_height: int = Field(default=1920, description="输出视频高度")
+    cover_url: str = Field(default="", description="自定义封面图片 URL")
+    custom_title: str = Field(default="", description="自定义视频标题")
+
+
 class CreateGenerationTaskRequest(BaseModel):
     """创建生成任务请求。
 
@@ -31,6 +40,13 @@ class CreateGenerationTaskRequest(BaseModel):
     asset_select_count: int = Field(
         default=0, ge=0, le=100, description="选取数量，0表示全部（仅 random/smart 模式有效）"
     )
+    # ── 预览 / 确认生成 ──
+    is_preview: bool = Field(default=True, description="是否为预览任务")
+    source_task_id: str = Field(default="", description="来源预览任务 ID（确认生成时传入）")
+    output_width: int = Field(default=1280, description="输出视频宽度")
+    output_height: int = Field(default=720, description="输出视频高度")
+    cover_url: str = Field(default="", description="封面图片 URL")
+    custom_title: str = Field(default="", description="自定义视频标题")
 
     @model_validator(mode="after")
     def _check_at_least_one_mode(self) -> "CreateGenerationTaskRequest":
@@ -58,6 +74,12 @@ class GenerationTaskResponse(BaseModel):
     source_edit_plan_id: str = ""
     asset_select_mode: str = ""
     batch_id: str = ""
+    is_preview: bool = True
+    source_task_id: str = ""
+    output_width: int = 1280
+    output_height: int = 720
+    cover_url: str = ""
+    custom_title: str = ""
     status: str
     progress: float
     result_count: int

apps/web/vitest.config.ts

@@ -11,6 +11,12 @@ export default defineConfig({
     globals: true,
     environment: "jsdom",
     setupFiles: "./src/test/setup.ts",
+    exclude: [
+      "node_modules",
+      "e2e",
+      "dist",
+      "build",
+    ],
     coverage: {
       provider: "v8",
       reporter: ["text", "json", "html"],

apps/worker/worker_app/tasks/generation.py

@@ -96,7 +96,9 @@ def _probe_duration(local_path: Path) -> float:
         return OUTPUT_DURATION_SECONDS
 
 
-def _create_fallback_clip(output_path: Path, title: str) -> None:
+def _create_fallback_clip(
+    output_path: Path, title: str, width: int = OUTPUT_WIDTH, height: int = OUTPUT_HEIGHT
+) -> None:
     """创建 fallback 视频（无素材时）"""
     safe_title = title.replace(":", "\\:").replace("'", "\\'")[:80]
     _run_ffmpeg(
@@ -106,7 +108,7 @@ def _create_fallback_clip(output_path: Path, title: str) -> None:
             "-f",
             "lavfi",
             "-i",
-            f"color=c=#111827:s={OUTPUT_WIDTH}x{OUTPUT_HEIGHT}:d={OUTPUT_DURATION_SECONDS}:r={int(OUTPUT_FPS)}",
+            f"color=c=#111827:s={width}x{height}:d={OUTPUT_DURATION_SECONDS}:r={int(OUTPUT_FPS)}",
             "-vf",
             f"drawtext=text='{safe_title}':fontcolor=white:fontsize=48:x=(w-text_w)/2:y=(h-text_h)/2",
             "-c:v",
@@ -204,6 +206,8 @@ def _process_with_editing_mode(
     audio_path: Optional[str],
     mode: str,
     output_path: Path,
+    output_width: int = OUTPUT_WIDTH,
+    output_height: int = OUTPUT_HEIGHT,
 ) -> None:
     """根据剪辑模式处理视频"""
     from video_processing.editing_modes import (
@@ -215,8 +219,8 @@ def _process_with_editing_mode(
 
     config = EditingModeConfig(
         mode=EditingMode(mode),
-        output_width=OUTPUT_WIDTH,
-        output_height=OUTPUT_HEIGHT,
+        output_width=output_width,
+        output_height=output_height,
         output_fps=int(OUTPUT_FPS),
         pip_position=PIPPosition.TOP_RIGHT,
         pip_scale=0.25,
@@ -262,6 +266,11 @@ def generate_video(self, task_id: str) -> dict:
         mode = gen_task.strategy_id or "one_take"
         task_asset_ids = list(gen_task.asset_ids or [])
         batch_id = getattr(gen_task, "batch_id", "") or ""
+        # 新增字段：分辨率、封面、标题
+        output_width = getattr(gen_task, "output_width", OUTPUT_WIDTH) or OUTPUT_WIDTH
+        output_height = getattr(gen_task, "output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT
+        cover_url = getattr(gen_task, "cover_url", "") or ""
+        custom_title = getattr(gen_task, "custom_title", "") or ""
     finally:
         session.close()
 
@@ -293,9 +302,13 @@ def generate_video(self, task_id: str) -> dict:
                     audio_path=audio_path,
                     mode=editing_mode.value,
                     output_path=output_path,
+                    output_width=output_width,
+                    output_height=output_height,
                 )
             else:
-                _create_fallback_clip(output_path, f"Generated Video {task_id[:8]}")
+                _create_fallback_clip(
+                    output_path, f"Generated Video {task_id[:8]}", width=output_width, height=output_height
+                )
 
             file_size = output_path.stat().st_size
             duration = _probe_duration(output_path)
@@ -324,6 +337,8 @@ def generate_video(self, task_id: str) -> dict:
                 duration=duration,
                 video_path=str(output_path),
                 mode=editing_mode.value,
+                width=output_width,
+                height=output_height,
             )
 
             return {
@@ -332,8 +347,8 @@ def generate_video(self, task_id: str) -> dict:
                 "output_path": str(output_path),
                 "file_size": file_size,
                 "duration": duration,
-                "width": OUTPUT_WIDTH,
-                "height": OUTPUT_HEIGHT,
+                "width": output_width,
+                "height": output_height,
                 "mode": editing_mode.value,
             }
     except Exception as error:
@@ -355,6 +370,8 @@ def _create_video_record_and_dedup(
     duration: float,
     video_path: str,
     mode: str,
+    width: int = OUTPUT_WIDTH,
+    height: int = OUTPUT_HEIGHT,
 ) -> None:
     """创建 GeneratedVideo 记录，计算指纹并执行查重（历史 + 批次）。"""
     from uuid import uuid4
@@ -378,8 +395,8 @@ def _create_video_record_and_dedup(
             file_url=file_url,
             file_size=file_size,
             duration=duration,
-            width=OUTPUT_WIDTH,
-            height=OUTPUT_HEIGHT,
+            width=width,
+            height=height,
             fps=OUTPUT_FPS,
             status="completed",
             generation_params={"mode": mode},

docs/schema-metadata-snapshot.json

@@ -1533,6 +1533,54 @@
           "type": "VARCHAR(32)",
           "unique": false
         },
+        {
+          "index": false,
+          "name": "is_preview",
+          "nullable": false,
+          "primary_key": false,
+          "type": "BOOLEAN",
+          "unique": false
+        },
+        {
+          "index": true,
+          "name": "source_task_id",
+          "nullable": false,
+          "primary_key": false,
+          "type": "VARCHAR(32)",
+          "unique": false
+        },
+        {
+          "index": false,
+          "name": "output_width",
+          "nullable": false,
+          "primary_key": false,
+          "type": "INTEGER",
+          "unique": false
+        },
+        {
+          "index": false,
+          "name": "output_height",
+          "nullable": false,
+          "primary_key": false,
+          "type": "INTEGER",
+          "unique": false
+        },
+        {
+          "index": false,
+          "name": "cover_url",
+          "nullable": false,
+          "primary_key": false,
+          "type": "VARCHAR(1000)",
+          "unique": false
+        },
+        {
+          "index": false,
+          "name": "custom_title",
+          "nullable": false,
+          "primary_key": false,
+          "type": "VARCHAR(500)",
+          "unique": false
+        },
         {
           "index": false,
           "name": "metadata",
@@ -1593,6 +1641,13 @@
           "name": "ix_generation_tasks_source_edit_plan_id",
           "unique": false
         },
+        {
+          "columns": [
+            "source_task_id"
+          ],
+          "name": "ix_generation_tasks_source_task_id",
+          "unique": false
+        },
         {
           "columns": [
             "status"

packages/adapters/sqlalchemy_impl/generation_task_repository.py

@@ -27,6 +27,12 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
         source_edit_plan_id=model.source_edit_plan_id or "",
         asset_select_mode=model.asset_select_mode or "",
         batch_id=model.batch_id or "",
+        is_preview=getattr(model, "is_preview", True),
+        source_task_id=getattr(model, "source_task_id", "") or "",
+        output_width=getattr(model, "output_width", 1280) or 1280,
+        output_height=getattr(model, "output_height", 720) or 720,
+        cover_url=getattr(model, "cover_url", "") or "",
+        custom_title=getattr(model, "custom_title", "") or "",
         created_at=model.created_at,
     )
 
@@ -56,6 +62,12 @@ class SQLAlchemyGenerationTaskRepository:
             source_edit_plan_id=task.source_edit_plan_id or None,
             asset_select_mode=task.asset_select_mode or "",
             batch_id=task.batch_id or "",
+            is_preview=task.is_preview,
+            source_task_id=task.source_task_id or "",
+            output_width=task.output_width,
+            output_height=task.output_height,
+            cover_url=task.cover_url or "",
+            custom_title=task.custom_title or "",
             created_at=task.created_at,
         )
         self.session.add(model)
@@ -129,5 +141,11 @@ class SQLAlchemyGenerationTaskRepository:
         model.source_edit_plan_id = task.source_edit_plan_id or None
         model.asset_select_mode = task.asset_select_mode or ""
         model.batch_id = task.batch_id or ""
+        model.is_preview = task.is_preview
+        model.source_task_id = task.source_task_id or ""
+        model.output_width = task.output_width
+        model.output_height = task.output_height
+        model.cover_url = task.cover_url or ""
+        model.custom_title = task.custom_title or ""
         self.session.commit()
         return task

packages/adapters/sqlalchemy_impl/models.py

@@ -255,6 +255,12 @@ class GenerationTaskModel(Base):
     source_edit_plan_id = Column(String(32), nullable=True, index=True)
     asset_select_mode = Column(String(20), nullable=False, default="")
     batch_id = Column(String(32), nullable=False, default="", index=True)
+    is_preview = Column(Boolean, nullable=False, default=True)
+    source_task_id = Column(String(32), nullable=False, default="", index=True)
+    output_width = Column(Integer, nullable=False, default=1280)
+    output_height = Column(Integer, nullable=False, default=720)
+    cover_url = Column(String(1000), nullable=False, default="")
+    custom_title = Column(String(500), nullable=False, default="")
     extra_meta = Column("metadata", JSON, nullable=False, default=dict)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
 

packages/application/generation_tasks.py

@@ -21,6 +21,12 @@ class CreateGenerationTaskCommand:
     source_edit_plan_id: str = ""
     asset_select_mode: str = ""
     batch_id: str = ""
+    is_preview: bool = True
+    source_task_id: str = ""
+    output_width: int = 1280
+    output_height: int = 720
+    cover_url: str = ""
+    custom_title: str = ""
 
 
 class CreateGenerationTaskUseCase:
@@ -48,6 +54,12 @@ class CreateGenerationTaskUseCase:
             source_edit_plan_id=command.source_edit_plan_id,
             asset_select_mode=command.asset_select_mode,
             batch_id=command.batch_id,
+            is_preview=command.is_preview,
+            source_task_id=command.source_task_id,
+            output_width=command.output_width,
+            output_height=command.output_height,
+            cover_url=command.cover_url,
+            custom_title=command.custom_title,
         )
         return self.generation_task_repository.create(task)
 

packages/domain/generation_task.py

@@ -45,6 +45,12 @@ class GenerationTask:
     created_by_user_id: str = ""
     asset_select_mode: str = ""
     batch_id: str = ""
+    is_preview: bool = True
+    source_task_id: str = ""
+    output_width: int = 1280
+    output_height: int = 720
+    cover_url: str = ""
+    custom_title: str = ""
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 
     @classmethod
@@ -63,6 +69,12 @@ class GenerationTask:
         source_edit_plan_id: str = "",
         asset_select_mode: str = "",
         batch_id: str = "",
+        is_preview: bool = True,
+        source_task_id: str = "",
+        output_width: int = 1280,
+        output_height: int = 720,
+        cover_url: str = "",
+        custom_title: str = "",
     ) -> "GenerationTask":
         if not project_id.strip() and not template_id.strip():
             raise ValueError("project_id 或 template_id 至少需要提供一个")
@@ -82,4 +94,10 @@ class GenerationTask:
             source_edit_plan_id=source_edit_plan_id.strip(),
             asset_select_mode=asset_select_mode,
             batch_id=batch_id,
+            is_preview=is_preview,
+            source_task_id=source_task_id,
+            output_width=output_width,
+            output_height=output_height,
+            cover_url=cover_url,
+            custom_title=custom_title,
         )

scripts/ci/auto_merge.sh

@@ -4,8 +4,6 @@
 # 环境变量：GITHUB_TOKEN, MERGE_TOKEN, PR_NUMBER, PR_HEAD_SHA, BASE_REF, GITHUB_API_URL, GITHUB_REPOSITORY
 set -eu
 
-set -eu
-
 echo "PR #${PR_NUMBER} - 检查CI状态+审批并自动合并到${BASE_REF}"
 echo
 echo "模式: 短作业（只检查一次，不满足则退出，由pr-auto-scan定时兜底）"
@@ -25,6 +23,7 @@ TOTAL=$(echo "$FILES" | grep -cv '^$' || true)
 FRONTEND_COUNT=$(echo "$FILES" | grep -c '^apps/web/' || true)
 BACKEND_COUNT=$((TOTAL - FRONTEND_COUNT))
 echo "变更文件: ${TOTAL} 个 (前端: ${FRONTEND_COUNT}, 后端/公共: ${BACKEND_COUNT})"
+echo
 
 # 使用统一的CI Gate门禁（单一检查点，自动处理前端/后端/全栈跳过逻辑）
 CONTEXTS=(
@@ -115,6 +114,11 @@ check_and_merge() {
     if [ "$MERGE_405_COUNT" -ge "$MAX_405_RETRIES" ]; then
       echo "⚠️  连续${MAX_405_RETRIES}次合并返回405，放弃本次自动合并"
       echo "   （pr-auto-scan会继续尝试，需人工确认是否有冲突或门禁问题）"
+      curl -s -X POST \
+        -H "Authorization: token ${MERGE_TOKEN}" \
+        -H "Content-Type: application/json" \
+        -d '{"body": "Auto merge skipped after multiple 405 errors: PR may have conflicts or unresolved checks. Please review manually. This is not a CI failure."}' \
+        "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/issues/${PR_NUMBER}/comments" > /dev/null 2>&1 || true
       exit 0
     fi
     echo "30秒后重试..."

scripts/ci/ci_health_report.py

@@ -96,7 +96,7 @@ def build_feishu_card(data: dict) -> dict:
 
     # 失败详情（最多显示5条）
     fail_detail_lines = []
-    for i, run in enumerate(failed_runs[:5]):
+    for _i, run in enumerate(failed_runs[:5]):
         run_id = run["id"]
         title = run.get("title", "")[:35]
         branch = run.get("branch", "")

scripts/ci/ci_repeated_failure_detector.py

@@ -122,8 +122,8 @@ def analyze_failures(runs):
 
     for run in sorted_runs:
         run_id = run.get("id")
-        run_status = run.get("status", "")
-        run_conclusion = run.get("conclusion", "")
+        run.get("status", "")
+        run.get("conclusion", "")
         run_started = run.get("started_at", run.get("created_at", ""))
         event = run.get("event", "")
 
@@ -135,7 +135,7 @@ def analyze_failures(runs):
 
         for job in jobs:
             name = job.get("name", "")
-            status = job.get("status", "")
+            job.get("status", "")
             conclusion = job.get("conclusion", "")
 
             # 跳过非CI核心job（如AI Code Review、Preview等）
@@ -176,7 +176,7 @@ def analyze_failures(runs):
                 # cancelled不算失败也不打断
 
     # 计算失败率
-    for name, stats in job_stats.items():
+    for _name, stats in job_stats.items():
         total_actual = stats["total"] - stats["skipped"] - stats["cancelled"]
         if total_actual > 0:
             stats["failure_rate"] = round((stats["failure"] + stats["error"]) / total_actual * 100, 1)
@@ -240,10 +240,10 @@ def generate_report(critical, warning, info, days, total_runs):
     lines.append(f"**生成时间**: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")
     lines.append("")
 
-    lines.append(f"## 概览")
+    lines.append("## 概览")
     lines.append("")
-    lines.append(f"| 级别 | 数量 |")
-    lines.append(f"|------|------|")
+    lines.append("| 级别 | 数量 |")
+    lines.append("|------|------|")
     lines.append(f"| 🔴 严重 (连续失败≥{CONSECUTIVE_FAIL_THRESHOLD}次 或 失败率≥50%) | {len(critical)} |")
     lines.append(f"| 🟡 警告 (失败率≥{FAIL_RATE_THRESHOLD}% 且 失败≥{FAIL_THRESHOLD}次) | {len(warning)} |")
     lines.append(f"| 🔵 关注 (失败≥2次) | {len(info)} |")
@@ -350,7 +350,7 @@ def send_feishu_notification(critical, warning, info, days):
 
 
 def main():
-    print(f"=== CI重复失败检测 ===")
+    print("=== CI重复失败检测 ===")
     print(f"统计周期: 最近{DAYS}天")
     print(f"仓库: {REPO}")
     print()

scripts/ci/docker_build_only.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 # PR构建专用：只构建不输出，验证Dockerfile能否正常构建
-# 优先用buildx + 远程缓存，失败自动回退到普通docker build（DooD模式下buildx builder偶发崩溃）
+# 无本地缓存（12个runner不共享，反而添乱），只用ACR远程缓存
 set -eu
 
 NO_CACHE_FLAG=""
@@ -19,64 +19,25 @@ for arg in "$@"; do
 done
 
 BUILDER_NAME="ci-pr-builder-${GITHUB_RUN_ID:-local}"
+if ! docker buildx inspect "$BUILDER_NAME" > /dev/null 2>&1; then
+  docker buildx create --use --name "$BUILDER_NAME" --driver docker-container
+else
+  docker buildx use "$BUILDER_NAME"
+fi
+docker buildx inspect --bootstrap
 
-echo "=== PR Build: buildx + remote cache (attempt 1) ==="
+echo "=== PR Build: build only, no output, remote cache only ==="
 echo "Dockerfile: ${DOCKERFILE}"
 echo "Image tag: ${IMAGE_TAG}"
 echo ""
 
-# --- 尝试 buildx docker-container driver ---
-if ! docker buildx inspect "$BUILDER_NAME" > /dev/null 2>&1; then
-  docker buildx create --use --name "$BUILDER_NAME" --driver docker-container 2>/dev/null || true
-else
-  docker buildx use "$BUILDER_NAME" 2>/dev/null || true
-fi
-docker buildx inspect --bootstrap > /dev/null 2>&1 || true
-
-set +e
 docker buildx build \
   $NO_CACHE_FLAG \
   $BUILD_ARGS \
   --cache-from "type=registry,ref=${CACHE_REF}" \
   -f "${DOCKERFILE}" \
   -t "${IMAGE_TAG}" \
-  --load \
   .
-BUILDX_EXIT=$?
-set -e
-
-if [ $BUILDX_EXIT -eq 0 ]; then
-  echo ""
-  echo "PR build OK (buildx): ${IMAGE_TAG}"
-  exit 0
-fi
 
 echo ""
-echo "⚠️  buildx build失败，回退到普通docker build"
-echo "   原因：buildx builder在DooD模式下偶发不稳定（graceful_stop / buildkitd.sock）"
-echo ""
-
-# 清理 buildx builder
-docker buildx rm "$BUILDER_NAME" 2>/dev/null || true
-
-# --- 回退：普通 docker build ---
-# 注意：普通docker build不支持远程缓存，但更稳定
-set +e
-docker build \
-  $NO_CACHE_FLAG \
-  $BUILD_ARGS \
-  -f "${DOCKERFILE}" \
-  -t "${IMAGE_TAG}" \
-  .
-DOCKER_EXIT=$?
-set -e
-
-if [ $DOCKER_EXIT -eq 0 ]; then
-  echo ""
-  echo "PR build OK (fallback docker build): ${IMAGE_TAG}"
-  exit 0
-fi
-
-echo ""
-echo "❌ PR build failed (both buildx and docker build)"
-exit 1
+echo "PR build OK (build only, no output): ${IMAGE_TAG}"

scripts/ci/run_integration_tests.sh

@@ -4,6 +4,11 @@
 # 支持 pytest-xdist 并行执行：每个 worker 使用独立数据库，预期加速 2-4 倍
 set -eu
 
+# 加载CI共享常量
+SCRIPT_DIR="$(dirname "${BASH_SOURCE[0]}")"
+# shellcheck source=ci_env.sh
+source "${SCRIPT_DIR}/ci_env.sh"
+
 echo "=== CI Integration Tests 开始 ==="
 
 # --- 安装依赖 ---
@@ -47,7 +52,7 @@ bash scripts/ci/step_install_ffmpeg.sh
 # 需要用宿主机IP访问映射端口
 # 检测策略：host.docker.internal -> docker0桥接IP -> 容器IP直连 -> 默认网关 -> 127.0.0.1
 detect_docker_host() {
-  local test_port="${1:-5432}"
+  local test_port="${1:-${CI_LOCAL_PG_PORT}}"
 
   # 候选IP列表
   local candidates=()
@@ -106,7 +111,7 @@ except:
 # 获取宿主机IP（先尝试用共享PG端口5433测试，再回退到其他端口）
 if [ -S /var/run/docker.sock ]; then
   # 先用共享PG端口5433探测
-  DOCKER_HOST_IP=$(detect_docker_host 5433)
+  DOCKER_HOST_IP=$(detect_docker_host "${CI_SHARED_PG_PORT}")
   if [ "$DOCKER_HOST_IP" = "127.0.0.1" ]; then
     # 如果共享PG端口探测失败，说明不在DooD或共享PG不可用，再试其他端口
     DOCKER_HOST_IP=$(detect_docker_host 22)
@@ -182,9 +187,9 @@ if [ "$USE_SHARED_PG" = "true" ]; then
   # 使用常驻共享PG实例
   echo "使用常驻共享PG实例（CI_USE_SHARED_PG=true）"
   SHARED_PG_HOST="$PG_HOST"
-  SHARED_PG_PORT="5433"
-  SHARED_PG_USER="postgres"
-  SHARED_PG_PASSWORD="ci_pg_2026!"
+  SHARED_PG_PORT="${CI_SHARED_PG_PORT}"
+  SHARED_PG_USER="${CI_SHARED_PG_USER}"
+  SHARED_PG_PASSWORD="${CI_SHARED_PG_PASSWORD}"
 
   echo "等待共享PG连接就绪..."
   wait_tcp_ready "$SHARED_PG_HOST" "$SHARED_PG_PORT" 5
@@ -220,9 +225,9 @@ else
     --health-timeout 5s \
     --health-retries 12 \
     postgres:16
-  PG_PORT=$(docker port "$PG_CONTAINER" 5432/tcp | cut -d: -f2)
+  PG_PORT=$(docker port "$PG_CONTAINER" ${CI_LOCAL_PG_PORT}/tcp | cut -d: -f2)
   echo "PostgreSQL port: $PG_PORT"
-  export DATABASE_URL="postgresql+psycopg://postgres:postgres@${PG_HOST}:${PG_PORT}/xiaoxia_saas"
+  export DATABASE_URL="postgresql+psycopg://${CI_SHARED_PG_USER}:${CI_SHARED_PG_PASSWORD}@${PG_HOST}:${PG_PORT}/${CI_DEFAULT_DB}"
 
   # 等待容器健康
   for i in $(seq 1 30); do

scripts/ci/run_unit_tests.sh

@@ -3,6 +3,9 @@
 # 包含：依赖安装、增量测试选择、覆盖率测试、diff覆盖率门禁
 set -eu
 
+# 测试环境必须的密钥变量
+export JWT_SECRET_KEY=${JWT_SECRET_KEY:-test-jwt-secret-for-ci-only-2026}
+
 JOB_NAME="${1:-Unit Tests}"
 
 echo "=== CI Unit Tests 开始 ==="
@@ -42,6 +45,9 @@ for i in 1 2 3; do
 done
 pytest --version
 
+# --- 安装 ffmpeg（视频处理相关测试依赖）---
+bash scripts/ci/step_install_ffmpeg.sh
+
 # 双保险：确保numpy已安装
 python3 -m pip install -q numpy==1.26.4 || true
 
@@ -94,7 +100,7 @@ else
     -m pytest tests/unit -q
   python3 -m coverage report --show-missing
   python3 -m coverage xml -o coverage.xml
-  python3 -m coverage report --fail-under=65 > /dev/null || true  # 全量覆盖率仅作参考，不阻塞合并
+  python3 -m coverage report --fail-under=55 > /dev/null
 fi
 
 # --- Diff 覆盖率检查（仅PR） ---

scripts/ci/run_validate.sh

@@ -14,6 +14,11 @@
 # 所有子任务同时启动，最后汇总结果。
 set -eu
 
+# 加载CI共享常量
+SCRIPT_DIR="$(dirname "${BASH_SOURCE[0]}")"
+# shellcheck source=ci_env.sh
+source "${SCRIPT_DIR}/ci_env.sh"
+
 echo "=== CI Validate: 并行化代码质量检查 ==="
 echo ""
 
@@ -302,7 +307,7 @@ task_alembic() {
 
   # --- DooD模式检测：确定宿主机访问地址 ---
   detect_docker_host() {
-    local test_port="${1:-5432}"
+    local test_port="${1:-${CI_LOCAL_PG_PORT}}"
     local candidates=()
 
     # 1. host.docker.internal
@@ -376,7 +381,7 @@ except:
   # 获取宿主机IP
   local PG_HOST
   if [ -S /var/run/docker.sock ]; then
-    PG_HOST=$(detect_docker_host 5433)
+    PG_HOST=$(detect_docker_host "${CI_SHARED_PG_PORT}")
     if [ "$PG_HOST" = "127.0.0.1" ]; then
       PG_HOST=$(detect_docker_host 22)
     fi
@@ -394,9 +399,9 @@ except:
     # 使用常驻共享PG实例
     echo "使用常驻共享PG实例（CI_USE_SHARED_PG=true）"
     local SHARED_PG_HOST="$PG_HOST"
-    local SHARED_PG_PORT="5433"
-    local SHARED_PG_USER="postgres"
-    local SHARED_PG_PASSWORD="ci_pg_2026!"
+    local SHARED_PG_PORT="${CI_SHARED_PG_PORT}"
+    local SHARED_PG_USER="${CI_SHARED_PG_USER}"
+    local SHARED_PG_PASSWORD="${CI_SHARED_PG_PASSWORD}"
     local CI_DB_NAME="ci_run_${GITHUB_RUN_ID:-$$}"
 
     echo "等待共享PG连接就绪..."
@@ -456,9 +461,9 @@ conn.close()
 
     if [ $exit_code -eq 0 ]; then
       local PG_PORT
-      PG_PORT=$(docker port "$PG_CONTAINER" 5432/tcp | cut -d: -f2)
+      PG_PORT=$(docker port "$PG_CONTAINER" ${CI_LOCAL_PG_PORT}/tcp | cut -d: -f2)
       echo "PostgreSQL port: $PG_PORT"
-      export DATABASE_URL="postgresql+psycopg://postgres:postgres@${PG_HOST}:${PG_PORT}/xiaoxia_saas"
+      export DATABASE_URL="postgresql+psycopg://${CI_SHARED_PG_USER}:${CI_SHARED_PG_PASSWORD}@${PG_HOST}:${PG_PORT}/${CI_DEFAULT_DB}"
 
       # 等待容器健康
       local i

scripts/ci/validate_code_quality.sh

@@ -56,95 +56,18 @@ for fpath, items in data.get('results', {}).items():
 fi
 echo "✅ Secret scan passed"
 
-# --- 增量/全量模式判断 ---
+# --- 代码质量检查（全量，PR 和 push 统一标准）---
+# 历史：PR 侧用增量检查以加速，但会导致 push 侧全量检查失败时 PR 侧感知不到
+# 现在统一全量检查，确保 CI 真正保护主分支（black/isort/ruff 全量仅多几十秒）
 echo ""
-echo "=== [2/6] Code quality checks ==="
+echo "=== [2/6] Code quality checks (full scan) ==="
 SCAN_MODE="full"
-CHANGED_PY_FILES=""
+echo "Full scan mode"
+python3 -m compileall -q alembic apps packages tests scripts
+python3 -m black --check --fast alembic apps packages tests scripts
+python3 -m isort --check-only alembic apps packages tests scripts
+python3 -m ruff check apps packages tests --statistics
 
-if [ "${GITHUB_EVENT_NAME:-}" = "pull_request" ] && [ -n "${GITHUB_REF_NAME:-}" ] && [ -n "${GITHUB_TOKEN:-}" ]; then
-  PR_NUMBER=$(echo "$GITHUB_REF" | sed 's|refs/pull/||; s|/.*||')
-  API_URL="${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}/files?limit=100"
-  set +e
-  RESPONSE=$(curl -s -w "\n%{http_code}" -H "Authorization: token ${GITHUB_TOKEN}" "$API_URL")
-  HTTP_CODE=$(echo "$RESPONSE" | tail -n1)
-  BODY=$(echo "$RESPONSE" | sed '$d')
-  set -e
-  if [ "$HTTP_CODE" = "200" ]; then
-    CHANGED_PY_FILES=$(echo "$BODY" | python3 -c "
-import json, sys
-try:
-    files = json.load(sys.stdin)
-    py_files = [f['filename'] for f in files if f['filename'].endswith('.py') and f['status'] != 'removed']
-    print(' '.join(py_files))
-except Exception:
-    print('')
-")
-    # 新增文件(added)强制全量检查，防止增量漏检
-    ADDED_PY_FILES=$(echo "$BODY" | python3 -c "
-import json, sys
-try:
-    files = json.load(sys.stdin)
-    added = [f['filename'] for f in files if f['filename'].endswith('.py') and f['status'] == 'added']
-    print(' '.join(added))
-except Exception:
-    print('')
-")
-    MODIFIED_PY_FILES=$(echo "$BODY" | python3 -c "
-import json, sys
-try:
-    files = json.load(sys.stdin)
-    modified = [f['filename'] for f in files if f['filename'].endswith('.py') and f['status'] not in ('removed', 'added')]
-    print(' '.join(modified))
-except Exception:
-    print('')
-")
-    if [ -n "$CHANGED_PY_FILES" ]; then
-      SCAN_MODE="incremental"
-      echo "Incremental mode: $(echo "$CHANGED_PY_FILES" | wc -w) Python files changed"
-    else
-      SCAN_MODE="skip_py"
-      echo "No Python files changed in this PR"
-    fi
-  else
-    echo "WARN: API returned HTTP $HTTP_CODE, falling back to full scan"
-  fi
-else
-  echo "Full scan mode (not a PR event)"
-fi
-
-if [ "$SCAN_MODE" = "incremental" ]; then
-  # 防御性过滤
-  EXISTING_PY_FILES=""
-  for f in $CHANGED_PY_FILES; do
-    if [ -f "$f" ]; then
-      if [ -z "$EXISTING_PY_FILES" ]; then
-        EXISTING_PY_FILES="$f"
-      else
-        EXISTING_PY_FILES="$EXISTING_PY_FILES $f"
-      fi
-    fi
-  done
-  CHANGED_PY_FILES="$EXISTING_PY_FILES"
-
-  python3 -m compileall -q $CHANGED_PY_FILES
-  python3 -m black --check --fast $CHANGED_PY_FILES
-  python3 -m isort --check-only $CHANGED_PY_FILES
-  RUFF_FILES=$(echo "$CHANGED_PY_FILES" | tr ' ' '\n' | grep -v '^scripts/' | grep -v '^$' | xargs)
-  if [ -n "$RUFF_FILES" ]; then
-    python3 -m ruff check $RUFF_FILES --statistics
-  else
-    echo "No ruff-checkable files changed, skipping"
-  fi
-elif [ "$SCAN_MODE" = "skip_py" ]; then
-  echo "No Python files changed - skipping Python lint checks"
-else
-  echo "Full scan mode"
-  python3 -m compileall -q alembic apps packages tests scripts
-  python3 -m black --check --fast alembic apps packages tests scripts
-  python3 -m isort --check-only alembic apps packages tests scripts
-  python3 -m ruff check apps packages tests --statistics
-fi
 echo "✅ Code quality checks passed"
 
 # --- Bandit 安全扫描（仅告警） ---

scripts/ci_notify.py

@@ -1,68 +1,185 @@
 #!/usr/bin/env python3
 """
-CI 通知脚本 - 发送飞书消息通知
+统一CI通知脚本 - 发送飞书卡片通知
+支持三种模式: start / success / failure
+包含: PR链接、耗时、失败阶段、分支、提交者、Run链接、Runner信息
+
 用法:
-  NOTIFY_MODE=start JOB_NAME="Build API" python3 scripts/ci_notify.py
-  NOTIFY_MODE=success JOB_NAME="Deploy Staging" python3 scripts/ci_notify.py
-  NOTIFY_MODE=failure JOB_NAME="Unit Tests" python3 scripts/ci_notify.py
+  NOTIFY_MODE=start   JOB_NAME="xxx" python3 scripts/ci_notify.py
+  NOTIFY_MODE=success JOB_NAME="xxx" JOB_DURATION="2m30s" python3 scripts/ci_notify.py
+  NOTIFY_MODE=failure JOB_NAME="xxx" FAILED_STEP="xxx" JOB_DURATION="2m30s" python3 scripts/ci_notify.py
 
 环境变量:
-  NOTIFY_MODE     - 通知类型: start/success/failure
-  JOB_NAME        - Job名称
-  CI_NOTIFY_WEBHOOK - 飞书Webhook地址
-  GITHUB_SHA      - Commit SHA (可选)
-  GITHUB_REF_NAME - 分支名 (可选)
-  GITHUB_RUN_ID   - Run ID (可选)
-  GITHUB_REPOSITORY - 仓库名 (可选)
-  GITHUB_SERVER_URL - Gitea地址 (可选)
+  CI_NOTIFY_WEBHOOK - 飞书webhook地址 (必填)
+  NOTIFY_MODE       - 通知模式: start / success / failure (必填)
+  JOB_NAME          - Job名称 (必填)
+  JOB_DURATION      - 耗时，如"2m30s" (成功/失败时建议传)
+  FAILED_STEP       - 失败的步骤名 (失败时建议传)
+  GITHUB_REF_NAME   - 分支名
+  GITHUB_SHA        - commit SHA
+  GITHUB_ACTOR      - 提交者
+  GITHUB_RUN_ID     - Run ID
+  GITHUB_REPOSITORY - 仓库路径
+  GITHUB_EVENT_NAME - 事件类型 (pull_request / push / ...)
+  GITHUB_PR_NUMBER  - PR编号 (PR事件时)
+  GITHUB_PR_TITLE   - PR标题 (PR事件时)
+  RUNNER_NAME       - Runner名称 (可选，自动获取)
 
 设计原则:
-  通知失败永远不阻断主流程（永远返回0）
+  1. 通知失败永远不阻断CI主流程（返回exit code 0）
+  2. 标题包含"CI通知"/"CI告警"关键词，适配飞书webhook关键词校验
+  3. 卡片信息尽量丰富，方便快速定位问题
 """
 
 import json
 import os
 import sys
 import urllib.request
-from datetime import datetime
 
 
 def get_env(name, default=""):
+    """读取环境变量"""
     return os.environ.get(name, default)
 
 
-def send_feishu_notify(webhook_url, title, content, color="blue"):
-    """发送飞书通知（简单卡片格式）"""
-    if not webhook_url:
-        print("[INFO] 未配置 CI_NOTIFY_WEBHOOK，跳过通知")
-        return True
+def format_duration(seconds_str):
+    """将秒数格式化为易读形式"""
+    try:
+        seconds = int(float(seconds_str))
+        mins = seconds // 60
+        secs = seconds % 60
+        if mins > 0:
+            return f"{mins}m{secs}s"
+        return f"{secs}s"
+    except (ValueError, TypeError):
+        return seconds_str or "未知"
 
-    # 状态颜色映射
-    color_map = {
-        "green": "green",
-        "red": "red",
-        "blue": "blue",
-        "yellow": "yellow",
-    }
-    header_color = color_map.get(color, "blue")
 
-    # 构造卡片
-    card = {
-        "config": {"wide_screen_mode": True},
-        "header": {
-            "title": {"tag": "plain_text", "content": title},
-            "template": header_color,
+def classify_job(job_name):
+    """根据Job名称判断所属阶段"""
+    name = job_name.lower()
+    if any(k in name for k in ["validate", "lint", "unit test", "integration test"]):
+        return "门禁检查"
+    if any(k in name for k in ["build", "image"]):
+        return "镜像构建"
+    if any(k in name for k in ["deploy", "staging", "production"]):
+        return "部署发布"
+    if any(k in name for k in ["e2e", "test", "smoke"]):
+        return "测试验证"
+    return "其他"
+
+
+def main() -> int:
+    webhook = get_env("CI_NOTIFY_WEBHOOK")
+    if not webhook:
+        print("未配置 CI_NOTIFY_WEBHOOK，跳过通知")
+        print("如需启用，请在仓库 Settings -> Secrets and variables -> Actions 中添加 CI_NOTIFY_WEBHOOK")
+        return 0
+
+    mode = get_env("NOTIFY_MODE", "failure").lower()
+    job_name = get_env("JOB_NAME", "Unknown Job")
+    duration = get_env("JOB_DURATION")
+    if not duration:
+        duration_sec = get_env("JOB_DURATION_SECONDS")
+        duration = format_duration(duration_sec) if duration_sec else "计算中..."
+
+    failed_step = get_env("FAILED_STEP", "")
+    branch = get_env("GITHUB_REF_NAME", "unknown")
+    commit = get_env("GITHUB_SHA", "unknown")[:8]
+    actor = get_env("GITHUB_ACTOR", "unknown")
+    run_id = get_env("GITHUB_RUN_ID", "unknown")
+    repo = get_env("GITHUB_REPOSITORY", "unknown")
+    event_name = get_env("GITHUB_EVENT_NAME", "")
+    pr_number = get_env("GITHUB_PR_NUMBER", "")
+    pr_title = get_env("GITHUB_PR_TITLE", "")
+    runner_name = get_env("RUNNER_NAME", "")
+
+    run_url = f"https://git.xiaoxiajianji.com/{repo}/actions/runs/{run_id}"
+    job_stage = classify_job(job_name)
+
+    # 根据模式设置标题、状态、颜色
+    # 注意：标题中必须包含飞书webhook配置的关键词，否则会报"Key Words Not Found"
+    # 这里加入"CI通知"/"CI告警"关键词提高命中率
+    if mode == "start":
+        title = f"🔄 CI通知：{job_name} 开始构建"
+        status = "blue"
+        button_text = "查看进度"
+        button_type = "primary"
+    elif mode == "success":
+        title = f"✅ CI通知：{job_name} 构建成功"
+        status = "green"
+        button_text = "查看详情"
+        button_type = "primary"
+    else:  # failure
+        title = f"❌ CI告警：{job_name} 构建失败"
+        status = "red"
+        button_text = "查看失败日志"
+        button_type = "danger"
+
+    # 构建卡片内容 - 左侧标签+右侧值的结构化展示
+    fields = []
+
+    fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**阶段**\n{job_stage}"}})
+    fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**任务**\n{job_name}"}})
+
+    if mode != "start":
+        fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**耗时**\n{duration}"}})
+    else:
+        fields.append({"is_short": True, "text": {"tag": "lark_md", "content": "**状态**\n进行中"}})
+
+    if runner_name:
+        fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**Runner**\n{runner_name}"}})
+
+    if mode == "failure" and failed_step:
+        fields.append({"is_short": False, "text": {"tag": "lark_md", "content": f"**失败步骤**\n{failed_step}"}})
+
+    # PR/分支信息
+    if event_name == "pull_request" and pr_number:
+        pr_url = f"https://git.xiaoxiajianji.com/{repo}/pulls/{pr_number}"
+        pr_display = f"#{pr_number}"
+        if pr_title:
+            pr_display += f" {pr_title[:30]}"
+        fields.append({"is_short": False, "text": {"tag": "lark_md", "content": f"**PR**\n[{pr_display}]({pr_url})"}})
+    elif event_name == "push":
+        fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**分支**\n{branch}"}})
+
+    fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**提交**\n`{commit}`"}})
+    fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**提交者**\n{actor}"}})
+    fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**Run ID**\n{run_id}"}})
+
+    payload = {
+        "msg_type": "interactive",
+        "card": {
+            "header": {
+                "title": {
+                    "tag": "plain_text",
+                    "content": title,
+                },
+                "status": status,
+            },
+            "elements": [
+                {
+                    "tag": "div",
+                    "fields": fields,
+                },
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": button_text},
+                            "url": run_url,
+                            "type": button_type,
+                        }
+                    ],
+                },
+            ],
         },
-        "elements": [
-            {"tag": "div", "text": {"tag": "lark_md", "content": content}},
-        ],
     }
 
-    payload = {"msg_type": "interactive", "card": card}
     data = json.dumps(payload).encode("utf-8")
-
     req = urllib.request.Request(
-        webhook_url,
+        webhook,
         data=data,
         headers={"Content-Type": "application/json"},
         method="POST",
@@ -70,64 +187,20 @@ def send_feishu_notify(webhook_url, title, content, color="blue"):
     try:
         with urllib.request.urlopen(req, timeout=10) as resp:
             resp_body = resp.read().decode("utf-8")
-        result = json.loads(resp_body)
-        if result.get("code", 0) != 0:
-            print(f"[WARN] 飞书通知返回错误: {result.get('msg', resp_body)}", file=sys.stderr)
-            return False
-        print("[INFO] 飞书通知发送成功")
-        return True
+        # 飞书返回code=0表示成功
+        try:
+            result = json.loads(resp_body)
+            if result.get("code", 0) != 0:
+                print(f"通知发送告警: 飞书返回错误 - {result.get('msg', resp_body)}", file=sys.stderr)
+                print(f"通知已发送 ({mode}) - 飞书返回非0，但不阻断CI流程")
+            else:
+                print(f"通知已发送 ({mode})")
+        except json.JSONDecodeError:
+            print(f"通知已发送 ({mode})")
     except Exception as e:
-        print(f"[WARN] 飞书通知发送失败: {e}", file=sys.stderr)
-        return False
+        print(f"通知发送告警: {e}", file=sys.stderr)
 
-
-def build_message():
-    """根据环境变量构造通知消息"""
-    notify_mode = get_env("NOTIFY_MODE", "info").lower()
-    job_name = get_env("JOB_NAME", "未知Job")
-    branch = get_env("GITHUB_REF_NAME", "未知分支")
-    sha = get_env("GITHUB_SHA", "")[:8]
-    run_id = get_env("GITHUB_RUN_ID", "")
-    repo = get_env("GITHUB_REPOSITORY", "")
-    server_url = get_env("GITHUB_SERVER_URL", "https://git.xiaoxiajianji.com")
-
-    # 状态映射
-    status_map = {
-        "start": ("🔔 CI 任务开始", "blue", "开始执行"),
-        "success": ("✅ CI 任务成功", "green", "执行成功"),
-        "failure": ("❌ CI 任务失败", "red", "执行失败"),
-        "info": ("ℹ️ CI 通知", "blue", "通知"),
-    }
-    title, color, status_text = status_map.get(notify_mode, status_map["info"])
-
-    # 构造内容
-    content_lines = [
-        f"**任务**: {job_name}",
-        f"**状态**: {status_text}",
-        f"**分支**: {branch}",
-    ]
-    if sha:
-        content_lines.append(f"**Commit**: `{sha}`")
-    if run_id and repo and server_url:
-        run_url = f"{server_url}/{repo}/actions/runs/{run_id}"
-        content_lines.append(f"**详情**: [点击查看]({run_url})")
-
-    content_lines.append(f"**时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
-
-    content = "\n".join(content_lines)
-    return title, content, color
-
-
-def main():
-    webhook = get_env("CI_NOTIFY_WEBHOOK", "")
-    title, content, color = build_message()
-
-    print(f"[CI Notify] 模式: {get_env('NOTIFY_MODE')}")
-    print(f"[CI Notify] 任务: {get_env('JOB_NAME')}")
-
-    send_feishu_notify(webhook, title, content, color)
-
-    # 永远返回0，不阻断主流程
+    # 通知无论成功失败都不阻断CI主流程，统一返回0
     return 0
 
 

scripts/ci_staging_deploy.sh

@@ -1,99 +1,340 @@
 #!/bin/sh
-# ============================================
-# CI Staging 部署脚本 - 通过 SSH 在 Staging 服务器执行
-# 用法：IMAGE_TAG=<sha> ACR_USERNAME=<user> ACR_PASSWORD=<pass> sh ci_staging_deploy.sh
-# ============================================
-
+# ===========================================
+# Staging 部署脚本（SSH 模式，并行优化版）
+# ===========================================
 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)"
+  export REGISTRY_TOKEN REGISTRY_HOST REGISTRY_USER
+  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:-}"
-ACR_REGISTRY="${ACR_REGISTRY:-xiaoxia-registry.cn-hangzhou.cr.aliyuncs.com/xiaoxiakeji}"
-ACR_USERNAME="${ACR_USERNAME:-}"
-ACR_PASSWORD="${ACR_PASSWORD:-}"
+REGISTRY="${REGISTRY:-xiaoxia-registry.cn-hangzhou.cr.aliyuncs.com/xiaoxiakeji}"
+REGISTRY_USER="${ACR_USERNAME:-${REGISTRY_USER:-nick0415343655}}"
+REGISTRY_TOKEN="${ACR_PASSWORD:-${REGISTRY_TOKEN:-}}"
 
 ENV_FILE="${ENV_FILE:-/var/lib/xiaoxia-saas-staging/.env}"
 GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-staging/generated}"
-STAGING_NETWORK="${STAGING_NETWORK:-xiaoxia-net-staging}"
+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"
   exit 1
 fi
 
-echo "============================================"
-echo "  Staging 部署 - CI触发"
-echo "  Version: $IMAGE_TAG"
-echo "============================================"
-echo ""
+test -f "$ENV_FILE"
+mkdir -p "$GENERATED_DIR"
+mkdir -p "$LEGACY_ASSETS_DIR"
 
-# ---- 登录 ACR ----
-if [ -n "$ACR_PASSWORD" ] && [ -n "$ACR_USERNAME" ]; then
-  echo "登录 ACR..."
-  printf '%s' "$ACR_PASSWORD" | docker login "$(echo "$ACR_REGISTRY" | cut -d/ -f1)" -u "$ACR_USERNAME" --password-stdin 2>/dev/null || {
-    echo "WARN: docker login failed, will try to pull anyway"
-  }
+echo "==========================================="
+echo "  Staging 部署 - $IMAGE_TAG (并行优化版)"
+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"
+
+  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" &
+  fi
+
+  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" &
+  fi
+
+  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" &
+  fi
+
+  wait
+
+  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
+}
+
+if [ -n "$REGISTRY_TOKEN" ]; then
+  echo "=========================================="
+  echo "  Login to Registry (with retries)"
+  echo "=========================================="
+  REGISTRY_HOST=$(echo "$REGISTRY" | cut -d/ -f1)
+  retry_docker_login
 fi
 
-# ---- Pull 3个镜像 ----
-API_IMAGE="${ACR_REGISTRY}/xiaoxia-saas-api:${IMAGE_TAG}"
-WORKER_IMAGE="${ACR_REGISTRY}/xiaoxia-saas-worker:${IMAGE_TAG}"
-WEB_IMAGE="${ACR_REGISTRY}/xiaoxia-saas-web:${IMAGE_TAG}"
+# ---- 并行 Pull 三个镜像 ----
+REGISTRY_API="${REGISTRY}/xiaoxia-saas-api:${IMAGE_TAG}"
+REGISTRY_WORKER="${REGISTRY}/xiaoxia-saas-worker:${IMAGE_TAG}"
+REGISTRY_WEB="${REGISTRY}/xiaoxia-saas-web:${IMAGE_TAG}"
+
+echo "=========================================="
+echo "  Pull images (parallel, up to 3 retries each)"
+echo "=========================================="
+PULL_LOG_DIR="/tmp/staging-pull-$$"
+mkdir -p "$PULL_LOG_DIR"
+
+retry_docker_pull "$REGISTRY_API" > "$PULL_LOG_DIR/api.log" 2>&1 &
+PID_API=$!
+retry_docker_pull "$REGISTRY_WORKER" > "$PULL_LOG_DIR/worker.log" 2>&1 &
+PID_WORKER=$!
+retry_docker_pull "$REGISTRY_WEB" > "$PULL_LOG_DIR/web.log" 2>&1 &
+PID_WEB=$!
+
+wait $PID_API $PID_WORKER $PID_WEB
 
 echo ""
-echo "拉取 API 镜像..."
-docker pull "$API_IMAGE"
-echo "拉取 Worker 镜像..."
-docker pull "$WORKER_IMAGE"
-echo "拉取 Web 镜像..."
-docker pull "$WEB_IMAGE"
-echo ""
-echo "所有镜像拉取完成"
+echo "Pull 结果:"
+PULL_FAILED=0
+for svc in api worker web; do
+  if tail -1 "$PULL_LOG_DIR/$svc.log" 2>/dev/null | grep -qE "Status:|Downloaded|already exists|is up to date"; then
+    echo "  OK $svc"
+  elif grep -qE "Digest:|Status: Downloaded" "$PULL_LOG_DIR/$svc.log" 2>/dev/null; then
+    echo "  OK $svc"
+  else
+    # 检查docker pull返回值不直接，用镜像是否存在来判断
+    img_var="REGISTRY_$(echo $svc | tr '[:lower:]' '[:upper:]')"
+    img_val=$(eval echo "\$$img_var")
+    if docker image inspect "$img_val" >/dev/null 2>&1; then
+      echo "  OK $svc"
+    else
+      echo "  FAIL $svc"
+      tail -5 "$PULL_LOG_DIR/$svc.log" 2>/dev/null || true
+      PULL_FAILED=$((PULL_FAILED + 1))
+    fi
+  fi
+done
 
-# ---- 确保基础设施在运行 ----
-echo ""
-echo "检查基础设施容器..."
+rm -rf "$PULL_LOG_DIR"
+
+if [ "$PULL_FAILED" -gt 0 ]; then
+  echo ""
+  echo "ERROR: $PULL_FAILED 个镜像 pull 失败"
+  exit 1
+fi
+
+echo "All images pulled."
+
+echo "Backing up legacy assets from current web container..."
+if docker inspect xiaoxia-web-staging >/dev/null 2>&1; then
+  _tmpdir="/tmp/legacy-assets-$$"
+  rm -rf "$_tmpdir"
+  mkdir -p "$_tmpdir"
+  docker cp xiaoxia-web-staging:/usr/share/nginx/html/assets/. "$_tmpdir/" 2>/dev/null || true
+  if [ -d "$_tmpdir" ] && [ "$(ls -A "$_tmpdir" 2>/dev/null)" ]; then
+    cp -an "$_tmpdir"/. "$LEGACY_ASSETS_DIR"/ 2>/dev/null || true
+    echo "Legacy assets backed up: $(ls "$_tmpdir" | wc -l) files"
+  fi
+  rm -rf "$_tmpdir"
+else
+  echo "No existing web container, skipping legacy assets backup"
+fi
+
+if [ -d "$LEGACY_ASSETS_DIR" ]; then
+  find "$LEGACY_ASSETS_DIR" -type f -mtime +7 -delete 2>/dev/null || true
+  echo "Legacy assets cleanup done (retain 7 days)"
+fi
+
+echo "Checking infrastructure containers..."
 for c in xiaoxia-postgres-staging xiaoxia-redis-staging; do
   if ! docker inspect "$c" >/dev/null 2>&1; then
-    echo "ERROR: 基础设施容器不存在: $c"
+    echo "ERROR: Required container not found: $c"
     exit 1
   fi
   state=$(docker inspect -f '{{.State.Status}}' "$c")
   if [ "$state" != "running" ]; then
-    echo "ERROR: 基础设施容器未运行: $c ($state)"
+    echo "ERROR: Container not running: $c ($state)"
     exit 1
   fi
 done
-echo "基础设施正常"
 
-# ---- 确保 staging 网络存在 ----
-docker network create "$STAGING_NETWORK" 2>/dev/null || true
+docker network create xiaoxia-net-staging 2>/dev/null || true
 
-mkdir -p "$GENERATED_DIR"
+if [ "$SKIP_MIGRATION" != "true" ]; then
+  echo "Running database migrations..."
+  docker run --rm \
+    --env-file "$ENV_FILE" \
+    --network xiaoxia-net-staging \
+    -e APP_ENV=staging \
+    "$REGISTRY_API" sh -c "cd /app && alembic upgrade head" || {
+    echo "ERROR: Database migration failed"
+    exit 1
+  }
+  echo "Migrations completed."
+else
+  echo "Skipping migrations (SKIP_MIGRATION=true)"
+fi
 
-# ---- 数据库 Migration ----
-echo ""
-echo "执行数据库迁移..."
-docker run --rm --env-file "$ENV_FILE" --network "$STAGING_NETWORK" "$API_IMAGE" sh -c "cd /app && /opt/venv/bin/alembic upgrade head"
-echo "数据库迁移完成"
-
-# ---- 停止旧容器 ----
-echo ""
-echo "停止旧容器..."
+echo "Stopping old containers..."
 docker rm -f xiaoxia-api-staging 2>/dev/null || true
 docker rm -f xiaoxia-worker-staging 2>/dev/null || true
 docker rm -f xiaoxia-web-staging 2>/dev/null || true
 
-# ---- 启动 API ----
-echo "启动 API 容器..."
+LOG_OPTS="--log-driver json-file --log-opt max-size=50m --log-opt max-file=3"
+
+# ---- 并行启动三个容器 ----
+echo "Starting all containers (parallel)..."
+
+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-api-staging \
   --env-file "$ENV_FILE" \
-  --network "$STAGING_NETWORK" \
+  --network xiaoxia-net-staging \
   -p 127.0.0.1:8000:8000 \
   -e APP_ENV=staging \
   -e APP_VERSION="$IMAGE_TAG" \
   -e GENERATED_FILES_DIR=/app/generated \
   -e GENERATED_FILES_URL_PREFIX=/generated-files \
+  -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
   -v "$GENERATED_DIR:/app/generated" \
   --restart unless-stopped \
   --health-cmd "python -c \"import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=5)\"" \
@@ -101,20 +342,21 @@ docker run -d \
   --health-timeout 10s \
   --health-retries 3 \
   --health-start-period 40s \
-  "$API_IMAGE"
+  $LOG_OPTS \
+  "$REGISTRY_API" &
+PID_API_START=$!
 
-# ---- 启动 Worker ----
-echo "启动 Worker 容器..."
 docker run -d \
   --name xiaoxia-worker-staging \
   --env-file "$ENV_FILE" \
-  --network "$STAGING_NETWORK" \
+  --network xiaoxia-net-staging \
   -e APP_ENV=staging \
   -e APP_VERSION="$IMAGE_TAG" \
   -e WORKER_CONCURRENCY=1 \
   -e WORKER_MAX_TASKS_PER_CHILD=100 \
   -e GENERATED_FILES_DIR=/app/generated \
   -e GENERATED_FILES_URL_PREFIX=/generated-files \
+  -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
   -v "$GENERATED_DIR:/app/generated" \
   --restart unless-stopped \
   --health-cmd "sh -c \"grep -q celery /proc/1/cmdline || exit 1\"" \
@@ -122,52 +364,127 @@ docker run -d \
   --health-timeout 10s \
   --health-retries 3 \
   --health-start-period 30s \
-  "$WORKER_IMAGE"
+  $LOG_OPTS \
+  "$REGISTRY_WORKER" &
+PID_WORKER_START=$!
 
-# ---- 启动 Web ----
-echo "启动 Web 容器..."
 docker run -d \
   --name xiaoxia-web-staging \
-  --network "$STAGING_NETWORK" \
+  --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 \
-  "$WEB_IMAGE"
+  $LOG_OPTS \
+  "$REGISTRY_WEB" &
+PID_WEB_START=$!
 
-# ---- 等待 API 健康 ----
-echo ""
-echo "等待 API 健康检查..."
-i=0
-while [ "$i" -lt 30 ]; do
-  if curl -sf --max-time 5 http://127.0.0.1:8000/health >/dev/null 2>&1; then
-    echo "API 健康检查通过"
-    break
+wait $PID_API_START $PID_WORKER_START $PID_WEB_START
+
+START_FAILED=0
+for c in xiaoxia-api-staging xiaoxia-worker-staging xiaoxia-web-staging; do
+  if ! docker inspect "$c" >/dev/null 2>&1; then
+    echo "  FAIL $c: not created"
+    START_FAILED=$((START_FAILED + 1))
+  else
+    state=$(docker inspect -f '{{.State.Status}}' "$c")
+    if [ "$state" = "running" ] || [ "$state" = "starting" ]; then
+      echo "  OK $c: $state"
+    else
+      echo "  FAIL $c: $state"
+      docker logs --tail 20 "$c" 2>/dev/null || true
+      START_FAILED=$((START_FAILED + 1))
+    fi
   fi
-  i=$((i + 1))
-  echo "  等待中... ($i/30)"
-  sleep 2
 done
 
-if [ "$i" -ge 30 ]; then
-  echo "ERROR: API 在 60s 内未通过健康检查"
-  echo ""
-  echo "=== API 最近日志 ==="
-  docker logs --tail 50 xiaoxia-api-staging
-  exit 1
+if [ "$START_FAILED" -gt 0 ]; then
+  echo "ERROR: $START_FAILED 个容器启动失败"
+  rollback
 fi
 
-# ---- 清理旧镜像 ----
-docker image prune -af --filter "until=72h" 2>/dev/null || true
+# ---- 并行等待 API 和 Web 健康 ----
+echo ""
+echo "Waiting for API + Web health (parallel)..."
+
+HEALTH_LOG_DIR="/tmp/staging-health-$$"
+mkdir -p "$HEALTH_LOG_DIR"
+
+(
+  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 "API healthy after $((i * 3))s"
+      exit 0
+    fi
+    i=$((i + 1))
+    sleep 3
+  done
+  echo "API FAILED after 120s"
+  exit 1
+) > "$HEALTH_LOG_DIR/api.log" 2>&1 &
+PID_API_HEALTH=$!
+
+(
+  i=0
+  while [ "$i" -lt 15 ]; do
+    if curl -sf --max-time 5 http://127.0.0.1:3001/ >/dev/null 2>&1; then
+      echo "Web healthy after $((i * 2))s"
+      exit 0
+    fi
+    i=$((i + 1))
+    sleep 2
+  done
+  echo "Web FAILED after 30s"
+  exit 1
+) > "$HEALTH_LOG_DIR/web.log" 2>&1 &
+PID_WEB_HEALTH=$!
+
+set +e
+wait $PID_API_HEALTH
+API_EXIT=$?
+wait $PID_WEB_HEALTH
+WEB_EXIT=$?
+set -e
 
 echo ""
-echo "============================================"
-echo "  ✅ Staging 部署完成"
-echo "============================================"
+echo "健康检查结果:"
+API_OK=0
+WEB_OK=0
+if [ "$API_EXIT" -eq 0 ]; then
+  echo "  OK API: $(cat "$HEALTH_LOG_DIR/api.log")"
+  API_OK=1
+else
+  echo "  FAIL API: 120s未就绪"
+  docker logs --tail 50 xiaoxia-api-staging
+fi
+
+if [ "$WEB_EXIT" -eq 0 ]; then
+  echo "  OK Web: $(cat "$HEALTH_LOG_DIR/web.log")"
+  WEB_OK=1
+else
+  echo "  FAIL Web: 30s未就绪"
+  docker logs --tail 30 xiaoxia-web-staging
+fi
+
+rm -rf "$HEALTH_LOG_DIR"
+
+if [ "$API_OK" -eq 0 ] || [ "$WEB_OK" -eq 0 ]; then
+  echo ""
+  echo "ERROR: 健康检查失败"
+  rollback
+fi
+
+echo "Cleaning up old images..."
+docker image prune -af --filter "until=168h" 2>/dev/null || true
+docker builder prune -af --filter "until=168h" 2>/dev/null || true
+
+echo ""
+echo "=== Staging deployment complete (并行优化版) ==="
 echo "API:      http://127.0.0.1:8000"
 echo "Web:      http://127.0.0.1:3001"
 echo "Version:  $IMAGE_TAG"
-echo ""
-docker ps --format "table {{.Names}}\t{{.Status}}" | grep staging
+docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Image}}" | grep staging

scripts/ci_staging_healthcheck.sh

@@ -1,110 +1,473 @@
 #!/bin/bash
-# ============================================
-# CI Staging 健康检查脚本 - 在 CI Runner 上执行
-# 通过 SSH 连接 Staging 服务器，检查 API/Worker/Web 健康状态
-# ============================================
+# ===========================================
+# CI Staging 健康检查 + 自动回滚脚本（SSH 部署模式）
+# ===========================================
+#
+# 在 CI Runner 上执行，通过公网 URL 检查 Staging 部署健康状态。
+# 不健康则通过 SSH 自动回滚到上一个版本的镜像。
+#
+# 用法:
+#   ./ci_staging_healthcheck.sh
+#
+# 环境变量:
+#   STAGING_API_URL       - Staging API 地址 (默认 https://staging-api.xiaoxiajianji.com)
+#   STAGING_WEB_URL       - Staging Web 地址 (默认 https://staging.xiaoxiajianji.com)
+#   HEALTH_CHECK_TIMEOUT  - 健康检查总超时秒数 (默认 120)
+#   SKIP_ROLLBACK         - 失败时不自动回滚 (true/false, 默认 false)
+#   SKIP_NOTIFY           - 跳过通知 (true/false, 默认 false)
+#   CI_NOTIFY_WEBHOOK     - 通知 Webhook URL
+#
+#   STAGING_SSH_HOST      - Staging 服务器 SSH 地址 (默认 47.98.113.167)
+#   STAGING_SSH_USER      - SSH 用户名 (默认 root)
+#   STAGING_SSH_PORT      - SSH 端口 (默认 22222)
+#   STAGING_SSH_KEY       - SSH 私钥内容
+#   REGISTRY_TOKEN        - Registry Token（回滚时拉取旧镜像需要）
+#
+#   GITHUB_SHA            - 当前 commit SHA
+#   GITHUB_REF_NAME       - 分支名
+#   GITHUB_RUN_ID         - CI Run ID
+#   GITHUB_REPOSITORY     - 仓库名
+#   GITHUB_ACTOR          - 提交者
 
 set -eu
 
+SCRIPT_DIR="$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)"
+
+# 配置
+STAGING_API_URL="${STAGING_API_URL:-https://staging-api.xiaoxiajianji.com}"
+STAGING_WEB_URL="${STAGING_WEB_URL:-https://staging.xiaoxiajianji.com}"
+HEALTH_CHECK_TIMEOUT="${HEALTH_CHECK_TIMEOUT:-120}"
+SKIP_ROLLBACK="${SKIP_ROLLBACK:-false}"
+SKIP_NOTIFY="${SKIP_NOTIFY:-false}"
+
 STAGING_SSH_HOST="${STAGING_SSH_HOST:-47.98.113.167}"
 STAGING_SSH_USER="${STAGING_SSH_USER:-root}"
 STAGING_SSH_PORT="${STAGING_SSH_PORT:-22222}"
-STAGING_SSH_KEY="${STAGING_SSH_KEY:-}"
 
-HEALTH_TIMEOUT="${HEALTH_TIMEOUT:-120}"
-HEALTH_INTERVAL="${HEALTH_INTERVAL:-5}"
+REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
+REGISTRY_USER="${REGISTRY_USER:-xiaoxia}"
 
-echo "============================================"
-echo "  Staging 健康检查"
-echo "  Host: $STAGING_SSH_HOST:$STAGING_SSH_PORT"
-echo "============================================"
-echo ""
+# 颜色
+RED='\033[0;31m'
+GREEN='\033[0;32m'
+YELLOW='\033[1;33m'
+BLUE='\033[0;34m'
+NC='\033[0m'
 
-# ---- 准备 SSH key ----
-mkdir -p ~/.ssh
-key_path=""
-if [ -f /root/.ssh/xiaoxia_runtime_builder ]; then
-  key_path="/root/.ssh/xiaoxia_runtime_builder"
-elif [ -f "$HOME/.ssh/xiaoxia_runtime_builder" ]; then
-  key_path="$HOME/.ssh/xiaoxia_runtime_builder"
-elif [ -n "${STAGING_SSH_KEY:-}" ]; then
-  key_path="$HOME/.ssh/staging_deploy_key"
-  printf '%s\n' "$STAGING_SSH_KEY" > "$key_path"
-  chmod 600 "$key_path"
-else
-  echo "ERROR: 没有可用的 SSH key"
-  exit 1
+log_info()  { echo -e "${GREEN}[INFO]${NC} $1"; }
+log_warn()  { echo -e "${YELLOW}[WARN]${NC} $1"; }
+log_error() { echo -e "${RED}[ERROR]${NC} $1"; }
+log_step()  { echo -e "${BLUE}[STEP]${NC} $1"; }
+
+# ===========================================
+# SSH 工具函数
+# ===========================================
+SSH_KEY_PATH=""
+
+setup_ssh() {
+    # 查找或创建 SSH 密钥
+    if [ -f /root/.ssh/xiaoxia_runtime_builder ]; then
+        SSH_KEY_PATH="/root/.ssh/xiaoxia_runtime_builder"
+    elif [ -f "$HOME/.ssh/xiaoxia_runtime_builder" ]; then
+        SSH_KEY_PATH="$HOME/.ssh/xiaoxia_runtime_builder"
+    elif [ -n "${STAGING_SSH_KEY:-}" ]; then
+        SSH_KEY_PATH="$HOME/.ssh/staging_deploy_key"
+        mkdir -p "$HOME/.ssh"
+        printf '%s\n' "$STAGING_SSH_KEY" > "$SSH_KEY_PATH"
+        chmod 600 "$SSH_KEY_PATH"
+    else
+        log_error "没有可用的 SSH 密钥"
+        return 1
+    fi
+
+    ssh-keyscan -p "$STAGING_SSH_PORT" -H "$STAGING_SSH_HOST" >> ~/.ssh/known_hosts 2>/dev/null || true
+    log_info "SSH 已配置: ${STAGING_SSH_USER}@${STAGING_SSH_HOST}:${STAGING_SSH_PORT}"
+}
+
+run_ssh() {
+    local cmd="$1"
+    ssh -p "$STAGING_SSH_PORT" -i "$SSH_KEY_PATH" -o StrictHostKeyChecking=no \
+        "${STAGING_SSH_USER}@${STAGING_SSH_HOST}" "$cmd"
+}
+
+# ===========================================
+# 1. 记录部署前各服务的镜像版本（用于回滚）
+# ===========================================
+ROLLBACK_API_TAG=""
+ROLLBACK_WORKER_TAG=""
+ROLLBACK_WEB_TAG=""
+
+save_rollback_target() {
+    log_step "记录当前 staging 各服务镜像版本（回滚目标）..."
+
+    # 通过 SSH 获取当前运行的容器镜像
+    local api_image worker_image web_image
+    api_image=$(run_ssh "docker inspect --format '{{.Config.Image}}' xiaoxia-api-staging 2>/dev/null || echo ''")
+    worker_image=$(run_ssh "docker inspect --format '{{.Config.Image}}' xiaoxia-worker-staging 2>/dev/null || echo ''")
+    web_image=$(run_ssh "docker inspect --format '{{.Config.Image}}' xiaoxia-web-staging 2>/dev/null || echo ''")
+
+    # 提取 tag（镜像名是 xiaoxia-saas-api:abc123 或 git.xiaoxiajianji.com/.../xiaoxia-saas-api:staging 格式）
+    ROLLBACK_API_TAG=$(echo "$api_image" | sed 's/.*://' || echo "")
+    ROLLBACK_WORKER_TAG=$(echo "$worker_image" | sed 's/.*://' || echo "")
+    ROLLBACK_WEB_TAG=$(echo "$web_image" | sed 's/.*://' || echo "")
+
+    log_info "  API:     ${ROLLBACK_API_TAG:-未知}"
+    log_info "  Worker:  ${ROLLBACK_WORKER_TAG:-未知}"
+    log_info "  Web:     ${ROLLBACK_WEB_TAG:-未知}"
+
+    # 验证三个服务版本是否一致
+    if [ -n "$ROLLBACK_API_TAG" ] && [ -n "$ROLLBACK_WORKER_TAG" ] && [ -n "$ROLLBACK_WEB_TAG" ]; then
+        if [ "$ROLLBACK_API_TAG" = "$ROLLBACK_WORKER_TAG" ] && [ "$ROLLBACK_API_TAG" = "$ROLLBACK_WEB_TAG" ]; then
+            log_info "  ✅ 三个服务版本一致: $ROLLBACK_API_TAG"
+            export ROLLBACK_TAG="$ROLLBACK_API_TAG"
+        else
+            log_warn "  ⚠️  三个服务版本不一致，回滚时将分别使用各自版本"
+            export ROLLBACK_API_TAG ROLLBACK_WORKER_TAG ROLLBACK_WEB_TAG
+            export ROLLBACK_TAG_MIXED="true"
+        fi
+    else
+        log_warn "  ⚠️  未能获取全部服务版本，回滚功能可能受限"
+    fi
+}
+
+# ===========================================
+# 2. 健康检查（公网视角）
+# ===========================================
+health_check() {
+    local timeout="$HEALTH_CHECK_TIMEOUT"
+    local start_time
+    start_time=$(date +%s)
+
+    log_step "公网健康检查（超时 ${timeout}s）..."
+    log_info "  API: ${STAGING_API_URL}/health"
+    log_info "  Web: ${STAGING_WEB_URL}/"
+
+    local api_ok=false
+    local web_ok=false
+    local api_docs_ok=false
+    local login_api_ok=false
+
+    while [ $(( $(date +%s) - start_time )) -lt "$timeout" ]; do
+        # 检查 API health
+        if [ "$api_ok" = false ] && curl -sf --max-time 10 "${STAGING_API_URL}/health" >/dev/null 2>&1; then
+            log_info "✅ API 健康检查通过"
+            api_ok=true
+        fi
+
+        # 检查 Web 首页
+        if [ "$web_ok" = false ] && curl -sf --max-time 10 "$STAGING_WEB_URL/" >/dev/null 2>&1; then
+            log_info "✅ Web 前端检查通过"
+            web_ok=true
+        fi
+
+        # 检查 API docs（服务完全启动的标志）
+        if [ "$api_docs_ok" = false ]; then
+            HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" --max-time 10 "${STAGING_API_URL}/docs" 2>/dev/null || echo "000")
+            if [ "$HTTP_CODE" = "200" ]; then
+                log_info "✅ API Docs 检查通过"
+                api_docs_ok=true
+            fi
+        fi
+
+        # 检查登录 API（业务逻辑正常的标志）
+        if [ "$login_api_ok" = false ]; then
+            HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" --max-time 10 -X POST \
+                "${STAGING_API_URL}/api/v1/auth/login" \
+                -H "Content-Type: application/json" \
+                -d '{"email":"smoke@test.com","password":"wrong"}' 2>/dev/null || echo "000")
+            if [ "$HTTP_CODE" = "401" ] || [ "$HTTP_CODE" = "422" ]; then
+                log_info "✅ 登录 API 检查通过（HTTP $HTTP_CODE，符合预期）"
+                login_api_ok=true
+            fi
+        fi
+
+        # 都通过了就退出
+        if [ "$api_ok" = true ] && [ "$web_ok" = true ] && [ "$api_docs_ok" = true ] && [ "$login_api_ok" = true ]; then
+            log_info "🎉 所有健康检查通过！"
+            return 0
+        fi
+
+        sleep 5
+    done
+
+    # 超时了
+    log_error "❌ 健康检查超时 (${timeout}s)"
+    [ "$api_ok" = false ]       && log_error "  - API health 未通过"
+    [ "$web_ok" = false ]       && log_error "  - Web 前端未通过"
+    [ "$api_docs_ok" = false ]  && log_error "  - API Docs 未通过"
+    [ "$login_api_ok" = false ] && log_error "  - 登录 API 未通过"
+
+    return 1
+}
+
+# ===========================================
+# 3. 执行回滚（SSH 重新部署旧版本）
+# ===========================================
+do_rollback() {
+    log_step "执行回滚：通过 SSH 重新部署旧版本镜像..."
+
+    local rollback_tag="${ROLLBACK_TAG:-}"
+    if [ -z "$rollback_tag" ] && [ "${ROLLBACK_TAG_MIXED:-}" != "true" ]; then
+        log_error "没有可回滚的版本记录，无法自动回滚"
+        return 1
+    fi
+
+    # 如果版本不一致，用 API 的版本作为回滚目标
+    if [ -z "$rollback_tag" ]; then
+        rollback_tag="$ROLLBACK_API_TAG"
+    fi
+
+    if [ -z "$rollback_tag" ]; then
+        log_error "无法确定回滚版本"
+        return 1
+    fi
+
+    log_info "回滚目标版本: $rollback_tag"
+
+    # 构建回滚脚本（直接部署旧版本镜像，不跑 migration）
+    local rollback_script=$(cat << 'ROLLBACK_EOF'
+#!/bin/sh
+set -eu
+
+IMAGE_TAG="$1"
+REGISTRY_TOKEN="$2"
+REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
+REGISTRY_USER="${REGISTRY_USER:-xiaoxia}"
+
+ENV_FILE="${ENV_FILE:-/var/lib/xiaoxia-saas-staging/.env}"
+GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-staging/generated}"
+LEGACY_ASSETS_DIR="${LEGACY_ASSETS_DIR:-/var/lib/xiaoxia-saas-staging/legacy-assets}"
+
+echo "=== Rollback to $IMAGE_TAG ==="
+
+# 登录 Registry
+if [ -n "$REGISTRY_TOKEN" ]; then
+    REGISTRY_HOST=$(echo "$REGISTRY" | cut -d/ -f1)
+    printf %s "$REGISTRY_TOKEN" | docker login "$REGISTRY_HOST" -u "$REGISTRY_USER" --password-stdin 2>/dev/null || true
 fi
 
-ssh-keyscan -p "$STAGING_SSH_PORT" -H "$STAGING_SSH_HOST" >> ~/.ssh/known_hosts 2>/dev/null || true
+# Pull 旧版本镜像
+LOCAL_API="xiaoxia-saas-api:${IMAGE_TAG}"
+LOCAL_WORKER="xiaoxia-saas-worker:${IMAGE_TAG}"
+LOCAL_WEB="xiaoxia-saas-web:${IMAGE_TAG}"
 
-SSH_CMD="ssh -p $STAGING_SSH_PORT -i $key_path -o StrictHostKeyChecking=no ${STAGING_SSH_USER}@${STAGING_SSH_HOST}"
+docker pull "${REGISTRY}/xiaoxia-saas-api:${IMAGE_TAG}"
+docker pull "${REGISTRY}/xiaoxia-saas-worker:${IMAGE_TAG}"
+docker pull "${REGISTRY}/xiaoxia-saas-web:${IMAGE_TAG}"
 
-# ---- 检查函数 ----
-check_api() {
-  $SSH_CMD "curl -sf --max-time 5 http://127.0.0.1:8000/health" 2>/dev/null >/dev/null
-}
+docker tag "${REGISTRY}/xiaoxia-saas-api:${IMAGE_TAG}" "$LOCAL_API"
+docker tag "${REGISTRY}/xiaoxia-saas-worker:${IMAGE_TAG}" "$LOCAL_WORKER"
+docker tag "${REGISTRY}/xiaoxia-saas-web:${IMAGE_TAG}" "$LOCAL_WEB"
 
-check_web() {
-  $SSH_CMD "curl -sf --max-time 5 -o /dev/null -w '%{http_code}' http://127.0.0.1:3001/" 2>/dev/null | grep -q "200"
-}
+echo "Rollback images pulled."
 
-check_worker() {
-  $SSH_CMD "docker inspect --format='{{.State.Status}}' xiaoxia-worker-staging 2>/dev/null | grep -q running" 2>/dev/null
-}
+# 停止当前容器（回滚不跑 migration，避免数据问题）
+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
 
-# ---- 循环等待健康 ----
-echo "等待 Staging 环境健康 (超时: ${HEALTH_TIMEOUT}s)..."
-elapsed=0
-api_ok=0
-web_ok=0
-worker_ok=0
+LOG_OPTS="--log-driver json-file --log-opt max-size=50m --log-opt max-file=3"
 
-while [ "$elapsed" -lt "$HEALTH_TIMEOUT" ]; do
-  # 检查 API
-  if [ "$api_ok" -eq 0 ] && check_api; then
-    echo "  ✅ API 健康"
-    api_ok=1
+# 启动 API（回滚不跑 migration）
+echo "Starting API (rollback)..."
+docker run -d \
+  --name xiaoxia-api-staging \
+  --env-file "$ENV_FILE" \
+  --network xiaoxia-net-staging \
+  -p 127.0.0.1:8000:8000 \
+  -e APP_ENV=staging \
+  -e APP_VERSION="$IMAGE_TAG" \
+  -e GENERATED_FILES_DIR=/app/generated \
+  -e GENERATED_FILES_URL_PREFIX=/generated-files \
+  -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
+  -v "$GENERATED_DIR:/app/generated" \
+  --restart unless-stopped \
+  --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 \
+  "$LOCAL_API"
+
+# 启动 Worker
+echo "Starting Worker (rollback)..."
+docker run -d \
+  --name xiaoxia-worker-staging \
+  --env-file "$ENV_FILE" \
+  --network xiaoxia-net-staging \
+  -e APP_ENV=staging \
+  -e APP_VERSION="$IMAGE_TAG" \
+  -e WORKER_CONCURRENCY=1 \
+  -e WORKER_MAX_TASKS_PER_CHILD=100 \
+  -e GENERATED_FILES_DIR=/app/generated \
+  -e GENERATED_FILES_URL_PREFIX=/generated-files \
+  -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
+  -v "$GENERATED_DIR:/app/generated" \
+  --restart unless-stopped \
+  --health-cmd "sh -c \"grep -q celery /proc/1/cmdline || exit 1\"" \
+  --health-interval 30s \
+  --health-timeout 10s \
+  --health-retries 3 \
+  --health-start-period 30s \
+  $LOG_OPTS \
+  "$LOCAL_WORKER"
+
+# 启动 Web
+LEGACY_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
+
+echo "Starting Web (rollback)..."
+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 \
+  "$LOCAL_WEB"
+
+# 等待 API 健康
+echo "Waiting for API (rollback)..."
+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 "API healthy (rollback)."
+    break
   fi
-
-  # 检查 Web
-  if [ "$web_ok" -eq 0 ] && check_web; then
-    echo "  ✅ Web 健康"
-    web_ok=1
-  fi
-
-  # 检查 Worker
-  if [ "$worker_ok" -eq 0 ] && check_worker; then
-    echo "  ✅ Worker 运行中"
-    worker_ok=1
-  fi
-
-  # 全部通过
-  if [ "$api_ok" -eq 1 ] && [ "$web_ok" -eq 1 ] && [ "$worker_ok" -eq 1 ]; then
-    echo ""
-    echo "============================================"
-    echo "  ✅ 全部健康检查通过"
-    echo "============================================"
-    exit 0
-  fi
-
-  sleep "$HEALTH_INTERVAL"
-  elapsed=$((elapsed + HEALTH_INTERVAL))
-  echo "  等待中... (${elapsed}s/${HEALTH_TIMEOUT}s)"
+  i=$((i + 1))
+  sleep 3
 done
 
-echo ""
-echo "============================================"
-echo "  ❌ 健康检查超时"
-echo "============================================"
-echo ""
-echo "未通过的检查项:"
-[ "$api_ok" -eq 0 ] && echo "  - API (http://127.0.0.1:8000/health)"
-[ "$web_ok" -eq 0 ] && echo "  - Web (http://127.0.0.1:3001/)"
-[ "$worker_ok" -eq 0 ] && echo "  - Worker 容器"
+# 等待 Web 健康
+echo "Waiting for Web (rollback)..."
+i=0
+while [ "$i" -lt 15 ]; do
+  if curl -sf --max-time 5 http://127.0.0.1:3001/ >/dev/null 2>&1; then
+    echo "Web healthy (rollback)."
+    break
+  fi
+  i=$((i + 1))
+  sleep 2
+done
 
-echo ""
-echo "=== API 最近日志 ==="
-$SSH_CMD "docker logs --tail 30 xiaoxia-api-staging" 2>/dev/null || echo "无法获取日志"
+echo "=== Rollback complete: $IMAGE_TAG ==="
+docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Image}}" | grep staging
+ROLLBACK_EOF
+)
 
-exit 1
+    # 将脚本 base64 编码后通过 SSH 执行
+    local script_b64
+    script_b64=$(echo "$rollback_script" | base64 -w 0)
+
+    log_info "在 staging 服务器上执行回滚脚本..."
+    if run_ssh "echo '$script_b64' | base64 -d | sh -s -- '$rollback_tag' '${REGISTRY_TOKEN:-}'" 2>&1; then
+        log_info "✅ 回滚命令执行完成"
+        return 0
+    else
+        log_error "❌ 回滚命令执行失败"
+        return 1
+    fi
+}
+
+# ===========================================
+# 4. 发送通知
+# ===========================================
+send_notification() {
+    local status="$1"   # success / failure / rollback
+    local detail="$2"
+
+    if [ "${SKIP_NOTIFY:-false}" = "true" ]; then
+        log_info "跳过通知（SKIP_NOTIFY=true）"
+        return 0
+    fi
+
+    local webhook="${CI_NOTIFY_WEBHOOK:-}"
+    if [ -z "$webhook" ]; then
+        log_warn "未配置 CI_NOTIFY_WEBHOOK，跳过通知"
+        return 0
+    fi
+
+    if [ -f "$SCRIPT_DIR/deploy_notify.py" ]; then
+        python3 "$SCRIPT_DIR/deploy_notify.py" \
+            --status "$status" \
+            --detail "$detail" \
+            --webhook "$webhook" \
+            --env staging \
+            2>/dev/null || log_warn "通知发送失败（非致命）"
+    else
+        log_warn "找不到 deploy_notify.py，跳过通知"
+    fi
+}
+
+# ===========================================
+# 主流程
+# ===========================================
+main() {
+    echo ""
+    echo "==========================================="
+    echo "  CI Staging 健康检查 + 自动回滚（SSH模式）"
+    echo "==========================================="
+    echo ""
+
+    local deploy_status="success"
+    local deploy_detail=""
+
+    # 1. 设置 SSH
+    if ! setup_ssh; then
+        log_error "SSH 配置失败，无法执行回滚"
+    fi
+
+    # 2. 记录部署前状态（回滚目标）
+    save_rollback_target || true
+
+    # 3. 健康检查（公网视角）
+    if ! health_check; then
+        log_error "健康检查失败"
+        deploy_status="failure"
+        deploy_detail="公网健康检查超时，部署后服务未正常响应"
+
+        # 自动回滚
+        if [ "${SKIP_ROLLBACK:-false}" != "true" ]; then
+            log_warn "开始自动回滚..."
+            if do_rollback; then
+                deploy_status="rollback"
+                deploy_detail="健康检查失败，已自动回滚到上一版本 (${ROLLBACK_TAG:-未知})"
+
+                # 回滚后再检查一下公网状态
+                log_info "回滚完成，重新检查公网健康状态..."
+                if health_check; then
+                    log_info "✅ 回滚后服务已恢复"
+                    deploy_detail="${deploy_detail}，回滚后服务已恢复"
+                else
+                    log_error "⚠️  回滚后健康检查仍未通过，请手动排查"
+                    deploy_detail="${deploy_detail}，但回滚后仍未恢复，请紧急排查"
+                fi
+            else
+                deploy_detail="健康检查失败且回滚失败，请手动排查"
+            fi
+        fi
+
+        send_notification "$deploy_status" "$deploy_detail"
+
+        # 失败时退出非零，让 CI Job 标记为失败
+        exit 1
+    fi
+
+    # 4. 成功
+    log_info ""
+    log_info "=================================="
+    log_info "  ✅ Staging 部署成功！"
+    log_info "=================================="
+
+    deploy_detail="部署成功，所有健康检查通过 (${GITHUB_SHA:-未知版本})"
+    send_notification "success" "$deploy_detail"
+}
+
+main "$@"

tests/unit/test_confirm_generation.py

@@ -0,0 +1,379 @@
+"""确认生成 API 单元测试.
+
+覆盖 POST /tasks/{task_id}/confirm 端点：
+  - 正常确认流程
+  - 预览任务不存在 → 404
+  - 权限不足 → 403
+  - is_preview=False 及分辨率正确
+  - cover_url 和 custom_title 正确传递
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Optional
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
+
+from packages.domain.generation_task import GenerationTask, GenerationTaskStatus
+
+# ── Stub Repository ──────────────────────────────────────────────────────────
+
+
+class StubGenerationTaskRepository:
+    """内存中模拟 GenerationTask 仓储"""
+
+    def __init__(self) -> None:
+        self._store: dict[str, Any] = {}
+
+    def create(self, task: Any) -> Any:
+        self._store[task.id] = task
+        return task
+
+    def get(self, task_id: str) -> Optional[Any]:
+        return self._store.get(task_id)
+
+    def update(self, task: Any) -> Any:
+        if task.id not in self._store:
+            raise ValueError(f"GenerationTask {task.id} not found")
+        self._store[task.id] = task
+        return task
+
+    def list_by_project(self, project_id: str) -> list[Any]:
+        return [t for t in self._store.values() if t.project_id == project_id]
+
+    def list_by_user(self, user_id: str) -> list[Any]:
+        return [t for t in self._store.values() if t.created_by_user_id == user_id]
+
+    def count_by_user(self, user_id: str) -> int:
+        return len([t for t in self._store.values() if t.created_by_user_id == user_id])
+
+    def list_recent_by_user(self, user_id: str, limit: int = 5) -> list[Any]:
+        items = [t for t in self._store.values() if t.created_by_user_id == user_id]
+        items.sort(key=lambda t: t.created_at, reverse=True)
+        return items[:limit]
+
+    def list_by_source_edit_plan(self, plan_id: str) -> list[Any]:
+        return [t for t in self._store.values() if (t.source_edit_plan_id or "") == plan_id]
+
+
+# ── Stub Project Repository ──────────────────────────────────────────────────
+
+
+@dataclass
+class FakeProject:
+    id: str = "project-001"
+    owner_user_id: str = "user-001"
+    shared_users: list[str] = field(default_factory=list)
+    name: str = "Test Project"
+
+    def can_access(self, user_id: str) -> bool:
+        return user_id == self.owner_user_id or user_id in self.shared_users
+
+
+class StubProjectRepository:
+    def __init__(self) -> None:
+        self._projects: dict[str, FakeProject] = {}
+
+    def add(self, project: FakeProject) -> None:
+        self._projects[project.id] = project
+
+    def find_by_id(self, project_id: str) -> Optional[FakeProject]:
+        return self._projects.get(project_id)
+
+
+# ── Fixtures ─────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class FakeUser:
+    id: str = "user-001"
+    email: str = "test@example.com"
+
+
+@dataclass
+class FakeAuthenticatedUser:
+    user: FakeUser = field(default_factory=FakeUser)
+    session_id: str | None = None
+    token_type: str | None = None
+
+
+@pytest.fixture
+def gen_task_repo() -> StubGenerationTaskRepository:
+    return StubGenerationTaskRepository()
+
+
+@pytest.fixture
+def project_repo() -> StubProjectRepository:
+    repo = StubProjectRepository()
+    repo.add(FakeProject())
+    return repo
+
+
+@pytest.fixture
+def app(
+    gen_task_repo: StubGenerationTaskRepository,
+    project_repo: StubProjectRepository,
+) -> FastAPI:
+    """构建测试 FastAPI 应用，注入 Stub Repository"""
+    from app.api.routes.generation_tasks import router
+    from app.auth import get_current_user
+    from app.dependencies import (
+        get_asset_library_repository,
+        get_asset_repository,
+        get_generated_video_repository,
+        get_generation_task_repository,
+        get_project_repository,
+    )
+
+    test_app = FastAPI()
+    test_app.include_router(router, prefix="/api/v1")
+
+    def override_get_current_user():
+        return FakeAuthenticatedUser()
+
+    def override_get_generation_task_repository():
+        return gen_task_repo
+
+    def override_get_project_repository():
+        return project_repo
+
+    test_app.dependency_overrides[get_current_user] = override_get_current_user
+    test_app.dependency_overrides[get_generation_task_repository] = override_get_generation_task_repository
+    test_app.dependency_overrides[get_project_repository] = override_get_project_repository
+    # Stubs for repositories not used by confirm endpoint but required by router
+    test_app.dependency_overrides[get_asset_library_repository] = lambda: MagicMock()
+    test_app.dependency_overrides[get_asset_repository] = lambda: MagicMock()
+    test_app.dependency_overrides[get_generated_video_repository] = lambda: MagicMock()
+
+    yield test_app
+    test_app.dependency_overrides.clear()
+
+
+@pytest.fixture
+def client(app: FastAPI) -> TestClient:
+    return TestClient(app)
+
+
+def _make_preview_task(**kwargs: Any) -> GenerationTask:
+    """创建预览任务"""
+    defaults = dict(
+        id="preview-task-001",
+        project_id="project-001",
+        asset_library_id="library-001",
+        strategy_id="one_take",
+        voice_library_id="",
+        template_id="",
+        asset_ids=["asset-1"],
+        title_ids=[],
+        voice_ids=[],
+        status=GenerationTaskStatus.COMPLETED,
+        progress=100.0,
+        result_count=1,
+        error_message="",
+        created_by_user_id="user-001",
+        source_edit_plan_id="",
+        asset_select_mode="all",
+        is_preview=True,
+        source_task_id="",
+        output_width=1280,
+        output_height=720,
+        cover_url="",
+        custom_title="",
+    )
+    defaults.update(kwargs)
+    return GenerationTask(**defaults)
+
+
+# ── Tests ────────────────────────────────────────────────────────────────────
+
+
+class TestConfirmGeneration:
+    def test_confirm_success(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """正常确认流程：预览任务存在、权限正确 → 创建正式任务"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            resp = client.post(
+                f"/api/v1/tasks/{preview.id}/confirm",
+                json={
+                    "output_width": 1080,
+                    "output_height": 1920,
+                    "cover_url": "https://example.com/cover.jpg",
+                    "custom_title": "我的视频",
+                },
+            )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["total"] == 1
+        item = data["items"][0]
+        assert item["is_preview"] is False
+        assert item["source_task_id"] == preview.id
+        assert item["output_width"] == 1080
+        assert item["output_height"] == 1920
+        assert item["cover_url"] == "https://example.com/cover.jpg"
+        assert item["custom_title"] == "我的视频"
+        # 复制了预览任务的配置
+        assert item["project_id"] == "project-001"
+        assert item["asset_library_id"] == "library-001"
+        assert item["strategy_id"] == "one_take"
+        assert item["asset_ids"] == ["asset-1"]
+
+        # 验证 celery 任务被调度
+        mock_celery.send_task.assert_called_once()
+        call_args = mock_celery.send_task.call_args
+        assert call_args[0][0] == "worker.generate_video"
+
+    def test_confirm_not_found(self, client: TestClient) -> None:
+        """预览任务不存在 → 404"""
+        resp = client.post(
+            "/api/v1/tasks/nonexistent-task/confirm",
+            json={"output_width": 1080, "output_height": 1920},
+        )
+        assert resp.status_code == 404
+        assert "not found" in resp.json()["detail"]
+
+    def test_confirm_access_denied(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """权限不足 → 403"""
+        preview = _make_preview_task(created_by_user_id="other-user-999")
+        gen_task_repo.create(preview)
+
+        resp = client.post(
+            f"/api/v1/tasks/{preview.id}/confirm",
+            json={"output_width": 1080, "output_height": 1920},
+        )
+        assert resp.status_code == 403
+        assert "denied" in resp.json()["detail"].lower() or "Access" in resp.json()["detail"]
+
+    def test_confirm_preserves_config(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """确认后的任务 is_preview=False，分辨率已更新，其余配置从预览任务复制"""
+        preview = _make_preview_task(
+            voice_library_id="voice-001",
+            template_id="tmpl-001",
+            title_ids=["title-1", "title-2"],
+            voice_ids=["voice-a"],
+        )
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            resp = client.post(
+                f"/api/v1/tasks/{preview.id}/confirm",
+                json={"output_width": 1920, "output_height": 1080},
+            )
+
+        assert resp.status_code == 200
+        item = resp.json()["items"][0]
+        assert item["is_preview"] is False
+        assert item["source_task_id"] == preview.id
+        assert item["output_width"] == 1920
+        assert item["output_height"] == 1080
+        # 默认分辨率
+        assert item["cover_url"] == ""
+        assert item["custom_title"] == ""
+        # 复制的配置
+        assert item["voice_library_id"] == "voice-001"
+        assert item["template_id"] == "tmpl-001"
+        assert item["title_ids"] == ["title-1", "title-2"]
+        assert item["voice_ids"] == ["voice-a"]
+
+    def test_confirm_cover_and_title(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """cover_url 和 custom_title 正确传递"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            resp = client.post(
+                f"/api/v1/tasks/{preview.id}/confirm",
+                json={
+                    "output_width": 1080,
+                    "output_height": 1920,
+                    "cover_url": "https://cdn.example.com/my-cover.png",
+                    "custom_title": "测试视频标题",
+                },
+            )
+
+        assert resp.status_code == 200
+        item = resp.json()["items"][0]
+        assert item["cover_url"] == "https://cdn.example.com/my-cover.png"
+        assert item["custom_title"] == "测试视频标题"
+
+    def test_confirm_default_resolution(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """不传分辨率时使用默认值 1080x1920"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            resp = client.post(
+                f"/api/v1/tasks/{preview.id}/confirm",
+                json={},
+            )
+
+        assert resp.status_code == 200
+        item = resp.json()["items"][0]
+        assert item["output_width"] == 1080
+        assert item["output_height"] == 1920
+
+    def test_confirm_creates_new_task_in_repo(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """确认生成的任务确实被存入 repository"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+
+        initial_count = len(gen_task_repo._store)
+
+        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            resp = client.post(
+                f"/api/v1/tasks/{preview.id}/confirm",
+                json={},
+            )
+
+        assert resp.status_code == 200
+        new_task_id = resp.json()["items"][0]["id"]
+        assert new_task_id != preview.id
+        assert len(gen_task_repo._store) == initial_count + 1
+
+        new_task = gen_task_repo.get(new_task_id)
+        assert new_task is not None
+        assert new_task.is_preview is False
+        assert new_task.source_task_id == preview.id