diff --git a/Makefile b/Makefile index a47a09e..d4da78f 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: setup build keys decrypt web export all status clean help +.PHONY: setup build keys decrypt web export emoticons all status clean help PYTHON ?= .venv/bin/python3 export SHELL := /bin/bash @@ -11,8 +11,9 @@ help: @echo " make keys 提取密钥（需要 root）" @echo " make decrypt 提取密钥 + 解密全部数据库" @echo " make web 启动 Web UI（实时消息监听）" - @echo " make export 解密 + 批量导出聊天记录" - @echo " make all 从零到完成：setup → keys → decrypt → export" + @echo " make export 解密 + 批量导出聊天记录" + @echo " make emoticons 导出收藏的表情包" + @echo " make all 从零到完成：setup → keys → decrypt → export" @echo " make status 显示当前数据状态和磁盘用量" @echo " make clean 交互式清理临时数据（解密库/导出/缓存）" @echo "" @@ -36,6 +37,9 @@ web: export: $(PYTHON) main.py export +emoticons: + $(PYTHON) main.py emoticons + all: $(PYTHON) main.py all diff --git a/README.md b/README.md index 2a30b60..3c667ed 100644 --- a/README.md +++ b/README.md @@ -225,6 +225,7 @@ py -m pip install --user -r requirements.txt | 导出时间窗口 delta JSON（不覆盖完整 JSON） | `python export_all_chats.py output_dir --delta-only --start "2026-05-26 00:00:00" --end "2026-05-26 08:00:00"` | | 批量导出 + 语音转录 | `python export_all_chats.py --with-transcriptions` | | 转录单个文件语音 | `python transcribe_chat.py input.json [output.json]` | +| 导出收藏的表情包 | `python main.py emoticons` 或 `python export_emoticons.py` | | 注册 MCP Server（Claude） | `claude mcp add wechat -- python /path/to/mcp_server.py` | 批量导出会在输出目录自动维护 `_export_index.json`，用稳定的 `username` @@ -258,6 +259,19 @@ py -m pip install --user -r requirements.txt | `GET /api/tags` | 联系人标签 | | `GET /stream` | SSE 实时消息推送 | +### 表情包导出 + +从 `emoticon.db` 解密后读取表情列表，通过 CDN 直接下载图片到本地目录。 + +```bash +python main.py emoticons # 全量导出到 ./exported_emoticons/ +python export_emoticons.py -o ./my_emojis # 指定输出目录 +python export_emoticons.py --dry-run # 预览模式，不下载 +python export_emoticons.py --filter "猫" # 按关键词过滤 +``` + +支持 NonStore（用户收藏）表情。导出格式为原始图片（GIF / JPG / PNG / WebP），文件名为表情 MD5。 + ### MCP Server（Claude AI 集成） 将微信数据查询能力接入 Claude Code，让 AI 直接读取你的微信消息。 @@ -409,6 +423,8 @@ make help # 列出所有命令 | `export_messages.py` | CSV / HTML / JSON 三种格式导出, 图片可内联 (PR #107) | | `export_wxwork_messages.py` | 企业微信版导出 (CSV / HTML / JSON) | | `export_sns.py` | 朋友圈 SnsTimeLine 导出 (JSON + HTML 时间线) | +| `emoticons.py` | 表情包公共模块 (emoticon.db 解析 + CDN 下载) | +| `export_emoticons.py` | 表情包批量导出 CLI (`--dry-run` / `--filter` / `-o`) | diff --git a/emoticons.py b/emoticons.py new file mode 100644 index 0000000..0a66cfb --- /dev/null +++ b/emoticons.py @@ -0,0 +1,286 @@ +""" +表情包公共模块 — 从 emoticon.db 构建 MD5→CDN 映射 + 下载 + +供 monitor_web.py、export_emoticons.py 复用。 +""" +import os +import re +import sqlite3 +import struct +import tempfile +import urllib.request + +from Crypto.Cipher import AES + +from key_utils import get_key_info +from decrypt_db import decrypt_page + +PAGE_SZ = 4096 +WAL_HEADER_SZ = 32 +WAL_FRAME_HEADER_SZ = 24 + + +def _full_decrypt(db_path, out_path, enc_key): + """全量解密数据库（跳过 HMAC 校验，适合运行时使用）""" + file_size = os.path.getsize(db_path) + total_pages = file_size // PAGE_SZ + os.makedirs(os.path.dirname(out_path), exist_ok=True) + with open(db_path, "rb") as fin, open(out_path, "wb") as fout: + for pgno in range(1, total_pages + 1): + page = fin.read(PAGE_SZ) + if len(page) < PAGE_SZ: + if len(page) > 0: + page = page + b"\x00" * (PAGE_SZ - len(page)) + else: + break + fout.write(decrypt_page(enc_key, page, pgno)) + + +def _decrypt_wal(wal_path, out_path, enc_key): + """解密 WAL 有效 frame 并 patch 到已解密的 DB 副本""" + if not os.path.exists(wal_path): + return + wal_size = os.path.getsize(wal_path) + if wal_size <= WAL_HEADER_SZ: + return + frame_size = WAL_FRAME_HEADER_SZ + PAGE_SZ + with open(wal_path, "rb") as wf, open(out_path, "r+b") as df: + wal_hdr = wf.read(WAL_HEADER_SZ) + wal_salt1 = struct.unpack(">I", wal_hdr[16:20])[0] + wal_salt2 = struct.unpack(">I", wal_hdr[20:24])[0] + while wf.tell() + frame_size <= wal_size: + fh = wf.read(WAL_FRAME_HEADER_SZ) + if len(fh) < WAL_FRAME_HEADER_SZ: + break + pgno = struct.unpack(">I", fh[0:4])[0] + frame_salt1 = struct.unpack(">I", fh[8:12])[0] + frame_salt2 = struct.unpack(">I", fh[12:16])[0] + ep = wf.read(PAGE_SZ) + if len(ep) < PAGE_SZ: + break + if pgno == 0 or pgno > 1000000: + continue + if frame_salt1 != wal_salt1 or frame_salt2 != wal_salt2: + continue + dec = decrypt_page(enc_key, ep, pgno) + df.seek((pgno - 1) * PAGE_SZ) + df.write(dec) + + +def build_emoji_lookup(keys_dict, db_dir): + """从 emoticon.db 构建 emoji md5 → URL 映射。 + + 返回: {md5: {cdn_url, aes_key, encrypt_url, caption, product_id}} + """ + key_info = get_key_info(keys_dict, os.path.join("emoticon", "emoticon.db")) + if not key_info: + return {} + + src = os.path.join(db_dir, "emoticon", "emoticon.db") + if not os.path.exists(src): + return {} + + dst = os.path.join(tempfile.gettempdir(), "wechat_emoticon_dec.db") + enc_key = bytes.fromhex(key_info["enc_key"]) + + try: + _full_decrypt(src, dst, enc_key) + wal = src + "-wal" + if os.path.exists(wal): + _decrypt_wal(wal, dst, enc_key) + except Exception as e: + print(f"[emoticons] emoticon.db 解密失败: {e}", flush=True) + return {} + + try: + conn = sqlite3.connect(f"file:{dst}?mode=ro", uri=True) + lookup = {} + + # 1. NonStore 表情（有独立 cdn_url） + rows = conn.execute( + "SELECT md5, aes_key, cdn_url, encrypt_url, product_id FROM kNonStoreEmoticonTable" + ).fetchall() + pkg_cdn_template = {} + for md5, aes_key, cdn_url, encrypt_url, product_id in rows: + if md5: + lookup[md5] = { + "cdn_url": cdn_url or "", + "aes_key": aes_key or "", + "encrypt_url": encrypt_url or "", + "product_id": product_id or "", + } + if product_id and cdn_url: + pkg_cdn_template[product_id] = cdn_url + + non_store_count = len(lookup) + + # 2. Store 表情（尝试构造 cdn_url） + store_rows = conn.execute( + "SELECT package_id_, md5_ FROM kStoreEmoticonFilesTable" + ).fetchall() + store_added = 0 + for pkg_id, md5 in store_rows: + if md5 and md5 not in lookup: + template = pkg_cdn_template.get(pkg_id, "") + if template and "&" in template: + constructed = re.sub(r"m=[0-9a-f]+", f"m={md5}", template) + lookup[md5] = { + "cdn_url": constructed, + "aes_key": "", + "encrypt_url": "", + "product_id": pkg_id or "", + } + store_added += 1 + + # 3. 收集 caption（表情描述） + try: + captions = conn.execute( + "SELECT md5_, caption_ FROM kStoreEmoticonCaptionsTable WHERE language_='default'" + ).fetchall() + for md5, caption in captions: + if md5 in lookup: + lookup[md5]["caption"] = caption or "" + except Exception: + pass + + conn.close() + print( + f"[emoticons] 已加载 {non_store_count} NonStore + {store_added} Store = {len(lookup)} 个表情映射", + flush=True, + ) + return lookup + except Exception as e: + print(f"[emoticons] 构建映射失败: {e}", flush=True) + return {} + finally: + try: + os.unlink(dst) + except OSError: + pass + + +def convert_hevc_to_jpeg(hevc_path, jpeg_path): + """将 wxgf/HEVC 文件转为 JPEG + + wxgf 是微信自有格式: wxgf header + ICC profile + HEVC NAL units + 通过扫描 HEVC VPS start code (00 00 00 01 40 01) 定位 Annex B 流， + 再用 PyAV (ffmpeg) 解码首帧为 JPEG。 + """ + try: + import av + + with open(hevc_path, 'rb') as f: + data = f.read() + + # 扫描 HEVC Annex B VPS start code: 00 00 00 01 40 01 + vps_sig = b'\x00\x00\x00\x01\x40\x01' + hevc_start = data.find(vps_sig) + if hevc_start < 0: + # fallback: 找 SPS (00 00 00 01 42 01) + hevc_start = data.find(b'\x00\x00\x00\x01\x42\x01') + if hevc_start < 0: + return None + + # 提取 HEVC Annex B 流并用 PyAV 解码 + h265_path = hevc_path + '.h265' + with open(h265_path, 'wb') as f: + f.write(data[hevc_start:]) + + try: + container = av.open(h265_path, format='hevc') + for frame in container.decode(video=0): + img = frame.to_image() + img.save(jpeg_path, "JPEG", quality=90) + container.close() + return jpeg_path + container.close() + finally: + if os.path.exists(h265_path): + os.unlink(h265_path) + + except ImportError: + pass + except Exception: + pass + return None + + +def download_emoji(md5, lookup, out_dir): + """从 CDN 下载单个表情到 out_dir，返回文件名或 None。 + + lookup: build_emoji_lookup() 返回的字典 + """ + info = lookup.get(md5) + if not info: + return None + + # 检查是否已缓存 + for ext in (".gif", ".png", ".jpg", ".webp"): + cached = os.path.join(out_dir, f"{md5}{ext}") + if os.path.exists(cached): + return f"{md5}{ext}" + + cdn_url = info.get("cdn_url", "") + aes_key = info.get("aes_key", "") + encrypt_url = info.get("encrypt_url", "") + + data = None + # 方法1: 从 cdn_url 直接下载（未加密） + if cdn_url: + try: + req = urllib.request.Request(cdn_url, headers={"User-Agent": "Mozilla/5.0"}) + resp = urllib.request.urlopen(req, timeout=15) + data = resp.read() + except Exception: + pass + + # 方法2: 从 encrypt_url 下载 + AES-CBC 解密 + if not data and encrypt_url and aes_key: + try: + req = urllib.request.Request(encrypt_url, headers={"User-Agent": "Mozilla/5.0"}) + resp = urllib.request.urlopen(req, timeout=15) + enc_data = resp.read() + key_bytes = bytes.fromhex(aes_key) + cipher = AES.new(key_bytes, AES.MODE_CBC, iv=key_bytes) + data = cipher.decrypt(enc_data) + if data: + pad = data[-1] + if 1 <= pad <= 16 and data[-pad:] == bytes([pad]) * pad: + data = data[:-pad] + except Exception: + pass + + if not data or len(data) < 4: + return None + + # 检测格式 + if data[:3] == b"\xff\xd8\xff": + ext = ".jpg" + elif data[:4] == b"\x89PNG": + ext = ".png" + elif data[:3] == b"GIF": + ext = ".gif" + elif data[:4] == b"RIFF": + ext = ".webp" + elif data[:4] == b"WXGF" or b"\x00\x00\x00\x01\x40\x01" in data[:256]: + # wxgf/wxam (HEVC): 先存临时文件再转 JPEG + os.makedirs(out_dir, exist_ok=True) + tmp_path = os.path.join(out_dir, f"{md5}.wxgf") + with open(tmp_path, "wb") as f: + f.write(data) + jpg_path = os.path.join(out_dir, f"{md5}.jpg") + if convert_hevc_to_jpeg(tmp_path, jpg_path): + os.unlink(tmp_path) + return f"{md5}.jpg" + # 转换失败，保留 .bin + os.replace(tmp_path, os.path.join(out_dir, f"{md5}.bin")) + return f"{md5}.bin" + else: + ext = ".bin" + + os.makedirs(out_dir, exist_ok=True) + out_name = f"{md5}{ext}" + out_path = os.path.join(out_dir, out_name) + with open(out_path, "wb") as f: + f.write(data) + return out_name diff --git a/export_emoticons.py b/export_emoticons.py new file mode 100644 index 0000000..dfcdfa8 --- /dev/null +++ b/export_emoticons.py @@ -0,0 +1,125 @@ +""" +导出收藏的表情包 + +python3 export_emoticons.py # 导出全部到 ./exported_emoticons/ +python3 export_emoticons.py -o ./my_emojis # 指定输出目录 +python3 export_emoticons.py --dry-run # 仅列出，不下载 +python3 export_emoticons.py --filter "猫" # 按关键词过滤（匹配 caption 或 product_id） +""" +import argparse +import json +import os +import sys + +import functools +print = functools.partial(print, flush=True) + +from config import load_config +from emoticons import build_emoji_lookup, download_emoji +from key_utils import strip_key_metadata + + +def main(): + parser = argparse.ArgumentParser( + description="导出微信收藏的表情包图片", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +示例: + python3 export_emoticons.py 全量导出 + python3 export_emoticons.py -o ./my_emojis 指定输出目录 + python3 export_emoticons.py --dry-run 预览模式 + python3 export_emoticons.py --filter "猫" 按关键词过滤 +""", + ) + parser.add_argument( + "output_dir", + nargs="?", + default=None, + help="输出目录路径 (默认: ./exported_emoticons)", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="预览模式：显示表情列表，不实际下载", + ) + parser.add_argument( + "--filter", + default=None, + help="按关键词过滤（匹配 caption 或 product_id）", + ) + args = parser.parse_args() + + cfg = load_config() + keys_file = cfg["keys_file"] + db_dir = cfg["db_dir"] + + if not os.path.exists(keys_file): + print("[!] 密钥文件不存在，请先运行 python main.py 提取密钥") + sys.exit(1) + + with open(keys_file, encoding="utf-8") as f: + keys = json.load(f) + keys = strip_key_metadata(keys) + if not keys: + print("[!] 密钥为空，请先运行 python main.py 提取密钥") + sys.exit(1) + + script_dir = os.path.dirname(os.path.abspath(__file__)) + output_dir = args.output_dir or os.path.join(script_dir, "exported_emoticons") + + print(f"[*] 构建表情映射...") + lookup = build_emoji_lookup(keys, db_dir) + if not lookup: + print("[!] 未找到任何表情") + sys.exit(1) + + # 过滤 + items = list(lookup.items()) + if args.filter: + keyword = args.filter.lower() + items = [ + (md5, info) for md5, info in items + if keyword in (info.get("caption", "") or "").lower() + or keyword in (info.get("product_id", "") or "").lower() + ] + print(f"[*] 过滤后: {len(items)} 个表情") + + if args.dry_run: + print(f"\n{'MD5':<34} {'格式':>6} {'来源':>8} {'描述'}") + print("-" * 80) + for md5, info in items: + src = "NonStore" if info.get("product_id", "") == "" else "Store" + caption = info.get("caption", "") or "" + has_cdn = "Y" if info.get("cdn_url") else "N" + print(f"{md5} cdn={has_cdn} {src:>8} {caption}") + print(f"\n共 {len(items)} 个表情") + return + + os.makedirs(output_dir, exist_ok=True) + + print(f"[*] 开始下载 {len(items)} 个表情到 {output_dir}") + print() + + try: + from tqdm import tqdm + iterator = tqdm(items, desc="下载表情", unit="个") + except ImportError: + iterator = items + + success = 0 + failed = 0 + + for md5, info in iterator: + fname = download_emoji(md5, lookup, output_dir) + if fname: + success += 1 + else: + failed += 1 + + print() + print(f"[+] 完成: {success} 成功, {failed} 失败") + print(f" 输出目录: {output_dir}") + + +if __name__ == "__main__": + main() diff --git a/main.py b/main.py index 54d943f..83dfee8 100644 --- a/main.py +++ b/main.py @@ -262,6 +262,7 @@ def print_usage(): print(" python main.py decode-images --help 查看 decode-images 全部选项") print(" python main.py export 解密 + 批量导出聊天记录") print(" python main.py all 从零到完成：密钥 → 解密 → 导出") + print(" python main.py emoticons 导出收藏的表情包") print(" python main.py status 显示当前状态和磁盘用量") @@ -346,6 +347,10 @@ def main(): print(" 未配置语音转录 backend (config.json 中设置)") print(" 配置后运行: python export_all_chats.py --with-transcriptions") + elif cmd == "emoticons": + from export_emoticons import main as export_emojis + export_emojis() + elif cmd == "web": print("[*] 启动 Web UI...") print() diff --git a/monitor_web.py b/monitor_web.py index 8b3679b..8cf16ba 100644 --- a/monitor_web.py +++ b/monitor_web.py @@ -54,182 +54,52 @@ MAX_LOG = 500 _img_executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix='img') _hidden_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix='hidden') -# ---- Emoji 缓存 (md5 → {cdn_url, aes_key, encrypt_url}) ---- +# ---- Emoji 缓存 (使用 emoticons.py 公共模块) ---- +from emoticons import build_emoji_lookup as _build_emoji_lookup_mod, download_emoji as _download_emoji_mod, convert_hevc_to_jpeg as _convert_hevc_to_jpeg + _emoji_lookup = {} # md5 → dict _emoji_lookup_lock = threading.Lock() - -_emoji_keys_dict = None # 保存 keys 引用供刷新用 +_emoji_keys_dict = None _emoji_last_refresh = 0 + def _build_emoji_lookup(keys_dict): - """从 emoticon.db 构建 emoji md5 → URL 映射（直接解密，不走 cache）""" global _emoji_lookup, _emoji_keys_dict, _emoji_last_refresh _emoji_keys_dict = keys_dict - key_info = get_key_info(keys_dict, os.path.join("emoticon", "emoticon.db")) - if not key_info: - print("[emoji] 无 emoticon.db key，跳过", flush=True) - return - - src = os.path.join(DB_DIR, "emoticon", "emoticon.db") - if not os.path.exists(src): - return - - import tempfile - dst = os.path.join(tempfile.gettempdir(), "wechat_emoticon_dec.db") - enc_key = bytes.fromhex(key_info["enc_key"]) - - try: - full_decrypt(src, dst, enc_key) - wal = src + "-wal" - if os.path.exists(wal): - decrypt_wal_full(wal, dst, enc_key) - except Exception as e: - print(f"[emoji] emoticon.db 解密失败: {e}", flush=True) - return - - try: - conn = sqlite3.connect(f"file:{dst}?mode=ro", uri=True) - new_lookup = {} - - # 1. NonStore 表情（有独立 cdn_url） - rows = conn.execute( - "SELECT md5, aes_key, cdn_url, encrypt_url, product_id FROM kNonStoreEmoticonTable" - ).fetchall() - # 收集每个 package 的 cdn_url 模板 - pkg_cdn_template = {} # package_id → cdn_url (任意一个) - for md5, aes_key, cdn_url, encrypt_url, product_id in rows: - if md5: - new_lookup[md5] = { - 'cdn_url': cdn_url or '', - 'aes_key': aes_key or '', - 'encrypt_url': encrypt_url or '', - } - if product_id and cdn_url: - pkg_cdn_template[product_id] = cdn_url - - non_store_count = len(new_lookup) - - # 2. Store 表情（尝试构造 cdn_url） - store_rows = conn.execute( - "SELECT package_id_, md5_ FROM kStoreEmoticonFilesTable" - ).fetchall() - store_added = 0 - for pkg_id, md5 in store_rows: - if md5 and md5 not in new_lookup: - # 尝试用同 package 的模板构造 URL - template = pkg_cdn_template.get(pkg_id, '') - if template and '&' in template: - # 替换 m= 参数为新 md5 - import re - constructed = re.sub(r'm=[0-9a-f]+', f'm={md5}', template) - new_lookup[md5] = { - 'cdn_url': constructed, - 'aes_key': '', - 'encrypt_url': '', - } - store_added += 1 - - conn.close() + lookup = _build_emoji_lookup_mod(keys_dict, DB_DIR) + if lookup: with _emoji_lookup_lock: - _emoji_lookup = new_lookup + _emoji_lookup = lookup _emoji_last_refresh = time.time() - print(f"[emoji] 已加载 {non_store_count} NonStore + {store_added} Store = {len(new_lookup)} 个表情映射", flush=True) - except Exception as e: - print(f"[emoji] 构建映射失败: {e}", flush=True) - finally: - try: - os.unlink(dst) - except OSError: - pass + def _download_emoji(md5): - """从 CDN 下载表情并缓存到 decoded_images/，返回文件名或 None""" with _emoji_lookup_lock: info = _emoji_lookup.get(md5) if not info: - # Lookup miss: 刷新 emoticon.db（最多每60秒一次） if _emoji_keys_dict and time.time() - _emoji_last_refresh > 60: print(f" [emoji] lookup miss, 刷新 emoticon.db...", flush=True) _build_emoji_lookup(_emoji_keys_dict) - with _emoji_lookup_lock: - info = _emoji_lookup.get(md5) + with _emoji_lookup_lock: + info = _emoji_lookup.get(md5) if not info: return None - # 先检查是否已缓存 + # 先检查是否已缓存（旧文件名格式兼容） for ext in ('.gif', '.png', '.jpg', '.webp'): cached = os.path.join(DECODED_IMAGE_DIR, f"emoji_{md5}{ext}") if os.path.exists(cached): return f"emoji_{md5}{ext}" - cdn_url = info.get('cdn_url', '') - aes_key = info.get('aes_key', '') - encrypt_url = info.get('encrypt_url', '') - - data = None - # 方法1: 从 cdn_url 直接下载（未加密） - if cdn_url: - try: - import urllib.request - req = urllib.request.Request(cdn_url, headers={'User-Agent': 'Mozilla/5.0'}) - resp = urllib.request.urlopen(req, timeout=15) - data = resp.read() - except Exception as e: - print(f" [emoji] cdn下载失败 {md5[:12]}: {e}", flush=True) - - # 方法2: 从 encrypt_url 下载 + AES-CBC 解密 - if not data and encrypt_url and aes_key: - try: - import urllib.request - req = urllib.request.Request(encrypt_url, headers={'User-Agent': 'Mozilla/5.0'}) - resp = urllib.request.urlopen(req, timeout=15) - enc_data = resp.read() - key_bytes = bytes.fromhex(aes_key) - cipher = AES.new(key_bytes, AES.MODE_CBC, iv=key_bytes) - data = cipher.decrypt(enc_data) - # 去除 PKCS7 padding - if data: - pad = data[-1] - if 1 <= pad <= 16 and data[-pad:] == bytes([pad]) * pad: - data = data[:-pad] - except Exception as e: - print(f" [emoji] encrypt下载解密失败 {md5[:12]}: {e}", flush=True) - - if not data or len(data) < 4: - return None - - # 检测格式 - if data[:3] == b'\xff\xd8\xff': - ext = '.jpg' - elif data[:4] == b'\x89PNG': - ext = '.png' - elif data[:3] == b'GIF': - ext = '.gif' - elif data[:4] == b'RIFF': - ext = '.webp' - elif data[:4] in (b'wxgf', b'wxam'): - # WXGF/WXAM 需要转换 - ext = '.gif' - tmp_path = os.path.join(DECODED_IMAGE_DIR, f"emoji_{md5}.wxgf") - with open(tmp_path, 'wb') as f: - f.write(data) - jpg_path = _convert_hevc_to_jpeg(tmp_path, os.path.join(DECODED_IMAGE_DIR, f"emoji_{md5}.jpg")) - try: - os.unlink(tmp_path) - except OSError: - pass - if jpg_path: - return f"emoji_{md5}.jpg" - return None - else: - ext = '.bin' - - out_name = f"emoji_{md5}{ext}" - out_path = os.path.join(DECODED_IMAGE_DIR, out_name) - with open(out_path, 'wb') as f: - f.write(data) - print(f" [emoji] 下载缓存: {out_name} ({len(data)//1024}KB)", flush=True) - return out_name + result = _download_emoji_mod(md5, _emoji_lookup, DECODED_IMAGE_DIR) + if result: + # 统一用 emoji_ 前缀（兼容旧缓存） + src = os.path.join(DECODED_IMAGE_DIR, result) + dst = os.path.join(DECODED_IMAGE_DIR, f"emoji_{result}") + if src != dst and os.path.exists(src): + os.replace(src, dst) + return f"emoji_{result}" + return None class MonitorDBCache: @@ -598,51 +468,6 @@ def broadcast_sse(msg_data): sse_clients.remove(q) -def _convert_hevc_to_jpeg(hevc_path, jpeg_path): - """将 wxgf/HEVC 文件转为 JPEG - - wxgf 是微信自有格式: wxgf header + ICC profile + HEVC NAL units - 通过扫描 HEVC VPS start code (00 00 00 01 40 01) 定位 Annex B 流， - 再用 PyAV (ffmpeg) 解码首帧为 JPEG。 - """ - try: - import av - - with open(hevc_path, 'rb') as f: - data = f.read() - - # 扫描 HEVC Annex B VPS start code: 00 00 00 01 40 01 - vps_sig = b'\x00\x00\x00\x01\x40\x01' - hevc_start = data.find(vps_sig) - if hevc_start < 0: - # fallback: 找 SPS (00 00 00 01 42 01) - hevc_start = data.find(b'\x00\x00\x00\x01\x42\x01') - if hevc_start < 0: - print(f" [img] wxgf 中未找到 HEVC VPS/SPS", flush=True) - return None - - # 提取 HEVC Annex B 流并用 PyAV 解码 - h265_path = hevc_path + '.h265' - with open(h265_path, 'wb') as f: - f.write(data[hevc_start:]) - - try: - container = av.open(h265_path, format='hevc') - for frame in container.decode(video=0): - img = frame.to_image() - img.save(jpeg_path, "JPEG", quality=90) - container.close() - return jpeg_path - container.close() - finally: - if os.path.exists(h265_path): - os.unlink(h265_path) - - except ImportError: - print(f" [img] 需要 PyAV: pip install av", flush=True) - except Exception as e: - print(f" [img] HEVC→JPEG 失败: {e}", flush=True) - return None # ============ 监听器 ============