mirror of
https://github.com/LifeArchiveProject/WeChatDataAnalysis.git
synced 2026-02-19 14:20:51 +08:00
bd44601611
- 新增 /api/chat/messages/daily_counts 与 /api/chat/messages/anchor，用于月度热力图与按日/首条定位\n- messages/around 支持跨 message 分片定位，定位更稳定\n- 新增 /api/chat/chat_history/resolve 与 /api/chat/appmsg/resolve，合并转发/链接卡片可按 server_id 补全\n- 新增 /api/chat/media/favicon，并补齐 link 本地缩略图处理\n- HTML 导出支持分页加载（html_page_size），避免大聊天单文件卡顿\n- tests: 覆盖 heatmap/anchor、favicon 缓存、HTML 分页导出
222 lines
8.3 KiB
Python
222 lines
8.3 KiB
Python
import os
|
|
import json
|
|
import hashlib
|
|
import sqlite3
|
|
import sys
|
|
import unittest
|
|
import zipfile
|
|
import importlib
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
|
|
|
|
class TestChatExportHtmlPaging(unittest.TestCase):
|
|
def _reload_export_modules(self):
|
|
import wechat_decrypt_tool.app_paths as app_paths
|
|
import wechat_decrypt_tool.chat_helpers as chat_helpers
|
|
import wechat_decrypt_tool.media_helpers as media_helpers
|
|
import wechat_decrypt_tool.chat_export_service as chat_export_service
|
|
|
|
importlib.reload(app_paths)
|
|
importlib.reload(chat_helpers)
|
|
importlib.reload(media_helpers)
|
|
importlib.reload(chat_export_service)
|
|
return chat_export_service
|
|
|
|
def _seed_contact_db(self, path: Path, *, account: str, username: str) -> None:
|
|
conn = sqlite3.connect(str(path))
|
|
try:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE contact (
|
|
username TEXT,
|
|
remark TEXT,
|
|
nick_name TEXT,
|
|
alias TEXT,
|
|
local_type INTEGER,
|
|
verify_flag INTEGER,
|
|
big_head_url TEXT,
|
|
small_head_url TEXT
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE stranger (
|
|
username TEXT,
|
|
remark TEXT,
|
|
nick_name TEXT,
|
|
alias TEXT,
|
|
local_type INTEGER,
|
|
verify_flag INTEGER,
|
|
big_head_url TEXT,
|
|
small_head_url TEXT
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO contact VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(account, "", "Me", "", 1, 0, "", ""),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO contact VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(username, "", "Friend", "", 1, 0, "", ""),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
def _seed_session_db(self, path: Path, *, username: str) -> None:
|
|
conn = sqlite3.connect(str(path))
|
|
try:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE SessionTable (
|
|
username TEXT,
|
|
is_hidden INTEGER,
|
|
sort_timestamp INTEGER
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO SessionTable VALUES (?, ?, ?)",
|
|
(username, 0, 1735689600),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
def _seed_message_db(self, path: Path, *, account: str, username: str, total: int) -> None:
|
|
conn = sqlite3.connect(str(path))
|
|
try:
|
|
conn.execute("CREATE TABLE Name2Id (rowid INTEGER PRIMARY KEY, user_name TEXT)")
|
|
conn.execute("INSERT INTO Name2Id(rowid, user_name) VALUES (?, ?)", (1, account))
|
|
conn.execute("INSERT INTO Name2Id(rowid, user_name) VALUES (?, ?)", (2, username))
|
|
|
|
table_name = f"msg_{hashlib.md5(username.encode('utf-8')).hexdigest()}"
|
|
conn.execute(
|
|
f"""
|
|
CREATE TABLE {table_name} (
|
|
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
|
|
)
|
|
"""
|
|
)
|
|
|
|
# Generate lots of plain text messages with unique markers.
|
|
rows = []
|
|
base_ts = 1735689600
|
|
for i in range(1, total + 1):
|
|
marker = f"MSG{i:04d}"
|
|
real_sender_id = 1 if (i % 2 == 0) else 2
|
|
rows.append((i, 100000 + i, 1, i, real_sender_id, base_ts + i, marker, None))
|
|
|
|
conn.executemany(
|
|
f"INSERT INTO {table_name} (local_id, server_id, local_type, sort_seq, real_sender_id, create_time, message_content, compress_content) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
|
rows,
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
def _prepare_account(self, root: Path, *, account: str, username: str, total: int) -> Path:
|
|
account_dir = root / "output" / "databases" / account
|
|
account_dir.mkdir(parents=True, exist_ok=True)
|
|
self._seed_contact_db(account_dir / "contact.db", account=account, username=username)
|
|
self._seed_session_db(account_dir / "session.db", username=username)
|
|
self._seed_message_db(account_dir / "message_0.db", account=account, username=username, total=total)
|
|
return account_dir
|
|
|
|
def _create_job(self, manager, *, account: str, username: str, html_page_size: int):
|
|
job = manager.create_job(
|
|
account=account,
|
|
scope="selected",
|
|
usernames=[username],
|
|
export_format="html",
|
|
start_time=None,
|
|
end_time=None,
|
|
include_hidden=False,
|
|
include_official=False,
|
|
include_media=False,
|
|
media_kinds=[],
|
|
message_types=[],
|
|
output_dir=None,
|
|
allow_process_key_extract=False,
|
|
download_remote_media=False,
|
|
html_page_size=html_page_size,
|
|
privacy_mode=False,
|
|
file_name=None,
|
|
)
|
|
|
|
# Export is async (thread). Allow enough time for a few thousand messages + zip writes.
|
|
for _ in range(600):
|
|
latest = manager.get_job(job.export_id)
|
|
if latest and latest.status in {"done", "error", "cancelled"}:
|
|
return latest
|
|
import time as _time
|
|
|
|
_time.sleep(0.05)
|
|
self.fail("export job did not finish in time")
|
|
|
|
def test_html_export_paging_inlines_latest_page_only(self):
|
|
with TemporaryDirectory() as td:
|
|
root = Path(td)
|
|
account = "wxid_test"
|
|
username = "wxid_friend"
|
|
|
|
total_messages = 2300
|
|
page_size = 1000
|
|
self._prepare_account(root, account=account, username=username, total=total_messages)
|
|
|
|
prev_data = os.environ.get("WECHAT_TOOL_DATA_DIR")
|
|
try:
|
|
os.environ["WECHAT_TOOL_DATA_DIR"] = str(root)
|
|
svc = self._reload_export_modules()
|
|
job = self._create_job(
|
|
svc.CHAT_EXPORT_MANAGER,
|
|
account=account,
|
|
username=username,
|
|
html_page_size=page_size,
|
|
)
|
|
self.assertEqual(job.status, "done", msg=job.error)
|
|
|
|
self.assertTrue(job.zip_path and job.zip_path.exists())
|
|
with zipfile.ZipFile(job.zip_path, "r") as zf:
|
|
names = set(zf.namelist())
|
|
|
|
html_path = next((n for n in names if n.endswith("/messages.html")), "")
|
|
self.assertTrue(html_path, msg="missing messages.html")
|
|
html_text = zf.read(html_path).decode("utf-8", errors="ignore")
|
|
|
|
# Paging UI + meta should exist for multi-page exports.
|
|
self.assertIn('id="wcePageMeta"', html_text)
|
|
self.assertIn('id="wcePager"', html_text)
|
|
self.assertIn('id="wceMessageList"', html_text)
|
|
self.assertIn('id="wceLoadPrevBtn"', html_text)
|
|
|
|
# Latest page is inlined; earliest page should not be present in messages.html.
|
|
self.assertIn("MSG2300", html_text)
|
|
self.assertNotIn("MSG0001", html_text)
|
|
|
|
conv_dir = html_path.rsplit("/", 1)[0]
|
|
page1_js = f"{conv_dir}/pages/page-0001.js"
|
|
self.assertIn(page1_js, names)
|
|
page1_text = zf.read(page1_js).decode("utf-8", errors="ignore")
|
|
self.assertIn("MSG0001", page1_text)
|
|
finally:
|
|
if prev_data is None:
|
|
os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
|
|
else:
|
|
os.environ["WECHAT_TOOL_DATA_DIR"] = prev_data
|