diff --git a/app/DataBase/exporter_csv.py b/app/DataBase/exporter_csv.py index 5536110..941d89f 100644 --- a/app/DataBase/exporter_csv.py +++ b/app/DataBase/exporter_csv.py @@ -22,7 +22,7 @@ class CSVExporter(ExporterBase): # 写入数据 # writer.writerows(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]) print(f"【完成导出 CSV {self.contact.remark}】") self.okSignal.emit(1) diff --git a/app/DataBase/exporter_html.py b/app/DataBase/exporter_html.py index f6ff22f..c6c40c9 100644 --- a/app/DataBase/exporter_html.py +++ b/app/DataBase/exporter_html.py @@ -11,7 +11,7 @@ from app.DataBase.output import ExporterBase, escape_js_and_html from app.log import logger from app.person import Me 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.image import get_image_path, get_image 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''' ) + 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): print(f"【开始导出 HTML {self.contact.remark}】") 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) elif type_ == 49 and sub_type == 5 and self.message_types.get(4905): 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: print(f"【导出 HTML {self.contact.remark}】{index}/{len(messages)}") f.write(html_end) @@ -325,7 +359,7 @@ class HtmlExporter(ExporterBase): @return: """ self.num += 1 - print('子线程完成', self.num, '/', self.total_num) + print("子线程完成", self.num, "/", self.total_num) if self.num == self.total_num: # 所有子线程都完成之后就发送完成信号 self.okSignal.emit(1) @@ -335,6 +369,7 @@ class OutputMedia(QThread): """ 导出语音消息 """ + okSingal = pyqtSignal(int) progressSignal = pyqtSignal(int) @@ -349,7 +384,9 @@ class OutputMedia(QThread): is_send = message[4] msgSvrId = message[9] 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: logger.error(traceback.format_exc()) finally: @@ -361,6 +398,7 @@ class OutputEmoji(QThread): """ 导出表情包 """ + okSingal = pyqtSignal(int) progressSignal = pyqtSignal(int) @@ -387,6 +425,7 @@ class OutputImage(QThread): """ 导出图片 """ + okSingal = pyqtSignal(int) progressSignal = pyqtSignal(int) @@ -399,10 +438,10 @@ class OutputImage(QThread): def count1(self, num): self.num += 1 - print('图片导出完成一个') + print("图片导出完成一个") if self.num == self.child_thread_num: self.okSingal.emit(47) - print('图片导出完成') + print("图片导出完成") def run(self): origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}" @@ -412,13 +451,19 @@ class OutputImage(QThread): BytesExtra = message[10] timestamp = message[5] 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)): - 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)): continue 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: os.utime(origin_docx_path + image_path[1:], (timestamp, timestamp)) except: @@ -456,13 +501,19 @@ class OutputImageChild(QThread): BytesExtra = message[10] timestamp = message[5] 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)): - 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)): continue 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: os.utime(origin_docx_path + image_path[1:], (timestamp, timestamp)) except: @@ -472,4 +523,4 @@ class OutputImageChild(QThread): finally: self.progressSignal.emit(1) self.okSingal.emit(47) - print('图片子线程完成') + print("图片子线程完成") diff --git a/app/DataBase/msg.py b/app/DataBase/msg.py index 8a0f205..5b25a4f 100644 --- a/app/DataBase/msg.py +++ b/app/DataBase/msg.py @@ -35,7 +35,8 @@ def parser_chatroom_message(messages): a[9]: msgSvrId, a[10]: BytesExtra, a[11]: CompressContent, - a[12]: msg_sender, （ContactPC 或 ContactDefault 类型，这个才是群聊里的信息发送人，不是群聊或者自己是发送者没有这个字段） + a[12]: DisplayContent, + a[13]: msg_sender, （ContactPC 或 ContactDefault 类型，这个才是群聊里的信息发送人，不是群聊或者自己是发送者没有这个字段） ''' updated_messages = [] # 用于存储修改后的消息列表 for row in messages: @@ -157,13 +158,15 @@ class Msg: a[9]: msgSvrId, a[10]: BytesExtra, a[11]: CompressContent, + a[12]: DisplayContent, + a[13]: 联系人的类（如果是群聊就有，不是的话没有这个字段） """ if not self.open_flag: return None if time_range: start_time, end_time = time_range 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 where StrTalker=? {'AND CreateTime>' + str(start_time) + ' AND CreateTime<' + str(end_time) if time_range else ''} diff --git a/app/DataBase/output.py b/app/DataBase/output.py index b35abf4..359ee4a 100644 --- a/app/DataBase/output.py +++ b/app/DataBase/output.py @@ -122,12 +122,12 @@ class ExporterBase(QThread): def get_avatar_path(self, is_send, message, is_absolute_path=False) -> str: if is_absolute_path: if self.contact.is_chatroom: - avatar = message[12].avatar_path + avatar = message[13].avatar_path else: avatar = Me().avatar_path if is_send else self.contact.avatar_path else: if self.contact.is_chatroom: - avatar = message[12].smallHeadImgUrl + avatar = message[13].smallHeadImgUrl else: avatar = Me().smallHeadImgUrl if is_send else self.contact.smallHeadImgUrl return avatar @@ -137,7 +137,7 @@ class ExporterBase(QThread): if is_send: display_name = Me().name else: - display_name = message[12].remark + display_name = message[13].remark else: display_name = Me().name if is_send else self.contact.remark return escape_js_and_html(display_name) diff --git a/app/DataBase/package_msg.py b/app/DataBase/package_msg.py index a371d89..bd3ed99 100644 --- a/app/DataBase/package_msg.py +++ b/app/DataBase/package_msg.py @@ -107,7 +107,8 @@ class PackageMsg: a[9]: msgSvrId, a[10]: BytesExtra, a[11]: CompressContent, - a[12]: msg_sender, （ContactPC 或 ContactDefault 类型，这个才是群聊里的信息发送人，不是群聊或者自己是发送者没有这个字段） + a[12]: DisplayContent, + a[13]: msg_sender, （ContactPC 或 ContactDefault 类型，这个才是群聊里的信息发送人，不是群聊或者自己是发送者没有这个字段） ''' updated_messages = [] # 用于存储修改后的消息列表 messages = msg_db.get_messages(chatroom_wxid) diff --git a/app/ui/chat/chat_info.py b/app/ui/chat/chat_info.py index f71d5c9..a88232e 100644 --- a/app/ui/chat/chat_info.py +++ b/app/ui/chat/chat_info.py @@ -83,13 +83,13 @@ class ChatInfo(QWidget): def get_avatar_path(self, is_send, message, is_absolute_path=False) -> str: if self.contact.is_chatroom: - avatar = message[12].smallHeadImgUrl + avatar = message[13].smallHeadImgUrl else: avatar = Me().smallHeadImgUrl if is_send else self.contact.smallHeadImgUrl if is_absolute_path: if self.contact.is_chatroom: - # message[12].save_avatar() - avatar = message[12].avatar + # message[13].save_avatar() + avatar = message[13].avatar else: avatar = Me().avatar if is_send else self.contact.avatar return avatar @@ -99,7 +99,7 @@ class ChatInfo(QWidget): if is_send: display_name = Me().name else: - display_name = message[12].remark + display_name = message[13].remark else: display_name = None return display_name diff --git a/app/ui/contact/export/export_dialog.py b/app/ui/contact/export/export_dialog.py index df56700..bb354f8 100644 --- a/app/ui/contact/export/export_dialog.py +++ b/app/ui/contact/export/export_dialog.py @@ -22,7 +22,9 @@ types = { '音乐与音频': 4903, '文件': 4906, '分享卡片': 4905, - '拍一拍等系统消息': 10000 + '转账': 492000, + '音视频通话': 50, + '拍一拍等系统消息': 10000, } Stylesheet = """ QPushButton{ @@ -53,7 +55,7 @@ class ExportDialog(QDialog, Ui_Dialog): if file_type == 'html': self.export_type = Output.HTML self.export_choices = {"文本": True, "图片": True, "语音": False, "视频": False, "表情包": False, - '音乐与音频': False, '分享卡片': False, '文件': False, + '音乐与音频': False, '分享卡片': False, '文件': False, '转账': False, '音视频通话': False, '拍一拍等系统消息': True} # 定义导出的数据类型，默认全部选择 elif file_type == 'csv': self.export_type = Output.CSV diff --git a/app/ui/menu/export.py b/app/ui/menu/export.py index b168de0..788b6a1 100644 --- a/app/ui/menu/export.py +++ b/app/ui/menu/export.py @@ -26,7 +26,9 @@ types = { '音乐与音频': 4903, '文件': 4906, '分享卡片': 4905, - '拍一拍等系统消息': 10000 + '转账': 492000, + '音视频通话': 50, + '拍一拍等系统消息': 10000, } file_format = { 'Docx': Output.DOCX, @@ -69,7 +71,7 @@ class ExportDialog(QDialog, Ui_Dialog): self.textBrowser.setVerticalScrollBar(scroll_bar) self.export_choices = {"文本": True, "图片": True, "语音": False, "视频": False, "表情包": False, '音乐与音频': False, '分享卡片': False, '文件': False, - '拍一拍等系统消息': True} # 定义导出的数据类型，默认全部选择 + '转账': False, '音视频通话': False, '拍一拍等系统消息': True} # 定义导出的数据类型 self.setWindowTitle(title) self.checkBox_word.setEnabled(False) self.checkBox_word.setText('Docx(暂时不可用)') diff --git a/app/util/compress_content.py b/app/util/compress_content.py index 31a5b94..f21af90 100644 --- a/app/util/compress_content.py +++ b/app/util/compress_content.py @@ -19,31 +19,32 @@ def decompress_CompressContent(data): :return: """ if data is None or not isinstance(data, bytes): - return '' + return "" try: 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: - print("Decompression failed: potentially corrupt input or insufficient buffer size.") - return '' + print( + "Decompression failed: potentially corrupt input or insufficient buffer size." + ) + return "" return decoded_string def escape_js_and_html(input_str): if not input_str: - return '' + return "" # 转义HTML特殊字符 html_escaped = html.escape(input_str, quote=False) # 手动处理JavaScript转义字符 js_escaped = ( - html_escaped - .replace("\\", "\\\\") + html_escaped.replace("\\", "\\\\") .replace("'", r"\'") - .replace('"', r'\"') - .replace("\n", r'\n') - .replace("\r", r'\r') - .replace("\t", r'\t') + .replace('"', r"\"") + .replace("\n", r"\n") + .replace("\r", r"\r") + .replace("\t", r"\t") ) return js_escaped @@ -53,168 +54,232 @@ def parser_reply(data: bytes): xml_content = decompress_CompressContent(data) if not xml_content: return { - 'type': 57, - 'title': "发生错误", - 'refer': { - 'type': '1', - 'content': '引用错误', - 'displayname': '用户名', + "type": 57, + "title": "发生错误", + "refer": { + "type": "1", + "content": "引用错误", + "displayname": "用户名", }, - "is_error": True + "is_error": True, } try: root = ET.XML(xml_content) - appmsg = root.find('appmsg') - msg_type = int(appmsg.find('type').text) - title = appmsg.find('title').text - refermsg_content = appmsg.find('refermsg').find('content').text - refermsg_type = int(appmsg.find('refermsg').find('type').text) - refermsg_displayname = appmsg.find('refermsg').find('displayname').text + appmsg = root.find("appmsg") + msg_type = int(appmsg.find("type").text) + title = appmsg.find("title").text + refermsg_content = appmsg.find("refermsg").find("content").text + refermsg_type = int(appmsg.find("refermsg").find("type").text) + refermsg_displayname = appmsg.find("refermsg").find("displayname").text return { - 'type': msg_type, - 'title': title, - 'refer': None if refermsg_type != 1 else { - 'type': refermsg_type, - 'content': refermsg_content.lstrip("\n"), - 'displayname': refermsg_displayname, + "type": msg_type, + "title": title, + "refer": None + if refermsg_type != 1 + else { + "type": refermsg_type, + "content": refermsg_content.lstrip("\n"), + "displayname": refermsg_displayname, }, - "is_error": False + "is_error": False, } except: return { - 'type': 57, - 'title': "发生错误", - 'refer': { - 'type': '1', - 'content': '引用错误', - 'displayname': '用户名', + "type": 57, + "title": "发生错误", + "refer": { + "type": "1", + "content": "引用错误", + "displayname": "用户名", }, - "is_error": True + "is_error": True, } def music_share(data: bytes): xml_content = decompress_CompressContent(data) if not xml_content: - return { - 'type': 3, - 'title': "发生错误", - "is_error": True - } + return {"type": 3, "title": "发生错误", "is_error": True} try: root = ET.XML(xml_content) - appmsg = root.find('appmsg') - msg_type = int(appmsg.find('type').text) - title = appmsg.find('title').text + appmsg = root.find("appmsg") + msg_type = int(appmsg.find("type").text) + title = appmsg.find("title").text if len(title) >= 39: - title = title[:38] + '...' - artist = appmsg.find('des').text - link_url = appmsg.find('url').text # 链接地址 - audio_url = get_audio_url(appmsg.find('dataurl').text) # 播放地址 + title = title[:38] + "..." + artist = appmsg.find("des").text + link_url = appmsg.find("url").text # 链接地址 + audio_url = get_audio_url(appmsg.find("dataurl").text) # 播放地址 website_name = get_website_name(link_url) return { - 'type': msg_type, - 'title': escape_js_and_html(title), - 'artist': escape_js_and_html(artist), - 'link_url': link_url, - 'audio_url': audio_url, - 'website_name': escape_js_and_html(website_name), - "is_error": False + "type": msg_type, + "title": escape_js_and_html(title), + "artist": escape_js_and_html(artist), + "link_url": link_url, + "audio_url": audio_url, + "website_name": escape_js_and_html(website_name), + "is_error": False, } except Exception as e: print(f"Music Share Error: {e}") - return { - 'type': 3, - 'title': "发生错误", - "is_error": True - } + return {"type": 3, "title": "发生错误", "is_error": True} 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: xml = decompress_CompressContent(compress_content_) root = ET.XML(xml) - appmsg = root.find('appmsg') - title = appmsg.find('title').text + appmsg = root.find("appmsg") + title = appmsg.find("title").text try: - des = appmsg.find('des').text + des = appmsg.find("des").text except: - des = '' - url = appmsg.find('url').text - appinfo = root.find('appinfo') - show_display_name = appmsg.find('sourcedisplayname') - sourceusername = appmsg.find('sourceusername') + des = "" + url = appmsg.find("url").text + appinfo = root.find("appinfo") + show_display_name = appmsg.find("sourcedisplayname") + sourceusername = appmsg.find("sourceusername") if show_display_name is not None: show_display_name = show_display_name.text else: if appinfo is not None: - show_display_name = appinfo.find('appname').text + show_display_name = appinfo.find("appname").text msg_bytes = MessageBytesExtra() msg_bytes.ParseFromString(bytesExtra) - app_logo = '' - thumbnail = '' + app_logo = "" + thumbnail = "" for tmp in msg_bytes.message2: if tmp.field1 == 3: thumbnail = tmp.field2 - thumbnail = "\\".join(thumbnail.split('\\')[1:]) + thumbnail = "\\".join(thumbnail.split("\\")[1:]) if tmp.field2 == 4: app_logo = tmp.field2 - app_logo = "\\".join(app_logo.split('\\')[1:]) + app_logo = "\\".join(app_logo.split("\\")[1:]) if sourceusername is not None: from app.DataBase import micro_msg_db # 放上面会导致循环依赖 + contact = micro_msg_db.get_contact_by_username(sourceusername.text) if contact: app_logo = contact[7] finally: return { - 'title': escape_js_and_html(title), - 'description': escape_js_and_html(des), - 'url': escape_js_and_html(url), - 'app_name': escape_js_and_html(show_display_name), - 'thumbnail': thumbnail, - 'app_logo': app_logo + "title": escape_js_and_html(title), + "description": escape_js_and_html(des), + "url": escape_js_and_html(url), + "app_name": escape_js_and_html(show_display_name), + "thumbnail": thumbnail, + "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): parsed_url = urlparse(url) domain = f"{parsed_url.scheme}://{parsed_url.netloc}" - website_name = '' + website_name = "" try: response = requests.get(domain, allow_redirects=False) if response.status_code == 200: - soup = BeautifulSoup(response.content, 'html.parser') + soup = BeautifulSoup(response.content, "html.parser") website_name = soup.title.string.strip() elif response.status_code == 302: - domain = response.headers['Location'] + domain = response.headers["Location"] 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() else: response = requests.get(url, allow_redirects=False) if response.status_code == 200: - soup = BeautifulSoup(response.content, 'html.parser') + soup = BeautifulSoup(response.content, "html.parser") website_name = soup.title.string.strip() index = website_name.find("-") if index != -1: # 如果找到了 "-" - website_name = website_name[index + 1:].strip() + website_name = website_name[index + 1 :].strip() except Exception as e: print(f"Get Website Info Error: {e}") return website_name def get_audio_url(url): - path = '' + path = "" try: response = requests.get(url, allow_redirects=False) # 检查响应状态码 if response.status_code == 302: - path = response.headers['Location'] + path = response.headers["Location"] elif response.status_code == 200: - print('音乐文件已失效,url:' + url) + print("音乐文件已失效,url:" + url) else: - print('音乐文件地址获取失败,url:' + url + ',状态码' + str(response.status_code)) + print("音乐文件地址获取失败,url:" + url + ",状态码" + str(response.status_code)) except Exception as e: print(f"Get Audio Url Error: {e}") return path @@ -223,45 +288,37 @@ def get_audio_url(url): def file(bytes_extra, compress_content, output_path): xml_content = decompress_CompressContent(compress_content) if not xml_content: - return { - 'type': 6, - 'title': "发生错误", - "is_error": True - } + return {"type": 6, "title": "发生错误", "is_error": True} try: root = ET.XML(xml_content) - appmsg = root.find('appmsg') - msg_type = int(appmsg.find('type').text) - file_name = appmsg.find('title').text + appmsg = root.find("appmsg") + msg_type = int(appmsg.find("type").text) + file_name = appmsg.find("title").text pattern = r'[\\/:*?"<>|\r\n]+' file_name = re.sub(pattern, "_", file_name) - appattach = appmsg.find('appattach') - file_len = int(appattach.find('totallen').text) - app_name = '' + appattach = appmsg.find("appattach") + file_len = int(appattach.find("totallen").text) + app_name = "" 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: - app_info = root.find('appinfo') - app_name = app_info.find('appname').text + app_info = root.find("appinfo") + app_name = app_info.find("appname").text if app_name is None: - app_name = '' + app_name = "" file_path = get_file(bytes_extra, file_name, output_path) return { - 'type': msg_type, - 'file_name': escape_js_and_html(file_name), - 'file_len': file_len, - 'file_ext': file_ext, - 'file_path': file_path, - 'app_name': escape_js_and_html(app_name), - "is_error": False + "type": msg_type, + "file_name": escape_js_and_html(file_name), + "file_len": file_len, + "file_ext": file_ext, + "file_path": file_path, + "app_name": escape_js_and_html(app_name), + "is_error": False, } except Exception as e: print(f"File Get Info Error: {e}") - return { - 'type': 6, - 'title': "发生错误", - "is_error": True - } + return {"type": 6, "title": "发生错误", "is_error": True} def format_bytes(size):