feat(emoticon): add emoticon export CLI (#137, closes #118)

Closes #118 (bigmanBass666 提的表情包导出 feature).

作者 thedavidweng 实测验证 (983 NonStore + 204 Store emoticons / 1251 cache files):
- ❌ 解密本地 cache 文件不可行 — 加密方式未知, 不是 V1/V2/XOR/AES 任意标准方案
- ✅ CDN 下载 via cdn_url 可行 — 所有 20/20 测试条目返回 plain image (GIF/JPG/PNG), 无需登录态
- ❌ 构造 CDN URL 不可行 — filekey/storeid 是 per-emoticon

方案: 抽 emoji 公共逻辑到 emoticons.py + 新 CLI export_emoticons.py.

改动:
- emoticons.py 新增 (286 行) — build_emoji_lookup + download_emoji, 不依赖 monitor_web
- export_emoticons.py 新增 (125 行) — CLI 带 --dry-run / --filter / -o 参数
- monitor_web.py 净 -175 行 — emoji 逻辑迁到 emoticons.py, 用 thin wrapper 调用, 保留 emoji_ 文件名前缀兼容已有缓存
- main.py +5 行 — 加 emoticons subcommand
- Makefile + README 文档

注意: 现有 monitor_web 进程要重启才能用新代码 (refactor 内部接口, 行为对等).

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
 

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`) |
 
 </details>
 

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

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()

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()

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
 
 
 # ============ 监听器 ============