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

删除部分命令行不可用命令

Browse Source
This commit is contained in:
xaoyaoo 2024-04-20 18:57:47 +08:00
parent fa66aa1c81
commit 092db3d76b
1 changed files with 1 additions and 170 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

171
pywxdump/cli.py
View File

@ -287,175 +287,6 @@ class MainShowChatRecords(BaseSubMainClass):
start_falsk(merge_path=merge_path, wx_path=args.wx_path, key="", my_wxid=args.my_wxid, online=online)
class MainExportChatRecords(BaseSubMainClass):
mode = "export"
parser_kwargs = {"help": "聊天记录导出为html"}
def init_parses(self, parser):
# 添加 'decrypt' 子命令解析器
parser.add_argument("-u", "--username", type=str, help="微信账号(聊天对象账号)", required=True, metavar="")
parser.add_argument("-o", "--outpath", type=str, help="导出路径", required=True, metavar="")
parser.add_argument("-msg", "--msg_path", type=str, help="解密后的 MSG.db 的路径", required=True,
metavar="")
parser.add_argument("-micro", "--micro_path", type=str, help="解密后的 MicroMsg.db 的路径", required=True,
metavar="")
parser.add_argument("-media", "--media_path", type=str, help="解密后的 MediaMSG.db 的路径", required=True,
metavar="")
parser.add_argument("-fs", "--filestorage_path", type=str,
help="(可选)文件夹FileStorage的路径（用于显示图片）", required=False, metavar="")
parser.add_argument("-t", "--type", type=str, help="导出类型(可选:html,csv)", required=False,
default="html", metavar="")
return parser
def run(self, args):
print(f"[*] PyWxDump v{pywxdump.__version__}")
# 从命令行参数获取值
t = args.type
if t == "csv":
try:
code, ret = export_csv(args.username, args.outpath, args.msg_path, page_size=10000000)
if not code:
print(ret)
return
print(f"[+] {ret}")
return
except Exception as e:
print(e)
print("[-] 导出失败")
return
elif t == "html":
try:
from flask import Flask, request, jsonify, render_template, g
import logging
except Exception as e:
print(e)
print("[-] 请安装flask( pip install flask)")
return
if not os.path.exists(args.msg_path) or not os.path.exists(args.micro_path) or not os.path.exists(
args.media_path):
print(os.path.exists(args.msg_path), os.path.exists(args.micro_path), os.path.exists(args.media_path))
print("[-] 输入数据库路径不存在")
return
if not os.path.exists(args.outpath):
os.makedirs(args.outpath)
print(f"[+] 创建输出文件夹：{args.outpath}")
export(args.username, args.outpath, args.msg_path, args.micro_path, args.media_path, args.filestorage_path)
print(f"[+] 导出成功{args.outpath}")
else:
print("[-] 未知的导出类型")
return
class MainAll(BaseSubMainClass):
mode = "all"
parser_kwargs = {"help": "获取微信信息，解密微信数据库，查看聊天记录"}
def init_parses(self, parser):
# 添加 'all' 子命令解析器
parser.add_argument("-s", '--save_path', metavar="", type=str, help="(可选)wx_info保存路径【json文件】")
parser.add_argument("--online", action='store_true', help="(可选)是否在线查看(局域网查看)", default=False)
return parser
def run(self, args):
print(f"[*] PyWxDump v{pywxdump.__version__}")
# 获取微信信息
save_path = args.save_path
online = args.online
WxInfo = read_info(VERSION_LIST, True, save_path)
if isinstance(WxInfo, str): # 如果返回的是字符串，则表示出错
return
for user in WxInfo:
key = user.get("key", "")
if not key:
print("[-] 未获取到密钥")
return
wxid = user.get("wxid", None)
filePath = user.get("filePath", None)
WxDbPath = get_wechat_db('all', None, wxid=wxid, is_logging=True) # 获取微信数据库路径
if isinstance(WxDbPath, str): # 如果返回的是字符串，则表示出错
print(WxDbPath)
return
wxdbpaths = [path for user_dir in WxDbPath.values() for paths in user_dir.values() for path in paths]
if len(wxdbpaths) == 0:
print("[-] 未获取到数据库路径")
return
wxdbpaths = [i for i in wxdbpaths if "Backup.db" not in i and "xInfo.db" not in i] # 过滤掉无需解密的数据库
wxdbpaths = [i for i in wxdbpaths if "MicroMsg" in i or "MediaMSG" in i or r"Multi\MSG" in i] # 过滤掉无需解密的数据库
wxdblen = len(wxdbpaths)
print(f"[+] 共发现 {wxdblen} 个微信数据库")
print("=" * 32)
decrypted_path = os.path.join(os.getcwd(), "decrypted")
print(f"[*] 解密后文件夹：{decrypted_path} ")
print(f"[*] 解密中...（用时较久，耐心等待）")
# 判断out_path是否为空目录
if os.path.exists(decrypted_path) and os.listdir(decrypted_path):
isdel = input(f"[*] 输出文件夹不为空({decrypted_path})\n 是否删除?(y/n):")
if isdel.lower() == 'y' or isdel.lower() == 'yes':
for root, dirs, files in os.walk(decrypted_path, topdown=False):
for name in files:
os.remove(os.path.join(root, name))
for name in dirs:
os.rmdir(os.path.join(root, name))
out_path = os.path.join(decrypted_path, wxid) if wxid else decrypted_path
if not os.path.exists(out_path):
os.makedirs(out_path)
# 调用 decrypt 函数，并传入参数 # 解密
code, ret = batch_decrypt(key, wxdbpaths, out_path, False)
if not code:
print(ret)
return
print("[+] 解密完成")
print("-" * 32)
errors = []
out_dbs = []
for code1, ret1 in ret:
if code1 == False:
errors.append(ret1)
else:
print(
f'[+] success "{os.path.relpath(ret1[0], os.path.commonprefix(wxdbpaths))}" -> "{os.path.relpath(ret1[1], os.getcwd())}"')
out_dbs.append(ret1[1])
if len(errors) > 0:
print("-" * 32)
print(
"[-] " + f"警告：共 {len(errors)} 个文件未解密(可能原因:非当前登录用户数据库;非加密数据库),详见{out_path}下‘未解密.txt’;")
# print("; ".join([f'"{wxdbpaths[i]}"' for i in errors]))
with open(os.path.join(out_path, "未解密.txt"), "w", encoding="utf-8") as f:
f.write("\n".join([f'{i}' for i in errors]))
print("=" * 32)
if len(out_dbs) <= 0:
print("[-] 未获取到解密后的数据库路径")
return
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]
# 合并所有的数据库
print(f"[*] 合并数据库中...（用时较久，耐心等待）")
merge_save_path = merge_db(parpare_merge_db_path, os.path.join(out_path, "merge_all.db"))
time.sleep(1)
print(f"[+] 合并完成：{merge_save_path}")
print("=" * 32)
# # 查看聊天记录
args.merge_path = merge_save_path
args.msg_path = merge_save_path
args.micro_path = merge_save_path
args.media_path = merge_save_path
args.wx_path = filePath
args.my_wxid = wxid
args.online = online
MainShowChatRecords().run(args)
class MainUi(BaseSubMainClass):
mode = "ui"
@ -482,7 +313,7 @@ class MainUi(BaseSubMainClass):
class MainApi(BaseSubMainClass):
mode = "api"
parser_kwargs = {"help": "启动api"}
parser_kwargs = {"help": "启动api，不打开浏览器"}
def init_parses(self, parser):
# 添加 'api' 子命令解析器