ZSCGR/WeChatDataAnalysis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(detection): 改进微信账户检测逻辑

Browse Source
This commit is contained in:
2977094657
2025-07-17 22:44:56 +08:00
parent 4896e8a2ec
commit 8b3dec2b9d
5 changed files with 610 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wechat_decrypt_tool/wechat_detection.py
+247 -61
View File
@@ -201,10 +201,10 @@ def auto_detect_wechat_data_dirs():
# 策略1：常见驱动器扫描微信相关目录
common_wechat_patterns = [
"WeChat Files", "wechat_files", "xwechat_files", "wechatMSG",
"WeChat Files", "wechat_files", "xwechat_files", "wechatMSG",
"WeChat", "微信", "Weixin", "wechat"
]
# 扫描常见驱动器
drives = ['C:', 'D:', 'E:', 'F:']
for drive in drives:
@@ -326,30 +326,181 @@ def get_wx_dir_by_reg(wxid="all"):
return wx_dir if os.path.exists(wx_dir) else None
def detect_wechat_accounts_from_backup(backup_base_path: str = None) -> List[Dict[str, Any]]:
"""
从指定的备份路径检测微信账号
Args:
backup_base_path: 微信文件基础路径，如果为None则自动检测
Returns:
账号信息列表，每个账号包含：
- account_name: 账号名
- backup_dir: 备份目录路径
- data_dir: 实际数据目录路径
- databases: 数据库文件列表
"""
accounts = []
# 如果没有指定路径，尝试自动检测
if backup_base_path is None:
# 使用自动检测找到包含Backup的路径
detected_dirs = auto_detect_wechat_data_dirs()
for detected_dir in detected_dirs:
# 首先检查直接的Backup目录
backup_test_dir = os.path.join(detected_dir, "Backup")
if os.path.exists(backup_test_dir):
backup_base_path = detected_dir
break
# 然后检查子目录中的Backup目录（如xwechat_files/Backup）
try:
for subdir in os.listdir(detected_dir):
subdir_path = os.path.join(detected_dir, subdir)
if os.path.isdir(subdir_path):
backup_test_dir = os.path.join(subdir_path, "Backup")
if os.path.exists(backup_test_dir):
backup_base_path = subdir_path
break
if backup_base_path:
break
except (PermissionError, OSError):
continue
# 如果还是没找到，返回空列表
if backup_base_path is None:
return accounts
# 检查备份目录
backup_dir = os.path.join(backup_base_path, "Backup")
if not os.path.exists(backup_dir):
return accounts
try:
# 遍历备份目录下的所有子文件夹（每个代表一个账号）
for item in os.listdir(backup_dir):
account_backup_path = os.path.join(backup_dir, item)
if not os.path.isdir(account_backup_path):
continue
account_name = item
# 在上级目录中查找对应的实际数据文件夹
# 命名规则：{账号名}_{随机字符}
data_dir = None
try:
for data_item in os.listdir(backup_base_path):
data_item_path = os.path.join(backup_base_path, data_item)
if (os.path.isdir(data_item_path) and
data_item.startswith(f"{account_name}_") and
data_item != "Backup"):
data_dir = data_item_path
break
except (PermissionError, OSError):
continue
# 收集该账号的数据库文件
databases = []
if data_dir and os.path.exists(data_dir):
databases = collect_account_databases(data_dir, account_name)
account_info = {
"account_name": account_name,
"backup_dir": account_backup_path,
"data_dir": data_dir,
"databases": databases,
"database_count": len(databases)
}
accounts.append(account_info)
except (PermissionError, OSError) as e:
# 如果无法访问备份目录，返回空列表
pass
return accounts
def collect_account_databases(data_dir: str, account_name: str) -> List[Dict[str, Any]]:
"""
收集指定账号数据目录下的所有数据库文件
Args:
data_dir: 账号数据目录
account_name: 账号名
Returns:
数据库文件信息列表
"""
databases = []
if not os.path.exists(data_dir):
return databases
try:
# 递归查找所有.db文件
for root, dirs, files in os.walk(data_dir):
for file_name in files:
if not file_name.endswith('.db'):
continue
# 排除不需要解密的数据库
if file_name in ["key_info.db"]:
continue
db_path = os.path.join(root, file_name)
# 确定数据库类型
db_type = re.sub(r'\d*\.db$', '', file_name)
try:
file_size = os.path.getsize(db_path)
except OSError:
file_size = 0
db_info = {
"path": db_path,
"name": file_name,
"type": db_type,
"size": file_size,
"relative_path": os.path.relpath(db_path, data_dir)
}
databases.append(db_info)
except (PermissionError, OSError):
pass
return databases
def detect_wechat_installation() -> Dict[str, Any]:
"""
检测微信安装情况 - 完全按照PyWxDump的逻辑实现
检测微信安装情况 - 改进的多账户检测逻辑
"""
result = {
"wechat_version": None,
"wechat_install_path": None,
"wechat_exe_path": None,
"is_running": False,
"accounts": [],
"total_accounts": 0,
"total_databases": 0,
"detection_errors": [],
"detection_methods": [],
# 保持向后兼容性的字段
"wechat_data_dirs": [],
"message_dirs": [],
"databases": [],
"version_detected": None,
"is_running": False,
"user_accounts": [],
"detection_errors": [],
"detection_methods": []
"user_accounts": []
}
# 进程检测 - 只检测Weixin.exe（按照用户要求）
# 1. 进程检测 - 检测微信是否运行
result["detection_methods"].append("进程检测")
process_list = get_process_list()
for pid, process_name in process_list:
# 只检查Weixin.exe进程
# 检查Weixin.exe进程
if process_name.lower() == 'weixin.exe':
try:
exe_path = get_process_exe_path(pid)
@@ -377,63 +528,98 @@ def detect_wechat_installation() -> Dict[str, Any]:
if not result["is_running"]:
result["detection_methods"].append("未检测到微信进程")
# 2. 使用自动检测逻辑获取微信目录和数据库
result["detection_methods"].append("目录自动检测")
# 2. 使用新的账号检测逻辑
result["detection_methods"].append("多账户检测")
try:
wx_dir = get_wx_dir_by_reg()
if wx_dir and os.path.exists(wx_dir):
result["wechat_data_dirs"].append(wx_dir)
result["detection_methods"].append(f"通过自动检测找到微信目录: {wx_dir}")
# 检测指定路径下的微信账号
accounts = detect_wechat_accounts_from_backup()
result["accounts"] = accounts
result["total_accounts"] = len(accounts)
# 使用PyWxDump的get_wx_db函数获取数据库信息
db_list = get_wx_db(msg_dir=wx_dir) # 移除db_types限制，获取所有.db文件
# 统计总数据库数量
total_db_count = sum(account["database_count"] for account in accounts)
result["total_databases"] = total_db_count
# 统计用户账户和消息目录
user_accounts_set = set()
message_dirs_set = set()
if accounts:
result["detection_methods"].append(f"在指定路径检测到 {len(accounts)} 个微信账户")
result["detection_methods"].append(f"总计 {total_db_count} 个数据库文件")
for db_info in db_list:
wxid = db_info["wxid"]
wxid_dir = db_info["wxid_dir"]
db_path = db_info["db_path"]
db_type = db_info["db_type"]
# 添加用户账户
user_accounts_set.add(wxid)
message_dirs_set.add(wxid_dir)
# 添加数据库信息
if os.path.exists(db_path):
result["databases"].append({
"path": db_path,
"name": os.path.basename(db_path),
"type": db_type,
"size": os.path.getsize(db_path),
"user": wxid,
"user_dir": wxid_dir
})
# 转换为列表
result["user_accounts"] = list(user_accounts_set)
result["message_dirs"] = list(message_dirs_set)
result["detection_methods"].append(f"检测到 {len(result['user_accounts'])} 个用户账户")
result["detection_methods"].append(f"检测到 {len(result['databases'])} 个数据库文件")
# 按数据库类型统计
db_type_count = {}
for db in result["databases"]:
db_type = db["type"]
db_type_count[db_type] = db_type_count.get(db_type, 0) + 1
if db_type_count:
type_summary = ", ".join([f"{k}({v})" for k, v in db_type_count.items()])
result["detection_methods"].append(f"数据库类型分布: {type_summary}")
# 为每个账户添加详细信息
for account in accounts:
account_name = account["account_name"]
db_count = account["database_count"]
data_dir_status = "已找到" if account["data_dir"] else "未找到"
result["detection_methods"].append(f"账户 {account_name}: {db_count} 个数据库, 数据目录{data_dir_status}")
else:
result["detection_methods"].append("自动检测未找到微信目录")
except Exception as e:
result["detection_errors"].append(f"目录检测失败: {str(e)}")
result["detection_methods"].append("未在指定路径检测到微信账户")
# 填充向后兼容性字段
for account in accounts:
if account["data_dir"]:
result["wechat_data_dirs"].append(account["data_dir"])
result["message_dirs"].append(account["data_dir"])
result["user_accounts"].append(account["account_name"])
# 添加数据库到兼容性列表
for db in account["databases"]:
result["databases"].append({
"path": db["path"],
"name": db["name"],
"type": db["type"],
"size": db["size"],
"user": account["account_name"],
"user_dir": account["data_dir"]
})
except Exception as e:
result["detection_errors"].append(f"账户检测失败: {str(e)}")
# 3. 如果新检测方法没有找到账户，尝试旧的检测方法作为备用
if not result["accounts"]:
result["detection_methods"].append("备用检测方法")
try:
wx_dir = get_wx_dir_by_reg()
if wx_dir and os.path.exists(wx_dir):
result["wechat_data_dirs"].append(wx_dir)
result["detection_methods"].append(f"通过备用方法找到微信目录: {wx_dir}")
# 使用旧的检测逻辑
db_list = get_wx_db(msg_dir=wx_dir)
# 按账户组织数据库
account_db_map = {}
for db_info in db_list:
wxid = db_info["wxid"]
if wxid not in account_db_map:
account_db_map[wxid] = {
"account_name": wxid,
"backup_dir": None,
"data_dir": db_info["wxid_dir"],
"databases": [],
"database_count": 0
}
if os.path.exists(db_info["db_path"]):
db_entry = {
"path": db_info["db_path"],
"name": os.path.basename(db_info["db_path"]),
"type": db_info["db_type"],
"size": os.path.getsize(db_info["db_path"]),
"relative_path": os.path.relpath(db_info["db_path"], db_info["wxid_dir"])
}
account_db_map[wxid]["databases"].append(db_entry)
account_db_map[wxid]["database_count"] += 1
result["accounts"] = list(account_db_map.values())
result["total_accounts"] = len(result["accounts"])
result["total_databases"] = sum(account["database_count"] for account in result["accounts"])
result["detection_methods"].append(f"备用方法检测到 {result['total_accounts']} 个账户")
result["detection_methods"].append(f"总计 {result['total_databases']} 个数据库文件")
else:
result["detection_methods"].append("备用检测方法未找到微信目录")
except Exception as e:
result["detection_errors"].append(f"备用检测失败: {str(e)}")
return result