WechatAPIClient

基础

class WechatAPI.Client.base.Proxy(ip: str, port: int, username: str = '', password: str = '')[源代码]

基类:object

代理(无效果,别用!)

参数:
  • ip (str) -- 代理服务器IP地址

  • port (int) -- 代理服务器端口

  • username (str, optional) -- 代理认证用户名. 默认为空字符串

  • password (str, optional) -- 代理认证密码. 默认为空字符串

ip: str
password: str = ''
port: int
username: str = ''
class WechatAPI.Client.base.Section(data_len: int, start_pos: int)[源代码]

基类:object

数据段配置类

参数:
  • data_len (int) -- 数据长度

  • start_pos (int) -- 起始位置

data_len: int
start_pos: int
class WechatAPI.Client.base.WechatAPIClientBase(ip: str, port: int)[源代码]

基类:object

微信API客户端基类

参数:
  • ip (str) -- 服务器IP地址

  • port (int) -- 服务器端口

变量:
  • wxid (str) -- 微信ID

  • nickname (str) -- 昵称

  • alias (str) -- 别名

  • phone (str) -- 手机号

  • ignore_protect (bool) -- 是否忽略保护机制

static error_handler(json_resp)[源代码]

处理API响应中的错误码

参数:

json_resp (dict) -- API响应的JSON数据

抛出:
  • ValueError -- 参数错误时抛出

  • MarshallingError -- 序列化错误时抛出

  • UnmarshallingError -- 反序列化错误时抛出

  • MMTLSError -- MMTLS初始化错误时抛出

  • PacketError -- 数据包长度错误时抛出

  • UserLoggedOut -- 用户已退出登录时抛出

  • ParsePacketError -- 解析数据包错误时抛出

  • DatabaseError -- 数据库错误时抛出

  • Exception -- 其他类型错误时抛出

登录

class WechatAPI.Client.login.LoginMixin(ip: str, port: int)[源代码]

基类:WechatAPIClientBase

async awaken_login(wxid: str = '') str[源代码]

唤醒登录。

参数:

wxid (str, optional) -- 要唤醒的微信ID. Defaults to "".

返回:

返回新的登录UUID

返回类型:

str

抛出:
  • Exception -- 如果未提供wxid且未登录

  • LoginError -- 如果无法获取UUID

  • 根据error_handler处理错误 --

async check_login_uuid(uuid: str, device_id: str = '') tuple[bool, dict | int][源代码]

检查登录的UUID状态。

参数:
  • uuid (str) -- 登录的UUID

  • device_id (str, optional) -- 设备ID. Defaults to "".

返回:

如果登录成功返回(True, 用户信息),否则返回(False, 过期时间)

返回类型:

tuple[bool, Union[dict, int]]

抛出:

根据error_handler处理错误 --

static create_device_id(s: str = '') str[源代码]

生成设备ID。

参数:

s (str, optional) -- 用于生成ID的字符串. Defaults to "".

返回:

返回生成的设备ID

返回类型:

str

static create_device_name() str[源代码]

生成一个随机的设备名。

返回:

返回生成的设备名

返回类型:

str

async get_auto_heartbeat_status() bool[源代码]

获取自动心跳状态。

返回:

如果正在运行返回True,否则返回False

返回类型:

bool

抛出:
  • UserLoggedOut -- 如果未登录时调用

  • 根据error_handler处理错误 --

async get_cached_info(wxid: str = None) dict[源代码]

获取登录缓存信息。

参数:

wxid (str, optional) -- 要查询的微信ID. Defaults to None.

返回:

返回缓存信息,如果未提供wxid且未登录返回空字典

返回类型:

dict

async get_qr_code(device_name: str, device_id: str = '', proxy: ~WechatAPI.Client.base.Proxy = None, print_qr: bool = False) -> (<class 'str'>, <class 'str'>)[源代码]

获取登录二维码。

