Python实现微信自动欢迎新人入群

本文最后更新于：几秒前

今天一小姑娘问我能不能实现wx自动欢迎新人入群。刚开始听到这个要求，本以为会很简单，毕竟Github上有很多开源的项目，完全直接可以克隆一个，根据自己的需求进行一下修改。当时我这么想，也这么做了，结果发Wechat网页版被禁止登录了！！Github上很多开源项目都是基于wechat网页版，通过HTTP进行的交互。然后经过寻找发现了一个WechatPCAPI,解决了我的问题。

下面进入正题：

首先将WechatPCAPI克隆到我们的目录，如下。
图1.jpg

下面是需要导入的包。

1
2
3
4
5
from WechatPCAPI import WechatPCAPI
import time
import logging
from queue import Queue
import threading

首先定义三个队列，用来处理收到的消息：

1
2
3
queue_recved_message = Queue()#用来处理所有消息
queue_groups=Queue() #用来处理群ID消息
queue_Members=Queue()#用来处理群成员信息

然后再定义群成员这个数据结构，群成员有群ID，WXID，昵称三个属性：

1
2
3
4
5
class Person:
    def __init__(self):
        self.chatroom_id = ""
        self.wx_id = ""
        self.nick_name = ""

接下来定义处理消息队列的函数，要检测新成员进群，首先要获取群ID，然后再更新指定群中成员的变化。如果发现本次更新的群成员信息与上次群成员信息不一致，则多出来的群成员即为新加入的。所以需要获得的消息类型有‘friend::chatroom’,‘member::chatroom’两种。获取这两种消息后，再次分别放到对应的队列中，进行下一步处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
def thead_handle_mess(wx_inst):
    while True:
        if not queue_recved_message.empty():
            message = queue_recved_message.get()
            if 'friend::chatroom' in message.get('type'):
                if chatroom in message.get('data',{}).get('chatroom_name',""):
                queue_groups.put(message.get('data',{}).get('chatroom_id',""))
            elif 'member::chatroom' in message.get('type'):
                Per=Person()
                Per.chatroom_id=message.get('data',{}).get('chatroom_id',"")
                Per.wx_id = message.get('data', {}).get('wx_id', "")
                Per.nick_name = message.get('data', {}).get('wx_nickname', "")
                queue_Members.put(Per)

接下来在定义更新群成员的函数：

1
2
3
4
5
def thead_handle_getmember(wx_inst,Group_list):
    while True:
        for group in Group_list:
            wx_inst.get_member_of_chatroom(group)
        time.sleep(20)#间隔多长时间更新一次群成员列表

然后定义获取群列表的函数：

1
2
3
4
5
6
7
def get_group_list():
    Group_list=[]
    while queue_groups.empty():
        time.sleep(1)
    while not queue_groups.empty():
        Group_list.append(queue_groups.get())
    return Group_list

这里有个问题，那就是一开始程序没有对比的群成员列表，这样它就会向所有的群成员发送欢迎语。为了解决这个问题，我们首先得获取已经存在的群成员列表。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def get_existed_member(wx_inst,Group_list):
    member_groups={}
    for group in Group_list:
        wx_inst.get_member_of_chatroom(group)
    while queue_Members.empty():
        time.sleep(0.5)
    while not queue_Members.empty():
        Person=queue_Members.get()
        if Person.chatroom_id not in  member_groups.keys():
            member_group={Person.chatroom_id:[Person]}
            member_groups.update(member_group)
        elif Person.wx_id not in get_all_id(member_group[Person.chatroom_id]):
            member_group[Person.chatroom_id].append(Person)
    return member_groups

接下来我们再写欢迎群成员的函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def thread_handle_member_welcome(wx_inst,member_groups):
    groups=member_groups
    with open('config.json', 'r')as f:
        j = json.loads(f.read())
    mess = j['mess']
    while True:
        if not queue_Members.empty():
            Person=queue_Members.get()
            if Person.wx_id not in get_all_id(groups[Person.chatroom_id]):
                add_member(Person,groups)
                try:
                    
                    wx_inst.send_text(to_user=Person.chatroom_id,msg=mess, at_someone=Person.wx_id)
                except Exception as  e:
                    print(e)
                print("Say welcome to {}".format(Person.nick_name))
            else:
                pass
        else:
            pass

