From 7019cf180d8f4dde583f4c67ecb9256596e399c8 Mon Sep 17 00:00:00 2001 From: Nelson <1475262689@qq.com> Date: Wed, 8 Apr 2026 11:00:31 +0800 Subject: [PATCH] init --- .gitignore | 2 + README.md | 15 +- wechat-bot/.env.example | 24 ++++ wechat-bot/.gitignore | 29 ++++ wechat-bot/Dockerfile | 23 +++ wechat-bot/README.md | 137 ++++++++++++++++++ wechat-bot/bot/__init__.py | 0 wechat-bot/bot/agent_client.py | 237 +++++++++++++++++++++++++++++++ wechat-bot/bot/handlers.py | 91 ++++++++++++ wechat-bot/bot/message_parser.py | 57 ++++++++ wechat-bot/bot/reply_parser.py | 84 +++++++++++ wechat-bot/bot/wecom_bot.py | 104 ++++++++++++++ wechat-bot/config.py | 41 ++++++ wechat-bot/docker-compose.yml | 10 ++ wechat-bot/main.py | 86 +++++++++++ wechat-bot/pyproject.toml | 18 +++ wechat-bot/requirements.txt | 3 + 17 files changed, 960 insertions(+), 1 deletion(-) create mode 100644 wechat-bot/.env.example create mode 100644 wechat-bot/.gitignore create mode 100644 wechat-bot/Dockerfile create mode 100644 wechat-bot/README.md create mode 100644 wechat-bot/bot/__init__.py create mode 100644 wechat-bot/bot/agent_client.py create mode 100644 wechat-bot/bot/handlers.py create mode 100644 wechat-bot/bot/message_parser.py create mode 100644 wechat-bot/bot/reply_parser.py create mode 100644 wechat-bot/bot/wecom_bot.py create mode 100644 wechat-bot/config.py create mode 100644 wechat-bot/docker-compose.yml create mode 100644 wechat-bot/main.py create mode 100644 wechat-bot/pyproject.toml create mode 100644 wechat-bot/requirements.txt diff --git a/.gitignore b/.gitignore index 36b13f1..3ed95dd 100644 --- a/.gitignore +++ b/.gitignore @@ -174,3 +174,5 @@ cython_debug/ # PyPI configuration file .pypirc +.DS_Store + diff --git a/README.md b/README.md index 17d2fb0..61dcf8a 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,16 @@ # Wechat-Bot -企业微信消息分发中转站 \ No newline at end of file +企业微信消息分发中转站 +企业微信doc：https://developer.work.weixin.qq.com/document/path/101463 + +企业微信sdk: aibot-python-sdk(wecom-aibot-python-sdk 1.0.2) + +本项目为企业微信与agent的桥梁 +企业微信通过创建智能机器人，用socket长链接方式与本地服务进行数据交换。 +后端agent为adk开发，并使用fastapi形式部署，所以可以使用adk中的client库进行交互。 + +全部项目均使用python编写 + +企业微信的userid与agent的userid可以保持一一对应关系，直接使用。 + +所有配置均应放在.env文件中 \ No newline at end of file diff --git a/wechat-bot/.env.example b/wechat-bot/.env.example new file mode 100644 index 0000000..2b3e91c --- /dev/null +++ b/wechat-bot/.env.example @@ -0,0 +1,24 @@ +# ============================================ +# 企业微信机器人配置 +# ============================================ +# 在企业微信管理后台 -> 智能机器人 -> API模式 -> 长连接 获取 +WECHAT_BOT_ID=your_bot_id_here +WECHAT_BOT_SECRET=your_bot_secret_here + +# ============================================ +# 远程 Agent 服务配置（ADK api_server） +# ============================================ +# Agent 服务地址（adk api_server 启动地址） +AGENT_BASE_URL=http://your-agent-host:8000 +# Agent 应用名称（对应 agents 目录名） +AGENT_APP_NAME=root_agent +# Agent 请求超时时间（秒） +AGENT_TIMEOUT=120 + +# ============================================ +# 服务配置 +# ============================================ +# 日志级别: DEBUG, INFO, WARNING, ERROR +LOG_LEVEL=INFO +# 数据目录（用于存储日志数据库等） +DATA_DIR=./data diff --git a/wechat-bot/.gitignore b/wechat-bot/.gitignore new file mode 100644 index 0000000..45fac81 --- /dev/null +++ b/wechat-bot/.gitignore @@ -0,0 +1,29 @@ +# Dependencies +__pycache__/ +*.py[cod] +*$py.class +*.egg-info/ +dist/ +build/ +.eggs/ + +# Environment +.env +venv/ +.venv/ + +# Data & Logs +data/ +logs/ +*.db +*.sqlite3 + +# IDE +.vscode/ +.idea/ +*.swp +*.swo + +# OS +.DS_Store +Thumbs.db diff --git a/wechat-bot/Dockerfile b/wechat-bot/Dockerfile new file mode 100644 index 0000000..a3528ce --- /dev/null +++ b/wechat-bot/Dockerfile @@ -0,0 +1,23 @@ +FROM python:3.12-slim + +WORKDIR /wechat-bot + +# 安装系统依赖 +RUN apt-get update && \ + apt-get install -y --no-install-recommends gcc curl && \ + rm -rf /var/lib/apt/lists/* + +# 安装 Python 依赖（利用 Docker 层缓存） +COPY requirements.txt . +RUN pip install --no-cache-dir --break-system-packages -r requirements.txt + +# 复制项目代码 +COPY config.py . +COPY main.py . +COPY bot/ ./bot/ + +# 环境变量 +ENV PYTHONUNBUFFERED=1 +ENV PYTHONDONTWRITEBYTECODE=1 + +CMD ["python", "main.py"] diff --git a/wechat-bot/README.md b/wechat-bot/README.md new file mode 100644 index 0000000..48b7447 --- /dev/null +++ b/wechat-bot/README.md @@ -0,0 +1,137 @@ +# Wechat-Bot + +企业微信消息分发中转站 —— 企业微信智能机器人与远程 Agent 的桥梁。 + +企业微信文档：https://developer.work.weixin.qq.com/document/path/101463 + +## 架构说明 + +``` +企业微信用户 + │ + ▼ +企业微信服务器 (wss://openws.work.weixin.qq.com) + │ WebSocket 长连接 + ▼ +wecom-aibot-python-sdk (接收/发送消息) + │ SSE 流式 + HTTP 调用 + ▼ +远程 Agent 服务 (AGENT_BASE_URL) +``` + +- **企业微信侧**：通过 `wecom-aibot-python-sdk` 建立 WebSocket 长连接，接收用户消息 +- **Agent 侧**：通过 HTTP 调用远程 Agent 的 REST API（兼容 ADK `api_server` 格式） +- **流式回复**：优先使用 `/run_sse`（SSE 流式）实现打字机效果，失败时降级为 `/run`（同步） +- **会话管理**：自动为每个用户创建 Agent 会话，企业微信 userid 与 Agent user_id 一一对应 +- **思考过滤**：自动去除模型思考过程（`` 标签），只推送正式回复给用户 + +## 支持的消息类型 + +| 类型 | msgtype | 说明 | +|------|---------|------| +| 文本 | `text` | 直接转发给 Agent 处理 | +| 图片 | `image` | 下载并描述，转发给 Agent | +| 语音 | `voice` | 提取文本内容后处理 | +| 文件 | `file` | 下载并描述，转发给 Agent | +| 视频 | `video` | 下载并描述，转发给 Agent | +| 图文混排 | `mixed` | 提取文本内容后处理 | + +## 项目结构 + +``` +wechat-bot/ +├── main.py # 主入口，启动 WebSocket 客户端 +├── config.py # 全局配置（从 .env 加载） +├── requirements.txt # Python 依赖 +├── Dockerfile # Docker 镜像构建 +├── docker-compose.yml # Docker Compose 编排 +├── .env.example # 环境变量模板 +└── bot/ + ├── __init__.py + ├── wecom_bot.py # 组装层：创建 WSClient + 注册事件 + ├── agent_client.py # Agent 通信：HTTP 客户端、会话管理、SSE/同步调用 + ├── handlers.py # 消息处理：统一的 handle_message() + ├── message_parser.py # 消息解析：提取文本内容和元数据 + └── reply_parser.py # 回复解析：去除思考标签、去重 +``` + +## 快速开始 + +### 1. 配置环境变量 + +```bash +cp .env.example .env +``` + +编辑 `.env` 文件，填写以下必要配置： + +| 配置项 | 必填 | 说明 | 默认值 | +|--------|------|------|--------| +| `WECHAT_BOT_ID` | ✅ | 企业微信机器人 ID | - | +| `WECHAT_BOT_SECRET` | ✅ | 长连接密钥 | - | +| `AGENT_BASE_URL` | ✅ | Agent 服务地址 | `http://127.0.0.1:8000` | +| `AGENT_APP_NAME` | ❌ | Agent 应用名称 | `root_agent` | +| `AGENT_TIMEOUT` | ❌ | Agent 调用超时（秒） | `120` | +| `LOG_LEVEL` | ❌ | 日志级别 | `INFO` | + +> `WECHAT_BOT_ID` 和 `WECHAT_BOT_SECRET` 获取方式：企业微信管理后台 → 智能机器人 → API 模式 + +### 2. 启动 Agent 服务 + +确保你的 Agent 服务已启动并监听对应端口。 + +### 3. 启动 Bot + +**Docker 部署（推荐，适用于 NAS）：** + +```bash +docker compose up -d --build +docker compose logs -f +``` + +**本地运行：** + +```bash +pip install -r requirements.txt +python main.py +``` + +## Agent API 集成 + +本项目通过 HTTP 调用远程 Agent，兼容 ADK `api_server` 接口格式： + +| 接口 | 方法 | 用途 | +|------|------|------| +| `/apps/{app_name}/users/{user_id}/sessions` | POST | 为用户创建会话 | +| `/run_sse` | POST | SSE 流式对话（主要方式） | +| `/run` | POST | 同步对话（降级方案） | + +### 请求格式 + +```json +{ + "app_name": "root_agent", + "user_id": "企业微信用户ID", + "session_id": "会话ID", + "new_message": { + "role": "user", + "parts": [{"text": "用户消息"}] + } +} +``` + +### 响应处理 + +- SSE 流式：逐行解析 `data:` 事件，收集 `content.parts[].text` +- 自动过滤 `thought: true` 的思考内容 +- 自动去除 `= 3.10 +- **wecom-aibot-python-sdk** - 企业微信智能机器人 SDK +- **httpx** - 异步 HTTP 客户端（支持 SSE 流式） +- **python-dotenv** - 环境变量管理 +- **Docker** + **Docker Compose** - 容器化部署 diff --git a/wechat-bot/bot/__init__.py b/wechat-bot/bot/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/wechat-bot/bot/agent_client.py b/wechat-bot/bot/agent_client.py new file mode 100644 index 0000000..ad60ea3 --- /dev/null +++ b/wechat-bot/bot/agent_client.py @@ -0,0 +1,237 @@ +""" +Agent HTTP 客户端模块 +负责与远程 ADK Agent 服务通信，包括： +- HTTP 客户端生命周期管理 +- 用户会话管理（创建、缓存、清除） +- SSE 流式调用（主方案） +- 同步调用（降级方案） +""" +import json +import logging +from typing import Optional + +import httpx + +from config import Config +from bot.reply_parser import extract_final_reply + +logger = logging.getLogger(__name__) + +# 用户会话映射: user_id -> session_id +_user_sessions: dict[str, str] = {} + +# Agent HTTP 客户端 +_http_client: Optional[httpx.AsyncClient] = None + + +def get_http_client() -> httpx.AsyncClient: + """获取或创建 HTTP 客户端（全局单例）""" + global _http_client + if _http_client is None: + _http_client = httpx.AsyncClient( + base_url=Config.AGENT_BASE_URL, + timeout=Config.AGENT_TIMEOUT, + ) + return _http_client + + +async def close_http_client(): + """关闭 HTTP 客户端""" + global _http_client + if _http_client: + await _http_client.aclose() + _http_client = None + + +# ============================================================ +# 会话管理 +# ============================================================ + +async def get_or_create_session(user_id: str) -> str: + """获取或创建用户的 Agent 会话""" + if user_id in _user_sessions: + return _user_sessions[user_id] + + client = get_http_client() + try: + url = f"/apps/{Config.AGENT_APP_NAME}/users/{user_id}/sessions" + resp = await client.post(url, json={}) + resp.raise_for_status() + data = resp.json() + session_id = data["id"] + _user_sessions[user_id] = session_id + logger.info("为用户 %s 创建 Agent 会话: %s", user_id, session_id) + return session_id + except Exception as e: + logger.error("创建 Agent 会话失败: %s", e) + raise + + +def clear_session(user_id: str): + """清除用户的会话缓存（会话失效时调用）""" + _user_sessions.pop(user_id, None) + + +# ============================================================ +# Agent 调用 +# ============================================================ + +async def call_agent_stream( + ws_client, + frame: dict, + user_id: str, + message: str, +) -> str: + """ + 通过 SSE 流式调用 Agent，实现打字机效果。 + 失败时降级为同步调用。 + 返回完整的回复文本。 + """ + from aibot import generate_req_id + + session_id = await get_or_create_session(user_id) + client = get_http_client() + + try: + req_body = { + "app_name": Config.AGENT_APP_NAME, + "user_id": user_id, + "session_id": session_id, + "new_message": { + "role": "user", + "parts": [{"text": message}], + }, + "streaming": True, + } + + async with client.stream( + "POST", + "/run_sse", + json=req_body, + timeout=Config.AGENT_TIMEOUT, + ) as response: + if response.status_code != 200: + logger.warning( + "SSE 流式调用返回 %d，降级为同步调用", + response.status_code, + ) + return await call_agent_sync(user_id, message) + + stream_id = generate_req_id("stream") + full_text = "" + has_sent_final = False + + async for line in response.aiter_lines(): + if not line.startswith("data: "): + continue + + data_str = line[6:] + try: + data = json.loads(data_str) + except json.JSONDecodeError: + continue + + # 错误处理 + if "error" in data: + logger.error("Agent SSE 错误: %s", data["error"]) + if "not found" in data["error"].lower(): + clear_session(user_id) + return "抱歉，处理消息时出现错误，请稍后重试。" + + # 收集所有文本（含思考过程） + parts = data.get("content", {}).get("parts", []) + for part in parts: + text = part.get("text", "") + if not text: + continue + if part.get("thought"): + continue + full_text += text + + # 实时推送"正在思考"状态 + if full_text and not has_sent_final: + try: + await ws_client.reply_stream( + frame, stream_id, "💭 正在思考...", False + ) + except Exception: + pass + + # 最终完成时提取正式回复并发送 + is_final = data.get("finishReason") == "STOP" + if is_final and full_text and not has_sent_final: + clean_reply = extract_final_reply(full_text) + logger.info("Agent 回复完成，长度: %d", len(clean_reply)) + try: + await ws_client.reply_stream( + frame, stream_id, clean_reply, True + ) + has_sent_final = True + except Exception as e: + logger.warning("推送最终回复失败: %s", e) + + # 兜底：如果流结束但未发送最终回复 + if full_text and not has_sent_final: + clean_reply = extract_final_reply(full_text) + logger.info("Agent SSE 兜底发送，长度: %d", len(clean_reply)) + try: + await ws_client.reply_stream( + frame, stream_id, clean_reply, True + ) + except Exception as e: + logger.warning("推送兜底回复失败: %s", e) + + return extract_final_reply(full_text) if full_text else "" + + except httpx.ConnectError: + logger.error("无法连接到 Agent 服务: %s", Config.AGENT_BASE_URL) + return "抱歉，Agent 服务连接失败，请稍后重试。" + except Exception as e: + logger.error("SSE 流式调用异常，降级为同步: %s", e) + return await call_agent_sync(user_id, message) + + +async def call_agent_sync(user_id: str, message: str) -> str: + """ + 同步调用 Agent（/run 接口），作为流式调用的降级方案。 + """ + session_id = await get_or_create_session(user_id) + client = get_http_client() + + def _build_body(sid: str) -> dict: + return { + "app_name": Config.AGENT_APP_NAME, + "user_id": user_id, + "session_id": sid, + "new_message": { + "role": "user", + "parts": [{"text": message}], + }, + } + + def _extract_reply(events: list) -> str: + """从 Event[] 中提取最终回复（跳过 thought）""" + for event in events: + parts = event.get("content", {}).get("parts", []) + for part in parts: + if part.get("text") and not part.get("thought"): + return part["text"] + return "" + + try: + resp = await client.post("/run", json=_build_body(session_id)) + resp.raise_for_status() + return _extract_reply(resp.json()) + + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + clear_session(user_id) + new_session = await get_or_create_session(user_id) + resp = await client.post("/run", json=_build_body(new_session)) + resp.raise_for_status() + return _extract_reply(resp.json()) + logger.error("Agent HTTP 错误 [%d]: %s", e.response.status_code, e) + return "抱歉，服务暂时不可用，请稍后重试。" + except Exception as e: + logger.error("同步调用 Agent 失败: %s", e) + return "抱歉，处理消息时出现错误，请稍后重试。" diff --git a/wechat-bot/bot/handlers.py b/wechat-bot/bot/handlers.py new file mode 100644 index 0000000..ce8ddd7 --- /dev/null +++ b/wechat-bot/bot/handlers.py @@ -0,0 +1,91 @@ +""" +消息处理器模块 +提供统一的消息处理流程，消除各消息类型处理器之间的重复代码。 + +处理流程： +1. 解析消息元数据和内容 +2. （可选）下载附件（图片/文件/视频） +3. 调用 Agent 获取回复 +""" +import logging + +from aibot import generate_req_id + +from bot.agent_client import call_agent_stream +from bot.message_parser import extract_metadata, extract_text_content + +logger = logging.getLogger(__name__) + + +async def handle_message(ws_client, frame: dict, msg_type: str): + """ + 通用消息处理器，适用于所有消息类型。 + + 对于需要下载附件的消息类型（image/file/video）， + 会先尝试下载附件并附加描述信息。 + """ + meta = extract_metadata(frame) + user_id = meta["user_id"] + + # ---- 提取消息内容 ---- + content = await _resolve_content(ws_client, frame, msg_type, meta) + logger.info("收到%s消息 - 用户: %s, 内容: %s", msg_type, user_id, content[:100]) + + # ---- 调用 Agent ---- + reply = await call_agent_stream(ws_client, frame, user_id, content) + + # ---- 兜底：未收到回复 ---- + if not reply: + stream_id = generate_req_id("stream") + await ws_client.reply_stream(frame, stream_id, "（未收到回复）", True) + + +async def _resolve_content( + ws_client, frame: dict, msg_type: str, meta: dict +) -> str: + """ + 解析消息内容。 + 对于文本类消息直接提取文本；对于附件类消息尝试下载并附加描述。 + """ + if msg_type in ("text", "voice", "mixed"): + return extract_text_content(frame) + + if msg_type == "image": + return await _download_and_describe(ws_client, frame, "image", "图片") + + if msg_type == "file": + return await _download_and_describe(ws_client, frame, "file", "文件") + + if msg_type == "video": + return await _download_and_describe(ws_client, frame, "video", "视频") + + return extract_text_content(frame) + + +async def _download_and_describe( + ws_client, frame: dict, media_type: str, label: str +) -> str: + """ + 尝试下载附件并返回描述文本。 + 下载失败时返回简单描述，不阻断流程。 + """ + body = frame.get("body", {}) + media_info = body.get(media_type, {}) + + if not media_info: + return f"[用户发送了一个{label}]" + + url = media_info.get("url") + aeskey = media_info.get("aeskey") + + if not url: + return f"[用户发送了一个{label}]" + + try: + buffer, filename = await ws_client.download_file(url, aeskey) + desc = f"[用户发送了{label}: {filename}, 大小: {len(buffer)} bytes]" + logger.info("%s下载成功: %s, 大小: %d bytes", label, filename, len(buffer)) + return desc + except Exception as e: + logger.warning("%s下载失败: %s", label, e) + return f"[用户发送了一个{label}]" diff --git a/wechat-bot/bot/message_parser.py b/wechat-bot/bot/message_parser.py new file mode 100644 index 0000000..218b1a9 --- /dev/null +++ b/wechat-bot/bot/message_parser.py @@ -0,0 +1,57 @@ +""" +消息解析模块 +负责从企业微信 SDK 的消息帧中提取结构化数据。 + +frame 结构: {cmd, headers, body: {msgtype, text, from, ...}} +注意：msgtype 在 body 内部，不在 frame 顶层。 +""" + + +def extract_text_content(frame: dict) -> str: + """ + 从消息帧中提取文本内容。 + 根据不同的消息类型，提取可用于 Agent 处理的文本描述。 + """ + body = frame.get("body", {}) + msg_type = body.get("msgtype", "") + + if msg_type == "text": + return body.get("text", {}).get("content", "") + + elif msg_type == "image": + return "[用户发送了一张图片]" + + elif msg_type == "voice": + return body.get("voice", {}).get("content", "[用户发送了一条语音消息]") + + elif msg_type == "file": + filename = body.get("file", {}).get("filename", "未知文件") + return f"[用户发送了一个文件: {filename}]" + + elif msg_type == "video": + return "[用户发送了一段视频]" + + elif msg_type == "mixed": + mixed_items = body.get("mixed", {}).get("content", []) + text_parts = [] + for item in mixed_items: + if isinstance(item, dict) and item.get("type") == "text": + text_parts.append(item.get("content", "")) + elif isinstance(item, str): + text_parts.append(item) + return "\n".join(text_parts) if text_parts else "[用户发送了图文混排消息]" + + return f"[未知消息类型: {msg_type}]" + + +def extract_metadata(frame: dict) -> dict: + """提取消息的元数据（msgtype 在 body 内部）""" + body = frame.get("body", {}) + from_info = body.get("from", {}) + return { + "user_id": from_info.get("userid", ""), + "chat_id": body.get("chatid", ""), + "chat_type": body.get("chattype", ""), + "msg_id": body.get("msgid", ""), + "msg_type": body.get("msgtype", ""), + } diff --git a/wechat-bot/bot/reply_parser.py b/wechat-bot/bot/reply_parser.py new file mode 100644 index 0000000..87da722 --- /dev/null +++ b/wechat-bot/bot/reply_parser.py @@ -0,0 +1,84 @@ +""" +回复文本解析模块 +负责处理 Agent 返回的原始文本，提取最终正式回复。 + +模型思考标记格式（按优先级检测）： +1. str: + """ + 从 SSE 收集的完整文本中提取最终正式回复。 + + 策略： + 1. 去除 ?", "", text) + + # 步骤2：兼容 💭 emoji 分割（旧版格式） + if "💭" in text: + text = text.split("💭")[-1] + + # 清理前导空白 + text = text.strip() + + # 步骤3：去重 + text = _deduplicate(text) + + return text + + +def _deduplicate(text: str) -> str: + """ + 对回复文本去重。 + ADK 最后一个 SSE event 会重复发送完整内容（含思考+回复）， + 去除 后可能出现连续两份相同的正式回复。 + 支持行级和字符级去重。 + """ + if not text: + return text + + # 行级去重 + lines = text.split("\n") + non_empty = [l for l in lines if l.strip()] + + if len(non_empty) > 1: + mid = len(non_empty) // 2 + if non_empty[:mid] == non_empty[mid:]: + return "\n".join(non_empty[:mid]) + + for overlap_len in range(min(len(non_empty) // 2, 20), 0, -1): + if non_empty[-overlap_len:] == non_empty[:overlap_len]: + return "\n".join(non_empty[overlap_len:]) + + # 字符级去重：检查文本是否由两个相同的子串拼接而成 + length = len(text) + for sub_len in range(length // 2, 0, -1): + if length % sub_len == 0 and text[:sub_len] * (length // sub_len) == text: + return text[:sub_len] + + # 部分重叠：前缀和后缀相同，中间有重复 + for overlap in range(min(length // 2, 500), 0, -1): + if text[:overlap] == text[-overlap:]: + trimmed = text[overlap:] + trimmed_len = len(trimmed) + for sub_len in range(trimmed_len // 2, 0, -1): + if trimmed_len % sub_len == 0 and trimmed[:sub_len] * (trimmed_len // sub_len) == trimmed: + return trimmed[:sub_len] + + return text diff --git a/wechat-bot/bot/wecom_bot.py b/wechat-bot/bot/wecom_bot.py new file mode 100644 index 0000000..1f740d7 --- /dev/null +++ b/wechat-bot/bot/wecom_bot.py @@ -0,0 +1,104 @@ +""" +企业微信桥接模块 +使用 wecom-aibot-python-sdk 与企业微信建立 WebSocket 长连接， +接收用户消息后通过 ADK API 调用远程 Agent，再将 Agent 回复发送给用户。 + +本模块仅负责： +- 创建 WSClient 实例 +- 注册事件和消息处理器 +""" +import json +import logging +import traceback + +from aibot import WSClient, WSClientOptions + +from config import Config +from bot.agent_client import close_http_client +from bot.handlers import handle_message +from bot.message_parser import extract_metadata + +logger = logging.getLogger(__name__) + + +def create_bot_client() -> WSClient: + """ + 创建企业微信机器人 WebSocket 客户端，并注册所有事件和消息处理器。 + """ + ws_client = WSClient( + WSClientOptions( + bot_id=Config.WECHAT_BOT_ID, + secret=Config.WECHAT_BOT_SECRET, + ) + ) + + # ---- 事件处理器 ---- + + @ws_client.on("authenticated") + def on_authenticated(): + logger.info("企业微信机器人认证成功") + + @ws_client.on("event.enter_chat") + async def on_enter_chat(frame): + """用户进入会话，发送欢迎语""" + meta = extract_metadata(frame) + user_id = meta.get("user_id", "") + logger.info("用户进入会话: %s", user_id) + + try: + await ws_client.reply_welcome( + frame, + { + "msgtype": "text", + "text": { + "content": "您好！我是智能助手，有什么可以帮您的吗？" + }, + }, + ) + except Exception as e: + logger.error("发送欢迎语失败: %s", e) + + @ws_client.on("event.feedback_event") + async def on_feedback(frame): + """用户反馈事件""" + body = frame.get("body", {}) + logger.info("收到用户反馈: %s", json.dumps(body, ensure_ascii=False)) + + @ws_client.on("disconnected") + def on_disconnected(reason): + logger.warning("连接被断开（可能被新连接踢掉）: %s", reason) + + # ---- 消息处理器 ---- + + @ws_client.on("message.text") + async def on_text(frame): + await _safe_handle(ws_client, frame, "text") + + @ws_client.on("message.image") + async def on_image(frame): + await _safe_handle(ws_client, frame, "image") + + @ws_client.on("message.voice") + async def on_voice(frame): + await _safe_handle(ws_client, frame, "voice") + + @ws_client.on("message.file") + async def on_file(frame): + # file 类型可能包含视频 + body = frame.get("body", {}) + msg_type = "video" if body.get("video") else "file" + await _safe_handle(ws_client, frame, msg_type) + + @ws_client.on("message.mixed") + async def on_mixed(frame): + await _safe_handle(ws_client, frame, "mixed") + + return ws_client + + +async def _safe_handle(ws_client, frame: dict, msg_type: str): + """安全地处理消息，捕获异常防止崩溃""" + try: + await handle_message(ws_client, frame, msg_type) + except Exception as e: + logger.error("处理%s消息异常: %s\n%s", msg_type, e, traceback.format_exc()) diff --git a/wechat-bot/config.py b/wechat-bot/config.py new file mode 100644 index 0000000..1f3f3a6 --- /dev/null +++ b/wechat-bot/config.py @@ -0,0 +1,41 @@ +""" +全局配置模块 +从 .env 文件加载所有配置项，提供统一的配置访问接口。 +""" +import os +from pathlib import Path +from dotenv import load_dotenv + +# 项目根目录 +BASE_DIR = Path(__file__).parent + +# 加载 .env 文件 +load_dotenv(BASE_DIR / ".env") + + +class Config: + """全局配置类，所有配置项均从环境变量读取""" + + # ---- 企业微信机器人 ---- + WECHAT_BOT_ID: str = os.getenv("WECHAT_BOT_ID", "") + WECHAT_BOT_SECRET: str = os.getenv("WECHAT_BOT_SECRET", "") + + # ---- 远程 Agent 服务 ---- + AGENT_BASE_URL: str = os.getenv("AGENT_BASE_URL", "http://127.0.0.1:8000") + AGENT_APP_NAME: str = os.getenv("AGENT_APP_NAME", "root_agent") + AGENT_TIMEOUT: int = int(os.getenv("AGENT_TIMEOUT", "120")) + + # ---- 服务配置 ---- + LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO") + + @classmethod + def validate(cls) -> list[str]: + """校验必要配置项，返回缺失项列表""" + missing = [] + if not cls.WECHAT_BOT_ID: + missing.append("WECHAT_BOT_ID") + if not cls.WECHAT_BOT_SECRET: + missing.append("WECHAT_BOT_SECRET") + if not cls.AGENT_BASE_URL: + missing.append("AGENT_BASE_URL") + return missing diff --git a/wechat-bot/docker-compose.yml b/wechat-bot/docker-compose.yml new file mode 100644 index 0000000..28a3262 --- /dev/null +++ b/wechat-bot/docker-compose.yml @@ -0,0 +1,10 @@ +services: + wechat-bot: + build: . + container_name: wechat-bot + restart: unless-stopped + env_file: + - .env + volumes: + - ./data:/app/data + - ./logs:/app/logs diff --git a/wechat-bot/main.py b/wechat-bot/main.py new file mode 100644 index 0000000..a32f23f --- /dev/null +++ b/wechat-bot/main.py @@ -0,0 +1,86 @@ +""" +Wechat-Bot 主入口 +启动企业微信 WebSocket 长连接，接收消息后转发给远程 Agent。 +""" +import asyncio +import logging +import sys + +from config import Config +from bot.wecom_bot import create_bot_client +from bot.agent_client import close_http_client + +# 配置日志 +logging.basicConfig( + level=getattr(logging, Config.LOG_LEVEL, logging.INFO), + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +logger = logging.getLogger(__name__) + + +async def run_bot(): + """启动企业微信机器人 WebSocket 客户端""" + # 创建机器人客户端并注册事件处理器 + ws_client = create_bot_client() + logger.info("正在连接企业微信...") + + # 用 Event 等待认证完成 + authenticated = asyncio.Event() + + @ws_client.on("authenticated") + def _on_authenticated(): + logger.info("企业微信认证成功，机器人已上线") + authenticated.set() + + @ws_client.on("error") + def _on_error(err): + logger.error("SDK 错误: %s", err) + + try: + await ws_client.connect() + # 等待认证结果（最多 30 秒） + try: + await asyncio.wait_for(authenticated.wait(), timeout=30) + except asyncio.TimeoutError: + logger.error("认证超时（30秒），请检查 BOT_ID 和 BOT_SECRET 是否正确") + return + + # 认证成功，保持事件循环运行以持续接收消息 + logger.info("正在监听消息...") + await asyncio.Future() # 永远挂起，直到事件循环被取消 + + except asyncio.CancelledError: + logger.info("任务被取消") + except Exception as e: + logger.error("企业微信连接异常: %s", e) + finally: + await close_http_client() + logger.info("机器人已停止") + + +def main(): + """主函数""" + # 校验配置 + missing = Config.validate() + if missing: + logger.error("缺少必要配置项: %s", ", ".join(missing)) + logger.error("请复制 .env.example 为 .env 并填写配置") + sys.exit(1) + + logger.info("=" * 50) + logger.info(" Wechat-Bot 启动中") + logger.info(" Agent 地址: %s", Config.AGENT_BASE_URL) + logger.info(" 日志级别: %s", Config.LOG_LEVEL) + logger.info("=" * 50) + + try: + asyncio.run(run_bot()) + except KeyboardInterrupt: + logger.info("收到中断信号，正在停止...") + + logger.info("Wechat-Bot 已退出") + + +if __name__ == "__main__": + main() diff --git a/wechat-bot/pyproject.toml b/wechat-bot/pyproject.toml new file mode 100644 index 0000000..21a61ac --- /dev/null +++ b/wechat-bot/pyproject.toml @@ -0,0 +1,18 @@ +[project] +name = "wechat-bot" +version = "0.1.0" +description = "企业微信消息分发中转站 - 企业微信与 Agent 的桥梁" +requires-python = ">=3.10" +dependencies = [ + "wecom-aibot-python-sdk>=1.0.1", + "python-dotenv>=1.0.0", + "httpx>=0.27.0", + "aiosqlite>=0.20.0", +] + +[build-system] +requires = ["setuptools>=68.0"] +build-backend = "setuptools.build_meta" + +[tool.setuptools.packages.find] +include = ["bot*"] diff --git a/wechat-bot/requirements.txt b/wechat-bot/requirements.txt new file mode 100644 index 0000000..c1de7d7 --- /dev/null +++ b/wechat-bot/requirements.txt @@ -0,0 +1,3 @@ +wecom-aibot-python-sdk>=1.0.1 +python-dotenv>=1.0.0 +httpx>=0.27.0