xiaoxia-saas/scripts/ci/chatops/webhook_server.py

#!/usr/bin/env python3
"""
Gitea Webhook 接收服务 - FastAPI 实现

功能:
  1. 接收 Gitea Actions webhook 事件，触发飞书通知
  2. 接收飞书机器人回调消息，处理 /ci 交互命令
  3. 维护简单的状态缓存，检测分支恢复等状态变化

部署:
  部署到构建服务器，监听 8090 端口（可配置）
  Gitea webhook 指向: http://<server>:8090/webhook/gitea
  飞书消息回调指向: http://<server>:8090/webhook/feishu

依赖:
  fastapi + uvicorn（可选，未安装时仅模块可用，服务不可启动）

注意:
  本文件为第一版骨架，通知逻辑已实现，飞书交互命令待后续完善。
"""

import hashlib
import hmac
import json
import sys
import threading
import time
from typing import Optional

from . import config
from .gitea_client import GiteaClient

# FastAPI 是可选依赖，未安装时仅导出类不启动服务
try:
    from fastapi import FastAPI, Header, HTTPException, Request
    from fastapi.responses import JSONResponse

    FASTAPI_AVAILABLE = True
except ImportError:
    FASTAPI_AVAILABLE = False
    FastAPI = None  # type: ignore


# ── 状态缓存 ──────────────────────────────────────────


class StateCache:
    """简单的内存状态缓存，用于检测状态变化

    记录每个分支最后一次 run 的状态，用于判断:
    - 是否从失败变成功（恢复通知）
    - 是否连续失败（避免重复告警）
    """

    def __init__(self, max_entries=100):
        self._cache = {}  # {branch: {last_status, last_run_id, last_notified_failure}}
        self._lock = threading.Lock()
        self._max = max_entries

    def get(self, key):
        with self._lock:
            return self._cache.get(key)

    def set(self, key, value):
        with self._lock:
            self._cache[key] = value
            # 简单的淘汰策略
            if len(self._cache) > self._max:
                oldest_key = next(iter(self._cache))
                del self._cache[oldest_key]

    def check_and_update(self, branch, run_id, conclusion):
        """检查状态变化并更新缓存

        Returns:
            dict: {is_new_failure, is_recovery, previous_status, previous_run_id}
        """
        prev = self.get(branch) or {}
        prev_status = prev.get("last_status", "unknown")
        prev_run_id = prev.get("last_run_id")

        is_new_failure = False
        is_recovery = False

        if conclusion == "failure" and prev_status != "failure":
            is_new_failure = True
        if conclusion == "success" and prev_status == "failure":
            is_recovery = True

        self.set(
            branch,
            {
                "last_status": conclusion,
                "last_run_id": run_id,
                "last_updated": time.time(),
                "last_notified_failure": run_id if is_new_failure else prev.get("last_notified_failure"),
            },
        )

        return {
            "is_new_failure": is_new_failure,
            "is_recovery": is_recovery,
            "previous_status": prev_status,
            "previous_run_id": prev_run_id,
        }


state_cache = StateCache()


# ── Gitea Webhook 处理 ────────────────────────────────


def verify_gitea_signature(payload: bytes, signature: str) -> bool:
    """校验 Gitea webhook 签名（X-Gitea-Signature）

    Gitea 使用 HMAC-SHA256 签名，格式: sha256=xxx
    """
    if not config.WEBHOOK_SECRET:
        return True  # 未配置密钥则跳过校验

    if not signature:
        return False

    try:
        algo, sig_hex = signature.split("=", 1)
        if algo != "sha256":
            return False
        expected = hmac.new(config.WEBHOOK_SECRET.encode(), payload, hashlib.sha256).hexdigest()
        return hmac.compare_digest(expected, sig_hex)
    except Exception:
        return False


