Shikong/WeChatMsg
1
0
Fork 0
mirror of https://github.com/LC044/WeChatMsg synced 2025-02-22 19:02:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #325

Browse Source 
完成解析转账和音视频通话类型信息的两个后端接口
This commit is contained in:
SiYuan 2024-01-19 19:12:04 +08:00 committed by GitHub
commit 548d9b18f5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 262 additions and 146 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
app/DataBase/exporter_csv.py
View File

@ -22,7 +22,7 @@ class CSVExporter(ExporterBase):
# 写入数据 # 写入数据
# writer.writerows(messages) # writer.writerows(messages)
for msg in messages: for msg in messages:
other_data = [msg[12].remark, msg[12].nickName, msg[12].wxid] if self.contact.is_chatroom else [] other_data = [msg[13].remark, msg[13].nickName, msg[13].wxid] if self.contact.is_chatroom else []
writer.writerow([*msg[:9], *other_data]) writer.writerow([*msg[:9], *other_data])
print(f"【完成导出 CSV {self.contact.remark}】") print(f"【完成导出 CSV {self.contact.remark}】")
self.okSignal.emit(1) self.okSignal.emit(1)

75
app/DataBase/exporter_html.py
View File

@ -11,7 +11,7 @@ from app.DataBase.output import ExporterBase, escape_js_and_html
from app.log import logger from app.log import logger
from app.person import Me from app.person import Me
from app.util import path from app.util import path
from app.util.compress_content import parser_reply, share_card, music_share, file from app.util.compress_content import parser_reply, share_card, music_share, file, transfer_decompress, call_decompress
from app.util.emoji import get_emoji_url from app.util.emoji import get_emoji_url
from app.util.image import get_image_path, get_image from app.util.image import get_image_path, get_image
from app.util.music import get_music_path from app.util.music import get_music_path
@ -264,6 +264,36 @@ class HtmlExporter(ExporterBase):
f'''{{ type:49,sub_type:5, text:'',is_send:{is_send},avatar_path:'{avatar}',url:'{card_data.get('url')}',timestamp:{timestamp},is_chatroom:{is_chatroom},displayname:'{display_name}',title:'{card_data.get('title')}',description:'{card_data.get('description')}',thumbnail:'{thumbnail}',app_logo:'{app_logo}',app_name:'{card_data.get('app_name')}'}},\n''' f'''{{ type:49,sub_type:5, text:'',is_send:{is_send},avatar_path:'{avatar}',url:'{card_data.get('url')}',timestamp:{timestamp},is_chatroom:{is_chatroom},displayname:'{display_name}',title:'{card_data.get('title')}',description:'{card_data.get('description')}',thumbnail:'{thumbnail}',app_logo:'{app_logo}',app_name:'{card_data.get('app_name')}'}},\n'''
) )
def transfer(self, doc, message):
is_send = message[4]
timestamp = message[5]
compress_content_ = message[11]
open("test.bin", "wb").write(compress_content_)
transfer_detail = transfer_decompress(compress_content_)
is_chatroom = 1 if self.contact.is_chatroom else 0
avatar = self.get_avatar_path(is_send, message)
display_name = self.get_display_name(is_send, message)
text_info_map = {
1: transfer_detail["pay_memo"] or "转账",
3: "已收款",
4: "已退还",
}
doc.write(f"""{{ type:49, sub_type:2000,text:'{text_info_map[transfer_detail["paysubtype"]]}',is_send:{is_send},avatar_path:'{avatar}',timestamp:{timestamp},is_chatroom:{is_chatroom},displayname:'{display_name}',paysubtype:{transfer_detail["paysubtype"]},pay_memo:'{transfer_detail["pay_memo"]}',feedesc:'{transfer_detail["feedesc"]}',}},\n""")
def call(self, doc, message):
is_send = message[4]
timestamp = message[5]
str_content = message[7]
bytes_extra = message[10]
display_content = message[12]
call_detail = call_decompress(
is_send, bytes_extra, display_content, str_content
)
is_chatroom = 1 if self.contact.is_chatroom else 0
avatar = self.get_avatar_path(is_send, message)
display_name = self.get_display_name(is_send, message)
doc.write(f"""{{ type:50, text:'{call_detail["display_content"]}',call_type:{call_detail["call_type"]},avatar_path:'{avatar}',timestamp:{timestamp},is_chatroom:{is_chatroom},displayname:'{display_name}',}},\n""")
def export(self): def export(self):
print(f"【开始导出 HTML {self.contact.remark}】") print(f"【开始导出 HTML {self.contact.remark}】")
messages = msg_db.get_messages(self.contact.wxid, time_range=self.time_range) messages = msg_db.get_messages(self.contact.wxid, time_range=self.time_range)
@ -311,6 +341,10 @@ class HtmlExporter(ExporterBase):
self.music_share(f, message) self.music_share(f, message)
elif type_ == 49 and sub_type == 5 and self.message_types.get(4905): elif type_ == 49 and sub_type == 5 and self.message_types.get(4905):
self.share_card(f, message) self.share_card(f, message)
elif type_ == 49 and sub_type == 2000 and self.message_types.get(492000):
self.transfer(f, message)
elif type_ == 50 and self.message_types.get(50):
self.call(f, message)
if index % 2000 == 0: if index % 2000 == 0:
print(f"【导出 HTML {self.contact.remark}】{index}/{len(messages)}") print(f"【导出 HTML {self.contact.remark}】{index}/{len(messages)}")
f.write(html_end) f.write(html_end)
@ -325,7 +359,7 @@ class HtmlExporter(ExporterBase):
@return: @return:
""" """
self.num += 1 self.num += 1
print('子线程完成', self.num, '/', self.total_num) print("子线程完成", self.num, "/", self.total_num)
if self.num == self.total_num: if self.num == self.total_num:
# 所有子线程都完成之后就发送完成信号 # 所有子线程都完成之后就发送完成信号
self.okSignal.emit(1) self.okSignal.emit(1)
@ -335,6 +369,7 @@ class OutputMedia(QThread):
""" """
导出语音消息 导出语音消息
""" """
okSingal = pyqtSignal(int) okSingal = pyqtSignal(int)
progressSignal = pyqtSignal(int) progressSignal = pyqtSignal(int)
@ -349,7 +384,9 @@ class OutputMedia(QThread):
is_send = message[4] is_send = message[4]
msgSvrId = message[9] msgSvrId = message[9]
try: try:
audio_path = media_msg_db.get_audio(msgSvrId, output_path=origin_docx_path + "/voice") audio_path = media_msg_db.get_audio(
msgSvrId, output_path=origin_docx_path + "/voice"
)
except: except:
logger.error(traceback.format_exc()) logger.error(traceback.format_exc())
finally: finally:
@ -361,6 +398,7 @@ class OutputEmoji(QThread):
""" """
导出表情包 导出表情包
""" """
okSingal = pyqtSignal(int) okSingal = pyqtSignal(int)
progressSignal = pyqtSignal(int) progressSignal = pyqtSignal(int)
@ -387,6 +425,7 @@ class OutputImage(QThread):
""" """
导出图片 导出图片
""" """
okSingal = pyqtSignal(int) okSingal = pyqtSignal(int)
progressSignal = pyqtSignal(int) progressSignal = pyqtSignal(int)
@ -399,10 +438,10 @@ class OutputImage(QThread):
def count1(self, num): def count1(self, num):
self.num += 1 self.num += 1
print('图片导出完成一个') print("图片导出完成一个")
if self.num == self.child_thread_num: if self.num == self.child_thread_num:
self.okSingal.emit(47) self.okSingal.emit(47)
print('图片导出完成') print("图片导出完成")
def run(self): def run(self):
origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}" origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}"
@ -412,13 +451,19 @@ class OutputImage(QThread):
BytesExtra = message[10] BytesExtra = message[10]
timestamp = message[5] timestamp = message[5]
try: try:
image_path = hard_link_db.get_image(str_content, BytesExtra, thumb=False) image_path = hard_link_db.get_image(
str_content, BytesExtra, thumb=False
)
if not os.path.exists(os.path.join(Me().wx_dir, image_path)): if not os.path.exists(os.path.join(Me().wx_dir, image_path)):
image_thumb_path = hard_link_db.get_image(str_content, BytesExtra, thumb=True) image_thumb_path = hard_link_db.get_image(
str_content, BytesExtra, thumb=True
)
if not os.path.exists(os.path.join(Me().wx_dir, image_thumb_path)): if not os.path.exists(os.path.join(Me().wx_dir, image_thumb_path)):
continue continue
image_path = image_thumb_path image_path = image_thumb_path
image_path = get_image(image_path, base_path=f'/data/聊天记录/{self.contact.remark}/image') image_path = get_image(
image_path, base_path=f"/data/聊天记录/{self.contact.remark}/image"
)
try: try:
os.utime(origin_docx_path + image_path[1:], (timestamp, timestamp)) os.utime(origin_docx_path + image_path[1:], (timestamp, timestamp))
except: except:
@ -456,13 +501,19 @@ class OutputImageChild(QThread):
BytesExtra = message[10] BytesExtra = message[10]
timestamp = message[5] timestamp = message[5]
try: try:
image_path = hard_link_db.get_image(str_content, BytesExtra, thumb=False) image_path = hard_link_db.get_image(
str_content, BytesExtra, thumb=False
)
if not os.path.exists(os.path.join(Me().wx_dir, image_path)): if not os.path.exists(os.path.join(Me().wx_dir, image_path)):
image_thumb_path = hard_link_db.get_image(str_content, BytesExtra, thumb=True) image_thumb_path = hard_link_db.get_image(
str_content, BytesExtra, thumb=True
)
if not os.path.exists(os.path.join(Me().wx_dir, image_thumb_path)): if not os.path.exists(os.path.join(Me().wx_dir, image_thumb_path)):
continue continue
image_path = image_thumb_path image_path = image_thumb_path
image_path = get_image(image_path, base_path=f'/data/聊天记录/{self.contact.remark}/image') image_path = get_image(
image_path, base_path=f"/data/聊天记录/{self.contact.remark}/image"
)
try: try:
os.utime(origin_docx_path + image_path[1:], (timestamp, timestamp)) os.utime(origin_docx_path + image_path[1:], (timestamp, timestamp))
except: except:
@ -472,4 +523,4 @@ class OutputImageChild(QThread):
finally: finally:
self.progressSignal.emit(1) self.progressSignal.emit(1)
self.okSingal.emit(47) self.okSingal.emit(47)
print('图片子线程完成') print("图片子线程完成")

