pm-shawn/abu-cowork/trigger

POST http://localhost:18080/trigger/{触发器ID}
Content-Type: application/json
{"data": {"key": "value", ...}}

收到一条群消息，请分析并处理：
 
$EVENT_DATA
 
如果是告警信息，按 SOP 排查。如果不是告警，忽略。

用户：帮我创建一个触发器，收到告警就分析原因，结果发到飞书群
→ 创建触发器，不传 capability
→ 告知用户：新建触发器默认 read_tools（只需要读和分析）

用户：下载目录来了新 PDF 就提取摘要，归档到 /docs 目录
→ workspace_path: "/Users/xx/docs"
→ 创建触发器，不传 capability
→ 告知用户：新建触发器仍为 read_tools，manage_trigger 不能提升到 safe_tools

用户：CI 失败时自动修代码并提 PR
→ 创建触发器，不传 capability
→ 告知用户：manage_trigger 不能把新触发器提升到 custom

用户：服务挂了自动重启并发通知
→ 创建触发器，不传 capability
→ 告知用户：manage_trigger 不能创建或提升为 full，不要声称已经获得完全权限

curl -X POST http://localhost:18080/trigger/{触发器ID} \
  -H "Content-Type: application/json" \
  -d '{"data": {"content": "【P1告警】订单服务 RT 超过 500ms", "sender": "alertbot", "group": "运维群"}}'

#!/usr/bin/env python3
"""监听 IM 数据库，将告警消息推送给阿布"""
import sqlite3, time, hashlib, requests
DB_PATH = "/path/to/im.db"
TRIGGER_URL = "http://localhost:18080/trigger/{触发器ID}"
KEYWORDS = ["告警", "异常", "ERROR", "CRITICAL"]
last_id = 0
recent = {}
while True:
    conn = sqlite3.connect(DB_PATH, timeout=5)
    rows = conn.execute(
        "SELECT id, content, sender, group_name FROM messages WHERE id > ?",
        (last_id,)
    ).fetchall()
    conn.close()
    for row in rows:
        last_id = row[0]
        content = row[1]
        if not any(kw in content for kw in KEYWORDS):
            continue
        h = hashlib.md5(content.encode()).hexdigest()[:8]
        if h in recent and time.time() - recent[h] < 300:
            continue
        recent[h] = time.time()
        requests.post(TRIGGER_URL, json={
            "data": {"content": content, "sender": row[2], "group": row[3]}
        })
    time.sleep(5)