这是两个杂项的功能，一个是获取所有的群成员wxid，另一个是用来向已有的列表中加入群成员。

1
2
3
4
5
6
7
def get_all_id(List):
    id_list=[]
    for i in List:
        id_list.append(i.wx_id)
    return id_list
def add_member(Person,member_groups):
    member_groups[Person.chatroom_id].append(Person)

这样这个小程序的所有组件就写完了，接下来的工作是将它们组合起来：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def main():
    print("初始化中...请稍候！")
    wx_inst=WechatPCAPI(on_message=onmessage)
    wx_inst.start_wechat(block=True)
    time.sleep(3)
    threading.Thread(target=thead_handle_mess,args=(wx_inst,)).start()
    wx_inst.update_frinds()
    Group_list=get_group_list()
    member_groups=get_existed_member(wx_inst,Group_list)
    print("运行中....")
    threading.Thread(target=thead_handle_getmember,args=(wx_inst,Group_list,)).start() 
    threading.Thread(target=thread_handle_member_welcome,args=(wx_inst,member_groups,)).start()

def onmessage(message):
    queue_recved_message.put(message)
 #这是回调函数

设置下日志记录功能，方便日后排错。

1
2
file_name = str(time.strftime("%Y-%m-%d", time.localtime()))+".log"
logging.basicConfig(filename=file_name,level=logging.INFO)

完整代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
# coding: utf-8
from WechatPCAPI import WechatPCAPI
import time
import logging
from queue import Queue
import threading
file_name = str(time.strftime("%Y-%m-%d", time.localtime()))+".log"
logging.basicConfig(filename=file_name,level=logging.INFO)
queue_recved_message = Queue()#用来处理所有消息
queue_groups=Queue() #用来处理群ID消息
queue_Members=Queue()#用来处理群成员信息
class Person:
    def __init__(self):
        self.chatroom_id = ""
        self.wx_id = ""
        self.nick_name = ""
def onmessage(message):
    queue_recved_message.put(message)
def thead_handle_mess(wx_inst):
    while True:
        if not queue_recved_message.empty():
            message = queue_recved_message.get()
            if 'friend::chatroom' in message.get('type'):
                if chatroom in message.get('data',{}).get('chatroom_name',""):
                queue_groups.put(message.get('data',{}).get('chatroom_id',""))
            elif 'member::chatroom' in message.get('type'):
                Per=Person()#生成新的对象
                Per.chatroom_id=message.get('data',{}).get('chatroom_id',"")
                Per.wx_id = message.get('data', {}).get('wx_id', "")
                Per.nick_name = message.get('data', {}).get('wx_nickname', "")
                queue_Members.put(Per)
def thead_handle_getmember(wx_inst,Group_list):
    while True:
        for group in Group_list:
            wx_inst.get_member_of_chatroom(group)
        time.sleep(60)
def get_group_list():
    Group_list=[]
    while queue_groups.empty():
        time.sleep(1)
    while not queue_groups.empty():
        Group_list.append(queue_groups.get())
    return Group_list
def get_existed_member(wx_inst,Group_list):
    member_groups={}
    for group in Group_list:
        wx_inst.get_member_of_chatroom(group)
    while queue_Members.empty():
        time.sleep(0.5)
    while not queue_Members.empty():
        Person=queue_Members.get()
        if Person.chatroom_id not in  member_groups.keys():
            member_group={Person.chatroom_id:[Person]}
            member_groups.update(member_group)
        elif Person.wx_id not in get_all_id(member_group[Person.chatroom_id]):
            member_group[Person.chatroom_id].append(Person)
    return member_groups
