wxRobot.py

# -*- coding: utf-8 -*-
"""
Created on Thu Feb 24 16:19:48 2022

@author: ljc545w
"""

# Before use,execute `CWeChatRebot.exe /regserver` in cmd by admin user
# need `pip install comtypes`
import comtypes.client
import ast
import os
import threading
import time

class ChatSession():
    def __init__(self,robot,wxid):
        self.robot = robot
        self.chatwith = wxid
        
    def SendText(self,msg):
        return self.robot.CSendText(self.chatwith,msg)
        
    def SendImage(self,imgpath):
        return self.robot.CSendImage(self.chatwith,imgpath)
    
    def SendFile(self,filepath):
        return self.robot.CSendFile(self.chatwith,filepath)
    
    def SendMp4(self,mp4path):
        return self.robot.CSendImage(self.chatwith,mp4path)
        
    def SendArticle(self,title,abstract,url,imgpath = None):
        return self.robot.CSendArticle(self.chatwith,title,abstract,url,imgpath)
    
    def SendCard(self,sharedwxid,nickname):
        return self.robot.CSendCard(self.chatwith,sharedwxid,nickname)
    
    def SendAtText(self,wxid:list or str or tuple,msg,AutoNickName = True):
        """
        Parameters
        ----------
        wxid : list or str or tuple
            被艾特人的wxid，列表或字符串.
        msg : str
            消息内容.
        AutoNickName : BOOL, optional
            是否自动填充被艾特人昵称. 

        Returns None
        -------
        SendAtText
            发送群艾特消息.
        """
        if '@chatroom' not in self.chatwith:
            return 1
        return self.robot.CSendAtText(self.chatwith,wxid,msg,AutoNickName)

    def SendAppMsg(self,appid):
        return self.robot.CSendAppMsg(self.chatwith,appid)
        

