本文实例为大家分享了python实现微信消息防撤回的具体代码，供大家参考，具体内容如下
使用了sqlite3保存数据，当有人撤回消息时取出数据发送到文件传输助手。
文件的话会先保存到本地，语音会以文件的方式发送。
wxpy 和 itchat很久没更新了，有些功能没法用了，web微信也不知道什么时候会凉。
帮助信息在注释里。
|
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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
# -*- coding: utf-8 -*- # 使用sqlite3保存message，当有人撤回消息时在数据库中通过id检索该消息是否存在，如果存在则将撤回的消息发送到文件助手里。# 目前只支持 text picture map sharing recording video attachment 类型的消息。 import wxpyimport sqlite3import osimport re # 准备工作# 创建attachment目录用于存储 图像、地图/位置、分享、语音、视频、文件if not os.path.isdir('attachment'): os.mkdir('attachment')attachment_path = os.path.join(os.getcwd(), 'attachment')bot = wxpy.bot()# 用于获取msg idpattern = re.compile(r'\d{19}')# 测试wxpy能否正常工作myself = bot.friends()[0]myself.send('hello?') # 创建数据库和message表try: conn = sqlite3.connect('wxpy.db') cursor = conn.cursor() # cursor.execute('drop table messages') cursor.execute("""create table if not exists messages (id integer primary key autoincrement, msg_id integer not null, msg_text text, create_time date not null, revoke_time date, attachment_path text, msg_sender text not null, msg_type text not null, msg_url text, msg_raw_data text not null)""") # print('establish successfully')finally: conn.commit() cursor.close() conn.close() # 注册所有消息，在程序运行期间将插入所有支持的信息@bot.register()def store_data(msg): # print(msg.raw) # 如果消息是支持的类型就将数据插入数据库 if msg.type in [wxpy.text, wxpy.recording, wxpy.picture, wxpy.attachment, wxpy.video, wxpy.sharing, wxpy.map]: insert_data(msg) # 撤回的消息类型是note elif msg.type == wxpy.note: send_revoke(msg) # 插入数据def insert_data(msg): try: conn = sqlite3.connect('wxpy.db') cursor = conn.cursor() if msg.type == wxpy.text: cursor.execute("insert into messages (msg_id, msg_text, create_time, msg_sender, msg_type, msg_raw_data)\ values (?, ?, ?, ?, ?, ?)", (msg.id, msg.text, msg.create_time, str(msg.sender)[9:-1], msg.type, str(msg.raw))) # 将录音/图像/文件/视频下载到本地，插入保存路径。 elif msg.type in [wxpy.recording, wxpy.picture, wxpy.attachment, wxpy.video]: save_path = os.path.join(attachment_path, msg.file_name) msg.get_file(save_path) cursor.execute('insert into messages (msg_id, create_time, attachment_path, msg_sender, msg_type,\ msg_raw_data) values (?, ?, ?, ?, ?, ?)', (msg.id, msg.create_time, save_path, str(msg.sender)[9:-1], msg.type, str(msg.raw))) # 插入分享/位置链接 elif msg.type in [wxpy.sharing, wxpy.map]: cursor.execute('insert into messages (msg_id, msg_text, create_time, msg_sender, msg_type, msg_url,\ msg_raw_data) values (?, ?, ?, ?, ?, ?, ?)', (msg.id, msg.text, msg.create_time, str(msg.sender)[9:-1], msg.type, str(msg.url), str(msg.raw))) # print('insert data successfully') finally: conn.commit() cursor.close() conn.close() # 在数据库中检索消息是否存在，如果存在则将被撤回的消息发送到文件传输助手。def send_revoke(message): msg_id = pattern.search(message.raw['content']).group() try: conn = sqlite3.connect('wxpy.db') cursor = conn.cursor() cursor.execute('insert into messages (msg_id, create_time, msg_sender, msg_type, msg_raw_data)\ values (?, ?, ?, ?, ?)', (message.id, message.create_time, str(message.sender)[9:-1], message.type, str(message.raw))) msg_data = cursor.execute('select * from messages where msg_id=?', (msg_id, )).fetchall() # print('take out data successfully') finally: conn.commit() cursor.close() conn.close() if msg_data[0][7] == 'text': msg_info = '告诉你一个秘密 {} 在 {} 撤回了文本\n{}'.format(msg_data[0][6], msg_data[0][3], msg_data[0][2]) bot.file_helper.send(msg_info) else: send_revoke_nontext(msg_data) # 非文本信息发送def send_revoke_nontext(msg_data): if msg_data[0][7] == 'picture': if msg_data[0][5][-4:] == '.gif': # 现在wxpy & itchat发不了gif了 bot.file_helper('很抱歉，暂时不支持表情（gif）的撤回重发。') else: msg_info = '告诉你一个秘密 {} 在 {} 撤回了图像'.format(msg_data[0][6], msg_data[0][3]) bot.file_helper.send(msg_info) bot.file_helper.send_image(msg_data[0][5]) elif msg_data[0][7] == 'recording': msg_info = '告诉你一个秘密 {} 在 {} 撤回了语音'.format(msg_data[0][6], msg_data[0][3]) bot.file_helper.send(msg_info) bot.file_helper.send_file(msg_data[0][5]) elif msg_data[0][7] == 'attachment': msg_info = '告诉你一个秘密 {} 在 {} 撤回了文件'.format(msg_data[0][6], msg_data[0][3]) bot.file_helper.send(msg_info) bot.file_helper.send_file(msg_data[0][5]) elif msg_data[0][7] == 'video': msg_info = '告诉你一个秘密 {} 在 {} 撤回了视频'.format(msg_data[0][6], msg_data[0][3]) bot.file_helper.send(msg_info) bot.file_helper.send_video(msg_data[0][5]) elif msg_data[0][7] == 'sharing': msg_info = '告诉你一个秘密 {} 在 {} 撤回了分享\n{}\n{}'.format(msg_data[0][6], msg_data[0][3], msg_data[0][2],\ msg_data[0][8]) bot.file_helper.send(msg_info) elif msg_data[0][7] == 'map': msg_info = '告诉你一个秘密 {} 在 {} 撤回了位置\n{}\n{}'.format(msg_data[0][6], msg_data[0][3], msg_data[0][2],\ msg_data[0][8]) bot.file_helper.send(msg_info) wxpy.embed() |
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持服务器之家。
原文链接：https://blog.csdn.net/babaili_/article/details/85947411