参数:
  • device_name (str) -- 设备名称

  • device_id (str, optional) -- 设备ID. Defaults to "".

  • proxy (Proxy, optional) -- 代理信息. Defaults to None.

  • print_qr (bool, optional) -- 是否在控制台打印二维码. Defaults to False.

返回:

返回登录二维码的UUID和URL

返回类型:

tuple[str, str]

抛出:

根据error_handler处理错误 --

async heartbeat() bool[源代码]

发送心跳包。

返回:

成功返回True,否则返回False

返回类型:

bool

抛出:
  • UserLoggedOut -- 如果未登录时调用

  • 根据error_handler处理错误 --

async is_running() bool[源代码]

检查WechatAPI是否在运行。

返回:

如果WechatAPI正在运行返回True,否则返回False。

返回类型:

bool

async log_out() bool[源代码]

登出当前账号。

返回:

登出成功返回True,否则返回False

返回类型:

bool

抛出:
  • UserLoggedOut -- 如果未登录时调用

  • 根据error_handler处理错误 --

async start_auto_heartbeat() bool[源代码]

开始自动心跳。

返回:

成功返回True,否则返回False

返回类型:

bool

抛出:
  • UserLoggedOut -- 如果未登录时调用

  • 根据error_handler处理错误 --

async stop_auto_heartbeat() bool[源代码]

停止自动心跳。

返回:

成功返回True,否则返回False

返回类型:

bool

抛出:
  • UserLoggedOut -- 如果未登录时调用

  • 根据error_handler处理错误 --

消息

class WechatAPI.Client.message.MessageMixin(ip: str, port: int)[源代码]

基类:WechatAPIClientBase

async _process_message_queue()[源代码]

处理消息队列的异步方法

async _queue_message(func, *args, **kwargs)[源代码]

将消息添加到队列

async _send_text_message(wxid: str, content: str, at: list[str] = None) tuple[int, int, int][源代码]

实际发送文本消息的方法

async revoke_message(wxid: str, client_msg_id: int, create_time: int, new_msg_id: int) bool[源代码]

撤回消息。

参数:
  • wxid (str) -- 接收人wxid

  • client_msg_id (int) -- 发送消息的返回值

  • create_time (int) -- 发送消息的返回值

  • new_msg_id (int) -- 发送消息的返回值

返回:

成功返回True,失败返回False

返回类型:

bool

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 登录新设备后4小时内操作

  • 根据error_handler处理错误 --

async send_app_message(wxid: str, xml: str, type: int) tuple[str, int, int][源代码]

发送应用消息。

参数:
  • wxid (str) -- 接收人wxid

  • xml (str) -- 应用消息的xml内容

  • type (int) -- 应用消息类型

返回:

返回(ClientMsgid, CreateTime, NewMsgId)

返回类型:

tuple[str, int, int]

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 登录新设备后4小时内操作

  • 根据error_handler处理错误 --

async send_card_message(wxid: str, card_wxid: str, card_nickname: str, card_alias: str = '') tuple[int, int, int][源代码]

发送名片消息。

参数:
  • wxid (str) -- 接收人wxid

  • card_wxid (str) -- 名片用户的wxid

  • card_nickname (str) -- 名片用户的昵称

  • card_alias (str, optional) -- 名片用户的备注. Defaults to "".

返回:

返回(ClientMsgid, CreateTime, NewMsgId)

返回类型:

tuple[int, int, int]

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 登录新设备后4小时内操作

  • 根据error_handler处理错误 --

async send_cdn_file_msg(wxid: str, xml: str) tuple[str, int, int][源代码]

转发文件消息。

参数:
  • wxid (str) -- 接收人wxid

  • xml (str) -- 要转发的文件消息xml内容

返回:

返回(ClientMsgid, CreateTime, NewMsgId)

返回类型:

tuple[str, int, int]

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 登录新设备后4小时内操作

  • 根据error_handler处理错误 --

async send_cdn_img_msg(wxid: str, xml: str) tuple[str, int, int][源代码]

