Python 使用 WeChatFerry 部署微信机器人详细教程（持续更新）

下载安装 wcferry 库

通过 pip 快速安装 wcferry

pip install wcferry

免责声明：仅供学习和技术研究使用，不得用于任何商业或非法行为，否则后果自负。

基本原理

当微信收到消息时，抢在微信处理（显示到页面）前，先让工具处理，处理完之后再交还给原来的处理模块。需要发送消息时，模拟微信发送消息，组装好消息体，调用微信发送消息的模块。获取联系人，则是遍历一块特定的内存空间。通过好友验证，则是组装好验证信息，调用微信的验证模块。数据库相关功能，则是通过获取到数据库句柄，基于 sqlite3 的接口来执行。

博客地址：微信机器人 DIY 从 0 到 1

Wcf 函数说明

WeChatFerry：一个玩微信的工具（函数说明）

电脑端检测登录微信

from wcferry import Wcf

wcf = Wcf()

检测微信登陆状态

检查当前 PC 端微信登陆状态？

from wcferry import Wcf

wcf = Wcf()
print(wcf.is_login())

获取登录账号信息

获取当前 PC 端微信账号信息？

from wcferry import Wcf

wcf = Wcf()
print(wcf.get_user_info())

运行结果

{'wxid': 'wxid_***', 'name': '字里行间', 'mobile': '195********', 'home': 'C:\\Users\\Administrator\\Documents\\WeChat Files\\'}

开辟线程监听群消息

开启线程监听消息：判断是否是群消息？

from queue import Empty
from threading import Thread

from wcferry import Wcf, WxMsg

wcf = Wcf()


def processMsg(msg: WxMsg):
    if msg.from_group():
        print(msg.content)


def enableReceivingMsg():
    def innerWcFerryProcessMsg():
        while wcf.is_receiving_msg():
            try:
                msg = wcf.get_msg()
                processMsg(msg)
            except Empty:
                continue
            except Exception as e:
                print(f"ERROR: {e}")

    wcf.enable_receiving_msg()
    Thread(target=innerWcFerryProcessMsg, name="ListenMessageThread", daemon=True).start()


enableReceivingMsg()

wcf.keep_running()

微信消息属性说明

class WxMsg() 微信消息属性说明

属性说明

消息类型

from wcferry import Wcf

wcf = Wcf()
print(wcf.get_msg_types())

根据群名称查询群 wxid

特别注意：Wcf 没有提供根据群名称查询群 wxid 功能。我们可以先获取全部联系人数据（微信好友、微信群等等），基于 wxid 进行区分，因为微信群 wxid 后缀都是 “chatroom” 结尾。

from wcferry import Wcf, WxMsg

wcf = Wcf()

wcf_rooms = []

for contact in wcf.get_contacts():
    if contact['wxid'].endswith("chatroom"):
        wcf_rooms.append(contact)

def get_chatroom_roomid(wcf_rooms: list, room_name: str):
    for room in wcf_rooms:
        if room['name'] == room_name:
            return room['wxid']
    return None

room_id = get_chatroom_roomid(wcf_rooms=wcf_rooms, room_name="测试群")

定时发送群文件

如何进行定时发送群文件？通过 aspschedule 第三方库实现定时任务，然后调用 wcf.send_file 函数执行发送文件的消息。

from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime

from wcferry import Wcf

wcf = Wcf()

wcf_rooms = []

for contact in wcf.get_contacts():
    if contact['wxid'].endswith("chatroom"):
        wcf_rooms.append(contact)


def get_chatroom_roomid(wcf_rooms: list, room_name: str):
    for room in wcf_rooms:
        if room['name'] == room_name:
            return room['wxid']
    return None


def schedule_task_job(room_id: str, wcf: Wcf):
    wcf.send_file(path="test.txt", receiver=room_id)


customize_time = "2024-05-09 09:10:10"
customize_room = "唤醒手腕测试群"
run_date = datetime.strptime(customize_time, "%Y-%m-%d %H:%M:%S")
room_id = get_chatroom_roomid(wcf_rooms, customize_room)

scheduler = BackgroundScheduler()
scheduler.add_job(schedule_task_job, args=(room_id, wcf), run_date=run_date)
scheduler.start()

wcf.keep_running()

运行结果

监听保存语音消息

from queue import Empty
from threading import Thread

from wcferry import Wcf, WxMsg

wcf = Wcf()


def processMsg(msg: WxMsg):
    if msg.from_group():
        response = wcf.get_audio_msg(id=msg.id, dir=f"audio")
        print("语音地址：" + response)


def enableReceivingMsg():
    def innerWcFerryProcessMsg():
        while wcf.is_receiving_msg():
            try:
                msg = wcf.get_msg()
                processMsg(msg)
            except Empty:
                continue
            except Exception as e:
                print(f"ERROR: {e}")

    wcf.enable_receiving_msg()
    Thread(target=innerWcFerryProcessMsg, name="ListenMessageThread", daemon=True).start()


enableReceivingMsg()

wcf.keep_running()

更新中······

作者：唤醒手腕