xiaoxia-saas/scripts/deploy_notify.py

#!/usr/bin/env python3
"""
部署通知脚本（支持 Staging / Production）

与 CI 通知（ci_notify_success.py / ci_notify_failure.py）卡片格式对齐。
发送部署结果通知到飞书 webhook（卡片格式）。

支持三种状态：success / failure / rollback
支持两种环境：staging / production

用法:
    python3 deploy_notify.py --status success --detail "部署成功" --env staging
    python3 deploy_notify.py --status rollback --detail "健康检查失败，已回滚" --env production
    python3 deploy_notify.py --status failure --detail "部署过程出错" --env production --failed-step "构建镜像"
"""

import argparse
import json
import os
import sys
import urllib.request

STATUS_CONFIG = {
    "success": {
        "emoji": "✅",
        "title_suffix": "部署成功",
        "color": "green",
        "button_text": "查看构建详情",
        "button_type": "primary",
    },
    "failure": {
        "emoji": "❌",
        "title_suffix": "部署失败",
        "color": "red",
        "button_text": "查看失败日志",
        "button_type": "danger",
    },
    "rollback": {
        "emoji": "↩️",
        "title_suffix": "部署已回滚",
        "color": "yellow",
        "button_text": "查看构建详情",
        "button_type": "primary",
    },
}

ENV_CONFIG = {
    "staging": {
        "label": "Staging",
        "url_web": "https://staging.xiaoxiajianji.com",
        "url_api": "https://staging-api.xiaoxiajianji.com",
    },
    "production": {
        "label": "Production",
        "url_web": "https://saas.xiaoxiajianji.com",
        "url_api": "https://api.xiaoxiajianji.com",
    },
}


def build_card(
    status: str,
    detail: str,
    env: str = "staging",
    duration: str = "",
    failed_step: str = "",
    pr_url: str = "",
) -> dict:
    """构建飞书卡片消息（与 ci_notify_*.py 风格一致）。"""
    cfg = STATUS_CONFIG.get(status, STATUS_CONFIG["failure"])
    env_cfg = ENV_CONFIG.get(env, ENV_CONFIG["staging"])
    title = f"{cfg['emoji']} {env_cfg['label']} {cfg['title_suffix']}"

    commit = os.environ.get("GITHUB_SHA", "unknown")[:8]
    ref = 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")

    # 版本信息：tag 部署显示 tag，分支部署显示分支
    if ref.startswith("v"):
        version_info = f"版本 {ref}"
    else:
        version_info = ref

    # 构建内容行（与 ci_notify_*.py 风格一致：**标签**: 值）
    lines = []

    # 详情行（部署特有）
    if detail:
        lines.append(f"**详情**: {detail}")

    lines.append(f"**环境**: {env_cfg['label']}")
    lines.append(f"**版本**: {version_info}")

    # 失败阶段（失败/回滚时显示）
    if failed_step and status in ("failure", "rollback"):
        lines.append(f"**失败阶段**: {failed_step}")

    # 耗时（可选）
    if duration:
        lines.append(f"**耗时**: {duration}")

    lines.append(f"**提交**: {commit}")
    lines.append(f"**提交者**: {actor}")
    lines.append(f"**Run ID**: {run_id}")

    elements = [
        {
            "tag": "div",
            "text": {
                "tag": "lark_md",
                "content": "\n".join(lines),
            },
        },
    ]

    # 查看详情按钮
    run_url = f"https://git.xiaoxiajianji.com/{repo}/actions/runs/{run_id}"
    if run_id and run_id != "-":
        elements.append(
            {
                "tag": "action",
                "actions": [
                    {
                        "tag": "button",
                        "text": {"tag": "plain_text", "content": cfg["button_text"]},
                        "url": run_url,
                        "type": cfg["button_type"],
                    }
                ],
            }
        )

    # PR 链接（可选）
    if pr_url:
        elements.append(
            {
                "tag": "action",
                "actions": [
                    {
                        "tag": "button",
                        "text": {"tag": "plain_text", "content": "查看 PR"},
                        "url": pr_url,
                        "type": "default",
                    }
                ],
            }
        )

    # 访问地址（部署特有）
    elements.append(
        {
            "tag": "note",
            "elements": [
                {
                    "tag": "plain_text",
                    "content": f"Web: {env_cfg['url_web']}  |  API: {env_cfg['url_api']}",
                }
            ],
        }
    )

    return {
        "msg_type": "interactive",
        "card": {
            "header": {
                "title": {
                    "tag": "plain_text",
                    "content": title,
                },
                "status": cfg["color"],
            },
            "elements": elements,
        },
    }


def send_notification(
    webhook: str,
    status: str,
    detail: str,
    env: str = "staging",
    duration: str = "",
    failed_step: str = "",
    pr_url: str = "",
) -> bool:
    """发送通知到 webhook。"""
    payload = build_card(status, detail, env, duration, failed_step, pr_url)
    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"通知已发送: {env} {status}")
        return True
    except Exception as e:
        print(f"通知发送失败: {e}", file=sys.stderr)
        return False


def main():
    parser = argparse.ArgumentParser(description="部署通知脚本")
    parser.add_argument(
        "--status",
        required=True,
        choices=["success", "failure", "rollback"],
        help="部署状态",
    )
    parser.add_argument("--detail", default="", help="详情描述")
    parser.add_argument(
        "--env",
        default="staging",
        choices=["staging", "production"],
        help="部署环境 (默认 staging)",
    )
    parser.add_argument("--duration", default="", help="部署耗时")
    parser.add_argument("--failed-step", default="", help="失败阶段")
    parser.add_argument("--pr-url", default="", help="PR 链接")
    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("未配置 CI_NOTIFY_WEBHOOK，跳过通知")
        print("如需启用，请在仓库 Settings -> Secrets and variables -> Actions 中添加 CI_NOTIFY_WEBHOOK")
        return 0

    send_notification(
        webhook=args.webhook,
        status=args.status,
        detail=args.detail,
        env=args.env,
        duration=args.duration,
        failed_step=args.failed_step,
        pr_url=args.pr_url,
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())