@@ -455,7 +670,8 @@ const isGettingDbKey = ref(false)
const steps = [
{ title: '数据库解密' },
{ title: '填写图片密钥' },
- { title: '图片解密' }
+ { title: '图片解密' },
+ { title: '表情下载' }
]
// 表单数据
@@ -735,6 +951,10 @@ const clearManualKeys = () => {
// 图片解密相关
const mediaDecryptResult = ref(null)
const mediaDecrypting = ref(false)
+const mediaDecryptConcurrency = ref(10)
+const emojiDownloadResult = ref(null)
+const emojiDownloading = ref(false)
+const emojiDownloadConcurrency = ref(20)
// 数据库解密进度（SSE）
const dbDecryptProgress = reactive({
@@ -756,12 +976,14 @@ const dbProgressPercent = computed(() => {
const decryptProgress = reactive({
current: 0,
total: 0,
+ concurrency: 0,
success_count: 0,
skip_count: 0,
fail_count: 0,
current_file: '',
fileStatus: '',
- status: ''
+ status: '',
+ message: ''
})
// 进度百分比
@@ -770,6 +992,36 @@ const progressPercent = computed(() => {
return Math.round((decryptProgress.current / decryptProgress.total) * 100)
})
+const emojiDownloadProgress = reactive({
+ current: 0,
+ total: 0,
+ concurrency: 0,
+ success_count: 0,
+ skip_count: 0,
+ fail_count: 0,
+ current_file: '',
+ fileStatus: '',
+ status: '',
+ message: ''
+})
+
+const emojiProgressPercent = computed(() => {
+ if (emojiDownloadProgress.total === 0) return 0
+ return Math.round((emojiDownloadProgress.current / emojiDownloadProgress.total) * 100)
+})
+
+const getEmojiDownloadConcurrency = () => {
+ const raw = Number.parseInt(String(emojiDownloadConcurrency.value || 20), 10)
+ if (!Number.isFinite(raw)) return 20
+ return Math.max(1, Math.min(100, raw))
+}
+
+const getMediaDecryptConcurrency = () => {
+ const raw = Number.parseInt(String(mediaDecryptConcurrency.value || 10), 10)
+ if (!Number.isFinite(raw)) return 10
+ return Math.max(1, Math.min(64, raw))
+}
+
// 解密结果存储
const decryptResult = ref(null)
@@ -802,6 +1054,7 @@ const validateForm = () => {
let dbDecryptEventSource = null
let mediaDecryptEventSource = null
+let emojiDownloadEventSource = null
const closeMediaDecryptEventSource = () => {
try {
@@ -813,6 +1066,16 @@ const closeMediaDecryptEventSource = () => {
}
}
+const closeEmojiDownloadEventSource = () => {
+ try {
+ if (emojiDownloadEventSource) emojiDownloadEventSource.close()
+ } catch (e) {
+ // ignore
+ } finally {
+ emojiDownloadEventSource = null
+ }
+}
+
onBeforeUnmount(() => {
try {
if (dbDecryptEventSource) dbDecryptEventSource.close()
@@ -823,6 +1086,7 @@ onBeforeUnmount(() => {
}
closeMediaDecryptEventSource()
+ closeEmojiDownloadEventSource()
})
const resetDbDecryptProgress = () => {
@@ -838,12 +1102,27 @@ const resetDbDecryptProgress = () => {
const resetMediaDecryptProgress = () => {
decryptProgress.current = 0
decryptProgress.total = 0
+ decryptProgress.concurrency = 0
decryptProgress.success_count = 0
decryptProgress.skip_count = 0
decryptProgress.fail_count = 0
decryptProgress.current_file = ''
decryptProgress.fileStatus = ''
decryptProgress.status = ''
+ decryptProgress.message = ''
+}
+
+const resetEmojiDownloadProgress = () => {
+ emojiDownloadProgress.current = 0
+ emojiDownloadProgress.total = 0
+ emojiDownloadProgress.concurrency = 0
+ emojiDownloadProgress.success_count = 0
+ emojiDownloadProgress.skip_count = 0
+ emojiDownloadProgress.fail_count = 0
+ emojiDownloadProgress.current_file = ''
+ emojiDownloadProgress.fileStatus = ''
+ emojiDownloadProgress.status = ''
+ emojiDownloadProgress.message = ''
}
// 处理解密
@@ -861,6 +1140,14 @@ const handleDecrypt = async () => {
warning.value = ''
resetDbDecryptProgress()
+ resetMediaDecryptProgress()
+ resetEmojiDownloadProgress()
+ mediaDecryptResult.value = null
+ emojiDownloadResult.value = null
+ mediaDecrypting.value = false
+ emojiDownloading.value = false
+ closeMediaDecryptEventSource()
+ closeEmojiDownloadEventSource()
try {
const canSse = process.client && typeof window !== 'undefined' && typeof EventSource !== 'undefined'
@@ -1024,8 +1311,11 @@ const decryptAllImages = async () => {
mediaDecryptResult.value = null
error.value = ''
warning.value = ''
+ const configuredConcurrency = getMediaDecryptConcurrency()
+ mediaDecryptConcurrency.value = configuredConcurrency
logDecryptDebug('media-decrypt:start', {
account: mediaAccount.value,
+ concurrency: configuredConcurrency,
keys: summarizeKeyStateForLog(mediaKeys.xor_key, mediaKeys.aes_key)
})
@@ -1038,6 +1328,7 @@ const decryptAllImages = async () => {
if (mediaAccount.value) params.set('account', mediaAccount.value)
if (mediaKeys.xor_key) params.set('xor_key', mediaKeys.xor_key)
if (mediaKeys.aes_key) params.set('aes_key', mediaKeys.aes_key)
+ params.set('concurrency', String(configuredConcurrency))
const apiBase = useApiBase()
const url = `${apiBase}/media/decrypt_all_stream?${params.toString()}`
@@ -1053,28 +1344,37 @@ const decryptAllImages = async () => {
if (data.type === 'scanning') {
decryptProgress.current_file = '正在扫描文件...'
+ decryptProgress.message = data.message || '正在扫描图片文件...'
} else if (data.type === 'start') {
- decryptProgress.total = data.total
+ decryptProgress.total = data.total || 0
+ decryptProgress.concurrency = data.concurrency || configuredConcurrency
+ decryptProgress.message = data.message || ''
} else if (data.type === 'progress') {
- decryptProgress.current = data.current
- decryptProgress.total = data.total
- decryptProgress.success_count = data.success_count
- decryptProgress.skip_count = data.skip_count
- decryptProgress.fail_count = data.fail_count
- decryptProgress.current_file = data.current_file
- decryptProgress.fileStatus = data.status
+ decryptProgress.current = data.current || 0
+ decryptProgress.total = data.total || 0
+ decryptProgress.concurrency = data.concurrency || configuredConcurrency
+ decryptProgress.success_count = data.success_count || 0
+ decryptProgress.skip_count = data.skip_count || 0
+ decryptProgress.fail_count = data.fail_count || 0
+ decryptProgress.current_file = data.current_file || ''
+ decryptProgress.fileStatus = data.status || ''
+ decryptProgress.message = data.message || ''
} else if (data.type === 'complete') {
decryptProgress.status = 'complete'
- decryptProgress.current = data.total
- decryptProgress.total = data.total
- decryptProgress.success_count = data.success_count
- decryptProgress.skip_count = data.skip_count
- decryptProgress.fail_count = data.fail_count
+ decryptProgress.current = data.total || 0
+ decryptProgress.total = data.total || 0
+ decryptProgress.concurrency = data.concurrency || configuredConcurrency
+ decryptProgress.success_count = data.success_count || 0
+ decryptProgress.skip_count = data.skip_count || 0
+ decryptProgress.fail_count = data.fail_count || 0
+ decryptProgress.message = data.message || '解密完成'
mediaDecryptResult.value = data
mediaDecrypting.value = false
logDecryptDebug('media-decrypt:complete', {
account: mediaAccount.value,
total: data.total,
+ concurrency: data.concurrency,
+ decrypt_stats: data.decrypt_stats,
success_count: data.success_count,
skip_count: data.skip_count,
fail_count: data.fail_count
@@ -1115,11 +1415,148 @@ const cancelMediaDecrypt = () => {
if (!mediaDecrypting.value) return
decryptProgress.status = 'cancelled'
+ decryptProgress.message = '已停止图片解密'
mediaDecrypting.value = false
warning.value = '已停止图片解密，已完成的图片会保留。'
+ logDecryptDebug('media-decrypt:cancelled', {
+ account: mediaAccount.value,
+ current: decryptProgress.current,
+ total: decryptProgress.total,
+ concurrency: decryptProgress.concurrency || getMediaDecryptConcurrency()
+ })
closeMediaDecryptEventSource()
}
+const downloadAllEmojis = async () => {
+ closeEmojiDownloadEventSource()
+ emojiDownloading.value = true
+ emojiDownloadResult.value = null
+ error.value = ''
+ warning.value = ''
+ const configuredConcurrency = getEmojiDownloadConcurrency()
+ emojiDownloadConcurrency.value = configuredConcurrency
+ logDecryptDebug('emoji-download:start', {
+ account: mediaAccount.value,
+ concurrency: configuredConcurrency
+ })
+
+ resetEmojiDownloadProgress()
+
+ try {
+ const params = new URLSearchParams()
+ if (mediaAccount.value) params.set('account', mediaAccount.value)
+ params.set('concurrency', String(configuredConcurrency))
+ const apiBase = useApiBase()
+ const url = `${apiBase}/media/emoji/download_all_stream?${params.toString()}`
+
+ const eventSource = new EventSource(url)
+ emojiDownloadEventSource = eventSource
+
+ eventSource.onmessage = (event) => {
+ if (emojiDownloadEventSource !== eventSource) return
+
+ try {
+ const data = JSON.parse(event.data)
+
+ if (data.type === 'scanning') {
+ emojiDownloadProgress.current_file = '正在扫描表情资源...'
+ emojiDownloadProgress.message = data.message || '正在扫描表情资源...'
+ } else if (data.type === 'start') {
+ emojiDownloadProgress.total = data.total || 0
+ emojiDownloadProgress.concurrency = data.concurrency || configuredConcurrency
+ emojiDownloadProgress.message = data.message || ''
+ } else if (data.type === 'progress') {
+ emojiDownloadProgress.current = data.current || 0
+ emojiDownloadProgress.total = data.total || 0
+ emojiDownloadProgress.concurrency = data.concurrency || configuredConcurrency
+ emojiDownloadProgress.success_count = data.success_count || 0
+ emojiDownloadProgress.skip_count = data.skip_count || 0
+ emojiDownloadProgress.fail_count = data.fail_count || 0
+ emojiDownloadProgress.current_file = data.current_file || ''
+ emojiDownloadProgress.fileStatus = data.status || ''
+ emojiDownloadProgress.message = data.message || ''
+ } else if (data.type === 'complete') {
+ emojiDownloadProgress.status = 'complete'
+ emojiDownloadProgress.current = data.total || 0
+ emojiDownloadProgress.total = data.total || 0
+ emojiDownloadProgress.concurrency = data.concurrency || configuredConcurrency
+ emojiDownloadProgress.success_count = data.success_count || 0
+ emojiDownloadProgress.skip_count = data.skip_count || 0
+ emojiDownloadProgress.fail_count = data.fail_count || 0
+ emojiDownloadProgress.message = data.message || '表情下载完成'
+ emojiDownloadResult.value = data
+ emojiDownloading.value = false
+ logDecryptDebug('emoji-download:complete', {
+ account: mediaAccount.value,
+ total: data.total,
+ concurrency: data.concurrency,
+ download_stats: data.download_stats,
+ success_count: data.success_count,
+ skip_count: data.skip_count,
+ fail_count: data.fail_count
+ })
+ closeEmojiDownloadEventSource()
+ } else if (data.type === 'error') {
+ error.value = data.message || '表情下载失败'
+ logDecryptDebug('emoji-download:error-event', {
+ account: mediaAccount.value,
+ message: data.message
+ })
+ emojiDownloading.value = false
+ closeEmojiDownloadEventSource()
+ }
+ } catch (e) {
+ console.error('解析表情下载SSE消息失败:', e)
+ }
+ }
+
+ eventSource.onerror = (e) => {
+ if (emojiDownloadEventSource !== eventSource) return
+
+ console.error('表情下载SSE连接错误:', e)
+ closeEmojiDownloadEventSource()
+ if (emojiDownloading.value) {
+ error.value = '表情下载连接中断，请重试'
+ emojiDownloading.value = false
+ }
+ }
+ } catch (err) {
+ error.value = err.message || '表情下载过程中发生错误'
+ emojiDownloading.value = false
+ closeEmojiDownloadEventSource()
+ }
+}
+
+const cancelEmojiDownload = () => {
+ if (!emojiDownloading.value) return
+
+ emojiDownloadProgress.status = 'cancelled'
+ emojiDownloading.value = false
+ warning.value = '已停止表情下载，已完成的表情会保留。'
+ logDecryptDebug('emoji-download:cancelled', {
+ account: mediaAccount.value,
+ current: emojiDownloadProgress.current,
+ total: emojiDownloadProgress.total
+ })
+ closeEmojiDownloadEventSource()
+}
+
+const goToEmojiDownloadStep = () => {
+ if (mediaDecrypting.value) return
+
+ error.value = ''
+ warning.value = ''
+ currentStep.value = 3
+}
+
+const goBackToMediaDecryptStep = () => {
+ if (emojiDownloading.value) return
+
+ error.value = ''
+ warning.value = ''
+ currentStep.value = 2
+}
+
// 从密钥步骤进入图片解密步骤
const goToMediaDecryptStep = async () => {
error.value = ''
diff --git a/src/wechat_decrypt_tool/media_helpers.py b/src/wechat_decrypt_tool/media_helpers.py
index cec1049..60740bc 100644
--- a/src/wechat_decrypt_tool/media_helpers.py
+++ b/src/wechat_decrypt_tool/media_helpers.py
@@ -9,23 +9,56 @@ import os
import re
import sqlite3
import struct
+import time
+from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
-from typing import Any, Optional
+from typing import Any, Iterable, Optional
from urllib.parse import urlparse
from fastapi import HTTPException
from .app_paths import get_output_databases_dir
+from .chat_helpers import _decode_message_content
from .logging_config import get_logger
from .sqlite_diagnostics import is_usable_sqlite_db
logger = get_logger(__name__)
+_MEDIA_INDEX_FILE_EXTS = {
+ ".dat",
+ ".gif",
+ ".heic",
+ ".heif",
+ ".jpeg",
+ ".jpg",
+ ".m4v",
+ ".mov",
+ ".mp4",
+ ".png",
+ ".webp",
+}
+_MEDIA_INDEX_VIDEO_STREAM_EXTS = {
+ ".m4v",
+ ".mov",
+ ".mp4",
+}
+_MEDIA_INDEX_VIDEO_INDEX_EXTS = _MEDIA_INDEX_VIDEO_STREAM_EXTS | {".dat"}
+_MEDIA_INDEX_STRIP_SUFFIX_RE = re.compile(r"(?i)(?:_h|_t|_thumb)$")
+_MEDIA_INDEX_DB_VERSION = 2
+
# 运行时输出目录（桌面端可通过 WECHAT_TOOL_DATA_DIR 指向可写目录）
_PACKAGE_ROOT = Path(__file__).resolve().parent
_SQLITE_HEADER = b"SQLite format 3\x00"
+_EMOTICON_MD5_RE = re.compile(r"(?i)^[0-9a-f]{32}$")
+_EMOTICON_MD5_ATTR_RE = re.compile(r"(?i)\bmd5\s*=\s*['\"]([0-9a-f]{32})['\"]")
+_EMOTICON_MD5_TAG_RE = re.compile(r"(?is)
\s*([0-9a-f]{32})\s*")
+_EMOTICON_EXTERN_MD5_ATTR_RE = re.compile(r"(?i)\bextern_?md5\s*=\s*['\"]([0-9a-f]{32})['\"]")
+_EMOTICON_EXTERN_MD5_TAG_RE = re.compile(r"(?is)
\s*([0-9a-f]{32})\s*")
+_EMOTICON_AES_KEY_ATTR_RE = re.compile(r"(?i)\baes_?key\s*=\s*['\"]([0-9a-f]{32})['\"]")
+_EMOTICON_AES_KEY_TAG_RE = re.compile(r"(?is)
\s*([0-9a-f]{32})\s*")
+_EMOTICON_HTTP_URL_RE = re.compile(r"(?i)https?://[^\s<>\"']+")
def _is_valid_decrypted_sqlite(path: Path) -> bool:
@@ -390,6 +423,182 @@ def _decrypt_emoticon_aes_cbc(data: bytes, aes_key_hex: str) -> Optional[bytes]:
return None
+def _normalize_emoticon_md5(value: Any) -> str:
+ md5 = str(value or "").strip().lower()
+ return md5 if _EMOTICON_MD5_RE.fullmatch(md5) else ""
+
+
+def _normalize_emoticon_aes_key(value: Any) -> str:
+ key = str(value or "").strip().lower()
+ return key if _EMOTICON_MD5_RE.fullmatch(key) else ""
+
+
+def _first_emoticon_match(text: str, patterns: tuple[re.Pattern[str], ...]) -> str:
+ if not text:
+ return ""
+ for pattern in patterns:
+ try:
+ match = pattern.search(text)
+ except Exception:
+ match = None
+ if match:
+ return str(match.group(1) or "").strip()
+ return ""
+
+
+def _extract_emoticon_message_md5(text: str) -> str:
+ return _normalize_emoticon_md5(_first_emoticon_match(text, (_EMOTICON_MD5_ATTR_RE, _EMOTICON_MD5_TAG_RE)))
+
+
+def _extract_emoticon_message_extern_md5(text: str) -> str:
+ return _normalize_emoticon_md5(
+ _first_emoticon_match(text, (_EMOTICON_EXTERN_MD5_ATTR_RE, _EMOTICON_EXTERN_MD5_TAG_RE))
+ )
+
+
+def _extract_emoticon_message_aes_key(text: str) -> str:
+ return _normalize_emoticon_aes_key(_first_emoticon_match(text, (_EMOTICON_AES_KEY_ATTR_RE, _EMOTICON_AES_KEY_TAG_RE)))
+
+
+def _extract_emoticon_message_urls(text: str) -> list[str]:
+ if not text:
+ return []
+ out: list[str] = []
+ seen: set[str] = set()
+ for match in _EMOTICON_HTTP_URL_RE.finditer(text):
+ url = str(match.group(0) or "").strip()
+ if not url or url in seen or not _is_safe_http_url(url):
+ continue
+ seen.add(url)
+ out.append(url)
+ return out
+
+
+def _emoticon_message_db_paths(account_dir: Path) -> list[Path]:
+ return sorted(
+ p
+ for p in Path(account_dir).glob("message_*.db")
+ if p.is_file() and p.name.lower() != "message_resource.db"
+ )
+
+
+def _emoticon_source_fingerprint(account_dir: Path) -> str:
+ parts: list[str] = []
+ paths = [Path(account_dir) / "emoticon.db", *_emoticon_message_db_paths(account_dir)]
+ for path in paths:
+ try:
+ st = path.stat()
+ parts.append(f"{path.name}:{st.st_size}:{st.st_mtime_ns}")
+ except Exception:
+ parts.append(f"{path.name}:missing")
+ return "|".join(parts)
+
+
+def _list_emoticon_message_tables(conn: sqlite3.Connection) -> list[str]:
+ try:
+ rows = conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
+ except Exception:
+ return []
+ out: list[str] = []
+ for row in rows:
+ if not row:
+ continue
+ raw_name = row[0]
+ if isinstance(raw_name, memoryview):
+ raw_name = raw_name.tobytes()
+ if isinstance(raw_name, (bytes, bytearray)):
+ try:
+ name = bytes(raw_name).decode("utf-8", errors="ignore")
+ except Exception:
+ continue
+ else:
+ name = str(raw_name or "")
+ if name.lower().startswith(("msg_", "chat_")):
+ out.append(name)
+ return out
+
+
+def _quote_sqlite_ident(name: str) -> str:
+ return '"' + str(name or "").replace('"', '""') + '"'
+
+
+def _iter_emoticon_varints(data: bytes) -> list[tuple[int, int]]:
+ out: list[tuple[int, int]] = []
+ i = 0
+ n = len(data)
+ while i < n:
+ key = int(data[i])
+ i += 1
+ field = key >> 3
+ wire_type = key & 0x07
+ if field <= 0:
+ break
+
+ if wire_type == 0:
+ shift = 0
+ value = 0
+ while i < n:
+ b = int(data[i])
+ i += 1
+ value |= (b & 0x7F) << shift
+ if b < 0x80:
+ break
+ shift += 7
+ out.append((field, int(value)))
+ continue
+
+ if wire_type == 1:
+ i += 8
+ continue
+
+ if wire_type == 2:
+ shift = 0
+ ln = 0
+ while i < n:
+ b = int(data[i])
+ i += 1
+ ln |= (b & 0x7F) << shift
+ if b < 0x80:
+ break
+ shift += 7
+ i += int(ln)
+ continue
+
+ if wire_type == 5:
+ i += 4
+ continue
+
+ break
+ return out
+
+
+def _extract_emoticon_builtin_expr_id(packed_info_data: Any) -> Optional[int]:
+ data: bytes = b""
+ if packed_info_data is None:
+ return None
+ if isinstance(packed_info_data, memoryview):
+ data = packed_info_data.tobytes()
+ elif isinstance(packed_info_data, (bytes, bytearray)):
+ data = bytes(packed_info_data)
+ elif isinstance(packed_info_data, str):
+ s = packed_info_data.strip()
+ if s:
+ try:
+ data = bytes.fromhex(s) if (len(s) % 2 == 0 and re.fullmatch(r"(?i)[0-9a-f]+", s)) else s.encode(
+ "utf-8",
+ errors="ignore",
+ )
+ except Exception:
+ data = b""
+ if not data:
+ return None
+
+ for field, value in _iter_emoticon_varints(data):
+ if field == 2:
+ return int(value)
+ return None
+
+
@lru_cache(maxsize=2048)
def _lookup_emoticon_info(account_dir_str: str, md5: str) -> dict[str, str]:
account_dir = Path(account_dir_str)
@@ -405,9 +614,11 @@ def _lookup_emoticon_info(account_dir_str: str, md5: str) -> dict[str, str]:
conn.row_factory = sqlite3.Row
try:
row = conn.execute(
- "SELECT md5, aes_key, cdn_url, encrypt_url, extern_url, thumb_url, tp_url "
- "FROM kNonStoreEmoticonTable WHERE lower(md5) = lower(?) LIMIT 1",
- (md5s,),
+ "SELECT md5, extern_md5, aes_key, cdn_url, encrypt_url, extern_url, thumb_url, tp_url "
+ "FROM kNonStoreEmoticonTable "
+ "WHERE lower(md5) = lower(?) OR lower(extern_md5) = lower(?) "
+ "LIMIT 1",
+ (md5s, md5s),
).fetchone()
if not row:
return {}
@@ -421,19 +632,320 @@ def _lookup_emoticon_info(account_dir_str: str, md5: str) -> dict[str, str]:
pass
-def _try_fetch_emoticon_from_remote(account_dir: Path, md5: str) -> tuple[Optional[bytes], Optional[str]]:
- info = _lookup_emoticon_info(str(account_dir), str(md5 or "").lower())
- if not info:
- return None, None
+def _merge_emoticon_candidate(
+ catalog: dict[str, dict[str, Any]],
+ md5: str,
+ *,
+ urls: Optional[list[str]] = None,
+ aes_key: str = "",
+ source: str = "",
+) -> None:
+ md5s = _normalize_emoticon_md5(md5)
+ if not md5s:
+ return
- aes_key_hex = str(info.get("aes_key") or "").strip()
- urls: list[str] = []
- # Prefer plain CDN URL first; fall back to encrypt_url (needs AES-CBC decrypt).
- for k in ("cdn_url", "extern_url", "thumb_url", "tp_url", "encrypt_url"):
- u = str(info.get(k) or "").strip()
- if u and _is_safe_http_url(u):
- urls.append(u)
+ entry = catalog.get(md5s)
+ if entry is None:
+ entry = {"md5": md5s, "urls": [], "aes_keys": [], "sources": []}
+ catalog[md5s] = entry
+ if source and source not in entry["sources"]:
+ entry["sources"].append(source)
+
+ key = _normalize_emoticon_aes_key(aes_key)
+ if key and key not in entry["aes_keys"]:
+ entry["aes_keys"].append(key)
+
+ seen = set(entry["urls"])
+ for url in urls or []:
+ u = str(url or "").strip()
+ if not u or u in seen or not _is_safe_http_url(u):
+ continue
+ seen.add(u)
+ entry["urls"].append(u)
+
+
+def _emoticon_catalog_public_stats(
+ stats: dict[str, Any],
+ catalog: dict[str, dict[str, Any]],
+ *,
+ elapsed_ms: float,
+) -> dict[str, Any]:
+ source_counts: dict[str, int] = {}
+ with_urls = 0
+ for entry in catalog.values():
+ if entry.get("urls"):
+ with_urls += 1
+ for source in entry.get("sources") or []:
+ source_counts[source] = source_counts.get(source, 0) + 1
+
+ return {
+ "emoticon_db_rows": int(stats.get("emoticon_db_rows") or 0),
+ "emoticon_db_md5": int(stats.get("emoticon_db_md5") or 0),
+ "emoticon_db_extern_md5": int(stats.get("emoticon_db_extern_md5") or 0),
+ "emoticon_db_with_remote": int(stats.get("emoticon_db_with_remote") or 0),
+ "message_db_count": int(stats.get("message_db_count") or 0),
+ "message_table_count": int(stats.get("message_table_count") or 0),
+ "message_xml_rows": int(stats.get("message_xml_rows") or 0),
+ "message_xml_md5": int(stats.get("message_xml_md5") or 0),
+ "message_xml_md5_with_url": int(stats.get("message_xml_md5_with_url") or 0),
+ "message_xml_extern_md5": int(stats.get("message_xml_extern_md5") or 0),
+ "message_builtin_expr_ids": int(stats.get("message_builtin_expr_ids") or 0),
+ "message_builtin_expr_rows": int(stats.get("message_builtin_expr_rows") or 0),
+ "total_candidates": len(catalog),
+ "total_candidates_with_url": with_urls,
+ "source_counts": source_counts,
+ "elapsed_ms": round(float(elapsed_ms), 1),
+ }
+
+
+@lru_cache(maxsize=8)
+def _collect_emoticon_download_catalog_cached(
+ account_dir_str: str,
+ fingerprint: str,
+) -> tuple[dict[str, dict[str, Any]], dict[str, Any]]:
+ started_at = datetime.datetime.now().timestamp()
+ account_dir = Path(account_dir_str)
+ catalog: dict[str, dict[str, Any]] = {}
+ stats: dict[str, Any] = {}
+ emoticon_primary: set[str] = set()
+ emoticon_extern: set[str] = set()
+ emoticon_with_remote: set[str] = set()
+ message_md5: set[str] = set()
+ message_md5_with_url: set[str] = set()
+ message_extern_md5: set[str] = set()
+ builtin_expr_ids: set[int] = set()
+ builtin_expr_rows = 0
+ message_rows = 0
+ message_table_count = 0
+
+ db_path = account_dir / "emoticon.db"
+ if db_path.exists():
+ try:
+ conn = sqlite3.connect(str(db_path))
+ except Exception as exc:
+ conn = None
+ logger.warning("[media] emoticon_catalog emoticon_db_open_failed: account=%s error=%s", account_dir.name, exc)
+ if conn is None:
+ rows = []
+ else:
+ rows = None
+ if conn is not None:
+ conn.row_factory = sqlite3.Row
+ if conn is not None:
+ try:
+ rows = conn.execute(
+ "SELECT md5, extern_md5, aes_key, cdn_url, encrypt_url, extern_url, thumb_url, tp_url "
+ "FROM kNonStoreEmoticonTable ORDER BY rowid DESC"
+ ).fetchall()
+ except Exception as exc:
+ logger.warning(
+ "[media] emoticon_catalog emoticon_db_scan_failed: account=%s error=%s",
+ account_dir.name,
+ exc,
+ )
+ rows = []
+ finally:
+ try:
+ conn.close()
+ except Exception:
+ pass
+ stats["emoticon_db_rows"] = len(rows or [])
+ for row in rows or []:
+ urls = [
+ str(row[key] or "").strip()
+ for key in ("cdn_url", "extern_url", "thumb_url", "tp_url", "encrypt_url")
+ if str(row[key] or "").strip() and _is_safe_http_url(str(row[key] or "").strip())
+ ]
+ aes_key = str(row["aes_key"] or "").strip()
+ md5s = _normalize_emoticon_md5(row["md5"])
+ extern_md5 = _normalize_emoticon_md5(row["extern_md5"])
+ if md5s:
+ emoticon_primary.add(md5s)
+ if urls:
+ emoticon_with_remote.add(md5s)
+ _merge_emoticon_candidate(catalog, md5s, urls=urls, aes_key=aes_key, source="emoticon_db_md5")
+ if extern_md5:
+ emoticon_extern.add(extern_md5)
+ if urls:
+ emoticon_with_remote.add(extern_md5)
+ _merge_emoticon_candidate(
+ catalog,
+ extern_md5,
+ urls=urls,
+ aes_key=aes_key,
+ source="emoticon_db_extern_md5",
+ )
+
+ message_db_paths = _emoticon_message_db_paths(account_dir)
+ for message_db_path in message_db_paths:
+ try:
+ conn = sqlite3.connect(str(message_db_path))
+ except Exception as exc:
+ logger.warning(
+ "[media] emoticon_catalog message_db_open_failed: account=%s db=%s error=%s",
+ account_dir.name,
+ message_db_path.name,
+ exc,
+ )
+ continue
+ conn.row_factory = sqlite3.Row
+ try:
+ for table_name in _list_emoticon_message_tables(conn):
+ message_table_count += 1
+ quoted = _quote_sqlite_ident(table_name)
+ try:
+ rows = conn.execute(
+ f"SELECT compress_content, message_content, packed_info_data FROM {quoted} WHERE local_type = 47"
+ )
+ except Exception:
+ continue
+
+ for row in rows:
+ message_rows += 1
+ try:
+ builtin_id = _extract_emoticon_builtin_expr_id(row["packed_info_data"])
+ except Exception:
+ builtin_id = None
+ if builtin_id is not None:
+ builtin_expr_rows += 1
+ builtin_expr_ids.add(int(builtin_id))
+
+ try:
+ raw_text = _decode_message_content(row["compress_content"], row["message_content"])
+ except Exception:
+ raw_text = ""
+ md5s = _extract_emoticon_message_md5(raw_text)
+ if not md5s:
+ continue
+ message_md5.add(md5s)
+
+ extern_md5 = _extract_emoticon_message_extern_md5(raw_text)
+ if extern_md5:
+ message_extern_md5.add(extern_md5)
+
+ if md5s in message_md5_with_url:
+ continue
+
+ urls = _extract_emoticon_message_urls(raw_text)
+ if not urls:
+ continue
+ message_md5_with_url.add(md5s)
+ _merge_emoticon_candidate(
+ catalog,
+ md5s,
+ urls=urls,
+ aes_key=_extract_emoticon_message_aes_key(raw_text),
+ source="message_xml",
+ )
+ except Exception as exc:
+ logger.warning(
+ "[media] emoticon_catalog message_db_scan_failed: account=%s db=%s error=%s",
+ account_dir.name,
+ message_db_path.name,
+ exc,
+ )
+ finally:
+ try:
+ conn.close()
+ except Exception:
+ pass
+
+ stats.update(
+ {
+ "fingerprint": fingerprint,
+ "emoticon_db_md5": len(emoticon_primary),
+ "emoticon_db_extern_md5": len(emoticon_extern),
+ "emoticon_db_with_remote": len(emoticon_with_remote),
+ "message_db_count": len(message_db_paths),
+ "message_table_count": message_table_count,
+ "message_xml_rows": message_rows,
+ "message_xml_md5": len(message_md5),
+ "message_xml_md5_with_url": len(message_md5_with_url),
+ "message_xml_extern_md5": len(message_extern_md5),
+ "message_builtin_expr_ids": len(builtin_expr_ids),
+ "message_builtin_expr_rows": builtin_expr_rows,
+ }
+ )
+ elapsed_ms = (datetime.datetime.now().timestamp() - started_at) * 1000.0
+ public_stats = _emoticon_catalog_public_stats(stats, catalog, elapsed_ms=elapsed_ms)
+ logger.info(
+ "[media] emoticon_catalog scan_done: account=%s total_candidates=%s source_counts=%s "
+ "emoticon_db_rows=%s emoticon_db_md5=%s emoticon_db_extern_md5=%s message_rows=%s "
+ "message_md5=%s message_md5_with_url=%s message_extern_md5=%s builtin_expr_ids=%s elapsed_ms=%s",
+ account_dir.name,
+ public_stats["total_candidates"],
+ public_stats["source_counts"],
+ public_stats["emoticon_db_rows"],
+ public_stats["emoticon_db_md5"],
+ public_stats["emoticon_db_extern_md5"],
+ public_stats["message_xml_rows"],
+ public_stats["message_xml_md5"],
+ public_stats["message_xml_md5_with_url"],
+ public_stats["message_xml_extern_md5"],
+ public_stats["message_builtin_expr_ids"],
+ public_stats["elapsed_ms"],
+ )
+ return catalog, public_stats
+
+
+def _collect_emoticon_download_catalog(account_dir: Path) -> tuple[dict[str, dict[str, Any]], dict[str, Any]]:
+ fingerprint = _emoticon_source_fingerprint(Path(account_dir))
+ return _collect_emoticon_download_catalog_cached(str(Path(account_dir)), fingerprint)
+
+
+def _collect_emoticon_download_candidates(account_dir: Path) -> list[str]:
+ catalog, _stats = _collect_emoticon_download_catalog(Path(account_dir))
+ return list(catalog.keys())
+
+
+def _find_emoticon_message_remote_source(account_dir: Path, md5: str) -> dict[str, Any]:
+ md5s = _normalize_emoticon_md5(md5)
+ if not md5s:
+ return {}
+
+ for message_db_path in _emoticon_message_db_paths(Path(account_dir)):
+ try:
+ conn = sqlite3.connect(str(message_db_path))
+ except Exception:
+ continue
+ conn.row_factory = sqlite3.Row
+ try:
+ for table_name in _list_emoticon_message_tables(conn):
+ quoted = _quote_sqlite_ident(table_name)
+ try:
+ rows = conn.execute(
+ f"SELECT compress_content, message_content FROM {quoted} WHERE local_type = 47"
+ )
+ except Exception:
+ continue
+
+ for row in rows:
+ try:
+ raw_text = _decode_message_content(row["compress_content"], row["message_content"])
+ except Exception:
+ raw_text = ""
+ if _extract_emoticon_message_md5(raw_text) != md5s:
+ continue
+ urls = _extract_emoticon_message_urls(raw_text)
+ if not urls:
+ continue
+ aes_key = _extract_emoticon_message_aes_key(raw_text)
+ out = {"md5": md5s, "urls": urls, "aes_keys": [], "sources": ["message_xml"]}
+ if aes_key:
+ out["aes_keys"].append(aes_key)
+ return out
+ except Exception:
+ continue
+ finally:
+ try:
+ conn.close()
+ except Exception:
+ pass
+ return {}
+
+
+def _try_fetch_emoticon_from_sources(urls: list[str], aes_keys: list[str]) -> tuple[Optional[bytes], Optional[str]]:
for url in urls:
try:
payload = _download_http_bytes(url)
@@ -441,9 +953,10 @@ def _try_fetch_emoticon_from_remote(account_dir: Path, md5: str) -> tuple[Option
continue
candidates: list[bytes] = [payload]
- dec = _decrypt_emoticon_aes_cbc(payload, aes_key_hex)
- if dec is not None:
- candidates.insert(0, dec)
+ for aes_key_hex in aes_keys:
+ dec = _decrypt_emoticon_aes_cbc(payload, aes_key_hex)
+ if dec is not None:
+ candidates.insert(0, dec)
for data in candidates:
if not data:
@@ -470,6 +983,59 @@ def _try_fetch_emoticon_from_remote(account_dir: Path, md5: str) -> tuple[Option
return None, None
+def _try_fetch_emoticon_from_remote(
+ account_dir: Path,
+ md5: str,
+ source: Optional[dict[str, Any]] = None,
+) -> tuple[Optional[bytes], Optional[str]]:
+ md5s = _normalize_emoticon_md5(md5)
+ if not md5s:
+ return None, None
+
+ urls: list[str] = []
+ aes_keys: list[str] = []
+
+ if source:
+ for u in source.get("urls") or []:
+ u = str(u or "").strip()
+ if u and u not in urls and _is_safe_http_url(u):
+ urls.append(u)
+ for key in source.get("aes_keys") or []:
+ key = _normalize_emoticon_aes_key(key)
+ if key and key not in aes_keys:
+ aes_keys.append(key)
+ else:
+ info = _lookup_emoticon_info(str(account_dir), md5s)
+ if info:
+ for key in ("cdn_url", "extern_url", "thumb_url", "tp_url", "encrypt_url"):
+ u = str(info.get(key) or "").strip()
+ if u and u not in urls and _is_safe_http_url(u):
+ urls.append(u)
+ aes_key = _normalize_emoticon_aes_key(info.get("aes_key"))
+ if aes_key:
+ aes_keys.append(aes_key)
+
+ data, media_type = _try_fetch_emoticon_from_sources(urls, aes_keys)
+ if data is not None and media_type:
+ return data, media_type
+
+ if source:
+ return None, None
+
+ message_source = _find_emoticon_message_remote_source(Path(account_dir), md5s)
+ if not message_source:
+ return None, None
+
+ message_urls = [str(u or "").strip() for u in message_source.get("urls") or []]
+ message_aes_keys = [
+ _normalize_emoticon_aes_key(key) for key in (message_source.get("aes_keys") or []) if key
+ ]
+ return _try_fetch_emoticon_from_sources(
+ [u for u in message_urls if u and _is_safe_http_url(u)],
+ [k for k in message_aes_keys if k],
+ )
+
+
class _WxAMConfig(ctypes.Structure):
_fields_ = [
("mode", ctypes.c_int),
@@ -711,6 +1277,1219 @@ def _resolve_hardlink_dir2id_table_name(conn: sqlite3.Connection) -> Optional[st
return str(rows[0][0]) if rows[0] and rows[0][0] else None
+@dataclass(slots=True)
+class _HardlinkEntry:
+ file_name: str
+ file_size: int
+ modify_time: int
+ dir1: int
+ dir2: int
+ dir_name: str
+
+
+def _iter_files_under(root: Path):
+ try:
+ root_str = str(root)
+ except Exception:
+ return
+
+ for current_root, _dirnames, filenames in os.walk(root_str):
+ for filename in filenames:
+ try:
+ yield Path(current_root) / filename
+ except Exception:
+ continue
+
+
+def _iter_media_lookup_keys(name: str) -> list[str]:
+ lower_name = str(name or "").strip().lower()
+ if not lower_name:
+ return []
+
+ stem = Path(lower_name).stem
+ keys: list[str] = []
+ for value in (lower_name, stem):
+ if value and value not in keys:
+ keys.append(value)
+
+ stripped = _MEDIA_INDEX_STRIP_SUFFIX_RE.sub("", stem)
+ if stripped and stripped not in keys:
+ keys.append(stripped)
+
+ return keys
+
+
+def _iter_md5_candidates_from_name(name: str) -> list[str]:
+ candidates: list[str] = []
+ for key in _iter_media_lookup_keys(name):
+ if _EMOTICON_MD5_RE.fullmatch(key) and key not in candidates:
+ candidates.append(key)
+ return candidates
+
+
+def _build_hardlink_dir2id_map(conn: sqlite3.Connection) -> dict[int, str]:
+ table_name = _resolve_hardlink_dir2id_table_name(conn)
+ if not table_name:
+ return {}
+
+ quoted = _quote_ident(table_name)
+ mapping: dict[int, str] = {}
+ try:
+ rows = conn.execute(f"SELECT rowid, username FROM {quoted}").fetchall()
+ except Exception:
+ return {}
+
+ for rowid, username in rows:
+ try:
+ rid = int(rowid)
+ except Exception:
+ continue
+ text = str(username or "").strip()
+ if text:
+ mapping[rid] = text
+ return mapping
+
+
+def _resolve_hardlink_entry_path(
+ *,
+ kind: str,
+ entry: _HardlinkEntry,
+ wxid_dir: Path,
+ username: Optional[str],
+ extra_roots: Optional[list[Path]] = None,
+) -> Optional[Path]:
+ kind_key = str(kind or "").lower().strip()
+ file_name = str(entry.file_name or "").strip()
+ if not file_name:
+ return None
+
+ roots: list[Path] = []
+ for root in [wxid_dir] + list(extra_roots or []):
+ if not root:
+ continue
+ try:
+ resolved = root.resolve()
+ except Exception:
+ resolved = root
+ if resolved not in roots:
+ roots.append(resolved)
+
+ if not roots:
+ return None
+
+ if kind_key in {"video", "video_thumb"}:
+ guessed_month: Optional[str] = None
+ if entry.modify_time and entry.modify_time > 0:
+ try:
+ dt = datetime.datetime.fromtimestamp(int(entry.modify_time))
+ guessed_month = f"{dt.year:04d}-{dt.month:02d}"
+ except Exception:
+ guessed_month = None
+
+ if re.fullmatch(r"\d{4}-\d{2}", str(entry.dir_name or "").strip()):
+ guessed_month = str(entry.dir_name or "").strip()
+
+ stem = Path(file_name).stem
+ if kind_key == "video":
+ file_variants = [file_name]
+ else:
+ file_variants = [
+ f"{stem}_thumb.jpg",
+ f"{stem}_thumb.jpeg",
+ f"{stem}_thumb.png",
+ f"{stem}_thumb.webp",
+ f"{stem}.jpg",
+ f"{stem}.jpeg",
+ f"{stem}.png",
+ f"{stem}.gif",
+ f"{stem}.webp",
+ f"{stem}.dat",
+ file_name,
+ ]
+
+ def _iter_video_base_dirs(root: Path) -> list[Path]:
+ bases: list[Path] = []
+ candidates = [
+ root / "msg" / "video",
+ root / "video",
+ root if str(root.name).lower() == "video" else None,
+ ]
+ for candidate in candidates:
+ if not candidate:
+ continue
+ try:
+ if candidate.exists() and candidate.is_dir():
+ bases.append(candidate)
+ except Exception:
+ continue
+
+ seen: set[str] = set()
+ uniq: list[Path] = []
+ for base in bases:
+ try:
+ token = str(base.resolve())
+ except Exception:
+ token = str(base)
+ if token in seen:
+ continue
+ seen.add(token)
+ uniq.append(base)
+ return uniq
+
+ for root in roots:
+ for base_dir in _iter_video_base_dirs(root):
+ dirs_to_check: list[Path] = []
+ if guessed_month:
+ dirs_to_check.append(base_dir / guessed_month)
+ dirs_to_check.append(base_dir)
+ for directory in dirs_to_check:
+ try:
+ if not directory.exists() or not directory.is_dir():
+ continue
+ except Exception:
+ continue
+ for variant in file_variants:
+ path = directory / variant
+ try:
+ if path.exists() and path.is_file():
+ return path
+ except Exception:
+ continue
+ return None
+
+ if kind_key == "file":
+ file_size = int(entry.file_size) if int(entry.file_size or 0) > 0 else None
+ guessed_month: Optional[str] = None
+ if entry.modify_time and entry.modify_time > 0:
+ try:
+ dt = datetime.datetime.fromtimestamp(int(entry.modify_time))
+ guessed_month = f"{dt.year:04d}-{dt.month:02d}"
+ except Exception:
+ guessed_month = None
+
+ file_base_dirs: list[Path] = []
+ for root in roots:
+ candidates = [
+ root / "msg" / "file",
+ root / "file" if root.name.lower() == "msg" else None,
+ root if root.name.lower() == "file" else None,
+ ]
+ for candidate in candidates:
+ if not candidate:
+ continue
+ try:
+ if candidate.exists() and candidate.is_dir() and candidate not in file_base_dirs:
+ file_base_dirs.append(candidate)
+ except Exception:
+ continue
+
+ if not file_base_dirs:
+ return None
+
+ file_stem = Path(file_name).stem
+
+ def _iter_month_dirs(base: Path) -> list[Path]:
+ result: list[Path] = []
+ try:
+ for child in base.iterdir():
+ try:
+ if not child.is_dir():
+ continue
+ except Exception:
+ continue
+ name = str(child.name)
+ if re.fullmatch(r"\d{4}-\d{2}", name):
+ result.append(child)
+ except Exception:
+ return []
+ return sorted(result, key=lambda item: str(item.name))
+
+ def _pick_best_hit(hits: list[Path]) -> Optional[Path]:
+ if not hits:
+ return None
+ if file_size is not None and file_size >= 0:
+ for hit in hits:
+ try:
+ if hit.stat().st_size == file_size:
+ return hit
+ except Exception:
+ continue
+ return hits[0]
+
+ for base in file_base_dirs:
+ month_dirs = _iter_month_dirs(base)
+ month_names: list[str] = []
+ if guessed_month:
+ month_names.append(guessed_month)
+ for directory in month_dirs:
+ name = str(directory.name)
+ if name not in month_names:
+ month_names.append(name)
+
+ for month_name in month_names:
+ month_dir = base / month_name
+ try:
+ if not (month_dir.exists() and month_dir.is_dir()):
+ continue
+ except Exception:
+ continue
+
+ direct = month_dir / file_name
+ try:
+ if direct.exists() and direct.is_file():
+ return direct
+ except Exception:
+ pass
+
+ in_stem_dir = month_dir / file_stem / file_name
+ try:
+ if in_stem_dir.exists() and in_stem_dir.is_file():
+ return in_stem_dir
+ except Exception:
+ pass
+ return None
+
+ dir_name = str(entry.dir_name or "").strip()
+ file_stem = Path(file_name).stem
+ file_variants = [file_name, f"{file_stem}_h.dat", f"{file_stem}_t.dat"]
+
+ for root in roots:
+ if entry.dir1 and dir_name:
+ for variant in file_variants:
+ direct = (root / str(entry.dir1) / dir_name / variant).resolve()
+ try:
+ if direct.exists() and direct.is_file():
+ return direct
+ except Exception:
+ continue
+
+ if username:
+ chat_hash = hashlib.md5(str(username).encode()).hexdigest()
+ for variant in file_variants:
+ attach = (root / "msg" / "attach" / chat_hash / dir_name / "Img" / variant).resolve()
+ try:
+ if attach.exists() and attach.is_file():
+ return attach
+ except Exception:
+ continue
+ return None
+
+
+class MediaPathIndex:
+ def __init__(
+ self,
+ *,
+ account_dir: Path,
+ usernames: Optional[Iterable[str]] = None,
+ media_kinds: Optional[Iterable[str]] = None,
+ ) -> None:
+ self.account_dir = account_dir
+ self.usernames = list(dict.fromkeys([str(item or "").strip() for item in (usernames or []) if str(item or "").strip()]))
+ self.media_kinds = {
+ str(kind or "").strip()
+ for kind in (media_kinds or [])
+ if str(kind or "").strip() in {"image", "emoji", "video", "video_thumb", "file"}
+ }
+ self.wxid_dir = _resolve_account_wxid_dir(account_dir)
+ self.db_storage_dir = _resolve_account_db_storage_dir(account_dir)
+ self.resource_dir = _get_resource_dir(account_dir)
+ scope_text = "\n".join(sorted(self.usernames)) or "__all__"
+ self._scope_key = hashlib.sha1(scope_text.encode("utf-8", errors="ignore")).hexdigest()
+ self._cache_db_path = self.account_dir / "media_path_index.db"
+ self._loaded_from_cache = False
+
+ self._roots: list[Path] = []
+ for root in [self.wxid_dir, self.db_storage_dir]:
+ if not root:
+ continue
+ try:
+ resolved = root.resolve()
+ except Exception:
+ resolved = root
+ if resolved not in self._roots:
+ self._roots.append(resolved)
+
+ self._md5_hits: dict[str, dict[str, Path]] = {
+ "image": {},
+ "emoji": {},
+ "video": {},
+ "video_thumb": {},
+ "file": {},
+ }
+ self._file_id_hits: dict[str, dict[str, Path]] = {
+ "image": {},
+ "emoji": {},
+ "video": {},
+ "video_thumb": {},
+ "file": {},
+ }
+ self._user_file_id_hits: dict[str, dict[tuple[str, str], Path]] = {
+ "image": {},
+ "emoji": {},
+ "video": {},
+ "video_thumb": {},
+ "file": {},
+ }
+ self._hardlink_hits: dict[str, dict[str, _HardlinkEntry]] = {
+ "image": {},
+ "emoji": {},
+ "video": {},
+ "video_thumb": {},
+ "file": {},
+ }
+ self._query_cache: dict[tuple[str, str, str, str], Optional[Path]] = {}
+ self._negative_cache: set[tuple[str, str, str, str]] = set()
+ self._known_missing: set[tuple[str, str, str, str]] = set()
+ self.stats = {
+ "resourceFiles": 0,
+ "hardlinkRows": 0,
+ "scannedFiles": 0,
+ "md5Keys": 0,
+ "fileIdKeys": 0,
+ "loadedEntries": 0,
+ "loadedMisses": 0,
+ }
+
+ @classmethod
+ def build(
+ cls,
+ *,
+ account_dir: Path,
+ usernames: Optional[Iterable[str]] = None,
+ media_kinds: Optional[Iterable[str]] = None,
+ ) -> "MediaPathIndex":
+ index = cls(account_dir=account_dir, usernames=usernames, media_kinds=media_kinds)
+ index._build()
+ return index
+
+ def _wants(self, kind: str) -> bool:
+ if not self.media_kinds:
+ return True
+ return str(kind or "").strip() in self.media_kinds
+
+ def _put_md5(self, kind: str, md5: str, path: Path) -> None:
+ bucket = self._md5_hits.setdefault(kind, {})
+ if md5 and md5 not in bucket:
+ bucket[md5] = path
+ self.stats["md5Keys"] += 1
+
+ def _put_file_id(self, kind: str, key: str, path: Path, username: str = "") -> None:
+ if not key:
+ return
+ bucket = self._file_id_hits.setdefault(kind, {})
+ if key not in bucket:
+ bucket[key] = path
+ self.stats["fileIdKeys"] += 1
+ user_key = str(username or "").strip()
+ if user_key:
+ ub = self._user_file_id_hits.setdefault(kind, {})
+ ub.setdefault((user_key, key), path)
+
+ def _register_kind_path(self, kind: str, path: Path, *, username: str = "") -> None:
+ name = str(path.name or "").strip()
+ if not name:
+ return
+ for md5 in _iter_md5_candidates_from_name(name):
+ self._put_md5(kind, md5, path)
+ for key in _iter_media_lookup_keys(name):
+ self._put_file_id(kind, key, path, username=username)
+
+ def _normalize_cache_key(
+ self,
+ *,
+ kind: str,
+ md5: str = "",
+ file_id: str = "",
+ username: str = "",
+ ) -> tuple[str, str, str, str]:
+ return (
+ str(kind or "").strip().lower(),
+ str(md5 or "").strip().lower(),
+ str(file_id or "").strip().lower(),
+ str(username or "").strip(),
+ )
+
+ def is_known_missing(
+ self,
+ *,
+ kind: str,
+ md5: str = "",
+ file_id: str = "",
+ username: str = "",
+ ) -> bool:
+ cache_key = self._normalize_cache_key(kind=kind, md5=md5, file_id=file_id, username=username)
+ return cache_key in self._known_missing
+
+ def _drop_cached_miss_for_path(self, *, kind: str, path: Path, username: str = "") -> list[tuple[str, str, str, str]]:
+ kind_key = str(kind or "").strip().lower()
+ username_key = str(username or "").strip()
+ md5_values = set(_iter_md5_candidates_from_name(path.name))
+ file_keys = set(_iter_media_lookup_keys(path.name))
+ if not kind_key or (not md5_values and not file_keys):
+ return []
+
+ stale_keys = [
+ cache_key
+ for cache_key in self._known_missing
+ if cache_key[0] == kind_key
+ and cache_key[3] == username_key
+ and ((cache_key[1] and cache_key[1] in md5_values) or (cache_key[2] and cache_key[2] in file_keys))
+ ]
+ for cache_key in stale_keys:
+ self._known_missing.discard(cache_key)
+ self._negative_cache.discard(cache_key)
+ self._query_cache.pop(cache_key, None)
+ return stale_keys
+
+ def _persist_entry_rows(self, rows: list[tuple[str, str, str, str, str, str]]) -> None:
+ if not rows:
+ return
+ try:
+ conn = sqlite3.connect(str(self._cache_db_path))
+ except Exception:
+ return
+
+ try:
+ self._ensure_cache_schema(conn)
+ with conn:
+ conn.executemany(
+ "INSERT OR REPLACE INTO media_index_entries(scope, kind, key_type, key, username, path) VALUES (?, ?, ?, ?, ?, ?)",
+ rows,
+ )
+ except Exception:
+ logger.exception("[media-index] persist entry rows failed account=%s", str(self.account_dir.name or ""))
+ finally:
+ conn.close()
+
+ def _persist_missing_rows(self, rows: list[tuple[str, str, str, str, str]]) -> None:
+ if not rows:
+ return
+ try:
+ conn = sqlite3.connect(str(self._cache_db_path))
+ except Exception:
+ return
+
+ try:
+ self._ensure_cache_schema(conn)
+ with conn:
+ conn.executemany(
+ "INSERT OR REPLACE INTO media_index_misses(scope, kind, md5, file_id, username) VALUES (?, ?, ?, ?, ?)",
+ rows,
+ )
+ except Exception:
+ logger.exception("[media-index] persist miss rows failed account=%s", str(self.account_dir.name or ""))
+ finally:
+ conn.close()
+
+ def _delete_missing_rows(self, rows: list[tuple[str, str, str, str, str]]) -> None:
+ if not rows:
+ return
+ try:
+ conn = sqlite3.connect(str(self._cache_db_path))
+ except Exception:
+ return
+
+ try:
+ self._ensure_cache_schema(conn)
+ with conn:
+ conn.executemany(
+ "DELETE FROM media_index_misses WHERE scope = ? AND kind = ? AND md5 = ? AND file_id = ? AND username = ?",
+ rows,
+ )
+ except Exception:
+ logger.exception("[media-index] delete miss rows failed account=%s", str(self.account_dir.name or ""))
+ finally:
+ conn.close()
+
+ def remember_path(self, *, kind: str, path: Path, username: str = "") -> None:
+ kind_key = str(kind or "").strip().lower()
+ username_key = str(username or "").strip()
+ if not kind_key:
+ return
+ try:
+ path_obj = path if isinstance(path, Path) else Path(path)
+ except Exception:
+ return
+ name = str(path_obj.name or "").strip()
+ if not name:
+ return
+
+ self._register_kind_path(kind_key, path_obj, username=username_key)
+ stale_keys = self._drop_cached_miss_for_path(kind=kind_key, path=path_obj, username=username_key)
+
+ rows: list[tuple[str, str, str, str, str, str]] = []
+ for md5 in _iter_md5_candidates_from_name(name):
+ rows.append((self._scope_key, kind_key, "md5", md5, "", str(path_obj)))
+ for key in _iter_media_lookup_keys(name):
+ rows.append((self._scope_key, kind_key, "file_id", key, "", str(path_obj)))
+ if username_key:
+ rows.append((self._scope_key, kind_key, "file_id", key, username_key, str(path_obj)))
+ self._persist_entry_rows(rows)
+ self._delete_missing_rows(
+ [
+ (self._scope_key, stale_kind, stale_md5, stale_file_id, stale_username)
+ for stale_kind, stale_md5, stale_file_id, stale_username in stale_keys
+ ]
+ )
+
+ def mark_missing(
+ self,
+ *,
+ kind: str,
+ md5: str = "",
+ file_id: str = "",
+ username: str = "",
+ ) -> None:
+ cache_key = self._normalize_cache_key(kind=kind, md5=md5, file_id=file_id, username=username)
+ if not cache_key[0] or (not cache_key[1] and not cache_key[2]):
+ return
+ if cache_key in self._known_missing:
+ return
+ self._known_missing.add(cache_key)
+ self._negative_cache.add(cache_key)
+ self._query_cache[cache_key] = None
+ self._persist_missing_rows(
+ [
+ (
+ self._scope_key,
+ cache_key[0],
+ cache_key[1],
+ cache_key[2],
+ cache_key[3],
+ )
+ ]
+ )
+
+ def _build(self) -> None:
+ started_at = time.perf_counter()
+ if self._try_load_persisted():
+ logger.info(
+ "[media-index] loaded persisted account=%s usernames=%s kinds=%s md5Keys=%s fileIdKeys=%s loadedEntries=%s elapsedMs=%.1f",
+ str(self.account_dir.name or ""),
+ len(self.usernames),
+ ",".join(sorted(self.media_kinds)) if self.media_kinds else "all",
+ int(self.stats["md5Keys"]),
+ int(self.stats["fileIdKeys"]),
+ int(self.stats["loadedEntries"]),
+ (time.perf_counter() - started_at) * 1000.0,
+ )
+ return
+ self._index_decrypted_resources()
+ self._load_hardlink_index()
+ self._scan_media_roots()
+ self._persist()
+ logger.info(
+ "[media-index] built account=%s usernames=%s kinds=%s resourceFiles=%s hardlinkRows=%s scannedFiles=%s md5Keys=%s fileIdKeys=%s elapsedMs=%.1f",
+ str(self.account_dir.name or ""),
+ len(self.usernames),
+ ",".join(sorted(self.media_kinds)) if self.media_kinds else "all",
+ int(self.stats["resourceFiles"]),
+ int(self.stats["hardlinkRows"]),
+ int(self.stats["scannedFiles"]),
+ int(self.stats["md5Keys"]),
+ int(self.stats["fileIdKeys"]),
+ (time.perf_counter() - started_at) * 1000.0,
+ )
+
+ def _ensure_cache_schema(self, conn: sqlite3.Connection) -> None:
+ conn.executescript(
+ """
+ CREATE TABLE IF NOT EXISTS media_index_meta (
+ scope TEXT NOT NULL,
+ key TEXT NOT NULL,
+ value TEXT NOT NULL,
+ PRIMARY KEY (scope, key)
+ );
+ CREATE TABLE IF NOT EXISTS media_index_entries (
+ scope TEXT NOT NULL,
+ kind TEXT NOT NULL,
+ key_type TEXT NOT NULL,
+ key TEXT NOT NULL,
+ username TEXT NOT NULL DEFAULT '',
+ path TEXT NOT NULL,
+ PRIMARY KEY (scope, kind, key_type, key, username)
+ );
+ CREATE INDEX IF NOT EXISTS idx_media_index_entries_lookup
+ ON media_index_entries(scope, kind, key_type, key, username);
+ CREATE TABLE IF NOT EXISTS media_index_misses (
+ scope TEXT NOT NULL,
+ kind TEXT NOT NULL,
+ md5 TEXT NOT NULL DEFAULT '',
+ file_id TEXT NOT NULL DEFAULT '',
+ username TEXT NOT NULL DEFAULT '',
+ PRIMARY KEY (scope, kind, md5, file_id, username)
+ );
+ CREATE INDEX IF NOT EXISTS idx_media_index_misses_lookup
+ ON media_index_misses(scope, kind, md5, file_id, username);
+ """
+ )
+
+ def _iter_signature_targets(self) -> list[tuple[str, Path, int]]:
+ targets: list[tuple[str, Path, int]] = []
+ hardlink_db_path = self.account_dir / "hardlink.db"
+ if hardlink_db_path.exists():
+ targets.append(("hardlink.db", hardlink_db_path, 0))
+
+ try:
+ if self.resource_dir.exists() and self.resource_dir.is_dir():
+ targets.append(("resource", self.resource_dir, 1))
+ except Exception:
+ pass
+
+ for username, directory in self._iter_attach_scan_dirs():
+ targets.append((f"attach:{username or '*'}:{directory.name}", directory, 3))
+ for directory in self._iter_video_scan_dirs():
+ targets.append((f"video:{directory.name}", directory, 2))
+ for directory in self._iter_file_scan_dirs():
+ targets.append((f"file:{directory.name}", directory, 2))
+ for directory in self._iter_cache_scan_dirs():
+ targets.append((f"cache:{directory.name}", directory, 3))
+ return targets
+
+ def _snapshot_path(self, path: Path, max_depth: int) -> list[tuple[str, int, int, int]]:
+ try:
+ if not path.exists():
+ return [(".", -1, 0, 0)]
+ except Exception:
+ return [(".", -1, 0, 0)]
+
+ try:
+ if path.is_file():
+ stat = path.stat()
+ return [(".", int(getattr(stat, "st_mtime_ns", int(stat.st_mtime * 1_000_000_000))), int(stat.st_size), 0)]
+ except Exception:
+ return [(".", -2, 0, 0)]
+
+ rows: list[tuple[str, int, int, int]] = []
+ root_str = str(path)
+ for current_root, dirnames, _filenames in os.walk(root_str):
+ rel = os.path.relpath(current_root, root_str)
+ if rel == ".":
+ depth = 0
+ rel_key = "."
+ else:
+ depth = rel.count(os.sep) + 1
+ rel_key = rel.replace("\\", "/")
+ try:
+ stat = os.stat(current_root)
+ mtime_ns = int(getattr(stat, "st_mtime_ns", int(stat.st_mtime * 1_000_000_000)))
+ except Exception:
+ mtime_ns = -1
+ rows.append((rel_key, mtime_ns, len(dirnames), depth))
+ dirnames.sort()
+ if depth >= max_depth:
+ dirnames[:] = []
+ return rows
+
+ def _build_signature(self) -> str:
+ payload: list[Any] = [
+ ["version", _MEDIA_INDEX_DB_VERSION],
+ ["account", str(self.account_dir.name or "")],
+ ["scope", self._scope_key],
+ ["usernames", sorted(self.usernames)],
+ ["mediaKinds", sorted(self.media_kinds)],
+ ]
+ for label, path, max_depth in self._iter_signature_targets():
+ payload.append(
+ [
+ label,
+ str(path),
+ self._snapshot_path(path, max_depth=max_depth),
+ ]
+ )
+ raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":"), default=str)
+ return hashlib.sha256(raw.encode("utf-8", errors="ignore")).hexdigest()
+
+ def _iter_persist_rows(self):
+ for kind, bucket in self._md5_hits.items():
+ for key, path in bucket.items():
+ yield (self._scope_key, kind, "md5", key, "", str(path))
+ for kind, bucket in self._file_id_hits.items():
+ for key, path in bucket.items():
+ yield (self._scope_key, kind, "file_id", key, "", str(path))
+ for kind, bucket in self._user_file_id_hits.items():
+ for (username, key), path in bucket.items():
+ yield (self._scope_key, kind, "file_id", key, str(username or ""), str(path))
+
+ def _iter_persist_missing_rows(self):
+ for kind, md5, file_id, username in sorted(self._known_missing):
+ yield (self._scope_key, kind, md5, file_id, username)
+
+ def _persist(self) -> None:
+ try:
+ conn = sqlite3.connect(str(self._cache_db_path))
+ except Exception:
+ return
+
+ try:
+ self._ensure_cache_schema(conn)
+ signature = self._build_signature()
+ meta_rows = [
+ (self._scope_key, "version", str(_MEDIA_INDEX_DB_VERSION)),
+ (self._scope_key, "signature", signature),
+ (self._scope_key, "usernames", json.dumps(sorted(self.usernames), ensure_ascii=False)),
+ (self._scope_key, "mediaKinds", json.dumps(sorted(self.media_kinds), ensure_ascii=False)),
+ (self._scope_key, "resourceFiles", str(int(self.stats["resourceFiles"]))),
+ (self._scope_key, "hardlinkRows", str(int(self.stats["hardlinkRows"]))),
+ (self._scope_key, "scannedFiles", str(int(self.stats["scannedFiles"]))),
+ (self._scope_key, "md5Keys", str(int(self.stats["md5Keys"]))),
+ (self._scope_key, "fileIdKeys", str(int(self.stats["fileIdKeys"]))),
+ ]
+ with conn:
+ conn.execute("DELETE FROM media_index_entries WHERE scope = ?", (self._scope_key,))
+ conn.execute("DELETE FROM media_index_misses WHERE scope = ?", (self._scope_key,))
+ conn.execute("DELETE FROM media_index_meta WHERE scope = ?", (self._scope_key,))
+ conn.executemany(
+ "INSERT OR REPLACE INTO media_index_entries(scope, kind, key_type, key, username, path) VALUES (?, ?, ?, ?, ?, ?)",
+ self._iter_persist_rows(),
+ )
+ conn.executemany(
+ "INSERT OR REPLACE INTO media_index_misses(scope, kind, md5, file_id, username) VALUES (?, ?, ?, ?, ?)",
+ self._iter_persist_missing_rows(),
+ )
+ conn.executemany(
+ "INSERT OR REPLACE INTO media_index_meta(scope, key, value) VALUES (?, ?, ?)",
+ meta_rows,
+ )
+ except Exception:
+ logger.exception("[media-index] persist failed account=%s", str(self.account_dir.name or ""))
+ finally:
+ conn.close()
+
+ def _try_load_persisted(self) -> bool:
+ try:
+ if not self._cache_db_path.exists():
+ return False
+ except Exception:
+ return False
+
+ try:
+ conn = sqlite3.connect(str(self._cache_db_path))
+ except Exception:
+ return False
+
+ try:
+ self._ensure_cache_schema(conn)
+ rows = conn.execute(
+ "SELECT key, value FROM media_index_meta WHERE scope = ?",
+ (self._scope_key,),
+ ).fetchall()
+ if not rows:
+ return False
+ meta = {str(key): str(value) for key, value in rows}
+ if meta.get("version") != str(_MEDIA_INDEX_DB_VERSION):
+ return False
+
+ stored_kinds_raw = str(meta.get("mediaKinds") or "[]")
+ try:
+ stored_kinds = set(json.loads(stored_kinds_raw))
+ except Exception:
+ stored_kinds = set()
+ if self.media_kinds and not self.media_kinds.issubset(stored_kinds):
+ return False
+
+ current_signature = self._build_signature()
+ if meta.get("signature") != current_signature:
+ return False
+
+ entry_rows = conn.execute(
+ "SELECT kind, key_type, key, username, path FROM media_index_entries WHERE scope = ?",
+ (self._scope_key,),
+ ).fetchall()
+ miss_rows = conn.execute(
+ "SELECT kind, md5, file_id, username FROM media_index_misses WHERE scope = ?",
+ (self._scope_key,),
+ ).fetchall()
+ if not entry_rows and not miss_rows:
+ return False
+
+ for kind, key_type, key, username, path in entry_rows:
+ kind_s = str(kind or "").strip()
+ key_type_s = str(key_type or "").strip()
+ key_s = str(key or "").strip().lower()
+ username_s = str(username or "").strip()
+ path_obj = Path(str(path or "").strip())
+ if not kind_s or not key_s:
+ continue
+ if key_type_s == "md5":
+ self._md5_hits.setdefault(kind_s, {})[key_s] = path_obj
+ elif key_type_s == "file_id":
+ self._file_id_hits.setdefault(kind_s, {}).setdefault(key_s, path_obj)
+ if username_s:
+ self._user_file_id_hits.setdefault(kind_s, {})[(username_s, key_s)] = path_obj
+
+ for kind, md5, file_id, username in miss_rows:
+ cache_key = self._normalize_cache_key(
+ kind=str(kind or ""),
+ md5=str(md5 or ""),
+ file_id=str(file_id or ""),
+ username=str(username or ""),
+ )
+ if not cache_key[0] or (not cache_key[1] and not cache_key[2]):
+ continue
+ self._known_missing.add(cache_key)
+ self._query_cache[cache_key] = None
+
+ self.stats["resourceFiles"] = int(meta.get("resourceFiles") or 0)
+ self.stats["hardlinkRows"] = int(meta.get("hardlinkRows") or 0)
+ self.stats["scannedFiles"] = int(meta.get("scannedFiles") or 0)
+ self.stats["md5Keys"] = sum(len(bucket) for bucket in self._md5_hits.values())
+ self.stats["fileIdKeys"] = sum(len(bucket) for bucket in self._file_id_hits.values())
+ self.stats["loadedEntries"] = len(entry_rows)
+ self.stats["loadedMisses"] = len(miss_rows)
+ self._loaded_from_cache = True
+ return True
+ except Exception:
+ logger.exception("[media-index] load persisted failed account=%s", str(self.account_dir.name or ""))
+ return False
+ finally:
+ conn.close()
+
+ def _index_decrypted_resources(self) -> None:
+ try:
+ if not self.resource_dir.exists() or not self.resource_dir.is_dir():
+ return
+ except Exception:
+ return
+
+ for path in _iter_files_under(self.resource_dir):
+ try:
+ if not path.is_file():
+ continue
+ except Exception:
+ continue
+
+ md5_values = _iter_md5_candidates_from_name(path.name)
+ if not md5_values:
+ continue
+
+ suffix = str(path.suffix or "").lower()
+ if suffix in _MEDIA_INDEX_VIDEO_STREAM_EXTS:
+ kinds = ("video",)
+ else:
+ kinds = tuple(kind for kind in ("image", "emoji", "video_thumb") if self._wants(kind))
+ if not kinds:
+ continue
+
+ for md5 in md5_values:
+ for kind in kinds:
+ self._put_md5(kind, md5, path)
+ self.stats["resourceFiles"] += 1
+
+ def _load_hardlink_index(self) -> None:
+ hardlink_db_path = self.account_dir / "hardlink.db"
+ if not hardlink_db_path.exists():
+ return
+
+ try:
+ conn = sqlite3.connect(str(hardlink_db_path))
+ conn.row_factory = sqlite3.Row
+ except Exception:
+ return
+
+ table_specs: list[tuple[str, tuple[str, ...]]] = []
+ if self._wants("image") or self._wants("emoji"):
+ table_specs.append(("image_hardlink_info", ("image", "emoji")))
+ if self._wants("video") or self._wants("video_thumb"):
+ table_specs.append(("video_hardlink_info", ("video", "video_thumb")))
+ if self._wants("file"):
+ table_specs.append(("file_hardlink_info", ("file",)))
+
+ try:
+ dir2id_map = _build_hardlink_dir2id_map(conn)
+ for prefix, kinds in table_specs:
+ table_name = _resolve_hardlink_table_name(conn, prefix)
+ if not table_name:
+ continue
+
+ quoted = _quote_ident(table_name)
+ try:
+ rows = conn.execute(
+ f"SELECT md5, file_name, file_size, modify_time, dir1, dir2 FROM {quoted} "
+ "WHERE md5 IS NOT NULL AND md5 <> '' ORDER BY modify_time DESC, rowid DESC"
+ ).fetchall()
+ except Exception:
+ continue
+
+ for row in rows:
+ md5 = str(row["md5"] or "").strip().lower()
+ if not _EMOTICON_MD5_RE.fullmatch(md5):
+ continue
+
+ entry = _HardlinkEntry(
+ file_name=str(row["file_name"] or "").strip(),
+ file_size=int(row["file_size"] or 0),
+ modify_time=int(row["modify_time"] or 0),
+ dir1=int(row["dir1"] or 0),
+ dir2=int(row["dir2"] or 0),
+ dir_name=str(dir2id_map.get(int(row["dir2"] or 0)) or str(row["dir2"] or "")).strip(),
+ )
+
+ for kind in kinds:
+ bucket = self._hardlink_hits.setdefault(kind, {})
+ bucket.setdefault(md5, entry)
+
+ self.stats["hardlinkRows"] += len(rows)
+ finally:
+ conn.close()
+
+ def _scan_media_roots(self) -> None:
+ if not self._roots:
+ return
+
+ if self._wants("image"):
+ for username, directory in self._iter_attach_scan_dirs():
+ self._scan_attach_dir(directory, username=username)
+
+ if self._wants("video") or self._wants("video_thumb"):
+ for directory in self._iter_video_scan_dirs():
+ self._scan_video_dir(directory)
+
+ if self._wants("file"):
+ for directory in self._iter_file_scan_dirs():
+ self._scan_file_dir(directory)
+
+ if self._wants("emoji") or self._wants("video_thumb"):
+ for directory in self._iter_cache_scan_dirs():
+ self._scan_cache_dir(directory)
+
+ def _iter_attach_scan_dirs(self) -> list[tuple[str, Path]]:
+ result: list[tuple[str, Path]] = []
+ usernames = self.usernames
+ for root in self._roots:
+ attach_root = root / "msg" / "attach"
+ try:
+ if not attach_root.exists() or not attach_root.is_dir():
+ continue
+ except Exception:
+ continue
+
+ if usernames:
+ for username in usernames:
+ chat_hash = hashlib.md5(username.encode()).hexdigest()
+ directory = attach_root / chat_hash
+ try:
+ if directory.exists() and directory.is_dir():
+ result.append((username, directory))
+ except Exception:
+ continue
+ else:
+ try:
+ for child in attach_root.iterdir():
+ try:
+ if child.is_dir():
+ result.append(("", child))
+ except Exception:
+ continue
+ except Exception:
+ continue
+ return result
+
+ def _iter_video_scan_dirs(self) -> list[Path]:
+ result: list[Path] = []
+ for root in self._roots:
+ candidates = [
+ root / "msg" / "video",
+ root / "video",
+ root if str(root.name).lower() == "video" else None,
+ ]
+ for candidate in candidates:
+ if not candidate:
+ continue
+ try:
+ if candidate.exists() and candidate.is_dir() and candidate not in result:
+ result.append(candidate)
+ except Exception:
+ continue
+ return result
+
+ def _iter_file_scan_dirs(self) -> list[Path]:
+ result: list[Path] = []
+ for root in self._roots:
+ candidates = [
+ root / "msg" / "file",
+ root / "file",
+ root if str(root.name).lower() == "file" else None,
+ ]
+ for candidate in candidates:
+ if not candidate:
+ continue
+ try:
+ if candidate.exists() and candidate.is_dir() and candidate not in result:
+ result.append(candidate)
+ except Exception:
+ continue
+ return result
+
+ def _iter_cache_scan_dirs(self) -> list[Path]:
+ result: list[Path] = []
+ for root in self._roots:
+ candidate = root / "cache"
+ try:
+ if candidate.exists() and candidate.is_dir() and candidate not in result:
+ result.append(candidate)
+ except Exception:
+ continue
+ return result
+
+ def _scan_attach_dir(self, directory: Path, *, username: str = "") -> None:
+ for path in _iter_files_under(directory):
+ suffix = str(path.suffix or "").lower()
+ if suffix not in _MEDIA_INDEX_FILE_EXTS:
+ continue
+ self.stats["scannedFiles"] += 1
+ if suffix in _MEDIA_INDEX_VIDEO_STREAM_EXTS:
+ if self._wants("video"):
+ self._register_kind_path("video", path, username=username)
+ continue
+ if self._wants("image"):
+ self._register_kind_path("image", path, username=username)
+
+ def _scan_video_dir(self, directory: Path) -> None:
+ for path in _iter_files_under(directory):
+ suffix = str(path.suffix or "").lower()
+ if suffix not in _MEDIA_INDEX_FILE_EXTS:
+ continue
+ self.stats["scannedFiles"] += 1
+ if suffix in _MEDIA_INDEX_VIDEO_STREAM_EXTS:
+ self._register_kind_path("video", path)
+ elif suffix == ".dat":
+ if self._wants("video"):
+ self._register_kind_path("video", path)
+ if self._wants("video_thumb"):
+ self._register_kind_path("video_thumb", path)
+ else:
+ self._register_kind_path("video_thumb", path)
+
+ def _scan_file_dir(self, directory: Path) -> None:
+ for path in _iter_files_under(directory):
+ self.stats["scannedFiles"] += 1
+ self._register_kind_path("file", path)
+ suffix = str(path.suffix or "").lower()
+ if suffix in _MEDIA_INDEX_VIDEO_STREAM_EXTS and self._wants("video"):
+ self._register_kind_path("video", path)
+
+ def _scan_cache_dir(self, directory: Path) -> None:
+ for path in _iter_files_under(directory):
+ suffix = str(path.suffix or "").lower()
+ if suffix not in _MEDIA_INDEX_FILE_EXTS:
+ continue
+ self.stats["scannedFiles"] += 1
+ lowered_parts = {str(part or "").lower() for part in path.parts}
+ if {"emoji", "emoticon"} & lowered_parts:
+ self._register_kind_path("emoji", path)
+ continue
+ if suffix in _MEDIA_INDEX_VIDEO_STREAM_EXTS:
+ self._register_kind_path("video", path)
+ continue
+ self._register_kind_path("video_thumb", path)
+
+ def resolve(self, *, kind: str, md5: str = "", file_id: str = "", username: str = "") -> Optional[Path]:
+ cache_key = self._normalize_cache_key(kind=kind, md5=md5, file_id=file_id, username=username)
+ kind_key, md5_key, file_key, username_key = cache_key
+ if cache_key in self._known_missing:
+ self._query_cache[cache_key] = None
+ return None
+ if cache_key in self._query_cache:
+ return self._query_cache[cache_key]
+ if cache_key in self._negative_cache:
+ return None
+
+ path: Optional[Path] = None
+ if md5_key and _EMOTICON_MD5_RE.fullmatch(md5_key):
+ path = self._resolve_by_md5(kind_key, md5_key, username_key)
+ if path is None and file_key:
+ path = self._resolve_by_file_id(kind_key, file_key, username_key)
+
+ if path is not None:
+ self._query_cache[cache_key] = path
+ return path
+
+ self._negative_cache.add(cache_key)
+ self._query_cache[cache_key] = None
+ return None
+
+ def _resolve_by_md5(self, kind: str, md5: str, username: str) -> Optional[Path]:
+ preferred: list[str]
+ if kind == "emoji":
+ preferred = ["emoji", "image"]
+ elif kind == "video_thumb":
+ preferred = ["video_thumb", "image"]
+ else:
+ preferred = [kind]
+
+ for candidate_kind in preferred:
+ path = self._md5_hits.get(candidate_kind, {}).get(md5)
+ if path is not None:
+ try:
+ if path.exists() and path.is_file():
+ return path
+ except Exception:
+ pass
+
+ for candidate_kind in preferred:
+ entry = self._hardlink_hits.get(candidate_kind, {}).get(md5)
+ if entry is None or not self.wxid_dir:
+ continue
+ path = _resolve_hardlink_entry_path(
+ kind=candidate_kind,
+ entry=entry,
+ wxid_dir=self.wxid_dir,
+ username=username or None,
+ extra_roots=self._roots[1:],
+ )
+ if path is None:
+ continue
+ self._register_kind_path(candidate_kind, path, username=username)
+ return path
+
+ if self.wxid_dir:
+ hardlink_db_path = self.account_dir / "hardlink.db"
+ for candidate_kind in preferred:
+ path = _resolve_media_path_from_hardlink(
+ hardlink_db_path=hardlink_db_path,
+ wxid_dir=self.wxid_dir,
+ md5=md5,
+ kind=candidate_kind,
+ username=username or None,
+ extra_roots=self._roots[1:],
+ )
+ if path is None:
+ continue
+ self._register_kind_path(candidate_kind, path, username=username)
+ return path
+ return None
+
+ def _resolve_by_file_id(self, kind: str, file_id: str, username: str) -> Optional[Path]:
+ keys = _iter_media_lookup_keys(file_id)
+ if not keys:
+ return None
+
+ if username:
+ user_bucket = self._user_file_id_hits.get(kind, {})
+ for key in keys:
+ path = user_bucket.get((username, key))
+ if path is None:
+ continue
+ try:
+ if path.exists() and path.is_file():
+ return path
+ except Exception:
+ continue
+
+ bucket = self._file_id_hits.get(kind, {})
+ for key in keys:
+ path = bucket.get(key)
+ if path is None:
+ continue
+ try:
+ if path.exists() and path.is_file():
+ return path
+ except Exception:
+ continue
+ return None
+
+
def _resolve_media_path_from_hardlink(
hardlink_db_path: Path,
wxid_dir: Path,
@@ -742,6 +2521,7 @@ def _resolve_media_path_from_hardlink(
conn = sqlite3.connect(str(hardlink_db_path))
conn.row_factory = sqlite3.Row
try:
+ dir2id_map = _build_hardlink_dir2id_map(conn)
for prefix in prefixes:
table_name = _resolve_hardlink_table_name(conn, prefix)
if not table_name:
@@ -750,7 +2530,7 @@ def _resolve_media_path_from_hardlink(
quoted = _quote_ident(table_name)
try:
row = conn.execute(
- f"SELECT dir1, dir2, file_name, modify_time FROM {quoted} WHERE md5 = ? ORDER BY modify_time DESC, dir1 DESC, rowid DESC LIMIT 1",
+ f"SELECT dir1, dir2, file_name, file_size, modify_time FROM {quoted} WHERE md5 = ? ORDER BY modify_time DESC, dir1 DESC, rowid DESC LIMIT 1",
(md5,),
).fetchone()
except Exception:
@@ -758,366 +2538,23 @@ def _resolve_media_path_from_hardlink(
if not row:
continue
- file_name = str(row["file_name"] or "").strip()
- if not file_name:
- continue
-
- if kind_key in {"video", "video_thumb"}:
- roots: list[Path] = []
- for r in [wxid_dir] + (extra_roots or []):
- if not r:
- continue
- try:
- rr = r.resolve()
- except Exception:
- rr = r
- if rr not in roots:
- roots.append(rr)
-
- def _iter_video_base_dirs(r: Path) -> list[Path]:
- bases: list[Path] = []
- try:
- if r.exists() and r.is_dir():
- pass
- else:
- return bases
- except Exception:
- return bases
-
- candidates = [
- r / "msg" / "video",
- r / "video",
- r if str(r.name).lower() == "video" else None,
- ]
- for c in candidates:
- if not c:
- continue
- try:
- if c.exists() and c.is_dir():
- bases.append(c)
- except Exception:
- continue
-
- # de-dup while keeping order
- seen: set[str] = set()
- uniq: list[Path] = []
- for b in bases:
- try:
- k = str(b.resolve())
- except Exception:
- k = str(b)
- if k in seen:
- continue
- seen.add(k)
- uniq.append(b)
- return uniq
-
- modify_time = None
- try:
- if row["modify_time"] is not None:
- modify_time = int(row["modify_time"])
- except Exception:
- modify_time = None
-
- guessed_month: Optional[str] = None
- if modify_time and modify_time > 0:
- try:
- dt = datetime.datetime.fromtimestamp(int(modify_time))
- guessed_month = f"{dt.year:04d}-{dt.month:02d}"
- except Exception:
- guessed_month = None
-
- stem = Path(file_name).stem
- if kind_key == "video":
- file_variants = [file_name]
- else:
- # Prefer real thumbnails when possible.
- file_variants = [
- f"{stem}_thumb.jpg",
- f"{stem}_thumb.jpeg",
- f"{stem}_thumb.png",
- f"{stem}_thumb.webp",
- f"{stem}.jpg",
- f"{stem}.jpeg",
- f"{stem}.png",
- f"{stem}.gif",
- f"{stem}.webp",
- f"{stem}.dat",
- file_name,
- ]
-
- for root in roots:
- for base_dir in _iter_video_base_dirs(root):
- dirs_to_check: list[Path] = []
- if guessed_month:
- dirs_to_check.append(base_dir / guessed_month)
- dirs_to_check.append(base_dir)
- for d in dirs_to_check:
- try:
- if not d.exists() or not d.is_dir():
- continue
- except Exception:
- continue
- for fv in file_variants:
- p = d / fv
- try:
- if p.exists() and p.is_file():
- return p
- except Exception:
- continue
-
- # Fallback: scan within the month directory for the exact file_name.
- if guessed_month:
- try:
- for p in d.rglob(file_name):
- try:
- if p.is_file():
- return p
- except Exception:
- continue
- except Exception:
- pass
-
- # Final fallback: locate by name under msg/video and cache.
- for base in _iter_video_base_dirs(wxid_dir):
- try:
- for p in base.rglob(file_name):
- if p.is_file():
- return p
- except Exception:
- pass
- return None
-
- if kind_key == "file":
- try:
- full_row = conn.execute(
- f"SELECT file_name, file_size, modify_time FROM {quoted} WHERE md5 = ? ORDER BY modify_time DESC LIMIT 1",
- (md5,),
- ).fetchone()
- except Exception:
- full_row = None
-
- file_size: Optional[int] = None
- modify_time: Optional[int] = None
- if full_row is not None:
- try:
- if full_row["file_size"] is not None:
- file_size = int(full_row["file_size"])
- except Exception:
- file_size = None
- try:
- if full_row["modify_time"] is not None:
- modify_time = int(full_row["modify_time"])
- except Exception:
- modify_time = None
-
- roots: list[Path] = []
- for r in [wxid_dir] + (extra_roots or []):
- if not r:
- continue
- try:
- rr = r.resolve()
- except Exception:
- rr = r
- if rr not in roots:
- roots.append(rr)
-
- file_base_dirs: list[Path] = []
- for root in roots:
- candidates = [
- root / "msg" / "file",
- root / "file" if root.name.lower() == "msg" else None,
- root if root.name.lower() == "file" else None,
- ]
- for c in candidates:
- if not c:
- continue
- try:
- if c.exists() and c.is_dir() and c not in file_base_dirs:
- file_base_dirs.append(c)
- except Exception:
- continue
-
- if not file_base_dirs:
- return None
-
- guessed_month: Optional[str] = None
- if modify_time:
- try:
- dt = datetime.datetime.fromtimestamp(int(modify_time))
- guessed_month = f"{dt.year:04d}-{dt.month:02d}"
- except Exception:
- guessed_month = None
-
- file_stem = Path(file_name).stem
-
- def _iter_month_dirs(base: Path) -> list[Path]:
- out: list[Path] = []
- try:
- for child in base.iterdir():
- try:
- if not child.is_dir():
- continue
- except Exception:
- continue
- name = str(child.name)
- if re.fullmatch(r"\d{4}-\d{2}", name):
- out.append(child)
- except Exception:
- return []
- return sorted(out, key=lambda p: str(p.name))
-
- def _pick_best_hit(hits: list[Path]) -> Optional[Path]:
- if not hits:
- return None
- if file_size is not None and file_size >= 0:
- for h in hits:
- try:
- if h.stat().st_size == file_size:
- return h
- except Exception:
- continue
- return hits[0]
-
- for base in file_base_dirs:
- month_dirs = _iter_month_dirs(base)
- month_names: list[str] = []
- if guessed_month:
- month_names.append(guessed_month)
- for d in month_dirs:
- n = str(d.name)
- if n not in month_names:
- month_names.append(n)
-
- for month_name in month_names:
- month_dir = base / month_name
- try:
- if not (month_dir.exists() and month_dir.is_dir()):
- continue
- except Exception:
- continue
-
- direct = month_dir / file_name
- try:
- if direct.exists() and direct.is_file():
- return direct
- except Exception:
- pass
-
- in_stem_dir = month_dir / file_stem / file_name
- try:
- if in_stem_dir.exists() and in_stem_dir.is_file():
- return in_stem_dir
- except Exception:
- pass
-
- hits: list[Path] = []
- try:
- for p in month_dir.rglob(file_name):
- try:
- if p.is_file():
- hits.append(p)
- if len(hits) >= 20:
- break
- except Exception:
- continue
- except Exception:
- hits = []
-
- best = _pick_best_hit(hits)
- if best:
- return best
-
- # Final fallback: search across all months (covers rare nesting patterns)
- hits_all: list[Path] = []
- try:
- for p in base.rglob(file_name):
- try:
- if p.is_file():
- hits_all.append(p)
- if len(hits_all) >= 50:
- break
- except Exception:
- continue
- except Exception:
- hits_all = []
-
- best_all = _pick_best_hit(hits_all)
- if best_all:
- return best_all
-
- if guessed_month:
- fallback_dir = base / guessed_month
- try:
- if fallback_dir.exists() and fallback_dir.is_dir():
- return fallback_dir
- except Exception:
- pass
-
- return base
-
- return None
-
- dir1 = str(row["dir1"] if row["dir1"] is not None else "").strip()
- dir2 = str(row["dir2"] if row["dir2"] is not None else "").strip()
- if not dir1 or not dir2:
- continue
-
- dir_name = dir2
- dir2id_table = _resolve_hardlink_dir2id_table_name(conn)
-
- if dir2id_table:
- try:
- drow = conn.execute(
- f"SELECT username FROM {_quote_ident(dir2id_table)} WHERE rowid = ? LIMIT 1",
- (int(dir2),),
- ).fetchone()
- if drow and drow[0]:
- dir_name = str(drow[0])
- except Exception:
- if username:
- try:
- drow = conn.execute(
- f"SELECT dir_name FROM {_quote_ident(dir2id_table)} WHERE dir_id = ? AND username = ? LIMIT 1",
- (dir2, username),
- ).fetchone()
- if drow and drow[0]:
- dir_name = str(drow[0])
- except Exception:
- pass
-
- roots: list[Path] = []
- for r in [wxid_dir] + (extra_roots or []):
- if not r:
- continue
- try:
- rr = r.resolve()
- except Exception:
- rr = r
- if rr not in roots:
- roots.append(rr)
-
- file_stem = Path(file_name).stem
- file_variants = [file_name, f"{file_stem}_h.dat", f"{file_stem}_t.dat"]
-
- for root in roots:
- for fv in file_variants:
- p = (root / dir1 / dir_name / fv).resolve()
- try:
- if p.exists() and p.is_file():
- return p
- except Exception:
- continue
-
- if username:
- chat_hash = hashlib.md5(username.encode()).hexdigest()
- for fv in file_variants:
- p = (root / "msg" / "attach" / chat_hash / dir_name / "Img" / fv).resolve()
- try:
- if p.exists() and p.is_file():
- return p
- except Exception:
- continue
+ entry = _HardlinkEntry(
+ file_name=str(row["file_name"] or "").strip(),
+ file_size=int(row["file_size"] or 0),
+ modify_time=int(row["modify_time"] or 0),
+ dir1=int(row["dir1"] or 0),
+ dir2=int(row["dir2"] or 0),
+ dir_name=str(dir2id_map.get(int(row["dir2"] or 0)) or str(row["dir2"] or "")).strip(),
+ )
+ resolved = _resolve_hardlink_entry_path(
+ kind=kind_key,
+ entry=entry,
+ wxid_dir=wxid_dir,
+ username=username,
+ extra_roots=extra_roots,
+ )
+ if resolved is not None:
+ return resolved
return None
finally:
@@ -1262,6 +2699,8 @@ def _fallback_search_media_by_md5(weixin_root_str: str, md5: str, kind: str = ""
f"{md5}*.dat",
f"{md5}*.jpg",
f"{md5}*.jpeg",
+ f"{md5}*.m4v",
+ f"{md5}*.mov",
f"{md5}*.png",
f"{md5}*.gif",
f"{md5}*.webp",
@@ -2085,6 +3524,7 @@ def _resolve_media_path_for_kind(
kind: str,
md5: str,
username: Optional[str],
+ allow_fallback_scan: bool = True,
) -> Optional[Path]:
if not md5:
return None
@@ -2123,7 +3563,7 @@ def _resolve_media_path_for_kind(
username=username,
extra_roots=roots[1:],
)
- if (not p) and wxid_dir:
+ if (not p) and wxid_dir and allow_fallback_scan:
hit = _fallback_search_media_by_md5(str(wxid_dir), str(md5), kind=kind_key)
if hit:
p = Path(hit)
diff --git a/src/wechat_decrypt_tool/routers/media.py b/src/wechat_decrypt_tool/routers/media.py
index c9806c6..b0f675b 100644
--- a/src/wechat_decrypt_tool/routers/media.py
+++ b/src/wechat_decrypt_tool/routers/media.py
@@ -1,5 +1,8 @@
import asyncio
+import concurrent.futures
import json
+import time
+from pathlib import Path
from typing import Optional
from fastapi import APIRouter, HTTPException, Request
@@ -8,8 +11,11 @@ from pydantic import BaseModel, Field
from ..logging_config import get_logger
from ..media_helpers import (
+ _collect_emoticon_download_catalog,
_collect_all_dat_files,
_decrypt_and_save_resource,
+ _get_decrypted_resource_path,
+ _detect_image_extension,
_detect_image_media_type,
_is_probably_valid_image,
_get_resource_dir,
@@ -17,6 +23,7 @@ from ..media_helpers import (
_resolve_account_dir,
_resolve_account_wxid_dir,
_save_media_keys,
+ _try_fetch_emoticon_from_remote,
_try_find_decrypted_resource,
)
from ..path_fix import PathFixRoute
@@ -47,6 +54,81 @@ def _summarize_media_keys(*, xor_key: Optional[str] = None, aes_key: Optional[st
}
+def _guess_emoji_extension(payload: bytes, media_type: Optional[str]) -> str:
+ mt = str(media_type or "").strip().lower()
+ if mt.startswith("image/"):
+ return _detect_image_extension(payload)
+ if mt == "video/mp4":
+ return "mp4"
+ return "dat"
+
+
+def _is_valid_cached_emoji(path) -> bool:
+ if not path:
+ return False
+ try:
+ if not path.exists() or not path.is_file():
+ return False
+ data = path.read_bytes()
+ except Exception:
+ return False
+
+ if not data:
+ return False
+
+ try:
+ if len(data) >= 8 and data[4:8] == b"ftyp":
+ return True
+ except Exception:
+ pass
+
+ media_type = _detect_image_media_type(data[:32])
+ if media_type == "application/octet-stream":
+ return False
+ return _is_probably_valid_image(data, media_type)
+
+
+def _normalize_emoji_download_concurrency(value: Optional[int]) -> int:
+ try:
+ n = int(value or 20)
+ except Exception:
+ n = 20
+ if n < 1:
+ return 1
+ if n > 100:
+ return 100
+ return n
+
+
+def _normalize_media_decrypt_concurrency(value: Optional[int]) -> int:
+ try:
+ n = int(value or 10)
+ except Exception:
+ n = 10
+ if n < 1:
+ return 1
+ if n > 64:
+ return 64
+ return n
+
+
+def _is_valid_cached_image(path) -> bool:
+ if not path:
+ return False
+ try:
+ if not path.exists() or not path.is_file():
+ return False
+ data = path.read_bytes()
+ except Exception:
+ return False
+ if not data:
+ return False
+ media_type = _detect_image_media_type(data[:32])
+ if media_type == "application/octet-stream":
+ return False
+ return _is_probably_valid_image(data, media_type)
+
+
class MediaKeysSaveRequest(BaseModel):
"""媒体密钥保存请求模型（用户手动提供）"""
@@ -283,6 +365,7 @@ async def decrypt_all_media_stream(
account: Optional[str] = None,
xor_key: Optional[str] = None,
aes_key: Optional[str] = None,
+ concurrency: int = 10,
):
"""批量解密所有图片资源，通过SSE实时推送进度
@@ -312,7 +395,11 @@ async def decrypt_all_media_stream(
except Exception:
return False
+ def sse(payload: dict) -> str:
+ return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+
async def generate_progress():
+ started_at = time.perf_counter()
try:
if await is_client_disconnected():
logger.info("[SSE] 客户端已断开，取消图片解密任务")
@@ -320,18 +407,20 @@ async def decrypt_all_media_stream(
account_dir = _resolve_account_dir(account)
wxid_dir = _resolve_account_wxid_dir(account_dir)
+ worker_count = _normalize_media_decrypt_concurrency(concurrency)
logger.info(
- "[media] decrypt_all_stream start: request_account=%s resolved_account=%s provided_keys=%s",
+ "[media] decrypt_all_stream start: request_account=%s resolved_account=%s provided_keys=%s requested_concurrency=%s effective_concurrency=%s",
str(account or "").strip(),
account_dir.name,
_summarize_media_keys(xor_key=xor_key, aes_key=aes_key),
+ concurrency,
+ worker_count,
)
if not wxid_dir:
- yield f"data: {json.dumps({'type': 'error', 'message': '未找到微信数据目录'})}\n\n"
+ yield sse({"type": "error", "message": "未找到微信数据目录"})
return
- # 获取密钥
xor_key_int: Optional[int] = None
aes_key16: Optional[bytes] = None
@@ -340,7 +429,7 @@ async def decrypt_all_media_stream(
xor_hex = xor_key.strip().lower().replace("0x", "")
xor_key_int = int(xor_hex, 16)
except Exception:
- yield f"data: {json.dumps({'type': 'error', 'message': 'XOR密钥格式无效'})}\n\n"
+ yield sse({"type": "error", "message": "XOR密钥格式无效"})
return
if aes_key:
@@ -348,7 +437,6 @@ async def decrypt_all_media_stream(
if len(aes_str) >= 16:
aes_key16 = aes_str[:16].encode("ascii", errors="ignore")
- # 如果未提供密钥，尝试从缓存加载
if xor_key_int is None or aes_key16 is None:
cached = _load_media_keys(account_dir)
logger.info(
@@ -375,114 +463,818 @@ async def decrypt_all_media_stream(
)
if xor_key_int is None:
- yield f"data: {json.dumps({'type': 'error', 'message': '未找到XOR密钥，请先使用 wx_key 获取并保存/填写'})}\n\n"
+ yield sse({"type": "error", "message": "未找到XOR密钥，请先使用 wx_key 获取并保存/填写"})
return
- # 收集所有.dat文件
+ scan_started_at = time.perf_counter()
logger.info(f"[SSE] 开始扫描 {wxid_dir} 中的.dat文件...")
- yield f"data: {json.dumps({'type': 'scanning', 'message': '正在扫描图片文件...'})}\n\n"
+ yield sse({"type": "scanning", "message": "正在扫描图片文件..."})
await asyncio.sleep(0)
dat_files = _collect_all_dat_files(wxid_dir)
total_files = len(dat_files)
- logger.info(f"[SSE] 共发现 {total_files} 个.dat文件（仅图片）")
+ scan_elapsed_ms = round((time.perf_counter() - scan_started_at) * 1000, 1)
+ logger.info(
+ "[media] decrypt_all_stream scan_done: account=%s total=%s elapsed_ms=%s",
+ account_dir.name,
+ total_files,
+ scan_elapsed_ms,
+ )
if await is_client_disconnected():
logger.info("[SSE] 扫描完成后客户端已断开，停止图片解密任务")
return
- if total_files == 0:
- yield f"data: {json.dumps({'type': 'complete', 'message': '未发现需要解密的图片文件', 'total': 0, 'success_count': 0, 'skip_count': 0, 'fail_count': 0})}\n\n"
- return
-
- # 发送总数信息
- yield f"data: {json.dumps({'type': 'start', 'total': total_files, 'message': f'开始解密 {total_files} 个图片文件'})}\n\n"
- await asyncio.sleep(0)
-
- # 开始解密
- success_count = 0
- skip_count = 0
- fail_count = 0
- failed_files: list[dict] = []
-
resource_dir = _get_resource_dir(account_dir)
resource_dir.mkdir(parents=True, exist_ok=True)
- for i, (dat_path, md5) in enumerate(dat_files):
- if await is_client_disconnected():
- logger.info("[SSE] 客户端已断开，停止图片解密任务")
- return
+ if total_files == 0:
+ yield sse(
+ {
+ "type": "complete",
+ "message": "未发现需要解密的图片文件",
+ "total": 0,
+ "concurrency": worker_count,
+ "success_count": 0,
+ "skip_count": 0,
+ "fail_count": 0,
+ "output_dir": str(resource_dir),
+ }
+ )
+ return
- current = i + 1
+ yield sse(
+ {
+ "type": "start",
+ "total": total_files,
+ "concurrency": worker_count,
+ "requested_concurrency": concurrency,
+ "message": f"开始解密 {total_files} 个图片文件（并发 {worker_count}）",
+ }
+ )
+ await asyncio.sleep(0)
+
+ if await is_client_disconnected():
+ logger.info("[SSE] 开始解密前客户端已断开，停止图片解密任务")
+ return
+
+ success_count = 0
+ skip_count = 0
+ fail_count = 0
+ processed_count = 0
+ failed_files: list[dict] = []
+ cache_hit_count = 0
+ decrypt_attempt_count = 0
+ decrypt_fail_count = 0
+ slow_decrypt_count = 0
+ total_cache_ms = 0.0
+ total_decrypt_ms = 0.0
+ max_decrypt_ms = 0.0
+ last_summary_at = time.perf_counter()
+
+ stop_event = asyncio.Event()
+ work_queue: asyncio.Queue = asyncio.Queue()
+ result_queue: asyncio.Queue = asyncio.Queue()
+ for item_index, item in enumerate(dat_files, start=1):
+ work_queue.put_nowait((item_index, item[0], item[1]))
+ for _ in range(worker_count):
+ work_queue.put_nowait(None)
+
+ loop = asyncio.get_running_loop()
+ executor = concurrent.futures.ThreadPoolExecutor(
+ max_workers=worker_count,
+ thread_name_prefix="media-decrypt",
+ )
+
+ def process_one(item_index: int, dat_path: Path, md5: str, worker_id: int) -> dict:
file_name = dat_path.name
-
- # 检查是否已解密
+ item_started_at = time.perf_counter()
+ cache_started_at = time.perf_counter()
existing = _try_find_decrypted_resource(account_dir, md5)
+ if existing and _is_valid_cached_image(existing):
+ return {
+ "item_index": item_index,
+ "worker_id": worker_id,
+ "file_name": file_name,
+ "md5": md5,
+ "status": "skip",
+ "message": "已存在",
+ "cache_ms": round((time.perf_counter() - cache_started_at) * 1000, 1),
+ "decrypt_ms": 0.0,
+ "elapsed_ms": round((time.perf_counter() - item_started_at) * 1000, 1),
+ }
if existing:
try:
- cached = existing.read_bytes()
- cached_mt = _detect_image_media_type(cached[:32])
- if cached_mt != "application/octet-stream" and _is_probably_valid_image(cached, cached_mt):
- skip_count += 1
- # 每100个跳过的文件发送一次进度，减少消息量
- if skip_count % 100 == 0 or current == total_files:
- yield (
- f"data: {json.dumps({'type': 'progress', 'current': current, 'total': total_files, 'success_count': success_count, 'skip_count': skip_count, 'fail_count': fail_count, 'current_file': file_name, 'status': 'skip', 'message': '已存在'})}\n\n"
- )
- await asyncio.sleep(0)
- continue
- # Cache exists but looks corrupted: remove and regenerate.
existing.unlink(missing_ok=True)
except Exception:
- # If we can't read/validate it, try regenerating.
+ pass
+ cache_elapsed_ms = round((time.perf_counter() - cache_started_at) * 1000, 1)
+
+ decrypt_started_at = time.perf_counter()
+ success, msg = _decrypt_and_save_resource(dat_path, md5, account_dir, xor_key_int, aes_key16)
+ decrypt_elapsed_ms = round((time.perf_counter() - decrypt_started_at) * 1000, 1)
+ status = "success" if success else "fail"
+ message = "解密成功" if success else str(msg or "解密失败")
+ return {
+ "item_index": item_index,
+ "worker_id": worker_id,
+ "file_name": file_name,
+ "md5": md5,
+ "status": status,
+ "message": message,
+ "cache_ms": cache_elapsed_ms,
+ "decrypt_ms": decrypt_elapsed_ms,
+ "elapsed_ms": round((time.perf_counter() - item_started_at) * 1000, 1),
+ }
+
+ async def worker(worker_id: int):
+ while not stop_event.is_set():
+ item = await work_queue.get()
+ try:
+ if item is None:
+ return
+ item_index, dat_path, md5 = item
+ if stop_event.is_set():
+ return
try:
- existing.unlink(missing_ok=True)
- except Exception:
- pass
-
- # 解密并保存
- success, msg = _decrypt_and_save_resource(
- dat_path, md5, account_dir, xor_key_int, aes_key16
- )
-
- if success:
- success_count += 1
- status = "success"
- status_msg = "解密成功"
- logger.debug(f"[SSE] 解密成功: {file_name}")
- else:
- fail_count += 1
- status = "fail"
- status_msg = msg
- logger.warning(f"[SSE] 解密失败: {file_name} - {msg}")
- if len(failed_files) < 100:
- failed_files.append(
- {
- "file": file_name,
+ result = await loop.run_in_executor(executor, process_one, item_index, dat_path, md5, worker_id)
+ except Exception as exc:
+ result = {
+ "item_index": item_index,
+ "worker_id": worker_id,
+ "file_name": dat_path.name,
"md5": md5,
- "error": msg,
+ "status": "fail",
+ "message": f"处理失败: {exc}",
+ "cache_ms": 0.0,
+ "decrypt_ms": 0.0,
+ "elapsed_ms": 0.0,
}
+ if not stop_event.is_set():
+ await result_queue.put(result)
+ finally:
+ work_queue.task_done()
+
+ worker_tasks = [asyncio.create_task(worker(i + 1)) for i in range(worker_count)]
+ logger.info(
+ "[media] decrypt_all_stream workers_started: account=%s total=%s concurrency=%s",
+ account_dir.name,
+ total_files,
+ worker_count,
+ )
+
+ try:
+ while processed_count < total_files:
+ if await is_client_disconnected():
+ stop_event.set()
+ logger.info(
+ "[SSE] 客户端已断开，停止图片解密任务: account=%s current=%s total=%s success=%s skip=%s fail=%s concurrency=%s elapsed_ms=%s",
+ account_dir.name,
+ processed_count,
+ total_files,
+ success_count,
+ skip_count,
+ fail_count,
+ worker_count,
+ round((time.perf_counter() - started_at) * 1000, 1),
+ )
+ return
+
+ try:
+ result = await asyncio.wait_for(result_queue.get(), timeout=0.5)
+ except asyncio.TimeoutError:
+ continue
+
+ processed_count += 1
+ status = str(result.get("status") or "")
+ cache_ms = float(result.get("cache_ms") or 0.0)
+ decrypt_ms = float(result.get("decrypt_ms") or 0.0)
+ elapsed_ms = float(result.get("elapsed_ms") or 0.0)
+ total_cache_ms += cache_ms
+ total_decrypt_ms += decrypt_ms
+
+ if status == "success":
+ success_count += 1
+ decrypt_attempt_count += 1
+ max_decrypt_ms = max(max_decrypt_ms, decrypt_ms)
+ if decrypt_ms >= 1000:
+ slow_decrypt_count += 1
+ elif status == "skip":
+ skip_count += 1
+ cache_hit_count += 1
+ else:
+ fail_count += 1
+ decrypt_attempt_count += 1
+ decrypt_fail_count += 1
+ max_decrypt_ms = max(max_decrypt_ms, decrypt_ms)
+ if decrypt_ms >= 1000:
+ slow_decrypt_count += 1
+ if len(failed_files) < 100:
+ failed_files.append(
+ {
+ "file": str(result.get("file_name") or ""),
+ "md5": str(result.get("md5") or ""),
+ "error": str(result.get("message") or ""),
+ }
+ )
+ logger.warning(
+ "[media] decrypt_all_stream item_fail: account=%s file=%s md5=%s worker=%s cache_ms=%s decrypt_ms=%s elapsed_ms=%s message=%s",
+ account_dir.name,
+ result.get("file_name"),
+ result.get("md5"),
+ result.get("worker_id"),
+ cache_ms,
+ decrypt_ms,
+ elapsed_ms,
+ result.get("message"),
)
- # 每处理一个文件发送进度（成功或失败都发送）
- yield (
- f"data: {json.dumps({'type': 'progress', 'current': current, 'total': total_files, 'success_count': success_count, 'skip_count': skip_count, 'fail_count': fail_count, 'current_file': file_name, 'status': status, 'message': status_msg})}\n\n"
- )
+ if elapsed_ms >= 1000:
+ logger.info(
+ "[media] decrypt_all_stream slow_item: account=%s file=%s md5=%s status=%s worker=%s cache_ms=%s decrypt_ms=%s elapsed_ms=%s",
+ account_dir.name,
+ result.get("file_name"),
+ result.get("md5"),
+ status,
+ result.get("worker_id"),
+ cache_ms,
+ decrypt_ms,
+ elapsed_ms,
+ )
- # 每处理10个文件让出一次控制权，避免阻塞
- if current % 10 == 0:
- await asyncio.sleep(0)
+ now = time.perf_counter()
+ if processed_count % 200 == 0 or (now - last_summary_at) >= 15:
+ elapsed_s = max(now - started_at, 0.001)
+ logger.info(
+ "[media] decrypt_all_stream progress_stats: account=%s current=%s total=%s success=%s skip=%s fail=%s concurrency=%s throughput_per_s=%s avg_decrypt_ms=%s max_decrypt_ms=%s slow_decrypt=%s",
+ account_dir.name,
+ processed_count,
+ total_files,
+ success_count,
+ skip_count,
+ fail_count,
+ worker_count,
+ round(processed_count / elapsed_s, 2),
+ round(total_decrypt_ms / max(decrypt_attempt_count, 1), 1),
+ round(max_decrypt_ms, 1),
+ slow_decrypt_count,
+ )
+ last_summary_at = now
- logger.info(f"[SSE] 解密完成: 成功={success_count}, 跳过={skip_count}, 失败={fail_count}")
+ yield sse(
+ {
+ "type": "progress",
+ "current": processed_count,
+ "total": total_files,
+ "concurrency": worker_count,
+ "success_count": success_count,
+ "skip_count": skip_count,
+ "fail_count": fail_count,
+ "current_file": result.get("file_name"),
+ "status": status,
+ "message": result.get("message"),
+ "worker_id": result.get("worker_id"),
+ "cache_ms": round(cache_ms, 1),
+ "decrypt_ms": round(decrypt_ms, 1),
+ "elapsed_ms": round(elapsed_ms, 1),
+ }
+ )
+ result_queue.task_done()
- # 发送完成消息
- yield (
- f"data: {json.dumps({'type': 'complete', 'total': total_files, 'success_count': success_count, 'skip_count': skip_count, 'fail_count': fail_count, 'output_dir': str(resource_dir), 'failed_files': failed_files[:20], 'message': f'解密完成: 成功 {success_count}, 跳过 {skip_count}, 失败 {fail_count}'})}\n\n"
+ if processed_count % 20 == 0:
+ await asyncio.sleep(0)
+ finally:
+ stop_event.set()
+ for task in worker_tasks:
+ task.cancel()
+ if worker_tasks:
+ await asyncio.gather(*worker_tasks, return_exceptions=True)
+ executor.shutdown(wait=False, cancel_futures=True)
+
+ avg_decrypt_ms = round(total_decrypt_ms / max(decrypt_attempt_count, 1), 1)
+ total_elapsed_ms = round((time.perf_counter() - started_at) * 1000, 1)
+ logger.info(
+ "[media] decrypt_all_stream complete: account=%s total=%s success=%s skip=%s fail=%s concurrency=%s cache_hit=%s decrypt_attempt=%s decrypt_fail=%s avg_decrypt_ms=%s max_decrypt_ms=%s slow_decrypt=%s elapsed_ms=%s",
+ account_dir.name,
+ total_files,
+ success_count,
+ skip_count,
+ fail_count,
+ worker_count,
+ cache_hit_count,
+ decrypt_attempt_count,
+ decrypt_fail_count,
+ avg_decrypt_ms,
+ round(max_decrypt_ms, 1),
+ slow_decrypt_count,
+ total_elapsed_ms,
+ )
+
+ yield sse(
+ {
+ "type": "complete",
+ "total": total_files,
+ "concurrency": worker_count,
+ "success_count": success_count,
+ "skip_count": skip_count,
+ "fail_count": fail_count,
+ "output_dir": str(resource_dir),
+ "decrypt_stats": {
+ "cache_hit_count": cache_hit_count,
+ "decrypt_attempt_count": decrypt_attempt_count,
+ "decrypt_fail_count": decrypt_fail_count,
+ "avg_cache_ms": round(total_cache_ms / max(total_files, 1), 1),
+ "avg_decrypt_ms": avg_decrypt_ms,
+ "max_decrypt_ms": round(max_decrypt_ms, 1),
+ "slow_decrypt_count": slow_decrypt_count,
+ },
+ "failed_files": failed_files[:20],
+ "message": f"解密完成: 成功 {success_count}, 跳过 {skip_count}, 失败 {fail_count}（并发 {worker_count}）",
+ }
)
except Exception as e:
logger.error(f"[SSE] 解密过程出错: {e}")
- yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
+ yield sse({"type": "error", "message": str(e)})
+
+ return StreamingResponse(
+ generate_progress(),
+ media_type="text/event-stream",
+ headers={
+ "Cache-Control": "no-cache",
+ "Connection": "keep-alive",
+ "X-Accel-Buffering": "no",
+ },
+ )
+
+
+@router.get("/api/media/emoji/download_all_stream", summary="批量下载所有表情资源（SSE实时进度）")
+async def download_all_emojis_stream(
+ request: Request,
+ account: Optional[str] = None,
+ force: bool = False,
+ concurrency: int = 20,
+):
+ async def is_client_disconnected() -> bool:
+ try:
+ return await request.is_disconnected()
+ except Exception:
+ return False
+
+ def sse(payload: dict) -> str:
+ return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+
+ async def generate_progress():
+ started_at = time.perf_counter()
+ try:
+ if await is_client_disconnected():
+ logger.info("[SSE] 客户端已断开，取消表情下载任务")
+ return
+
+ account_dir = _resolve_account_dir(account)
+ worker_count = _normalize_emoji_download_concurrency(concurrency)
+ logger.info(
+ "[media] emoji_download_all_stream start: request_account=%s resolved_account=%s force=%s requested_concurrency=%s effective_concurrency=%s",
+ str(account or "").strip(),
+ account_dir.name,
+ bool(force),
+ concurrency,
+ worker_count,
+ )
+
+ scan_started_at = time.perf_counter()
+ yield sse({"type": "scanning", "message": "正在扫描表情资源..."})
+ await asyncio.sleep(0)
+
+ candidate_map, scan_stats = _collect_emoticon_download_catalog(account_dir)
+ md5_list = list(candidate_map.keys())
+ total_files = len(md5_list)
+ scan_elapsed_ms = round((time.perf_counter() - scan_started_at) * 1000, 1)
+ source_breakdown = scan_stats.get("source_counts") or {}
+ logger.info(
+ "[media] emoji_download_all_stream scan_done: account=%s total=%s source_breakdown=%s stats=%s elapsed_ms=%s",
+ account_dir.name,
+ total_files,
+ source_breakdown,
+ scan_stats,
+ scan_elapsed_ms,
+ )
+
+ if await is_client_disconnected():
+ logger.info("[SSE] 扫描完成后客户端已断开，停止表情下载任务")
+ return
+
+ resource_dir = _get_resource_dir(account_dir)
+ resource_dir.mkdir(parents=True, exist_ok=True)
+
+ if total_files == 0:
+ yield sse(
+ {
+ "type": "complete",
+ "message": "未发现可下载的表情资源",
+ "total": 0,
+ "success_count": 0,
+ "skip_count": 0,
+ "fail_count": 0,
+ "output_dir": str(resource_dir),
+ "source_breakdown": source_breakdown,
+ "source_stats": scan_stats,
+ }
+ )
+ return
+
+ yield sse(
+ {
+ "type": "start",
+ "total": total_files,
+ "concurrency": worker_count,
+ "requested_concurrency": concurrency,
+ "source_breakdown": source_breakdown,
+ "source_stats": scan_stats,
+ "message": f"开始下载 {total_files} 个表情资源（并发 {worker_count}）",
+ }
+ )
+ await asyncio.sleep(0)
+
+ if await is_client_disconnected():
+ logger.info("[SSE] 开始下载前客户端已断开，停止表情下载任务")
+ return
+
+ success_count = 0
+ skip_count = 0
+ fail_count = 0
+ processed_count = 0
+ failed_files: list[dict] = []
+ cache_hit_count = 0
+ fetch_attempt_count = 0
+ fetch_fail_count = 0
+ write_attempt_count = 0
+ write_fail_count = 0
+ slow_fetch_count = 0
+ slow_write_count = 0
+ total_cache_ms = 0.0
+ total_fetch_ms = 0.0
+ total_write_ms = 0.0
+ max_fetch_ms = 0.0
+ max_write_ms = 0.0
+ last_summary_at = time.perf_counter()
+
+ stop_event = asyncio.Event()
+ work_queue: asyncio.Queue = asyncio.Queue()
+ result_queue: asyncio.Queue = asyncio.Queue()
+ for item_index, md5 in enumerate(md5_list, start=1):
+ work_queue.put_nowait((item_index, md5))
+ for _ in range(worker_count):
+ work_queue.put_nowait(None)
+
+ loop = asyncio.get_running_loop()
+ executor = concurrent.futures.ThreadPoolExecutor(
+ max_workers=worker_count,
+ thread_name_prefix="emoji-download",
+ )
+
+ def process_one(item_index: int, md5: str, worker_id: int) -> dict:
+ item_started_at = time.perf_counter()
+ source_info = candidate_map.get(md5, {})
+ source_names = ",".join(str(s) for s in (source_info.get("sources") or []))
+ url_count = len(source_info.get("urls") or [])
+ cache_started_at = time.perf_counter()
+
+ existing = _try_find_decrypted_resource(account_dir, md5)
+ if existing:
+ if (not force) and _is_valid_cached_emoji(existing):
+ return {
+ "item_index": item_index,
+ "worker_id": worker_id,
+ "md5": md5,
+ "source": source_names,
+ "url_count": url_count,
+ "status": "skip",
+ "message": "已存在",
+ "path": str(existing),
+ "cache_ms": round((time.perf_counter() - cache_started_at) * 1000, 1),
+ "fetch_ms": 0.0,
+ "write_ms": 0.0,
+ "elapsed_ms": round((time.perf_counter() - item_started_at) * 1000, 1),
+ }
+ try:
+ existing.unlink(missing_ok=True)
+ except Exception:
+ pass
+ cache_elapsed_ms = round((time.perf_counter() - cache_started_at) * 1000, 1)
+
+ fetch_started_at = time.perf_counter()
+ try:
+ payload, media_type = _try_fetch_emoticon_from_remote(account_dir, md5, source_info)
+ fetch_error = ""
+ except Exception as exc:
+ payload, media_type = None, None
+ fetch_error = str(exc)
+ fetch_elapsed_ms = round((time.perf_counter() - fetch_started_at) * 1000, 1)
+
+ if not payload or not media_type:
+ message = "未找到可下载地址或下载失败"
+ if fetch_error:
+ message = f"{message}: {fetch_error}"
+ return {
+ "item_index": item_index,
+ "worker_id": worker_id,
+ "md5": md5,
+ "source": source_names,
+ "url_count": url_count,
+ "status": "fail",
+ "phase": "fetch",
+ "message": message,
+ "error": fetch_error,
+ "cache_ms": cache_elapsed_ms,
+ "fetch_ms": fetch_elapsed_ms,
+ "write_ms": 0.0,
+ "elapsed_ms": round((time.perf_counter() - item_started_at) * 1000, 1),
+ }
+
+ ext = _guess_emoji_extension(payload, media_type)
+ write_started_at = time.perf_counter()
+ try:
+ output_path = _get_decrypted_resource_path(account_dir, md5, ext)
+ output_path.parent.mkdir(parents=True, exist_ok=True)
+ output_path.write_bytes(payload)
+ except Exception as exc:
+ return {
+ "item_index": item_index,
+ "worker_id": worker_id,
+ "md5": md5,
+ "source": source_names,
+ "url_count": url_count,
+ "status": "fail",
+ "phase": "write",
+ "message": f"写入失败: {exc}",
+ "error": str(exc),
+ "cache_ms": cache_elapsed_ms,
+ "fetch_ms": fetch_elapsed_ms,
+ "write_ms": round((time.perf_counter() - write_started_at) * 1000, 1),
+ "elapsed_ms": round((time.perf_counter() - item_started_at) * 1000, 1),
+ }
+
+ return {
+ "item_index": item_index,
+ "worker_id": worker_id,
+ "md5": md5,
+ "source": source_names,
+ "url_count": url_count,
+ "status": "success",
+ "message": "下载成功",
+ "path": str(output_path),
+ "media_type": media_type,
+ "bytes": len(payload),
+ "cache_ms": cache_elapsed_ms,
+ "fetch_ms": fetch_elapsed_ms,
+ "write_ms": round((time.perf_counter() - write_started_at) * 1000, 1),
+ "elapsed_ms": round((time.perf_counter() - item_started_at) * 1000, 1),
+ }
+
+ async def worker(worker_id: int):
+ while not stop_event.is_set():
+ item = await work_queue.get()
+ try:
+ if item is None:
+ return
+ item_index, md5 = item
+ if stop_event.is_set():
+ return
+ try:
+ result = await loop.run_in_executor(executor, process_one, item_index, md5, worker_id)
+ except Exception as exc:
+ result = {
+ "item_index": item_index,
+ "worker_id": worker_id,
+ "md5": md5,
+ "source": "",
+ "url_count": 0,
+ "status": "fail",
+ "phase": "worker",
+ "message": f"处理失败: {exc}",
+ "error": str(exc),
+ "cache_ms": 0.0,
+ "fetch_ms": 0.0,
+ "write_ms": 0.0,
+ "elapsed_ms": 0.0,
+ }
+ if not stop_event.is_set():
+ await result_queue.put(result)
+ finally:
+ work_queue.task_done()
+
+ worker_tasks = [asyncio.create_task(worker(i + 1)) for i in range(worker_count)]
+ logger.info(
+ "[media] emoji_download_all_stream workers_started: account=%s total=%s concurrency=%s",
+ account_dir.name,
+ total_files,
+ worker_count,
+ )
+
+ try:
+ while processed_count < total_files:
+ if await is_client_disconnected():
+ stop_event.set()
+ logger.info(
+ "[SSE] 客户端已断开，停止表情下载任务: account=%s current=%s total=%s success=%s skip=%s fail=%s concurrency=%s elapsed_ms=%s",
+ account_dir.name,
+ processed_count,
+ total_files,
+ success_count,
+ skip_count,
+ fail_count,
+ worker_count,
+ round((time.perf_counter() - started_at) * 1000, 1),
+ )
+ return
+
+ try:
+ result = await asyncio.wait_for(result_queue.get(), timeout=0.5)
+ except asyncio.TimeoutError:
+ continue
+
+ processed_count += 1
+ status = str(result.get("status") or "")
+ phase = str(result.get("phase") or "")
+ md5 = str(result.get("md5") or "")
+ source_names = str(result.get("source") or "")
+ item_elapsed_ms = float(result.get("elapsed_ms") or 0.0)
+ cache_ms = float(result.get("cache_ms") or 0.0)
+ fetch_ms = float(result.get("fetch_ms") or 0.0)
+ write_ms = float(result.get("write_ms") or 0.0)
+ total_cache_ms += cache_ms
+ total_fetch_ms += fetch_ms
+ total_write_ms += write_ms
+ if fetch_ms:
+ fetch_attempt_count += 1
+ max_fetch_ms = max(max_fetch_ms, fetch_ms)
+ if fetch_ms >= 3000:
+ slow_fetch_count += 1
+ if write_ms:
+ write_attempt_count += 1
+ max_write_ms = max(max_write_ms, write_ms)
+ if write_ms >= 1000:
+ slow_write_count += 1
+
+ if status == "success":
+ success_count += 1
+ elif status == "skip":
+ skip_count += 1
+ cache_hit_count += 1
+ else:
+ fail_count += 1
+ if phase == "write":
+ write_fail_count += 1
+ else:
+ fetch_fail_count += 1
+ if len(failed_files) < 100:
+ failed_files.append({"md5": md5, "error": str(result.get("message") or "")})
+ logger.debug(
+ "[media] emoji_download_all_stream fail_detail: account=%s md5=%s phase=%s source=%s worker=%s url_count=%s cache_ms=%s fetch_ms=%s write_ms=%s elapsed_ms=%s error=%s",
+ account_dir.name,
+ md5,
+ phase,
+ source_names,
+ result.get("worker_id"),
+ result.get("url_count"),
+ cache_ms,
+ fetch_ms,
+ write_ms,
+ item_elapsed_ms,
+ result.get("error"),
+ )
+
+ if item_elapsed_ms >= 1000:
+ logger.info(
+ "[media] emoji_download_all_stream slow_item: account=%s md5=%s status=%s phase=%s source=%s worker=%s cache_ms=%s fetch_ms=%s write_ms=%s elapsed_ms=%s bytes=%s",
+ account_dir.name,
+ md5,
+ status,
+ phase,
+ source_names,
+ result.get("worker_id"),
+ cache_ms,
+ fetch_ms,
+ write_ms,
+ item_elapsed_ms,
+ result.get("bytes") or 0,
+ )
+
+ now = time.perf_counter()
+ if processed_count % 200 == 0 or (now - last_summary_at) >= 15:
+ elapsed_s = max(now - started_at, 0.001)
+ logger.info(
+ "[media] emoji_download_all_stream progress_stats: account=%s current=%s total=%s success=%s skip=%s fail=%s concurrency=%s throughput_per_s=%s avg_fetch_ms=%s max_fetch_ms=%s avg_write_ms=%s max_write_ms=%s slow_fetch=%s slow_write=%s",
+ account_dir.name,
+ processed_count,
+ total_files,
+ success_count,
+ skip_count,
+ fail_count,
+ worker_count,
+ round(processed_count / elapsed_s, 2),
+ round(total_fetch_ms / max(fetch_attempt_count, 1), 1),
+ round(max_fetch_ms, 1),
+ round(total_write_ms / max(write_attempt_count, 1), 1),
+ round(max_write_ms, 1),
+ slow_fetch_count,
+ slow_write_count,
+ )
+ last_summary_at = now
+
+ event_payload = {
+ "type": "progress",
+ "current": processed_count,
+ "total": total_files,
+ "success_count": success_count,
+ "skip_count": skip_count,
+ "fail_count": fail_count,
+ "current_file": md5,
+ "source": source_names,
+ "status": status,
+ "message": str(result.get("message") or ""),
+ "concurrency": worker_count,
+ "worker_id": result.get("worker_id"),
+ "elapsed_ms": round(item_elapsed_ms, 1),
+ "cache_ms": round(cache_ms, 1),
+ "fetch_ms": round(fetch_ms, 1),
+ "write_ms": round(write_ms, 1),
+ }
+ if result.get("path"):
+ event_payload["path"] = result.get("path")
+ if result.get("media_type"):
+ event_payload["media_type"] = result.get("media_type")
+ yield sse(event_payload)
+ result_queue.task_done()
+
+ if processed_count % 20 == 0:
+ await asyncio.sleep(0)
+ finally:
+ stop_event.set()
+ for task in worker_tasks:
+ task.cancel()
+ if worker_tasks:
+ await asyncio.gather(*worker_tasks, return_exceptions=True)
+ executor.shutdown(wait=False, cancel_futures=True)
+
+ total_elapsed_ms = round((time.perf_counter() - started_at) * 1000, 1)
+ avg_fetch_ms = round(total_fetch_ms / max(fetch_attempt_count, 1), 1)
+ avg_write_ms = round(total_write_ms / max(write_attempt_count, 1), 1)
+ logger.info(
+ "[media] emoji_download_all_stream complete: account=%s total=%s success=%s skip=%s fail=%s concurrency=%s cache_hit=%s fetch_attempt=%s fetch_fail=%s write_attempt=%s write_fail=%s avg_fetch_ms=%s max_fetch_ms=%s avg_write_ms=%s max_write_ms=%s slow_fetch=%s slow_write=%s source_breakdown=%s elapsed_ms=%s",
+ account_dir.name,
+ total_files,
+ success_count,
+ skip_count,
+ fail_count,
+ worker_count,
+ cache_hit_count,
+ fetch_attempt_count,
+ fetch_fail_count,
+ write_attempt_count,
+ write_fail_count,
+ avg_fetch_ms,
+ round(max_fetch_ms, 1),
+ avg_write_ms,
+ round(max_write_ms, 1),
+ slow_fetch_count,
+ slow_write_count,
+ source_breakdown,
+ total_elapsed_ms,
+ )
+ yield sse(
+ {
+ "type": "complete",
+ "total": total_files,
+ "success_count": success_count,
+ "skip_count": skip_count,
+ "fail_count": fail_count,
+ "output_dir": str(resource_dir),
+ "concurrency": worker_count,
+ "download_stats": {
+ "cache_hit_count": cache_hit_count,
+ "fetch_attempt_count": fetch_attempt_count,
+ "fetch_fail_count": fetch_fail_count,
+ "write_attempt_count": write_attempt_count,
+ "write_fail_count": write_fail_count,
+ "avg_cache_ms": round(total_cache_ms / max(total_files, 1), 1),
+ "avg_fetch_ms": avg_fetch_ms,
+ "max_fetch_ms": round(max_fetch_ms, 1),
+ "avg_write_ms": avg_write_ms,
+ "max_write_ms": round(max_write_ms, 1),
+ "slow_fetch_count": slow_fetch_count,
+ "slow_write_count": slow_write_count,
+ },
+ "source_breakdown": source_breakdown,
+ "source_stats": scan_stats,
+ "failed_files": failed_files[:20],
+ "message": f"表情下载完成: 成功 {success_count}, 跳过 {skip_count}, 失败 {fail_count}（并发 {worker_count}）",
+ }
+ )
+ except Exception as exc:
+ logger.error("[media] emoji_download_all_stream error: %s", exc, exc_info=True)
+ yield sse({"type": "error", "message": str(exc)})
return StreamingResponse(
generate_progress(),
diff --git a/tests/test_media_decrypt_stream_cancel.py b/tests/test_media_decrypt_stream_cancel.py
index 0324b55..ff6e2c2 100644
--- a/tests/test_media_decrypt_stream_cancel.py
+++ b/tests/test_media_decrypt_stream_cancel.py
@@ -24,7 +24,80 @@ class _FakeDisconnectingRequest:
return self._calls >= self._disconnect_after
+async def _read_sse_events(response) -> list[dict]:
+ chunks = []
+ async for chunk in response.body_iterator:
+ chunks.append(chunk.decode("utf-8") if isinstance(chunk, bytes) else str(chunk))
+
+ events = []
+ for chunk in chunks:
+ for line in chunk.splitlines():
+ if line.startswith("data: "):
+ events.append(json.loads(line[len("data: ") :]))
+ return events
+
+
class TestMediaDecryptStreamCancel(unittest.TestCase):
+ def test_stream_uses_default_concurrency(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account_dir = root / "account"
+ wxid_dir = root / "wxid"
+ dat_path = wxid_dir / "image.dat"
+ resource_dir = account_dir / "resource"
+ wxid_dir.mkdir(parents=True, exist_ok=True)
+ dat_path.write_bytes(b"encrypted")
+
+ with mock.patch.object(media_router, "_resolve_account_dir", return_value=account_dir):
+ with mock.patch.object(media_router, "_resolve_account_wxid_dir", return_value=wxid_dir):
+ with mock.patch.object(media_router, "_load_media_keys", return_value={"xor": 0xA5, "aes": ""}):
+ with mock.patch.object(media_router, "_collect_all_dat_files", return_value=[(dat_path, "abc123")]):
+ with mock.patch.object(media_router, "_get_resource_dir", return_value=resource_dir):
+ with mock.patch.object(media_router, "_try_find_decrypted_resource", return_value=None):
+ with mock.patch.object(media_router, "_decrypt_and_save_resource", return_value=(True, "ok")):
+ response = asyncio.run(
+ media_router.decrypt_all_media_stream(
+ request=_FakeDisconnectingRequest(disconnect_after=999),
+ account="wxid_demo",
+ )
+ )
+ events = asyncio.run(_read_sse_events(response))
+
+ self.assertEqual([event.get("type") for event in events], ["scanning", "start", "progress", "complete"])
+ self.assertEqual(events[1].get("concurrency"), 10)
+ self.assertEqual(events[2].get("concurrency"), 10)
+ self.assertEqual(events[3].get("concurrency"), 10)
+
+ def test_stream_uses_requested_concurrency(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account_dir = root / "account"
+ wxid_dir = root / "wxid"
+ dat_path = wxid_dir / "image.dat"
+ resource_dir = account_dir / "resource"
+ wxid_dir.mkdir(parents=True, exist_ok=True)
+ dat_path.write_bytes(b"encrypted")
+
+ with mock.patch.object(media_router, "_resolve_account_dir", return_value=account_dir):
+ with mock.patch.object(media_router, "_resolve_account_wxid_dir", return_value=wxid_dir):
+ with mock.patch.object(media_router, "_load_media_keys", return_value={"xor": 0xA5, "aes": ""}):
+ with mock.patch.object(media_router, "_collect_all_dat_files", return_value=[(dat_path, "abc123")]):
+ with mock.patch.object(media_router, "_get_resource_dir", return_value=resource_dir):
+ with mock.patch.object(media_router, "_try_find_decrypted_resource", return_value=None):
+ with mock.patch.object(media_router, "_decrypt_and_save_resource", return_value=(True, "ok")):
+ response = asyncio.run(
+ media_router.decrypt_all_media_stream(
+ request=_FakeDisconnectingRequest(disconnect_after=999),
+ account="wxid_demo",
+ concurrency=7,
+ )
+ )
+ events = asyncio.run(_read_sse_events(response))
+
+ self.assertEqual(events[1].get("concurrency"), 7)
+ self.assertEqual(events[2].get("concurrency"), 7)
+ self.assertEqual(events[3].get("concurrency"), 7)
+
def test_stream_stops_processing_when_client_disconnects(self):
with TemporaryDirectory() as td:
root = Path(td)
@@ -43,28 +116,15 @@ class TestMediaDecryptStreamCancel(unittest.TestCase):
with mock.patch.object(media_router, "_load_media_keys", return_value={"xor": 0xA5, "aes": ""}):
with mock.patch.object(media_router, "_collect_all_dat_files", return_value=[(dat_path, "abc123")]):
with mock.patch.object(media_router, "_get_resource_dir", return_value=resource_dir):
- with mock.patch.object(media_router, "_try_find_decrypted_resource", return_value=None):
- with mock.patch.object(media_router, "_decrypt_and_save_resource", decrypt_mock):
- response = asyncio.run(
- media_router.decrypt_all_media_stream(
- request=request,
- account="wxid_demo",
+ with mock.patch.object(media_router, "_try_find_decrypted_resource", return_value=None):
+ with mock.patch.object(media_router, "_decrypt_and_save_resource", decrypt_mock):
+ response = asyncio.run(
+ media_router.decrypt_all_media_stream(
+ request=request,
+ account="wxid_demo",
+ )
)
- )
-
- async def _read_chunks():
- chunks = []
- async for chunk in response.body_iterator:
- chunks.append(chunk.decode("utf-8") if isinstance(chunk, bytes) else str(chunk))
- return chunks
-
- chunks = asyncio.run(_read_chunks())
-
- events = []
- for chunk in chunks:
- for line in chunk.splitlines():
- if line.startswith("data: "):
- events.append(json.loads(line[len("data: ") :]))
+ events = asyncio.run(_read_sse_events(response))
self.assertEqual([event.get("type") for event in events], ["scanning", "start"])
decrypt_mock.assert_not_called()
diff --git a/tests/test_media_emoji_download_stream.py b/tests/test_media_emoji_download_stream.py
new file mode 100644
index 0000000..5696e56
--- /dev/null
+++ b/tests/test_media_emoji_download_stream.py
@@ -0,0 +1,190 @@
+import asyncio
+import json
+import sys
+import unittest
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest import mock
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+
+from wechat_decrypt_tool.routers import media as media_router # noqa: E402 pylint: disable=wrong-import-position
+
+
+PNG_1X1 = bytes.fromhex(
+ "89504E470D0A1A0A"
+ "0000000D49484452000000010000000108060000001F15C489"
+ "0000000D49444154789C6360606060000000050001A5F64540"
+ "0000000049454E44AE426082"
+)
+
+
+class _FakeRequest:
+ async def is_disconnected(self):
+ return False
+
+
+class _FakeDisconnectingRequest:
+ def __init__(self, disconnect_after: int):
+ self._disconnect_after = disconnect_after
+ self._calls = 0
+
+ async def is_disconnected(self):
+ self._calls += 1
+ return self._calls >= self._disconnect_after
+
+
+def _emoji_catalog(md5: str):
+ return (
+ {
+ md5: {
+ "md5": md5,
+ "urls": [f"https://example.com/{md5}.png"],
+ "aes_keys": [],
+ "sources": ["message_xml"],
+ }
+ },
+ {
+ "total_candidates": 1,
+ "total_candidates_with_url": 1,
+ "source_counts": {"message_xml": 1},
+ },
+ )
+
+
+async def _read_sse_events(response) -> list[dict]:
+ chunks = []
+ async for chunk in response.body_iterator:
+ chunks.append(chunk.decode("utf-8") if isinstance(chunk, bytes) else str(chunk))
+
+ events = []
+ for chunk in chunks:
+ for line in chunk.splitlines():
+ if line.startswith("data: "):
+ events.append(json.loads(line[len("data: ") :]))
+ return events
+
+
+class TestMediaEmojiDownloadStream(unittest.TestCase):
+ def test_stream_downloads_missing_emoji_and_saves_resource(self):
+ with TemporaryDirectory() as td:
+ account_dir = Path(td) / "account"
+ account_dir.mkdir(parents=True, exist_ok=True)
+ md5 = "a" * 32
+
+ with mock.patch.object(media_router, "_resolve_account_dir", return_value=account_dir):
+ with mock.patch.object(
+ media_router,
+ "_collect_emoticon_download_catalog",
+ return_value=_emoji_catalog(md5),
+ ):
+ with mock.patch.object(
+ media_router,
+ "_try_fetch_emoticon_from_remote",
+ return_value=(PNG_1X1, "image/png"),
+ ) as fetch_mock:
+ response = asyncio.run(
+ media_router.download_all_emojis_stream(
+ request=_FakeRequest(),
+ account="wxid_demo",
+ )
+ )
+ events = asyncio.run(_read_sse_events(response))
+
+ self.assertEqual([event.get("type") for event in events], ["scanning", "start", "progress", "complete"])
+ self.assertEqual(events[2].get("status"), "success")
+ self.assertEqual(events[3].get("success_count"), 1)
+ self.assertEqual(events[1].get("concurrency"), 20)
+ self.assertTrue((account_dir / "resource" / md5[:2] / f"{md5}.png").exists())
+ fetch_mock.assert_called_once()
+
+ def test_stream_uses_requested_concurrency(self):
+ with TemporaryDirectory() as td:
+ account_dir = Path(td) / "account"
+ account_dir.mkdir(parents=True, exist_ok=True)
+ md5 = "d" * 32
+
+ with mock.patch.object(media_router, "_resolve_account_dir", return_value=account_dir):
+ with mock.patch.object(
+ media_router,
+ "_collect_emoticon_download_catalog",
+ return_value=_emoji_catalog(md5),
+ ):
+ with mock.patch.object(
+ media_router,
+ "_try_fetch_emoticon_from_remote",
+ return_value=(PNG_1X1, "image/png"),
+ ):
+ response = asyncio.run(
+ media_router.download_all_emojis_stream(
+ request=_FakeRequest(),
+ account="wxid_demo",
+ concurrency=7,
+ )
+ )
+ events = asyncio.run(_read_sse_events(response))
+
+ self.assertEqual(events[1].get("concurrency"), 7)
+ self.assertEqual(events[2].get("concurrency"), 7)
+ self.assertEqual(events[3].get("concurrency"), 7)
+
+ def test_stream_skips_existing_downloaded_emoji(self):
+ with TemporaryDirectory() as td:
+ account_dir = Path(td) / "account"
+ md5 = "b" * 32
+ resource_dir = account_dir / "resource" / md5[:2]
+ account_dir.mkdir(parents=True, exist_ok=True)
+ resource_dir.mkdir(parents=True, exist_ok=True)
+ cached = resource_dir / f"{md5}.png"
+ cached.write_bytes(PNG_1X1)
+
+ with mock.patch.object(media_router, "_resolve_account_dir", return_value=account_dir):
+ with mock.patch.object(
+ media_router,
+ "_collect_emoticon_download_catalog",
+ return_value=_emoji_catalog(md5),
+ ):
+ with mock.patch.object(media_router, "_try_fetch_emoticon_from_remote") as fetch_mock:
+ response = asyncio.run(
+ media_router.download_all_emojis_stream(
+ request=_FakeRequest(),
+ account="wxid_demo",
+ )
+ )
+ events = asyncio.run(_read_sse_events(response))
+
+ self.assertEqual([event.get("type") for event in events], ["scanning", "start", "progress", "complete"])
+ self.assertEqual(events[2].get("status"), "skip")
+ self.assertEqual(events[3].get("skip_count"), 1)
+ fetch_mock.assert_not_called()
+
+ def test_stream_stops_before_processing_when_client_disconnects(self):
+ with TemporaryDirectory() as td:
+ account_dir = Path(td) / "account"
+ account_dir.mkdir(parents=True, exist_ok=True)
+ md5 = "c" * 32
+
+ with mock.patch.object(media_router, "_resolve_account_dir", return_value=account_dir):
+ with mock.patch.object(
+ media_router,
+ "_collect_emoticon_download_catalog",
+ return_value=_emoji_catalog(md5),
+ ):
+ with mock.patch.object(media_router, "_try_fetch_emoticon_from_remote") as fetch_mock:
+ response = asyncio.run(
+ media_router.download_all_emojis_stream(
+ request=_FakeDisconnectingRequest(disconnect_after=3),
+ account="wxid_demo",
+ )
+ )
+ events = asyncio.run(_read_sse_events(response))
+
+ self.assertEqual([event.get("type") for event in events], ["scanning", "start"])
+ fetch_mock.assert_not_called()
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_media_emoticon_catalog.py b/tests/test_media_emoticon_catalog.py
new file mode 100644
index 0000000..329ba0d
--- /dev/null
+++ b/tests/test_media_emoticon_catalog.py
@@ -0,0 +1,109 @@
+import sqlite3
+import sys
+import unittest
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+
+from wechat_decrypt_tool.media_helpers import ( # noqa: E402 pylint: disable=wrong-import-position
+ _collect_emoticon_download_catalog,
+ _lookup_emoticon_info,
+)
+
+
+class TestMediaEmoticonCatalog(unittest.TestCase):
+ def test_catalog_merges_emoticon_db_extern_md5_and_message_xml(self):
+ with TemporaryDirectory() as td:
+ account_dir = Path(td) / "account"
+ account_dir.mkdir(parents=True, exist_ok=True)
+ primary_md5 = "a" * 32
+ extern_md5 = "b" * 32
+ message_md5 = "c" * 32
+ no_url_md5 = "d" * 32
+ message_extern_md5 = "e" * 32
+ aes_key = "1" * 32
+
+ conn = sqlite3.connect(str(account_dir / "emoticon.db"))
+ conn.execute(
+ "CREATE TABLE kNonStoreEmoticonTable ("
+ "md5 TEXT, extern_md5 TEXT, aes_key TEXT, cdn_url TEXT, encrypt_url TEXT, "
+ "extern_url TEXT, thumb_url TEXT, tp_url TEXT)"
+ )
+ conn.execute(
+ "INSERT INTO kNonStoreEmoticonTable VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+ (
+ primary_md5,
+ extern_md5,
+ aes_key,
+ f"https://example.com/{primary_md5}.gif",
+ "",
+ "",
+ "",
+ "",
+ ),
+ )
+ conn.commit()
+ conn.close()
+
+ conn = sqlite3.connect(str(account_dir / "message_0.db"))
+ conn.execute(
+ "CREATE TABLE Msg_demo ("
+ "local_type INTEGER, compress_content BLOB, message_content BLOB, packed_info_data BLOB)"
+ )
+ conn.executemany(
+ "INSERT INTO Msg_demo VALUES (?, ?, ?, ?)",
+ [
+ (
+ 47,
+ None,
+ (
+ f'
'
+ ),
+ bytes([0x10, 0x45]),
+ ),
+ (
+ 47,
+ None,
+ f'
',
+ bytes([0x10, 0x45]),
+ ),
+ (
+ 47,
+ None,
+ f'
',
+ bytes([0x10, 0x45]),
+ ),
+ ],
+ )
+ conn.commit()
+ conn.close()
+
+ catalog, stats = _collect_emoticon_download_catalog(account_dir)
+
+ self.assertEqual(set(catalog), {primary_md5, extern_md5, message_md5})
+ self.assertIn("emoticon_db_md5", catalog[primary_md5]["sources"])
+ self.assertIn("message_xml", catalog[primary_md5]["sources"])
+ self.assertIn("emoticon_db_extern_md5", catalog[extern_md5]["sources"])
+ self.assertIn("message_xml", catalog[message_md5]["sources"])
+ self.assertNotIn(no_url_md5, catalog)
+ self.assertEqual(stats["emoticon_db_md5"], 1)
+ self.assertEqual(stats["emoticon_db_extern_md5"], 1)
+ self.assertEqual(stats["message_xml_rows"], 3)
+ self.assertEqual(stats["message_xml_md5"], 3)
+ self.assertEqual(stats["message_xml_md5_with_url"], 2)
+ self.assertEqual(stats["message_xml_extern_md5"], 1)
+ self.assertEqual(stats["message_builtin_expr_ids"], 1)
+ self.assertEqual(stats["source_counts"]["message_xml"], 2)
+
+ info = _lookup_emoticon_info(str(account_dir), extern_md5)
+ self.assertEqual(info["md5"], primary_md5)
+ self.assertEqual(info["extern_md5"], extern_md5)
+
+
+if __name__ == "__main__":
+ unittest.main()