跳转到主内容
极星编程网:以代码为星,赴技术山海!

搭建微信机器人的第4种方式,免费开源,轻量高效

作为国民级应用,你早已离不开微信了吧~ 把你的微信接入 GPT ,打造一个随时待命的 AI 助理,怎么样?

微信AI助手系列

已经陆续大家分享了三种实现: 手把手搭建微信机器人,帮你雇一个24小时在线的个人 AI 助理 NGCBot-打造基于Hook机制的微信机器人,Windows本地部署 搭建微信机器人的第3种方式,我又造了一个24H在线的个人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=http://129.150.39.xxx:3000/receive/ -e LOGIN_API_TOKEN=123 dannicool/docker-wechatbot-webhook

这里有几个参数需要说明一下,环境变量参数需要使用 -e:

ACCEPT_RECVD_MSG_MYSELF=true

:机器人能接收自己发的消息

RECVD_MSG_API=http://129.150.39.xxx:3000/receive/

:

接收消息

的 API 地址,实现处理接收消息的逻辑,如果你不想实现这个功能,可以不填,默认为空

LOGIN_API_TOKEN=123

:自定义登录 API token,可以不填,默认自动生成一个 第二步:查看日志并登录:

sudo docker logs -f wxBotWebhook

其中,

-f

表示 “follow”,即实时跟踪日志输出。

当然,也可以通过如下地址:http://129.150.39.xxx:3001/login?token=123,实现登录 or 查看登录情况。

下图显示已经登录,如果没登录,这里会出现二维码。

最后,给大家看下这个项目的内存占用情况:

才 100M 的内存占用,放到嵌入式移动端去跑吧,连电费都帮你省了!

2. 如何发送消息 项目部署成功后,拿到

发送消息

的 API 地址:http://129.150.39.xxx:3001/webhook/msg/v2?token=123 2.1 发送私聊信息 我们以 Python 为例,写一段代码,给你的大号发一条信息试试:

import requests

import json

url = f'http://129.150.39.xxx: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'http://129.150.39.xxx: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': []}}

给大家看下,通过上述命令,机器人给发送的消息和文件:

3. 如何接收消息 接收消息相对复杂一些,还记得刚部署服务时填入的

RECVD_MSG_API=http://129.150.39.xxx:3000/receive/

字段么?

接下来,我们需要实现

http://129.150.39.xxx: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 = "http://129.150.39.xxx: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. 文末福利:写个定时任务 说了半天,这玩意儿到底有啥用?

接口都有了,想干点啥还不容易?

比如我做了一个小工具:每天定时给我的大号发送一条【摸鱼】消息。

这里需要用到一个摸鱼接口: https://moyu.awsl.icu/api/moyu_json?day=10 ,浏览器也可直达!

咋用 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'http://129.150.39.xxx: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 助手,在哪里?

篇幅有点长了,先简单做到这里,下一篇带大家:接入

Coze

和

Dify

等外部工具,打造全能 AI 助理~ 如果本文对你有帮助,不妨点个 免费的赞 和 收藏 备用。

相关文章