308 lines
11 KiB
Python
308 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
企业微信媒体文件导出模块
|
||
从消息 content(protobuf) 中提取文件引用(UUID/CDN URL/文件名),
|
||
在本地 Cache 目录中查找匹配文件并复制到输出目录.
|
||
|
||
文件特性:
|
||
图片: PNG/JPG 明文, Cache/Image/ 按 UUID 命名
|
||
语音: SILK_V3 明文, Cache/Voice/ 按 UUID 命名
|
||
视频: MP4 明文, Cache/Video/ 混合命名
|
||
文件: 原始格式, Cache/File/ 用原始文件名
|
||
|
||
关联方式:
|
||
1. UUID 匹配 (图片/语音, 命中率低但有的话就关联)
|
||
2. 文件名匹配 (文件/视频, 按原始文件名搜索)
|
||
3. CDN URL 记录 (本地无缓存时, 记录 URL 供后续下载)
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
import shutil
|
||
import sqlite3
|
||
import sys
|
||
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, BASE_DIR)
|
||
|
||
# ---- 正则 ----
|
||
|
||
# UUID (在 raw bytes 中直接搜索 ASCII)
|
||
UUID_RE = re.compile(rb"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", re.I)
|
||
|
||
# CDN URL (wework/mmbiz qpic 域名 + 常见媒体扩展名)
|
||
URL_RE = re.compile(
|
||
rb"https?://[^\x00-\x1f\x20\"'<>]+", re.I)
|
||
|
||
# 文件名 (在去掉 URL 的文本中搜索)
|
||
FN_RE = re.compile(
|
||
r"[\w\u4e00-\u9fff\u3000-\u303f\s\-+()【】\[\]（）《》、,，.]{2,80}"
|
||
r"\.(?:jpg|jpeg|png|gif|bmp|webp|mp4|mov|avi|mkv|pdf|docx?|xlsx?|pptx?|txt|zip|rar|7z|silk|amr|csv)",
|
||
re.I)
|
||
|
||
# Cache 子目录映射
|
||
CACHE_TYPE_MAP = {
|
||
"Image": "图片",
|
||
"Voice": "语音",
|
||
"Video": "视频",
|
||
"File": "文件",
|
||
}
|
||
|
||
|
||
def extract_media_refs(content):
|
||
"""从消息 content(bytes) 中提取媒体引用
|
||
|
||
返回:
|
||
{"uuids": [str], "urls": [str], "filenames": [str]}
|
||
"""
|
||
result = {"uuids": [], "urls": [], "filenames": []}
|
||
if not content or not isinstance(content, bytes):
|
||
return result
|
||
|
||
# 1. ASCII UUID (直接在 bytes 中搜索)
|
||
for m in UUID_RE.finditer(content):
|
||
uid = m.group(0).decode("ascii").lower()
|
||
if uid not in result["uuids"]:
|
||
result["uuids"].append(uid)
|
||
|
||
# 2. 解码为文本
|
||
text = content.decode("utf-8", errors="ignore")
|
||
|
||
# 3. CDN URL (先提取, 然后从文本中移除以便提取文件名)
|
||
for m in re.finditer(r"https?://[^\s\x00-\x1f\"'<>]+", text):
|
||
url = m.group(0).rstrip(".,;)")
|
||
# 只保留含媒体特征的 URL
|
||
if any(kw in url.lower() for kw in (
|
||
"qpic.cn", "wwcdn", "wework", "mmbiz",
|
||
".jpg", ".png", ".jpeg", ".gif", ".mp4", ".pdf",
|
||
".docx", ".xlsx", ".silk", ".amr"
|
||
)):
|
||
if url not in result["urls"]:
|
||
result["urls"].append(url)
|
||
|
||
# 4. 文件名 (从去掉 URL 的文本中搜索)
|
||
text_no_url = re.sub(r"https?://\S+", "", text)
|
||
for m in FN_RE.finditer(text_no_url):
|
||
fn = m.group(0).strip()
|
||
# 去掉路径前缀
|
||
fn = re.sub(r"^[./\\]+", "", fn).strip()
|
||
# 过滤太短或纯数字的
|
||
if len(fn) > 4 and fn not in result["filenames"]:
|
||
# 确保有扩展名
|
||
if "." in fn:
|
||
result["filenames"].append(fn)
|
||
|
||
return result
|
||
|
||
|
||
def build_cache_index(account_dir):
|
||
"""扫描账号 Cache 目录, 建立 UUID→文件 和 文件名→文件 索引
|
||
|
||
返回:
|
||
{"by_uuid": {uuid: (filepath, type)}, "by_name": {filename: (filepath, type)}}
|
||
"""
|
||
index = {"by_uuid": {}, "by_name": {}}
|
||
cache_dir = os.path.join(account_dir, "Cache")
|
||
if not os.path.isdir(cache_dir):
|
||
return index
|
||
|
||
for subdir in ("Image", "Voice", "Video", "File"):
|
||
d = os.path.join(cache_dir, subdir)
|
||
if not os.path.isdir(d):
|
||
continue
|
||
for root, dirs, files in os.walk(d):
|
||
for fn in files:
|
||
fp = os.path.join(root, fn)
|
||
fn_lower = fn.lower()
|
||
# 按 UUID 索引
|
||
m = re.match(
|
||
r"([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})",
|
||
fn, re.I)
|
||
if m:
|
||
index["by_uuid"][m.group(1).lower()] = (fp, subdir)
|
||
# 按文件名索引 (原始名 + 去扩展名)
|
||
index["by_name"][fn_lower] = (fp, subdir)
|
||
base = os.path.splitext(fn_lower)[0]
|
||
if base and base not in index["by_name"]:
|
||
index["by_name"][base] = (fp, subdir)
|
||
|
||
return index
|
||
|
||
|
||
def match_media(refs, cache_index):
|
||
"""根据 media refs 在 cache 索引中查找匹配文件
|
||
|
||
返回: (filepath, media_type, match_method) 或 (None, None, method)
|
||
match_method: "uuid" / "filename" / "fuzzy" / "url_only" / None
|
||
"""
|
||
# 1. UUID 精确匹配
|
||
for uid in refs["uuids"]:
|
||
if uid in cache_index["by_uuid"]:
|
||
fp, mtype = cache_index["by_uuid"][uid]
|
||
return (fp, mtype, "uuid")
|
||
|
||
# 2. 文件名精确匹配
|
||
for fn in refs["filenames"]:
|
||
fn_lower = fn.lower()
|
||
if fn_lower in cache_index["by_name"]:
|
||
fp, mtype = cache_index["by_name"][fn_lower]
|
||
return (fp, mtype, "filename")
|
||
# 去路径的 basename
|
||
bn = os.path.basename(fn_lower)
|
||
if bn in cache_index["by_name"]:
|
||
fp, mtype = cache_index["by_name"][bn]
|
||
return (fp, mtype, "filename")
|
||
|
||
# 3. 文件名模糊匹配 (包含关系, 要求长度>4 避免误匹配)
|
||
for fn in refs["filenames"]:
|
||
fn_lower = fn.lower()
|
||
if len(fn_lower) < 5:
|
||
continue
|
||
for cache_fn, (fp, mtype) in cache_index["by_name"].items():
|
||
if fn_lower in cache_fn or cache_fn in fn_lower:
|
||
return (fp, mtype, "fuzzy")
|
||
|
||
# 4. 只有 URL, 无本地文件
|
||
if refs["urls"]:
|
||
return (None, None, "url_only")
|
||
|
||
return (None, None, None)
|
||
|
||
|
||
def export_media(decrypted_dbs, out_dir, date_from=None, date_to=None, log=print,
|
||
personal_only=True, blocked_conv_ids=None):
|
||
"""导出媒体文件, 返回 {server_id: media_info} 和统计
|
||
|
||
参数:
|
||
decrypted_dbs: [(db_path, db_name, account), ...]
|
||
out_dir: 输出根目录
|
||
date_from / date_to: "YYYY-MM-DD" 日期范围 (None=不过滤)
|
||
log: 日志函数
|
||
personal_only: True=只导出单聊会话 (M:/S:) 的媒体 (默认); False=全部会话
|
||
blocked_conv_ids: 会话ID黑名单集合 (按会话名称过滤出的官方/系统账号), 命中即跳过
|
||
"""
|
||
# 延迟导入, 避免与 wxwork_export_final 循环引用
|
||
from wxwork_export_final import DEFAULT_DB_BASE
|
||
|
||
blocked_conv_ids = set(blocked_conv_ids or ())
|
||
media_dir = os.path.join(out_dir, "媒体文件")
|
||
for sub in ("图片", "语音", "视频", "文件"):
|
||
os.makedirs(os.path.join(media_dir, sub), exist_ok=True)
|
||
|
||
result = {} # server_id -> media_info
|
||
stats = {
|
||
"total_media_msg": 0,
|
||
"matched_uuid": 0,
|
||
"matched_filename": 0,
|
||
"matched_fuzzy": 0,
|
||
"copied": 0,
|
||
"url_only": 0,
|
||
}
|
||
|
||
msg_dbs = [(d, a) for d, n, a in decrypted_dbs if n == "message.db"]
|
||
if not msg_dbs:
|
||
log("[-] 未找到 message.db, 跳过媒体导出")
|
||
return result, stats
|
||
|
||
for db_path, account in msg_dbs:
|
||
account_dir = os.path.join(DEFAULT_DB_BASE, account)
|
||
cache_index = build_cache_index(account_dir)
|
||
n_uuid = len(cache_index["by_uuid"])
|
||
n_name = len(cache_index["by_name"])
|
||
log(f" 账号 {account}: Cache 索引 {n_uuid} UUID + {n_name} 文件名")
|
||
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
cols = [r[1] for r in conn.execute(
|
||
"PRAGMA table_info(message_table)").fetchall()]
|
||
ct_idx = cols.index("content_type")
|
||
content_idx = cols.index("content")
|
||
sid_idx = (cols.index("server_id") if "server_id" in cols
|
||
else cols.index("message_id"))
|
||
conv_idx = (cols.index("conversation_id")
|
||
if "conversation_id" in cols else None)
|
||
time_col = "send_time" if "send_time" in cols else None
|
||
|
||
# 日期过滤
|
||
where = ""
|
||
params = ()
|
||
if date_from and time_col:
|
||
# send_time 是毫秒时间戳, 转为整数比较
|
||
from datetime import datetime as _dt
|
||
ts_from = int(_dt.strptime(date_from, "%Y-%m-%d").timestamp())
|
||
ts_to = int(_dt.strptime(date_to, "%Y-%m-%d").timestamp()) + 86399
|
||
where = f" WHERE {time_col} >= ? AND {time_col} <= ?"
|
||
params = (ts_from, ts_to)
|
||
|
||
# 只导出单聊会话 (M:xxx / S:xxx) 的媒体
|
||
if personal_only:
|
||
conv_filter = "conversation_id LIKE 'M:%' OR conversation_id LIKE 'S:%'"
|
||
where = (where + f" AND ({conv_filter})") if where \
|
||
else f" WHERE {conv_filter}"
|
||
|
||
rows = conn.execute(
|
||
f"SELECT * FROM message_table{where} ORDER BY {time_col or 'rowid'}",
|
||
params).fetchall()
|
||
|
||
for row in rows:
|
||
ct = row[ct_idx]
|
||
content = row[content_idx]
|
||
sid = row[sid_idx]
|
||
|
||
# 按会话名称过滤出的黑名单会话ID, 跳过
|
||
if blocked_conv_ids and conv_idx is not None \
|
||
and str(row[conv_idx]) in blocked_conv_ids:
|
||
continue
|
||
|
||
refs = extract_media_refs(content)
|
||
if not (refs["uuids"] or refs["urls"] or refs["filenames"]):
|
||
continue
|
||
|
||
stats["total_media_msg"] += 1
|
||
|
||
fp, mtype, method = match_media(refs, cache_index)
|
||
|
||
if fp and os.path.exists(fp):
|
||
if method == "uuid":
|
||
stats["matched_uuid"] += 1
|
||
elif method == "filename":
|
||
stats["matched_filename"] += 1
|
||
elif method == "fuzzy":
|
||
stats["matched_fuzzy"] += 1
|
||
|
||
# 复制文件
|
||
dest_sub = CACHE_TYPE_MAP.get(mtype, "文件")
|
||
dest_name = f"{sid}_{os.path.basename(fp)}"
|
||
dest_path = os.path.join(media_dir, dest_sub, dest_name)
|
||
|
||
if not os.path.exists(dest_path):
|
||
try:
|
||
shutil.copy2(fp, dest_path)
|
||
stats["copied"] += 1
|
||
except Exception as e:
|
||
log(f" [!] 复制失败 {fp}: {e}")
|
||
dest_path = fp # 退回原路径
|
||
|
||
result[str(sid)] = {
|
||
"path": os.path.relpath(dest_path, out_dir),
|
||
"type": dest_sub,
|
||
"url": refs["urls"][0] if refs["urls"] else "",
|
||
"method": method,
|
||
"filename": os.path.basename(fp),
|
||
}
|
||
elif method == "url_only":
|
||
stats["url_only"] += 1
|
||
result[str(sid)] = {
|
||
"path": "",
|
||
"type": "",
|
||
"url": refs["urls"][0] if refs["urls"] else "",
|
||
"method": "url_only",
|
||
"filename": "",
|
||
}
|
||
except sqlite3.Error as e:
|
||
log(f" [警告] {db_path}: {e}")
|
||
finally:
|
||
conn.close()
|
||
|
||
return result, stats
|