简单的 HTTP 处理器,用于捕获飞书事件的 chat_id
| 26 | def do_POST(self): |
| 27 | content_length = int(self.headers.get('Content-Length', 0)) |
| 28 | body = self.rfile.read(content_length) |
| 29 | |
| 30 | try: |
| 31 | event = json.loads(body.decode('utf-8')) |
| 32 | event_type = event.get("type", "") |
| 33 | |
| 34 | # 跳过 url_verification |
| 35 | if event_type == "url_verification": |
| 36 | challenge = event.get("challenge", "") |
| 37 | self.send_response(200) |
| 38 | self.send_header('Content-Type', 'application/json') |
| 39 | self.end_headers() |
| 40 | self.wfile.write(json.dumps({"challenge": challenge}).encode('utf-8')) |
| 41 | return |
| 42 | |
| 43 | # 解析事件 |
| 44 | header = event.get("header", {}) |
| 45 | event_type = header.get("event_type", "") |
| 46 | |
| 47 | if event_type == "im.message.receive_v1": |
| 48 | event_data = event.get("event", {}) |
| 49 | message = event_data.get("message", {}) |
| 50 | sender = event_data.get("sender", {}) |
| 51 | |
| 52 | chat_id = message.get("chat_id", "") |
| 53 | msg_type = message.get("msg_type", "") |
| 54 | content_raw = message.get("content", "{}") |
| 55 | |
| 56 | # 解析消息内容 |
| 57 | try: |
| 58 | content = json.loads(content_raw) if isinstance(content_raw, str) else content_raw |
| 59 | except json.JSONDecodeError: |
| 60 | content = {"text": content_raw} |
| 61 | |
| 62 | text = content.get("text", "") |
| 63 | sender_type = sender.get("sender_type", "") |
| 64 | open_id = sender.get("sender_id", {}).get("open_id", "") |
| 65 | user_id = sender.get("sender_id", {}).get("user_id", "") |
| 66 | |
| 67 | # 跳过 bot 消息 |
| 68 | if sender_type == "app_bot": |
| 69 | return |
| 70 | |
| 71 | # 打印 chat_id 信息 |
| 72 | print("\n" + "="*60) |
| 73 | print("📱 收到消息!") |
| 74 | print("="*60) |
| 75 | print(f" chat_id: {chat_id}") |
| 76 | print(f" open_id: {open_id}") |
| 77 | print(f" user_id: {user_id or '(空)'}") |
| 78 | print(f" sender_type:{sender_type}") |
| 79 | print(f" msg_type: {msg_type}") |
| 80 | print(f" 消息内容: {text[:50]}...") |
| 81 | print("="*60) |
| 82 | print("\n👉 请将上面的 chat_id 复制到 roles.json 中对应的角色下") |
| 83 | print(" 例如:\"rolea\": { \"feishu_chat_id\": \"" + chat_id + "\" }\n") |
| 84 | |
| 85 | # 发送响应 |
nothing calls this directly
no outgoing calls
no test coverage detected