abot/wechat_ipad/client/user.py

import json

import aiohttp

from wechat_ipad import UserLoggedOut
from wechat_ipad.client.base import WechatAPIClientBase

from loguru import logger


class UserMixin(WechatAPIClientBase):
    async def get_profile(self, wxid: str = None) -> dict:
        """获取用户信息。

        Args:
            wxid (str, optional): 用户wxid. Defaults to None.

        Returns:
            dict: 用户信息字典

        Raises:
            UserLoggedOut: 未登录时调用
            根据error_handler处理错误
        """
        if not self.wxid and not wxid:
            raise UserLoggedOut("请先登录")

        if not wxid:
            wxid = self.wxid

        async with aiohttp.ClientSession() as session:
            response = await session.post(f'http://{self.ip}:{self.port}/api/User/GetContractProfile?wxid={wxid}')
            json_resp = await response.json()

            if json_resp.get("Success"):
                return json_resp.get("Data").get("userInfo")
            else:
                self.error_handler(json_resp)

    async def get_profile_info_ext(self, wxid: str = None) -> dict:
        """获取用户扩展信息。

        Args:
            wxid (str, optional): 用户wxid. Defaults to None.

        Returns:
            dict: 用户信息字典

        Raises:
            UserLoggedOut: 未登录时调用
            根据error_handler处理错误
        """
        if not self.wxid and not wxid:
            raise UserLoggedOut("请先登录")

        if not wxid:
            wxid = self.wxid

        async with aiohttp.ClientSession() as session:
            response = await session.post(f'http://{self.ip}:{self.port}/api/User/GetContractProfile?wxid={wxid}')
            json_resp = await response.json()

            if json_resp.get("Success"):
                return json_resp.get("Data").get("userInfoExt")
            else:
                self.error_handler(json_resp)

    async def get_my_qrcode(self, style: int = 0) -> str:
        """获取个人二维码。

        Args:
            style (int, optional): 二维码样式. Defaults to 0.

        Returns:
            str: 图片的base64编码字符串

        Raises:
            UserLoggedOut: 未登录时调用
            BanProtection: 风控保护: 新设备登录后4小时内请挂机
            根据error_handler处理错误
        """
        if not self.wxid:
            raise UserLoggedOut("请先登录")

        async with aiohttp.ClientSession() as session:
            json_param = {"Wxid": self.wxid, "Style": style}
            response = await session.post(f'http://{self.ip}:{self.port}/GetMyQRCode', json=json_param)
            json_resp = await response.json()

            if json_resp.get("Success"):
                return json_resp.get("Data").get("qrcode").get("buffer")
            else:
                self.error_handler(json_resp)

    async def is_logged_in(self, wxid: str = None) -> bool:
        """检查是否登录。

        Args:
            wxid (str, optional): 用户wxid. Defaults to None.

        Returns:
            bool: 已登录返回True，未登录返回False
        """
        if not wxid:
            wxid = self.wxid
        try:
            await self.get_profile(wxid)
            return True
        except Exception as e:
            logger.error("is_logged_in:{}", e)
            return False
if __name__ == '__main__':
    json_resp = """{
  "Code": 0,
  "Success": true,
  "Message": "成功",
  "Data": {
    "baseResponse": {
      "ret": 0,
      "errMsg": {}
    },
    "userInfo": {
      "BitFlag": 190,
      "UserName": {
        "string": "wxid_72ow1edm3kea22"
      },
      "NickName": {
        "string": "小牛"
      },
      "BindUin": 0,
      "BindEmail": {},
      "BindMobile": {
        "string": "17279625053"
      },
      "Status": 37,
      "ImgLen": 0,
      "Sex": 0,
      "Province": "Hunan",
      "City": "Changsha",
      "Signature": "随便聊聊",
      "PersonalCard": 1,
      "DisturbSetting": {
        "NightSetting": 0,
        "NightTime": {
          "BeginTime": 0,
          "EndTime": 0
        },
        "AllDaySetting": 0,
        "AllDayTim": {
          "BeginTime": 0,
          "EndTime": 0
        }
      },
      "PluginFlag": 30945,
      "VerifyFlag": 0,
      "Point": 855,
      "Experience": 2763,
      "Level": 7,
      "LevelLowExp": 2701,
      "LevelHighExp": 3500,
      "PluginSwitch": 40960,
      "GmailList": {
        "Count": 0
      },
      "WeiboFlag": 0,
      "FaceBookFlag": 0,
      "FbuserId": 0,
      "AlbumStyle": 0,
      "AlbumFlag": 0,
      "TxnewsCategory": 0,
      "Country": "CN"
    },
    "userInfoExt": {
      "SnsUserInfo": {
        "SnsFlag": 1,
        "SnsBgimgId": "http://szmmsns.qpic.cn/mmsns/AcIhsXSWkDeNF6K5icia87Oia8uDIHUf81hsyKibVCFxJgBdgGYP6E0V6gzqvhWrBODtg9lVwiaRS2U0/0",
        "SnsBgobjectId": 14653643334646501000,
        "SnsFlagEx": 128
      },
      "MyBrandList": "<brandlist><brand><username>gh_6e99ff560306</username><nickname><![CDATA[]]></nickname><alias></alias><iconurl></iconurl><Hidden>1</Hidden></brand></brandlist>",
      "BigChatRoomSize": 0,
      "BigChatRoomQuota": 0,
      "BigChatRoomInvite": 0,
      "BigHeadImgUrl": "http://wx.qlogo.cn/mmhead/ver_1/F8B2PqJujqlmP78Jben9VYYbjfU7SjPTOBkogPHm2TLOibAS1HiaARDUZlUrSibJ6AUJTgzxJn3NicwzpAoVTcr8znfTfkqX4D9zQsbjajoudicbnK2Th5jsAbXgNWuKnA7KC/0",
      "SmallHeadImgUrl": "http://wx.qlogo.cn/mmhead/ver_1/F8B2PqJujqlmP78Jben9VYYbjfU7SjPTOBkogPHm2TLOibAS1HiaARDUZlUrSibJ6AUJTgzxJn3NicwzpAoVTcr8znfTfkqX4D9zQsbjajoudicbnK2Th5jsAbXgNWuKnA7KC/132",
      "MainAcctType": 0,
      "ExtXml": {},
      "SafeDeviceList": {
        "Count": 0
      },
      "SafeDevice": 0,
      "GrayscaleFlag": 359,
      "RegCountry": "CN",
      "LinkedinContactItem": {},
      "PatternLockInfo": {
        "PatternVersion": 1,
        "Sign": {
          "iLen": 156,
          "buffer": "MzA0YzAyMjQwMzg4ZDJlMDRjYzJlYzU2MmQxNmUzMDRkNTY5OTUyZmNiOTBkZTNlNDcyYzc0ZjZlYTU3Mzg2MmI1MzIyOTYxOTdhYmRiMTYwMjI0MDEwNTU1OTg4ZDYxODFmNjQ4NjQ2NmRjZjUyNWMyYzNkMzk4N2Y2MTEyNWEzZGMxOTMzMDYwMGQxZGJlYzlkYTNhNzI5YzJi"
        },
        "LockStatus": 0
      },
      "PayWalletType": 0,
      "WalletRegion": 1,
      "ExtStatus": 65536,
      "UserStatus": 1
    }
  },
  "Data62": "",
  "Debug": ""
}"""
    json_resp =json.loads(json_resp)
    if json_resp.get("Success"):
         print(json_resp.get("Data").get("userInfoExt").get("SmallHeadImgUrl"))