Hide navigation sidebar
Hide table of contents sidebar
Skip to content
Back to top
Toggle table of contents sidebar

# WechatAPI.Client.friend 源代码

    from typing import Union

    import aiohttp

    from .base import *
    from .protect import protector
    from ..errors import *

    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("请先登录")
            elif not self.ignore_protect and protector.check(14400):
                raise BanProtection("风控保护: 新设备登录后4小时内请挂机")

            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}/AcceptFriend', 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}/GetContact', 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, "RequestWxids": wxid, "Chatroom": chatroom}
                response = await session.post(f'http://{self.ip}:{self.port}/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) -> dict:
            """获取联系人列表

            Args:
                wx_seq: 联系人序列
                chatroom_seq: 群聊序列

            Returns:
                dict: 联系人列表数据
            """
            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}/GetContractList', json=json_param)
                json_resp = await response.json()

                if json_resp.get("Success"):
                    return json_resp.get("Data")
                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