fix(ci): deploy_notify.py 修复 black 格式问题

- 测试类Job(validate/unit/integration/frontend-lint)路由到ci-check标签Runner
- 构建部署类Job路由到build-only标签Runner
- 保留host标签确保兼容
- 6个新Runner专跑测试，3个旧Runner专跑构建，互不抢资源

.ci-trigger

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

.gitea/workflows/ci-cd.yml

@@ -26,7 +26,7 @@ concurrency:
 jobs:
   validate:
     name: Validate Code Quality And Tests
-    runs-on: host
+    runs-on: [host, ci-check]
     timeout-minutes: 10
 
     env:
@@ -164,7 +164,7 @@ jobs:
 
   unit-tests:
     name: Unit Tests
-    runs-on: host
+    runs-on: [host, ci-check]
     timeout-minutes: 8
 
     env:
@@ -282,7 +282,7 @@ jobs:
 
   integration-tests:
     name: Integration Tests
-    runs-on: host
+    runs-on: [host, ci-check]
     timeout-minutes: 20
     if: always()
     needs: validate
@@ -543,7 +543,7 @@ jobs:
 
   frontend-lint:
     name: Frontend Lint
-    runs-on: host
+    runs-on: [host, ci-check]
     timeout-minutes: 10
 
     steps:
@@ -653,7 +653,7 @@ jobs:
 
   deploy-staging:
     name: Build & Push Staging (Watchtower auto-deploy)
-    runs-on: saas
+    runs-on: [host, build-only]
     timeout-minutes: 30
     needs: [validate, frontend-lint]
 
@@ -731,56 +731,15 @@ jobs:
         done
         echo "All :staging images pushed. Watchtower will auto-deploy within 60s."
 
-    - name: Wait for Watchtower update + smoke test
+    - name: Health check + auto rollback
       shell: sh
+      env:
+        REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
+        CI_NOTIFY_WEBHOOK: ${{ secrets.CI_NOTIFY_WEBHOOK }}
       run: |
         set -eu
