fix: 修复脚本末尾多余空行，通过black格式检查

.gitea/workflows/test-ssh-secret.yml

@@ -1,69 +0,0 @@
-name: Test SSH Secret
-on:
-  push:
-    branches: [develop]
-    paths:
-      - '.gitea/workflows/test-ssh-secret.yml'
-
-jobs:
-  test-ssh:
-    runs-on: ubuntu-22.04
-    steps:
-      - name: Install SSH client
-        run: |
-          which ssh || (apt-get update && apt-get install -y openssh-client)
-          ssh -V
-
-      - name: Debug environment
-        run: |
-          echo "=== Environment ==="
-          echo "Runner hostname: $(hostname)"
-          echo "Runner IP: $(hostname -i || echo 'unknown')"
-          echo "Current user: $(whoami)"
-          echo "=== Secrets check ==="
-          if [ -n "$STAGING_SSH_HOST" ]; then
-            echo "STAGING_SSH_HOST: [SET] value_length=${#STAGING_SSH_HOST}"
-          else
-            echo "STAGING_SSH_HOST: [EMPTY]"
-          fi
-          if [ -n "$STAGING_SSH_USER" ]; then
-            echo "STAGING_SSH_USER: [SET] value_length=${#STAGING_SSH_USER}"
-          else
-            echo "STAGING_SSH_USER: [EMPTY]"
-          fi
-          if [ -n "$STAGING_SSH_KEY" ]; then
-            echo "STAGING_SSH_KEY: [SET] value_length=${#STAGING_SSH_KEY}"
-          else
-            echo "STAGING_SSH_KEY: [EMPTY]"
-          fi
-        env:
-          STAGING_SSH_HOST: ${{ secrets.STAGING_SSH_HOST }}
-          STAGING_SSH_USER: ${{ secrets.STAGING_SSH_USER }}
-          STAGING_SSH_KEY: ${{ secrets.STAGING_SSH_KEY }}
-
-      - name: Setup SSH key
-        run: |
-          mkdir -p ~/.ssh
-          chmod 700 ~/.ssh
-          echo "$STAGING_SSH_KEY" > ~/.ssh/id_ed25519
-          chmod 600 ~/.ssh/id_ed25519
-          ssh-keygen -y -f ~/.ssh/id_ed25519 > ~/.ssh/id_ed25519.pub 2>/dev/null || echo "No public key generated"
-          echo "=== SSH Key fingerprint ==="
-          ssh-keygen -lf ~/.ssh/id_ed25519 || echo "Key fingerprint failed"
-        env:
-          STAGING_SSH_KEY: ${{ secrets.STAGING_SSH_KEY }}
-
-      - name: Test SSH connection
-        run: |
-          echo "Attempting SSH connection to $STAGING_SSH_HOST..."
-          ssh -i ~/.ssh/id_ed25519 \
-            -o StrictHostKeyChecking=no \
-            -o UserKnownHostsFile=/dev/null \
-            -o ConnectTimeout=10 \
-            -o BatchMode=yes \
-            -v \
-            $STAGING_SSH_USER@$STAGING_SSH_HOST "echo 'SSH_CONNECTION_SUCCESS' && hostname && whoami"
-          echo "=== SSH Test Complete ==="
-        env:
-          STAGING_SSH_HOST: ${{ secrets.STAGING_SSH_HOST }}
-          STAGING_SSH_USER: ${{ secrets.STAGING_SSH_USER }}

scripts/check_migration_safety.py

