cai.client package

Application Client.

This module is main entry point for the application.

Copyright

Copyright (C) 2021-2021 cscs181

License

AGPL-3.0 or later. See LICENSE for detail.

Submodules

cai.client.client module

Application Client Class.

This module is used to control client actions (low-level api).

Copyright

Copyright (C) 2021-2021 cscs181

License

AGPL-3.0 or later. See LICENSE for detail.

class cai.client.client.Client(uin: int, password_md5: bytes)

基类：object

LISTENERS: Set[Callable[[cai.client.client.Client, cai.client.event.Event], Awaitable[None]]] = {}
property uin: int
返回

qq number of the client account.

返回类型

int

property nick: Optional[str]

Only available after login.

返回

nick name of the client account.

返回类型

Optional[str]

property age: Optional[int]

Only available after login.

返回

age of the client account.

返回类型

Optional[int]

property gender: Optional[int]

Only available after login.

返回

gender of the client account.

返回类型

Optional[int]

property status: Optional[cai.client.status_service.OnlineStatus]

See detail statuses in OnlineStatus Enum class.

返回

Online status of the client account.

返回类型

Optional[OnlineStatus]

property connection: cai.connection.Connection
返回

connection object for the client.

返回类型

Connection

引发

ConnectionError – no connection available.

property connected: bool
返回

True if the client has connected to the server.

返回类型

bool

async connect(server: Optional[cai.client.sso_server.jce.SsoServer] = None) → None

Connect to the server.

This should be called before sending any packets.

参数

server (Optional[SsoServer], optional) – The server you want to connect. Defaults to None.

引发

  • RuntimeError – Already connected to the server.

  • ConnectionError – Error when connecting the server.

async disconnect() → None

Disconnect if already connected to the server.

async reconnect(change_server: bool = False, server: Optional[cai.client.sso_server.jce.SsoServer] = None) → None

Reconnect to the server.

The server arg only take effect if change_server is True.

参数

  • change_server (bool, optional) – True if you want to change the server. Defaults to False.

  • server (Optional[SsoServer], optional) – Which server you want to connect to. Defaults to None.

async close() → None

Close the client and logout.

property seq: int
返回

current packet sequence number.

返回类型

int

next_seq() → int

Get next packet sequence number.

返回

next sequence number.

返回类型

int

async send(seq: int, command_name: str, packet: Union[bytes, cai.utils.binary.Packet]) → None

Send a packet with the given sequence but not wait for the response.

参数

  • seq (int) – Sequence number.

  • command_name (str) – Command name of the packet.

  • packet (Union[bytes, Packet]) – Packet to send.

返回

None.

async send_and_wait(seq: int, command_name: str, packet: Union[bytes, cai.utils.binary.Packet], timeout: Optional[float] = 10.0) → cai.client.command.Command

Send a packet with the given sequence and wait for the response.

参数

  • seq (int) – Sequence number.

  • command_name (str) – Command name of the packet.

  • packet (Union[bytes, Packet]) – Packet to send.

  • timeout (Optional[float], optional) – Timeout. Defaults to 10.

返回

Response.

返回类型

Command

async receive()

Receive data from connection reader and store it in sequence future.

注解

Source: com.tencent.mobileqq.msf.core.auth.n.a

property listeners: Set[Callable[[cai.client.client.Client, cai.client.event.Event], Awaitable[None]]]
dispatch_event(event: cai.client.event.Event) → None
add_event_listener(listener: Callable[[cai.client.client.Client, cai.client.event.Event], Awaitable[None]]) → None

Add event listener for this client.

参数

listener (Callable[[Client, Event], Awaitable[None]]) – Event listener.

async login() → cai.client.wtlogin.oicq.LoginSuccess

Login the account of the client.

This should be called before using any other apis.

返回

Success login command.

返回类型

LoginSuccess

引发
async submit_captcha(captcha: str, captcha_sign: bytes) → cai.client.wtlogin.oicq.LoginSuccess

Submit captcha when login captcha needed.

This should be called after LoginCaptchaNeeded occurred.

返回

Success login command.

返回类型

LoginSuccess

引发
async submit_slider_ticket(ticket: str) → cai.client.wtlogin.oicq.LoginSuccess

Submit slider ticket when login slider captcha needed.

This should be called after LoginSliderNeeded occurred.

返回

Success login command.

返回类型

LoginSuccess

引发
async request_sms() → bool

Request new sms message when login sms code needed.

This should be called after LoginSMSRequestError occurred.

返回

Success login command.

返回类型

LoginSuccess

引发
async submit_sms(sms_code: str) → cai.client.wtlogin.oicq.LoginSuccess