7
app/DataBase/msg.py
View File

@ -35,7 +35,8 @@ def parser_chatroom_message(messages):
a[9]: msgSvrId, a[9]: msgSvrId,
a[10]: BytesExtra, a[10]: BytesExtra,
a[11]: CompressContent, a[11]: CompressContent,
a[12]: msg_sender, （ContactPC 或 ContactDefault 类型，这个才是群聊里的信息发送人，不是群聊或者自己是发送者没有这个字段） a[12]: DisplayContent,
a[13]: msg_sender, （ContactPC 或 ContactDefault 类型，这个才是群聊里的信息发送人，不是群聊或者自己是发送者没有这个字段）
''' '''
updated_messages = [] # 用于存储修改后的消息列表 updated_messages = [] # 用于存储修改后的消息列表
for row in messages: for row in messages:
@ -157,13 +158,15 @@ class Msg:
a[9]: msgSvrId, a[9]: msgSvrId,
a[10]: BytesExtra, a[10]: BytesExtra,
a[11]: CompressContent, a[11]: CompressContent,
a[12]: DisplayContent,
a[13]: 联系人的类（如果是群聊就有，不是的话没有这个字段）
""" """
if not self.open_flag: if not self.open_flag:
return None return None
if time_range: if time_range:
start_time, end_time = time_range start_time, end_time = time_range
sql = f''' sql = f'''
select localId,TalkerId,Type,SubType,IsSender,CreateTime,Status,StrContent,strftime('%Y-%m-%d %H:%M:%S',CreateTime,'unixepoch','localtime') as StrTime,MsgSvrID,BytesExtra,CompressContent select localId,TalkerId,Type,SubType,IsSender,CreateTime,Status,StrContent,strftime('%Y-%m-%d %H:%M:%S',CreateTime,'unixepoch','localtime') as StrTime,MsgSvrID,BytesExtra,CompressContent,DisplayContent
from MSG from MSG
where StrTalker=? where StrTalker=?
{'AND CreateTime>' + str(start_time) + ' AND CreateTime<' + str(end_time) if time_range else ''} {'AND CreateTime>' + str(start_time) + ' AND CreateTime<' + str(end_time) if time_range else ''}

