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

作为国民级应用，你早已离不开微信了吧~

把你的微信接入 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
  • 1

第二步：启动容器:

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
  • 1

这里有几个参数需要说明一下，环境变量参数需要使用 -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
  • 1

其中，-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())
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16

2.2 发送群聊信息

想发送群聊信息怎么办？data 结构体中加上参数 "isRoom": true 即可。

# 发给群消息
data = {
    "to": "群昵称",
    "isRoom": true,
    "data": {"content": "你好?"}
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

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"}
}
  • 1
  • 2
  • 3
  • 4
  • 5

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)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

注意：上述请求的URL中不包含v2。

2.5 一次发多条信息

data = {
    "to": "账号昵称",
    "data": [{"content": "你好?"}, {"type": "fileUrl", "content": "https://download.samplelib.com/jpeg/sample-clouds-400x300.jpg?$alias=cloud.jpg"}]
}
  • 1
  • 2
  • 3
  • 4

2.6 给多人群发

data = [
  {
      "to": "昵称1",
      "data": {"content": "你好?"}
  },
  {
      "to": "昵称2",
      "data": {"content": "你好?"}
  }
]
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

最终的返回结果，格式如下：

{'success': True, 'message': 'Message sent successfully', 'task': {'successCount': 1, 'totalCount': 1, 'failedCount': 0, 'reject': [], 'sentFailed': [], 'notFound': []}}
  • 1

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

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)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33

注意：上述处理逻辑部分，完全可以定制化，比如接入 GPT 处理消息后进行回复。

第二步：启动服务器，把服务端程序放到后台去运行吧，即便关闭终端，程序也不会被杀。

nohup python3 server.py > log.txt 2>&1 &
  • 1

第三步：确保防火墙允许访问指定的端口（比如我这里的3000）。可以使用以下命令检查和修改防火墙设置：

sudo ufw allow 3000
sudo ufw status
  • 1
  • 2

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)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25

成功搞定！

{'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'}}
  • 1

再次提醒：确保你的服务端端口防火墙已经被打开哦。

4. 文末福利：写个定时任务

说了半天，这玩意儿到底有啥用？

接口都有了，想干点啥还不容易？

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

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

咋用 Python 写个定时任务呢？

首先，需要安装两个库：schedule 用于执行定时任务，pytz 用于指定时区，以获得正确的时间信息。

pip install schedule
pip install pytz
  • 1
  • 2

然后，编写如下代码：

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)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39

把这个定时任务的脚本放到后台去跑吧~

nohup python3 your_script.py > log.txt 2>&1 &
  • 1

注：上述代码有三个坑，折腾了不少时间，帮大家排个雷：

  • 关于时区：如果是租用的海外的云服务器，一定记得要修改时区为’Asia/Shanghai’，否则定时发送的时间是错乱的；
  • 关于 content：接口返回的是JSON格式的数据，换行符会被编码为 \\n 来确保JSON字符串的正确格式，所以需要content.replace('\\n', '\n')，才能得到markdown格式的数据。
  • 关于now.strftime('%H:%M')：这个字符串居然有空格，必须加上.strip()，才能匹配上。

发送成功！

写在最后

怎么样，是不是超简单？

动手试试吧，在搭建过程中遇到任何问题，有什么好的想法，欢迎评论区留言~

说好的 AI 助手，在哪里？

篇幅有点长了，先简单做到这里，下一篇带大家：接入 Coze 和 Dify 等外部工具，打造全能 AI 助理~

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

注：本文转载自blog.csdn.net的AI码上来的文章"https://blog.csdn.net/u010522887/article/details/141348878"。版权归原作者所有，此博客不拥有其著作权，亦不承担相应法律责任。如有侵权，请联系我们删除。
复制链接
相关推荐
发表评论

评论记录：

未查询到任何数据！