class WeChatRobot():
    
    def __init__(self):
        self.robot = comtypes.client.CreateObject("WeChatRobot.CWeChatRobot")
        self.AddressBook = []
        self.myinfo = {}
        self.ReceiveMessageStarted = False
        
    def StartService(self):
        status = self.robot.CStartRobotService()
        return status

    def SendText(self,receiver,msg):
        return self.robot.CSendText(receiver,msg)
        
    def SendImage(self,receiver,imgpath):
        return self.robot.CSendImage(receiver,imgpath)
    
    def SendFile(self,receiver,filepath):
        return self.robot.CSendFile(receiver,filepath)
        
    def SendArticle(self,receiver,title,abstract,url,imgpath = None):
        return self.robot.CSendArticle(receiver,title,abstract,url,imgpath)
    
    def SendCard(self,receiver,sharedwxid,nickname):
        return self.robot.CSendCard(receiver,sharedwxid,nickname)
    
    def SendAtText(self,chatroomid,AtUsers:list or str or tuple,msg,AutoNickName = True):
        if '@chatroom' not in chatroomid:
            return 1
        return self.robot.CSendAtText(chatroomid,AtUsers,msg,AutoNickName)

    def GetSelfInfo(self):
        myinfo = self.robot.CGetSelfInfo().replace('\n','\\n')
        try:
            myinfo = ast.literal_eval(myinfo)
        except SyntaxError:
            return {}
        self.myinfo = myinfo
        return self.myinfo
    
    def StopService(self):
        import psutil
        self.StopReceiveMessage()
        cpid = self.robot.CStopRobotService()
        try:
            cprocess = psutil.Process(cpid)
            cprocess.kill()
        except psutil.NoSuchProcess:
            pass
        return cpid
    
    def GetAddressBook(self):
        try:
            FriendTuple = self.robot.CGetFriendList()
            self.AddressBook = [dict(i) for i in list(FriendTuple)]
        except IndexError:
            self.AddressBook = []
        return self.AddressBook
    
    def GetFriendList(self):
        if not self.AddressBook:
            self.GetAddressBook()
        FriendList = []
        for item in self.AddressBook:
            if 'wxid_' == item['wxid'][0:5]:
                FriendList.append(item)
        return FriendList
    
    def GetChatRoomList(self):
        if not self.AddressBook:
            self.GetAddressBook()
        ChatRoomList = []
        for item in self.AddressBook:
            if '@chatroom' in item['wxid']:
                ChatRoomList.append(item)
        return ChatRoomList
    
    def GetOfficialAccountList(self):
        if not self.AddressBook:
            self.GetAddressBook()
        OfficialAccountList = []
        for item in self.AddressBook:
            if 'wxid_' != item['wxid'][0:5] and '@chatroom' not in item['wxid']:
                OfficialAccountList.append(item)
        return OfficialAccountList
    
    def GetFriendByWxRemark(self,remark:str):
        if not self.AddressBook:
            self.GetAddressBook()
        for item in self.AddressBook:
            if item['wxRemark'] == remark:
                return item
        return None
    
    def GetFriendByWxNumber(self,wxnumber:str):
        if not self.AddressBook:
            self.GetAddressBook()
        for item in self.AddressBook:
            if item['wxNumber'] == wxnumber:
                return item
        return None
    
    def GetFriendByWxNickName(self,wxnickname:str):
        if not self.AddressBook:
            self.GetAddressBook()
        for item in self.AddressBook:
            if item['wxNickName'] == wxnickname:
                return item
        return None
    
    def GetChatSession(self,wxid):
        return ChatSession(self.robot, wxid)
        
    def GetWxUserInfo(self,wxid):
        userinfo = self.robot.CGetWxUserInfo(wxid).replace('\n','\\n')
        return ast.literal_eval(userinfo)
    
    def GetChatRoomMembers(self,chatroomid):
        info = dict(self.robot.CGetChatRoomMembers(chatroomid))
        if not info:
            return None
        members = info['members'].split('^G')
        data = self.GetWxUserInfo(chatroomid)
        data['members'] = []
        for member in members:
            memberinfo = self.GetWxUserInfo(member)
            data['members'].append(memberinfo)
        return data
        
    def CheckFriendStatusInit(self):
        return self.robot.CCheckFriendStatusInit()
    
    def CheckFriendStatusFinish(self):
        return self.robot.CCheckFriendStatusFinish()
    
    def CheckFriendStatus(self,wxid):
        _EnumFriendStatus = {
            0xB0:'被删除',
            0xB1:'是好友',
            0xB2:'已拉黑',
            0xB5:'被拉黑',
            }
        status = self.robot.CCheckFriendStatus(wxid)
        if status == 0x0:
            print('请先初始化再进行检测！')
            assert False
        try:
            return _EnumFriendStatus[status]
        except KeyError:
            return "未知状态：{}".format(hex(status).upper().replace('0X','0x'))
        
    def ReceiveMessage(self,CallBackFunc = None):
        comtypes.CoInitialize()
        # 线程中必须新建一个对象，但无需重复注入
        ThreadRobot = WeChatRobot()
        while self.ReceiveMessageStarted:
            try:
                message = dict(ThreadRobot.robot.CReceiveMessage())
                if CallBackFunc and message:
                    CallBackFunc(ThreadRobot,message)
            except IndexError:
                message = None
            time.sleep(0.5)
        comtypes.CoUninitialize()
    
    # 接收消息的函数，可以添加一个回调
    def StartReceiveMessage(self,CallBackFunc = None):
        self.ReceiveMessageStarted = True
        status = self.robot.CStartReceiveMessage()
        self.ReceiveMessageThread = threading.Thread(target = self.ReceiveMessage,args= (CallBackFunc,))
        self.ReceiveMessageThread.daemon = True
        self.ReceiveMessageThread.start()
        return status
    
    def GetDbHandles(self):
        tablesTuple = self.robot.CGetDbHandles()
        tables = [dict(i) for i in tablesTuple]
        dbs = {}
        for table in tables:
            dbname = table['dbname']
            if dbname not in dbs.keys():
                dbs[dbname] = {'Handle':table['Handle'],'tables':[]}
            dbs[dbname]['tables'].append(
                {'name': table['name'],'tbl_name': table['tbl_name'],
                 'rootpage': table['rootpage'],'sql': table['sql']}
                )
        return dbs
    
    def ExecuteSQL(self,handle,sql):
        result = self.robot.CExecuteSQL(handle,sql)
        query_list = []
        keys = list(result[0])
        for item in result[1:]:
            query_dict = {}
            for key,value in zip(keys,item):
                query_dict[key] = value if not isinstance(value, tuple) else bytes(value)
            query_list.append(query_dict)
        return query_list
    
    def BackupSQLiteDB(self,handle,BackupFile):
        BackupFile = BackupFile.replace('/','\\')
        savepath = BackupFile.replace(BackupFile.split('\\')[-1],'')
        if not os.path.exists(savepath):
            os.makedirs(savepath)
        return self.robot.CBackupSQLiteDB(handle,BackupFile)

    def StopReceiveMessage(self):
        self.ReceiveMessageStarted = False
        try:
            self.ReceiveMessageThread.join()
        except:
            pass
        status = self.robot.CStopReceiveMessage()
        return status
        
    def VerifyFriendApply(self,v3,v4):
        return self.robot.CVerifyFriendApply(v3,v4)
    
    def AddFriendByWxid(self,wxid,message):
        return self.robot.CAddFriendByWxid(wxid,message)
    
    def AddFriendByV3(self,v3,message,AddType):
        """
        v3 : str
            要添加的人的V3数据.
        message : str
            附加信息.
        AddType : int
            手机号: 0xF;微信号: 0x3;QQ号: 0x1;朋友验证消息: 0x6.
        """
        return self.robot.CAddFriendByV3(v3,message,AddType)
    
    def GetWeChatVer(self):
        return self.robot.CGetWeChatVer()
    
    def StartWeChat(self):
        return self.robot.CStartWeChat()
    
    def GetUserInfoByNet(self,keyword):
        userinfo = self.robot.CSearchContactByNet(keyword)
        if userinfo:
            return dict(userinfo)
        return userinfo
    
    def AddBrandContact(self,PublicId):
        return self.robot.CAddBrandContact(PublicId)
    
    def ChangeWeChatVer(self,version:str):
        """
        version: like `3.7.0.26`
        """
        return self.robot.CChangeWeChatVer(version)
    
    def HookImageMsg(self,savepath):
        return self.robot.CHookImageMsg(savepath)
    
    def UnHookImageMsg(self):
        return self.robot.CUnHookImageMsg()
    
    def HookVoiceMsg(self,savepath):
        return self.robot.CHookVoiceMsg(savepath)
    
    def UnHookVoiceMsg(self):
        return self.robot.CUnHookVoiceMsg()

    def DeleteUser(self,wxid):
        return self.robot.CDeleteUser(wxid)

    def SendAppMsg(self,wxid,appid):
        return self.robot.CSendAppMsg(wxid,appid)