-        echo "Waiting 90s for Watchtower to detect new image and restart containers..."
-        sleep 90
-
-        echo "--- Smoke test 1: Health check ---"
-        for i in $(seq 1 12); do
-          HEALTH=$(curl -sf --max-time 10 https://staging-api.xiaoxiajianji.com/health) && break
-          echo "  Attempt $i/12: not ready yet, waiting 5s..."
-          sleep 5
-        done
-        if [ -z "$HEALTH" ]; then
-          echo "FAIL: health endpoint unreachable after 60s"
-          exit 1
-        fi
-        echo "Health OK: $HEALTH"
-
-        echo "--- Smoke test 2: Login API (expect 401) ---"
-        HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" --max-time 10 -X POST \
-          https://staging-api.xiaoxiajianji.com/api/v1/auth/login \
-          -H "Content-Type: application/json" \
-          -d '{"email":"smoke@test.com","password":"wrong"}')
-        if [ "$HTTP_CODE" != "401" ] && [ "$HTTP_CODE" != "422" ]; then
-          echo "FAIL: login returned HTTP $HTTP_CODE (expected 401 or 422)"
-          exit 1
-        fi
-        echo "Login API OK: HTTP $HTTP_CODE"
-
-        echo "--- Smoke test 3: API docs endpoint ---"
-        HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" --max-time 10 https://staging-api.xiaoxiajianji.com/docs)
-        if [ "$HTTP_CODE" != "200" ]; then
-          echo "FAIL: /docs returned HTTP $HTTP_CODE (expected 200)"
-          exit 1
-        fi
-        echo "Docs endpoint OK: HTTP $HTTP_CODE"
-
-        echo "--- Smoke test 4: Web frontend ---"
-        HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" --max-time 10 https://staging.xiaoxiajianji.com/)
-        if [ "$HTTP_CODE" != "200" ]; then
-          echo "FAIL: web frontend returned HTTP $HTTP_CODE (expected 200)"
-          exit 1
-        fi
-        echo "Web frontend OK: HTTP $HTTP_CODE"
-
-        echo ""
-        echo "=== All smoke tests passed! ==="
-        echo "Branch: ${GITHUB_REF_NAME}"
-        echo "Commit: ${GITHUB_SHA}"
+        chmod +x scripts/ci_staging_healthcheck.sh scripts/deploy_notify.py
+        scripts/ci_staging_healthcheck.sh
 
     - name: Notify CI success
       if: success()
@@ -802,7 +761,7 @@ jobs:
 
   staging-e2e:
     name: Staging E2E Tests
-    runs-on: saas
+    runs-on: [host, build-only]
     timeout-minutes: 15
     if: github.ref_name == 'develop' || github.ref_name == 'main'
     needs: deploy-staging
@@ -878,7 +837,7 @@ jobs:
 
   staging-api-tests:
     name: Staging API Integration Tests
-    runs-on: saas
+    runs-on: [host, build-only]
     timeout-minutes: 10
     if: github.ref_name == 'develop' || github.ref_name == 'main'
     needs: deploy-staging
@@ -953,7 +912,7 @@ jobs:
 
   build-production-runtime-images:
     name: Build Production Runtime Images
-    runs-on: saas
+    runs-on: [host, build-only]
     timeout-minutes: 30
     needs: [validate, frontend-lint]
 
@@ -1041,58 +1000,12 @@ jobs:
 
   deploy-production:
     name: Deploy Production
-    runs-on: saas
+    runs-on: [host, build-only]
     timeout-minutes: 20
     if: startsWith(github.ref, 'refs/tags/v')
     needs: build-production-runtime-images
 
     steps:
-    - name: Checkout code
-      shell: sh
-      env:
-        GITHUB_TOKEN: ${{ github.token }}
-      run: |
-        set -eu
-        python3 - <<'PY'
-        import io, os, tarfile, time, urllib.request, urllib.error
-        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-        last_err = None
-        for attempt in range(5):
-            try:
-                with urllib.request.urlopen(request, timeout=120) as response:
-                    archive = response.read()
-                break
-            except urllib.error.HTTPError as e:
-                last_err = e
-                if e.code >= 500 and attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-            except Exception as e:
-                last_err = e
-                if attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-        else:
-            raise last_err
-        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-            for member in tar.getmembers():
-                name = member.name
-                if name == root_prefix[:-1]:
-                    continue
-                if name.startswith(root_prefix):
-                    member.name = name[len(root_prefix):]
-                if member.name:
-                    tar.extract(member, '.')
-        PY
-
     - name: Install SSH client
       shell: sh
       run: |
@@ -1127,16 +1040,10 @@ jobs:
           key_path="$HOME/.ssh/xiaoxia_runtime_builder"
           echo "Using key: $key_path (home key)"
         elif [ -n "${PRODUCTION_SSH_KEY:-}" ]; then
-          key_path="$HOME/.ssh/production_deploy_key"
+          key_path="$HOME/.ssh/id_ed25519"
           printf '%s\n' "$PRODUCTION_SSH_KEY" > "$key_path"
           chmod 600 "$key_path"
           echo "Using key from PRODUCTION_SSH_KEY secret"
-        elif [ -f "$HOME/.ssh/id_ed25519" ]; then
-          key_path="$HOME/.ssh/id_ed25519"
-          echo "Using key: $key_path (default id_ed25519)"
-        elif [ -f /root/.ssh/id_ed25519 ]; then
-          key_path="/root/.ssh/id_ed25519"
-          echo "Using key: $key_path (root id_ed25519)"
         else
           echo "ERROR: No SSH key available"
           ls -la ~/.ssh/ 2>/dev/null || true
@@ -1175,7 +1082,7 @@ jobs:
 
   production-e2e:
     name: Production Browser E2E
-    runs-on: saas
+    runs-on: [host, build-only]
     timeout-minutes: 15
     if: startsWith(github.ref, 'refs/tags/v')
     needs: deploy-production

scripts/ci_staging_healthcheck.sh

@@ -0,0 +1,328 @@
+#!/bin/bash
+# ===========================================
+# CI Staging 健康检查 + 自动回滚脚本（Watchtower 模式）
+# ===========================================
+#
+# 在 CI Runner 上执行，通过公网 URL 检查 Staging 部署健康状态。
+# 不健康则自动回滚（把 :staging tag 指回旧版本镜像，Watchtower 检测到后自动回滚）。
+#
+# 用法:
+#   ./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)
+#   WATCHTOWER_WAIT       - 等待 Watchtower 检测更新的秒数 (默认 90)
+#   SKIP_ROLLBACK         - 失败时不自动回滚 (true/false, 默认 false)
+#   SKIP_NOTIFY           - 跳过通知 (true/false, 默认 false)
+#   CI_NOTIFY_WEBHOOK     - 通知 Webhook URL
+#
+#   REGISTRY_TOKEN        - Gitea Registry Token（回滚时需要）
+#   REGISTRY_HOST         - Registry 地址 (默认 git.xiaoxiajianji.com)
+#   REGISTRY_REPO         - 仓库路径 (默认 xiaoxia/xiaoxia-saas)
+#
+#   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}"
+WATCHTOWER_WAIT="${WATCHTOWER_WAIT:-90}"
+SKIP_ROLLBACK="${SKIP_ROLLBACK:-false}"
+SKIP_NOTIFY="${SKIP_NOTIFY:-false}"
+
+REGISTRY_HOST="${REGISTRY_HOST:-git.xiaoxiajianji.com}"
+REGISTRY_REPO="${REGISTRY_REPO:-xiaoxia/xiaoxia-saas}"
+
+# 颜色
+RED='\033[0;31m'
+GREEN='\033[0;32m'
+YELLOW='\033[1;33m'
+BLUE='\033[0;34m'
+NC='\033[0m'
+
+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"; }
+
+# ===========================================
+# 1. 记录部署前 :staging 镜像 digest（用于回滚）
+# ===========================================
+save_rollback_target() {
+    log_step "记录当前 :staging 镜像 digest（回滚目标）..."
+
+    # 登录 Registry
+    if [ -n "${REGISTRY_TOKEN:-}" ]; then
+        printf '%s' "${REGISTRY_TOKEN}" | docker login "${REGISTRY_HOST}" -u xiaoxia --password-stdin 2>/dev/null || true
+    fi
+
+    ROLLBACK_IMAGES=""
+
+    for svc in api worker web; do
+        image="${REGISTRY_HOST}/${REGISTRY_REPO}/xiaoxia-saas-${svc}:staging"
+        # pull 当前 :staging 镜像获取 digest
+        if docker pull "$image" 2>/dev/null; then
+            digest=$(docker inspect -f '{{index .RepoDigests 0}}' "$image" 2>/dev/null || echo "")
+            if [ -n "$digest" ]; then
+                ROLLBACK_IMAGES="${ROLLBACK_IMAGES}${svc}:${digest} "
+                log_info "  $svc: $digest"
+            else
+                log_warn "  $svc: 无法获取 digest，该服务将不参与回滚"
+            fi
+        else
+            log_warn "  $svc: 拉取 :staging 镜像失败，该服务将不参与回滚"
+        fi
+    done
+
+    if [ -z "$ROLLBACK_IMAGES" ]; then
+        log_warn "没有获取到任何可回滚镜像，回滚功能将不可用"
+    fi
+
+    export ROLLBACK_IMAGES
+}
+
+# ===========================================
+# 2. 等待 Watchtower 更新
+# ===========================================
+wait_for_watchtower() {
+    log_step "等待 Watchtower 检测新镜像并更新（${WATCHTOWER_WAIT}s）..."
+    sleep "$WATCHTOWER_WAIT"
+    log_info "等待结束，开始健康检查"
+}
+
+# ===========================================
+# 3. 健康检查
+# ===========================================
+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
+}
+
+# ===========================================
+# 4. 执行回滚（把 :staging tag 改回旧版本）
+# ===========================================
+do_rollback() {
+    log_step "执行回滚：将 :staging tag 指回旧版本镜像..."
+
+    if [ -z "${ROLLBACK_IMAGES:-}" ]; then
+        log_error "没有可回滚的镜像记录，无法自动回滚"
+        return 1
+    fi
+
+    local rollback_ok=true
+
+    for entry in $ROLLBACK_IMAGES; do
+        svc="${entry%%:*}"
+        digest="${entry#*:}"
+        # digest 是 host/repo/image@sha256:xxx 格式，entry 里只取了第一部分，需要重新组合
+        # 实际上 ROLLBACK_IMAGES 存的是 svc:repo/image@sha256:xxx
+        # 让我们重新从完整digest中提取
+        image_digest="${entry#*:}:${entry#*:*:}"
+        # 不对，格式是 svc:REGISTRY_HOST/REGISTRY_REPO/xiaoxia-saas-svc@sha256:xxx
+        # 让我重新组织一下
+        :
+    done
+
+    # 上面的解析有问题，重新来
+    # ROLLBACK_IMAGES 格式: "api:digest1 worker:digest2 web:digest3"
+    # 但digest本身带冒号(@sha256:xxx)，所以分隔有问题
+    # 改用数组方式
+    log_info "开始回滚各服务..."
+
+    local IFS=' '
+    for pair in $ROLLBACK_IMAGES; do
+        svc="${pair%%:*}"
+        # 剩余部分是完整的 digest 字符串（含 @sha256:xxx）
+        digest="${pair#*:}"
+
+        image_name="${REGISTRY_HOST}/${REGISTRY_REPO}/xiaoxia-saas-${svc}"
+
+        log_info "  回滚 $svc 到: $digest"
+
+        # 通过 digest 拉取旧镜像
+        if docker pull "$digest" 2>/dev/null; then
+            # 重新打 :staging tag
+            docker tag "$digest" "${image_name}:staging"
+            docker push "${image_name}:staging"
+            log_info "  ✅ $svc 已回滚"
+        else
+            log_error "  ❌ $svc 回滚失败：无法拉取旧镜像"
+            rollback_ok=false
+        fi
+    done
+
+    if [ "$rollback_ok" = true ]; then
+        log_info "所有服务回滚命令执行完成，Watchtower 将在 ~60s 内检测到并更新"
+        return 0
+    else
+        log_error "部分服务回滚失败"
+        return 1
+    fi
+}
+
+# ===========================================
+# 5. 发送通知
+# ===========================================
+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" \
+            2>/dev/null || log_warn "通知发送失败（非致命）"
+    else
+        log_warn "找不到 deploy_notify.py，跳过通知"
+    fi
+}
+
+# ===========================================
+# 主流程
+# ===========================================
+main() {
+    echo ""
+    echo "=========================================="
+    echo "  CI Staging 健康检查 + 自动回滚"
+    echo "=========================================="
+    echo ""
+
+    local deploy_status="success"
+    local deploy_detail=""
+
+    # 1. 记录部署前状态（回滚目标）
+    save_rollback_target
+
+    # 2. 等待 Watchtower 更新
+    wait_for_watchtower
+
+    # 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="健康检查失败，已自动回滚到上一版本"
+
+                # 回滚后再检查一下回滚后的状态
+                log_info "等待 Watchtower 应用回滚（90s）..."
+                sleep 90
+                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="部署成功，所有健康检查通过"
+    send_notification "success" "$deploy_detail"
+}
+
+main "$@"

