下载安装 wcferry 库

通过 pip 快速安装 wcferry

pip install wcferry

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

基本原理

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

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

Wcf 函数说明

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

电脑端检测登录微信

from wcferry import Wcfwcf = Wcf()

检测微信登陆状态

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

from wcferry import Wcfwcf = Wcf()print(wcf.is_login())

获取登录账号信息

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

from wcferry import Wcfwcf = Wcf()print(wcf.get_user_info())

运行结果

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

开辟线程监听群消息

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

from queue import Emptyfrom threading import Threadfrom wcferry import Wcf, WxMsgwcf = 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 Wcfwcf = Wcf()print(wcf.get_msg_types())

根据群名称查询群 wxid

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

from wcferry import Wcf, WxMsgwcf = 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 Noneroom_id = get_chatroom_roomid(wcf_rooms=wcf_rooms, room_name="测试群")

定时发送群文件

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

from apscheduler.schedulers.background import BackgroundSchedulerfrom datetime import datetimefrom wcferry import Wcfwcf = 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 Nonedef 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 Emptyfrom threading import Threadfrom wcferry import Wcf, WxMsgwcf = 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()

更新中······