@@ -0,0 +1,249 @@
+#!/usr/bin/env python3
+"""
+数据库迁移破坏性变更安全检查
+
+只检查 Alembic 迁移文件的 upgrade 函数中是否包含破坏性操作：
+- DROP TABLE
+- ALTER TABLE ... DROP COLUMN
+- 列类型变更（可能导致数据丢失）
+- NOT NULL 约束新增（无默认值时）
+- RENAME TABLE / RENAME COLUMN
+
+忽略 downgrade 函数中的操作（那是回滚逻辑，正常的）。
+
+使用方式：
+  # 检查所有迁移（不推荐，会扫历史已执行的迁移）
+  python3 scripts/check_migration_safety.py
+
+  # 只检查与目标分支相比新增的迁移（推荐用于CI）
+  python3 scripts/check_migration_safety.py --diff-against origin/main
+
+  # 只检查指定版本之后的迁移
+  python3 scripts/check_migration_safety.py --since 030_xxx
+
+退出码：
+  0 - 安全 / 只有非破坏性变更
+  1 - 检测到高风险破坏性变更
+  2 - 检测到中风险变更，需人工确认
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import List, Tuple
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+ALEMBIC_VERSIONS_DIR = REPO_ROOT / "alembic" / "versions"
+
+# 高风险模式：直接导致数据丢失（只在 upgrade 中检查）
+HIGH_RISK_PATTERNS = [
+    (r"\bop\.drop_table\(", "op.drop_table() - 删除表，数据永久丢失"),
+    (r"\bop\.drop_column\(", "op.drop_column() - 删除列，数据永久丢失"),
+]
+
+# 中风险模式：可能导致数据丢失或兼容性问题
+MEDIUM_RISK_PATTERNS = [
+    (r"op\.alter_column\([^)]*nullable\s*=\s*False", "新增 NOT NULL 约束 - 旧数据可能为空导致迁移失败"),
+    (r"op\.alter_column\([^)]*type_\s*=", "列类型变更 - 可能导致数据截断或转换失败"),
+    (r"\bop\.rename_table\(", "op.rename_table() - 重命名表，可能导致依赖该表的代码报错"),
+    (r"\bop\.rename_column\(", "op.rename_column() - 重命名列，可能导致依赖该列的代码报错"),
+    (r"\bop\.drop_index\(", "op.drop_index() - 删除索引，可能影响查询性能"),
+    (r"\bop\.drop_constraint\(", "op.drop_constraint() - 删除约束，可能影响数据完整性"),
+]
+
+# 安全模式：这些是安全的新增操作
+SAFE_PATTERNS = [
+    (r"\bop\.create_table\(", "新建表"),
+    (r"\bop\.add_column\(", "新增列"),
+    (r"\bop\.create_index\(", "新建索引"),
+    (r"\bop\.create_unique_constraint\(", "新建唯一约束"),
+    (r"\bop\.create_foreign_key\(", "新建外键约束"),
+]
+
+
+def extract_upgrade_content(content: str) -> str:
+    """
+    从迁移文件中提取 upgrade 函数的内容。
+    只检查 upgrade 中的操作，忽略 downgrade。
+    """
+    upgrade_match = re.search(r"def upgrade\b[^:]*:", content)
+    if not upgrade_match:
+        return ""
+
+    upgrade_start = upgrade_match.end()
+
+    # 找到下一个顶层 def（通常是 def downgrade）作为结束位置
+    rest = content[upgrade_start:]
+    downgrade_match = re.search(r"\n\ndef\s+\w+\b", rest)
+    if downgrade_match:
+        upgrade_end = upgrade_start + downgrade_match.start()
+    else:
+        upgrade_end = len(content)
+
+    return content[upgrade_start:upgrade_end]
+
+
+def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
+    """
+    通过 git diff 对比目标分支/commit，找出 alembic/versions/ 下新增的迁移文件。
+    只包含新增文件（A状态），不包含修改或删除的文件。
+    """
+    try:
+        result = subprocess.run(
+            ["git", "diff", "--name-only", "--diff-filter=A", diff_target, "HEAD", "--", "alembic/versions/"],
+            cwd=str(REPO_ROOT),
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        files = [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]
+        return [REPO_ROOT / f for f in files]
+    except subprocess.CalledProcessError as e:
+        print(f"⚠️  git diff 失败（{diff_target}）：{e.stderr.strip()}")
+        print(f"   降级为检查所有迁移文件")
+        return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
+
+
+def find_new_migrations(since_revision: str | None = None, diff_against: str | None = None) -> List[Path]:
+    """
+    找出需要检查的迁移文件。
+    优先级：diff_against > since_revision > 全部
+    """
+    if diff_against:
+        return get_new_migrations_via_diff(diff_against)
+
+    all_migrations = sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
+    if not since_revision:
+        return all_migrations
+
+    result = []
+    found = False
+    for m in all_migrations:
+        if since_revision in m.name or since_revision in m.stem:
+            found = True
+            continue
+        if found:
+            result.append(m)
+
+    return result if found else all_migrations
+
+
+def analyze_migration(file_path: Path) -> Tuple[List[str], List[str], List[str]]:
+    """分析单个迁移文件 upgrade 部分的风险等级"""
+    content = file_path.read_text()
+    upgrade_content = extract_upgrade_content(content)
+
+    if not upgrade_content:
+        return [], [], [f"{file_path.name}: 未找到 upgrade 函数"]
+
+    high_risks = []
+    medium_risks = []
+    safes = []
+
+    for pattern, desc in HIGH_RISK_PATTERNS:
+        if re.search(pattern, upgrade_content):
+            high_risks.append(f"{file_path.name}: {desc}")
+
+    for pattern, desc in MEDIUM_RISK_PATTERNS:
+        if re.search(pattern, upgrade_content):
+            medium_risks.append(f"{file_path.name}: {desc}")
+
+    for pattern, desc in SAFE_PATTERNS:
+        if re.search(pattern, upgrade_content):
+            safes.append(f"{file_path.name}: {desc}")
+
+    return high_risks, medium_risks, safes
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument(
+        "--since",
+        default=os.getenv("MIGRATION_SINCE_REVISION"),
+        help="只检查指定版本之后的迁移（如：030_xxx），不传则检查所有迁移",
+    )
+    parser.add_argument(
+        "--diff-against",
+        default=os.getenv("MIGRATION_DIFF_AGAINST"),
+        help="对比指定分支/commit，只检查新增的迁移文件（推荐用于CI，如 origin/main）",
+    )
+    parser.add_argument(
+        "--warn-only",
+        action="store_true",
+        help="只警告不失败（用于非强制门禁场景）",
+    )
+    parser.add_argument(
+        "--allow-medium-risk",
+        action="store_true",
+        help="允许中风险变更（只拦截高风险）",
+    )
+    args = parser.parse_args()
+
+    migrations = find_new_migrations(args.since, args.diff_against)
+
+    if not migrations:
+        print("✅ 未找到需要检查的新增迁移文件，跳过")
+        return 0
+
+    print(f"🔍 正在检查 {len(migrations)} 个迁移文件的 upgrade 操作...")
+    if args.diff_against:
+        print(f"   （对比基准：{args.diff_against}，仅检查新增迁移）")
+    print()
+
+    all_high = []
+    all_medium = []
+    all_safe = []
+
+    for m in migrations:
+        high, medium, safe = analyze_migration(m)
+        all_high.extend(high)
+        all_medium.extend(medium)
+        all_safe.extend(safe)
+
+    if all_safe:
+        print("✅ 安全变更：")
+        for s in all_safe:
+            print(f"   - {s}")
+        print()
+
+    if all_medium:
+        print("⚠️  中风险变更（需人工确认）：")
+        for m_item in all_medium:
+            print(f"   - {m_item}")
+        print()
+
+    if all_high:
+        print("❌ 高风险破坏性变更（禁止自动部署）：")
+        for h in all_high:
+            print(f"   - {h}")
+        print()
+
+    print("=" * 60)
+    print(f"检查结果：{len(all_safe)} 项安全 / {len(all_medium)} 项中风险 / {len(all_high)} 项高风险")
+    print()
+
+    if all_high:
+        print("❌ 检测到高风险破坏性变更，CI 检查失败！")
+        print("   如果确认这是预期操作，请在 MR/PR 中说明原因并获得审批。")
+        if args.warn_only:
+            return 0
+        return 1
+
+    if all_medium and not args.allow_medium_risk:
+        print("⚠️  检测到中风险变更，请人工确认后再部署。")
+        if args.warn_only:
+            return 0
+        print("（如需仅拦截高风险，可使用 --allow-medium-risk 参数）")
+        return 2
+
+    print("✅ 未检测到破坏性变更")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci_coverage_summary.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+"""解析 coverage.xml 并输出覆盖率汇总。"""
+
+import os
+import sys
+import xml.etree.ElementTree as ET
+
+THRESHOLD = int(os.environ.get("COVERAGE_THRESHOLD", 65))  # 行覆盖率门槛，百分比，可通过环境变量覆盖
+
+
+def main() -> int:
+    try:
+        tree = ET.parse("coverage.xml")
+    except FileNotFoundError:
+        print("coverage.xml 不存在，跳过汇总")
+        return 0
+
+    root = tree.getroot()
+    line_rate = float(root.get("line-rate", 0)) * 100
+    branch_rate = float(root.get("branch-rate", 0)) * 100
+    lines_covered = int(root.get("lines-covered", 0))
+    lines_valid = int(root.get("lines-valid", 0))
+
+    print(f"行覆盖率: {line_rate:.2f}% ({lines_covered}/{lines_valid})")
+    print(f"分支覆盖率: {branch_rate:.2f}%")
+    print(f"门槛: {THRESHOLD}%")
+    status = "PASS ✅" if line_rate >= THRESHOLD else "FAIL ❌"
+    print(f"状态: {status}")
+
+    return 0 if line_rate >= THRESHOLD else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci_notify_failure.py

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

scripts/ci_notify_success.py

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

scripts/rollback_production.sh

@@ -0,0 +1,289 @@
+#!/bin/bash
+set -eu
+
+# ============================================================
+# 生产环境一键回滚脚本
+#
+# 用法:
+#   IMAGE_TAG=<版本号> REGISTRY_TOKEN=<token> sh rollback_production.sh
+#
+# 功能:
+#   1. 拉取指定版本镜像
+#   2. 数据库回滚到对应版本（alembic downgrade）
+#   3. 重启 api/worker/web 三个服务
+#   4. 健康检查确认服务正常
+#
+# 环境变量:
+#   IMAGE_TAG           - 要回滚到的版本标签（必填）
+#   REGISTRY_TOKEN      - Registry 访问 token（可选）
+#   SKIP_DB_ROLLBACK    - 跳过数据库回滚（1=跳过，默认不跳过）
+#   DB_ROLLBACK_REV     - 数据库回滚到的版本（默认自动用镜像里的 head）
+# ============================================================
+
+IMAGE_TAG="${IMAGE_TAG:-}"
+REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
+REGISTRY_USER="${REGISTRY_USER:-xiaoxia}"
+REGISTRY_TOKEN="${REGISTRY_TOKEN:-}"
+SKIP_DB_ROLLBACK="${SKIP_DB_ROLLBACK:-0}"
+DB_ROLLBACK_REV="${DB_ROLLBACK_REV:-}"
+
+ENV_FILE="${ENV_FILE:-/var/lib/xiaoxia-saas-production/.env}"
+GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-production/generated}"
+LEGACY_ASSETS_DIR="${LEGACY_ASSETS_DIR:-/var/lib/xiaoxia-saas-production/legacy-assets}"
+
+if [ -z "$IMAGE_TAG" ]; then
+  echo "❌ IMAGE_TAG 是必填参数"
+  echo "用法: IMAGE_TAG=v0.1.125 REGISTRY_TOKEN=xxx sh rollback_production.sh"
+  exit 1
+fi
+
+test -f "$ENV_FILE"
+mkdir -p "$GENERATED_DIR"
+mkdir -p "$LEGACY_ASSETS_DIR"
+
+echo "=========================================="
+echo "  生产环境回滚  →  $IMAGE_TAG"
+echo "=========================================="
+echo ""
+
+# 先获取当前版本
+CURRENT_VERSION=""
+if docker inspect xiaoxia-api-production >/dev/null 2>&1; then
+  CURRENT_VERSION=$(docker inspect --format '{{ index .Config.Env 0 }}' xiaoxia-api-production 2>/dev/null | grep APP_VERSION | cut -d= -f2 || echo "unknown")
+fi
+echo "当前版本: ${CURRENT_VERSION:-unknown}"
+echo "回滚目标: $IMAGE_TAG"
+echo ""
+
+# 确认
+read -p "⚠️  确认要回滚生产环境到 $IMAGE_TAG 吗？(输入 YES 确认): " confirm
+if [ "$confirm" != "YES" ]; then
+  echo "已取消"
+  exit 0
+fi
+echo ""
+
+# ----- 登录 Registry -----
+if [ -n "$REGISTRY_TOKEN" ]; then
+  echo "登录 Registry: $REGISTRY"
+  REGISTRY_HOST=$(echo "$REGISTRY" | cut -d/ -f1)
+  printf '%s' "$REGISTRY_TOKEN" | docker login "$REGISTRY_HOST" -u "$REGISTRY_USER" --password-stdin 2>/dev/null || {
+    echo "WARN: docker login failed, will try to pull anyway"
+  }
+fi
+
+# ----- Pull 镜像 -----
+REGISTRY_API="${REGISTRY}/xiaoxia-saas-api:${IMAGE_TAG}"
+REGISTRY_WORKER="${REGISTRY}/xiaoxia-saas-worker:${IMAGE_TAG}"
+REGISTRY_WEB="${REGISTRY}/xiaoxia-saas-web:${IMAGE_TAG}"
+
+LOCAL_API="xiaoxia-saas-api:${IMAGE_TAG}"
+LOCAL_WORKER="xiaoxia-saas-worker:${IMAGE_TAG}"
+LOCAL_WEB="xiaoxia-saas-web:${IMAGE_TAG}"
+
+echo "Pulling API image..."
+docker pull "$REGISTRY_API"
+echo "Pulling Worker image..."
+docker pull "$REGISTRY_WORKER"
+echo "Pulling Web image..."
+docker pull "$REGISTRY_WEB"
+
+docker tag "$REGISTRY_API" "$LOCAL_API"
+docker tag "$REGISTRY_WORKER" "$LOCAL_WORKER"
+docker tag "$REGISTRY_WEB" "$LOCAL_WEB"
+echo "所有镜像拉取完成"
+echo ""
+
+# ----- 检查基础设施容器 -----
+echo "检查基础设施容器..."
+for c in xiaoxia-postgres-production xiaoxia-redis-production; do
+  if ! docker inspect "$c" >/dev/null 2>&1; then
+    echo "ERROR: Required container not found: $c"
+    exit 1
+  fi
+  state=$(docker inspect -f '{{.State.Status}}' "$c")
+  if [ "$state" != "running" ]; then
+    echo "ERROR: Container not running: $c ($state)"
+    exit 1
+  fi
+done
+
+docker network create xiaoxia-net-production 2>/dev/null || true
+echo ""
+
+# ----- 数据库回滚 -----
+if [ "$SKIP_DB_ROLLBACK" = "1" ]; then
+  echo "⏭️  跳过数据库回滚（SKIP_DB_ROLLBACK=1）"
+else
+  echo "🔄 执行数据库回滚..."
+  if [ -n "$DB_ROLLBACK_REV" ]; then
+    # 回滚到指定版本
+    echo "回滚到版本: $DB_ROLLBACK_REV"
+    docker run --rm \
+      --env-file "$ENV_FILE" \
+      --network xiaoxia-net-production \
+      -e APP_ENV=production \
+      "$LOCAL_API" sh -c "cd /app && alembic downgrade $DB_ROLLBACK_REV"
+  else
+    # 用目标镜像的 alembic head 来判断是否需要回滚
+    # 先检查当前DB版本和目标版本的关系
+    echo "检测数据库当前版本与目标版本..."
+    CURRENT_DB_REV=$(docker run --rm \
+      --env-file "$ENV_FILE" \
+      --network xiaoxia-net-production \
+      -e APP_ENV=production \
+      "$LOCAL_API" sh -c "cd /app && alembic current" 2>&1 | tail -1 | awk '{print $1}')
+    TARGET_DB_HEAD=$(docker run --rm \
+      "$LOCAL_API" sh -c "cd /app && alembic head" 2>&1 | tail -1 | awk '{print $1}')
+
+    echo "当前 DB 版本: ${CURRENT_DB_REV:-unknown}"
+    echo "目标 DB 版本: ${TARGET_DB_HEAD:-unknown}"
+
+    if [ "$CURRENT_DB_REV" = "$TARGET_DB_HEAD" ]; then
+      echo "✅ 数据库版本与目标版本一致，无需回滚"
+    else
+      echo "⚠️  数据库版本不一致，尝试回滚..."
+      echo "注意：自动回滚可能无法正确处理，请确认 DB_ROLLBACK_REV 参数"
+      echo "如果需要跳过数据库回滚，请设置 SKIP_DB_ROLLBACK=1"
+      exit 1
+    fi
+  fi
+  echo "数据库回滚完成"
+fi
+echo ""
+
+# ----- 停止旧容器 -----
+echo "停止旧容器..."
+docker rm -f xiaoxia-api-production 2>/dev/null || true
+docker rm -f xiaoxia-worker-production 2>/dev/null || true
+docker rm -f xiaoxia-web-production 2>/dev/null || true
+echo ""
+
+# ----- 启动新容器 -----
+LOG_OPTS="--log-driver json-file --log-opt max-size=50m --log-opt max-file=3"
+
+echo "启动 API 容器..."
+docker run -d \
+  --name xiaoxia-api-production \
+  --env-file "$ENV_FILE" \
+  --network xiaoxia-net-production \
+  -p 127.0.0.1:8001:8000 \
+  -e APP_ENV=production \
+  -e APP_VERSION="$IMAGE_TAG" \
+  -e GENERATED_FILES_DIR=/app/generated \
+  -e GENERATED_FILES_URL_PREFIX=/generated-files \
+  -e PUBLIC_API_BASE_URL=https://api.xiaoxiajianji.com \
+  -v "$GENERATED_DIR:/app/generated" \
+  --restart unless-stopped \
+  --cpus 2 \
+  --memory 2g \
+  --health-cmd "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"
+
+echo "启动 Worker 容器..."
+docker run -d \
+  --name xiaoxia-worker-production \
+  --env-file "$ENV_FILE" \
+  --network xiaoxia-net-production \
+  -e APP_ENV=production \
+  -e APP_VERSION="$IMAGE_TAG" \
+  -e WORKER_CONCURRENCY=1 \
+  -e WORKER_MAX_TASKS_PER_CHILD=100 \
+  -e GENERATED_FILES_DIR=/app/generated \
+  -e GENERATED_FILES_URL_PREFIX=/generated-files \
+  -e PUBLIC_API_BASE_URL=https://api.xiaoxiajianji.com \
+  -v "$GENERATED_DIR:/app/generated" \
+  --restart unless-stopped \
+  --cpus 2 \
+  --memory 2g \
+  --health-cmd "sh -c \"grep -q celery /proc/1/cmdline || exit 1\"" \
+  --health-interval 30s \
+  --health-timeout 10s \
+  --health-retries 3 \
+  --health-start-period 30s \
+  $LOG_OPTS \
+  "$LOCAL_WORKER"
+
+# Web legacy assets
+LEGACY_VOLUME=""
+if [ -d "$LEGACY_ASSETS_DIR" ] && [ "$(ls -A "$LEGACY_ASSETS_DIR" 2>/dev/null)" ]; then
+  LEGACY_VOLUME="-v ${LEGACY_ASSETS_DIR}:/usr/share/nginx/html/assets-legacy/assets:ro"
+  echo "Web 容器: legacy assets 已挂载"
+else
+  echo "Web 容器: 没有 legacy assets"
+fi
+
+echo "启动 Web 容器..."
+docker run -d \
+  --name xiaoxia-web-production \
+  --network xiaoxia-net-production \
+  -p 127.0.0.1:3002:80 \
+  --restart unless-stopped \
+  --cpus 0.5 \
+  --memory 512m \
+  $LEGACY_VOLUME \
+  --health-cmd "wget --spider -q http://127.0.0.1:80" \
+  --health-interval 30s \
+  --health-timeout 5s \
+  --health-retries 3 \
+  $LOG_OPTS \
+  "$LOCAL_WEB"
+echo ""
+
+# ----- 健康检查 -----
+echo "等待 API 健康..."
+i=0
+while [ "$i" -lt 40 ]; do
+  if curl -sf --max-time 5 http://127.0.0.1:8001/health >/dev/null 2>&1; then
+    echo "API is healthy!"
+    break
+  fi
+  i=$((i + 1))
+  echo "  Waiting... ($i/40)"
+  sleep 3
+done
+
+if [ "$i" -ge 40 ]; then
+  echo "❌ API 在 120s 内未就绪"
+  docker logs --tail 50 xiaoxia-api-production
+  exit 1
+fi
+
+echo "等待 Web 健康..."
+i=0
+while [ "$i" -lt 15 ]; do
+  if curl -sf --max-time 5 http://127.0.0.1:3002/ >/dev/null 2>&1; then
+    echo "Web is healthy!"
+    break
+  fi
+  i=$((i + 1))
+  echo "  Waiting... ($i/15)"
+  sleep 2
+done
+
+if [ "$i" -ge 15 ]; then
+  echo "❌ Web 在 30s 内未就绪"
+  docker logs --tail 30 xiaoxia-web-production
+  exit 1
+fi
+echo ""
+
+# ----- 清理 -----
+echo "清理旧镜像..."
+docker image prune -af --filter "until=168h" 2>/dev/null || true
+docker builder prune -af --filter "until=168h" 2>/dev/null || true
+
+echo ""
+echo "=========================================="
+echo "  ✅ 生产环境回滚完成"
+echo "=========================================="
+echo "API:      http://127.0.0.1:8001"
+echo "Web:      http://127.0.0.1:3002"
+echo "Version:  $IMAGE_TAG"
+docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Image}}" | grep production
+