转发图片消息。

参数:
  • wxid (str) -- 接收人wxid

  • xml (str) -- 要转发的图片消息xml内容

返回:

返回(ClientImgId, CreateTime, NewMsgId)

返回类型:

tuple[str, int, int]

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 登录新设备后4小时内操作

  • 根据error_handler处理错误 --

async send_cdn_video_msg(wxid: str, xml: str) tuple[str, int][源代码]

转发视频消息。

参数:
  • wxid (str) -- 接收人wxid

  • xml (str) -- 要转发的视频消息xml内容

返回:

返回(ClientMsgid, NewMsgId)

返回类型:

tuple[str, int]

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 登录新设备后4小时内操作

  • 根据error_handler处理错误 --

async send_emoji_message(wxid: str, md5: str, total_length: int) list[dict][源代码]

发送表情消息。

参数:
  • wxid (str) -- 接收人wxid

  • md5 (str) -- 表情md5值

  • total_length (int) -- 表情总长度

返回:

返回表情项列表(list of emojiItem)

返回类型:

list[dict]

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 登录新设备后4小时内操作

  • 根据error_handler处理错误 --

async send_image_message(wxid: str, image: str | bytes | PathLike) tuple[int, int, int][源代码]

发送图片消息。

参数:
  • wxid (str) -- 接收人wxid

  • image (str, byte, os.PathLike) -- 图片,支持base64字符串,图片byte,图片路径

返回:

返回(ClientImgId, CreateTime, NewMsgId)

返回类型:

tuple[int, int, int]

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 登录新设备后4小时内操作

  • ValueError -- image_path和image_base64都为空或都不为空时

  • 根据error_handler处理错误 --

发送链接消息。

参数:
  • wxid (str) -- 接收人wxid

  • url (str) -- 跳转链接

  • title (str, optional) -- 标题. Defaults to "".

  • description (str, optional) -- 描述. Defaults to "".

  • thumb_url (str, optional) -- 缩略图链接. Defaults to "".

返回:

返回(ClientMsgid, CreateTime, NewMsgId)

返回类型:

tuple[str, int, int]

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 登录新设备后4小时内操作

  • 根据error_handler处理错误 --

async send_text_message(wxid: str, content: str, at: list | str = '') tuple[int, int, int][源代码]

发送文本消息。

参数:
  • wxid (str) -- 接收人wxid

  • content (str) -- 消息内容

  • at (list, str, optional) -- 要@的用户

返回:

返回(ClientMsgid, CreateTime, NewMsgId)

返回类型:

tuple[int, int, int]

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 登录新设备后4小时内操作

  • 根据error_handler处理错误 --

async send_video_message(wxid: str, video: str | bytes | ~os.PathLike, image: [<class 'str'>, <class 'bytes'>, <class 'os.PathLike'>] = None)[源代码]

发送视频消息。不推荐使用,上传速度很慢300KB/s。如要使用,可压缩视频,或者发送链接卡片而不是视频。

参数:
  • wxid (str) -- 接收人wxid

  • video (str, bytes, os.PathLike) -- 视频 接受base64字符串,字节,文件路径

  • image (str, bytes, os.PathLike) -- 视频封面图片 接受base64字符串,字节,文件路径

返回:

返回(ClientMsgid, NewMsgId)

返回类型:

tuple[int, int]

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 登录新设备后4小时内操作

  • ValueError -- 视频或图片参数都为空或都不为空时

  • 根据error_handler处理错误 --

async send_voice_message(wxid: str, voice: str | bytes | PathLike, format: str = 'amr') tuple[int, int, int][源代码]

发送语音消息。

参数:
  • wxid (str) -- 接收人wxid

  • voice (str, bytes, os.PathLike) -- 语音 接受base64字符串,字节,文件路径

  • format (str, optional) -- 语音格式,支持amr/wav/mp3. Defaults to "amr".

返回:

返回(ClientMsgid, CreateTime, NewMsgId)

