!!!
123
你好呀
import csv
import html
import os
from re import findall
from PyQt5.QtCore import pyqtSignal, QThread
from PyQt5.QtWidgets import QFileDialog
from . import msg_db, micro_msg_db
from .package_msg import PackageMsg
from ..DataBase import hard_link_db
from ..person_pc import MePC
from ..util import path
import shutil
from ..util.emoji import get_emoji
os.makedirs('./data/聊天记录', exist_ok=True)
def makedirs(path):
os.makedirs(path, exist_ok=True)
os.makedirs(os.path.join(path, 'image'), exist_ok=True)
os.makedirs(os.path.join(path, 'emoji'), exist_ok=True)
os.makedirs(os.path.join(path, 'video'), exist_ok=True)
os.makedirs(os.path.join(path, 'voice'), exist_ok=True)
os.makedirs(os.path.join(path, 'file'), exist_ok=True)
def escape_js_and_html(input_str):
# 转义HTML特殊字符
html_escaped = html.escape(input_str, quote=False)
# 手动处理JavaScript转义字符
js_escaped = (
html_escaped
.replace("\\", "\\\\")
.replace("'", r"\'")
.replace('"', r'\"')
.replace("\n", r'\n')
.replace("\r", r'\r')
.replace("\t", r'\t')
)
return js_escaped
class Output(QThread):
"""
发送信息线程
"""
progressSignal = pyqtSignal(int)
rangeSignal = pyqtSignal(int)
okSignal = pyqtSignal(int)
i = 1
CSV = 0
DOCX = 1
HTML = 2
CSV_ALL = 3
CONTACT_CSV = 4
TXT = 5
def __init__(self, contact, type_=DOCX, message_types={}, parent=None):
super().__init__(parent)
self.Child0 = None
self.last_timestamp = 0
self.message_types = message_types
self.sec = 2 # 默认1000秒
self.contact = contact
self.ta_username = contact.wxid if contact else ''
self.msg_id = 0
self.output_type = type_
self.total_num = 0
self.num = 0
def progress(self, value):
self.progressSignal.emit(value)
def to_csv_all(self):
origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/"
os.makedirs(origin_docx_path, exist_ok=True)
filename = QFileDialog.getSaveFileName(None, "save file", os.path.join(os.getcwd(), 'messages.csv'),
"csv files (*.csv);;all files(*.*)")
if not filename[0]:
return
filename = filename[0]
# columns = ["用户名", "消息内容", "发送时间", "发送状态", "消息类型", "isSend", "msgId"]
columns = ['localId', 'TalkerId', 'Type', 'SubType',
'IsSender', 'CreateTime', 'Status', 'StrContent',
'StrTime', 'Remark', 'NickName', 'Sender']
packagemsg = PackageMsg()
messages = packagemsg.get_package_message_all()
# 写入CSV文件
with open(filename, mode='w', newline='', encoding='utf-8') as file:
writer = csv.writer(file)
writer.writerow(columns)
# 写入数据
writer.writerows(messages)
self.okSignal.emit(1)
def contact_to_csv(self):
filename = QFileDialog.getSaveFileName(None, "save file", os.path.join(os.getcwd(), 'contacts.csv'),
"csv files (*.csv);;all files(*.*)")
if not filename[0]:
return
filename = filename[0]
# columns = ["用户名", "消息内容", "发送时间", "发送状态", "消息类型", "isSend", "msgId"]
columns = ['UserName', 'Alias', 'Type', 'Remark', 'NickName', 'PYInitial', 'RemarkPYInitial', 'smallHeadImgUrl',
'bigHeadImgUrl']
contacts = micro_msg_db.get_contact()
# 写入CSV文件
with open(filename, mode='w', newline='', encoding='utf-8') as file:
writer = csv.writer(file)
writer.writerow(columns)
# 写入数据
writer.writerows(contacts)
self.okSignal.emit(1)
def run(self):
if self.output_type == self.DOCX:
return
elif self.output_type == self.CSV_ALL:
self.to_csv_all()
elif self.output_type == self.CONTACT_CSV:
self.contact_to_csv()
else:
self.Child = ChildThread(self.contact, type_=self.output_type, message_types=self.message_types)
self.Child.progressSignal.connect(self.progress)
self.Child.rangeSignal.connect(self.rangeSignal)
self.Child.okSignal.connect(self.okSignal)
self.Child.start()
def cancel(self):
self.requestInterruption()
class ChildThread(QThread):
"""
子线程,用于导出部分聊天记录
"""
progressSignal = pyqtSignal(int)
rangeSignal = pyqtSignal(int)
okSignal = pyqtSignal(int)
i = 1
CSV = 0
DOCX = 1
HTML = 2
def __init__(self, contact, type_=DOCX, message_types={}, parent=None):
super().__init__(parent)
self.contact = contact
self.message_types = message_types
self.last_timestamp = 0
self.sec = 2 # 默认1000秒
self.msg_id = 0
self.output_type = type_
def is_5_min(self, timestamp):
if abs(timestamp - self.last_timestamp) > 300:
self.last_timestamp = timestamp
return True
return False
def text(self, doc, message):
type_ = message[2]
str_content = message[7]
str_time = message[8]
is_send = message[4]
avatar = 'myhead.png' if is_send else 'tahead.png'
timestamp = message[5]
if self.output_type == Output.HTML:
str_content = escape_js_and_html(str_content)
if self.is_5_min(timestamp):
doc.write(
f'''{{ type:0, text: '{str_time}',is_send:0,avatar_path:''}},'''
)
emojiText = findall(r"(\[.+?\])", str_content)
for emoji_text in emojiText:
if emoji_text in emoji:
str_content = str_content.replace(emoji_text, emoji[emoji_text])
doc.write(
f'''{{ type:{type_}, text: '{str_content}',is_send:{is_send},avatar_path:'{avatar}'}},'''
)
elif self.output_type == Output.TXT:
name = '你' if is_send else self.contact.remark
doc.write(
f'''{str_time} {name}\n{str_content}\n\n'''
)
def image(self, doc, message):
type_ = message[2]
str_content = message[7]
str_time = message[8]
is_send = message[4]
avatar = 'myhead.png' if is_send else 'tahead.png'
timestamp = message[5]
BytesExtra = message[10]
if self.output_type == Output.HTML:
str_content = escape_js_and_html(str_content)
image_path = hard_link_db.get_image(str_content, BytesExtra, thumb=False)
image_thumb_path = hard_link_db.get_image(str_content, BytesExtra, thumb=True)
if image_path is None and image_thumb_path is not None:
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 = image_path.replace('\\', '/')
# print(f"tohtml:---{image_path}")
if self.is_5_min(timestamp):
doc.write(
f'''{{ type:0, text: '{str_time}',is_send:0,avatar_path:''}},'''
)
doc.write(
f'''{{ type:{type_}, text: '{image_path}',is_send:{is_send},avatar_path:'{avatar}'}},'''
)
elif self.output_type == Output.TXT:
name = '你' if is_send else self.contact.remark
doc.write(
f'''{str_time} {name}\n[图片]\n\n'''
)
def emoji(self, doc, message):
origin_docx_path = f"{os.path.abspath('.')}/data/聊天记录/{self.contact.remark}"
str_content = message[7]
str_time = message[8]
is_send = message[4]
avatar = 'myhead.png' if is_send else 'tahead.png'
timestamp = message[5]
if self.output_type == Output.HTML:
emoji_path = get_emoji(str_content, thumb=True, output_path=origin_docx_path + '/emoji')
emoji_path = './emoji/' + os.path.basename(emoji_path)
if self.is_5_min(timestamp):
doc.write(
f'''{{ type:0, text: '{str_time}',is_send:0,avatar_path:''}},'''
)
doc.write(
f'''{{ type:{3}, text: '{emoji_path}',is_send:{is_send},avatar_path:'{avatar}'}},'''
)
elif self.output_type == Output.TXT:
name = '你' if is_send else self.contact.remark
doc.write(
f'''{str_time} {name}\n[表情包]\n\n'''
)
def wx_file(self, doc, isSend, content, status):
return
def retract_message(self, doc, isSend, content, status):
return
def reply(self, doc, isSend, content, status):
return
def system_msg(self, doc, message):
str_content = message[7]
is_send = message[4]
str_time = message[8]
str_content = escape_js_and_html(str_content.lstrip('