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

搭建微信机器人的第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 助理~

如果本文对你有帮助,不妨点个免费的赞和收藏备用。

相关文章

365结束投注 滤镜总排行榜

滤镜总排行榜

🗓️ 06-29 👁️ 2123
365结束投注 使用git命令提交代码 到远程仓库(命令提交;详细版本)
365体育投注网址亚洲下载 小鬼離世半年…峮峮默默親赴祭拜 放不下1年舊情悲吐思念
365体育投注网址亚洲下载 青松基金投资机构信息

青松基金投资机构信息

🗓️ 07-06 👁️ 6170
365结束投注 如何给视频添加字幕和配音

如何给视频添加字幕和配音

🗓️ 07-06 👁️ 7387
365结束投注 探访沩山十景 追寻源远流长的文化根脉