返回类型:

tuple[int, int, int]

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 登录新设备后4小时内操作

  • ValueError -- voice_path和voice_base64都为空或都不为空时,或format不支持时

  • 根据error_handler处理错误 --

async sync_message() dict[源代码]

同步消息。

返回:

返回同步到的消息数据

返回类型:

dict

抛出:
  • UserLoggedOut -- 未登录时调用

  • 根据error_handler处理错误 --

用户

class WechatAPI.Client.user.UserMixin(ip: str, port: int)[源代码]

基类:WechatAPIClientBase

async get_my_qrcode(style: int = 0) str[源代码]

获取个人二维码。

参数:

style (int, optional) -- 二维码样式. Defaults to 0.

返回:

图片的base64编码字符串

返回类型:

str

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 风控保护: 新设备登录后4小时内请挂机

  • 根据error_handler处理错误 --

async get_profile(wxid: str = None) dict[源代码]

获取用户信息。

参数:

wxid (str, optional) -- 用户wxid. Defaults to None.

返回:

用户信息字典

返回类型:

dict

抛出:
  • UserLoggedOut -- 未登录时调用

  • 根据error_handler处理错误 --

async is_logged_in(wxid: str = None) bool[源代码]

检查是否登录。

参数:

wxid (str, optional) -- 用户wxid. Defaults to None.

返回:

已登录返回True,未登录返回False

返回类型:

bool

群聊

class WechatAPI.Client.chatroom.ChatroomMixin(ip: str, port: int)[源代码]

基类:WechatAPIClientBase

async add_chatroom_member(chatroom: str, wxid: str) bool[源代码]

添加群成员(群聊最多40人)

参数:
  • chatroom -- 群聊wxid

  • wxid -- 要添加的wxid

返回:

成功返回True, 失败False或者报错

返回类型:

bool

async get_chatroom_announce(chatroom: str) dict[源代码]

获取群聊公告

参数:

chatroom -- 群聊id

返回:

群聊信息字典

返回类型:

dict

async get_chatroom_info(chatroom: str) dict[源代码]

获取群聊信息

参数:

chatroom -- 群聊id

返回:

群聊信息字典

返回类型:

dict

async get_chatroom_member_list(chatroom: str) list[dict][源代码]

获取群聊成员列表

参数:

chatroom -- 群聊id

返回:

群聊成员列表

返回类型:

list[dict]

async get_chatroom_qrcode(chatroom: str) dict[str, Any][源代码]

获取群聊二维码

参数:

chatroom -- 群聊id

返回:

{"base64": 二维码的base64, "description": 二维码描述}

返回类型:

dict

async invite_chatroom_member(wxid: str | list, chatroom: str) bool[源代码]

邀请群聊成员(群聊大于40人)

参数:
  • wxid -- 要邀请的用户wxid或wxid列表

  • chatroom -- 群聊id

返回:

成功返回True, 失败False或者报错

返回类型:

bool

好友

class WechatAPI.Client.friend.FriendMixin(ip: str, port: int)[源代码]

基类:WechatAPIClientBase

async accept_friend(scene: int, v1: str, v2: str) bool[源代码]

接受好友请求

主动添加好友单天上限如下所示:1小时内上限为 5个,超过上限时,无法发出好友请求,也收不到好友请求。

  • 新账号:5/天

  • 注册超过7天:10个/天

  • 注册满3个月&&近期登录过该电脑:15/天

  • 注册满6个月&&近期经常登录过该电脑:20/天

  • 注册满6个月&&近期频繁登陆过该电脑:30/天

  • 注册1年以上&&一直登录:50/天

  • 上一次通过好友到下一次通过间隔20-40s

  • 收到加人申请,到通过好友申请(每天最多通过300个好友申请),间隔30s+(随机时间)

参数:
  • scene -- 来源 在消息的xml获取

  • v1 -- v1key

  • v2 -- v2key

返回:

操作是否成功

返回类型:

bool

