.
Works via depth tracking: when we enter a top-level message div, start a
new collection buffer; when it closes, emit.
"""
def __init__(self, html_file: str):
super().__init__(convert_charrefs=True)
self.html_file = html_file
self.messages: list[TGMessage] = []
# outer stack of (tag, class, attrs)
self._stack: list[tuple[str, str, dict]] = []
# index into self._stack where the current message div starts, or None
self._msg_depth: int | None = None
# current message being built
self._cur: dict | None = None
# which sub-section we're currently capturing text into
# one of: None | "from_name" | "text" | "reply_to" | "reaction_emoji"
# | "forwarded_from" | "service_body"
self._capture: str | None = None
self._capture_buf: list[str] = []
# reaction accumulator: last seen emoji waiting for count (user-pics length is count)
self._pending_reactions: list[tuple[str, int]] = []
self._reaction_userpic_depth: int | None = None
self._current_reaction_emoji: str | None = None
self._current_reaction_count: int = 0
# track seen sender to fill in "joined" continuations
self._last_sender: str | None = None
# -- helpers -----------------------------------------------------------
def _classes(self, attrs: dict) -> set[str]:
return set((attrs.get("class") or "").split())
def _in_message(self) -> bool:
return self._msg_depth is not None
def _flush_text(self) -> str:
s = "".join(self._capture_buf)
self._capture_buf.clear()
return s
# -- HTMLParser hooks --------------------------------------------------
def handle_starttag(self, tag, attrs):
attr_d = dict(attrs)
self._stack.append((tag, attr_d.get("class", ""), attr_d))
classes = self._classes(attr_d)
if tag == "div" and "message" in classes and attr_d.get("id", "").startswith("message"):
self._msg_depth = len(self._stack) - 1
mid = attr_d["id"][len("message"):]
self._cur = {
"message_id": mid,
"html_file": self.html_file,
"sender": None,
"timestamp": None,
"text_parts": [],
"is_service": "service" in classes,
"is_joined": "joined" in classes,
"reply_to_id": None,
"mentions": [],
"attachments": [],
"reactions": [],
"forwarded_from": None,
}
return
if not self._in_message() or self._cur is None:
return
# inside a message
if tag == "div":
if "from_name" in classes:
self._capture = "from_name"
self._capture_buf.clear()
elif "pull_right" in classes and "date" in classes and "details" in classes:
ts = _parse_title_ts(attr_d.get("title", ""))
if ts:
self._cur["timestamp"] = ts
elif "reply_to" in classes and "details" in classes:
self._capture = "reply_to"
self._capture_buf.clear()
elif "text" in classes and self._capture != "from_name":
self._capture = "text"
self._capture_buf.clear()
elif "forwarded" in classes and "body" in classes:
# forwarded block; the inner .from_name capture will fire
self._cur["forwarded_from"] = "" # sentinel: we saw a forward
elif "body" in classes and "details" in classes and self._cur["is_service"]:
# service message body
self._capture = "service_body"
self._capture_buf.clear()
elif tag == "span":
if "emoji" in classes and self._capture != "text":
# reaction emoji
self._capture = "reaction_emoji"
self._capture_buf.clear()
elif "userpics" in classes:
self._reaction_userpic_depth = len(self._stack) - 1
self._current_reaction_count = 0
elif tag == "a":
if self._capture == "text":
# mention link
onclick = attr_d.get("onclick", "") or ""
if "ShowMentionName" in onclick:
# keep accumulating; the link text lands in capture buffer
pass
if self._capture == "reply_to":
m = _REPLY_ID_RE.search(attr_d.get("onclick", "") or "")
if m:
self._cur["reply_to_id"] = m.group(1)
href = attr_d.get("href", "") or ""
if href.startswith(("photos/", "files/", "voice_messages/", "video_files/", "round_video_messages/", "stickers/")):
self._cur["attachments"].append(href)
elif tag == "div" and self._reaction_userpic_depth is not None:
# count userpic divs inside a reactions .userpics (one per reactor)
pass
if tag == "div" and self._reaction_userpic_depth is not None:
classes = self._classes(attr_d)
if "userpic" in classes:
self._current_reaction_count += 1
if tag == "br" and self._capture == "text":
self._capture_buf.append("\n")
def handle_endtag(self, tag):
if not self._stack:
return
# pop current tag frame
self._stack.pop()
depth = len(self._stack)
# close of the message div?
if self._msg_depth is not None and depth == self._msg_depth and self._cur is not None:
# finalize
sender = self._cur["sender"]
if self._cur["is_joined"] and not sender:
sender = self._last_sender
if sender:
self._last_sender = sender
text = "".join(self._cur["text_parts"]).strip()
ts = self._cur["timestamp"]
self.messages.append(TGMessage(
message_id=self._cur["message_id"],
html_file=self._cur["html_file"],
sender=sender,
timestamp=ts,
text=text,
is_service=self._cur["is_service"],
is_joined=self._cur["is_joined"],
reply_to_id=self._cur["reply_to_id"],
mentions=self._cur["mentions"],
attachments=self._cur["attachments"],
reactions=self._cur["reactions"],
forwarded_from=self._cur["forwarded_from"] or None,
))
self._cur = None
self._msg_depth = None
self._capture = None
self._capture_buf.clear()
return
if self._reaction_userpic_depth is not None and depth == self._reaction_userpic_depth:
# closed a userpics span — commit reaction
if self._cur and self._current_reaction_emoji:
self._cur["reactions"].append(
(self._current_reaction_emoji, max(1, self._current_reaction_count))
)
self._current_reaction_emoji = None
self._current_reaction_count = 0
self._reaction_userpic_depth = None
return
# close of a capturing section?
if self._capture == "from_name" and tag == "div":
text = self._flush_text().strip()
if self._cur and self._cur["forwarded_from"] == "":
# forwarded header fills forwarded_from, not sender
self._cur["forwarded_from"] = text
elif self._cur:
self._cur["sender"] = text
self._capture = None
elif self._capture == "text" and tag == "div":
# finalize text
raw = self._flush_text()
if self._cur:
self._cur["text_parts"].append(raw)
self._capture = None
elif self._capture == "reply_to" and tag == "div":
self._capture = None
self._capture_buf.clear()
elif self._capture == "reaction_emoji" and tag == "span":
self._current_reaction_emoji = self._flush_text().strip()
self._capture = None
elif self._capture == "service_body" and tag == "div":
svc = self._flush_text().strip()
if self._cur:
self._cur["text_parts"].append(svc)
self._capture = None
def handle_data(self, data):
if self._capture:
self._capture_buf.append(data)
def parse_html_file(path: Path) -> list[TGMessage]:
with open(path, encoding="utf-8") as f:
raw = f.read()
parser = _TelegramExportParser(html_file=path.name)
parser.feed(raw)
parser.close()
return parser.messages
# ---------------------------------------------------------------------------
# Export directory handling
# ---------------------------------------------------------------------------
def _sorted_message_htmls(root: Path) -> list[Path]:
"""messages.html, messages2.html, messages3.html … in the order TG exports them."""
files = list(root.glob("messages*.html"))
def key(p: Path) -> int:
stem = p.stem # "messages" or "messages3"
num = stem[len("messages"):]
return int(num) if num.isdigit() else 1
return sorted(files, key=key)
def _resolve_export_dir(source: Path) -> tuple[Path, tempfile.TemporaryDirectory | None]:
"""Return a directory containing messages*.html, extracting the zip if needed."""
if source.is_dir():
return source, None
if not source.exists() or source.suffix.lower() != ".zip":
raise SystemExit(f"not a zip or directory: {source}")
tmp = tempfile.TemporaryDirectory(prefix="tg_export_")
with zipfile.ZipFile(source) as z:
z.extractall(tmp.name)
# look for messages.html at any depth
root = Path(tmp.name)
candidates = list(root.rglob("messages.html"))
if not candidates:
raise SystemExit(f"no messages.html inside {source}")
export_root = candidates[0].parent
return export_root, tmp
# ---------------------------------------------------------------------------
# Memory-row builder
# ---------------------------------------------------------------------------
def _deterministic_uuid(*parts: str) -> str:
h = hashlib.sha256("||".join(parts).encode()).hexdigest()
# Craft a UUIDv4-shaped string from the hash so qdrant accepts it.
return str(uuid.UUID(h[:32]))
def _importance(msg: TGMessage) -> float:
if msg.is_service:
return 0.15
if not msg.text:
return 0.25 if msg.attachments else 0.1
n = len(msg.text)
if n < 4:
return 0.25
if n < 20:
return 0.35
if n < 200:
return 0.5
return 0.6
def _content(msg: TGMessage, chat_title: str) -> str:
"""Human+embedding friendly line: 'Sender: text' plus hints."""
sender = msg.sender or "(unknown)"
body = msg.text or ""
if msg.is_service:
return f"[service] {body}"
if msg.attachments and not body:
body = f"(attachment: {', '.join(msg.attachments[:3])})"
prefix = f"{sender}"
if msg.forwarded_from:
prefix += f" (forwarded from {msg.forwarded_from})"
return f"{prefix}: {body}".strip()
def _entities(msg: TGMessage) -> list[str]:
ents: list[str] = []
if msg.sender:
ents.append(msg.sender)
ents.extend(msg.mentions)
# pull @handles and #tags from the body
if msg.text:
ents.extend(re.findall(r"@[A-Za-z0-9_]+", msg.text))
ents.extend(re.findall(r"#\w+", msg.text))
# dedupe, keep order
seen = set()
out = []
for e in ents:
k = e.lower()
if k in seen:
continue
seen.add(k)
out.append(e)
return out
def messages_to_rows(
messages: list[TGMessage],
*,
agent_id: str,
chat_title: str,
chat_id: str,
chat_slug: str,
skip_service: bool = False,
) -> Iterator[dict]:
for msg in messages:
if skip_service and msg.is_service:
continue
ts = msg.timestamp or datetime(2026, 3, 15, tzinfo=timezone.utc)
row_id = _deterministic_uuid(chat_id, msg.html_file, msg.message_id)
yield {
"id": row_id,
"content": _content(msg, chat_title),
"memory_type": "episodic",
"agent_id": agent_id,
"importance": _importance(msg),
"initial_importance": _importance(msg),
"timestamp": ts.astimezone(timezone.utc).isoformat(),
"entities": _entities(msg),
"session_id": f"tg:{chat_slug}",
"metadata": {
"chat_title": chat_title,
"chat_id": chat_id,
"chat_slug": chat_slug,
"tg_message_id": msg.message_id,
"html_file": msg.html_file,
"sender": msg.sender,
"reply_to_tg_id": msg.reply_to_id,
"mentions": msg.mentions,
"attachments": msg.attachments,
"reactions": [[e, c] for e, c in msg.reactions],
"forwarded_from": msg.forwarded_from,
"is_service": msg.is_service,
"is_joined": msg.is_joined,
},
"consolidated": False,
"consolidated_into": [],
"consolidation_batch_id": None,
"retrieval_count": 0,
"utilization_count": 0,
"outcome_count": 0,
"last_retrieved": None,
"last_utilized": None,
"last_boosted": None,
"importance_history": [_importance(msg)],
"boost_cooldown_until": None,
"has_colbert": False,
"colbert_token_count": 0,
"_source": "telegram_export",
"_source_file": msg.html_file,
}
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("source", help="path to .zip export or extracted directory")
p.add_argument("output", help="output .jsonl path")
p.add_argument("--agent-id", default="m2")
p.add_argument("--chat-id", default="-1003815414577",
help="Telegram numeric chat id (recorded in metadata)")
p.add_argument("--chat-slug", default="parlobyg")
p.add_argument("--chat-title", default=None,
help="override auto-detected chat title")
p.add_argument("--skip-service", action="store_true",
help="drop service messages (joins/renames/etc.)")
args = p.parse_args(argv)
source = Path(args.source).expanduser().resolve()
export_dir, tmp_holder = _resolve_export_dir(source)
html_files = _sorted_message_htmls(export_dir)
if not html_files:
print(f"no messages*.html in {export_dir}", file=sys.stderr)
return 1
chat_title = args.chat_title
if not chat_title:
# extract from first messages.html
…
try:
head = html_files[0].read_text(encoding="utf-8")
m = re.search(r'class="text bold"\s*>\s*([^<]+?)\s*<', head)
if m:
chat_title = html.unescape(m.group(1)).strip()
except Exception:
pass
chat_title = chat_title or args.chat_slug
total = 0
per_file: dict[str, int] = {}
out_path = Path(args.output).expanduser().resolve()
out_path.parent.mkdir(parents=True, exist_ok=True)
with open(out_path, "w", encoding="utf-8") as out:
for htmlf in html_files:
msgs = parse_html_file(htmlf)
rows = list(messages_to_rows(
msgs,
agent_id=args.agent_id,
chat_title=chat_title,
chat_id=args.chat_id,
chat_slug=args.chat_slug,
skip_service=args.skip_service,
))
for r in rows:
out.write(json.dumps(r, ensure_ascii=False) + "\n")
total += len(rows)
per_file[htmlf.name] = len(rows)
print(f" {htmlf.name}: {len(rows)} rows", file=sys.stderr)
print(f"wrote {total} rows to {out_path}", file=sys.stderr)
print(f" chat_title={chat_title!r} chat_id={args.chat_id} agent_id={args.agent_id}",
file=sys.stderr)
if tmp_holder is not None:
tmp_holder.cleanup()
return 0
if __name__ == "__main__":
sys.exit(main())