WxHook是一个基于dll注入实现的python微信机器人框架，支持多种接口、高扩展性、多线程消息处理，让你轻松应对海量消息，为你的需求实现提供便捷灵活的支持。
支持的接口
- hook同步消息
- 取消hook同步消息
- hook日志
- 取消hook日志
- 检查登录状态
- 获取用户信息
- 发送文本消息
- 发送图片消息
- 发送文件消息
- 发送表情消息
- 发送小程序消息
- 发送群@消息
- 发送拍一拍消息
- 获取联系人列表
- 获取联系人详情
- 创建群聊
- 退出群聊
- 获取群详情
- 获取群成员列表
- 添加群成员
- 删除群成员
- 邀请群成员
- 修改群成员昵称
- 设置群置顶消息
- 移除群置顶消息
- 转发消息
- 获取朋友圈首页
- 获取朋友圈下一页
- 收藏消息
- 收藏图片
- 下载附件
- 转发公众号消息
- 转发公众号消息通过消息ID
- 解码图片
- 获取语音通过消息ID
- 图片文本识别
- 获取数据库句柄
- 执行SQL命令
- 测试
pip install wxhookclass Bot:
def __init__(
self,
on_login: Optional[Callable[["Bot", Event], Any]] = None,
on_before_message: Optional[Callable[["Bot", Event], Any]] = None,
on_after_message: Optional[Callable[["Bot", Event], Any]] = None,
on_start: Optional[Callable[["Bot"], Any]] = None,
on_stop: Optional[Callable[["Bot"], Any]] = None,
faked_version: Optional[str] = None
):
# ...初始化 Bot 实例，可以设置各种回调函数和微信版本伪装。
def handle(self, events: Union[List[str], str, None] = None, once: bool = False) -> Callable[[Callable], None]:
# ...用于注册事件处理函数的装饰器。
示例：
@bot.handle(events.TEXT_MESSAGE)
def on_message(bot: Bot, event: Event):
# 处理文本消息发送文本消息。
- URL:
/api/sendTextMsg - 方法: POST
- 参数:
- wxid: 接收者的微信ID
- msg: 要发送的文本消息
def send_text(self, wxid: str, msg: str) -> Response:
# ...发送图片消息。
- URL:
/api/sendImagesMsg - 方法: POST
- 参数:
- wxid: 接收者的微信ID
- imagePath: 图片文件的路径
def send_image(self, wxid: str, image_path: str) -> Response:
# ...发送表情消息。
- URL:
/api/sendCustomEmotion - 方法: POST
- 参数:
- wxid: 接收者的微信ID
- filePath: 表情文件的路径
def send_emotion(self, wxid: str, file_path: str) -> Response:
# ...发送文件消息。
- URL:
/api/sendFileMsg - 方法: POST
- 参数:
- wxid: 接收者的微信ID
- filePath: 文件的路径
def send_file(self, wxid: str, file_path: str) -> Response:
# ...发送小程序消息。
- URL:
/api/sendApplet - 方法: POST
- 参数:
- wxid: 接收者的微信ID
- waidConcat: 小程序联系人
- waid: 小程序ID
- appletWxid: 小程序的微信ID
- jsonParam: JSON参数
- headImgUrl: 头像图片URL
- mainImg: 主图片
- indexPage: 索引页面
def send_applet(
self,
wxid: str,
waid_contact: str,
waid: str,
applet_wxid: str,
json_param: str,
head_img_url: str,
main_img: str,
index_page: str
) -> Response:
# ...发送群@消息。
- URL:
/api/sendAtText - 方法: POST
- 参数:
- chatRoomId: 群聊ID
- wxids: 要@的用户ID列表，用逗号分隔
- msg: 要发送的消息内容
def send_room_at(self, room_id: str, wxids: List[str], msg: str) -> Response:
# ...发送拍一拍消息。
- URL:
/api/sendPatMsg - 方法: POST
- 参数:
- receiver: 群聊ID
- wxid: 要拍一拍的用户ID
def send_pat(self, room_id: str, wxid: str) -> Response:
# ...获取联系人列表。
- URL:
/api/getContactList - 方法: POST
def get_contacts(self) -> List[Contact]:
# ...获取联系人详情。
- URL:
/api/getContactProfile - 方法: POST
- 参数:
- wxid: 联系人的微信ID
def get_contact(self, wxid: str) -> ContactDetail:
# ...创建群聊。
- URL:
/api/createChatRoom - 方法: POST
- 参数:
- memberIds: 群成员ID列表，用逗号分隔
def create_room(self, member_ids: List[str]) -> Response:
# ...退出群聊。
- URL:
/api/quitChatRoom - 方法: POST
- 参数:
- chatRoomId: 群聊ID
def quit_room(self, room_id: str) -> Response:
# ...获取群详情。
- URL:
/api/getChatRoomDetailInfo - 方法: POST
- 参数:
- chatRoomId: 群聊ID
def get_room(self, room_id: str) -> Room:
# ...获取群成员列表。
- URL:
/api/getMemberFromChatRoom - 方法: POST
- 参数:
- chatRoomId: 群聊ID
def get_room_members(self, room_id: str) -> RoomMembers:
# ...添加群成员。
- URL:
/api/addMemberToChatRoom - 方法: POST
- 参数:
- chatRoomId: 群聊ID
- memberIds: 要添加的成员ID列表，用逗号分隔
def add_room_member(self, room_id: str, member_ids: List[str]) -> Response:
# ...删除群成员。
- URL:
/api/delMemberFromChatRoom - 方法: POST
- 参数:
- chatRoomId: 群聊ID
- memberIds: 要删除的成员ID列表，用逗号分隔
def delete_room_member(self, room_id: str, member_ids: List[str]) -> Response:
# ...邀请群成员。
- URL:
/api/InviteMemberToChatRoom - 方法: POST
- 参数:
- chatRoomId: 群聊ID
- memberIds: 要邀请的成员ID列表，用逗号分隔
def invite_room_member(self, room_id: str, member_ids: List[str]) -> Response:
# ...修改群成员昵称。
- URL:
/api/modifyNickname - 方法: POST
- 参数:
- chatRoomId: 群聊ID
- wxid: 要修改昵称的成员ID
- nickName: 新昵称
def modify_member_nickname(self, room_id: str, wxid: str, nickname: str) -> Response:
# ...获取朋友圈首页。
- URL:
/api/getSNSFirstPage - 方法: POST
def get_sns_first_page(self) -> Response:
# ...获取朋友圈下一页。
- URL:
/api/getSNSNextPage - 方法: POST
- 参数:
- snsId: 朋友圈ID
def get_sns_next_page(self, sns_id: int) -> Response:
# ...收藏消息。
- URL:
/api/addFavFromMsg - 方法: POST
- 参数:
- msgId: 消息ID
def collect_msg(self, msg_id: int) -> Response:
# ...收藏图片。
- URL:
/api/addFavFromImage - 方法: POST
- 参数:
- wxid: 图片所属的微信ID
- imagePath: 图片路径
def collect_image(self, wxid: str, image_path: str) -> Response:
# ...下载附件。
- URL:
/api/downloadAttach - 方法: POST
- 参数:
- msgId: 消息ID
def download_attachment(self, msg_id: int) -> Response:
# ...转发公众号消息。
- URL:
/api/forwardPublicMsg - 方法: POST
- 参数:
- wxid: 接收者的微信ID
- appName: 应用名称
- userName: 用户名
- title: 标题
- url: URL
- thumbUrl: 缩略图URL
- digest: 摘要
def forward_public_msg(
self,
wxid: str,
app_name: str,
username: str,
title: str,
url: str,
thumb_url: str,
digest: str
) -> Response:
# ...图片文本识别。
- URL:
/api/ocr - 方法: POST
- 参数:
- imagePath: 图片路径
def ocr(self, image_path: str) -> Response:
# ...获取数据库句柄。
- URL:
/api/getDBInfo - 方法: POST
def get_db_info(self) -> List[DB]:
# ...执行SQL命令。
- URL:
/api/execSql - 方法: POST
- 参数:
- dbHandle: 数据库句柄
- sql: SQL语句
def exec_sql(self, db_handle: int, sql: str) -> Response:
# ...# import os
# os.environ["WXHOOK_LOG_LEVEL"] = "INFO" # 修改日志输出级别
# os.environ["WXHOOK_LOG_FORMAT"] = "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{message}</level>" # 修改日志输出格式
from wxhook import Bot
from wxhook import events
from wxhook.model import Event
def on_login(bot: Bot, event: Event):
print("登录成功之后会触发这个函数")
def on_start(bot: Bot):
print("微信客户端打开之后会触发这个函数")
def on_stop(bot: Bot):
print("关闭微信客户端之前会触发这个函数")
def on_before_message(bot: Bot, event: Event):
print("消息事件处理之前")
def on_after_message(bot: Bot, event: Event):
print("消息事件处理之后")
bot = Bot(
faked_version="3.9.10.19", # 解除微信低版本限制
on_login=on_login,
on_start=on_start,
on_stop=on_stop,
on_before_message=on_before_message,
on_after_message=on_after_message
)
# 消息回调地址
# bot.set_webhook_url("http://127.0.0.1:8000")
@bot.handle(events.TEXT_MESSAGE)
def on_message(bot: Bot, event: Event):
bot.send_text("filehelper", "hello world!")
bot.run()