6
app/DataBase/output.py
View File

@ -122,12 +122,12 @@ class ExporterBase(QThread):
def get_avatar_path(self, is_send, message, is_absolute_path=False) -> str: def get_avatar_path(self, is_send, message, is_absolute_path=False) -> str:
if is_absolute_path: if is_absolute_path:
if self.contact.is_chatroom: if self.contact.is_chatroom:
avatar = message[12].avatar_path avatar = message[13].avatar_path
else: else:
avatar = Me().avatar_path if is_send else self.contact.avatar_path avatar = Me().avatar_path if is_send else self.contact.avatar_path
else: else:
if self.contact.is_chatroom: if self.contact.is_chatroom:
avatar = message[12].smallHeadImgUrl avatar = message[13].smallHeadImgUrl
else: else:
avatar = Me().smallHeadImgUrl if is_send else self.contact.smallHeadImgUrl avatar = Me().smallHeadImgUrl if is_send else self.contact.smallHeadImgUrl
return avatar return avatar
@ -137,7 +137,7 @@ class ExporterBase(QThread):
if is_send: if is_send:
display_name = Me().name display_name = Me().name
else: else:
display_name = message[12].remark display_name = message[13].remark
else: else:
display_name = Me().name if is_send else self.contact.remark display_name = Me().name if is_send else self.contact.remark
return escape_js_and_html(display_name) return escape_js_and_html(display_name)

