# -*- 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