""" chatbot/whatsapp_bot.py - WhatsApp 机器人 =========================================== 使用 whatsapp-web.js 或 Baileys 库接入 WhatsApp。 支持二维码绑定（通过 Node.js Baileys bridge）。 纯 Python 实现，异步运行。 """ from __future__ import annotations import asyncio import json import time import os import subprocess from pathlib import Path from typing import Optional, List from chatbot.base import BaseChatBot, ChatMessage, ChatResponse try: import aiohttp HAS_AIOHTTP = True except ImportError: HAS_AIOHTTP = False class WhatsAppBot(BaseChatBot): """ WhatsApp 机器人适配器。 配置要求 (Cloud API 模式): - token: Access Token (从 Meta Developer Portal 获取) - phone_number_id: Phone Number ID (从 Meta 获取) - verify_token: Webhook 验证 Token (自动生成) 配置要求 (QR 码绑定模式): - 无需 token / phone_number_id - 需要 Node.js 环境 + Baileys 依赖 - 首次运行会生成 QR 码，扫码后自动保存会话 支持两种模式: 1. Cloud API 模式: 使用 Meta WhatsApp Business API 2. Baileys 模式: 通过 QR Code 绑定 (推荐个人使用) """ platform_name = "whatsapp" def __init__(self, **kwargs): super().__init__(**kwargs) self._app_id = kwargs.get("app_id", "") self._app_secret = kwargs.get("app_secret", "") self._phone_number_id = self.config.get("phone_number_id", "") self._verify_token = self.config.get("verify_token", "") self._webhook_base = self.config.get("webhook_base", "") self._qr_code = "" self._connected = False self._bridge_process: Optional[subprocess.Popen] = None self._bridge_mode = not self._phone_number_id # 没有 phone_number_id 则使用 bridge self._session_dir = str(Path(__file__).parent / "whatsapp_bridge" / "session") async def start(self): """启动 WhatsApp 机器人""" if self._bridge_mode: await self._start_bridge_mode() else: await self._start_cloud_mode() async def _start_cloud_mode(self): """Meta Cloud API 模式""" if not self.token: self.logger.error("WhatsApp Access Token 未配置") return if not self._phone_number_id: self.logger.error("Cloud API 模式需要配置 phone_number_id") return self.logger.info("WhatsApp Cloud API 模式启动") self._connected = True while self._running: await asyncio.sleep(1) async def _start_bridge_mode(self): """Baileys Bridge 模式 (QR 码绑定)""" bridge_dir = Path(__file__).parent / "whatsapp_bridge" bridge_script = bridge_dir / "bridge.mjs" if not bridge_script.exists(): self.logger.error(f"Bridge 脚本不存在: {bridge_script}") self.logger.error("请运行: cd chatbot/whatsapp_bridge && npm install @whiskeysockets/baileys qrcode-terminal") return # 检查是否已安装依赖 node_modules = bridge_dir / "node_modules" if not node_modules.exists(): self.logger.warning("Baileys 依赖未安装，请手动执行:") self.logger.warning(f" cd {bridge_dir} && npm install") return env = os.environ.copy() env["SESSION_DIR"] = self._session_dir try: self._bridge_process = await asyncio.create_subprocess_exec( "node", str(bridge_script), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env, cwd=str(bridge_dir), ) self.logger.info("WhatsApp Bridge 进程已启动") # 启动消息读取循环 asyncio.create_task(self._read_bridge_output()) except Exception as e: self.logger.error(f"Bridge 启动失败: {e}") while self._running: await asyncio.sleep(1) async def _read_bridge_output(self): """读取 Bridge 进程的 stdout (JSON 协议)""" if not self._bridge_process or not self._bridge_process.stdout: return try: reader = self._bridge_process.stdout while self._running and self._bridge_process.returncode is None: line = await reader.readline() if not line: break line = line.decode('utf-8', errors='replace').strip() if not line: continue try: msg = json.loads(line) msg_type = msg.get("type", "") if msg_type == "qr": self._qr_code = msg.get("qr", "") self.logger.info("QR 码已生成，等待扫码...") elif msg_type == "connected": self._connected = True phone = msg.get("phone", "") name = msg.get("name", "") self.logger.info(f"WhatsApp 已连接: {name} ({phone})") # 更新配置中的连接状态 self.config["connection_status"] = "connected" self.config["connected_phone"] = phone elif msg_type == "disconnected": self._connected = False reason = msg.get("reason", "unknown") self.logger.warning(f"WhatsApp 已断开: {reason}") self.config["connection_status"] = "disconnected" elif msg_type == "message": from_id = msg.get("from", "") text = msg.get("text", "") push_name = msg.get("pushName", "") if from_id and text: chat_msg = ChatMessage( platform=self.platform_name, chat_id=from_id, user_id=from_id, username=push_name or from_id.split('@')[0], text=text, is_group='@g.us' in from_id, raw_data=msg, ) await self._handle_message(chat_msg) elif msg_type == "error": error = msg.get("error", "") self.logger.error(f"Bridge 错误: {error}") except json.JSONDecodeError: self.logger.debug(f"非 JSON 输出: {line[:100]}") except Exception as e: self.logger.error(f"Bridge 输出读取异常: {e}") async def stop(self): """停止 WhatsApp 机器人""" self._running = False self._connected = False if self._bridge_process: try: self._bridge_process.terminate() await asyncio.sleep(1) if self._bridge_process.returncode is None: self._bridge_process.kill() except Exception: pass self._bridge_process = None self.logger.info("WhatsApp 机器人已停止") async def send_message(self, response: ChatResponse) -> bool: """发送消息到 WhatsApp""" if not self._connected or not response.chat_id: return False if self._bridge_mode: return await self._send_via_bridge(response.chat_id, response.text) else: return await self._send_via_cloud_api(response.chat_id, response.text) async def _send_via_bridge(self, chat_id: str, text: str) -> bool: """通过 Bridge 发送消息""" if not self._bridge_process or not self._bridge_process.stdin: return False try: msg = json.dumps({"action": "send", "to": chat_id, "text": text}) + "\n" self._bridge_process.stdin.write(msg.encode('utf-8')) await self._bridge_process.stdin.drain() return True except Exception as e: self.logger.error(f"Bridge 发送失败: {e}") return False async def _send_via_cloud_api(self, chat_id: str, text: str) -> bool: """通过 Cloud API 发送消息""" if not HAS_AIOHTTP: self.logger.error("请安装 aiohttp: pip install aiohttp") return False url = f"https://graph.facebook.com/v18.0/{self._phone_number_id}/messages" headers = { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", } payload = { "messaging_product": "whatsapp", "to": chat_id, "type": "text", "text": {"body": text[:4096]}, } try: async with aiohttp.ClientSession() as session: async with session.post(url, headers=headers, json=payload) as resp: if resp.status == 200: return True error_text = await resp.text() self.logger.error(f"WhatsApp 发送失败 ({resp.status}): {error_text[:200]}") return False except Exception as e: self.logger.error(f"Cloud API 发送失败: {e}") return False # ========================================================================== # Webhook 处理 (Cloud API 模式) # ========================================================================== def verify_webhook(self, mode: str, token: str, challenge: str) -> Optional[str]: """验证 Webhook""" if mode == "subscribe" and token == self._verify_token: return challenge return None async def handle_webhook_event(self, event_data: dict): """处理 Webhook 事件 (Cloud API 模式)""" try: for entry in event_data.get("entry", []): for change in entry.get("changes", []): value = change.get("value", {}) messages = value.get("messages", []) contacts = value.get("contacts", []) contact_map = {} for c in contacts: wa_id = c.get("wa_id", "") name = c.get("profile", {}).get("name", "") contact_map[wa_id] = name for msg in messages: msg_type = msg.get("type", "") if msg_type != "text": continue from_id = msg.get("from", "") text_body = msg.get("text", {}).get("body", "") msg_id = msg.get("id", "") timestamp = msg.get("timestamp", "") username = contact_map.get(from_id, from_id) message = ChatMessage( platform=self.platform_name, chat_id=from_id, user_id=from_id, username=username, text=text_body, is_group=False, reply_to=msg_id, raw_data={"timestamp": timestamp, "msg_id": msg_id}, ) await self._handle_message(message) except Exception as e: self.logger.error(f"处理 Webhook 事件失败: {e}") # ========================================================================== # QR 码绑定 # ========================================================================== def get_qr_code(self) -> str: """获取当前 QR Code（base64 PNG 图片）""" return self._qr_code async def generate_qr_code(self) -> str: """请求生成 QR 码。Bridge 模式下，首次启动会自动生成。""" if self._bridge_mode: # 如果 bridge 已在运行且有 QR 码，直接返回 if self._qr_code: return self._qr_code # 如果 bridge 未运行，尝试启动 if not self._bridge_process or self._bridge_process.returncode is not None: asyncio.create_task(self._start_bridge_mode()) # 等待 QR 码生成（最多 30 秒） for _ in range(60): await asyncio.sleep(0.5) if self._qr_code: return self._qr_code if self._connected: return "" # 已连接，不需要 QR return self._qr_code # Cloud API 模式不需要 QR self.logger.info("Cloud API 模式不需要 QR Code") return ""