3
app/DataBase/package_msg.py
View File

@ -107,7 +107,8 @@ class PackageMsg:
a[9]: msgSvrId, a[9]: msgSvrId,
a[10]: BytesExtra, a[10]: BytesExtra,
a[11]: CompressContent, a[11]: CompressContent,
a[12]: msg_sender, （ContactPC 或 ContactDefault 类型，这个才是群聊里的信息发送人，不是群聊或者自己是发送者没有这个字段） a[12]: DisplayContent,
a[13]: msg_sender, （ContactPC 或 ContactDefault 类型，这个才是群聊里的信息发送人，不是群聊或者自己是发送者没有这个字段）
''' '''
updated_messages = [] # 用于存储修改后的消息列表 updated_messages = [] # 用于存储修改后的消息列表
messages = msg_db.get_messages(chatroom_wxid) messages = msg_db.get_messages(chatroom_wxid)

8
app/ui/chat/chat_info.py
View File

@ -83,13 +83,13 @@ class ChatInfo(QWidget):
def get_avatar_path(self, is_send, message, is_absolute_path=False) -> str: def get_avatar_path(self, is_send, message, is_absolute_path=False) -> str:
if self.contact.is_chatroom: if self.contact.is_chatroom:
avatar = message[12].smallHeadImgUrl avatar = message[13].smallHeadImgUrl
else: else:
avatar = Me().smallHeadImgUrl if is_send else self.contact.smallHeadImgUrl avatar = Me().smallHeadImgUrl if is_send else self.contact.smallHeadImgUrl
if is_absolute_path: if is_absolute_path:
if self.contact.is_chatroom: if self.contact.is_chatroom:
# message[12].save_avatar() # message[13].save_avatar()
avatar = message[12].avatar avatar = message[13].avatar
else: else:
avatar = Me().avatar if is_send else self.contact.avatar avatar = Me().avatar if is_send else self.contact.avatar
return avatar return avatar
@ -99,7 +99,7 @@ class ChatInfo(QWidget):
if is_send: if is_send:
display_name = Me().name display_name = Me().name
else: else:
display_name = message[12].remark display_name = message[13].remark
else: else:
display_name = None display_name = None
return display_name return display_name

6
app/ui/contact/export/export_dialog.py
View File

