From ed3ab3402811a42a887cfcc3e58445acbf61d667 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 14 Jul 2026 21:34:15 +0800 Subject: [PATCH] =?UTF-8?q?feat(ci):=20=E5=8D=87=E7=BA=A7=E9=83=A8?= =?UTF-8?q?=E7=BD=B2=E9=80=9A=E7=9F=A5=E8=84=9A=E6=9C=AC=EF=BC=8C=E5=8D=A1?= =?UTF-8?q?=E7=89=87=E6=A0=BC=E5=BC=8F=E4=B8=8ECI=E9=80=9A=E7=9F=A5?= =?UTF-8?q?=E5=AF=B9=E9=BD=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/deploy_notify.py | 154 +++++++++++++++++++++++++++++++-------- 1 file changed, 125 insertions(+), 29 deletions(-) diff --git a/scripts/deploy_notify.py b/scripts/deploy_notify.py index 6e8139c89..03a058278 100644 --- a/scripts/deploy_notify.py +++ b/scripts/deploy_notify.py @@ -1,13 +1,17 @@ #!/usr/bin/env python3 """ -Staging 部署通知脚本 +部署通知脚本（支持 Staging / Production） +与 CI 通知（ci_notify_success.py / ci_notify_failure.py）卡片格式对齐。 发送部署结果通知到飞书 webhook（卡片格式）。 + 支持三种状态：success / failure / rollback +支持两种环境：staging / production 用法: - python3 deploy_notify.py --status success --detail "部署成功" --webhook - python3 deploy_notify.py --status rollback --detail "健康检查失败，已回滚" + 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 @@ -19,73 +23,139 @@ import urllib.request STATUS_CONFIG = { "success": { "emoji": "✅", - "title": "Staging 部署成功", + "title_suffix": "部署成功", "color": "green", + "button_text": "查看构建详情", + "button_type": "primary", }, "failure": { "emoji": "❌", - "title": "Staging 部署失败", + "title_suffix": "部署失败", "color": "red", + "button_text": "查看失败日志", + "button_type": "danger", }, "rollback": { "emoji": "↩️", - "title": "Staging 部署已回滚", + "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) -> dict: - """构建飞书卡片消息。""" +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] - branch = os.environ.get("GITHUB_REF_NAME", "unknown") + 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": ( - f"**状态**: {cfg['emoji']} {cfg['title']}\n" - f"**详情**: {detail}\n" - f"**分支**: {branch}\n" - f"**提交**: {commit}\n" - f"**提交者**: {actor}\n" - ), + "content": "\n".join(lines), }, }, ] - # 如果有 run_id，加一个查看详情按钮 + # 查看详情按钮 + run_url = f"https://git.xiaoxiajianji.com/{repo}/actions/runs/{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": "查看构建详情"}, + "text": {"tag": "plain_text", "content": cfg["button_text"]}, "url": run_url, - "type": "primary", + "type": cfg["button_type"], } ], } ) - # 加 Staging 访问链接 + # 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": "Staging: https://staging.xiaoxiajianji.com", + "content": f"Web: {env_cfg['url_web']} | API: {env_cfg['url_api']}", } ], } @@ -97,7 +167,7 @@ def build_card(status: str, detail: str) -> dict: "header": { "title": { "tag": "plain_text", - "content": f"{cfg['emoji']} {cfg['title']}", + "content": title, }, "status": cfg["color"], }, @@ -106,9 +176,17 @@ def build_card(status: str, detail: str) -> dict: } -def send_notification(webhook: str, status: str, detail: str) -> bool: +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) + payload = build_card(status, detail, env, duration, failed_step, pr_url) data = json.dumps(payload).encode("utf-8") req = urllib.request.Request( @@ -121,7 +199,7 @@ def send_notification(webhook: str, status: str, detail: str) -> bool: try: with urllib.request.urlopen(req, timeout=10) as resp: resp.read() - print(f"通知已发送: {status}") + print(f"通知已发送: {env} {status}") return True except Exception as e: print(f"通知发送失败: {e}", file=sys.stderr) @@ -129,7 +207,7 @@ def send_notification(webhook: str, status: str, detail: str) -> bool: def main(): - parser = argparse.ArgumentParser(description="Staging 部署通知") + parser = argparse.ArgumentParser(description="部署通知脚本") parser.add_argument( "--status", required=True, @@ -137,6 +215,15 @@ def main(): 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", ""), @@ -146,10 +233,19 @@ def main(): args = parser.parse_args() if not args.webhook: - print("未配置 webhook URL，跳过通知") + print("未配置 CI_NOTIFY_WEBHOOK，跳过通知") + print("如需启用，请在仓库 Settings -> Secrets and variables -> Actions 中添加 CI_NOTIFY_WEBHOOK") return 0 - send_notification(args.webhook, args.status, args.detail) + 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