{title}
{body}
"""
def _html_escape(s: str) -> str:
return s.replace("&","&").replace("<","<").replace(">",">").replace('"','"')
def _write_html(path: str, title: str, is_group: bool, messages: list, image_map: dict = None, out_dir: str = None):
parts = []
last_date = None
for m in messages:
dt = datetime.fromtimestamp(m["create_time"])
day = dt.strftime("%Y年%m月%d日")
if day != last_date:
parts.append(f'{day}
')
last_date = day
if m["is_system"]:
parts.append(
f''
f'{_html_escape(m["display_content"])}
{_html_escape(m["sender"])}
'
type_tag = ""
if m["type"] != 1:
type_tag = f'{m["type_name"]}
'
# 图片消息嵌入
bubble_content = _html_escape(m["display_content"])
if m["type"] == 3 and image_map and m["local_id"] in image_map:
rel_path = image_map[m["local_id"]]
if out_dir:
abs_img = os.path.join(out_dir, rel_path)
if os.path.exists(abs_img):
ext = os.path.splitext(abs_img)[1].lstrip('.').lower()
mime = {'jpg': 'image/jpeg', 'jpeg': 'image/jpeg', 'png': 'image/png',
'gif': 'image/gif', 'webp': 'image/webp'}.get(ext, 'image/jpeg')
try:
with open(abs_img, 'rb') as imgf:
b64 = base64.b64encode(imgf.read()).decode('ascii')
bubble_content = f''
f'
'
)
body = "\n".join(parts)
with open(path, "w", encoding="utf-8") as f:
f.write(HTML_TEMPLATE.format(title=_html_escape(title), body=body))
# ─── 加载联系人信息 ─────────────────────────────────────────────────────────────
contact_map: dict[str, dict] = {}
try:
cconn = sqlite3.connect(CONTACT_DB_PATH)
for uname, alias, remark, nick_name in cconn.execute(
"SELECT username, alias, remark, nick_name FROM contact"
):
contact_map[uname] = {
"username": uname,
"alias": alias or "",
"remark": remark or "",
"nick_name": nick_name or "",
}
cconn.close()
print(f"联系人数据库: {len(contact_map)} 条")
except Exception as e:
print(f"联系人数据库读取失败: {e}")
def display_name(username: str) -> str:
info = contact_map.get(username, {})
return info.get("remark") or info.get("nick_name") or username
# ─── 遍历所有 message_*.db ──────────────────────────────────────────────────────
db_files = sorted(
f for f in glob.glob(os.path.join(MSG_DB_DIR, "message_*.db"))
if not f.endswith(("_fts.db", "_resource.db"))
)
print(f"找到 {len(db_files)} 个消息数据库")
total_chats = 0
total_msgs = 0
# ── 阶段1: 收集所有联系人的消息 ──────────────────────────────────────────────
# chat_data[chat_username] -> { dname, is_group, db_messages: [(db_name, messages)] }
chat_data: dict[str, dict] = {}
for db_path in sorted(db_files):
db_name = os.path.basename(db_path)
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
# rowid -> username
sender_map: dict[int, str] = {}
for row in conn.execute("SELECT rowid, user_name FROM Name2Id"):
sender_map[row[0]] = row[1]
# 计算 username -> hash 映射
hash_to_username: dict[str, str] = {}
for username in sender_map.values():
if username:
h = hashlib.md5(username.encode()).hexdigest()
hash_to_username[h] = username
# 找出所有 Msg_{initial}
'
f''
f'{sender_label}'
f'
{bubble_content}
'
f'{type_tag}'
f'{m["time_str"]}
'
f'