@ -22,7 +22,9 @@ types = {
'音乐与音频': 4903, '音乐与音频': 4903,
'文件': 4906, '文件': 4906,
'分享卡片': 4905, '分享卡片': 4905,
'拍一拍等系统消息': 10000 '转账': 492000,
'音视频通话': 50,
'拍一拍等系统消息': 10000,
} }
Stylesheet = """ Stylesheet = """
QPushButton{ QPushButton{
@ -53,7 +55,7 @@ class ExportDialog(QDialog, Ui_Dialog):
if file_type == 'html': if file_type == 'html':
self.export_type = Output.HTML self.export_type = Output.HTML
self.export_choices = {"文本": True, "图片": True, "语音": False, "视频": False, "表情包": False, self.export_choices = {"文本": True, "图片": True, "语音": False, "视频": False, "表情包": False,
'音乐与音频': False, '分享卡片': False, '文件': False, '音乐与音频': False, '分享卡片': False, '文件': False, '转账': False, '音视频通话': False,
'拍一拍等系统消息': True} # 定义导出的数据类型，默认全部选择 '拍一拍等系统消息': True} # 定义导出的数据类型，默认全部选择
elif file_type == 'csv': elif file_type == 'csv':
self.export_type = Output.CSV self.export_type = Output.CSV

6
app/ui/menu/export.py
View File

@ -26,7 +26,9 @@ types = {
'音乐与音频': 4903, '音乐与音频': 4903,
'文件': 4906, '文件': 4906,
'分享卡片': 4905, '分享卡片': 4905,
'拍一拍等系统消息': 10000 '转账': 492000,
'音视频通话': 50,
'拍一拍等系统消息': 10000,
} }
file_format = { file_format = {
'Docx': Output.DOCX, 'Docx': Output.DOCX,
@ -69,7 +71,7 @@ class ExportDialog(QDialog, Ui_Dialog):
self.textBrowser.setVerticalScrollBar(scroll_bar) self.textBrowser.setVerticalScrollBar(scroll_bar)
self.export_choices = {"文本": True, "图片": True, "语音": False, "视频": False, "表情包": False, self.export_choices = {"文本": True, "图片": True, "语音": False, "视频": False, "表情包": False,
'音乐与音频': False, '分享卡片': False, '文件': False, '音乐与音频': False, '分享卡片': False, '文件': False,
'拍一拍等系统消息': True} # 定义导出的数据类型，默认全部选择 '转账': False, '音视频通话': False, '拍一拍等系统消息': True} # 定义导出的数据类型
self.setWindowTitle(title) self.setWindowTitle(title)
self.checkBox_word.setEnabled(False) self.checkBox_word.setEnabled(False)
self.checkBox_word.setText('Docx(暂时不可用)') self.checkBox_word.setText('Docx(暂时不可用)')

295
app/util/compress_content.py
View File

@ -19,31 +19,32 @@ def decompress_CompressContent(data):
:return: :return:
""" """
if data is None or not isinstance(data, bytes): if data is None or not isinstance(data, bytes):
return '' return ""
try: try:
dst = lz4.block.decompress(data, uncompressed_size=len(data) << 10) dst = lz4.block.decompress(data, uncompressed_size=len(data) << 10)
decoded_string = dst.decode().replace('\x00', '') # Remove any null characters decoded_string = dst.decode().replace("\x00", "") # Remove any null characters
except: except:
print("Decompression failed: potentially corrupt input or insufficient buffer size.") print(
return '' "Decompression failed: potentially corrupt input or insufficient buffer size."
)
return ""
return decoded_string return decoded_string
def escape_js_and_html(input_str): def escape_js_and_html(input_str):
if not input_str: if not input_str:
return '' return ""
# 转义HTML特殊字符 # 转义HTML特殊字符
html_escaped = html.escape(input_str, quote=False) html_escaped = html.escape(input_str, quote=False)
# 手动处理JavaScript转义字符 # 手动处理JavaScript转义字符
js_escaped = ( js_escaped = (
html_escaped html_escaped.replace("\\", "\\\\")
.replace("\\", "\\\\")
.replace("'", r"\'") .replace("'", r"\'")
.replace('"', r'\"') .replace('"', r"\"")
.replace("\n", r'\n') .replace("\n", r"\n")
.replace("\r", r'\r') .replace("\r", r"\r")
.replace("\t", r'\t') .replace("\t", r"\t")
) )
return js_escaped return js_escaped
@ -53,168 +54,232 @@ def parser_reply(data: bytes):
xml_content = decompress_CompressContent(data) xml_content = decompress_CompressContent(data)
if not xml_content: if not xml_content:
return { return {
'type': 57, "type": 57,
'title': "发生错误", "title": "发生错误",
'refer': { "refer": {
'type': '1', "type": "1",
'content': '引用错误', "content": "引用错误",
'displayname': '用户名', "displayname": "用户名",
}, },
"is_error": True "is_error": True,
} }
try: try:
root = ET.XML(xml_content) root = ET.XML(xml_content)
appmsg = root.find('appmsg') appmsg = root.find("appmsg")
msg_type = int(appmsg.find('type').text) msg_type = int(appmsg.find("type").text)
title = appmsg.find('title').text title = appmsg.find("title").text
refermsg_content = appmsg.find('refermsg').find('content').text refermsg_content = appmsg.find("refermsg").find("content").text
refermsg_type = int(appmsg.find('refermsg').find('type').text) refermsg_type = int(appmsg.find("refermsg").find("type").text)
refermsg_displayname = appmsg.find('refermsg').find('displayname').text refermsg_displayname = appmsg.find("refermsg").find("displayname").text
return { return {
'type': msg_type, "type": msg_type,
'title': title, "title": title,
'refer': None if refermsg_type != 1 else { "refer": None
'type': refermsg_type, if refermsg_type != 1
'content': refermsg_content.lstrip("\n"), else {
'displayname': refermsg_displayname, "type": refermsg_type,
"content": refermsg_content.lstrip("\n"),
"displayname": refermsg_displayname,
}, },
"is_error": False "is_error": False,
} }
except: except:
return { return {
'type': 57, "type": 57,
'title': "发生错误", "title": "发生错误",
'refer': { "refer": {
'type': '1', "type": "1",
'content': '引用错误', "content": "引用错误",
'displayname': '用户名', "displayname": "用户名",
}, },
"is_error": True "is_error": True,
} }
def music_share(data: bytes): def music_share(data: bytes):
xml_content = decompress_CompressContent(data) xml_content = decompress_CompressContent(data)
if not xml_content: if not xml_content:
return { return {"type": 3, "title": "发生错误", "is_error": True}
'type': 3,
'title': "发生错误",
"is_error": True
}
try: try:
root = ET.XML(xml_content) root = ET.XML(xml_content)
appmsg = root.find('appmsg') appmsg = root.find("appmsg")
msg_type = int(appmsg.find('type').text) msg_type = int(appmsg.find("type").text)
title = appmsg.find('title').text title = appmsg.find("title").text
if len(title) >= 39: if len(title) >= 39:
title = title[:38] + '...' title = title[:38] + "..."
artist = appmsg.find('des').text artist = appmsg.find("des").text
link_url = appmsg.find('url').text # 链接地址 link_url = appmsg.find("url").text # 链接地址
audio_url = get_audio_url(appmsg.find('dataurl').text) # 播放地址 audio_url = get_audio_url(appmsg.find("dataurl").text) # 播放地址
website_name = get_website_name(link_url) website_name = get_website_name(link_url)
return { return {
'type': msg_type, "type": msg_type,
'title': escape_js_and_html(title), "title": escape_js_and_html(title),
'artist': escape_js_and_html(artist), "artist": escape_js_and_html(artist),
'link_url': link_url, "link_url": link_url,
'audio_url': audio_url, "audio_url": audio_url,
'website_name': escape_js_and_html(website_name), "website_name": escape_js_and_html(website_name),
"is_error": False "is_error": False,
} }
except Exception as e: except Exception as e:
print(f"Music Share Error: {e}") print(f"Music Share Error: {e}")
return { return {"type": 3, "title": "发生错误", "is_error": True}
'type': 3,
'title': "发生错误",
"is_error": True
}
def share_card(bytesExtra, compress_content_): def share_card(bytesExtra, compress_content_):
title, des, url, show_display_name, thumbnail, app_logo = '', '', '', '', '', '' title, des, url, show_display_name, thumbnail, app_logo = "", "", "", "", "", ""
try: try:
xml = decompress_CompressContent(compress_content_) xml = decompress_CompressContent(compress_content_)
root = ET.XML(xml) root = ET.XML(xml)
appmsg = root.find('appmsg') appmsg = root.find("appmsg")
title = appmsg.find('title').text title = appmsg.find("title").text
try: try:
des = appmsg.find('des').text des = appmsg.find("des").text
except: except:
des = '' des = ""
url = appmsg.find('url').text url = appmsg.find("url").text
appinfo = root.find('appinfo') appinfo = root.find("appinfo")
show_display_name = appmsg.find('sourcedisplayname') show_display_name = appmsg.find("sourcedisplayname")
sourceusername = appmsg.find('sourceusername') sourceusername = appmsg.find("sourceusername")
if show_display_name is not None: if show_display_name is not None:
show_display_name = show_display_name.text show_display_name = show_display_name.text
else: else:
if appinfo is not None: if appinfo is not None:
show_display_name = appinfo.find('appname').text show_display_name = appinfo.find("appname").text
msg_bytes = MessageBytesExtra() msg_bytes = MessageBytesExtra()
msg_bytes.ParseFromString(bytesExtra) msg_bytes.ParseFromString(bytesExtra)
app_logo = '' app_logo = ""
thumbnail = '' thumbnail = ""
for tmp in msg_bytes.message2: for tmp in msg_bytes.message2:
if tmp.field1 == 3: if tmp.field1 == 3:
thumbnail = tmp.field2 thumbnail = tmp.field2
thumbnail = "\\".join(thumbnail.split('\\')[1:]) thumbnail = "\\".join(thumbnail.split("\\")[1:])
if tmp.field2 == 4: if tmp.field2 == 4:
app_logo = tmp.field2 app_logo = tmp.field2
app_logo = "\\".join(app_logo.split('\\')[1:]) app_logo = "\\".join(app_logo.split("\\")[1:])
if sourceusername is not None: if sourceusername is not None:
from app.DataBase import micro_msg_db # 放上面会导致循环依赖 from app.DataBase import micro_msg_db # 放上面会导致循环依赖
contact = micro_msg_db.get_contact_by_username(sourceusername.text) contact = micro_msg_db.get_contact_by_username(sourceusername.text)
if contact: if contact:
app_logo = contact[7] app_logo = contact[7]
finally: finally:
return { return {
'title': escape_js_and_html(title), "title": escape_js_and_html(title),
'description': escape_js_and_html(des), "description": escape_js_and_html(des),
'url': escape_js_and_html(url), "url": escape_js_and_html(url),
'app_name': escape_js_and_html(show_display_name), "app_name": escape_js_and_html(show_display_name),
'thumbnail': thumbnail, "thumbnail": thumbnail,
'app_logo': app_logo "app_logo": app_logo,
} }
def transfer_decompress(compress_content_):
"""
return dict
feedesc: 钱数，str类型，包含一个前缀币种符号（除人民币￥之外未测试）;
pay_memo: 转账备注;
receiver_username: 接受转账人的 wxid; （因为电脑上只有私聊页面会显示收款所以这个字段没有也罢，不要轻易使用，因为可能为空）
paysubtype: int 类型，1 为发出转账，3 为接受转账，4 为退还转账;
"""
feedesc, pay_memo, receiver_username, paysubtype = "", "", "", ""
try:
xml = decompress_CompressContent(compress_content_)
root = ET.XML(xml)
appmsg = root.find("appmsg")
wcpayinfo = appmsg.find("wcpayinfo")
paysubtype = int(wcpayinfo.find("paysubtype").text)
feedesc = wcpayinfo.find("feedesc").text
pay_memo = wcpayinfo.find("pay_memo").text
receiver_username = wcpayinfo.find("receiver_username").text
finally:
return {
"feedesc": feedesc,
"pay_memo": escape_js_and_html(pay_memo),
"receiver_username": receiver_username,
"paysubtype": paysubtype,
}
def call_decompress(is_send, bytes_extra, display_content, str_content): # 音视频通话
"""
return dict
call_type: int 类型，0 为视频，1为语音; （返回为 2 是未知错误）
display_content: str 类型，页面显示的话;
"""
call_type = 2
call_length = 0
msg_bytes = MessageBytesExtra()
msg_bytes.ParseFromString(bytes_extra)
# message2 字段 1: 发送人wxid; 字段 3: "1"是语音，"0"是视频; 字段 4: 通话时长
for i in msg_bytes.message2:
if i.field1 == 3:
call_type = int(i.field2)
elif i.field1 == 4:
call_length = int(i.field2)
try:
if display_content == "":
if str_content == "11":
h, m, s = (
call_length // 3600,
(call_length % 3600) // 60,
call_length % 60,
)
display_content = f"通话时长 {f'{h:02d}:' if h else ''}{m:02d}:{s:02d}"
else:
display_content = {
"5": ("" if is_send else "对方") + "已取消",
"8": ("对方" if is_send else "") + "已拒绝",
"7": "已在其他设备接听",
"12": "已在其他设备拒绝",
}[str_content]
except KeyError:
display_content = "未知类型，您可以把这条消息对应的微信界面消息反馈给我们"
return {
"call_type": call_type,
"display_content": display_content,
}
def get_website_name(url): def get_website_name(url):
parsed_url = urlparse(url) parsed_url = urlparse(url)
domain = f"{parsed_url.scheme}://{parsed_url.netloc}" domain = f"{parsed_url.scheme}://{parsed_url.netloc}"
website_name = '' website_name = ""
try: try:
response = requests.get(domain, allow_redirects=False) response = requests.get(domain, allow_redirects=False)
if response.status_code == 200: if response.status_code == 200:
soup = BeautifulSoup(response.content, 'html.parser') soup = BeautifulSoup(response.content, "html.parser")
website_name = soup.title.string.strip() website_name = soup.title.string.strip()
elif response.status_code == 302: elif response.status_code == 302:
domain = response.headers['Location'] domain = response.headers["Location"]
response = requests.get(domain, allow_redirects=False) response = requests.get(domain, allow_redirects=False)
soup = BeautifulSoup(response.content, 'html.parser') soup = BeautifulSoup(response.content, "html.parser")
website_name = soup.title.string.strip() website_name = soup.title.string.strip()
else: else:
response = requests.get(url, allow_redirects=False) response = requests.get(url, allow_redirects=False)
if response.status_code == 200: if response.status_code == 200:
soup = BeautifulSoup(response.content, 'html.parser') soup = BeautifulSoup(response.content, "html.parser")
website_name = soup.title.string.strip() website_name = soup.title.string.strip()
index = website_name.find("-") index = website_name.find("-")
if index != -1: # 如果找到了 "-" if index != -1: # 如果找到了 "-"
website_name = website_name[index + 1:].strip() website_name = website_name[index + 1 :].strip()
except Exception as e: except Exception as e:
print(f"Get Website Info Error: {e}") print(f"Get Website Info Error: {e}")
return website_name return website_name
def get_audio_url(url): def get_audio_url(url):
path = '' path = ""
try: try:
response = requests.get(url, allow_redirects=False) response = requests.get(url, allow_redirects=False)
# 检查响应状态码 # 检查响应状态码
if response.status_code == 302: if response.status_code == 302:
path = response.headers['Location'] path = response.headers["Location"]
elif response.status_code == 200: elif response.status_code == 200:
print('音乐文件已失效,url:' + url) print("音乐文件已失效,url:" + url)
else: else:
print('音乐文件地址获取失败,url:' + url + ',状态码' + str(response.status_code)) print("音乐文件地址获取失败,url:" + url + ",状态码" + str(response.status_code))
except Exception as e: except Exception as e:
print(f"Get Audio Url Error: {e}") print(f"Get Audio Url Error: {e}")
return path return path
@ -223,45 +288,37 @@ def get_audio_url(url):
def file(bytes_extra, compress_content, output_path): def file(bytes_extra, compress_content, output_path):
xml_content = decompress_CompressContent(compress_content) xml_content = decompress_CompressContent(compress_content)
if not xml_content: if not xml_content:
return { return {"type": 6, "title": "发生错误", "is_error": True}
'type': 6,
'title': "发生错误",
"is_error": True
}
try: try:
root = ET.XML(xml_content) root = ET.XML(xml_content)
appmsg = root.find('appmsg') appmsg = root.find("appmsg")
msg_type = int(appmsg.find('type').text) msg_type = int(appmsg.find("type").text)
file_name = appmsg.find('title').text file_name = appmsg.find("title").text
pattern = r'[\\/:*?"<>|\r\n]+' pattern = r'[\\/:*?"<>|\r\n]+'
file_name = re.sub(pattern, "_", file_name) file_name = re.sub(pattern, "_", file_name)
appattach = appmsg.find('appattach') appattach = appmsg.find("appattach")
file_len = int(appattach.find('totallen').text) file_len = int(appattach.find("totallen").text)
app_name = '' app_name = ""
file_len = format_bytes(file_len) file_len = format_bytes(file_len)
file_ext = appattach.find('fileext').text file_ext = appattach.find("fileext").text
if root.find("appinfo") is not None: if root.find("appinfo") is not None:
app_info = root.find('appinfo') app_info = root.find("appinfo")
app_name = app_info.find('appname').text app_name = app_info.find("appname").text
if app_name is None: if app_name is None:
app_name = '' app_name = ""
file_path = get_file(bytes_extra, file_name, output_path) file_path = get_file(bytes_extra, file_name, output_path)
return { return {
'type': msg_type, "type": msg_type,
'file_name': escape_js_and_html(file_name), "file_name": escape_js_and_html(file_name),
'file_len': file_len, "file_len": file_len,
'file_ext': file_ext, "file_ext": file_ext,
'file_path': file_path, "file_path": file_path,
'app_name': escape_js_and_html(app_name), "app_name": escape_js_and_html(app_name),
"is_error": False "is_error": False,
} }
except Exception as e: except Exception as e:
print(f"File Get Info Error: {e}") print(f"File Get Info Error: {e}")
return { return {"type": 6, "title": "发生错误", "is_error": True}
'type': 6,
'title': "发生错误",
"is_error": True
}
def format_bytes(size): def format_bytes(size):