150 lines
5.4 KiB
Python
150 lines
5.4 KiB
Python
from typing import Union
|
|
|
|
import aiohttp
|
|
|
|
from wechat_ipad import UserLoggedOut
|
|
from wechat_ipad.client.base import WechatAPIClientBase
|
|
|
|
|
|
class FriendMixin(WechatAPIClientBase):
|
|
async def accept_friend(self, scene: int, v1: str, v2: str) -> bool:
|
|
"""接受好友请求
|
|
|
|
主动添加好友单天上限如下所示：1小时内上限为 5个，超过上限时，无法发出好友请求，也收不到好友请求。
|
|
|
|
- 新账号：5/天
|
|
- 注册超过7天：10个/天
|
|
- 注册满3个月&&近期登录过该电脑：15/天
|
|
- 注册满6个月&&近期经常登录过该电脑：20/天
|
|
- 注册满6个月&&近期频繁登陆过该电脑：30/天
|
|
- 注册1年以上&&一直登录：50/天
|
|
- 上一次通过好友到下一次通过间隔20-40s
|
|
- 收到加人申请，到通过好友申请（每天最多通过300个好友申请），间隔30s+（随机时间）
|
|
|
|
Args:
|
|
scene: 来源 在消息的xml获取
|
|
v1: v1key
|
|
v2: v2key
|
|
|
|
Returns:
|
|
bool: 操作是否成功
|
|
"""
|
|
if not self.wxid:
|
|
raise UserLoggedOut("请先登录")
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
json_param = {"Wxid": self.wxid, "Scene": scene, "V1": v1, "V2": v2}
|
|
response = await session.post(f'http://{self.ip}:{self.port}/api/Friend/PassVerify', json=json_param)
|
|
json_resp = await response.json()
|
|
|
|
if json_resp.get("Success"):
|
|
return True
|
|
else:
|
|
self.error_handler(json_resp)
|
|
|
|
async def get_contact(self, wxid: Union[str, list[str]]) -> Union[dict, list[dict]]:
|
|
"""获取联系人信息
|
|
|
|
Args:
|
|
wxid: 联系人wxid, 可以是多个wxid在list里，也可查询chatroom
|
|
|
|
Returns:
|
|
Union[dict, list[dict]]: 单个联系人返回dict，多个联系人返回list[dict]
|
|
"""
|
|
if not self.wxid:
|
|
raise UserLoggedOut("请先登录")
|
|
|
|
if isinstance(wxid, list):
|
|
wxid = ",".join(wxid)
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
json_param = {"Wxid": self.wxid, "RequestWxids": wxid}
|
|
response = await session.post(f'http://{self.ip}:{self.port}/api/Friend/GetContractDetail', json=json_param)
|
|
json_resp = await response.json()
|
|
|
|
if json_resp.get("Success"):
|
|
contact_list = json_resp.get("Data").get("ContactList")
|
|
if len(contact_list) == 1:
|
|
return contact_list[0]
|
|
else:
|
|
return contact_list
|
|
else:
|
|
self.error_handler(json_resp)
|
|
|
|
async def get_contract_detail(self, wxid: Union[str, list[str]], chatroom: str = "") -> list:
|
|
"""获取联系人详情
|
|
|
|
Args:
|
|
wxid: 联系人wxid
|
|
chatroom: 群聊wxid
|
|
|
|
Returns:
|
|
list: 联系人详情列表
|
|
"""
|
|
if not self.wxid:
|
|
raise UserLoggedOut("请先登录")
|
|
|
|
if isinstance(wxid, list):
|
|
if len(wxid) > 20:
|
|
raise ValueError("一次最多查询20个联系人")
|
|
wxid = ",".join(wxid)
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
json_param = {"Wxid": self.wxid, "Towxids": wxid, "Chatroom": chatroom}
|
|
response = await session.post(f'http://{self.ip}:{self.port}/api/Friend/GetContractDetail', json=json_param)
|
|
json_resp = await response.json()
|
|
|
|
if json_resp.get("Success"):
|
|
return json_resp.get("Data").get("ContactList")
|
|
else:
|
|
self.error_handler(json_resp)
|
|
|
|
async def get_contract_list(self, wx_seq: int = 0, chatroom_seq: int = 0) -> list:
|
|
"""获取联系人用户名列表
|
|
|
|
Args:
|
|
wx_seq: 联系人序列
|
|
chatroom_seq: 群聊序列
|
|
|
|
Returns:
|
|
list: 联系人用户名列表
|
|
"""
|
|
if not self.wxid:
|
|
raise UserLoggedOut("请先登录")
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
json_param = {"Wxid": self.wxid, "CurrentWxcontactSeq": wx_seq, "CurrentChatroomContactSeq": chatroom_seq}
|
|
response = await session.post(f'http://{self.ip}:{self.port}/api/Friend/GetContractList', json=json_param)
|
|
json_resp = await response.json()
|
|
|
|
if json_resp.get("Success"):
|
|
# 直接返回联系人用户名列表
|
|
return json_resp.get("Data").get("ContactUsernameList", [])
|
|
else:
|
|
self.error_handler(json_resp)
|
|
|
|
async def get_nickname(self, wxid: Union[str, list[str]]) -> Union[str, list[str]]:
|
|
"""获取用户昵称
|
|
|
|
Args:
|
|
wxid: 用户wxid，可以是单个wxid或最多20个wxid的列表
|
|
|
|
Returns:
|
|
Union[str, list[str]]: 如果输入单个wxid返回str，如果输入wxid列表则返回对应的昵称列表
|
|
"""
|
|
data = await self.get_contract_detail(wxid)
|
|
|
|
if isinstance(wxid, str):
|
|
try:
|
|
return data[0].get("NickName").get("string")
|
|
except:
|
|
return ""
|
|
else:
|
|
result = []
|
|
for contact in data:
|
|
try:
|
|
result.append(contact.get("NickName").get("string"))
|
|
except:
|
|
result.append("")
|
|
return result
|