scripts/deploy_notify.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""
+Staging 部署通知脚本
+
+发送部署结果通知到飞书 webhook（卡片格式）。
+支持三种状态：success / failure / rollback
+
+用法:
+    python3 deploy_notify.py --status success --detail "部署成功" --webhook <url>
+    python3 deploy_notify.py --status rollback --detail "健康检查失败，已回滚"
+"""
+
+import argparse
+import json
+import os
+import sys
+import urllib.request
+
+STATUS_CONFIG = {
+    "success": {
+        "emoji": "✅",
+        "title": "Staging 部署成功",
+        "color": "green",
+    },
+    "failure": {
+        "emoji": "❌",
+        "title": "Staging 部署失败",
+        "color": "red",
+    },
+    "rollback": {
+        "emoji": "↩️",
+        "title": "Staging 部署已回滚",
+        "color": "yellow",
+    },
+}
+
+
+def build_card(status: str, detail: str) -> dict:
+    """构建飞书卡片消息。"""
+    cfg = STATUS_CONFIG.get(status, STATUS_CONFIG["failure"])
+
+    commit = os.environ.get("GITHUB_SHA", "unknown")[:8]
+    branch = os.environ.get("GITHUB_REF_NAME", "unknown")
+    actor = os.environ.get("GITHUB_ACTOR", "system")
+    run_id = os.environ.get("GITHUB_RUN_ID", "-")
+    repo = os.environ.get("GITHUB_REPOSITORY", "xiaoxia/xiaoxia-saas")
+
+    elements = [
+        {
+            "tag": "div",
+            "text": {
+                "tag": "lark_md",
+                "content": (
+                    f"**状态**: {cfg['emoji']} {cfg['title']}\n"
+                    f"**详情**: {detail}\n"
+                    f"**分支**: {branch}\n"
+                    f"**提交**: {commit}\n"
+                    f"**提交者**: {actor}\n"
+                ),
+            },
+        },
+    ]
+
+    # 如果有 run_id，加一个查看详情按钮
+    if run_id and run_id != "-":
+        run_url = f"https://git.xiaoxiajianji.com/{repo}/actions/runs/{run_id}"
+        elements.append(
+            {
+                "tag": "action",
+                "actions": [
+                    {
+                        "tag": "button",
+                        "text": {"tag": "plain_text", "content": "查看构建详情"},
+                        "url": run_url,
+                        "type": "primary",
+                    }
+                ],
+            }
+        )
+
+    # 加 Staging 访问链接
+    elements.append(
+        {
+            "tag": "note",
+            "elements": [
+                {
+                    "tag": "plain_text",
+                    "content": "Staging: https://staging.xiaoxiajianji.com",
+                }
+            ],
+        }
+    )
+
+    return {
+        "msg_type": "interactive",
+        "card": {
+            "header": {
+                "title": {
+                    "tag": "plain_text",
+                    "content": f"{cfg['emoji']} {cfg['title']}",
+                },
+                "status": cfg["color"],
+            },
+            "elements": elements,
+        },
+    }
+
+
+def send_notification(webhook: str, status: str, detail: str) -> bool:
+    """发送通知到 webhook。"""
+    payload = build_card(status, detail)
+    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(f"通知已发送: {status}")
+        return True
+    except Exception as e:
+        print(f"通知发送失败: {e}", file=sys.stderr)
+        return False
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Staging 部署通知")
+    parser.add_argument(
+        "--status",
+        required=True,
+        choices=["success", "failure", "rollback"],
+        help="部署状态",
+    )
+    parser.add_argument("--detail", default="", help="详情描述")
+    parser.add_argument(
+        "--webhook",
+        default=os.environ.get("CI_NOTIFY_WEBHOOK", ""),
+        help="Webhook URL (也可通过 CI_NOTIFY_WEBHOOK 环境变量设置)",
+    )
+
+    args = parser.parse_args()
+
+    if not args.webhook:
+        print("未配置 webhook URL，跳过通知")
+        return 0
+
+    send_notification(args.webhook, args.status, args.detail)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())