158 lines
4.2 KiB
Python
158 lines
4.2 KiB
Python
#!/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())
|