Submit sms code when login sms code needed.

This should be called after LoginSMSRequestError occurred.

返回

Success login command.

返回类型

LoginSuccess

引发
async refresh_siginfo() → cai.client.wtlogin.oicq.LoginSuccess

Submit sms code when login sms code needed.

This should be called after LoginSMSRequestError occurred.

返回

Success login command.

返回类型

LoginSuccess

引发
async register(status: cai.client.status_service.OnlineStatus = OnlineStatus.Online, register_reason: cai.client.status_service.RegPushReason = RegPushReason.AppRegister) → cai.client.status_service.command.RegisterSuccess

Register app client and get login status.

This should be called after Client.login() successed.

参数
返回

Register success response.

返回类型

RegisterSuccess

引发

  • RuntimeError – Error response type got. This should not happen.

  • ApiResponseError – Register failed.

async set_status(status: Union[int, cai.client.status_service.OnlineStatus], battery_status: Optional[int] = None, is_power_connected: bool = False) → cai.client.status_service.command.RegisterSuccess

Register app client and get login status.

This should be called after Client.login() successed.

参数

  • status (OnlineStatus, optional) – Client status. Defaults to Online.

  • register_reason (RegPushReason, optional) – Register reason. Defaults to AppRegister.

  • battery_status (Optional[int], optional) – Battery capacity. Only works when status is OnlineStatus.Battery. Defaults to None.

  • is_power_connected (bool, optional) – Is power connected to phone. Only works when status is OnlineStatus.Battery. Defaults to False.

返回

Register success response.

返回类型

RegisterSuccess

引发

  • RuntimeError – Error response type got. This should not happen.

  • ApiResponseError – Register failed.

async heartbeat() → None

Do heartbeat.

Calling this method more than once takes no effect.

Heartbeat will be down when an error occurred.

Example

Create a heartbeat task using asyncio.

>>> import asyncio
>>> asyncio.create_task(client.heartbeat())
async get_friend(uin: int, cache: bool = True) → Optional[cai.client.models.Friend]

Get Friend.

Return cached friend if cache is True.

参数

  • uin (int) – Friend uin.

  • cache (bool, optional) – Use cached friend list. Defaults to True.

返回

Friend object. None: Friend not exists.

返回类型

Friend

引发

  • RuntimeError – Error response type got. This should not happen.

  • ApiResponseError – Get friend list failed.

  • FriendListException – Get friend list returned non-zero ret code.

async get_friend_list(cache: bool = True) → List[cai.client.models.Friend]

Get Friend List.

Return cached friend list if cache is True.

参数

cache (bool, optional) – Use cached friend list. Defaults to True.

返回

Friend list.

返回类型

List[Friend]

引发

  • RuntimeError – Error response type got. This should not happen.

  • ApiResponseError – Get friend list failed.

  • FriendListException – Get friend list returned non-zero ret code.

async get_friend_group(group_id: int, cache: bool = True) → Optional[cai.client.models.FriendGroup]

Get Friend Group.

Return cached friend group if cache is True.

参数

  • group_id (int) – Friend group id.

  • cache (bool, optional) – Use cached friend group list. Defaults to True.

返回

Friend group object. None: Friend group not exists.

返回类型

FriendGroup

引发

  • RuntimeError – Error response type got. This should not happen.

  • ApiResponseError – Get friend group list failed.

  • FriendListException – Get friend group list returned non-zero ret code.

async get_friend_group_list(cache: bool = True) → List[cai.client.models.FriendGroup]

Get Friend Group List.

Return cached friend group list if cache is True.

参数

cache (bool, optional) – Use cached friend group list. Defaults to True.

返回

Friend group list.

返回类型

List[FriendGroup]

引发

  • RuntimeError – Error response type got. This should not happen.

  • ApiResponseError – Get friend group list failed.

  • FriendListException – Get friend group list returned non-zero ret code.

async get_group(group_id: int, cache: bool = True) → Optional[cai.client.models.Group]

Get Group.

Return cached group if cache is True.

参数

cache (bool, optional) – Use cached group list. Defaults to True.

返回

Group object. None: Group not exists.

返回类型

Group

引发

  • RuntimeError – Error response type got. This should not happen.

  • ApiResponseError – Get group list failed.

  • GroupListException – Get group list returned non-zero ret code.

async get_group_list(cache: bool = True) → List[cai.client.models.Group]

Get Group List.

Return cached group list if cache is True.

参数

cache (bool, optional) – Use cached group list. Defaults to True.

返回

Group list.

返回类型

List[Group]

