"""Telegram HTML 내보내기 파일을 conversation_memory.py 호환 JSONL로 변환.
사용법:
python html_to_jsonl.py [INPUT_HTML] [OUTPUT_JSONL]
기본값:
INPUT_HTML = /home/jay/.cokacdir/workspace/autoset/messages.html
OUTPUT_JSONL = /home/jay/workspace/memory/groupchat/test-import-2026-03-15.jsonl
"""
from __future__ import annotations
import json
import sys
from collections import Counter
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Optional
from bs4 import BeautifulSoup, Tag
# ---------------------------------------------------------------------------
# 상수
# ---------------------------------------------------------------------------
DEFAULT_INPUT = "/home/jay/.cokacdir/workspace/autoset/messages.html"
DEFAULT_OUTPUT = "/home/jay/workspace/memory/groupchat/test-import-2026-03-15.jsonl"
CHAT_ID: int = 99999
TOPIC_TAG: str = "general"
# 발신자 매핑: HTML 표시명 → (정규화 이름, is_bot)
SENDER_MAP: dict[str, tuple[str, bool]] = {
"JJongs": ("제이회장님", False),
"잼민이": ("잼민이", True),
"코덱스": ("코덱스", True),
"클로디": ("클로디", True),
"아누(ANU) AI비서": ("아누", True),
}
# UTC+09:00 타임존
TZ_KST = timezone(timedelta(hours=9))
# ---------------------------------------------------------------------------
# 파싱 헬퍼
# ---------------------------------------------------------------------------
def parse_timestamp(title: str) -> Optional[datetime]:
"""'15.03.2026 14:52:14 UTC+09:00' 형식을 datetime으로 변환.
파싱 실패 시 None 반환.
"""
# "15.03.2026 14:52:14 UTC+09:00" → "15.03.2026 14:52:14 +09:00"
normalized = title.replace("UTC", "").strip()
# 날짜 부분: DD.MM.YYYY
try:
dt = datetime.strptime(normalized, "%d.%m.%Y %H:%M:%S %z")
return dt
except ValueError:
return None
def resolve_sender(raw_name: str) -> tuple[str, bool]:
"""HTML 발신자 이름을 (정규화 이름, is_bot) 으로 변환.
매핑에 없는 이름은 원본 그대로 + is_bot=False 로 처리.
"""
stripped = raw_name.strip()
return SENDER_MAP.get(stripped, (stripped, False))
def extract_text(text_div: Tag) -> str:
"""text div 내 HTML 태그를 제거하고 순수 텍스트만 반환.
a 태그는 href 대신 텍스트만 남긴다.
태그는 개행으로 처리.
"""
#
태그를 개행문자로 교체
for br in text_div.find_all("br"):
br.replace_with("\n")
return text_div.get_text(separator="")
# ---------------------------------------------------------------------------
# 핵심 파싱 함수
# ---------------------------------------------------------------------------
def parse_html_to_records(html_path: Path) -> list[dict]:
"""HTML 파일을 파싱하여 JSONL 레코드 리스트를 반환."""
with open(html_path, "r", encoding="utf-8") as fh:
soup = BeautifulSoup(fh, "html.parser")
records: list[dict] = []
last_sender_raw: str = ""
# .history 아래의 모든 .message div 탐색
history_div = soup.find("div", class_="history")
if history_div is None:
print("[WARN] .history div를 찾을 수 없습니다.", file=sys.stderr)
return records
for msg_div in history_div.find_all("div", class_="message"):
classes = set(msg_div.get("class") or [])
# service 메시지 스킵
if "service" in classes:
continue
# default 메시지만 처리
if "default" not in classes:
continue
body_div = msg_div.find("div", class_="body")
if body_div is None:
continue
# --- 발신자 추출 ---
if "joined" in classes:
# 연속 메시지: 이전 발신자 상속
sender_raw = last_sender_raw
else:
from_name_div = body_div.find("div", class_="from_name")
if from_name_div is None:
# from_name 없는 default 메시지는 이전 발신자 상속
sender_raw = last_sender_raw
else:
sender_raw = from_name_div.get_text(strip=True)
last_sender_raw = sender_raw
if not sender_raw:
# 발신자를 끝내 알 수 없으면 스킵
continue
# --- 타임스탬프 추출 ---
date_div = body_div.find("div", class_="date")
if date_div is None:
continue
title_attr = date_div.get("title") or ""
ts = parse_timestamp(str(title_attr))
if ts is None:
print(f"[WARN] 타임스탬프 파싱 실패: {title_attr!r}", file=sys.stderr)
continue
# --- 텍스트 추출 ---
text_div = body_div.find("div", class_="text")
if text_div is None:
# 텍스트 없는 메시지(스티커, 미디어 등) 스킵
continue
text = extract_text(text_div).strip()
if not text:
continue
# --- 발신자 정규화 ---
sender_name, is_bot = resolve_sender(sender_raw)
record = {
"sender": sender_name,
"text": text,
"timestamp": ts.isoformat(),
"is_bot": is_bot,
"chat_id": CHAT_ID,
"topic_tag": TOPIC_TAG,
}
records.append(record)
return records
# ---------------------------------------------------------------------------
# 통계 출력
# ---------------------------------------------------------------------------
def print_stats(records: list[dict]) -> None:
"""변환 결과 통계를 출력한다."""
total = len(records)
bot_count = sum(1 for r in records if r["is_bot"])
human_count = total - bot_count
sender_counter: Counter = Counter(r["sender"] for r in records)
print(f"\n{'=' * 50}")
print("변환 통계")
print(f"{'=' * 50}")
print(f"총 메시지 수 : {total:,}")
print(f"봇 메시지 : {bot_count:,} ({bot_count / total * 100:.1f}%)" if total else "봇 메시지 : 0")
print(f"사람 메시지 : {human_count:,} ({human_count / total * 100:.1f}%)" if total else "사람 메시지 : 0")
print()
print("발신자별 메시지 수:")
for sender, count in sender_counter.most_common():
is_bot = any(r["is_bot"] for r in records if r["sender"] == sender)
label = "[봇]" if is_bot else "[사람]"
print(f" {label} {sender}: {count:,}")
print(f"{'=' * 50}\n")
# ---------------------------------------------------------------------------
# 파일 출력
# ---------------------------------------------------------------------------
def write_jsonl(records: list[dict], output_path: Path) -> None:
"""레코드 리스트를 JSONL 파일로 저장."""
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, "w", encoding="utf-8") as fh:
for record in records:
fh.write(json.dumps(record, ensure_ascii=False) + "\n")
print(f"저장 완료: {output_path} ({len(records):,}줄)")
# ---------------------------------------------------------------------------
# 진입점
# ---------------------------------------------------------------------------
if __name__ == "__main__":
input_path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(DEFAULT_INPUT)
output_path = Path(sys.argv[2]) if len(sys.argv) > 2 else Path(DEFAULT_OUTPUT)
if not input_path.exists():
print(f"[ERROR] 입력 파일을 찾을 수 없습니다: {input_path}", file=sys.stderr)
sys.exit(1)
print(f"파싱 중: {input_path}")
records = parse_html_to_records(input_path)
if not records:
print("[WARN] 변환된 메시지가 없습니다.")
sys.exit(0)
print_stats(records)
write_jsonl(records, output_path)