MisakaNet/hub/sync/feishu_notifier.py at main · ghzhost/MisakaNet · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
"""
Feishu Notifier - 飞书通知模块
当仲裁案例创建时，通过飞书机器人通知用户
"""
import requests
import json
from typing import Optional


class FeishuNotifier:
    """
    飞书通知器
    使用飞书机器人 webhook 发送通知
    """

    def __init__(self, webhook_url: str = None):
        self.webhook_url = webhook_url

    def notify_arbitration_case(self, case: dict, feishu_webhook: str = None) -> bool:
        """
        发送仲裁案例通知到飞书（卡片+按钮）
        case: 包含 id, skill_name, versions 等字段
        """
        webhook = feishu_webhook or self.webhook_url
        if not webhook:
            print("[Feishu] 未配置 webhook，跳过通知")
            return False

        versions_text = ""
        buttons = []
        for i, v in enumerate(case.get("versions", [])):
            desc = v.get("description", "")[:50]
            source = v.get("source", "unknown")
            versions_text += f"• **v{i+1}**: {desc}... (来源: {source})\n"
            version_id = v.get("id", f"v{i+1}")
            buttons.append({
                "tag": "button",
                "text": {"tag": "plain_text", "content": f"选择 v{i+1}"},
                "type": "primary",
                "value": {"action": "arbitrate", "case_id": case.get("id", ""), "winner_id": version_id}
            })

        payload = {
            "msg_type": "interactive",
            "card": {
                "header": {
                    "title": {"tag": "plain_text", "content": f"⚖️ 仲裁请求: {case.get('skill_name', 'Unknown')}"},
                    "template": "red"
                },
                "elements": [
                    {
                        "tag": "markdown",
                        "content": f"**案例 ID**: `{case.get('id', '')}`\n\n**候选版本**:\n{versions_text}"
                    },
                    {"tag": "hr"},
                    {
                        "tag": "action",
                        "actions": buttons
                    },
                    {
                        "tag": "markdown",
                        "content": "或回复: `选择 v1` / `保留 xxx`"
                    },
                    {
                        "tag": "note",
                        "elements": [
                            {"tag": "plain_text", "content": "虫群记忆系统 v3.0"}
                        ]
                    }
                ]
            }
        }

        try:
            resp = requests.post(webhook, json=payload, timeout=10)
            if resp.status_code == 200:
                print(f"[Feishu] ✅ 仲裁通知已发送: {case.get('id')}")
                return True
            else:
                print(f"[Feishu] ❌ 发送失败: {resp.status_code}")
                return False
        except Exception as e:
            print(f"[Feishu] ❌ 发送异常: {e}")
            return False

    def notify_sync_ready(self, agent_id: str, skill_count: int, feishu_webhook: str = None) -> bool:
        """通知同步就绪"""
        webhook = feishu_webhook or self.webhook_url
        if not webhook:
            return False

        payload = {
            "msg_type": "text",
            "content": {"text": f"🔔 同步就绪: {agent_id} - {skill_count} 个 skills"}
        }

        try:
            resp = requests.post(webhook, json=payload, timeout=10)
            return resp.status_code == 200
        except Exception:
            return False

    def send_hook_stats(self, stats: list[dict], feishu_webhook: str = None) -> bool:
        """发送 Hook 统计数据到飞书"""
        webhook = feishu_webhook or self.webhook_url
        if not webhook:
            print("[Feishu] 未配置 webhook，跳过 hook 统计通知")
            return False

        lines = ["📊 MisakaNet Hook 统计\n"]
        for s in stats:
            node = s.get("node", "?")
            cat_lines = []
            for cat in ["network", "pip", "permission", "disk", "package_conflict", "model_output"]:
                t = s.get("triggers", {}).get(cat, 0)
                h = s.get("hits", {}).get(cat, 0)
                if t > 0:
                    icons = {"network": "🔴", "pip": "🟡", "permission": "🔵", "disk": "🟣",
                             "package_conflict": "🟠", "model_output": "⚪"}
                    cat_lines.append(f"  {icons.get(cat, '⚪')} {cat}  {t}次  lessons 有答案{h}次")
            if cat_lines:
                lines.append(f"节点 {node}:")
                lines.extend(cat_lines)
            else:
                lines.append(f"节点 {node}: 无触发")

        payload = {
            "msg_type": "text",
            "content": {"text": "\n".join(lines)}
        }

        try:
            resp = requests.post(webhook, json=payload, timeout=10)
            if resp.status_code == 200:
                print(f"[Feishu] Hook stats 已推送 ({len(stats)} 节点)")
            return resp.status_code == 200
        except Exception as e:
            print(f"[Feishu] 推送失败: {e}")
            return False

    def notify_agent_registered(self, agent_id: str, feishu_webhook: str = None) -> bool:
        """通知新节点注册"""
        webhook = feishu_webhook or self.webhook_url
        if not webhook:
            return False

        payload = {
            "msg_type": "text",
            "content": {"text": f"🆕 节点注册: {agent_id}"}
        }

        try:
            resp = requests.post(webhook, json=payload, timeout=10)
            return resp.status_code == 200
        except Exception:
            return False