引发

  • RuntimeError – Error response type got. This should not happen.

  • ApiResponseError – Get group list failed.

  • GroupListException – Get group list returned non-zero ret code.

async get_group_member_list(group: int, cache: bool = True) → Optional[List[cai.client.models.GroupMember]]
async get_group_member_list(group: cai.client.models.Group, cache: bool = True) → List[cai.client.models.GroupMember]

Get Group Member List.

Return cached group member list if cache is True.

参数

  • group (Union[int, Group]) – Group id or group object want to get members.

  • cache (bool, optional) – Use cached group list. Defaults to True.

返回

Group member list. None: Group not exists.

返回类型

List[GroupMember]

引发

cai.client.command module

Client Base Command.

This module is used to build command from packet.

Copyright

Copyright (C) 2021-2021 cscs181

License

AGPL-3.0 or later. See LICENSE for detail.

class cai.client.command.Command(uin: int, seq: int, ret_code: int, command_name: str)

基类：object

uin: int
seq: int
ret_code: int
command_name: str
class cai.client.command.UnhandledCommand(uin: int, seq: int, ret_code: int, command_name: str, data: bytes)

基类：cai.client.command.Command

data: bytes

cai.client.event module

Client Base Command.

This module is used to build command from packet.

Copyright

Copyright (C) 2021-2021 cscs181

License

AGPL-3.0 or later. See LICENSE for detail.

class cai.client.event.Event

基类：abc.ABC

abstract property type: str

cai.client.models module

Application Client Info Releated Models.

This module is used to define account info data models.

Copyright

Copyright (C) 2021-2021 cscs181

License

AGPL-3.0 or later. See LICENSE for detail.

class cai.client.models.SigInfo(d2: bytes = b'', d2key: bytes = b'', g: bytes = b'', tgt: bytes = b'', tgt_key: bytes = b'', device_token: bytes = b'', dpwd: bytes = b'', no_pic_sig: bytes = b'', encrypted_a1: bytes = b'', login_bitmap: int = 0, ps_key_map: Dict[str, bytes] = <factory>, pt4_token_map: Dict[str, bytes] = <factory>, rand_seed: bytes = b'', _s_key: bytes = b'', s_key_expire_time: int = 0, user_st_key: bytes = b'', user_st_web_sig: bytes = b'', wt_session_ticket: bytes = b'', wt_session_ticket_key: bytes = b'')

基类：object

Account Siginfo.

注解

Source: oicq.wlogin_sdk.sharemem.WloginSigInfo

d2: bytes = b''
d2key: bytes = b''
g: bytes = b''
tgt: bytes = b''
tgt_key: bytes = b''
device_token: bytes = b''
dpwd: bytes = b''
no_pic_sig: bytes = b''
encrypted_a1: bytes = b''
login_bitmap: int = 0
ps_key_map: Dict[str, bytes]
pt4_token_map: Dict[str, bytes]
rand_seed: bytes = b''
s_key_expire_time: int = 0
user_st_key: bytes = b''
user_st_web_sig: bytes = b''
wt_session_ticket: bytes = b''
wt_session_ticket_key: bytes = b''
property s_key: bytes
class cai.client.models.Friend(friend_uin: int, group_id: int, face_id: int, remark: str, is_mqq_online: bool, is_iphone_online: bool, show_name: str, is_remark: bool, nick: str, network_type: int, vip_font: int, term_description: str, sex: int, battery_status: int, _client: 'Client')

基类：cai.utils.dataclass.JsonableDataclass

friend_uin: int
group_id: int
face_id: int
remark: str
is_mqq_online: bool
is_iphone_online: bool
show_name: str
is_remark: bool
nick: str
network_type: int
vip_font: int
term_description: str
sex: int
battery_status: int
property uin: int
async get_group() → cai.client.models.FriendGroup
class cai.client.models.FriendGroup(group_id: int, group_name: str, friend_count: int, online_friend_count: int, _client: 'Client')

基类：cai.utils.dataclass.JsonableDataclass

group_id: int
group_name: str
friend_count: int
online_friend_count: int
class cai.client.models.Group(group_uin: int, group_code: int, group_name: str, group_memo: str, shutup_timestamp: int, my_shutup_timestamp: int, member_num: int, group_owner_uin: int, cmd_uin_join_time: int, max_group_member_num: int, _client: 'Client', _cached_member_list: List[ForwardRef('GroupMember')] = <factory>)

基类：cai.utils.dataclass.JsonableDataclass

group_uin: int
group_code: int
group_name: str
group_memo: str
shutup_timestamp: int
my_shutup_timestamp: int
member_num: int
group_owner_uin: int
cmd_uin_join_time: int
max_group_member_num: int
property group_id: int