def handle_gitea_webhook(payload: dict, event_type: str) -> dict:
    """处理 Gitea webhook 事件

    Args:
        payload: webhook 请求体
        event_type: X-Gitea-Event 头

    Returns:
        dict: {handled, notifications_sent, message}
    """
    if event_type != "create" and event_type != "push":
        # 我们主要关心 push 和 actions 事件
        # Gitea Actions 的 webhook 事件类型可能是 "push" 或专门的 actions 事件
        pass

    # 尝试提取 run 信息
    run_info = _extract_run_info(payload)
    if not run_info:
        return {"handled": False, "notifications_sent": 0, "message": "非 CI 事件，跳过"}

    branch = run_info["branch"]
    run_id = run_info["run_id"]
    status = run_info["status"]
    conclusion = run_info.get("conclusion", "")

    # 只处理已完成的 run
    if status != "completed":
        return {"handled": True, "notifications_sent": 0, "message": f"Run {run_id} 仍在运行中 ({status})"}

    # 检查是否在通知分支列表中
    if branch not in config.NOTIFY_BRANCHES:
        return {
            "handled": True,
            "notifications_sent": 0,
            "message": f"分支 {branch} 不在通知列表中",
        }

    # 检查状态变化
    change_info = state_cache.check_and_update(branch, run_id, conclusion)
    notifications = 0

    # 延迟导入，避免循环依赖
    from .feishu_notify import FeishuNotifier

    notifier = FeishuNotifier()

    if conclusion == "failure" and change_info["is_new_failure"]:
        # 新失败 → 发失败通知
        notifier.notify_branch_failure(run_id, branch)
        notifications += 1

        # 检查是否是 E2E 失败
        gitea = GiteaClient()
        failed_jobs = gitea.get_failed_jobs_summary(run_id)
        has_e2e = any("e2e" in j["name"].lower() for j in failed_jobs)
        if has_e2e:
            notifier.notify_e2e_failure(run_id, branch=branch)
            notifications += 1

    elif conclusion == "success" and change_info["is_recovery"]:
        # 从失败恢复 → 发恢复通知
        prev_run_id = change_info.get("previous_run_id")
        notifier.notify_branch_recovery(run_id, branch, previous_failure_run_id=prev_run_id)
        notifications += 1

    return {
        "handled": True,
        "notifications_sent": notifications,
        "message": f"分支 {branch} run {run_id} {conclusion}",
    }


def _extract_run_info(payload: dict) -> Optional[dict]:
    """从 webhook payload 中提取 run 信息

    Gitea Actions webhook 的 payload 结构可能不同，这里做兼容处理。
    如果 payload 不是 run 事件，返回 None。
    """
    # 尝试多种可能的结构
    if "workflow_run" in payload:
        wr = payload["workflow_run"]
        return {
            "run_id": wr.get("id"),
            "branch": wr.get("head_branch", ""),
            "status": wr.get("status", ""),
            "conclusion": wr.get("conclusion", ""),
            "name": wr.get("name", ""),
        }

    if "action" in payload and "pull_request" in payload:
        # PR 事件，暂不处理
        return None

    if "ref" in payload and "head_commit" in payload:
        # push 事件，不是 run 事件
        return None

    return None


# ── 飞书消息处理 ──────────────────────────────────────


def handle_feishu_message(payload: dict) -> dict:
    """处理飞书机器人回调消息

    支持命令:
      /ci status [branch] - 查询分支 CI 状态
      /ci rerun <run-id>  - 重跑失败的 jobs
      /ci help            - 帮助

    注意: 第一版骨架，仅解析命令，实际执行逻辑待完善。
    """
    # 飞书消息回调格式
    header = payload.get("header", {})
    event_type = header.get("event_type", "")

    if event_type == "url_verification":
        # 飞书 URL 验证
        return {"challenge": payload.get("challenge", "")}

    if event_type != "im.message.receive_v1":
        return {"handled": False, "message": f"非消息事件: {event_type}"}

    event = payload.get("event", {})
    message = event.get("message", {})
    content_str = message.get("content", "{}")

    try:
        content = json.loads(content_str)
    except json.JSONDecodeError:
        content = {}

    text = content.get("text", "")
    if not text:
        return {"handled": False, "message": "空消息"}

    # 解析命令
    text = text.strip()
    if not text.startswith("/ci"):
        return {"handled": False, "message": "非 CI 命令"}

    parts = text.split()
    if len(parts) < 2:
        return _help_response()

    cmd = parts[1].lower()

    if cmd == "status":
        branch = parts[2] if len(parts) > 2 else "develop"
        return _handle_status_command(branch)

    elif cmd == "rerun":
        if len(parts) < 3:
            return {"text": "用法: /ci rerun <run-id> 或 /ci rerun latest [branch]"}
        arg = parts[2]
        if arg == "latest":
            branch = parts[3] if len(parts) > 3 else "develop"
            return _handle_rerun_latest(branch)
        return _handle_rerun_command(arg)

    elif cmd == "help":
        return _help_response()

    else:
        return {"text": f"未知命令: {cmd}\n输入 /ci help 查看帮助"}


