252 lines
8.1 KiB
Python
252 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
CI健康度每日巡检报告脚本
|
|
- 调用ci_health_check.py获取数据
|
|
- 有失败时生成飞书卡片通知并发送
|
|
- 无失败时静默退出（不打扰）
|
|
- 用于每日定时巡检
|
|
|
|
用法:
|
|
python3 scripts/ci/ci_health_report.py [--limit 30] [--dry-run]
|
|
|
|
环境变量:
|
|
GITEA_TOKEN API token（必需）
|
|
CI_NOTIFY_WEBHOOK 飞书webhook地址（必需，用于发报告）
|
|
GITEA_API_URL Gitea API 地址
|
|
GITEA_REPO 仓库
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import urllib.request
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
|
|
def run_health_check(limit: int) -> dict:
|
|
"""调用ci_health_check.py获取JSON结果"""
|
|
script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
cmd = [
|
|
sys.executable,
|
|
os.path.join(script_dir, "ci_health_check.py"),
|
|
"--json",
|
|
"--limit",
|
|
str(limit),
|
|
]
|
|
env = os.environ.copy()
|
|
# 确保GITEA_TOKEN传递
|
|
if not env.get("GITEA_TOKEN") and env.get("GITHUB_TOKEN"):
|
|
env["GITEA_TOKEN"] = env["GITHUB_TOKEN"]
|
|
|
|
result = subprocess.run(cmd, capture_output=True, text=True, env=env)
|
|
if result.returncode != 0:
|
|
print(f"health check failed: {result.stderr}")
|
|
return {"workflows": {}, "failed_runs": []}
|
|
try:
|
|
return json.loads(result.stdout)
|
|
except json.JSONDecodeError:
|
|
print(f"failed to parse health check output: {result.stdout[:200]}")
|
|
return {"workflows": {}, "failed_runs": []}
|
|
|
|
|
|
def build_feishu_card(data: dict) -> dict:
|
|
"""构建飞书卡片消息"""
|
|
wf_stats = data.get("workflows", {})
|
|
failed_runs = data.get("failed_runs", [])
|
|
|
|
# 统计数据
|
|
total_all = sum(s["total"] for s in wf_stats.values())
|
|
succ_all = sum(s["success"] for s in wf_stats.values())
|
|
fail_all = sum(s["failure"] for s in wf_stats.values())
|
|
rate_all = (succ_all / total_all * 100) if total_all > 0 else 0
|
|
|
|
# 失败分类
|
|
infra_fail = 0
|
|
biz_fail = 0
|
|
unknown_fail = 0
|
|
for run in failed_runs:
|
|
for job in run.get("jobs", []):
|
|
cat = job.get("category", "unknown")
|
|
if cat == "infra":
|
|
infra_fail += 1
|
|
elif cat == "business":
|
|
biz_fail += 1
|
|
else:
|
|
unknown_fail += 1
|
|
|
|
now = datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%d %H:%M")
|
|
|
|
# 各workflow成功率行
|
|
wf_lines = []
|
|
for wf, s in sorted(wf_stats.items()):
|
|
total = s["total"]
|
|
succ = s["success"]
|
|
fail = s["failure"]
|
|
rate = (succ / total * 100) if total > 0 else 0
|
|
icon = "🟢" if rate >= 90 else ("🟡" if rate >= 70 else "🔴")
|
|
wf_name = (
|
|
wf.replace("ci-pipeline.yml", "CI Pipeline")
|
|
.replace("code-review.yml", "Code Review")
|
|
.replace("daily-check.yml", "Daily Check")
|
|
.replace("preview-deploy.yml", "Preview Deploy")
|
|
)
|
|
wf_lines.append(f"{icon} **{wf_name}**: {rate:.0f}% ({succ}/{total}，失败{fail})")
|
|
|
|
# 失败详情（最多显示5条）
|
|
fail_detail_lines = []
|
|
for _i, run in enumerate(failed_runs[:5]):
|
|
run_id = run["id"]
|
|
title = run.get("title", "")[:35]
|
|
branch = run.get("branch", "")
|
|
jobs_str = ", ".join(j["name"][:15] for j in run.get("jobs", [])[:3])
|
|
fail_detail_lines.append(f"• **#{run_id}** {title}\n 分支: {branch} | 失败: {jobs_str}")
|
|
|
|
if len(failed_runs) > 5:
|
|
fail_detail_lines.append(f"... 还有 {len(failed_runs) - 5} 条失败记录")
|
|
|
|
# 整体状态
|
|
if fail_all == 0:
|
|
status_text = "✅ 全部通过"
|
|
status_color = "green"
|
|
elif infra_fail > biz_fail:
|
|
status_text = "⚠️ 基础设施问题为主"
|
|
status_color = "yellow"
|
|
else:
|
|
status_text = "🔴 存在业务失败"
|
|
status_color = "red"
|
|
|
|
card = {
|
|
"config": {"wide_screen_mode": True},
|
|
"header": {
|
|
"title": {"tag": "plain_text", "content": f"CI告警 - 每日健康度巡检 ({now})"},
|
|
"template": status_color,
|
|
},
|
|
"elements": [
|
|
{
|
|
"tag": "div",
|
|
"text": {
|
|
"tag": "lark_md",
|
|
"content": f"**统计范围**: 最近 {total_all} 条 run\n**整体状态**: {status_text}\n**总成功率**: {rate_all:.1f}% ({succ_all}/{total_all})",
|
|
},
|
|
},
|
|
{"tag": "hr"},
|
|
{
|
|
"tag": "div",
|
|
"text": {
|
|
"tag": "lark_md",
|
|
"content": "**📊 各Workflow成功率**\n" + "\n".join(wf_lines) if wf_lines else "暂无数据",
|
|
},
|
|
},
|
|
],
|
|
}
|
|
|
|
# 失败分类统计
|
|
if fail_all > 0:
|
|
card["elements"].append({"tag": "hr"})
|
|
card["elements"].append(
|
|
{
|
|
"tag": "div",
|
|
"text": {
|
|
"tag": "lark_md",
|
|
"content": f"**失败原因分类**\n🏗️ 基础设施: {infra_fail} 个\n🐛 业务代码: {biz_fail} 个\n❓ 待确认: {unknown_fail} 个",
|
|
},
|
|
}
|
|
)
|
|
|
|
# 失败详情
|
|
if fail_detail_lines:
|
|
card["elements"].append({"tag": "hr"})
|
|
card["elements"].append(
|
|
{
|
|
"tag": "div",
|
|
"text": {
|
|
"tag": "lark_md",
|
|
"content": "**❌ 失败详情**\n" + "\n\n".join(fail_detail_lines),
|
|
},
|
|
}
|
|
)
|
|
|
|
# 查看更多
|
|
card["elements"].append({"tag": "hr"})
|
|
base_url = os.environ.get("GITEA_BASE_URL", "https://git.xiaoxiajianji.com")
|
|
repo = os.environ.get("GITEA_REPO", "xiaoxia/xiaoxia-saas")
|
|
card["elements"].append(
|
|
{
|
|
"tag": "action",
|
|
"actions": [
|
|
{
|
|
"tag": "button",
|
|
"text": {"tag": "plain_text", "content": "查看CI面板"},
|
|
"type": "primary",
|
|
"url": f"{base_url}/{repo}/actions",
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
return {"msg_type": "interactive", "card": card}
|
|
|
|
|
|
def send_feishu(webhook: str, payload: dict) -> bool:
|
|
"""发送飞书webhook"""
|
|
data = json.dumps(payload).encode()
|
|
req = urllib.request.Request(
|
|
webhook,
|
|
data=data,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
result = json.loads(resp.read().decode())
|
|
return result.get("code", -1) == 0 or result.get("StatusCode", -1) == 0
|
|
except Exception as e:
|
|
print(f"send feishu failed: {e}")
|
|
return False
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="CI健康度每日巡检报告")
|
|
parser.add_argument("--limit", type=int, default=30, help="统计最近N条run")
|
|
parser.add_argument("--dry-run", action="store_true", help="只打印不发送")
|
|
parser.add_argument("--always-notify", action="store_true", help="即使全部通过也发送通知")
|
|
args = parser.parse_args()
|
|
|
|
webhook = os.environ.get("CI_NOTIFY_WEBHOOK", "")
|
|
if not webhook and not args.dry_run:
|
|
print("未配置 CI_NOTIFY_WEBHOOK，跳过通知")
|
|
# 还是执行健康检查输出到日志，方便排查
|
|
data = run_health_check(args.limit)
|
|
print(f"health check done: {len(data.get('failed_runs', []))} failed")
|
|
return 0
|
|
|
|
# 执行健康检查
|
|
data = run_health_check(args.limit)
|
|
failed_count = len(data.get("failed_runs", []))
|
|
|
|
# 无失败且不强制通知 → 静默退出
|
|
if failed_count == 0 and not args.always_notify:
|
|
print("✅ 全部通过，静默退出")
|
|
return 0
|
|
|
|
# 构建并发送卡片
|
|
card = build_feishu_card(data)
|
|
|
|
if args.dry_run:
|
|
print(json.dumps(card, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
success = send_feishu(webhook, card)
|
|
if success:
|
|
print(f"📤 已发送健康度报告，失败 {failed_count} 条")
|
|
else:
|
|
print("❌ 发送飞书通知失败")
|
|
|
|
# 通知失败不阻断流程
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|