xiaoxia-saas/scripts/ci_trigger_monitor.py

#!/usr/bin/env python3
"""
CI触发可靠性监控 - 定时检查PR的CI触发状态
- 监控open PR的最新commit是否在5分钟内触发了CI
- 异常时通过飞书webhook告警

环境变量:
  GITEA_API_TOKEN - Gitea API Token (必填)
  GITEA_REPO - 仓库路径，如 xiaoxia/xiaoxia-saas
  GITEA_URL - Gitea地址，如 https://git.xiaoxiajianji.com
  CI_NOTIFY_WEBHOOK - 飞书告警webhook (必填)
  CHECK_INTERVAL_MIN - 检查间隔（分钟），默认5
  STALE_THRESHOLD_MIN - CI未触发告警阈值（分钟），默认5
"""

import json
import os
import sys
import time
import urllib.error
import urllib.request


def get_env(name, default=""):
    return os.environ.get(name, default)


def api_get(path):
    """调用Gitea API"""
    token = get_env("GITEA_API_TOKEN")
    base_url = get_env("GITEA_URL", "https://git.xiaoxiajianji.com")
    repo = get_env("GITEA_REPO", "xiaoxia/xiaoxia-saas")

    url = f"{base_url}/api/v1/repos/{repo}{path}"
    req = urllib.request.Request(url)
    req.add_header("Authorization", f"token {token}")

    for attempt in range(3):
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                return json.loads(resp.read())
        except urllib.error.HTTPError as e:
            if e.code >= 500 and attempt < 2:
                time.sleep(2**attempt)
                continue
            raise
        except Exception:
            if attempt < 2:
                time.sleep(2**attempt)
                continue
            raise


def get_open_prs():
    """获取所有open PR"""
    prs = []
    page = 1
    while True:
        batch = api_get(f"/pulls?state=open&sort=updated&direction=desc&limit=50&page={page}")
        if not batch:
            break
        prs.extend(batch)
        if len(batch) < 50:
            break
        page += 1
    return prs


def get_commit_status(sha):
    """获取commit的CI状态"""
    try:
        return api_get(f"/commits/{sha}/status")
    except Exception as e:
        print(f"  ⚠️  获取commit状态失败: {e}")
        return {"state": "error", "statuses": []}


def has_ci_started(statuses):
    """判断是否有CI job已经启动（pending/running/success/failure都算启动了）"""
    pr_statuses = [s for s in statuses if "pull_request" in s.get("context", "")]
    if not pr_statuses:
        return False
    # 只要有非pending且非空的状态，就算启动了
    for s in pr_statuses:
        if s.get("status") in ["success", "failure", "running"]:
            return True
        if s.get("status") == "pending" and "Has started running" in s.get("description", ""):
            return True
    # 全是"Blocked by required conditions"的pending也算（说明CI系统收到了事件）
    for s in pr_statuses:
        if "Blocked" in s.get("description", ""):
            return True
    return False