async get_contact(wxid: str | list[str]) dict | list[dict][源代码]

获取联系人信息

参数:

wxid -- 联系人wxid, 可以是多个wxid在list里,也可查询chatroom

返回:

单个联系人返回dict,多个联系人返回list[dict]

返回类型:

Union[dict, list[dict]]

async get_contract_detail(wxid: str | list[str], chatroom: str = '') list[源代码]

获取联系人详情

参数:
  • wxid -- 联系人wxid

  • chatroom -- 群聊wxid

返回:

联系人详情列表

返回类型:

list

async get_contract_list(wx_seq: int = 0, chatroom_seq: int = 0) dict[源代码]

获取联系人列表

参数:
  • wx_seq -- 联系人序列

  • chatroom_seq -- 群聊序列

返回:

联系人列表数据

返回类型:

dict

async get_nickname(wxid: str | list[str]) str | list[str][源代码]

获取用户昵称

参数:

wxid -- 用户wxid,可以是单个wxid或最多20个wxid的列表

返回:

如果输入单个wxid返回str,如果输入wxid列表则返回对应的昵称列表

返回类型:

Union[str, list[str]]

红包

class WechatAPI.Client.hongbao.HongBaoMixin(ip: str, port: int)[源代码]

基类:WechatAPIClientBase

async get_hongbao_detail(xml: str, encrypt_key: str, encrypt_userinfo: str) dict[源代码]

获取红包详情

参数:
  • xml -- 红包 XML 数据

  • encrypt_key -- 加密密钥

  • encrypt_userinfo -- 加密的用户信息

返回:

红包详情数据

返回类型:

dict

保护

class WechatAPI.Client.protect.Protect(*args, **kwargs)[源代码]

基类:object

保护类,风控保护机制。

使用单例模式确保全局只有一个实例。

变量:
  • login_stat_path (str) -- 登录状态文件的路径

  • login_stat (dict) -- 登录状态信息

  • login_time (int) -- 最后登录时间戳

  • login_device_id (str) -- 最后登录的设备ID

__init__()[源代码]

初始化保护类实例。

创建或加载登录状态文件,初始化登录时间和设备ID。

check(second: int) bool[源代码]

检查是否在指定时间内,风控保护。

参数:

second (int) -- 指定的秒数

返回:

如果当前时间与上次登录时间的差小于指定秒数,返回True;否则返回False

返回类型:

bool

update_login_status(device_id: str = '')[源代码]

更新登录状态。

如果设备ID发生变化,更新登录时间和设备ID,并保存到文件。

参数:

device_id (str, optional) -- 设备ID. Defaults to "".

class WechatAPI.Client.protect.Singleton[源代码]

基类:type

单例模式的元类。

用于确保一个类只有一个实例。

变量:

_instances (dict) -- 存储类的实例的字典

__call__(*args, **kwargs)[源代码]

创建或返回类的单例实例。

参数:
  • *args -- 位置参数

  • **kwargs -- 关键字参数

返回:

类的单例实例

返回类型:

object

工具

class WechatAPI.Client.tool.ToolMixin(ip: str, port: int)[源代码]

基类:WechatAPIClientBase

static base64_to_byte(base64_str: str) bytes[源代码]

将base64字符串转换为bytes。

参数:

base64_str (str) -- base64编码的字符串

返回:

解码后的字节数据

返回类型:

bytes

static base64_to_file(base64_str: str, file_name: str, file_path: str) bool[源代码]

将base64字符串转换为文件并保存。

参数:
  • base64_str (str) -- base64编码的字符串

  • file_name (str) -- 要保存的文件名

  • file_path (str) -- 文件保存路径

返回:

转换成功返回True,失败返回False

返回类型:

bool

static byte_to_base64(byte: bytes) str[源代码]

将bytes转换为base64字符串。

参数:

byte (bytes) -- 字节数据

返回:

base64编码的字符串

返回类型:

str

async check_database() bool[源代码]

检查数据库状态。