def _handle_status_command(branch: str) -> dict:
    """处理 /ci status 命令"""
    from .ci_query import CIQuery

    query = CIQuery()
    result = query.get_branch_status(branch)
    reply = CIQuery.format_branch_status(result)
    return {"text": reply}


def _handle_rerun_command(run_id: str) -> dict:
    """处理 /ci rerun 命令"""
    from .ci_trigger import CITrigger

    trigger = CITrigger()
    try:
        result = trigger.rerun_failed(int(run_id))
    except ValueError:
        return {"text": f"无效的 run id: {run_id}"}

    if result["success"]:
        return {"text": f"✅ {result['message']}\n{result.get('run_url', '')}"}
    return {"text": f"❌ {result['message']}"}


def _handle_rerun_latest(branch: str) -> dict:
    """处理 /ci rerun latest 命令"""
    from .ci_trigger import CITrigger

    trigger = CITrigger()
    result = trigger.rerun_latest_failed(branch=branch)

    if result["success"]:
        return {"text": f"✅ {result['message']}\n{result.get('run_url', '')}"}
    return {"text": f"❌ {result['message']}"}


def _help_response() -> dict:
    """返回帮助信息"""
    help_text = """**CI ChatOps 命令帮助**

`/ci status [branch]`  查询分支 CI 状态（默认 develop）
`/ci rerun <run-id>`   重跑指定 run 的失败 jobs
`/ci rerun latest [branch]`  重跑分支最近一次失败的 run
`/ci help`             显示此帮助

**环境变量配置:**
  `GITEA_TOKEN` / `GITEA_USERNAME + GITEA_PASSWORD`
  `FEISHU_WEBHOOK_URL`
  `CHATOPS_NOTIFY_BRANCHES=main,develop`
"""
    return {"text": help_text}


# ── FastAPI 应用 ──────────────────────────────────────


def create_app():
    """创建 FastAPI 应用

    如果 FastAPI 未安装，返回 None
    """
    if not FASTAPI_AVAILABLE:
        print(
            "[WARN] FastAPI 未安装，无法启动 webhook 服务。" " 请运行: pip install fastapi uvicorn",
            file=sys.stderr,
        )
        return None

    app = FastAPI(title="CI ChatOps Webhook", version="0.1.0")

    @app.post("/webhook/gitea")
    async def gitea_webhook(
        request: Request,
        x_gitea_event: str = Header(default=""),
        x_gitea_signature: str = Header(default=""),
    ):
        body = await request.body()

        # 签名校验
        if not verify_gitea_signature(body, x_gitea_signature):
            raise HTTPException(status_code=401, detail="Invalid signature")

        try:
            payload = json.loads(body.decode())
        except json.JSONDecodeError as e:
            raise HTTPException(status_code=400, detail="Invalid JSON") from e

        result = handle_gitea_webhook(payload, x_gitea_event)
        return JSONResponse(content=result)

    @app.post("/webhook/feishu")
    async def feishu_webhook(request: Request):
        body = await request.body()
        try:
            payload = json.loads(body.decode())
        except json.JSONDecodeError as e:
            raise HTTPException(status_code=400, detail="Invalid JSON") from e

        result = handle_feishu_message(payload)
        return JSONResponse(content=result)

    @app.get("/health")
    async def health():
        return {"status": "ok", "service": "ci-chatops"}

    return app


# ── CLI 入口 ──────────────────────────────────────────


def main():
    """启动 webhook 服务"""
    import argparse

    parser = argparse.ArgumentParser(description="CI ChatOps Webhook 服务")
    parser.add_argument("--port", type=int, default=config.CHATOPS_WEBHOOK_PORT, help="监听端口")
    parser.add_argument("--host", default="0.0.0.0", help="监听地址")
    args = parser.parse_args()

    app = create_app()
    if not app:
        print("[ERROR] FastAPI 不可用，请先安装: pip install fastapi uvicorn")
        return 1

    try:
        import uvicorn
    except ImportError:
        print("[ERROR] uvicorn 未安装，请先安装: pip install uvicorn")
        return 1

    print(f"[INFO] CI ChatOps Webhook 服务启动: http://{args.host}:{args.port}")
    print("[INFO] Gitea webhook:   POST /webhook/gitea")
    print("[INFO] 飞书 webhook:    POST /webhook/feishu")
    print("[INFO] 健康检查:        GET  /health")
    print(f"[INFO] 通知分支:        {', '.join(config.NOTIFY_BRANCHES)}")

    uvicorn.run(app, host=args.host, port=args.port)
    return 0


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