8987b52361
- 按 864 router.go 修正联系人详情与群公告的真实接口路径 - 结合 CheckLoginStatus 返回结构补充 uuid 与 effective_time 同步，支持 Dashboard 倒计时与二维码自动刷新 - 更新多版本适配路线图，记录首轮 864 实服联调结论
95 lines
3.7 KiB
Python
95 lines
3.7 KiB
Python
from typing import Any, Union
|
||
|
||
from wechat_ipad.providers.server_864.base import Server864APIClientBase
|
||
|
||
|
||
class ChatroomMixin(Server864APIClientBase):
|
||
"""864 群聊相关接口。"""
|
||
|
||
async def get_chatroom_info(self, chatroom: str) -> dict:
|
||
"""获取单个群详情。"""
|
||
data = await self._request_data(
|
||
"post",
|
||
"/group/GetChatRoomInfo",
|
||
json_body={"ChatRoomWxIdList": [chatroom]},
|
||
timeout=30,
|
||
)
|
||
if isinstance(data, list):
|
||
return dict(data[0] or {}) if data else {}
|
||
if not isinstance(data, dict):
|
||
return {}
|
||
|
||
contact_list = data.get("ContactList") or data.get("contactList") or []
|
||
if contact_list:
|
||
return dict(contact_list[0] or {})
|
||
return dict(data)
|
||
|
||
async def get_chatroom_member_list(self, chatroom: str) -> list[dict]:
|
||
"""获取群成员列表。"""
|
||
data = await self._request_data(
|
||
"post",
|
||
"/group/GetChatroomMemberDetail",
|
||
json_body={"ChatRoomName": chatroom},
|
||
timeout=30,
|
||
)
|
||
if isinstance(data, list):
|
||
return list(data)
|
||
if not isinstance(data, dict):
|
||
return []
|
||
return list(
|
||
data.get("ChatRoomMember")
|
||
or data.get("chatRoomMember")
|
||
or (data.get("NewChatroomData") or {}).get("ChatRoomMember", [])
|
||
or (data.get("newChatroomData") or {}).get("chatRoomMember", [])
|
||
or data.get("MemberList")
|
||
or []
|
||
)
|
||
|
||
async def get_chatroom_announce(self, chatroom: str) -> dict:
|
||
"""获取群公告详情。"""
|
||
data = await self._request_data(
|
||
"post",
|
||
# 864 的群公告接口路由名保留了历史命名 `SetGetChatRoomInfoDetail`：
|
||
# 1. controller 中实际导出的就是这个路径；
|
||
# 2. 直接用语义更自然的 `GetChatRoomInfoDetail` 会 404；
|
||
# 3. 对外方法名继续保持 `get_chatroom_announce`，内部路径单独适配即可。
|
||
"/group/SetGetChatRoomInfoDetail",
|
||
json_body={"ChatRoomName": chatroom},
|
||
timeout=30,
|
||
)
|
||
return dict(data or {})
|
||
|
||
async def get_chatroom_qrcode(self, chatroom: str) -> dict[str, Any]:
|
||
"""获取群二维码。"""
|
||
data = await self._request_data(
|
||
"post",
|
||
"/group/GetChatroomQrCode",
|
||
json_body={"ChatRoomName": chatroom},
|
||
timeout=30,
|
||
)
|
||
return {
|
||
"base64": str(self._pick_first(data, "QrCode", "qrcode", "QrCodeBase64", "base64") or ""),
|
||
"description": str(self._pick_first(data, "Desc", "desc", "Description", "description") or ""),
|
||
}
|
||
|
||
async def get_chatroom_nickname(self, wxid: Union[str, list[str]], chatroom: str) -> Union[str, list[str]]:
|
||
"""获取成员在群内的展示昵称。"""
|
||
members = await self.get_chatroom_member_list(chatroom)
|
||
member_map = {}
|
||
for member in members:
|
||
member_wxid = str(member.get("UserName", "") or "")
|
||
nickname = member.get("DisplayName") or member.get("NickName") or member_wxid
|
||
member_map[member_wxid] = nickname
|
||
|
||
if isinstance(wxid, str):
|
||
return str(member_map.get(wxid, ""))
|
||
return [str(member_map.get(single_wxid, "")) for single_wxid in wxid]
|
||
|
||
async def get_chatroom_member_detail(self, wxid: str, chatroom: str) -> dict:
|
||
"""获取单个群成员详情。"""
|
||
members = await self.get_chatroom_member_list(chatroom)
|
||
for member in members:
|
||
if str(member.get("UserName", "") or "") == wxid:
|
||
return dict(member)
|
||
return {}
|