def thread_handle_member_welcome(wx_inst,member_groups):
    groups=member_groups
    with open('config.json', 'r')as f:
        j = json.loads(f.read())
    mess = j['mess']
    while True:
        if not queue_Members.empty():
            Person=queue_Members.get()
            if Person.wx_id not in get_all_id(groups[Person.chatroom_id]):
                add_member(Person,groups)
                try:
                    
                    wx_inst.send_text(to_user=Person.chatroom_id,msg=mess, at_someone=Person.wx_id)
                except Exception as  e:
                    print(e)
                print("Say welcome to {}".format(Person.nick_name))
            else:
                pass
        else:
            pass
def main():
    print("初始化中...请稍候！")
    wx_inst=WechatPCAPI(on_message=onmessage)
    wx_inst.start_wechat(block=True)
    time.sleep(3)
    threading.Thread(target=thead_handle_mess,args=(wx_inst,)).start()
    wx_inst.update_frinds()
    Group_list=get_group_list()
    member_groups=get_existed_member(wx_inst,Group_list)
    print("运行中....")
    threading.Thread(target=thead_handle_getmember,args=(wx_inst,Group_list,)).start() 
    threading.Thread(target=thread_handle_member_welcome,args=(wx_inst,member_groups,)).start()
def get_all_id(List):
    id_list=[]
    for i in List:
        id_list.append(i.wx_id)
    return id_list
def add_member(Person,member_groups):
    member_groups[Person.chatroom_id].append(Person)
if __name__ == "__main__":
    main()

下面是对脚本进行部分优化
首先，目录结构不用变动，我们只是修改Main.py的一些功能函数。

下面是需要导入的包：

1
2
3
4
5
from WechatPCAPI import WechatPCAPI
import time
import logging
from queue import Queue
import threading

导入的包较上次也是没有变动的。

同上篇一样，需要定义三个队列用来处理接收到的信息：

1
2
3
queue_recved_message = Queue()#用来处理所有消息
queue_groups=Queue() #用来处理群ID消息
queue_Members=Queue()#用来处理群成员信息

群成员这个数据结构有了些许变化，群成员现在只有群ID，昵称这两个属性了：

1
2
3
4
class Person:
    def __init__(self):
        self.chatroom_id = ""
        self.nick_name = ""

接下来定义处理消息队列的函数，首先还是要获取指定关键词的群列表。由于群成员新进群或者被移除群聊时，wx都会产生一条消息，这条消息的类型是‘msg::chatroom’ ，数据类型是‘10000’，具体的消息内容就是‘xxx通过xx方式加入了群聊’或者‘xxx被移除了群聊’，所以要响应新成员进群，我们只需要判断‘msg::chatroom’的chatroom_id，以及消息内容便可。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def thead_handle_mess(wx_inst):
    with open('config.json', 'r')as f:
        j = json.loads(f.read())
    chatroom=j['chatroom']
    while True:
        if not queue_recved_message.empty():
            message = queue_recved_message.get()
            if 'friend::chatroom' in message.get('type'):
                if chatroom in message.get('data', {}).get('chatroom_name', ""):
                    queue_groups.put(message.get(
                    'data', {}).get('chatroom_id', ""))
            elif 'msg::chatroom' in message.get('type'):
                if 'data_type' in message.get('data').keys():
                    if message.get('data').get('data_type') == '10000' and "加入" in message.get('data').get('msg'):
                        Per = Person()  # 生成新的对象
                        Per.chatroom_id =message.get('data').get('from_chatroom_wxid')
                        Per.nick_name = message.get('data').get('msg').split("\"")[1]
                        queue_Members.put(Per)

接下来定义获取群列表的函数：

1
2
3
4
5
6
7
def get_group_list():
    Group_list = []
    while queue_groups.empty():
        time.sleep(1)
    while not queue_groups.empty():
        Group_list.append(queue_groups.get())
    return Group_list

然后就是欢迎群成员的函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
def thread_handle_member_welcome(wx_inst, group_list):
    groups = group_list
    with open('config.json', 'r')as f:
        j = json.loads(f.read())
    mess=j['mess']
    while True:
        if not queue_Members.empty():
            Person = queue_Members.get()
            if Person.chatroom_id in groups:
                M=str(Person.nick_name)+","+mess
                wx_inst.send_text(to_user=Person.chatroom_id,msg=M)
                print("say welcome to {}".format(Person.nick_name))
            time.sleep(2)