def send_alert(pr_num, pr_title, pr_url, head_sha, commit_age_min):
    """发送飞书告警"""
    webhook = get_env("CI_NOTIFY_WEBHOOK")
    if not webhook:
        print("  ⚠️  未配置CI_NOTIFY_WEBHOOK，跳过告警")
        return

    get_env("GITEA_URL", "https://git.xiaoxiajianji.com")

    content = {
        "msg_type": "interactive",
        "card": {
            "header": {
                "title": {"tag": "plain_text", "content": f"⚠️ CI告警 - PR#{pr_num} CI未触发"},
                "template": "red",
            },
            "elements": [
                {
                    "tag": "div",
                    "text": {
                        "tag": "lark_md",
                        "content": f"**PR**: [{pr_title}]({pr_url})\n**最新commit**: `{head_sha[:12]}`\n**已等待**: {commit_age_min:.0f} 分钟仍无CI启动\n**可能原因**: Gitea Actions事件丢失 / Webhook失败 / Runner资源不足",
                    },
                },
                {
                    "tag": "action",
                    "actions": [
                        {
                            "tag": "button",
                            "text": {"tag": "plain_text", "content": "查看PR"},
                            "url": pr_url,
                            "type": "primary",
                        },
                        {
                            "tag": "button",
                            "text": {"tag": "plain_text", "content": "查看Actions"},
                            "url": f"{pr_url}/files",
                            "type": "default",
                        },
                    ],
                },
                {
                    "tag": "note",
                    "elements": [
                        {"tag": "plain_text", "content": f"CI触发监控 | 检测时间: {time.strftime('%Y-%m-%d %H:%M:%S')}"}
                    ],
                },
            ],
        },
    }

    try:
        data = json.dumps(content).encode()
        req = urllib.request.Request(webhook, data=data, method="POST")
        req.add_header("Content-Type", "application/json")
        with urllib.request.urlopen(req, timeout=10) as resp:
            resp.read()
        print(f"  📢 告警已发送: PR#{pr_num}")
    except Exception as e:
        print(f"  ⚠️  告警发送失败: {e}")


def main():
    stale_threshold = int(get_env("STALE_THRESHOLD_MIN", "5"))

    print("=" * 60)
    print(f"CI触发监控 - 检测时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"告警阈值: {stale_threshold}分钟无CI启动")
    print("=" * 60)

    # 获取open PR列表
    try:
        prs = get_open_prs()
    except Exception as e:
        print(f"❌ 获取PR列表失败: {e}")
        sys.exit(0)  # 告警脚本不阻断CI

    print(f"\n共 {len(prs)} 个open PR\n")

    stale_prs = []
    now = time.time()

    for pr in prs:
        pr_num = pr["number"]
        pr_title = pr["title"]
        pr_url = pr["html_url"]
        head_sha = pr["head"]["sha"]
        updated_at = pr["updated_at"]

        # 解析updated_at（ISO格式）
        try:
            # 2026-07-17T09:22:43+08:00
            from datetime import datetime

            # 简化处理：直接用字符串解析
            ts_str = updated_at.replace("Z", "+00:00")
            # 手动解析
            dt = datetime.fromisoformat(ts_str)
            commit_time = dt.timestamp()
        except Exception as e:
            print(f"  ⚠️  PR#{pr_num} 时间解析失败: {e}")
            continue

        age_min = (now - commit_time) / 60

        print(f"PR#{pr_num:3d} | {pr_title[:45]:45s} | 更新于 {age_min:.0f}min前")

        # 少于2分钟的跳过，给CI一点启动时间
        if age_min < 2:
            print("  ⏳ 刚更新，等待CI启动...")
            continue

        # 获取commit状态
        status = get_commit_status(head_sha)
        statuses = status.get("statuses", [])

        if has_ci_started(statuses):
            print(f"  ✅ CI已启动 (state={status.get('state')})")
            continue

        # CI未启动，判断是否超过阈值
        if age_min >= stale_threshold:
            print(f"  🚨 CI未触发！已等待 {age_min:.0f} 分钟")
            stale_prs.append({"num": pr_num, "title": pr_title, "url": pr_url, "sha": head_sha, "age_min": age_min})
        else:
            print(f"  ⏳ CI尚未启动 ({age_min:.0f}min < {stale_threshold}min阈值)")

    # 发送告警
    print(f"\n{'=' * 60}")
    print(f"检测结果: {len(stale_prs)} 个PR CI未触发超过阈值")

    if stale_prs:
        print("\n告警列表:")
        for pr in stale_prs:
            print(f"  - PR#{pr['num']}: {pr['title'][:40]} ({pr['age_min']:.0f}min)")
            send_alert(pr["num"], pr["title"], pr["url"], pr["sha"], pr["age_min"])
    else:
        print("✅ 所有PR CI触发正常")

    print("=" * 60)


if __name__ == "__main__":
    main()