"""Webhook 通知服务。 事件类型: - chat.completed 聊天成功 - chat.failed 聊天失败 - file.uploaded 文件上传 - config.updated 配置更新 - session.created 会话创建 - session.deleted 会话删除 配置存 SQLite webhook_config 表;触发时异步 httpx POST, 失败不影响主流程。事件用 events 字段(逗号分隔)过滤。 """ from __future__ import annotations import asyncio import json import time from typing import Any, Optional from ..database import get_conn from ..http_client import get_http_client # ===== 配置 CRUD ===== def _now_ts() -> int: return int(time.time()) def list_webhooks() -> list[dict]: with get_conn() as conn: rows = conn.execute( "SELECT id, name, url, events, enabled, created_at FROM webhook_config " "ORDER BY id ASC" ).fetchall() out = [] for r in rows: out.append({ "id": int(r["id"]), "name": r["name"], "url": r["url"], "events": [e for e in (r["events"] or "").split(",") if e], "enabled": bool(r["enabled"]), "created_at": int(r["created_at"]), }) return out def create_webhook(*, name: str, url: str, events: list[str], enabled: bool = True) -> dict: events_str = ",".join([e for e in (events or []) if e]) now = _now_ts() with get_conn() as conn: cur = conn.execute( "INSERT INTO webhook_config(name, url, events, enabled, created_at) " "VALUES(?,?,?,?,?)", (name, url, events_str, 1 if enabled else 0, now), ) wid = int(cur.lastrowid) return {"id": wid, "name": name, "url": url, "events": events or [], "enabled": enabled, "created_at": now} def update_webhook( webhook_id: int, *, name: Optional[str] = None, url: Optional[str] = None, events: Optional[list[str]] = None, enabled: Optional[bool] = None, ) -> Optional[dict]: sets: list[str] = [] args: list = [] if name is not None: sets.append("name = ?") args.append(name) if url is not None: sets.append("url = ?") args.append(url) if events is not None: sets.append("events = ?") args.append(",".join([e for e in events if e])) if enabled is not None: sets.append("enabled = ?") args.append(1 if enabled else 0) if not sets: return get_webhook(webhook_id) args.append(webhook_id) with get_conn() as conn: cur = conn.execute( f"UPDATE webhook_config SET {', '.join(sets)} WHERE id = ?", tuple(args), ) if cur.rowcount == 0: return None return get_webhook(webhook_id) def get_webhook(webhook_id: int) -> Optional[dict]: with get_conn() as conn: row = conn.execute( "SELECT id, name, url, events, enabled, created_at FROM webhook_config WHERE id = ?", (webhook_id,), ).fetchone() if not row: return None return { "id": int(row["id"]), "name": row["name"], "url": row["url"], "events": [e for e in (row["events"] or "").split(",") if e], "enabled": bool(row["enabled"]), "created_at": int(row["created_at"]), } def delete_webhook(webhook_id: int) -> bool: with get_conn() as conn: cur = conn.execute("DELETE FROM webhook_config WHERE id = ?", (webhook_id,)) return cur.rowcount > 0 # ===== 事件触发 ===== def _matching_webhooks(event: str) -> list[dict]: """返回订阅了该事件且启用的 webhook 列表。""" with get_conn() as conn: rows = conn.execute( "SELECT id, name, url, events FROM webhook_config WHERE enabled = 1" ).fetchall() out = [] for r in rows: events = [e for e in (r["events"] or "").split(",") if e] if event in events or "*" in events: out.append({"id": int(r["id"]), "name": r["name"], "url": r["url"]}) return out async def notify(event: str, payload: Any) -> None: """异步通知所有订阅了 event 的 webhook。失败静默。""" targets = _matching_webhooks(event) if not targets: return body = json.dumps( {"event": event, "ts": _now_ts(), "data": payload}, ensure_ascii=False, ).encode("utf-8") client = get_http_client() for t in targets: try: resp = await client.post( t["url"], content=body, headers={ "Content-Type": "application/json", "X-XTC-Event": event, "X-XTC-Webhook-Id": str(t["id"]), "User-Agent": "xtc-backend-hf-webhook/1.0", }, timeout=10.0, ) if resp.status_code >= 400: print(f"[webhook] {t['name']} ({t['url']}) returned {resp.status_code}") except Exception as e: print(f"[webhook] {t['name']} ({t['url']}) failed: {e}") def notify_fire_and_forget(event: str, payload: Any) -> None: """同步上下文用:把通知丢到事件循环后台跑,不阻塞主流程。""" try: loop = asyncio.get_event_loop() except RuntimeError: return loop.create_task(notify(event, payload))