这里面有个问题，群成员的nickname是自己这一端显示的成员nickname。例如：有个成员叫member，你给他的备注是猪，那么欢迎语中的nickname就是猪，这点使用时需要注意下。

这样这个小程序的所有组件就写完了，接下来的工作是将它们组合起来：

1
2
3
4
5
6
7
8
9
10
11
def main():
    print("初始化中...请稍候！")
    wx_inst = WechatPCAPI(on_message=onmessage)
    wx_inst.start_wechat(block=True)
    while not wx_inst.get_myself():
        time.sleep(3)
    threading.Thread(target=thead_handle_mess, args=(wx_inst,)).start()
    wx_inst.update_frinds()
    Group_list = get_group_list()
    print("运行中....")
    threading.Thread(target=thread_handle_member_welcome,args=(wx_inst, Group_list,)).start()

设置下日志记录功能，方便日后排错。

1
2
file_name = str(time.strftime("%Y-%m-%d", time.localtime()))+".log"
logging.basicConfig(filename=file_name,level=logging.INFO)

完整的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
# coding: utf-8
from WechatPCAPI import WechatPCAPI
import time
import logging
from queue import Queue
import threading
file_name = str(time.strftime("%Y-%m-%d", time.localtime()))+".log"
logging.basicConfig(filename=file_name, level=logging.INFO)
queue_recved_message = Queue()  # 用来处理所有消息
queue_groups = Queue()  # 用来处理群ID消息
queue_Members = Queue()  # 用来处理群成员信息
class Person:
    def __init__(self):
        self.chatroom_id = ""
        self.nick_name = ""
def onmessage(message):
    queue_recved_message.put(message)
def thead_handle_mess(wx_inst):
    with open('config.json', 'r')as f:
        j = json.loads(f.read())
    chatroom=j['chatroom']
    while True:
        if not queue_recved_message.empty():
            message = queue_recved_message.get()
            if 'friend::chatroom' in message.get('type'):
                if chatroom in message.get('data', {}).get('chatroom_name', ""):
                    queue_groups.put(message.get(
                    'data', {}).get('chatroom_id', ""))
            elif 'msg::chatroom' in message.get('type'):
                if 'data_type' in message.get('data').keys():
                    if message.get('data').get('data_type') == '10000' and "加入" in message.get('data').get('msg'):
                        Per = Person()  # 生成新的对象
                        Per.chatroom_id =message.get('data').get('from_chatroom_wxid')
                        Per.nick_name = message.get('data').get('msg').split("\"")[1]
                        queue_Members.put(Per)
def get_group_list():
    Group_list = []
    while queue_groups.empty():
        time.sleep(1)
    while not queue_groups.empty():
        Group_list.append(queue_groups.get())
    return Group_list
def thread_handle_member_welcome(wx_inst, group_list):
    groups = group_list
    with open('config.json', 'r')as f:
        j = json.loads(f.read())
    mess=j['mess']
    while True:
        if not queue_Members.empty():
            Person = queue_Members.get()
            if Person.chatroom_id in groups:
                M=str(Person.nick_name)+","+mess
                wx_inst.send_text(to_user=Person.chatroom_id,msg=M)
                print("say welcome to {}".format(Person.nick_name))
            time.sleep(2)
def main():
    print("初始化中...请稍候！")
    wx_inst = WechatPCAPI(on_message=onmessage)
    wx_inst.start_wechat(block=True)
    while not wx_inst.get_myself():
        time.sleep(3)
    threading.Thread(target=thead_handle_mess, args=(wx_inst,)).start()
    wx_inst.update_frinds()
    Group_list = get_group_list()
    print("运行中....")
    threading.Thread(target=thread_handle_member_welcome,args=(wx_inst, Group_list,)).start()
if __name__ == "__main__":
    main()
Python
#Python #Wechat
Python实现微信自动欢迎新人入群
https://blog.icansudo.top/2020/04/23/Python实现微信自动欢迎新人入群/
作者
odin
发布于
2020年4月23日
许可协议