ff33edb0d1
- 新增 server_864 独立 provider 目录，接入登录、消息轮询、联系人、群资料、用户资料与朋友圈基础能力 - 扩展 gateway、robot 与配置归一化逻辑，支持 server_864/864 别名和 WECHAT_SERVER_KEY - 更新配置示例与多版本适配路线图，明确 864 第一版接入范围和后续待补项
85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
from typing import Union
|
|
|
|
from wechat_ipad.providers.server_864.base import Server864APIClientBase
|
|
|
|
|
|
class FriendMixin(Server864APIClientBase):
|
|
"""864 联系人与通讯录相关接口。"""
|
|
|
|
async def get_contact(self, wxid: Union[str, list[str]]) -> Union[dict, list[dict]]:
|
|
"""获取联系人详情，兼容旧调用面。"""
|
|
detail_list = await self.get_contract_detail(wxid)
|
|
if isinstance(wxid, str):
|
|
return detail_list[0] if detail_list else {}
|
|
return detail_list
|
|
|
|
async def get_contract_detail(self, wxid: Union[str, list[str]], chatroom: str = "") -> list:
|
|
"""批量获取联系人详情。"""
|
|
user_names = [wxid] if isinstance(wxid, str) else list(wxid or [])
|
|
room_wxid_list = [chatroom] if chatroom else []
|
|
data = await self._request_data(
|
|
"post",
|
|
"/friend/GetContactDetail",
|
|
json_body={"UserNames": user_names, "RoomWxIDList": room_wxid_list},
|
|
timeout=30,
|
|
)
|
|
|
|
if isinstance(data, list):
|
|
return data
|
|
if not isinstance(data, dict):
|
|
return []
|
|
|
|
contact_list = (
|
|
data.get("ContactList")
|
|
or data.get("contactList")
|
|
or data.get("MemberList")
|
|
or data.get("memberList")
|
|
or data.get("UserList")
|
|
or data.get("userList")
|
|
or []
|
|
)
|
|
return list(contact_list or [])
|
|
|
|
async def get_contract_list(self, wx_seq: int = 0, chatroom_seq: int = 0) -> list:
|
|
"""获取通讯录用户名列表。"""
|
|
data = await self._request_data(
|
|
"post",
|
|
"/friend/GetContactList",
|
|
json_body={
|
|
"CurrentWxcontactSeq": int(wx_seq),
|
|
"CurrentChatRoomContactSeq": int(chatroom_seq),
|
|
},
|
|
timeout=30,
|
|
)
|
|
if isinstance(data, list):
|
|
return data
|
|
if not isinstance(data, dict):
|
|
return []
|
|
return list(
|
|
data.get("ContactUsernameList")
|
|
or data.get("contactUsernameList")
|
|
or data.get("UserNames")
|
|
or data.get("userNames")
|
|
or []
|
|
)
|
|
|
|
async def get_nickname(self, wxid: Union[str, list[str]]) -> Union[str, list[str]]:
|
|
"""根据联系人详情返回昵称。"""
|
|
data = await self.get_contract_detail(wxid)
|
|
if isinstance(wxid, str):
|
|
if not data:
|
|
return ""
|
|
nickname = data[0].get("NickName")
|
|
if isinstance(nickname, dict):
|
|
return nickname.get("string", "")
|
|
return nickname or ""
|
|
|
|
result = []
|
|
for contact in data:
|
|
nickname = contact.get("NickName")
|
|
if isinstance(nickname, dict):
|
|
result.append(nickname.get("string", ""))
|
|
else:
|
|
result.append(nickname or "")
|
|
return result
|