返回:

数据库正常返回True,否则返回False

返回类型:

bool

async download_attach(attach_id: str) dict[源代码]

下载附件。

参数:

attach_id (str) -- 附件ID

返回:

附件数据

返回类型:

dict

抛出:
  • UserLoggedOut -- 未登录时调用

  • 根据error_handler处理错误 --

async download_image(aeskey: str, cdnmidimgurl: str) str[源代码]

CDN下载高清图片。

参数:
  • aeskey (str) -- 图片的AES密钥

  • cdnmidimgurl (str) -- 图片的CDN URL

返回:

图片的base64编码字符串

返回类型:

str

抛出:
  • UserLoggedOut -- 未登录时调用

  • 根据error_handler处理错误 --

async download_video(msg_id) str[源代码]

下载视频。

参数:

msg_id (str) -- 消息的msg_id

返回:

视频的base64编码字符串

返回类型:

str

抛出:
  • UserLoggedOut -- 未登录时调用

  • 根据error_handler处理错误 --

async download_voice(msg_id: str, voiceurl: str, length: int) str[源代码]

下载语音文件。

参数:
  • msg_id (str) -- 消息的msgid

  • voiceurl (str) -- 语音的url,从xml获取

  • length (int) -- 语音长度,从xml获取

返回:

语音的base64编码字符串

返回类型:

str

抛出:
  • UserLoggedOut -- 未登录时调用

  • 根据error_handler处理错误 --

static file_to_base64(file_path: str) str[源代码]

将文件转换为base64字符串。

参数:

file_path (str) -- 文件路径

返回:

base64编码的字符串

返回类型:

str

async set_proxy(proxy: Proxy) bool[源代码]

设置代理。

参数:

proxy (Proxy) -- 代理配置对象

返回:

成功返回True,失败返回False

返回类型:

bool

抛出:
  • UserLoggedOut -- 未登录时调用

  • 根据error_handler处理错误 --

async set_step(count: int) bool[源代码]

设置步数。

参数:

count (int) -- 要设置的步数

返回:

成功返回True,失败返回False

返回类型:

bool

抛出:
  • UserLoggedOut -- 未登录时调用

  • BanProtection -- 风控保护: 新设备登录后4小时内请挂机

  • 根据error_handler处理错误 --

async static silk_base64_to_wav_byte(silk_base64: str) bytes[源代码]

将silk格式的base64字符串转换为WAV字节数据。

参数:

silk_base64 (str) -- silk格式的base64编码字符串

返回:

WAV格式的字节数据

返回类型:

bytes

async static silk_byte_to_byte_wav_byte(silk_byte: bytes) bytes[源代码]

将silk字节转换为wav字节。

参数:

silk_byte (bytes) -- silk格式的字节数据

返回:

wav格式的字节数据

返回类型:

bytes

static wav_byte_to_amr_base64(wav_byte: bytes) str[源代码]

将WAV字节数据转换为AMR格式的base64字符串。

参数:

wav_byte (bytes) -- WAV格式的字节数据

返回:

AMR格式的base64编码字符串

返回类型:

str

static wav_byte_to_amr_byte(wav_byte: bytes) bytes[源代码]

将WAV字节数据转换为AMR格式。

参数:

wav_byte (bytes) -- WAV格式的字节数据

返回:

AMR格式的字节数据

返回类型:

bytes

抛出:

Exception -- 转换失败时抛出异常

async static wav_byte_to_silk_base64(wav_byte: bytes) str[源代码]

将WAV字节数据转换为silk格式的base64字符串。

参数:

wav_byte (bytes) -- WAV格式的字节数据

返回:

silk格式的base64编码字符串

返回类型:

str

async static wav_byte_to_silk_byte(wav_byte: bytes) bytes[源代码]

将WAV字节数据转换为silk格式。

参数:

wav_byte (bytes) -- WAV格式的字节数据

返回:

silk格式的字节数据

返回类型:

bytes

索引