dcd1241a94
Ingestion scripts: - scripts/ingest/telegram_export.py (stdlib only, from spark3) - scripts/ingest/push_to_memory.py (JSONL → memory API, importance scoring) - scripts/ingest/memory_api_fix.py (memory.py ownership-aware close fix) memory-api bug fixed (deployed to running container): - memory.py: AgentMemory.close() was closing shared _qdrant client - Fix: track ownership (_owns_qdrant/embeddings/redis) — only close what we created - main.py: qdrant_client= → qdrant= parameter name fix agents/nasr/MEMORY.md: - Added GMI clinic context (machine.machine.zip ingested 2026-04-22) - Added Hermes agent note (Mariusz to give details) - Added active projects and skills inventory telegram/: - machine.machine.zip (572KB — MM group chat, GMI context) - manifest.json (checksums for all zips) - .gitignore (large zips excluded: m2.zip, parlobyg.zip, MuhlAI.zip)
531 lines
19 KiB
Python
531 lines
19 KiB
Python
"""
|
|
Parse a Telegram Desktop HTML export (e.g. parlobyg.zip) into a JSONL file
|
|
whose rows match the agent.memory.system Memory schema.
|
|
|
|
Stdlib only — no bs4/lxml dependency. The export structure is regular enough
|
|
that a depth-tracking HTMLParser handles it.
|
|
|
|
Usage:
|
|
python -m ingest.telegram_export <zip-or-dir> <output.jsonl> \
|
|
[--agent-id m2] [--chat-id -1003815414577] [--chat-slug parlobyg]
|
|
|
|
The zip can be a Telegram Desktop "Export chat history" archive OR an already-
|
|
extracted directory containing messages*.html.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import html
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import uuid
|
|
import zipfile
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone, timedelta
|
|
from html.parser import HTMLParser
|
|
from pathlib import Path
|
|
from typing import Iterator
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Message model (flat, pre-serialization)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class TGMessage:
|
|
message_id: str
|
|
html_file: str
|
|
sender: str | None
|
|
timestamp: datetime | None
|
|
text: str
|
|
is_service: bool
|
|
is_joined: bool # continuation of prior sender (no userpic/from_name)
|
|
reply_to_id: str | None
|
|
mentions: list[str]
|
|
attachments: list[str]
|
|
reactions: list[tuple[str, int]]
|
|
forwarded_from: str | None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HTML parser
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
_TS_RE = re.compile(
|
|
r"(\d{2})\.(\d{2})\.(\d{4})\s+(\d{2}):(\d{2}):(\d{2})\s+UTC([+\-]\d{2}:\d{2})?"
|
|
)
|
|
_REPLY_ID_RE = re.compile(r"GoToMessage\((\d+)\)")
|
|
|
|
|
|
def _parse_title_ts(title: str) -> datetime | None:
|
|
"""Parse a '.date.details' title like '15.03.2026 17:10:23 UTC+01:00'."""
|
|
m = _TS_RE.search(title or "")
|
|
if not m:
|
|
return None
|
|
dd, mm, yyyy, hh, mi, ss, tz = m.groups()
|
|
if tz:
|
|
sign = 1 if tz[0] == "+" else -1
|
|
hrs, mins = tz[1:].split(":")
|
|
offset = timezone(sign * timedelta(hours=int(hrs), minutes=int(mins)))
|
|
else:
|
|
offset = timezone.utc
|
|
return datetime(
|
|
int(yyyy), int(mm), int(dd),
|
|
int(hh), int(mi), int(ss),
|
|
tzinfo=offset,
|
|
)
|
|
|
|
|
|
class _TelegramExportParser(HTMLParser):
|
|
"""
|
|
Walks a messages*.html file and yields one TGMessage per <div class="message ...">.
|
|
|
|
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 <div class="text bold">…</div>
|
|
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())
|