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

增加导出解密数据库、csv方式

Browse Source
This commit is contained in:
xaoyaoo 2024-01-20 22:03:33 +08:00
parent 10767f2abe
commit 226baa5d13
5 changed files with 86 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pywxdump/__init__.py
View File

@ -5,7 +5,7 @@
# Author: xaoyaoo
# Date: 2023/10/14
# -------------------------------------------------------------------------------
from .wx_info import BiasAddr, read_info, get_wechat_db, encrypt, batch_decrypt, decrypt
from .wx_info import BiasAddr, read_info, get_wechat_db, encrypt, batch_decrypt, decrypt, get_core_db
from .wx_info import merge_copy_db, merge_msg_db, merge_media_msg_db, merge_db, decrypt_merge
from .analyzer.db_parsing import read_img_dat, read_emoji, decompress_CompressContent, read_audio_buf, read_audio, \
parse_xml_string, read_BytesExtra

79
pywxdump/api/api.py
View File

@ -12,7 +12,7 @@ import time
import shutil
from flask import Flask, request, render_template, g, Blueprint, send_file, make_response, session
from pywxdump import analyzer, read_img_dat, read_audio, get_wechat_db
from pywxdump import analyzer, read_img_dat, read_audio, get_wechat_db, get_core_db
from pywxdump.api.rjson import ReJson, RqJson
from pywxdump.api.utils import read_session, save_session
from pywxdump import read_info, VERSION_LIST, batch_decrypt, BiasAddr, merge_db, decrypt_merge
@ -282,34 +282,22 @@ def export():
:return:
"""
export_type = request.json.get("export_type")
start_time = request.json.get("start_time")
end_time = request.json.get("end_time")
start_time = request.json.get("start_time", 0)
end_time = request.json.get("end_time", 0)
chat_type = request.json.get("chat_type")
username = request.json.get("username")
# 可选参数
wx_path = request.json.get("wx_path", read_session(g.sf, "wx_path"))
key = request.json.get("key", read_session(g.sf, "key"))
if not export_type:
return ReJson(1002)
chat_type_tups = []
for t in chat_type:
tup = analyzer.get_name_typeid(t)
if tup:
chat_type_tups += tup
if not chat_type_tups:
if not export_type or not isinstance(export_type, str):
return ReJson(1002)
# 导出路径
outpath = os.path.join(g.tmp_path, "export")
outpath = os.path.join(g.tmp_path, "export", export_type)
if not os.path.exists(outpath):
os.makedirs(outpath)
if export_type == "endb":
outpath = os.path.join(outpath, export_type)
if not os.path.exists(outpath):
os.makedirs(outpath)
if export_type == "endb": # 导出加密数据库
# 获取微信文件夹路径
if not wx_path:
return ReJson(1002)
@ -317,36 +305,55 @@ def export():
return ReJson(1001, body=wx_path)
# 分割wx_path的文件名和父目录
msg_dir = os.path.dirname(wx_path)
my_wxid = os.path.basename(wx_path)
WxDbPath = get_wechat_db('all', msg_dir, wxid=my_wxid, is_logging=False) # 获取微信数据库路径
if isinstance(WxDbPath, str): # 如果返回的是字符串，则表示出错
return False, WxDbPath
wxdbpaths = [path for user_dir in WxDbPath.values() for paths in user_dir.values() for path in paths]
if len(wxdbpaths) == 0:
return False, "未获取到数据库路径"
wxdbpaths = [i for i in wxdbpaths if "MicroMsg" in i or "MediaMSG" in i or r"Multi\MSG" in i] # 过滤掉无需解密的数据库
code, wxdbpaths = get_core_db(wx_path)
if not code:
return ReJson(2001, body=wxdbpaths)
for wxdb in wxdbpaths:
# 复制wxdb->dboutpath
dboutpath = os.path.join(outpath, os.path.basename(wxdb))
shutil.copy(wxdb, dboutpath)
# 复制wxdb->outpath, os.path.basename(wxdb)
shutil.copy(wxdb, os.path.join(outpath, os.path.basename(wxdb)))
return ReJson(0, body=outpath)
elif export_type == "dedb":
pass
if isinstance(start_time, int) and isinstance(end_time, int):
msg_path = read_session(g.sf, "msg_path")
micro_path = read_session(g.sf, "micro_path")
media_path = read_session(g.sf, "media_path")
dbpaths = [msg_path, msg_path, micro_path]
dbpaths = list(set(dbpaths))
mergepath = merge_db(dbpaths, os.path.join(outpath, "merge.db"), start_time, end_time)
return ReJson(0, body=mergepath)
# if msg_path == media_path and msg_path == media_path:
# shutil.copy(msg_path, os.path.join(outpath, "merge.db"))
# return ReJson(0, body=msg_path)
# else:
# dbpaths = [msg_path, msg_path, micro_path]
# dbpaths = list(set(dbpaths))
# mergepath = merge_db(dbpaths, os.path.join(outpath, "merge.db"), start_time, end_time)
# return ReJson(0, body=mergepath)
else:
return ReJson(1002, body={"start_time": start_time, "end_time": end_time})
elif export_type == "csv":
# 导出聊天记录
outpath = os.path.join(outpath, "csv")
if not os.path.exists(outpath):
os.makedirs(outpath)
code, ret = analyzer.export_csv(username, outpath, read_session(g.sf, "msg_path"))
code, ret = analyzer.export_csv(username, os.path.join(outpath, username), read_session(g.sf, "msg_path"))
if code:
return ReJson(0, ret)
else:
return ReJson(2001, body=ret)
elif export_type == "json":
pass
elif export_type == "html":
pass
chat_type_tups = []
for ct in chat_type:
tup = analyzer.get_name_typeid(ct)
if tup:
chat_type_tups += tup
if not chat_type_tups:
return ReJson(1002)
elif export_type == "pdf":
pass
elif export_type == "docx":
@ -354,7 +361,7 @@ def export():
else:
return ReJson(1002)
return ReJson(0, "")
return ReJson(9999, "")
# 这部分为专业工具的api

2
pywxdump/wx_info/__init__.py
View File

@ -5,7 +5,7 @@
# Author: xaoyaoo
# Date: 2023/08/21
# -------------------------------------------------------------------------------
from .get_wx_info import read_info, get_wechat_db
from .get_wx_info import read_info, get_wechat_db, get_core_db
from .get_bias_addr import BiasAddr
from .decryption import batch_decrypt, encrypt, decrypt
from .merge_db import merge_msg_db, merge_copy_db, merge_media_msg_db, merge_db, decrypt_merge

35
pywxdump/wx_info/get_wx_info.py
View File

@ -293,6 +293,41 @@ def get_wechat_db(require_list: Union[List[str], str] = "all", msg_dir: str = No
return user_dirs
def get_core_db(wx_path: str, db_type: list = None) -> [str]:
"""
获取聊天消息核心数据库路径
:param wx_path: 微信文件夹路径 eg：C:\*****\WeChat Files\wxid*******
:param db_type: 数据库类型 eg: ["MSG", "MediaMSG", "MicroMsg"]，三个中选择一个或多个
:return: 返回数据库路径 eg:["",""]
"""
if not os.path.exists(wx_path):
return False, f"[-] 目录不存在: {wx_path}"
db_type_all = ["MSG", "MediaMSG", "MicroMsg"]
db_type = [dt for dt in db_type if dt in db_type_all]
if not db_type:
db_type = db_type_all
msg_dir = os.path.dirname(wx_path)
my_wxid = os.path.basename(wx_path)
WxDbPath = get_wechat_db('all', msg_dir, wxid=my_wxid, is_logging=False) # 获取微信数据库路径
if isinstance(WxDbPath, str): # 如果返回的是字符串，则表示出错
return False, WxDbPath
wxdbpaths = [path for user_dir in WxDbPath.values() for paths in user_dir.values() for path in paths]
if len(wxdbpaths) == 0:
return False, "未获取到数据库路径"
rdbpaths = []
for dbp in wxdbpaths:
if "MicroMsg" in db_type and "MicroMsg" in dbp:
rdbpaths.append(dbp)
if "MediaMSG" in db_type and "MediaMSG" in dbp:
rdbpaths.append(dbp)
if "MSG" in db_type and "Multi\MSG" in dbp:
rdbpaths.append(dbp)
return True, rdbpaths
if __name__ == '__main__':
from pywxdump import VERSION_LIST

18
pywxdump/wx_info/merge_db.py
View File

@ -279,7 +279,7 @@ def merge_db(db_paths, save_path="merge.db", CreateTime: int = 0, endCreateTime:
return save_path
def decrypt_merge(wx_path, key, outpath="", CreateTime: int = 0) -> (bool, str):
def decrypt_merge(wx_path, key, outpath="", CreateTime: int = 0, endCreateTime: int = 0) -> (bool, str):
"""
解密合并数据库 msg.db, microMsg.db, media.db
:param wx_path: 微信路径 eg: C:\*******\WeChat Files\wxid_*********
@ -287,28 +287,21 @@ def decrypt_merge(wx_path, key, outpath="", CreateTime: int = 0) -> (bool, str):
:return: (true,解密后的数据库路径) or (false,错误信息)
"""
from .decryption import batch_decrypt
from .get_wx_info import get_wechat_db
from .get_wx_info import get_core_db
outpath = outpath if outpath else "decrypt_merge_tmp"
merge_save_path = os.path.join(outpath, "merge_all.db")
decrypted_path = os.path.join(outpath, "decrypted")
if not wx_path or not key:
return True, "参数错误"
return False, "参数错误"
# 分割wx_path的文件名和父目录
msg_dir = os.path.dirname(wx_path)
my_wxid = os.path.basename(wx_path)
# 解密
WxDbPath = get_wechat_db('all', msg_dir, wxid=my_wxid, is_logging=False) # 获取微信数据库路径
if isinstance(WxDbPath, str): # 如果返回的是字符串，则表示出错
return False, WxDbPath
wxdbpaths = [path for user_dir in WxDbPath.values() for paths in user_dir.values() for path in paths]
if len(wxdbpaths) == 0:
return False, "未获取到数据库路径"
wxdbpaths = [i for i in wxdbpaths if "MicroMsg" in i or "MediaMSG" in i or r"Multi\MSG" in i] # 过滤掉无需解密的数据库
code, wxdbpaths = get_core_db(wx_path)
# 判断out_path是否为空目录
if os.path.exists(decrypted_path) and os.listdir(decrypted_path):
@ -333,6 +326,7 @@ def decrypt_merge(wx_path, key, outpath="", CreateTime: int = 0) -> (bool, str):
parpare_merge_db_path = [i for i in out_dbs if "de_MicroMsg" in i or "de_MediaMSG" in i or "de_MSG" in i]
merge_save_path = merge_db(parpare_merge_db_path, merge_save_path, CreateTime=CreateTime)
merge_save_path = merge_db(parpare_merge_db_path, merge_save_path, CreateTime=CreateTime,
endCreateTime=endCreateTime)
return True, merge_save_path