diff --git a/app/DataBase/output_pc.py b/app/DataBase/output_pc.py index e186e600..0127f2b8 100644 --- a/app/DataBase/output_pc.py +++ b/app/DataBase/output_pc.py @@ -18,6 +18,39 @@ os.makedirs('./data/聊天记录', exist_ok=True) +WINDOWS_RESERVED_NAMES = { + "CON", "PRN", "AUX", "NUL", + "COM1", "COM2", "COM3", "COM4", "COM5", + "COM6", "COM7", "COM8", "COM9", + "LPT1", "LPT2", "LPT3", "LPT4", "LPT5", + "LPT6", "LPT7", "LPT8", "LPT9", +} + + + + +def sanitize_filename(name, fallback="unknown"): + """Return a Windows-safe file or directory name.""" + name = str(name or fallback) + + invalid_chars = '<>:"/\\|?*' + + name = "".join( + "_" if char in invalid_chars or ord(char) < 32 else char + for char in name + ) + + # Windows filenames cannot end with spaces or periods. + name = name.rstrip(" .") + + if not name: + name = fallback + + # Windows reserved device names are invalid even with an extension. + if name.split(".")[0].upper() in WINDOWS_RESERVED_NAMES: + name = f"_{name}" + + return name def makedirs(path): os.makedirs(path, exist_ok=True) @@ -190,7 +223,7 @@ def text(self, doc, message): ) def image(self, doc, message): - origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}" + origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{sanitize_filename(self.contact.remark or self.contact.wxid)}" type_ = message[2] str_content = message[7] str_time = message[8] @@ -210,7 +243,7 @@ def image(self, doc, message): image_path = image_thumb_path if image_path is None and image_thumb_path is None: return - image_path = path.get_relative_path(image_path, base_path=f'/data/聊天记录/{self.contact.remark}/image') + image_path = path.get_relative_path(image_path, base_path=f'/data/聊天记录/{sanitize_filename(self.contact.remark or self.contact.wxid)}/image') image_path = image_path.replace('/', '\\') os.utime(origin_docx_path + image_path[1:], (timestamp, timestamp)) print(origin_docx_path + image_path[1:]) @@ -230,7 +263,7 @@ def image(self, doc, message): ) def audio(self, doc, message): - origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}" + origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{sanitize_filename(self.contact.remark or self.contact.wxid)}" str_content = message[7] str_time = message[8] is_send = message[4] @@ -261,7 +294,7 @@ def audio(self, doc, message): def emoji(self, doc, message): - origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}" + origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{sanitize_filename(self.contact.remark or self.contact.wxid)}" str_content = message[7] str_time = message[8] is_send = message[4] @@ -350,7 +383,7 @@ def system_msg(self, doc, message): ) def video(self, doc, message): - origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}" + origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{sanitize_filename(self.contact.remark or self.contact.wxid)}" type_ = message[2] str_content = message[7] str_time = message[8] @@ -362,7 +395,7 @@ def video(self, doc, message): video_path = hard_link_db.get_video(str_content, BytesExtra, thumb=False) image_path = hard_link_db.get_video(str_content, BytesExtra, thumb=True) if video_path is None and image_path is not None: - image_path = path.get_relative_path(image_path, base_path=f'/data/聊天记录/{self.contact.remark}/image') + image_path = path.get_relative_path(image_path, base_path=f'/data/聊天记录/{sanitize_filename(self.contact.remark or self.contact.wxid)}/image') image_path = image_path try: # todo 网络图片问题 @@ -406,9 +439,9 @@ def video(self, doc, message): ) def to_csv(self): - origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}" + origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{sanitize_filename(self.contact.remark or self.contact.wxid)}" os.makedirs(origin_docx_path, exist_ok=True) - filename = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}/{self.contact.remark}_utf8.csv" + filename = f"{os.path.abspath('.')}/data/聊天记录/{sanitize_filename(self.contact.remark or self.contact.wxid)}/{sanitize_filename(self.contact.remark or self.contact.wxid)}_utf8.csv" # columns = ["用户名", "消息内容", "发送时间", "发送状态", "消息类型", "isSend", "msgId"] columns = ['localId', 'TalkerId', 'Type', 'SubType', 'IsSender', 'CreateTime', 'Status', 'StrContent', @@ -428,10 +461,10 @@ def to_csv(self): self.okSignal.emit('ok') def to_html_(self): - origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}" + origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{sanitize_filename(self.contact.remark or self.contact.wxid)}" makedirs(origin_docx_path) messages = msg_db.get_messages(self.contact.wxid) - filename = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}/{self.contact.remark}.html" + filename = f"{os.path.abspath('.')}/data/聊天记录/{sanitize_filename(self.contact.remark or self.contact.wxid)}/{sanitize_filename(self.contact.remark or self.contact.wxid)}.html" f = open(filename, 'w', encoding='utf-8') f.write(html_head) MePC().avatar.save(os.path.join(origin_docx_path, 'myhead.png')) @@ -461,9 +494,9 @@ def to_html_(self): self.okSignal.emit(1) def to_txt(self): - origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}" + origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{sanitize_filename(self.contact.remark or self.contact.wxid)}" os.makedirs(origin_docx_path, exist_ok=True) - filename = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}/{self.contact.remark}.txt" + filename = f"{os.path.abspath('.')}/data/聊天记录/{sanitize_filename(self.contact.remark or self.contact.wxid)}/{sanitize_filename(self.contact.remark or self.contact.wxid)}.txt" messages = msg_db.get_messages(self.contact.wxid) total_steps = len(messages) with open(filename, mode='w', newline='', encoding='utf-8') as f: