Python 打造微信群聊天机器人（带操作界面）

客官°小女子只卖身不卖艺 2022-02-22 09:53 1107阅读 0赞

## 前言 ##

这几天我的一个小伙伴问我能不能给 Ta 做一个配置灵活的微信群聊天机器人，之前了解过 `itchat` 库的使用，我就爽快的答应了，花了一个晚上，终于做出了雏形。

电脑上运行程序如下:

![在这里插入图片描述][20190409092137595.gif]

手机上的信息如下:

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lnZHh0_size_16_color_FFFFFF_t_70]

其实基于 `itchat` 的微信机器人早已经烂大街了，但大多数过于简单，相比较而言，我的这个程序有下面几大鲜明的特色:

1、支持打开/关闭指定群聊的自动回复，只需用记事本打开 `group.csv` 文件，填写想要打开自动回复群聊名称即可，每次输完一个群聊名称，必须换行。

2、支持自定义设置关键词回复，只需用记事本打开 `keyword.csv` 文件，按照\*\*\*\{关键词,回复\}\*\*\*的格式添加即可，而不需要在代码中做任何修改。输完一个键值对，同样也需要换行，注意输入的是英文逗号。

3、支持定时群发消息，而且时间、消息可以在程序运行中动态修改。

4、有较为良好的 `GUI` 界面，其中色彩搭配参考了微信的简约设计。

## DIY 玩法 ##

1、面向商户

> 作为商户，维护群的时候可能有这样的需求，定时发送消息。比如`回复 xxx 可获得 yyy`,同时在`keyword.csv` 文件中事先写好。这样，可以让群成员各取所需，你又不需要打字、复制粘贴，还可以同时处理多个群，省心还高效。

2、面向普通个人

> 定时向男女朋友，父母亲人发送晚安祝福消息等。

3、部署至阿里云服务器

> 有个缺陷就是如果想一直自动群发消息的话，你的电脑就必须一直开着，但是部署至云服务器可以解决这个问题。部署流程可以参考我之前的文章,同时给出阿里云服务器优惠购买传送门。

## 代码的详细设计 ##

1、代码的架构

由于引入了 `GUI`，`GUI`代码块和负责群发消息的代码块一样，都是阻塞型的，为此，程序就必须引入多线程机制，其中 `GUI` 界面是主线程，负责群发消息的代码块运行在子线程，线程间的通信我用的是 `wxPython` 内置的 `wx.lib.pubsub` 模块，一旦子线程执行了相应的动作，就通过 `wx.CallAfter(pub.sendMessage)` 接口发送消息给通知 `GUI` 线程，从而保证 `GUI` 能够及时刷新并不至于卡顿。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lnZHh0_size_16_color_FFFFFF_t_70 1]

2、代码的流程

首先是加载相应的配置文件，确定要开启哪些群聊的自动回复，以及关键词回复信息。也正因如此，在程序执行过程中，这些信息是不能被动态改变的。

其中加载 `keyword`的代码如下:

def load_keyword(self):
    	global keywords
    	with open('keyword.csv', 'r', encoding='utf-8', newline='') as f:
    		reader = csv.reader(f)
    		for i, line in enumerate(reader):
    			if i == 0:
    				continue
    			keywords[line[0]] = line[1]

把 `keywords` 设置为全局变量方便后面使用，避免传参调用，判断 `i == 0` 是为了去掉 `csv` 文件的第一行头部信息。

负责群发的主要代码块如下，代码注释较为清晰，不再赘述

@itchat.msg_register(TEXT, isGroupChat=True)
    def group_text(msg):
    	global keywords
    	groups = itchat.get_chatrooms(update=True)
    	for group in groups:
    		# 群的 NickName 是群名称，UserName 是群id(以两个@开始)
    		# Python/Java 学习交流群
    		if group['NickName'] in group_names:  # 从群中找到指定的群聊
    			group_id = msg['FromUserName']
    			# 防止其他群消息的的干扰
    			if not group_id == group['UserName']:
    				break
    			# 准备回复的消息
    			keys = keywords.keys()
    			key = ''
    			for i in keys:
    				if i in msg['Text']:
    					key = i
    					break
    			if key == '':
    				return
    			message = keywords.get(key)
    			# 在消息中找到 发送人的id
    			sender_id = msg['ActualUserName']
    			# 有时 group['MemberList'] 为空，改变思路由群 id 获取群聊成员
    			# group_info = itchat.update_chatroom(msg['ToUserName'], detailedMember=True)
    			# if len(group_info) == 0:
    			# toUserName 是自己在群聊发消息时，群 id 在消息里的 key
    			# FromUserName 是别人在群里发时，群 id 在消息里的 key
    			group_info = itchat.update_chatroom(group_id, detailedMember=True)
    			memberlist = group_info['MemberList']
    			for member in memberlist:
    				# 找到消息的发送者
    				if member['UserName'] == sender_id:
    					# 如果有备注名，群聊显示的是备注名
    					to_user = member['RemarkName']
    					if len(to_user) == 0:
    						# 否则显示成员自己修改的在群里的昵称
    						to_user = member['DisplayName']
    					if len(to_user) == 0:
    						# 否则显示他微信号的昵称
    						to_user = member['NickName']
    					itchat.send_msg('@{}\n{}'.format(to_user, message), group['UserName'])
    					wx.CallAfter(pub.sendMessage, "update", msg="回复群聊[{}]成员[{}]成功:[{}]".format(group['NickName'],to_user,message))

负责定时群发的代码和上面的代码比较独立，在子线程开始的同时，开始执行定时群发的逻辑

def run(self):
        global t
        t = threading.Timer(minutes * 60, self.auto_timer)
        t.start()
        self.load_keyword()
        self.load_group()
        itchat.auto_login(hotReload=True)
        itchat.run()

其中主要的函数是 `threading.Timer(minutes * 60, self.auto_timer)`,它的意思是在负责执行群发的线程里，再开一个线程，这个线程间隔`minutes * 60` 秒后去执行回调函数 `self.auto_timer`,但是这样只能触发一次，没办法一直轮询，解决办法是在回调函数里面再去执行 `threading.Timer(minutes * 60, self.auto_timer)`,有点儿类似于递归调用，和递归不同的是，调用是没有终止条件的，但并不会产生内存溢出，因为定时器的存在，时间一到一触发回调函数，这个线程的生命就到此为止了，因此在整个程序运行期间，活跃线程的数目仅仅只是个位数：

def auto_timer(self):
        global auto_message
        groups = itchat.get_chatrooms(update=True)
        for group in groups:
            if group['NickName'] in group_names:
                itchat.send_msg('{}'.format(auto_message), group['UserName'])
                wx.CallAfter(pub.sendMessage, "update",
                             msg="群聊[{}]定时消息:[{}]发送成功".format(group['NickName'], auto_message))
    
        global t  # 把 t 设置为全局变量
        t = threading.Timer(minutes * 60, self.auto_timer)
        t.start()

`GUI` 部分的代码由于篇幅限制,就不贴出来了。

# 如何体验 #

关注公众号月小水长,后台回复微信群机器人即可获得

[20190409092137595.gif]: /images/20220222/89427693e48c4f6f8ba20dc896f8a6ff.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lnZHh0_size_16_color_FFFFFF_t_70]: /images/20220222/f5050075add44416bb9328dead0c5f68.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lnZHh0_size_16_color_FFFFFF_t_70 1]: /images/20220222/7bb7574cad74451c8ea77bb27858c729.png