|
作为国民级应用，你早已离不开微信了吧~ 把你的微信接入 GPT ，打造一个随时待命的 AI 助理，怎么样？ 微信AI助手系列已经陆续大家分享了三种实现： 不过，多少都有一定限制，其中智能微秘书需要付费才能把 Coze 、FastGPT、Dify 等知识库工具接入微信。 今天又发现一个开源项目-wechatbot-webhook，项目介绍干脆利落：抹平开发障碍，一个小小的微信机器人webhook。 大白话原理：使用 web 协议登录小号微信，接收消息发给指定的 API 进行处理，再将处理结果回复给大号微信。 大部分开源微信机器人项目非常复杂，且很难嵌入到自己的项目中，而wechatbot-webhook可定制化非常高，你可以用任意擅长的语言进行开发，简直就是开发者的福音。 今天，就手把手带大家，以一种更简洁&优雅的方式搭建你的微信超级 AI 助理，保证丝滑有效！ 即便没有任何编程知识，只要跟着一步步实操，你也可以拥有! 进入实操之前，小伙伴们可以思考一个问题：一个完整的微信机器人至少需要实现几个功能？ 你别看很多项目开发的一堆花里胡哨的功能，其实最核心的就是三个： 扫码登录； 发送消息； 接收消息。 下面，我们一一搞定它~ 友情提醒：注册一个小号使用，严禁用于违法用途（如发送广告/群发/诈骗、色情、政治等内容），否则封号是早晚的事哦。 1. 如何扫码登录为了实现用你的小号扫码登录，我们首先需要将 wechatbot-webhook 成功部署。 这里依然采用最简单的 docker 方式进行部署，还不知道如何使用 docker 的小伙伴，可以看上一篇。 以下展示均基于云服务器。如果你在本地部署，流程和指令都是一致的。 第一步：拉取镜像： sudo docker pull dannicool/docker-wechatbot-webhook第二步：启动容器: sudo docker run -d --name wxBotWebhook --restart unless-stopped -p 3001:3001 -v ~/wxBot_logs:/app/log -e ACCEPT_RECVD_MSG_MYSELF=true -e RECVD_MSG_API=:3000/receive/ -e LOGIN_API_TOKEN=123 dannicool/docker-wechatbot-webhook这里有几个参数需要说明一下，环境变量参数需要使用 -e： ACCEPT_RECVD_MSG_MYSELF=true：机器人能接收自己发的消息 RECVD_MSG_API=:3000/receive/：接收消息的 API 地址，实现处理接收消息的逻辑，如果你不想实现这个功能，可以不填，默认为空 LOGIN_API_TOKEN=123：自定义登录 API token，可以不填，默认自动生成一个 第二步：查看日志并登录： sudo docker logs -f wxBotWebhook其中，-f 表示 “follow”，即实时跟踪日志输出。
当然，也可以通过如下地址：:3001/login?token=123，实现登录 or 查看登录情况。下图显示已经登录，如果没登录，这里会出现二维码。
最后，给大家看下这个项目的内存占用情况：
才 100M 的内存占用，放到嵌入式移动端去跑吧，连电费都帮你省了！ 2. 如何发送消息项目部署成功后，拿到发送消息的 API 地址：:3001/webhook/msg/v2?token=123 2.1 发送私聊信息我们以 Python 为例，写一段代码，给你的大号发一条信息试试： import requests import json url = f':3001/webhook/msg/v2?token=123' # 发给好友消息 data = { "to": "账号昵称", "data": {"content": "你好👋"} } # 发送 POST 请求 response = requests.post(url, headers={'Content-Type': 'application/json'}, data=json.dumps(data)) print(response.status_code) print(response.json()) 2.2 发送群聊信息想发送群聊信息怎么办？data 结构体中加上参数 "isRoom": true 即可。 # 发给群消息 data = { "to": "群昵称", "isRoom": true, "data": {"content": "你好👋"} } 2.3 发送 Url 文件要发文件怎么办？依然只需修改 data 结构即可： # 给 url 拼接 query 参数 $alias 可用于指定发送给目标的文件名 data = { "to": "账号昵称", "data": {"type": "fileUrl", "content": "https://download.samplelib.com/jpeg/sample-clouds-400x300.jpg?$alias=cloud.jpg"} } 2.4 发送本地文件要发送本地文件怎么办？ 注意：这里的请求地址变了（不用v2），且 headers 中的发送类型需改为ContentType: multipart/form-data，因为 request 库中默认就是这个类型，所以无需手动添加 headers。 url = f':3001/webhook/msg?token=123' data = { "to": "账号昵称", } response = requests.post(url, data=data, files={'content': open('pipe.py', 'rb')}) print(response.status_code) print(response.text)注意：上述请求的URL中不包含v2。 2.5 一次发多条信息 data = { "to": "账号昵称", "data": [{"content": "你好👋"}, {"type": "fileUrl", "content": "https://download.samplelib.com/jpeg/sample-clouds-400x300.jpg?$alias=cloud.jpg"}] } 2.6 给多人群发 data = [ { "to": "昵称1", "data": {"content": "你好👋"} }, { "to": "昵称2", "data": {"content": "你好👋"} } ]最终的返回结果，格式如下： {'success': True, 'message': 'Message sent successfully', 'task': {'successCount': 1, 'totalCount': 1, 'failedCount': 0, 'reject': [], 'sentFailed': [], 'notFound': []}}给大家看下，通过上述命令，机器人给发送的消息和文件：
接收消息相对复杂一些，还记得刚部署服务时填入的RECVD_MSG_API=:3000/receive/字段么？ 接下来，我们需要实现:3000/receive/这个服务端端口，让它能够接收并处理发给机器人的消息。 3.1 服务端实现既然用 Python，那么用 fastapi 来写一个服务端函数，再合适不过了。 第一步：新建一个 server.py 文件，编写如下代码： from fastapi import FastAPI, UploadFile, Form from fastapi.responses import JSONResponse import uvicorn app = FastAPI() @app.post("/receive") async def receive_message( type: str = Form(...), content: str = Form(...), source: str = Form(...), isMentioned: str = Form(...), isMsgFromSelf: str = Form(...), ): # 处理请求数据 response_data = { "type": type, "content": content, "source": source, "isMentioned": isMentioned, "isMsgFromSelf": isMsgFromSelf, } try: # 填写处理逻辑-开始 print(response_data) # 填写处理逻辑-结束 return JSONResponse(content={"status": "success", "data": response_data}) except Exception as e: print(e) return JSONResponse(content={"status": "error", "data": "处理失败"}) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=3000)注意：上述处理逻辑部分，完全可以定制化，比如接入 GPT 处理消息后进行回复。 第二步：启动服务器，把服务端程序放到后台去运行吧，即便关闭终端，程序也不会被杀。 nohup python3 server.py > log.txt 2>&1 &第三步：确保防火墙允许访问指定的端口（比如我这里的3000）。可以使用以下命令检查和修改防火墙设置： sudo ufw allow 3000 sudo ufw status 3.2 客户端测试最后，我们来编写客户端函数，来测试下能否成功调用服务端： import requests def send_message(api_url, message_type, content, source, is_mentioned="0", is_msg_from_self="0"): data = { "type": message_type, "content": content, "source": source, "isMentioned": is_mentioned, "isMsgFromSelf": is_msg_from_self, } response = requests.post(api_url, data=data) if response.status_code == 200: return response.json() return None # 示例调用 if __name__ == "__main__": api_url = ":3000/receive/" message_type = "text" content = "你好" source = '{"room":"","to":{"id":"@f387910fa45","payload":{"alias":"","avatar":""}},"from":{"id":"@6b5111dcc269b6901fbb58","payload":{"city":"Mars","name":"Daniel"}}}' response = send_message(api_url, message_type, content, source) print(response)成功搞定！ {'status': 'success', 'data': {'type': 'text', 'content': '你好', 'source': '{"room":"","to":{"id":"@f387910fa45","payload":{"alias":"","avatar":""}},"from":{"id":"@6b5111dcc269b6901fbb58","payload":{"city":"Mars","name":"Daniel"}}}', 'isMentioned': '0', 'isMsgFromSelf': '0'}}再次提醒：确保你的服务端端口防火墙已经被打开哦。 4. 文末福利：写个定时任务说了半天，这玩意儿到底有啥用？ 接口都有了，想干点啥还不容易？ 比如我做了一个小工具：每天定时给我的大号发送一条【摸鱼】消息。 这里需要用到一个摸鱼接口：，浏览器也可直达！ 咋用 Python 写个定时任务呢？ 首先，需要安装两个库：schedule 用于执行定时任务，pytz 用于指定时区，以获得正确的时间信息。 pip install schedule pip install pytz然后，编写如下代码： import json import pytz import time import requests import schedule from datetime import datetime # 设置上海时区 shanghai_tz = pytz.timezone('Asia/Shanghai') def get_moyu_text(): response = requests.get("https://moyu.awsl.icu/api/moyu_json?day=10") return response.text if response.status_code == 200 else '接口调用失败' def send_message(): url = f':3001/webhook/msg/v2?token=123' content = get_moyu_text() content = content.replace('\\n', '\n').replace('"', '') data = { "to": "昵称", "data": {"content": content} } response = requests.post(url, headers={'Content-Type': 'application/json'}, data=json.dumps(data)) if response.status_code == 200: print(response.json()) else: print('发送接口调用失败') def schedule_job(t='10:00'): now = datetime.now(shanghai_tz) if now.strftime('%H:%M').strip() == t: send_message() if __name__ == '__main__': # 设置每天的定时任务 schedule.every().minute.do(schedule_job) # 每天10:00发送信息 while True: schedule.run_pending() time.sleep(1)把这个定时任务的脚本放到后台去跑吧~ nohup python3 your_script.py > log.txt 2>&1 &注：上述代码有三个坑，折腾了不少时间，帮大家排个雷： 关于时区：如果是租用的海外的云服务器，一定记得要修改时区为’Asia/Shanghai’，否则定时发送的时间是错乱的； 关于 content：接口返回的是JSON格式的数据，换行符会被编码为 \\n 来确保JSON字符串的正确格式，所以需要content.replace('\\n', '\n')，才能得到markdown格式的数据。 关于now.strftime('%H:%M')：这个字符串居然有空格，必须加上.strip()，才能匹配上。 发送成功！
怎么样，是不是超简单？ 动手试试吧，在搭建过程中遇到任何问题，有什么好的想法，欢迎评论区留言~ 说好的 AI 助手，在哪里？ (责任编辑：) |
