cyonjan/PyWxDump
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

改善python3.8以下版本的容错

Browse Source
This commit is contained in:
xaoyaoo 2023-12-18 16:39:27 +08:00
parent a1d72a0595
commit 3d7a03d4e4
3 changed files with 44 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
pywxdump/analyzer/export_chat.py
View File

@ -20,9 +20,9 @@ import time
from functools import wraps
from .utils import get_md5, detach_databases, attach_databases, execute_sql
from .db_parsing import read_img_dat, decompress_CompressContent, read_audio, parse_xml_string
# from .db_parsing import read_img_dat, decompress_CompressContent, read_audio, parse_xml_string
from flask import Flask, request, render_template, g, Blueprint
# from flask import Flask, request, render_template, g, Blueprint
def get_contact_list(MicroMsg_db_path):
@ -35,14 +35,19 @@ def get_contact_list(MicroMsg_db_path):
# 连接 MicroMsg.db 数据库，并执行查询
db = sqlite3.connect(MicroMsg_db_path)
cursor = db.cursor()
cursor.execute(
"SELECT A.UserName, A.NickName, A.Remark,B.bigHeadImgUrl FROM Contact A,ContactHeadImgUrl B ORDER BY NickName ASC")
sql = ("SELECT A.UserName, A.NickName, A.Remark,A.Alias,A.Reserved6,B.bigHeadImgUrl "
"FROM Contact A,ContactHeadImgUrl B "
"where UserName==usrName "
"ORDER BY NickName ASC;")
cursor.execute(sql)
result = cursor.fetchall()
for row in result:
# 获取用户名、昵称、备注和聊天记录数量
username, nickname, remark, headImgUrl = row
users.append({"username": username, "nickname": nickname, "remark": remark, "headImgUrl": headImgUrl})
username, nickname, remark, Alias, describe, headImgUrl = row
users.append(
{"username": username, "nickname": nickname, "remark": remark, "account": Alias, "describe": describe,
"headImgUrl": headImgUrl})
cursor.close()
db.close()
return users
@ -264,61 +269,7 @@ def export(username, outpath, MSG_ALL_db_path, MicroMsg_db_path, MediaMSG_all_db
return export_html(user, outpath, MSG_ALL_db_path, MediaMSG_all_db_path, FileStorage_path)
app_show_chat = Blueprint('show_chat_main', __name__, template_folder='templates')
app_show_chat.debug = False
# 主页 - 显示用户列表
@app_show_chat.route('/')
def index():
g.USER_LIST = get_user_list(g.MSG_ALL_db_path, g.MicroMsg_db_path)
return render_template("index.html", users=g.USER_LIST)
# 获取聊天记录
@app_show_chat.route('/get_chat_data', methods=["GET", 'POST'])
def get_chat_data():
username = request.args.get("username", "")
user = list(filter(lambda x: x["username"] == username, g.USER_LIST))
if username and len(user) > 0:
user = user[0]
limit = int(request.args.get("limit", 100)) # 每页显示的条数
page = int(request.args.get("page", user.get("chat_count", limit) / limit)) # 当前页数
start_index = (page - 1) * limit
page_size = limit
data = load_chat_records(username, start_index, page_size, user, g.MSG_ALL_db_path, g.MediaMSG_all_db_path,
g.FileStorage_path)
return render_template("chat.html", msgs=data)
else:
return "error"
# 聊天记录导出为html
@app_show_chat.route('/export_chat_data', methods=["GET", 'POST'])
def get_export():
username = request.args.get("username", "")
user = list(filter(lambda x: x["username"] == username, g.USER_LIST))
if username and len(user) > 0:
user = user[0]
n = f"{user.get('username', '')}_{user.get('nickname', '')}_{user.get('remark', '')}"
outpath = os.path.join(os.getcwd(), "export" + os.sep + n)
if not os.path.exists(outpath):
os.makedirs(outpath)
ret = export_html(user, outpath, g.MSG_ALL_db_path, g.MediaMSG_all_db_path, g.FileStorage_path, page_size=200)
if ret[0]:
return ret[1]
else:
return ret[1]
else:
return "error"
if __name__ == '__main__':
pass
msg_all = r"D:\_code\py_code\test\a2023\b0821wxdb\merge_wfwx_db\kkWxMsg\MSG_all.db"
a = get_contact_list(msg_all)
print(a)

29
pywxdump/analyzer/utils.py
View File

@ -56,12 +56,29 @@ def execute_sql(connection, sql, params=None):
- sql：要执行的SQL语句
- params：SQL语句中的参数
"""
cursor = connection.cursor()
if params:
cursor.execute(sql, params)
else:
cursor.execute(sql)
return cursor.fetchall()
try:
# connection.text_factory = bytes
cursor = connection.cursor()
if params:
cursor.execute(sql, params)
else:
cursor.execute(sql)
return cursor.fetchall()
except Exception as e:
try:
connection.text_factory = bytes
cursor = connection.cursor()
if params:
cursor.execute(sql, params)
else:
cursor.execute(sql)
rdata = cursor.fetchall()
connection.text_factory = str
return rdata
except Exception as e:
print(f"**********\nSQL: {sql}\nparams: {params}\n{e}\n**********")
return None
if __name__ == '__main__':
pass

12
pywxdump/cli.py
View File

@ -6,7 +6,6 @@
# Date: 2023/10/14
# -------------------------------------------------------------------------------
import argparse
import importlib.metadata
import os
import sys
@ -353,8 +352,11 @@ class MainAll():
args.filestorage_path = FileStorage_path
MainShowChatRecords().run(args)
PYWXDUMP_VERSION = importlib.metadata.version('pywxdump')
try:
import importlib.metadata as importlib
except ImportError:
import importlib_metadata as importlib
PYWXDUMP_VERSION = importlib.version('pywxdump')
class CustomArgumentParser(argparse.ArgumentParser):
@ -362,7 +364,7 @@ class CustomArgumentParser(argparse.ArgumentParser):
# 首先显示软件简介
# 定义软件简介文本并进行格式化
line_len = 70
PYWXDUMP_VERSION = importlib.metadata.version('pywxdump')
PYWXDUMP_VERSION = importlib.version('pywxdump')
wxdump_line = '\n'.join([f'\033[36m{line:^{line_len}}\033[0m' for line in wxdump_ascii.split('\n') if line])
first_line = f'\033[36m{" PyWxDump v" + PYWXDUMP_VERSION + " ":=^{line_len}}\033[0m'
brief = 'PyWxDump功能：获取账号信息、解密数据库、查看聊天记录、导出聊天记录为html等'
@ -379,7 +381,7 @@ class CustomArgumentParser(argparse.ArgumentParser):
def console_run():
# 创建命令行参数解析器
parser = CustomArgumentParser(formatter_class=argparse.RawTextHelpFormatter)
PYWXDUMP_VERSION = importlib.metadata.version('pywxdump')
PYWXDUMP_VERSION = importlib.version('pywxdump')
parser.add_argument('-V', '--version', action='version', version=f"PyWxDump v{PYWXDUMP_VERSION}")
# 添加子命令解析器