int: Group ID.

Caculated by group_uin and group_code.

property join_time: int

int: Group join time. Same as cmd_uin_join_time.

async get_members(cache: bool = True) → List[cai.client.models.GroupMember]
class cai.client.models.GroupMemberRole(value)

基类：str, enum.Enum

An enumeration.

owner = 'owner'
admin = 'admin'
member = 'member'
class cai.client.models.GroupMember(member_uin: int, age: int, gender: int, nick: str, show_name: str, name: str, phone: str, email: str, memo: str, member_level: int, join_time: int, last_speak_time: int, flag: int, concerned: bool, shielded: bool, special_title: str, special_title_expire_time: int, shutup_timestamp: int, _client: 'Client', _group: cai.client.models.Group)

基类：cai.utils.dataclass.JsonableDataclass

member_uin: int
age: int
gender: int
nick: str
show_name: str
name: str
phone: str
email: str
memo: str
member_level: int
join_time: int
last_speak_time: int
flag: int
concerned: bool
shielded: bool
special_title: str
special_title_expire_time: int
shutup_timestamp: int
property uin: int
property role: cai.client.models.GroupMemberRole
property group: cai.client.models.Group

cai.client.packet module

Client packets.

This module is used to packet data into outgoing format.

Copyright

Copyright (C) 2021-2021 cscs181

License

AGPL-3.0 or later. See LICENSE for detail.

class cai.client.packet.CSsoBodyPacket(*args, **kwargs)

基类：cai.utils.binary.Packet

CSSOBody Packet.

注解

Source:

com.tencent.qphone.base.util.CodecWarpper

/data/data/com.tencent.mobileqq/lib/libcodecwrapperV2.so

CSSOReqHead::serialize_verFull

classmethod build(seq: int, sub_app_id: int, command_name: str, imei: str, session_id: bytes, ksid: bytes, body: Union[bytes, cai.utils.binary.Packet], extra_data: Union[bytes, cai.utils.binary.Packet] = b'', unknown_bytes: bytes = b'\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00') → cai.client.packet.CSsoBodyPacket

Build CSSOBody head and append body.

注解

Source: CSSOReqHead::serialize_verFull

class cai.client.packet.CSsoDataPacket(*args, **kwargs)

基类：cai.utils.binary.Packet

CSSOData Packet.

KSSOVersion: Full: 0xA Simple: 0xB.

注解

Source:

com.tencent.qphone.base.util.CodecWarpper

/data/data/com.tencent.mobileqq/lib/libcodecwrapperV2.so

CSSOData::serialize

classmethod build(uin: int, body_type: int, body: Union[bytes, cai.utils.binary.Packet], ksso_version: int = 10, key: Optional[bytes] = None, extra_data: bytes = b'') → cai.client.packet.CSsoDataPacket

Build CSSOPacket head and append body.

Packet body was encrypted in CSSOData::serialize.

注解

Source: CSSOHead::serialize_verFull

class cai.client.packet.UniPacket(*args, **kwargs)

基类：cai.utils.binary.Packet

classmethod build(uin: int, seq: int, command_name: str, session_id: bytes, body_type: int, body: Union[bytes, cai.utils.binary.Packet], key: bytes, extra_data: bytes = b'') → cai.client.packet.UniPacket

Build new packet and write data into it.

参数

*data (Union[bytes, Packet]) – Data to write

返回

Current Packet

返回类型

Packet

class cai.client.packet.IncomingPacket(uin: int, seq: int, ret_code: int, extra: bytes, command_name: str, session_id: bytes, data: bytes)

基类：object

uin: int
seq: int
ret_code: int
extra: bytes
command_name: str
session_id: bytes
data: bytes
classmethod parse(data: Union[bytes, bytearray], key: bytes, d2key: bytes, session_key: bytes) → cai.client.packet.IncomingPacket
classmethod parse_sso_frame(sso_frame: Union[bytes, bytearray], encrypt_type: int, key: bytes, session_key: bytes, **kwargs) → cai.client.packet.IncomingPacket
classmethod parse_oicq_body(data: Union[bytes, bytearray], key: bytes, session_key: bytes) → bytes

Parse incoming OICQ packet.

注解

Source: oicq.wlogin_sdk.request.oicq_request.b

参数

  • data (Union[bytes, Packet]) – OICQ data.

  • key (bytes) – Random key.

  • session_key (bytes) – Siginfo wt_session_ticket_key.

引发

ValueError – Flag error or unknown encrypt type.

返回

Decode data.

返回类型

bytes