# -*- 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 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): return self.robot.CSendArticle(self.chatwith,title,abstract,url) def SendCard(self,sharedwxid,nickname): return self.robot.CSendCard(self.chatwith,sharedwxid,nickname) def SendAtText(self,wxid,msg): if '@chatroom' not in self.chatwith: return 1 return self.robot.CSendAtText(self.chatwith,wxid,msg) 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 GetSelfInfo(self): myinfo = self.robot.CGetSelfInfo().replace('\n','\\n') try: myinfo = ast.literal_eval(myinfo) except SyntaxError: return {} myinfo['wxBigAvatar'] = myinfo['wxBigAvatar'].replace("/132","/0") self.myinfo = myinfo return self.myinfo def StopService(self): self.StopReceiveMessage() return self.robot.CStopRobotService() 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 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 StopReceiveMessage(self): self.ReceiveMessageStarted = False try: self.ReceiveMessageThread.join() except: pass status = self.robot.CStopReceiveMessage() return status # 一个示例回调，将收到的文本消息转发给filehelper def ReceiveMessageCallBack(robot,message): if message['type'] == 1 and message['sender'] != 'filehelper': robot.robot.CSendText('filehelper',message['message']) if message['sender'] != 'filehelper': wxSender = robot.GetWxUserInfo(message['sender']) sender = wxSender['wxNickName'] if wxSender['wxNickName'] != 'null' else message['sender'] print("来自 {}".format(sender)) print(message) def test_SendText(): import os path = os.path.split(os.path.realpath(__file__))[0] # image full path imgpath = os.path.join(path,'test\\测试图片.png') # file full path filepath = os.path.join(path,'test\\测试文件') wx = WeChatRobot() wx.StartService() myinfo = wx.GetSelfInfo() chatwith = wx.GetFriendByWxNickName("文件传输助手") session = wx.GetChatSession(chatwith.get('wxid')) filehelper = wx.GetWxUserInfo(chatwith.get('wxid')) session.SendText('个人信息：{}'.format(str(myinfo.get('wxNickName')))) session.SendText('好友信息：{}'.format(str(filehelper.get('wxNickName')))) if os.path.exists(imgpath): session.SendImage(imgpath) if os.path.exists(filepath): session.SendFile(filepath) session.SendArticle("天气预报","点击查看","http://www.baidu.com") shared = wx.GetFriendByWxNickName("码农翻身") if shared: session.SendCard(shared.get('wxid'),shared.get('wxNickName')) wx.StopService() def test_FriendStatus(): f = open('Friendstatus.txt','wt',encoding = 'utf-8') wx = WeChatRobot() wx.StartService() FriendList = wx.GetFriendList() wx.CheckFriendStatusInit() index = "\t".join(['微信号','昵称','备注','状态','\n']) f.writelines(index) for Friend in FriendList: result = '\t'.join( [Friend.get('wxNumber'),Friend.get('wxNickName'),Friend.get('wxRemark'), wx.CheckFriendStatus(Friend.get('wxid'))]) print(result) result += '\n' f.writelines(result) time.sleep(1) break f.close() wx.StopService() def test_ReceiveMessage(): wx = WeChatRobot() wx.StartService() wx.StartReceiveMessage(CallBackFunc = ReceiveMessageCallBack) try: while True: pass except KeyboardInterrupt: pass wx.StopService() if __name__ == '__main__': test_ReceiveMessage()