ZIP 文件名
可选，留空时自动生成 @@ -282,6 +300,36 @@ />
+
+
+
+
+
+
+
+
+
+
+ {{ exportFolderNamePreview }}
+
+ 未选中的既有会话会继续保留
+ 增量目录
+ + {{ exportBaselineStatusLabel }} + +保存目录
@@ -293,7 +341,7 @@
{{ exportFolder || '尚未选择保存目录' }}
- {{ exportFolder ? '导出完成后会写入此目录' : '开始导出前需要先完成此项' }}
+ {{ exportFolder ? (exportOutputMode === 'folder' ? '将创建或更新其中的增量目录' : '导出完成后会写入此目录') : '开始导出前需要先完成此项' }}
- 缺失媒体
- {{ exportJob.progress?.mediaMissing || 0 }}
+ {{ exportJob.options?.outputMode === 'folder' ? '待补媒体（去重）' : '缺失媒体' }}
+
+ {{ exportJob.options?.outputMode === 'folder' && exportJob.status === 'done'
+ ? (exportJob.unresolvedMedia?.uniqueCount || 0)
+ : (exportJob.progress?.mediaMissing || 0) }}
+
{{ exportJob.progress?.currentConversationMessagesExported || 0 }} / {{ exportJob.progress?.currentConversationMessagesTotal || 0 }} 条消息
+
\n"
+ )
)
parts.append('
+
+
+
+
+
+
+
+ 增量目录已更新
+
+ {{ exportJob.folderPath || exportJob.folderName || exportFolderNamePreview }}
+
+
+
+ 新增 {{ exportJob.incremental?.messagesAdded || 0 }}
+ 更新 {{ exportJob.incremental?.conversationsUpdated || 0 }}
+ 复用 {{ exportJob.incremental?.conversationsReused || 0 }}
+
+ 补回 {{ exportJob.incremental.filesRecovered }}
+
+
+ 同步历史 {{ exportJob.incremental.historyChangesSynced }}
+
+
+
+ 浏览器目录：{{ exportFolder || '未选择' }}
+
+ {{ exportSaveProgressText }}
+ {{ exportSaveMsg }}
+
+
+
+
+
+
+
+
+
+ {{ exportJob.repairCandidates.length }} 个会话存在可恢复差异
+ 已确认修复会产生变化，仅重建对应会话。
+
+
+
+
+
+
+ {{ exportJob.unresolvedMedia?.uniqueCount || 0 }} 个媒体当前无法获取
+
+ 影响 {{ exportJob.unresolvedMedia?.referenceCount || 0 }} 条消息；源端暂不可用，重复修复不会改变结果。
+
+
+
+
+
+ 查看完整任务说明
+{{ exportJob.warning }}
+
- {{ isExportCreating ? '正在创建' : '开始导出' }}
+ {{ isExportCreating ? '正在创建' : (exportOutputMode === 'folder' ? '开始更新' : '开始导出') }}
\n"
+ if folder_context is not None
+ else "提示：此页面需要 JavaScript 才能使用“合并聊天记录”等交互功能。若该提示一直存在，请确认已完整解压导出目录，并检查 assets/_wce/ 下的运行时文件是否完整。\n')
parts.append('
\n')
@@ -2545,7 +3586,10 @@ def esc_attr(v: Any) -> str:
}
zf.writestr("manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2))
zf.writestr("report.json", json.dumps(report, ensure_ascii=False, indent=2))
- if export_format == "html":
+ if folder_context is not None:
+ # 可持续目录由基线保存逐文件摘要；不可变 ZIP 的阻断式完整性链不适用于原地更新。
+ pass
+ elif export_format == "html":
try:
html_assets = dict(job.options.get("_htmlAssets") or {})
write_active_html_zip_integrity(zf, job.export_id, html_assets)
@@ -2569,7 +3613,39 @@ def esc_attr(v: Any) -> str:
_raise_if_job_cancelled(job, "before_finalize", trace)
phase_started = time.perf_counter()
- if job.content_key is not None:
+ if folder_context is not None:
+ final_out = materialize_folder_archive(
+ job=job,
+ archive_path=tmp_zip,
+ context=folder_context,
+ )
+ tmp_zip.unlink(missing_ok=True)
+ warning_parts: list[str] = []
+ if folder_context.reset_baseline:
+ warning_parts.append("已重置基线并完整重建本次选择的会话。")
+ recovered_files = int(job.incremental.get("filesRecovered") or 0)
+ if recovered_files:
+ warning_parts.append(f"已补回 {recovered_files} 个缺失或异常的受管理文件。")
+ if folder_context.repair_candidates:
+ warning_parts.append(
+ f"发现 {len(folder_context.repair_candidates)} 个会话存在可恢复差异。"
+ )
+ unresolved_unique = int(job.unresolved_media.get("uniqueCount") or 0)
+ unresolved_references = int(job.unresolved_media.get("referenceCount") or 0)
+ if unresolved_unique:
+ warning_parts.append(
+ f"有 {unresolved_unique} 个媒体当前在源端不可用，影响 {unresolved_references} 条消息；重复修复不会产生变化。"
+ )
+ if folder_context.history_synced:
+ warning_parts.append(
+ f"已在追加新消息时同步 {len(folder_context.history_synced)} 个会话的历史变化。"
+ )
+ if folder_context.unresolved_missing_owner_keys:
+ warning_parts.append(
+ f"有 {len(folder_context.unresolved_missing_owner_keys)} 个缺失文件所属会话已不在当前数据源中，将在后续更新时继续尝试补回。"
+ )
+ job.warning = " ".join(warning_parts)
+ elif job.content_key is not None:
final_out = final_zip.with_name(final_zip.name + ".wec")
if final_out.exists():
final_out = final_out.with_name(
@@ -2591,13 +3667,14 @@ def esc_attr(v: Any) -> str:
with self._lock:
job.status = "done"
- job.zip_path = final_out
+ job.zip_path = final_out if folder_context is None else None
job.finished_at = time.time()
_safe_trace(
trace,
"job_done",
durationMs=round(((job.finished_at or time.time()) - (job.started_at or job.created_at)) * 1000.0, 1),
- finalZip=str(final_out),
+ finalZip=str(final_out) if folder_context is None else "",
+ folderPath=str(final_out) if folder_context is not None else "",
messagesExported=job.progress.messages_exported,
mediaCopied=job.progress.media_copied,
mediaMissing=job.progress.media_missing,
@@ -2622,10 +3699,16 @@ def esc_attr(v: Any) -> str:
mediaMissing=job.progress.media_missing,
)
except Exception:
- if job.content_key is not None:
+ if job.content_key is not None or folder_context is not None:
tmp_zip.unlink(missing_ok=True)
if encrypted_output_path is not None:
encrypted_output_path.unlink(missing_ok=True)
+ if job.staging_dir is not None and job.staging_dir.exists():
+ import shutil
+
+ shutil.rmtree(job.staging_dir, ignore_errors=True)
+ job.staging_dir = None
+ job.staged_files = {}
raise
finally:
if realtime_paused:
@@ -3490,6 +4573,207 @@ def sort_key(r: _Row) -> tuple[int, int, int]:
return heapq.merge(*streams, key=sort_key)
+def _incremental_row_key(row: _Row) -> tuple[int, int, int, int, str, str]:
+ return (
+ int(row.create_time or 0),
+ int(row.sort_seq or 0),
+ int(row.local_id or 0),
+ int(row.server_id or 0),
+ str(row.db_stem or ""),
+ str(row.table_name or ""),
+ )
+
+
+def _incremental_row_payload(row: _Row, message: dict[str, Any]) -> bytes:
+ normalized_message = {
+ key: value
+ for key, value in message.items()
+ if key not in {"createTimeText", "conversationUsername", "_mediaUsername"}
+ }
+ value = {
+ "key": list(_incremental_row_key(row)),
+ "message": normalized_message,
+ }
+ return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
+
+
+def _incremental_watermark(value: Any) -> tuple[int, int, int, int, str, str] | None:
+ raw = value if isinstance(value, list) else []
+ if len(raw) < 6:
+ return None
+ try:
+ return (
+ int(raw[0] or 0),
+ int(raw[1] or 0),
+ int(raw[2] or 0),
+ int(raw[3] or 0),
+ str(raw[4] or ""),
+ str(raw[5] or ""),
+ )
+ except Exception:
+ return None
+
+
+def _probe_incremental_append(
+ *,
+ account_dir: Path,
+ conv_username: str,
+ start_time: Optional[int],
+ end_time: Optional[int],
+ source: str,
+ rt_conn: Any | None,
+ old_state: dict[str, Any],
+ want_types: Optional[set[str]] = None,
+ is_group: bool = False,
+ checkpoint: Optional[Callable[[], None]] = None,
+) -> dict[str, Any]:
+ """只读取旧水位之后的消息，供可分页格式执行真正的追加更新。"""
+
+ old_watermark = _incremental_watermark(old_state.get("watermark"))
+ if old_watermark is None:
+ raise ValueError("incremental append requires one valid watermark")
+
+ lower_bound = int(old_watermark[0] or 0)
+ if start_time is not None:
+ lower_bound = max(lower_bound, int(start_time))
+ prepared_messages: list[dict[str, Any]] = []
+ new_payloads: list[bytes] = []
+ watermark = old_watermark
+ rows = _iter_rows_for_conversation(
+ account_dir=account_dir,
+ conv_username=conv_username,
+ start_time=lower_bound,
+ end_time=end_time,
+ local_types=None,
+ source=source,
+ rt_conn=rt_conn,
+ checkpoint=checkpoint,
+ )
+ for row in rows:
+ key = _incremental_row_key(row)
+ if key <= old_watermark:
+ continue
+ watermark = max(watermark, key)
+ parsed = _parse_message_for_export(
+ row=row,
+ conv_username=conv_username,
+ is_group=is_group,
+ resource_conn=None,
+ resource_chat_id=None,
+ )
+ if not _is_render_type_selected(parsed.get("renderType"), want_types):
+ continue
+ prepared_messages.append(parsed)
+ new_payloads.append(_incremental_row_payload(row, parsed))
+
+ old_count = int(old_state.get("sourceRowCount") or old_state.get("messageCount") or 0)
+ chained_hash = hashlib.sha256()
+ chained_hash.update(b"wce-incremental-append-v1\0")
+ chained_hash.update(str(old_state.get("historyFingerprint") or "").encode("ascii", errors="ignore"))
+ for payload in new_payloads:
+ chained_hash.update(len(payload).to_bytes(4, "big"))
+ chained_hash.update(payload)
+ return {
+ "messageCount": old_count + len(prepared_messages),
+ "newMessageCount": len(prepared_messages),
+ "watermark": list(watermark),
+ "historyFingerprint": (
+ chained_hash.hexdigest()
+ if prepared_messages
+ else str(old_state.get("historyFingerprint") or "")
+ ),
+ "historyChanged": False,
+ "preparedMessages": prepared_messages,
+ "appendProbe": True,
+ }
+
+
+def _probe_incremental_conversation(
+ *,
+ account_dir: Path,
+ conv_username: str,
+ start_time: Optional[int],
+ end_time: Optional[int],
+ source: str,
+ rt_conn: Any | None,
+ old_state: dict[str, Any],
+ want_types: Optional[set[str]] = None,
+ is_group: bool = False,
+ privacy_mode: bool = False,
+ checkpoint: Optional[Callable[[], None]] = None,
+) -> dict[str, Any]:
+ """流式计算旧水位前的历史指纹和当前会话水位。"""
+
+ old_watermark = _incremental_watermark(old_state.get("watermark"))
+
+ full_hash = hashlib.sha256()
+ prefix_hash = hashlib.sha256()
+ count = 0
+ prefix_count = 0
+ new_count = 0
+ watermark: tuple[int, int, int, int, str, str] | None = None
+ privacy_sender_alias_map: dict[str, int] = {}
+ rows = _iter_rows_for_conversation(
+ account_dir=account_dir,
+ conv_username=conv_username,
+ start_time=start_time,
+ end_time=end_time,
+ local_types=None,
+ source=source,
+ rt_conn=rt_conn,
+ checkpoint=checkpoint,
+ )
+ for row in rows:
+ parsed = _parse_message_for_export(
+ row=row,
+ conv_username=conv_username,
+ is_group=is_group,
+ resource_conn=None,
+ resource_chat_id=None,
+ )
+ if not _is_render_type_selected(parsed.get("renderType"), want_types):
+ continue
+ if privacy_mode:
+ _privacy_scrub_message(
+ parsed,
+ conv_is_group=is_group,
+ sender_alias_map=privacy_sender_alias_map,
+ )
+ payload = _incremental_row_payload(row, parsed)
+ key = _incremental_row_key(row)
+ full_hash.update(len(payload).to_bytes(4, "big"))
+ full_hash.update(payload)
+ count += 1
+ watermark = key
+ if old_watermark is not None and key <= old_watermark:
+ prefix_hash.update(len(payload).to_bytes(4, "big"))
+ prefix_hash.update(payload)
+ prefix_count += 1
+ elif old_watermark is not None:
+ new_count += 1
+
+ old_exists = bool(old_state)
+ if not old_exists or old_watermark is None:
+ new_count = count
+ old_count = int(old_state.get("sourceRowCount") or old_state.get("messageCount") or 0)
+ old_fingerprint = str(old_state.get("historyFingerprint") or "")
+ history_changed = bool(
+ old_exists
+ and old_watermark is not None
+ and (
+ prefix_count != old_count
+ or prefix_hash.hexdigest() != old_fingerprint
+ )
+ )
+ return {
+ "messageCount": count,
+ "newMessageCount": max(0, new_count),
+ "watermark": list(watermark) if watermark is not None else [],
+ "historyFingerprint": full_hash.hexdigest(),
+ "historyChanged": history_changed,
+ }
+
+
def _parse_message_for_export(
*,
row: _Row,
@@ -4189,6 +5473,7 @@ def lookup_alias(username: str) -> str:
zf=zf,
account_dir=account_dir,
conv_username=media_conv_username,
+ owner_username=conv_username,
msg=msg,
media_written=media_written,
report=report,
@@ -4551,6 +5836,7 @@ def lookup_alias(username: str) -> str:
zf=zf,
account_dir=account_dir,
conv_username=media_conv_username,
+ owner_username=conv_username,
msg=msg,
media_written=media_written,
report=report,
@@ -4611,6 +5897,85 @@ def lookup_alias(username: str) -> str:
return exported
+def _html_incremental_snapshot_text(text: str) -> Optional[dict[str, Any]]:
+ """读取 HTML 的最后一页和分页元数据；结构不匹配时安全回退全量生成。"""
+
+ marker = '![]()
\n'
+ marker_at = text.find(marker)
+ if marker_at < 0:
+ return None
+ content_start = marker_at + len(marker)
+ page_pattern = re.compile(
+ r'',
+ flags=re.DOTALL,
+ )
+ page_match = page_pattern.search(text, content_start)
+ brand_at = text.find('
\n"
+ )
)
# Root
@@ -5468,6 +6846,7 @@ def _mark_exported() -> None:
zf=zf,
account_dir=account_dir,
conv_username=media_conv_username,
+ owner_username=conv_username,
msg=msg,
media_written=media_written,
report=report,
@@ -5475,6 +6854,7 @@ def _mark_exported() -> None:
allow_process_key_extract=allow_process_key_extract,
media_db_path=media_db_path,
media_index=media_index,
+ remote_written=remote_written,
lock=lock,
job=job,
)
@@ -6304,7 +7684,11 @@ def get_quoted_link_text() -> str:
num = str(page_no).zfill(int(paged_pad_width or 4))
arc_js = f"{conv_dir}/pages/page-{num}.js"
js_payload = _html_export_page_fragment_js(
- export_id=str(getattr(job, "export_id", "") or ""),
+ export_id=str(
+ (getattr(job, "options", {}) or {}).get("_folderRuntimeId")
+ or getattr(job, "export_id", "")
+ or ""
+ ),
arc_js=arc_js,
page_no=int(page_no),
fragment_html=frag_text,
@@ -6321,6 +7705,121 @@ def get_quoted_link_text() -> str:
return exported
+def _write_conversation_html_append(
+ *,
+ zf: _ZipIntegrityWriter,
+ conv_dir: str,
+ existing_snapshot: dict[str, Any],
+ old_state: dict[str, Any],
+ new_messages: list[dict[str, Any]],
+ runtime_id: str,
+ writer_options: dict[str, Any],
+) -> int:
+ """把旧内联末页转为历史分页，仅渲染并内联本轮新增消息。"""
+
+ old_count = int(old_state.get("messageCount") or 0)
+ if old_count <= 0 or not str(existing_snapshot.get("fragment") or "").strip():
+ return _write_conversation_html(
+ zf=zf,
+ conv_dir=conv_dir,
+ prepared_messages=new_messages,
+ **writer_options,
+ )
+
+ temp_buffer = io.BytesIO()
+ with zipfile.ZipFile(temp_buffer, mode="w", compression=zipfile.ZIP_DEFLATED, compresslevel=6) as temp_archive:
+ temp_writer = _ZipIntegrityWriter(temp_archive, native_integrity=None)
+ added_count = _write_conversation_html(
+ zf=temp_writer,
+ conv_dir=conv_dir,
+ prepared_messages=new_messages,
+ **writer_options,
+ )
+ if added_count <= 0:
+ return old_count
+
+ temp_buffer.seek(0)
+ with zipfile.ZipFile(temp_buffer, mode="r") as temp_archive:
+ temp_html = temp_archive.read(f"{conv_dir}/messages.html").decode("utf-8")
+ delta_snapshot = _html_incremental_snapshot_text(temp_html)
+ if delta_snapshot is None or not str(delta_snapshot.get("fragment") or "").strip():
+ raise RuntimeError("新增 HTML 分页生成失败，无法安全追加。")
+ for name in temp_archive.namelist():
+ normalized = _zip_arcname(name)
+ if not normalized or normalized.startswith(f"{conv_dir}/"):
+ continue
+ zf.writestr(normalized, temp_archive.read(name))
+
+ old_text = str(existing_snapshot.get("text") or "")
+ content_start = int(existing_snapshot.get("contentStart") or 0)
+ content_end = int(existing_snapshot.get("contentEnd") or 0)
+ close_marker = str(existing_snapshot.get("closeMarker") or "")
+ old_fragment = str(existing_snapshot.get("fragment") or "")
+ new_fragment = str(delta_snapshot.get("fragment") or "")
+ old_page_meta = dict(existing_snapshot.get("pageMeta") or {})
+ old_total_pages = max(1, int(old_page_meta.get("totalPages") or 1))
+ new_total_pages = old_total_pages + 1
+ pad_width = max(4, int(old_page_meta.get("padWidth") or 4), len(str(new_total_pages)))
+
+ old_page_number = str(old_total_pages).zfill(pad_width)
+ old_page_arc = f"{conv_dir}/pages/page-{old_page_number}.js"
+ zf.writestr(
+ old_page_arc,
+ _html_export_page_fragment_js(
+ export_id=str(runtime_id or ""),
+ arc_js=old_page_arc,
+ page_no=old_total_pages,
+ fragment_html=_minify_html_for_export(old_fragment),
+ ),
+ )
+
+ merged_html = old_text[:content_start] + new_fragment + old_text[content_end:]
+ page_meta = {
+ "schemaVersion": 1,
+ "pageSize": int(old_page_meta.get("pageSize") or writer_options.get("html_page_size") or 1000),
+ "totalPages": new_total_pages,
+ "initialPage": new_total_pages,
+ "totalMessages": old_count + added_count,
+ "padWidth": pad_width,
+ "pageFilePrefix": "pages/page-",
+ "pageFileSuffix": ".js",
+ "inlinedPages": [new_total_pages],
+ }
+ page_script = (
+ '"
+ )
+ page_pattern = re.compile(
+ r'',
+ flags=re.DOTALL,
+ )
+ if page_pattern.search(merged_html):
+ merged_html = page_pattern.sub(lambda _match: page_script, merged_html, count=1)
+ else:
+ inserted_at = content_start + len(new_fragment) + len(close_marker)
+ merged_html = merged_html[:inserted_at] + page_script + "\n" + merged_html[inserted_at:]
+
+ merged_media = _merge_html_media_index(
+ existing_snapshot.get("mediaIndex") or {},
+ delta_snapshot.get("mediaIndex") or {},
+ )
+ media_script = (
+ '"
+ )
+ media_pattern = re.compile(
+ r'',
+ flags=re.DOTALL,
+ )
+ if media_pattern.search(merged_html):
+ merged_html = media_pattern.sub(lambda _match: media_script, merged_html, count=1)
+
+ zf.writestr(f"{conv_dir}/messages.html", merged_html)
+ return old_count + added_count
+
+
def _format_message_line_txt(*, msg: dict[str, Any]) -> str:
ts = int(msg.get("createTime") or 0)
time_text = _format_ts(ts)
@@ -6552,11 +8051,109 @@ def _attach_voice_transcript(
stats["failed"] = int(stats.get("failed") or 0) + 1
+def _pending_media_local_repairability(
+ *,
+ account_dir: Path,
+ conv_username: str,
+ item: dict[str, Any],
+ media_index: Optional[MediaPathIndex],
+) -> tuple[bool, str]:
+ """只在确认本地资源能生成有效产物时，才允许进入媒体修复流程。"""
+
+ kind = str(item.get("kind") or "").strip().lower()
+ ident = str(item.get("id") or "").strip()
+ if not kind or not ident:
+ return False, "SOURCE_ID_MISSING"
+ md5 = ident.lower() if _is_md5(ident.lower()) else ""
+ file_id = "" if md5 else ident
+ source: Optional[Path] = None
+ if md5:
+ try:
+ source = _try_find_decrypted_resource(account_dir, md5)
+ except Exception:
+ source = None
+ if source is None and media_index is not None:
+ try:
+ source = media_index.resolve(
+ kind=kind,
+ md5=md5,
+ file_id=file_id,
+ username=str(conv_username or "").strip(),
+ )
+ except Exception:
+ source = None
+ if source is None and md5:
+ try:
+ source = _resolve_media_path_for_kind(
+ account_dir,
+ kind=kind,
+ md5=md5,
+ username=conv_username,
+ allow_fallback_scan=False,
+ )
+ except Exception:
+ source = None
+ try:
+ if source is None or not source.is_file():
+ return False, "SOURCE_NOT_FOUND"
+ except Exception:
+ return False, "SOURCE_NOT_FOUND"
+
+ if kind == "file":
+ return True, "LOCAL_SOURCE_READY"
+ if kind == "voice":
+ # 语音需要按 server_id 从媒体库读取，不能只凭待补标识判断可恢复。
+ return False, "SOURCE_NOT_FOUND"
+ try:
+ data, media_type = _read_and_maybe_decrypt_media(source, account_dir=account_dir)
+ except Exception:
+ return False, "SOURCE_DECRYPT_FAILED"
+ media_type = str(media_type or "").strip().lower()
+ if kind in {"image", "emoji", "video_thumb"}:
+ return (
+ (True, "LOCAL_SOURCE_READY")
+ if media_type.startswith("image/") and bool(data)
+ else (False, "SOURCE_DECRYPT_FAILED")
+ )
+ if kind == "video":
+ return (
+ (True, "LOCAL_SOURCE_READY")
+ if media_type == "video/mp4" and bool(data)
+ else (False, "SOURCE_DECRYPT_FAILED")
+ )
+ return False, "SOURCE_FORMAT_UNSUPPORTED"
+
+
+def _classify_pending_media(
+ *,
+ account_dir: Path,
+ conv_username: str,
+ values: Any,
+ media_index: Optional[MediaPathIndex],
+) -> tuple[list[dict[str, Any]], bool]:
+ normalized = normalize_pending_media(values)
+ classified: list[dict[str, Any]] = []
+ for raw in normalized:
+ item = dict(raw)
+ repairable, reason_code = _pending_media_local_repairability(
+ account_dir=account_dir,
+ conv_username=conv_username,
+ item=item,
+ media_index=media_index,
+ )
+ item["repairable"] = repairable
+ item["state"] = "recoverable_local" if repairable else "source_unavailable"
+ item["reasonCode"] = reason_code
+ classified.append(item)
+ return classified, classified != normalized
+
+
def _attach_offline_media(
*,
zf: zipfile.ZipFile,
account_dir: Path,
conv_username: str,
+ owner_username: str,
msg: dict[str, Any],
media_written: dict[str, str],
report: dict[str, Any],
@@ -6564,6 +8161,7 @@ def _attach_offline_media(
allow_process_key_extract: bool,
media_db_path: Path,
media_index: Optional[MediaPathIndex],
+ remote_written: Optional[dict[str, str]] = None,
lock: threading.Lock,
job: ExportJob,
) -> None:
@@ -6582,7 +8180,15 @@ def _attach_offline_media(
def record_missing(kind: str, ident: str) -> None:
with lock:
- job.progress.media_missing += 1
+ job.progress.media_missing_references += 1
+ if str(job.options.get("outputMode") or "zip") == "folder":
+ key = (str(kind or ""), str(ident or ""))
+ if key not in job.missing_media_keys:
+ job.missing_media_keys.add(key)
+ job.progress.media_missing += 1
+ else:
+ # ZIP 全量保留原有按消息引用计数的行为。
+ job.progress.media_missing += 1
try:
report["missingMedia"].append(
{
@@ -6595,6 +8201,31 @@ def record_missing(kind: str, ident: str) -> None:
except Exception:
pass
+ def try_remote_image(kind: str, ident: str, url: Any) -> tuple[str, bool]:
+ if (
+ str(job.options.get("outputMode") or "zip") != "folder"
+ or remote_written is None
+ or not bool(job.options.get("downloadRemoteMedia"))
+ ):
+ return "", False
+ raw = str(url or "").strip()
+ try:
+ parsed = urlparse(raw)
+ except Exception:
+ parsed = None
+ if parsed is None or parsed.scheme not in {"http", "https"} or not parsed.netloc:
+ return "", False
+ was_known = raw in remote_written
+ arc = _download_remote_image_to_zip(
+ zf=zf,
+ url=raw,
+ remote_written=remote_written,
+ report=report,
+ )
+ if arc and ident:
+ media_written[f"{kind}:{ident}"] = arc
+ return arc, bool(arc and not was_known)
+
offline: list[dict[str, Any]] = []
if rt == "image" and "image" in media_kinds:
@@ -6666,6 +8297,16 @@ def add_file_id(v: Any) -> None:
used_file_id = file_id
break
+ if not arc:
+ arc, is_new = try_remote_image(
+ "image",
+ primary_md5 or primary_file_id,
+ msg.get("imageUrl"),
+ )
+ if arc:
+ used_md5 = primary_md5
+ used_file_id = primary_file_id
+
if arc:
# Keep primary fields in sync with what actually resolved.
try:
@@ -6697,6 +8338,8 @@ def add_file_id(v: Any) -> None:
suggested_name="",
media_index=media_index,
)
+ if not arc:
+ arc, is_new = try_remote_image("emoji", md5 or file_id, msg.get("emojiUrl"))
if arc:
offline.append({"kind": "emoji", "path": arc, "md5": md5, "fileId": file_id})
if is_new:
@@ -6792,6 +8435,17 @@ def add_file_id(v: Any) -> None:
if offline:
msg["offlineMedia"] = offline
+ if str(job.options.get("outputMode") or "") == "folder":
+ owners_by_path = job.options.setdefault("_folderResourceOwners", {})
+ owner = str(owner_username or "").strip()
+ if isinstance(owners_by_path, dict) and owner:
+ for item in offline:
+ path = str(item.get("path") or "").strip()
+ if not path:
+ continue
+ owners = owners_by_path.setdefault(path, [])
+ if isinstance(owners, list) and owner not in owners:
+ owners.append(owner)
def _materialize_avatar(
diff --git a/src/wechat_decrypt_tool/chat_incremental_export.py b/src/wechat_decrypt_tool/chat_incremental_export.py
new file mode 100644
index 00000000..9a433422
--- /dev/null
+++ b/src/wechat_decrypt_tool/chat_incremental_export.py
@@ -0,0 +1,922 @@
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import os
+import re
+import shutil
+import tempfile
+import uuid
+import zipfile
+from dataclasses import dataclass, field
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Optional
+
+
+STATE_FILE_NAME = ".wechat-chat-export.json"
+SCHEMA_VERSION = 1
+ARTIFACT_TYPE = "wechat-chat-incremental-folder"
+PENDING_MEDIA_STATES = {
+ "unclassified",
+ "recoverable_local",
+ "recoverable_remote",
+ "retryable",
+ "source_unavailable",
+ "decrypt_blocked",
+ "unsupported",
+}
+
+
+class ChatIncrementalError(ValueError):
+ """聊天增量目录无法安全继续更新。"""
+
+ def __init__(self, code: str, message: str):
+ super().__init__(message)
+ self.code = str(code or "incremental_error")
+
+
+def _safe_component(value: Any, *, fallback: str) -> str:
+ text = str(value or "").strip()
+ text = re.sub(r"[<>:\"/\\|?*\x00-\x1f]", "_", text)
+ text = re.sub(r"\s+", " ", text).strip(" .")
+ if not text:
+ text = fallback
+ return text[:96].rstrip(" .") or fallback
+
+
+def normalize_relative_path(value: Any) -> str:
+ text = str(value or "").replace("\\", "/").lstrip("/")
+ parts = [part for part in text.split("/") if part not in {"", "."}]
+ if not parts or any(part == ".." for part in parts):
+ return ""
+ return "/".join(parts)
+
+
+def _require_managed_path(value: Any) -> str:
+ raw = str(value or "").strip().replace("\\", "/")
+ normalized = normalize_relative_path(raw)
+ if (
+ not normalized
+ or raw.startswith("/")
+ or raw != normalized
+ or ":" in normalized.split("/", 1)[0]
+ or "\x00" in raw
+ ):
+ raise ChatIncrementalError("incremental_unsafe_path", "增量基线包含不安全的文件路径。")
+ return normalized
+
+
+def _validate_baseline_paths(value: dict[str, Any]) -> None:
+ files = value.get("files") if isinstance(value.get("files"), dict) else {}
+ for path, metadata in files.items():
+ _require_managed_path(path)
+ if not isinstance(metadata, dict):
+ raise ChatIncrementalError("incremental_baseline_invalid", "增量基线文件摘要损坏，请选择新目录。")
+ digest = str(metadata.get("sha256") or "")
+ try:
+ size = int(metadata.get("size"))
+ except Exception as exc:
+ raise ChatIncrementalError("incremental_baseline_invalid", "增量基线文件摘要损坏，请选择新目录。") from exc
+ if size < 0 or not re.fullmatch(r"[0-9a-f]{64}", digest):
+ raise ChatIncrementalError("incremental_baseline_invalid", "增量基线文件摘要损坏，请选择新目录。")
+ conversations = value.get("conversations") if isinstance(value.get("conversations"), dict) else {}
+ for state in conversations.values():
+ if not isinstance(state, dict):
+ raise ChatIncrementalError("incremental_baseline_invalid", "增量基线会话信息损坏，请选择新目录。")
+ directory = _require_managed_path(state.get("directory"))
+ if not directory.startswith("conversations/"):
+ raise ChatIncrementalError("incremental_unsafe_path", "增量基线包含不安全的会话目录。")
+ managed_files = state.get("managedFiles") or []
+ if not isinstance(managed_files, list):
+ raise ChatIncrementalError("incremental_baseline_invalid", "增量基线会话信息损坏，请选择新目录。")
+ for path in managed_files:
+ _require_managed_path(path)
+
+
+def _canonical_json(value: Any) -> bytes:
+ return json.dumps(
+ value,
+ ensure_ascii=False,
+ sort_keys=True,
+ separators=(",", ":"),
+ ).encode("utf-8")
+
+
+def build_config(
+ *,
+ export_format: str,
+ start_time: Optional[int],
+ end_time: Optional[int],
+ message_types: list[str],
+ include_media: bool,
+ media_kinds: list[str],
+ download_remote_media: bool,
+ html_page_size: int,
+ privacy_mode: bool,
+ transcribe_voice: bool,
+) -> dict[str, Any]:
+ return {
+ "format": str(export_format or "").strip().lower(),
+ "startTime": int(start_time) if start_time is not None else None,
+ "endTime": int(end_time) if end_time is not None else None,
+ "messageTypes": sorted({str(item or "").strip() for item in message_types if str(item or "").strip()}),
+ "includeMedia": bool(include_media),
+ "mediaKinds": sorted({str(item or "").strip() for item in media_kinds if str(item or "").strip()}),
+ "downloadRemoteMedia": bool(download_remote_media),
+ "htmlPageSize": int(html_page_size) if str(export_format or "").lower() == "html" else None,
+ "privacyMode": bool(privacy_mode),
+ "transcribeVoice": bool(transcribe_voice),
+ }
+
+
+def config_fingerprint(config: dict[str, Any]) -> str:
+ return hashlib.sha256(_canonical_json(config)).hexdigest()
+
+
+def conversation_key(*, salt: str, username: str) -> str:
+ payload = f"{str(salt or '')}\0{str(username or '')}".encode("utf-8", errors="replace")
+ return hashlib.sha256(payload).hexdigest()
+
+
+def account_fingerprint(account: str) -> str:
+ return hashlib.sha256(str(account or "").encode("utf-8", errors="replace")).hexdigest()
+
+
+def privacy_account_token(account: str) -> str:
+ """生成与浏览器端一致的短标识，避免隐私目录名包含账号明文。"""
+
+ value = 0x811C9DC5
+ raw = str(account or "").encode("utf-16-le", errors="surrogatepass")
+ for index in range(0, len(raw), 2):
+ unit = raw[index] | (raw[index + 1] << 8)
+ value ^= unit
+ value = (value * 0x01000193) & 0xFFFFFFFF
+ return f"{value:08x}"
+
+
+def allocate_conversation_directory(
+ *,
+ old_state: dict[str, Any],
+ key: str,
+ display_name: str,
+ privacy_mode: bool,
+) -> str:
+ conversations = old_state.get("conversations") if isinstance(old_state.get("conversations"), dict) else {}
+ old = conversations.get(key) if isinstance(conversations.get(key), dict) else {}
+ existing = normalize_relative_path(old.get("directory"))
+ if existing and existing.startswith("conversations/"):
+ return existing
+ stem = "conversation" if privacy_mode else _safe_component(display_name, fallback="conversation")
+ return f"conversations/{stem}_{key[:10]}"
+
+
+def normalize_pending_media(
+ values: Any,
+ *,
+ default_state: str = "unclassified",
+ default_reason: str = "",
+) -> list[dict[str, Any]]:
+ """按媒体唯一标识合并待补项，同时保留受影响消息引用数。"""
+
+ state_default = str(default_state or "unclassified").strip()
+ if state_default not in PENDING_MEDIA_STATES:
+ state_default = "unclassified"
+ grouped: dict[tuple[str, str], dict[str, Any]] = {}
+ for raw in values if isinstance(values, list) else []:
+ if not isinstance(raw, dict):
+ continue
+ kind = str(raw.get("kind") or "").strip().lower()
+ ident = str(raw.get("id") or "").strip()
+ if not kind or not ident:
+ continue
+ key = (kind, ident)
+ try:
+ occurrences = max(1, int(raw.get("occurrenceCount") or 1))
+ except Exception:
+ occurrences = 1
+ state = str(raw.get("state") or state_default).strip()
+ if state not in PENDING_MEDIA_STATES:
+ state = state_default
+ repairable = bool(raw.get("repairable")) or state in {"recoverable_local", "recoverable_remote"}
+ reason_code = str(raw.get("reasonCode") or default_reason or "").strip()
+ existing = grouped.get(key)
+ if existing is None:
+ existing = {
+ "kind": kind,
+ "id": ident,
+ "occurrenceCount": occurrences,
+ "state": state,
+ "reasonCode": reason_code,
+ "repairable": repairable,
+ }
+ message_id = str(raw.get("messageId") or "").strip()
+ if message_id:
+ existing["messageId"] = message_id
+ grouped[key] = existing
+ continue
+ existing["occurrenceCount"] = int(existing.get("occurrenceCount") or 0) + occurrences
+ # 只要任一来源已确认可恢复，就不能被另一个旧的不可用记录覆盖。
+ if repairable and not bool(existing.get("repairable")):
+ existing["state"] = state
+ existing["reasonCode"] = reason_code
+ existing["repairable"] = True
+ return [grouped[key] for key in sorted(grouped)]
+
+
+def summarize_pending_media(conversations: dict[str, dict[str, Any]]) -> dict[str, Any]:
+ """汇总唯一缺失媒体和消息引用数，避免把重复表情误报成多个文件。"""
+
+ unique: dict[tuple[str, str], dict[str, Any]] = {}
+ by_kind: dict[str, dict[str, int]] = {}
+ for state in conversations.values():
+ if not isinstance(state, dict):
+ continue
+ for item in normalize_pending_media(state.get("pendingMedia") or []):
+ if bool(item.get("repairable")):
+ continue
+ kind = str(item.get("kind") or "")
+ ident = str(item.get("id") or "")
+ key = (kind, ident)
+ references = max(1, int(item.get("occurrenceCount") or 1))
+ bucket = by_kind.setdefault(kind, {"uniqueCount": 0, "referenceCount": 0})
+ bucket["referenceCount"] += references
+ if key not in unique:
+ unique[key] = dict(item)
+ bucket["uniqueCount"] += 1
+ else:
+ unique[key]["occurrenceCount"] = int(unique[key].get("occurrenceCount") or 0) + references
+ return {
+ "uniqueCount": len(unique),
+ "referenceCount": sum(int(item.get("occurrenceCount") or 0) for item in unique.values()),
+ "byKind": by_kind,
+ }
+
+
+@dataclass
+class ChatFolderContext:
+ account: str
+ folder_name: str
+ config: dict[str, Any]
+ config_hash: str
+ privacy_mode: bool
+ desktop_output: bool
+ exports_root: Path
+ target_root: Optional[Path]
+ old_state: dict[str, Any]
+ salt: str
+ missing_files: set[str] = field(default_factory=set)
+ reset_baseline: bool = False
+ selected_keys: set[str] = field(default_factory=set)
+ current_conversations: dict[str, dict[str, Any]] = field(default_factory=dict)
+ repair_candidates: list[dict[str, Any]] = field(default_factory=list)
+ history_synced: list[dict[str, Any]] = field(default_factory=list)
+ unresolved_media_conversations: list[dict[str, Any]] = field(default_factory=list)
+ unresolved_missing_owner_keys: set[str] = field(default_factory=set)
+ metadata_changed: bool = False
+
+ @property
+ def export_runtime_id(self) -> str:
+ return str(self.old_state.get("runtimeId") or hashlib.sha256(self.salt.encode("utf-8")).hexdigest()[:12])
+
+
+def _read_json_file(path: Path) -> dict[str, Any]:
+ try:
+ value = json.loads(path.read_text(encoding="utf-8"))
+ except Exception as exc:
+ raise ChatIncrementalError("incremental_baseline_invalid", "增量基线损坏，请选择新目录。") from exc
+ if not isinstance(value, dict):
+ raise ChatIncrementalError("incremental_baseline_invalid", "增量基线损坏，请选择新目录。")
+ return value
+
+
+def _baseline_is_owned(value: dict[str, Any]) -> bool:
+ return (
+ int(value.get("schemaVersion") or 0) == SCHEMA_VERSION
+ and str(value.get("artifactType") or "") == ARTIFACT_TYPE
+ and isinstance(value.get("conversations"), dict)
+ and isinstance(value.get("files"), dict)
+ )
+
+
+def _directory_has_user_files(path: Path) -> bool:
+ try:
+ return path.is_dir() and any(path.iterdir())
+ except Exception:
+ # 无法检查时按非空处理，避免在未知目录中覆盖用户文件。
+ return True
+
+
+def prepare_folder_context(
+ *,
+ account: str,
+ exports_root: Path,
+ requested_folder_name: str,
+ config: dict[str, Any],
+ privacy_mode: bool,
+ desktop_output: bool,
+ supplied_baseline: Optional[dict[str, Any]],
+ missing_files: list[str],
+ reset_baseline: bool,
+) -> ChatFolderContext:
+ folder_name = _safe_component(
+ requested_folder_name or f"微信聊天记录_{account}",
+ fallback="微信聊天记录",
+ )
+ if privacy_mode:
+ folder_name = f"微信聊天记录_隐私_{privacy_account_token(account)}"
+ exports_root = Path(exports_root).resolve()
+ nested_root = (exports_root / folder_name).resolve()
+ direct_state_path = exports_root / STATE_FILE_NAME
+ direct_state: dict[str, Any] = {}
+ if desktop_output and direct_state_path.is_file():
+ direct_state = _read_json_file(direct_state_path)
+
+ direct_matches = bool(
+ exports_root.name == folder_name
+ or (
+ direct_state
+ and str(direct_state.get("artifactType") or "") == ARTIFACT_TYPE
+ and (
+ str(direct_state.get("account") or "") == str(account or "")
+ or str(direct_state.get("accountFingerprint") or "") == account_fingerprint(account)
+ )
+ )
+ )
+ target_root = exports_root if desktop_output and direct_matches else (nested_root if desktop_output else None)
+
+ old_state: dict[str, Any] = {}
+ if desktop_output and target_root is not None:
+ disk_state_path = target_root / STATE_FILE_NAME
+ if disk_state_path.is_file():
+ old_state = direct_state if disk_state_path == direct_state_path and direct_state else _read_json_file(disk_state_path)
+ elif isinstance(supplied_baseline, dict):
+ old_state = dict(supplied_baseline)
+
+ owned = bool(old_state and _baseline_is_owned(old_state))
+ if old_state and not owned:
+ raise ChatIncrementalError("incremental_baseline_invalid", "增量基线损坏或不属于聊天导出，请选择新目录。")
+ if owned:
+ _validate_baseline_paths(old_state)
+
+ desired_hash = config_fingerprint(config)
+ if owned:
+ baseline_account = str(old_state.get("account") or "")
+ baseline_account_fingerprint = str(old_state.get("accountFingerprint") or "")
+ account_matches = (
+ baseline_account == str(account or "")
+ if baseline_account
+ else baseline_account_fingerprint == account_fingerprint(account)
+ )
+ if not account_matches:
+ raise ChatIncrementalError("incremental_account_mismatch", "该增量目录属于其他微信账号，请选择新目录。")
+ if str(old_state.get("configFingerprint") or "") != desired_hash and not reset_baseline:
+ raise ChatIncrementalError(
+ "incremental_config_mismatch",
+ "导出格式或筛选配置与该增量目录不一致，请选择新目录或重置后完整重建。",
+ )
+
+ if reset_baseline:
+ if old_state and not owned:
+ raise ChatIncrementalError("incremental_baseline_invalid", "无法安全重置损坏的增量目录，请选择空目录。")
+ if not owned and desktop_output and target_root is not None and _directory_has_user_files(target_root):
+ raise ChatIncrementalError(
+ "incremental_directory_not_empty",
+ "无法重置来源不明的非空目录，请选择新目录。",
+ )
+ old_state = {} if not owned else old_state
+ elif not old_state and desktop_output and target_root is not None and _directory_has_user_files(target_root):
+ raise ChatIncrementalError(
+ "incremental_directory_not_empty",
+ "所选目录非空且没有可识别的聊天增量基线，请选择其父目录或一个空目录。",
+ )
+
+ salt = str(old_state.get("conversationSalt") or uuid.uuid4().hex)
+ missing = {_require_managed_path(raw) for raw in missing_files}
+ if owned and desktop_output and target_root is not None:
+ managed_files = old_state.get("files") if isinstance(old_state.get("files"), dict) else {}
+ for raw_path, metadata in managed_files.items():
+ path = _require_managed_path(raw_path)
+ destination = (target_root / Path(*path.split("/"))).resolve()
+ if target_root not in destination.parents or not destination.is_file():
+ missing.add(path)
+ continue
+ try:
+ expected_size = int((metadata or {}).get("size"))
+ except Exception:
+ expected_size = -1
+ if expected_size >= 0 and destination.stat().st_size != expected_size:
+ missing.add(path)
+ return ChatFolderContext(
+ account=str(account or ""),
+ folder_name=str(old_state.get("folderName") or folder_name),
+ config=dict(config),
+ config_hash=desired_hash,
+ privacy_mode=bool(privacy_mode),
+ desktop_output=bool(desktop_output),
+ exports_root=exports_root,
+ target_root=target_root,
+ old_state=old_state,
+ salt=salt,
+ missing_files=missing,
+ reset_baseline=bool(reset_baseline),
+ )
+
+
+def missing_conversation_keys(
+ context: ChatFolderContext,
+ *,
+ preferred_keys: Optional[set[str]] = None,
+) -> set[str]:
+ """把全部缺失受管理文件反向映射到需要重建的会话。"""
+
+ missing = set(context.missing_files)
+ if not missing:
+ return set()
+
+ conversations = (
+ context.old_state.get("conversations")
+ if isinstance(context.old_state.get("conversations"), dict)
+ else {}
+ )
+ files = (
+ context.old_state.get("files")
+ if isinstance(context.old_state.get("files"), dict)
+ else {}
+ )
+ result: set[str] = set()
+ preferred = {str(key or "") for key in (preferred_keys or set())}
+
+ for raw_key, raw_state in conversations.items():
+ if not isinstance(raw_state, dict):
+ continue
+ key = str(raw_key or "")
+ managed = {
+ normalize_relative_path(path)
+ for path in (raw_state.get("managedFiles") or [])
+ }
+ directory = normalize_relative_path(raw_state.get("directory"))
+ if managed & missing or any(
+ directory and (path == directory or path.startswith(directory + "/"))
+ for path in missing
+ ):
+ result.add(key)
+
+ known_keys = {str(key or "") for key in conversations}
+ for path in missing:
+ if not _is_owned_resource_path(path):
+ continue
+ metadata = files.get(path) if isinstance(files.get(path), dict) else {}
+ owners = {
+ str(owner or "")
+ for owner in (metadata.get("owners") or [])
+ if str(owner or "") in known_keys
+ }
+ if owners:
+ # 共享资源只需一个所有者重新物化；优先复用已经需要重建或本次已选择的会话。
+ candidates = owners & result
+ if not candidates:
+ candidates = owners & preferred
+ result.add(sorted(candidates or owners)[0])
+ else:
+ # 老基线若没有资源归属信息，保守重建全部会话，避免缺失媒体永久无法补回。
+ result.update(known_keys)
+ return result
+
+
+def _safe_archive_infos(archive: zipfile.ZipFile) -> dict[str, zipfile.ZipInfo]:
+ result: dict[str, zipfile.ZipInfo] = {}
+ for info in archive.infolist():
+ if info.is_dir():
+ continue
+ path = normalize_relative_path(info.filename)
+ if path:
+ result[path] = info
+ return result
+
+
+def _file_meta(payload: bytes, *, owners: Optional[list[str]] = None) -> dict[str, Any]:
+ result: dict[str, Any] = {
+ "sha256": hashlib.sha256(payload).hexdigest(),
+ "size": len(payload),
+ }
+ if owners:
+ result["owners"] = sorted(set(owners))
+ return result
+
+
+def _resource_search_payload(payload: bytes, path: str) -> bytes:
+ """Excel 文件本身是 ZIP，需要展开 XML 后才能识别其中的资源引用。"""
+
+ if not str(path or "").lower().endswith(".xlsx"):
+ return payload
+ try:
+ with zipfile.ZipFile(io.BytesIO(payload), "r") as workbook:
+ return b"\n".join(
+ workbook.read(info)
+ for info in workbook.infolist()
+ if not info.is_dir() and info.filename.lower().endswith((".xml", ".rels"))
+ )
+ except Exception:
+ return payload
+
+
+def _is_owned_resource_path(path: str) -> bool:
+ return str(path or "").startswith(("media/", "avatars/"))
+
+
+def _is_conversation_path(path: str) -> bool:
+ return str(path or "").startswith("conversations/")
+
+
+def _conversation_owner(path: str, conversations: dict[str, dict[str, Any]]) -> str:
+ for key, state in conversations.items():
+ directory = normalize_relative_path(state.get("directory"))
+ if directory and (path == directory or path.startswith(directory + "/")):
+ return key
+ return ""
+
+
+def _write_staged_file(staging_dir: Path, relative: str, payload: bytes) -> Path:
+ destination = (staging_dir / Path(*relative.split("/"))).resolve()
+ if staging_dir not in destination.parents:
+ raise ChatIncrementalError("incremental_unsafe_path", "增量文件路径不安全。")
+ destination.parent.mkdir(parents=True, exist_ok=True)
+ destination.write_bytes(payload)
+ return destination
+
+
+def materialize_folder_archive(
+ *,
+ job: Any,
+ archive_path: Path,
+ context: ChatFolderContext,
+) -> Path:
+ staging_dir = Path(
+ tempfile.mkdtemp(
+ prefix=f".chat-folder-{job.export_id}-",
+ dir=str(context.exports_root),
+ )
+ ).resolve()
+ job.staging_dir = staging_dir
+
+ old_files = context.old_state.get("files") if isinstance(context.old_state.get("files"), dict) else {}
+ old_conversations = context.old_state.get("conversations") if isinstance(context.old_state.get("conversations"), dict) else {}
+ conversations: dict[str, dict[str, Any]] = {} if context.reset_baseline else {
+ str(key): dict(value)
+ for key, value in old_conversations.items()
+ if isinstance(value, dict)
+ }
+ conversations.update({str(key): dict(value) for key, value in context.current_conversations.items()})
+
+ current_files: dict[str, Any] = {} if context.reset_baseline else {
+ str(path): dict(meta)
+ for path, meta in old_files.items()
+ if normalize_relative_path(path) == str(path) and isinstance(meta, dict)
+ }
+ selected_old_managed: set[str] = set()
+ rendered_keys = {
+ key for key, state in context.current_conversations.items() if bool(state.get("rendered"))
+ }
+ append_keys = {
+ key for key, state in context.current_conversations.items()
+ if bool(state.get("rendered")) and bool(state.get("appendOnly"))
+ }
+ replaced_keys = rendered_keys - append_keys
+ for key in replaced_keys:
+ old = old_conversations.get(key) if isinstance(old_conversations.get(key), dict) else {}
+ selected_old_managed.update(
+ path
+ for raw in (old.get("managedFiles") or [])
+ if (path := normalize_relative_path(raw))
+ )
+ for path in selected_old_managed:
+ current_files.pop(path, None)
+
+ # 先撤销本轮重建会话对旧共享资源的引用；稍后再按新产物重新建立归属。
+ for path, meta in list(current_files.items()):
+ if not _is_owned_resource_path(path) or not isinstance(meta.get("owners"), list):
+ continue
+ meta["owners"] = sorted({str(owner) for owner in meta.get("owners") or []} - replaced_keys)
+
+ quiet_noop = bool(
+ context.old_state
+ and not context.reset_baseline
+ and not rendered_keys
+ and not context.repair_candidates
+ and not context.history_synced
+ and not context.missing_files
+ and not context.metadata_changed
+ )
+ preserve_dynamic_indexes = bool(
+ context.old_state
+ and not context.reset_baseline
+ and not rendered_keys
+ and not context.missing_files
+ )
+ staged_entries: list[dict[str, Any]] = []
+ staged_payloads: dict[str, bytes] = {}
+ new_managed: dict[str, set[str]] = {
+ key: {
+ path
+ for raw in (
+ (old_conversations.get(key) or {}).get("managedFiles")
+ if isinstance(old_conversations.get(key), dict)
+ else []
+ ) or []
+ if (path := normalize_relative_path(raw))
+ }
+ if key in append_keys
+ else set()
+ for key in rendered_keys
+ }
+ with zipfile.ZipFile(archive_path, "r") as archive:
+ infos = _safe_archive_infos(archive)
+ owner_search_payloads: dict[str, list[bytes]] = {key: [] for key in rendered_keys}
+ for path, info in infos.items():
+ if not _is_conversation_path(path):
+ continue
+ owner = _conversation_owner(path, context.current_conversations)
+ if not owner or owner not in rendered_keys:
+ continue
+ payload = archive.read(info)
+ owner_search_payloads.setdefault(owner, []).append(_resource_search_payload(payload, path))
+
+ for path, info in infos.items():
+ if path in {"manifest.json", "report.json"} or path.startswith("_integrity/"):
+ continue
+ owner = _conversation_owner(path, context.current_conversations)
+ if _is_conversation_path(path):
+ if not owner or owner not in rendered_keys:
+ continue
+ new_managed.setdefault(owner, set()).add(path)
+ payload = archive.read(info)
+ if _is_owned_resource_path(path):
+ path_bytes = path.encode("utf-8", errors="replace")
+ referenced_by = {
+ key
+ for key, search_payloads in owner_search_payloads.items()
+ if any(path_bytes in search_payload for search_payload in search_payloads)
+ }
+ explicit_owners = (getattr(job, "options", {}) or {}).get("_folderResourceOwners") or {}
+ for username in explicit_owners.get(path, []) if isinstance(explicit_owners, dict) else []:
+ key = conversation_key(salt=context.salt, username=str(username or ""))
+ if key in rendered_keys:
+ referenced_by.add(key)
+ previous_current = current_files.get(path) if isinstance(current_files.get(path), dict) else {}
+ owners = sorted(
+ {
+ str(value)
+ for value in (previous_current.get("owners") or [])
+ if str(value or "")
+ }
+ | referenced_by
+ )
+ else:
+ owners = [owner] if owner else []
+ meta = _file_meta(payload, owners=owners)
+ previous = old_files.get(path) if isinstance(old_files.get(path), dict) else {}
+ current_files[path] = meta
+ missing = path in context.missing_files
+ if preserve_dynamic_indexes and not missing and path in {"index.html", "index.xlsx"} and path in old_files:
+ current_files[path] = dict(previous)
+ continue
+ if (
+ not context.reset_baseline
+ and not missing
+ and str(previous.get("sha256") or "") == meta["sha256"]
+ and int(previous.get("size") or -1) == meta["size"]
+ and (
+ context.target_root is None
+ or (context.target_root / Path(*path.split("/"))).is_file()
+ )
+ ):
+ continue
+ staged_payloads[path] = payload
+
+ # 只有旧基线明确记录过归属的资源，才会在最后一个会话解除引用后被清理。
+ for path, meta in list(current_files.items()):
+ previous = old_files.get(path) if isinstance(old_files.get(path), dict) else {}
+ if (
+ _is_owned_resource_path(path)
+ and isinstance(previous.get("owners"), list)
+ and not list(meta.get("owners") or [])
+ ):
+ current_files.pop(path, None)
+ staged_payloads.pop(path, None)
+
+ for key in rendered_keys:
+ state = conversations.get(key) or {}
+ state["managedFiles"] = sorted(new_managed.get(key) or [])
+ conversations[key] = state
+
+ generated_at = datetime.now().isoformat(timespec="seconds")
+ unresolved_media = summarize_pending_media(conversations)
+ unresolved_conversations: list[dict[str, Any]] = []
+ for raw in context.unresolved_media_conversations:
+ key = str(raw.get("conversationKey") or "")
+ state = conversations.get(key) if isinstance(conversations.get(key), dict) else {}
+ pending = [
+ item
+ for item in normalize_pending_media(state.get("pendingMedia") or [])
+ if not bool(item.get("repairable"))
+ ]
+ if not pending:
+ continue
+ unresolved_conversations.append(
+ {
+ **dict(raw),
+ "uniqueCount": len(pending),
+ "referenceCount": sum(
+ max(1, int(item.get("occurrenceCount") or 1)) for item in pending
+ ),
+ }
+ )
+ manifest = {
+ "schemaVersion": 1,
+ "artifactType": ARTIFACT_TYPE,
+ "account": "hidden" if context.privacy_mode else context.account,
+ "format": context.config.get("format"),
+ "folderName": context.folder_name,
+ "updatedAt": generated_at,
+ "stats": {
+ "conversations": len(conversations),
+ "repairPending": len(context.repair_candidates),
+ "unresolvedMedia": int(unresolved_media.get("uniqueCount") or 0),
+ "unresolvedMediaReferences": int(unresolved_media.get("referenceCount") or 0),
+ },
+ }
+ persisted_repairs = [dict(item) for item in context.repair_candidates]
+ persisted_history = [dict(item) for item in context.history_synced]
+ if context.privacy_mode:
+ for item in [*persisted_repairs, *persisted_history]:
+ item["username"] = ""
+ item["displayName"] = ""
+ report = {
+ "schemaVersion": 1,
+ "updatedAt": manifest["updatedAt"],
+ "repairCandidates": persisted_repairs,
+ "historyChangesSynced": persisted_history,
+ "unresolvedMedia": unresolved_media,
+ }
+ for path, payload in {
+ "manifest.json": json.dumps(manifest, ensure_ascii=False, indent=2).encode("utf-8"),
+ "report.json": json.dumps(report, ensure_ascii=False, indent=2).encode("utf-8"),
+ }.items():
+ if quiet_noop and path in old_files:
+ current_files[path] = dict(old_files[path])
+ continue
+ meta = _file_meta(payload)
+ previous = old_files.get(path) if isinstance(old_files.get(path), dict) else {}
+ current_files[path] = meta
+ if (
+ context.reset_baseline
+ or str(previous.get("sha256") or "") != meta["sha256"]
+ or int(previous.get("size") or -1) != meta["size"]
+ or context.target_root is None
+ or not (context.target_root / path).is_file()
+ ):
+ staged_payloads[path] = payload
+
+ old_managed = set(old_files)
+ stale = sorted(
+ path
+ for path in old_managed - set(current_files)
+ if normalize_relative_path(path) == path
+ )
+ # 全局运行时升级也属于真实变更，必须最后更新基线，避免下次重复迁移。
+ quiet_noop = quiet_noop and not staged_payloads and not stale
+
+ for path, payload in sorted(staged_payloads.items()):
+ staged_path = _write_staged_file(staging_dir, path, payload)
+ file_id = uuid.uuid4().hex
+ job.staged_files[file_id] = staged_path
+ meta = current_files[path]
+ staged_entries.append(
+ {
+ "fileId": file_id,
+ "path": path,
+ "size": int(meta.get("size") or 0),
+ "sha256": str(meta.get("sha256") or ""),
+ }
+ )
+
+ persisted_conversations: dict[str, dict[str, Any]] = {}
+ for key, value in conversations.items():
+ cleaned = {
+ field_name: field_value
+ for field_name, field_value in value.items()
+ if field_name not in {"rendered", "appendOnly", "newMessageCount"}
+ }
+ persisted_conversations[key] = cleaned
+
+ state = {
+ "schemaVersion": SCHEMA_VERSION,
+ "artifactType": ARTIFACT_TYPE,
+ "account": "" if context.privacy_mode else context.account,
+ "accountFingerprint": account_fingerprint(context.account),
+ "folderName": context.folder_name,
+ "runtimeId": context.export_runtime_id,
+ "conversationSalt": context.salt,
+ "config": context.config,
+ "configFingerprint": context.config_hash,
+ "updatedAt": (
+ str(context.old_state.get("updatedAt") or generated_at)
+ if quiet_noop
+ else generated_at
+ ),
+ "conversations": persisted_conversations,
+ "files": current_files,
+ }
+ state_bytes = json.dumps(state, ensure_ascii=False, indent=2, sort_keys=True).encode("utf-8")
+ state_path = _write_staged_file(staging_dir, STATE_FILE_NAME, state_bytes)
+ state_file_id = uuid.uuid4().hex
+ job.staged_files[state_file_id] = state_path
+
+ updated_count = len(rendered_keys)
+ reused_count = max(0, len(context.selected_keys) - updated_count)
+ appended = sum(max(0, int(item.get("newMessageCount") or 0)) for item in context.current_conversations.values())
+ recovered_count = sum(1 for entry in staged_entries if str(entry.get("path") or "") in context.missing_files)
+ job.incremental = {
+ "messagesAdded": appended,
+ "conversationsUpdated": updated_count,
+ "conversationsReused": reused_count,
+ "conversationsRepairPending": len(context.repair_candidates),
+ "historyChangesSynced": len(context.history_synced),
+ "filesChanged": len(staged_entries),
+ "filesReused": max(0, len(current_files) - len(staged_entries)),
+ "filesRemoved": len(stale),
+ "filesRecovered": recovered_count,
+ }
+ job.repair_candidates = list(context.repair_candidates)
+ job.unresolved_media = {
+ **unresolved_media,
+ "conversations": unresolved_conversations,
+ }
+ job.change_manifest = {
+ "folderName": context.folder_name,
+ "files": staged_entries,
+ "stale": stale,
+ "state": {
+ "fileId": state_file_id,
+ "path": STATE_FILE_NAME,
+ "size": len(state_bytes),
+ "sha256": hashlib.sha256(state_bytes).hexdigest(),
+ "unchanged": quiet_noop,
+ },
+ "stats": dict(job.incremental),
+ }
+
+ if not context.desktop_output:
+ return staging_dir
+
+ target_root = context.target_root
+ if target_root is None:
+ raise ChatIncrementalError("incremental_target_missing", "增量目录不可用。")
+ target_root.mkdir(parents=True, exist_ok=True)
+ for entry in staged_entries:
+ source = job.staged_files[str(entry["fileId"])]
+ relative = normalize_relative_path(entry["path"])
+ destination = (target_root / Path(*relative.split("/"))).resolve()
+ if target_root not in destination.parents:
+ raise ChatIncrementalError("incremental_unsafe_path", "增量文件路径不安全。")
+ destination.parent.mkdir(parents=True, exist_ok=True)
+ os.replace(source, destination)
+
+ # 只清理由旧基线明确管理、且本轮已经失效的文件。
+ for relative in stale:
+ destination = (target_root / Path(*relative.split("/"))).resolve()
+ if target_root not in destination.parents or not destination.is_file():
+ continue
+ destination.unlink(missing_ok=True)
+
+ state_destination = target_root / STATE_FILE_NAME
+ if not quiet_noop or not state_destination.is_file():
+ os.replace(state_path, state_destination)
+ job.folder_path = target_root
+ job.staged_files = {}
+ shutil.rmtree(staging_dir, ignore_errors=True)
+ job.staging_dir = None
+ return target_root
+
+
+__all__ = [
+ "ARTIFACT_TYPE",
+ "ChatFolderContext",
+ "ChatIncrementalError",
+ "SCHEMA_VERSION",
+ "STATE_FILE_NAME",
+ "account_fingerprint",
+ "allocate_conversation_directory",
+ "build_config",
+ "config_fingerprint",
+ "conversation_key",
+ "materialize_folder_archive",
+ "missing_conversation_keys",
+ "normalize_pending_media",
+ "normalize_relative_path",
+ "prepare_folder_context",
+ "privacy_account_token",
+ "summarize_pending_media",
+]
diff --git a/src/wechat_decrypt_tool/routers/chat_export.py b/src/wechat_decrypt_tool/routers/chat_export.py
index dbbd36de..db355699 100644
--- a/src/wechat_decrypt_tool/routers/chat_export.py
+++ b/src/wechat_decrypt_tool/routers/chat_export.py
@@ -1,13 +1,14 @@
import asyncio
import json
import time
-from typing import Literal, Optional
+from typing import Any, Literal, Optional
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import FileResponse, StreamingResponse
from pydantic import BaseModel, Field, SecretStr
from ..chat_export_service import CHAT_EXPORT_MANAGER, get_chat_export_targets_preview
+from ..chat_incremental_export import ChatIncrementalError
from ..native_core_export import decode_export_content_key, erase_export_content_key
from ..path_fix import PathFixRoute
from ..voice_transcription import VoiceTranscriptionError, get_voice_transcription_service
@@ -16,6 +17,7 @@
ExportFormat = Literal["json", "txt", "html", "excel"]
ExportScope = Literal["selected", "all", "groups", "singles"]
+ExportOutputMode = Literal["zip", "folder"]
ChatSource = Literal["auto", "decrypted", "realtime"]
MediaKind = Literal["image", "emoji", "video", "video_thumb", "voice", "file"]
MessageType = Literal[
@@ -78,10 +80,25 @@ class ChatExportCreateRequest(BaseModel):
description="WEC1 的 32 字节 Base64 内容密钥；仅 encrypt=true 时使用",
)
transcribe_voice: bool = Field(False, description="使用本地 Whisper 将语音消息转成中文并写入导出文件")
+ output_mode: ExportOutputMode = Field("zip", description="输出方式：zip=全量压缩包；folder=可持续更新目录")
+ folder_name: Optional[str] = Field(None, description="增量导出根目录名")
+ baseline: Optional[dict[str, Any]] = Field(None, description="浏览器端读取的上轮聊天增量基线")
+ missing_files: list[str] = Field(default_factory=list, description="浏览器端发现缺失的受管理文件")
+ reset_baseline: bool = Field(False, description="明确重置基线并完整重建")
+ repair_usernames: list[str] = Field(default_factory=list, description="需要重建的可恢复差异会话")
+ recheck_media: bool = Field(False, description="明确重新探测待补媒体；不会把源端不可用项当成可修复差异")
-@router.post("/api/chat/exports", summary="创建聊天记录导出任务（离线 zip）")
+@router.post("/api/chat/exports", summary="创建聊天记录导出任务（ZIP 全量或增量目录）")
async def create_chat_export(req: ChatExportCreateRequest):
+ if req.baseline is not None:
+ baseline_size = len(json.dumps(req.baseline, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))
+ if baseline_size > 128 * 1024 * 1024:
+ raise HTTPException(status_code=413, detail={"code": "incremental_baseline_too_large", "message": "增量基线过大。"})
+ if len(req.missing_files) > 100_000:
+ raise HTTPException(status_code=413, detail={"code": "incremental_missing_files_too_many", "message": "缺失文件清单过大。"})
+ if req.output_mode == "folder" and req.encrypt:
+ raise HTTPException(status_code=400, detail={"code": "incremental_encryption_unsupported", "message": "增量目录不支持整包加密，请使用 ZIP 全量导出。"})
if req.transcribe_voice and not req.privacy_mode:
try:
await asyncio.to_thread(get_voice_transcription_service().ensure_available)
@@ -120,7 +137,17 @@ async def create_chat_export(req: ChatExportCreateRequest):
encrypt=bool(req.encrypt),
content_key=content_key,
transcribe_voice=req.transcribe_voice,
+ output_mode=req.output_mode,
+ folder_name=req.folder_name,
+ baseline=req.baseline,
+ missing_files=req.missing_files,
+ reset_baseline=bool(req.reset_baseline),
+ repair_usernames=req.repair_usernames,
+ recheck_media=bool(req.recheck_media),
)
+ except ChatIncrementalError as e:
+ erase_export_content_key(content_key)
+ raise HTTPException(status_code=409, detail={"code": e.code, "message": str(e)}) from e
except ValueError as e:
erase_export_content_key(content_key)
raise HTTPException(status_code=400, detail=str(e))
@@ -180,6 +207,34 @@ async def download_chat_export(export_id: str):
)
+@router.get("/api/chat/exports/{export_id}/files", summary="获取聊天增量目录变化文件清单")
+async def list_chat_export_files(export_id: str):
+ job = CHAT_EXPORT_MANAGER.get_job(str(export_id or "").strip())
+ if not job:
+ raise HTTPException(status_code=404, detail="Export not found.")
+ if job.status != "done" or not job.change_manifest or not job.staged_files:
+ raise HTTPException(status_code=409, detail="Incremental export not ready.")
+ return {"status": "success", "manifest": job.change_manifest}
+
+
+@router.get("/api/chat/exports/{export_id}/files/{file_id}", summary="下载单个聊天增量变化文件")
+async def download_chat_export_file(export_id: str, file_id: str):
+ path = CHAT_EXPORT_MANAGER.get_staged_file(
+ str(export_id or "").strip(),
+ str(file_id or "").strip(),
+ )
+ if path is None:
+ raise HTTPException(status_code=404, detail="Export file not found.")
+ return FileResponse(str(path), media_type="application/octet-stream", filename=path.name)
+
+
+@router.post("/api/chat/exports/{export_id}/commit", summary="确认浏览器已完成聊天增量目录写入")
+async def commit_chat_export_files(export_id: str):
+ if not CHAT_EXPORT_MANAGER.commit_staged_files(str(export_id or "").strip()):
+ raise HTTPException(status_code=404, detail="Export not found.")
+ return {"status": "success"}
+
+
@router.get("/api/chat/exports/{export_id}/events", summary="导出任务进度 SSE")
async def stream_chat_export_events(export_id: str, request: Request):
export_id = str(export_id or "").strip()
diff --git a/tests/test_chat_export_panel_frontend.py b/tests/test_chat_export_panel_frontend.py
index 74cf5756..4244f3ff 100644
--- a/tests/test_chat_export_panel_frontend.py
+++ b/tests/test_chat_export_panel_frontend.py
@@ -33,6 +33,55 @@ def test_scope_bulk_selection_button_never_wraps(self):
self.assertIn("min-width: 88px", dialog)
self.assertIn("white-space: nowrap", dialog)
+ def test_incremental_folder_mode_keeps_zip_as_default_and_writes_baseline_last(self):
+ dialog = (ROOT / "frontend" / "components" / "chat" / "ChatExportDialog.vue").read_text(encoding="utf-8")
+ export_state = (ROOT / "frontend" / "composables" / "chat" / "useChatExport.js").read_text(encoding="utf-8")
+ api_state = (ROOT / "frontend" / "composables" / "useApi.js").read_text(encoding="utf-8")
+
+ self.assertIn("const exportOutputMode = ref('zip')", export_state)
+ self.assertIn("ZIP 全量", dialog)
+ self.assertIn("增量目录", dialog)
+ self.assertIn("output_mode: exportOutputMode.value", export_state)
+ self.assertIn("folder_name:", export_state)
+ self.assertIn("repair_usernames:", export_state)
+ self.assertIn("recheck_media:", export_state)
+ self.assertIn("重新探测缺失媒体", dialog)
+ self.assertIn("修复可恢复差异", dialog)
+ self.assertIn("微信聊天记录_隐私_${privacyAccountToken(account)}", export_state)
+ self.assertIn("output_mode: data.output_mode === 'folder' ? 'folder' : 'zip'", api_state)
+ self.assertIn("baseline: data.baseline && typeof data.baseline === 'object'", api_state)
+ self.assertIn("repair_usernames: Array.isArray(data.repair_usernames)", api_state)
+ self.assertIn("recheck_media: !!data.recheck_media", api_state)
+ file_loop = export_state.index("for (const entry of files)")
+ stale_loop = export_state.index("for (const stalePath")
+ state_write = export_state.index("writeResponseToBrowserFile(root, CHAT_EXPORT_BASELINE_FILE")
+ commit = export_state.index("/commit`, { method: 'POST' }")
+ self.assertLess(file_loop, stale_loop)
+ self.assertLess(stale_loop, state_write)
+ self.assertLess(state_write, commit)
+
+ def test_incremental_result_separates_success_repair_and_unavailable_media(self):
+ dialog = (ROOT / "frontend" / "components" / "chat" / "ChatExportDialog.vue").read_text(encoding="utf-8")
+
+ summary = dialog.index('class="chat-export-folder-result__summary"')
+ followups = dialog.index('class="chat-export-folder-result__followups"')
+ details = dialog.index('class="chat-export-folder-result__details"')
+ self.assertLess(summary, followups)
+ self.assertLess(followups, details)
+ self.assertIn("已确认修复会产生变化，仅重建对应会话。", dialog)
+ self.assertIn("源端暂不可用，重复修复不会改变结果。", dialog)
+ self.assertIn("查看完整任务说明", dialog)
+
+ def test_incremental_baseline_card_uses_compact_status_and_custom_checkbox(self):
+ dialog = (ROOT / "frontend" / "components" / "chat" / "ChatExportDialog.vue").read_text(encoding="utf-8")
+
+ self.assertIn('class="chat-export-incremental-card__folder"', dialog)
+ self.assertIn('class="chat-export-baseline-status"', dialog)
+ self.assertIn(':data-status="exportBaselineStatus"', dialog)
+ self.assertIn("auto: '自动检查基线'", dialog)
+ self.assertIn('type="checkbox" class="sr-only"', dialog)
+ self.assertIn(".chat-export-reset-option:focus-within", dialog)
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_chat_incremental_export.py b/tests/test_chat_incremental_export.py
new file mode 100644
index 00000000..6f882507
--- /dev/null
+++ b/tests/test_chat_incremental_export.py
@@ -0,0 +1,1295 @@
+import hashlib
+import io
+import json
+import os
+import sqlite3
+import threading
+import time
+import unittest
+import zipfile
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest import mock
+
+from test_chat_export_message_types_semantics import TestChatExportMessageTypesSemantics as _BaseChatExportTest
+
+
+class TestChatIncrementalExport(unittest.TestCase):
+ _prepare_account = _BaseChatExportTest._prepare_account
+ _reload_export_modules = _BaseChatExportTest._reload_export_modules
+ _seed_contact_db = _BaseChatExportTest._seed_contact_db
+ _seed_session_db = _BaseChatExportTest._seed_session_db
+ _seed_message_db = _BaseChatExportTest._seed_message_db
+ _seed_media_files = _BaseChatExportTest._seed_media_files
+ _seed_wxid_media_files = _BaseChatExportTest._seed_wxid_media_files
+ _seed_source_info = _BaseChatExportTest._seed_source_info
+
+ def _wait_for_job(self, manager, export_id: str):
+ for _ in range(400):
+ job = manager.get_job(export_id)
+ if job and job.status in {"done", "error", "cancelled"}:
+ return job
+ time.sleep(0.05)
+ self.fail("incremental export job did not finish in time")
+
+ def _create_folder_job(
+ self,
+ manager,
+ *,
+ account: str,
+ username: str | None = None,
+ usernames=None,
+ output_dir: Path,
+ export_format: str = "json",
+ privacy_mode: bool = False,
+ reset_baseline: bool = False,
+ repair_usernames=None,
+ message_types=None,
+ include_media: bool = False,
+ missing_files=None,
+ baseline=None,
+ ):
+ selected_usernames = list(usernames or ([username] if username else []))
+ job = manager.create_job(
+ account=account,
+ source="decrypted",
+ scope="selected",
+ usernames=selected_usernames,
+ export_format=export_format,
+ start_time=None,
+ end_time=None,
+ include_hidden=False,
+ include_official=False,
+ include_media=include_media,
+ media_kinds=["image", "emoji", "video", "video_thumb", "voice", "file"] if include_media else [],
+ message_types=list(message_types or ["text"]),
+ output_dir=str(output_dir) if output_dir is not None else None,
+ allow_process_key_extract=False,
+ download_remote_media=False,
+ html_page_size=1000,
+ privacy_mode=privacy_mode,
+ file_name=None,
+ output_mode="folder",
+ folder_name="聊天增量测试",
+ reset_baseline=reset_baseline,
+ repair_usernames=list(repair_usernames or []),
+ missing_files=list(missing_files or []),
+ baseline=baseline,
+ )
+ return self._wait_for_job(manager, job.export_id)
+
+ @staticmethod
+ def _message_table(username: str) -> str:
+ return f"msg_{hashlib.md5(username.encode('utf-8')).hexdigest()}"
+
+ def _add_conversation(self, account_dir: Path, *, username: str, display_name: str, local_id: int = 20) -> None:
+ connection = sqlite3.connect(str(account_dir / "contact.db"))
+ try:
+ connection.execute(
+ "INSERT INTO contact VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+ (username, "", display_name, "", 1, 0, "", ""),
+ )
+ connection.commit()
+ finally:
+ connection.close()
+
+ connection = sqlite3.connect(str(account_dir / "session.db"))
+ try:
+ connection.execute("INSERT INTO SessionTable VALUES (?, ?, ?)", (username, 0, 1735689700))
+ connection.commit()
+ finally:
+ connection.close()
+
+ table = self._message_table(username)
+ connection = sqlite3.connect(str(account_dir / "message_0.db"))
+ try:
+ connection.execute("INSERT INTO Name2Id(rowid, user_name) VALUES (?, ?)", (local_id, username))
+ connection.execute(
+ f"""
+ CREATE TABLE {table} (
+ local_id INTEGER,
+ server_id INTEGER,
+ local_type INTEGER,
+ sort_seq INTEGER,
+ real_sender_id INTEGER,
+ create_time INTEGER,
+ message_content TEXT,
+ compress_content BLOB
+ )
+ """
+ )
+ connection.execute(
+ f"INSERT INTO {table} VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+ (1, 2001, 1, 1, local_id, 1735689701, f"{display_name}的文本", None),
+ )
+ connection.commit()
+ finally:
+ connection.close()
+
+ @staticmethod
+ def _managed_message_file(folder: Path, suffix: str) -> Path:
+ matches = list((folder / "conversations").glob(f"*/messages.{suffix}"))
+ assert len(matches) == 1
+ return matches[0]
+
+ def test_unavailable_pending_media_is_deduplicated_without_repair_prompt(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account = "wxid_pending_account"
+ username = "wxid_pending_friend"
+ self._prepare_account(root, account=account, username=username)
+ output_dir = root / "exports"
+ previous = os.environ.get("WECHAT_TOOL_DATA_DIR")
+ try:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = str(root)
+ service = self._reload_export_modules()
+ first = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ )
+ self.assertEqual(first.status, "done", msg=first.error)
+ state_path = first.folder_path / ".wechat-chat-export.json"
+ baseline = json.loads(state_path.read_text(encoding="utf-8"))
+ conversation = next(iter(baseline["conversations"].values()))
+ missing_id = "f" * 32
+ conversation["pendingMedia"] = [
+ {"kind": "emoji", "id": missing_id, "messageId": "1"},
+ {"kind": "emoji", "id": missing_id, "messageId": "2"},
+ {"kind": "emoji", "id": missing_id, "messageId": "3"},
+ ]
+ state_path.write_text(json.dumps(baseline, ensure_ascii=False, indent=2), encoding="utf-8")
+
+ checked = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ )
+ self.assertEqual(checked.status, "done", msg=checked.error)
+ self.assertFalse(checked.repair_candidates)
+ self.assertEqual(checked.unresolved_media.get("uniqueCount"), 1)
+ self.assertEqual(checked.unresolved_media.get("referenceCount"), 3)
+ self.assertIn("重复修复不会产生变化", checked.warning)
+
+ migrated = json.loads(state_path.read_text(encoding="utf-8"))
+ pending = next(iter(migrated["conversations"].values()))["pendingMedia"]
+ self.assertEqual(len(pending), 1)
+ self.assertEqual(pending[0].get("occurrenceCount"), 3)
+ self.assertEqual(pending[0].get("state"), "source_unavailable")
+ self.assertFalse(pending[0].get("repairable"))
+
+ before_repeat = (state_path.stat().st_mtime_ns, state_path.read_bytes())
+ repeated = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ )
+ self.assertEqual(repeated.status, "done", msg=repeated.error)
+ self.assertFalse(repeated.repair_candidates)
+ self.assertEqual(repeated.unresolved_media.get("uniqueCount"), 1)
+ self.assertEqual((state_path.stat().st_mtime_ns, state_path.read_bytes()), before_repeat)
+ finally:
+ if previous is None:
+ os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
+ else:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = previous
+
+ def test_html_direct_emoji_uses_remote_download_before_marking_missing(self):
+ with TemporaryDirectory() as td:
+ account_dir = Path(td) / "wxid_remote_emoji"
+ account_dir.mkdir(parents=True)
+ service = self._reload_export_modules()
+ job = service.ExportJob(
+ export_id="remote-emoji",
+ account=account_dir.name,
+ options={"outputMode": "folder", "downloadRemoteMedia": True},
+ )
+ message = {
+ "id": "1",
+ "renderType": "emoji",
+ "emojiMd5": "f" * 32,
+ "emojiUrl": "https://example.com/emoji.png",
+ }
+ report = {"missingMedia": [], "errors": []}
+ with io.BytesIO() as buffer, zipfile.ZipFile(buffer, "w") as archive:
+ with mock.patch.object(
+ service,
+ "_download_remote_image_to_zip",
+ return_value="media/remote/emoji.png",
+ ) as downloader:
+ service._attach_offline_media(
+ zf=archive,
+ account_dir=account_dir,
+ conv_username="wxid_friend",
+ owner_username="wxid_friend",
+ msg=message,
+ media_written={},
+ report=report,
+ media_kinds=["emoji"],
+ allow_process_key_extract=False,
+ media_db_path=account_dir / "media.db",
+ media_index=None,
+ remote_written={},
+ lock=threading.Lock(),
+ job=job,
+ )
+ downloader.assert_called_once()
+ self.assertEqual(message["offlineMedia"][0]["path"], "media/remote/emoji.png")
+ self.assertEqual(job.progress.media_copied, 1)
+ self.assertEqual(job.progress.media_missing, 0)
+ self.assertFalse(report["missingMedia"])
+
+ def test_repair_prompt_only_appears_after_pending_media_becomes_recoverable(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account = "wxid_recoverable_account"
+ username = "wxid_recoverable_friend"
+ self._prepare_account(root, account=account, username=username)
+ output_dir = root / "exports"
+ previous = os.environ.get("WECHAT_TOOL_DATA_DIR")
+ try:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = str(root)
+ service = self._reload_export_modules()
+ first = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ include_media=True,
+ message_types=["text", "image"],
+ )
+ self.assertEqual(first.status, "done", msg=first.error)
+ state_path = first.folder_path / ".wechat-chat-export.json"
+ baseline = json.loads(state_path.read_text(encoding="utf-8"))
+ conversation = next(iter(baseline["conversations"].values()))
+ conversation["pendingMedia"] = [
+ {"kind": "image", "id": "a" * 32, "messageId": "2"},
+ ]
+ state_path.write_text(json.dumps(baseline, ensure_ascii=False, indent=2), encoding="utf-8")
+
+ checked = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ include_media=True,
+ message_types=["text", "image"],
+ )
+ self.assertEqual(checked.status, "done", msg=checked.error)
+ self.assertEqual(len(checked.repair_candidates), 1)
+ self.assertEqual(checked.repair_candidates[0].get("reasons"), ["media_recoverable"])
+ self.assertEqual(checked.unresolved_media.get("uniqueCount"), 0)
+
+ repaired = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ include_media=True,
+ message_types=["text", "image"],
+ repair_usernames=[username],
+ )
+ self.assertEqual(repaired.status, "done", msg=repaired.error)
+ self.assertFalse(repaired.repair_candidates)
+ repaired_state = json.loads(state_path.read_text(encoding="utf-8"))
+ self.assertFalse(next(iter(repaired_state["conversations"].values()))["pendingMedia"])
+ finally:
+ if previous is None:
+ os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
+ else:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = previous
+
+ def test_folder_missing_progress_is_unique_but_zip_keeps_legacy_reference_count(self):
+ with TemporaryDirectory() as td:
+ account_dir = Path(td) / "wxid_missing_counts"
+ account_dir.mkdir(parents=True)
+ service = self._reload_export_modules()
+ for output_mode, expected_missing in (("folder", 1), ("zip", 2)):
+ with self.subTest(output_mode=output_mode):
+ job = service.ExportJob(
+ export_id=f"missing-{output_mode}",
+ account=account_dir.name,
+ options={"outputMode": output_mode, "downloadRemoteMedia": False},
+ )
+ report = {"missingMedia": [], "errors": []}
+ media_written = {}
+ with io.BytesIO() as buffer, zipfile.ZipFile(buffer, "w") as archive:
+ for message_id in ("1", "2"):
+ service._attach_offline_media(
+ zf=archive,
+ account_dir=account_dir,
+ conv_username="wxid_friend",
+ owner_username="wxid_friend",
+ msg={
+ "id": message_id,
+ "renderType": "emoji",
+ "emojiMd5": "f" * 32,
+ },
+ media_written=media_written,
+ report=report,
+ media_kinds=["emoji"],
+ allow_process_key_extract=False,
+ media_db_path=account_dir / "media.db",
+ media_index=None,
+ lock=threading.Lock(),
+ job=job,
+ )
+ self.assertEqual(job.progress.media_missing, expected_missing)
+ self.assertEqual(job.progress.media_missing_references, 2)
+ self.assertEqual(len(report["missingMedia"]), 2)
+
+ def test_first_run_and_no_change_reuse_all_formats(self):
+ for export_format, suffix in (("html", "html"), ("json", "json"), ("txt", "txt"), ("excel", "xlsx")):
+ with self.subTest(export_format=export_format), TemporaryDirectory() as td:
+ root = Path(td)
+ account = "wxid_incremental"
+ username = "wxid_friend"
+ self._prepare_account(root, account=account, username=username)
+ output_dir = root / "exports"
+
+ previous = os.environ.get("WECHAT_TOOL_DATA_DIR")
+ try:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = str(root)
+ service = self._reload_export_modules()
+ first = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ export_format=export_format,
+ )
+ self.assertEqual(first.status, "done", msg=first.error)
+ folder = output_dir / "聊天增量测试"
+ message_file = self._managed_message_file(folder, suffix)
+ before = (message_file.stat().st_mtime_ns, message_file.read_bytes())
+
+ second = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ export_format=export_format,
+ )
+ self.assertEqual(second.status, "done", msg=second.error)
+ after = (message_file.stat().st_mtime_ns, message_file.read_bytes())
+ self.assertEqual(before, after)
+ self.assertEqual(second.incremental.get("filesChanged"), 0)
+ self.assertEqual(second.incremental.get("conversationsUpdated"), 0)
+ self.assertEqual(second.incremental.get("conversationsReused"), 1)
+
+ if export_format == "json":
+ json.loads(message_file.read_text(encoding="utf-8"))
+ elif export_format == "txt":
+ message_file.read_text(encoding="utf-8")
+ elif export_format == "excel":
+ import zipfile
+
+ self.assertTrue(zipfile.is_zipfile(message_file))
+ else:
+ html_text = message_file.read_text(encoding="utf-8").lower()
+ self.assertIn("", html_text)
+ self.assertNotIn("data-wce-sri", html_text)
+ self.assertTrue((folder / "assets" / "chat-export.css").is_file())
+ self.assertTrue((folder / "assets" / "chat-export.js").is_file())
+ self.assertFalse((folder / "_integrity").exists())
+ finally:
+ if previous is None:
+ os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
+ else:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = previous
+
+ def test_html_folder_runtime_disables_zip_integrity_and_migrates_old_asset(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account = "wxid_incremental"
+ username = "wxid_friend"
+ self._prepare_account(root, account=account, username=username)
+ output_dir = root / "exports"
+
+ previous = os.environ.get("WECHAT_TOOL_DATA_DIR")
+ try:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = str(root)
+ service = self._reload_export_modules()
+ first = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ export_format="html",
+ )
+ self.assertEqual(first.status, "done", msg=first.error)
+ folder = output_dir / "聊天增量测试"
+ runtime_path = folder / "assets" / "chat-export.js"
+ folder_runtime = runtime_path.read_text(encoding="utf-8")
+ self.assertIn("data-wce-folder-mode", folder_runtime)
+ self.assertIn("window.__WCE_VERIFY_FRAGMENT__ = () => true", folder_runtime)
+ self.assertNotIn("const integrityOk = await initExportIntegrity()", folder_runtime)
+
+ html_files = [folder / "index.html", self._managed_message_file(folder, "html")]
+ for html_file in html_files:
+ html_text = html_file.read_text(encoding="utf-8")
+ self.assertNotIn("data-wce-sri", html_text)
+ self.assertNotIn("data-wce-integrity-bundle", html_text)
+ message_before = (html_files[1].stat().st_mtime_ns, html_files[1].read_bytes())
+
+ # 模拟已经导出的旧版目录：文件存在且基线摘要也匹配，但运行时仍会阻断目录页。
+ old_runtime = service._html_export_runtime_js(service._load_wce_integrity_native())
+ old_runtime_bytes = old_runtime.encode("utf-8")
+ runtime_path.write_bytes(old_runtime_bytes)
+ state_path = folder / ".wechat-chat-export.json"
+ baseline = json.loads(state_path.read_text(encoding="utf-8"))
+ baseline["files"]["assets/chat-export.js"] = {
+ "sha256": hashlib.sha256(old_runtime_bytes).hexdigest(),
+ "size": len(old_runtime_bytes),
+ }
+ state_path.write_text(
+ json.dumps(baseline, ensure_ascii=False, indent=2, sort_keys=True),
+ encoding="utf-8",
+ )
+
+ migrated = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ export_format="html",
+ )
+ self.assertEqual(migrated.status, "done", msg=migrated.error)
+ self.assertEqual(runtime_path.read_text(encoding="utf-8"), folder_runtime)
+ self.assertEqual((html_files[1].stat().st_mtime_ns, html_files[1].read_bytes()), message_before)
+ self.assertEqual(migrated.incremental.get("filesChanged"), 1)
+ migrated_baseline = json.loads(state_path.read_text(encoding="utf-8"))
+ self.assertEqual(
+ migrated_baseline["files"]["assets/chat-export.js"]["sha256"],
+ hashlib.sha256(runtime_path.read_bytes()).hexdigest(),
+ )
+
+ no_change = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ export_format="html",
+ )
+ self.assertEqual(no_change.status, "done", msg=no_change.error)
+ self.assertEqual(no_change.incremental.get("filesChanged"), 0)
+ finally:
+ if previous is None:
+ os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
+ else:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = previous
+
+ def test_html_folder_uses_one_shared_session_catalog_across_incremental_pages(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account = "wxid_incremental"
+ first_username = "wxid_friend"
+ second_username = "wxid_second_friend"
+ account_dir = self._prepare_account(root, account=account, username=first_username)
+ output_dir = root / "exports"
+
+ previous = os.environ.get("WECHAT_TOOL_DATA_DIR")
+ try:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = str(root)
+ service = self._reload_export_modules()
+ first = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=first_username,
+ output_dir=output_dir,
+ export_format="html",
+ )
+ self.assertEqual(first.status, "done", msg=first.error)
+ folder = output_dir / "聊天增量测试"
+ first_page = self._managed_message_file(folder, "html")
+
+ # 模拟旧版增量页：聊天正文仍受基线管理，但页面本身还没有公共目录脚本标签。
+ legacy_text = first_page.read_text(encoding="utf-8").replace(
+ ' \n',
+ "",
+ )
+ self.assertNotIn("data-wce-folder-sessions", legacy_text)
+ first_page.write_text(legacy_text, encoding="utf-8")
+ state_path = folder / ".wechat-chat-export.json"
+ baseline = json.loads(state_path.read_text(encoding="utf-8"))
+ first_page_relative = first_page.relative_to(folder).as_posix()
+ first_page_bytes = first_page.read_bytes()
+ baseline["files"][first_page_relative] = {
+ "sha256": hashlib.sha256(first_page_bytes).hexdigest(),
+ "size": len(first_page_bytes),
+ }
+ state_path.write_text(
+ json.dumps(baseline, ensure_ascii=False, indent=2, sort_keys=True),
+ encoding="utf-8",
+ )
+ legacy_before = (first_page.stat().st_mtime_ns, first_page_bytes)
+
+ self._add_conversation(
+ account_dir,
+ username=second_username,
+ display_name="第二个联系人",
+ )
+ updated = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=second_username,
+ output_dir=output_dir,
+ export_format="html",
+ )
+ self.assertEqual(updated.status, "done", msg=updated.error)
+ self.assertEqual((first_page.stat().st_mtime_ns, first_page.read_bytes()), legacy_before)
+
+ catalog_path = folder / "assets" / "chat-sessions.js"
+ catalog_text = catalog_path.read_text(encoding="utf-8")
+ prefix = "window.__WCE_FOLDER_SESSIONS__="
+ self.assertTrue(catalog_text.startswith(prefix))
+ catalog = json.loads(catalog_text[len(prefix):].rstrip(";\r\n"))
+ self.assertEqual(len(catalog.get("items") or []), 2)
+ catalog_directories = {
+ str(item.get("convDir") or "")
+ for item in catalog.get("items") or []
+ }
+ current_baseline = json.loads(state_path.read_text(encoding="utf-8"))
+ self.assertEqual(
+ catalog_directories,
+ {
+ str(value.get("directory") or "")
+ for value in current_baseline.get("conversations", {}).values()
+ },
+ )
+
+ runtime_text = (folder / "assets" / "chat-export.js").read_text(encoding="utf-8")
+ self.assertIn("loadFolderSessionCatalog", runtime_text)
+ self.assertIn("syncFolderSessionCatalog", runtime_text)
+ self.assertIn("new URL('chat-sessions.js', wceFolderRuntimeSrc)", runtime_text)
+
+ second_page = next(
+ page
+ for page in (folder / "conversations").glob("*/messages.html")
+ if page != first_page
+ )
+ self.assertIn("data-wce-folder-sessions", second_page.read_text(encoding="utf-8"))
+
+ no_change = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=second_username,
+ output_dir=output_dir,
+ export_format="html",
+ )
+ self.assertEqual(no_change.status, "done", msg=no_change.error)
+ self.assertEqual(no_change.incremental.get("filesChanged"), 0)
+ finally:
+ if previous is None:
+ os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
+ else:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = previous
+
+ def test_html_folder_shared_session_catalog_respects_privacy_mode(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account = "wxid_private_account"
+ username = "wxid_private_friend"
+ self._prepare_account(root, account=account, username=username)
+ output_dir = root / "exports"
+
+ previous = os.environ.get("WECHAT_TOOL_DATA_DIR")
+ try:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = str(root)
+ service = self._reload_export_modules()
+ job = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ export_format="html",
+ privacy_mode=True,
+ )
+ self.assertEqual(job.status, "done", msg=job.error)
+ catalog_text = (job.folder_path / "assets" / "chat-sessions.js").read_text(encoding="utf-8")
+ self.assertNotIn(account, catalog_text)
+ self.assertNotIn(username, catalog_text)
+ self.assertNotIn("测试好友", catalog_text)
+ self.assertNotIn("普通文本消息", catalog_text)
+ catalog = json.loads(
+ catalog_text.removeprefix("window.__WCE_FOLDER_SESSIONS__=").rstrip(";\r\n")
+ )
+ self.assertEqual(len(catalog.get("items") or []), 1)
+ item = catalog["items"][0]
+ self.assertEqual(item.get("username"), "")
+ self.assertEqual(item.get("avatarPath"), "")
+ self.assertEqual(item.get("previewText"), "")
+ finally:
+ if previous is None:
+ os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
+ else:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = previous
+
+ def test_html_existing_conversation_reads_and_renders_only_new_messages(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account = "wxid_incremental"
+ username = "wxid_friend"
+ account_dir = self._prepare_account(root, account=account, username=username)
+ output_dir = root / "exports"
+
+ previous = os.environ.get("WECHAT_TOOL_DATA_DIR")
+ try:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = str(root)
+ service = self._reload_export_modules()
+ first = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ export_format="html",
+ )
+ self.assertEqual(first.status, "done", msg=first.error)
+ folder = output_dir / "聊天增量测试"
+ message_file = self._managed_message_file(folder, "html")
+ self.assertIn("普通文本消息", message_file.read_text(encoding="utf-8"))
+ first_baseline = json.loads((folder / ".wechat-chat-export.json").read_text(encoding="utf-8"))
+ first_conversation = next(iter(first_baseline["conversations"].values()))
+ first_watermark_time = int(first_conversation["watermark"][0])
+
+ table = self._message_table(username)
+ connection = sqlite3.connect(str(account_dir / "message_0.db"))
+ try:
+ connection.execute(
+ f"INSERT INTO {table} VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+ (8, 1008, 1, 8, 2, 1735689608, "只渲染这一条增量消息", None),
+ )
+ connection.commit()
+ finally:
+ connection.close()
+
+ original_full_probe = service._probe_incremental_conversation
+ original_estimator = service._estimate_conversation_message_count
+ estimate_start_times = []
+
+ def reject_full_probe(**_kwargs):
+ raise AssertionError("已有分页会话不应重新扫描完整历史")
+
+ def track_incremental_estimate(**kwargs):
+ estimate_start_times.append(kwargs.get("start_time"))
+ return original_estimator(**kwargs)
+
+ service._probe_incremental_conversation = reject_full_probe
+ service._estimate_conversation_message_count = track_incremental_estimate
+ try:
+ appended = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ export_format="html",
+ )
+ self.assertEqual(appended.status, "done", msg=appended.error)
+ self.assertEqual(appended.incremental.get("messagesAdded"), 1)
+ self.assertEqual(appended.progress.messages_exported, 1)
+ self.assertEqual(appended.progress.current_conversation_messages_total, 1)
+ self.assertTrue(estimate_start_times)
+ self.assertEqual(estimate_start_times[0], first_watermark_time)
+
+ current_html = message_file.read_text(encoding="utf-8")
+ self.assertIn("只渲染这一条增量消息", current_html)
+ self.assertNotIn("普通文本消息", current_html)
+ page_file = message_file.parent / "pages" / "page-0001.js"
+ self.assertTrue(page_file.is_file())
+ self.assertIn("普通文本消息", page_file.read_text(encoding="utf-8"))
+ self.assertIn('"totalPages": 2', current_html)
+
+ before_noop = (message_file.stat().st_mtime_ns, message_file.read_bytes())
+ no_change = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ export_format="html",
+ )
+ self.assertEqual(no_change.status, "done", msg=no_change.error)
+ self.assertEqual(no_change.incremental.get("filesChanged"), 0)
+ self.assertEqual((message_file.stat().st_mtime_ns, message_file.read_bytes()), before_noop)
+ finally:
+ service._probe_incremental_conversation = original_full_probe
+ service._estimate_conversation_message_count = original_estimator
+ finally:
+ if previous is None:
+ os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
+ else:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = previous
+
+ def test_new_message_then_history_repair(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account = "wxid_incremental"
+ username = "wxid_friend"
+ account_dir = self._prepare_account(root, account=account, username=username)
+ output_dir = root / "exports"
+
+ previous = os.environ.get("WECHAT_TOOL_DATA_DIR")
+ try:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = str(root)
+ service = self._reload_export_modules()
+ first = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ )
+ self.assertEqual(first.status, "done", msg=first.error)
+ message_file = self._managed_message_file(output_dir / "聊天增量测试", "json")
+
+ table = self._message_table(username)
+ connection = sqlite3.connect(str(account_dir / "message_0.db"))
+ try:
+ connection.execute(
+ f"INSERT INTO {table} VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+ (8, 1008, 1, 8, 2, 1735689608, "增量新增消息", None),
+ )
+ connection.commit()
+ finally:
+ connection.close()
+
+ appended = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ )
+ self.assertEqual(appended.status, "done", msg=appended.error)
+ self.assertEqual(appended.incremental.get("messagesAdded"), 1)
+ payload = json.loads(message_file.read_text(encoding="utf-8"))
+ contents = [str(item.get("content") or "") for item in payload.get("messages", [])]
+ self.assertEqual(contents.count("增量新增消息"), 1)
+
+ before_repair = message_file.read_bytes()
+ connection = sqlite3.connect(str(account_dir / "message_0.db"))
+ try:
+ connection.execute(
+ f"UPDATE {table} SET message_content = ? WHERE local_id = 4",
+ ("历史消息已修改",),
+ )
+ connection.commit()
+ finally:
+ connection.close()
+
+ detected = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ )
+ self.assertEqual(detected.status, "done", msg=detected.error)
+ self.assertEqual(message_file.read_bytes(), before_repair)
+ self.assertEqual(len(detected.repair_candidates), 1)
+
+ repaired = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ repair_usernames=[username],
+ )
+ self.assertEqual(repaired.status, "done", msg=repaired.error)
+ payload = json.loads(message_file.read_text(encoding="utf-8"))
+ contents = [str(item.get("content") or "") for item in payload.get("messages", [])]
+ self.assertIn("历史消息已修改", contents)
+
+ connection = sqlite3.connect(str(account_dir / "message_0.db"))
+ try:
+ connection.execute(
+ f"UPDATE {table} SET message_content = ? WHERE local_id = 4",
+ ("历史消息再次修改",),
+ )
+ connection.execute(
+ f"INSERT INTO {table} VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+ (9, 1009, 1, 9, 2, 1735689609, "历史变化并存的新消息", None),
+ )
+ connection.commit()
+ finally:
+ connection.close()
+ synchronized = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ )
+ self.assertEqual(synchronized.status, "done", msg=synchronized.error)
+ self.assertEqual(synchronized.incremental.get("messagesAdded"), 1)
+ self.assertEqual(synchronized.incremental.get("historyChangesSynced"), 1)
+ self.assertFalse(synchronized.repair_candidates)
+ payload = json.loads(message_file.read_text(encoding="utf-8"))
+ contents = [str(item.get("content") or "") for item in payload.get("messages", [])]
+ self.assertIn("历史消息再次修改", contents)
+ self.assertEqual(contents.count("历史变化并存的新消息"), 1)
+ finally:
+ if previous is None:
+ os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
+ else:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = previous
+ def test_config_conflict_and_privacy_baseline(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account = "wxid_private_account"
+ username = "wxid_private_friend"
+ self._prepare_account(root, account=account, username=username)
+ output_dir = root / "exports"
+
+ previous = os.environ.get("WECHAT_TOOL_DATA_DIR")
+ try:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = str(root)
+ service = self._reload_export_modules()
+ first = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ privacy_mode=True,
+ )
+ self.assertEqual(first.status, "done", msg=first.error)
+ baseline_text = (first.folder_path / ".wechat-chat-export.json").read_text(encoding="utf-8")
+ self.assertNotIn(account, baseline_text)
+ self.assertNotIn(username, baseline_text)
+ self.assertNotIn("测试好友", baseline_text)
+ self.assertNotIn("普通文本消息", baseline_text)
+ user_file = first.folder_path / "用户保留文件.txt"
+ user_file.write_text("不要删除", encoding="utf-8")
+
+ with self.assertRaises(service.ChatIncrementalError) as captured:
+ self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ privacy_mode=True,
+ message_types=["text", "system"],
+ )
+ self.assertEqual(captured.exception.code, "incremental_config_mismatch")
+ rebuilt = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=username,
+ output_dir=output_dir,
+ privacy_mode=True,
+ message_types=["text", "system"],
+ reset_baseline=True,
+ )
+ self.assertEqual(rebuilt.status, "done", msg=rebuilt.error)
+ self.assertEqual(user_file.read_text(encoding="utf-8"), "不要删除")
+ finally:
+ if previous is None:
+ os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
+ else:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = previous
+
+ def test_unselected_conversation_is_preserved_and_missing_file_is_restored(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account = "wxid_incremental"
+ first_username = "wxid_friend"
+ second_username = "wxid_friend_two"
+ account_dir = self._prepare_account(root, account=account, username=first_username)
+ self._add_conversation(account_dir, username=second_username, display_name="第二位好友")
+ output_dir = root / "exports"
+
+ previous = os.environ.get("WECHAT_TOOL_DATA_DIR")
+ try:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = str(root)
+ service = self._reload_export_modules()
+ first = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ usernames=[first_username, second_username],
+ output_dir=output_dir,
+ )
+ self.assertEqual(first.status, "done", msg=first.error)
+ folder = output_dir / "聊天增量测试"
+ baseline = json.loads((folder / ".wechat-chat-export.json").read_text(encoding="utf-8"))
+ self.assertEqual(len(baseline["conversations"]), 2)
+ with self.assertRaises(service.ChatIncrementalError) as incomplete_reset:
+ self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=first_username,
+ output_dir=output_dir,
+ reset_baseline=True,
+ )
+ self.assertEqual(incomplete_reset.exception.code, "incremental_reset_incomplete")
+
+ second_state = next(
+ state
+ for state in baseline["conversations"].values()
+ if state.get("displayName") == "第二位好友"
+ )
+ second_message = folder / Path(second_state["directory"]) / "messages.json"
+ second_before = (second_message.stat().st_mtime_ns, second_message.read_bytes())
+ user_file = folder / "我的说明.txt"
+ user_file.write_text("不能删除", encoding="utf-8")
+
+ table = self._message_table(first_username)
+ connection = sqlite3.connect(str(account_dir / "message_0.db"))
+ try:
+ connection.execute(
+ f"INSERT INTO {table} VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+ (8, 1008, 1, 8, 2, 1735689800, "只更新第一个会话", None),
+ )
+ connection.commit()
+ finally:
+ connection.close()
+
+ updated = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=first_username,
+ output_dir=output_dir,
+ )
+ self.assertEqual(updated.status, "done", msg=updated.error)
+ self.assertEqual((second_message.stat().st_mtime_ns, second_message.read_bytes()), second_before)
+ self.assertEqual(user_file.read_text(encoding="utf-8"), "不能删除")
+
+ current_baseline = json.loads((folder / ".wechat-chat-export.json").read_text(encoding="utf-8"))
+ first_state = next(
+ state
+ for state in current_baseline["conversations"].values()
+ if state.get("displayName") == "测试好友"
+ )
+ first_message = folder / Path(first_state["directory"]) / "messages.json"
+ first_message.unlink()
+
+ restored = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=second_username,
+ output_dir=folder,
+ )
+ self.assertEqual(restored.status, "done", msg=restored.error)
+ self.assertTrue(first_message.is_file())
+ self.assertEqual((second_message.stat().st_mtime_ns, second_message.read_bytes()), second_before)
+ self.assertEqual(restored.incremental.get("filesRecovered"), 1)
+ self.assertFalse((folder / "聊天增量测试").exists())
+ self.assertTrue(user_file.is_file())
+
+ connection = sqlite3.connect(str(account_dir / "contact.db"))
+ try:
+ connection.execute(
+ "UPDATE contact SET nick_name = ? WHERE username = ?",
+ ("改名后的好友", first_username),
+ )
+ connection.commit()
+ finally:
+ connection.close()
+ renamed = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=first_username,
+ output_dir=output_dir,
+ )
+ self.assertEqual(renamed.status, "done", msg=renamed.error)
+ self.assertFalse(renamed.repair_candidates)
+ renamed_baseline = json.loads((folder / ".wechat-chat-export.json").read_text(encoding="utf-8"))
+ renamed_state = next(
+ state
+ for state in renamed_baseline["conversations"].values()
+ if state.get("displayName") == "改名后的好友"
+ )
+ self.assertEqual(renamed_state.get("directory"), first_state.get("directory"))
+ finally:
+ if previous is None:
+ os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
+ else:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = previous
+
+ def test_missing_chat_and_media_are_restored_for_all_formats(self):
+ format_suffixes = (
+ ("html", "html"),
+ ("json", "json"),
+ ("txt", "txt"),
+ ("excel", "xlsx"),
+ )
+ for export_format, suffix in format_suffixes:
+ with self.subTest(export_format=export_format), TemporaryDirectory() as td:
+ root = Path(td)
+ account = "wxid_incremental"
+ first_username = "wxid_friend"
+ second_username = "wxid_friend_two"
+ account_dir = self._prepare_account(root, account=account, username=first_username)
+ self._add_conversation(account_dir, username=second_username, display_name="第二位好友")
+ output_dir = root / "exports"
+
+ previous = os.environ.get("WECHAT_TOOL_DATA_DIR")
+ try:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = str(root)
+ service = self._reload_export_modules()
+ initial = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ usernames=[first_username, second_username],
+ output_dir=output_dir,
+ export_format=export_format,
+ message_types=["text", "image"],
+ include_media=True,
+ )
+ self.assertEqual(initial.status, "done", msg=initial.error)
+
+ folder = output_dir / "聊天增量测试"
+ baseline = json.loads((folder / ".wechat-chat-export.json").read_text(encoding="utf-8"))
+ first_state = next(
+ state
+ for state in baseline["conversations"].values()
+ if state.get("displayName") == "测试好友"
+ )
+ second_state = next(
+ state
+ for state in baseline["conversations"].values()
+ if state.get("displayName") == "第二位好友"
+ )
+ first_message = folder / Path(first_state["directory"]) / f"messages.{suffix}"
+ second_message = folder / Path(second_state["directory"]) / f"messages.{suffix}"
+ second_before = (second_message.stat().st_mtime_ns, second_message.read_bytes())
+ media_paths = [
+ path for path in baseline["files"]
+ if path.startswith("media/images/")
+ ]
+ self.assertTrue(media_paths)
+
+ first_message.unlink()
+ for media_path in media_paths:
+ (folder / Path(media_path)).unlink()
+
+ restored = self._create_folder_job(
+ service.CHAT_EXPORT_MANAGER,
+ account=account,
+ username=second_username,
+ output_dir=output_dir,
+ export_format=export_format,
+ message_types=["text", "image"],
+ include_media=True,
+ )
+ self.assertEqual(restored.status, "done", msg=restored.error)
+ self.assertTrue(first_message.is_file())
+ for media_path in media_paths:
+ self.assertTrue((folder / Path(media_path)).is_file())
+ self.assertEqual(
+ (second_message.stat().st_mtime_ns, second_message.read_bytes()),
+ second_before,
+ )
+ self.assertEqual(
+ restored.incremental.get("filesRecovered"),
+ 1 + len(media_paths),
+ )
+ self.assertEqual(
+ restored.incremental.get("conversationsUpdated"),
+ 1,
+ msg=export_format,
+ )
+ self.assertEqual(restored.incremental.get("conversationsReused"), 1)
+ finally:
+ if previous is None:
+ os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
+ else:
+ os.environ["WECHAT_TOOL_DATA_DIR"] = previous
+
+ def test_shared_media_is_removed_only_after_last_owner_is_rebuilt(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account = "wxid_incremental"
+ first_username = "wxid_friend"
+ second_username = "wxid_friend_two"
+ account_dir = self._prepare_account(root, account=account, username=first_username)
+ self._add_conversation(account_dir, username=second_username, display_name="第二位好友")
+ table_two = self._message_table(second_username)
+ image_xml = '(.*?)',
+ flags=re.DOTALL,
+ )
+ media_match = media_pattern.search(text, boundary)
+ media_index: dict[str, Any] = {}
+ if media_match is not None:
+ try:
+ value = json.loads(media_match.group(1))
+ if isinstance(value, dict):
+ media_index = value
+ except Exception:
+ media_index = {}
+
+ return {
+ "text": text,
+ "contentStart": content_start,
+ "contentEnd": content_end,
+ "closeMarker": close_marker,
+ "fragment": text[content_start:content_end],
+ "pageMeta": page_meta,
+ "pageSpan": page_span,
+ "mediaIndex": media_index,
+ }
+
+
+def _html_incremental_snapshot(path: Path) -> Optional[dict[str, Any]]:
+ try:
+ return _html_incremental_snapshot_text(path.read_text(encoding="utf-8"))
+ except Exception:
+ return None
+
+
+def _merge_html_media_index(old_value: Any, new_value: Any) -> Any:
+ if not isinstance(old_value, dict) or not isinstance(new_value, dict):
+ return copy.deepcopy(new_value if new_value not in ({}, None) else old_value)
+ merged = copy.deepcopy(old_value)
+ for key, value in new_value.items():
+ if key in merged and isinstance(merged[key], dict) and isinstance(value, dict):
+ merged[key] = _merge_html_media_index(merged[key], value)
+ else:
+ merged[key] = copy.deepcopy(value)
+ return merged
+
+
def _write_conversation_html(
*,
zf: zipfile.ZipFile,
@@ -4675,12 +6040,15 @@ def _write_conversation_html(
html_assets = dict(getattr(job, "options", {}).get("_htmlAssets") or {})
css_asset_path = str(html_assets.get("cssPath") or _html_export_asset_paths(job.export_id)[0])
js_asset_path = str(html_assets.get("jsPath") or _html_export_asset_paths(job.export_id)[1])
+ session_catalog_path = str(html_assets.get("sessionCatalogPath") or "assets/chat-sessions.js")
integrity_asset_path = str(html_assets.get("integrityPath") or _html_export_asset_paths(job.export_id)[2])
css_integrity = str(html_assets.get("cssIntegrity") or "")
js_integrity = str(html_assets.get("jsIntegrity") or "")
+ folder_mode = bool(html_assets.get("folderMode"))
css_href = rel_root + css_asset_path
integrity_src = rel_root + integrity_asset_path
js_src = rel_root + js_asset_path
+ session_catalog_src = rel_root + session_catalog_path
def esc_text(v: Any) -> str:
return html.escape(str(v or ""), quote=False)
@@ -5114,17 +6482,27 @@ def flush(self) -> None:
tw.write(' \n')
tw.write(' \n')
tw.write(f" {esc_text(page_title)} \n")
- tw.write(_html_export_gate_style())
- # Do not use native `integrity=` for offline file:// exports; Chrome blocks
- # those resources before our runtime can show the page.
- tw.write(f' \n')
- tw.write(_html_export_integrity_script_tag(src=integrity_src))
- tw.write(f' \n')
+ if folder_mode:
+ tw.write(f' \n')
+ tw.write(
+ f' \n'
+ )
+ tw.write(f' \n')
+ else:
+ tw.write(_html_export_gate_style())
+ # file:// 下由导出运行时核对 data-wce-sri，不能使用浏览器原生 SRI。
+ tw.write(f' \n')
+ tw.write(_html_export_integrity_script_tag(src=integrity_src))
+ tw.write(f' \n')
tw.write("\n")
tw.write("\n")
tw.write(
'
\n"
+ if folder_mode
+ else "提示：此页面需要 JavaScript 才能使用“合并聊天记录”等交互功能。若该提示一直存在，请确认已完整解压导出目录，并检查 assets/_wce/ 下的运行时文件是否完整。'
- "提示：此页面需要 JavaScript 才能使用“合并聊天记录”等交互功能。若该提示一直存在，请确认已完整解压导出目录，并检查 assets/_wce/ 下的运行时文件是否完整。
\n"
+ + (
+ "提示：此页面需要 JavaScript 才能使用“合并聊天记录”等交互功能。若该提示一直存在，请确认导出目录中的运行时文件完整。