4dae5477ce
Some checks failed
Build & Push Docker Image / build-and-push (push) Failing after 5s
- FastAPI-based Matrix chat export tool with E2E support - Dockerfile with Python 3.12-slim base image - Docker Compose with dedicated network configuration - Gitea Actions workflow for automated image builds - README with setup instructions and feature overview - MIT license Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
838 lines
32 KiB
Python
838 lines
32 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
import hashlib
|
|
import zipfile
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
from typing import AsyncGenerator
|
|
import shutil
|
|
from urllib.parse import quote
|
|
|
|
import aiohttp
|
|
import aiofiles
|
|
|
|
from nio import (
|
|
AsyncClient,
|
|
AsyncClientConfig,
|
|
LoginResponse,
|
|
LoginError,
|
|
RoomMessagesResponse,
|
|
RoomMessagesError,
|
|
MegolmEvent,
|
|
DownloadResponse,
|
|
DownloadError,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SENDER_COLORS = [
|
|
"#ff6b6b", "#ffa502", "#7bed9f", "#70a1ff", "#5352ed", "#ff6348",
|
|
"#2ed573", "#1e90ff", "#ff4757", "#eccc68", "#a29bfe", "#fd79a8",
|
|
"#00cec9", "#e17055", "#6c5ce7", "#fab1a0",
|
|
]
|
|
|
|
|
|
def _color_for_user(user_id: str) -> str:
|
|
h = int(hashlib.md5(user_id.encode()).hexdigest()[:8], 16)
|
|
return SENDER_COLORS[h % len(SENDER_COLORS)]
|
|
|
|
|
|
def _ts_to_datetime(ts_ms: int) -> datetime:
|
|
return datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
|
|
|
|
|
|
def _format_size(size: int) -> str:
|
|
if size > 1_048_576:
|
|
return f"{size / 1_048_576:.1f} MB"
|
|
if size > 1024:
|
|
return f"{size / 1024:.0f} KB"
|
|
return f"{size} B"
|
|
|
|
|
|
EXT_MAP = {
|
|
"image/jpeg": ".jpg", "image/png": ".png", "image/gif": ".gif",
|
|
"image/webp": ".webp", "video/mp4": ".mp4", "video/webm": ".webm",
|
|
"video/quicktime": ".mov", "audio/ogg": ".ogg", "audio/mpeg": ".mp3",
|
|
"audio/mp4": ".m4a", "audio/webm": ".weba", "audio/aac": ".aac",
|
|
"application/pdf": ".pdf",
|
|
}
|
|
|
|
|
|
MEDIA_PURGED = "__PURGED__"
|
|
|
|
|
|
class MatrixExporter:
|
|
|
|
def __init__(self, store_path: str, export_path: str):
|
|
self.store_path = Path(store_path)
|
|
self.export_path = Path(export_path)
|
|
self.client: AsyncClient | None = None
|
|
self.user_id: str | None = None
|
|
self.homeserver: str = ""
|
|
self.access_token: str = ""
|
|
self.credentials_file = self.store_path / "credentials.json"
|
|
self.store_path.mkdir(parents=True, exist_ok=True)
|
|
self.export_path.mkdir(parents=True, exist_ok=True)
|
|
self._keys_imported = False
|
|
|
|
# ------------------------------------------------------------------
|
|
# Connection
|
|
# ------------------------------------------------------------------
|
|
|
|
async def connect_password(self, homeserver: str, username: str, password: str) -> dict:
|
|
self.homeserver = homeserver.rstrip("/")
|
|
nio_store = self.store_path / "nio_store"
|
|
nio_store.mkdir(parents=True, exist_ok=True)
|
|
|
|
cfg = AsyncClientConfig(store_sync_tokens=True, encryption_enabled=True)
|
|
self.client = AsyncClient(self.homeserver, username, store_path=str(nio_store), config=cfg)
|
|
|
|
resp = await self.client.login(password)
|
|
if isinstance(resp, LoginError):
|
|
raise Exception(f"Login fehlgeschlagen: {resp.message}")
|
|
|
|
self.user_id = resp.user_id
|
|
self.access_token = resp.access_token
|
|
|
|
self.credentials_file.write_text(json.dumps({
|
|
"homeserver": self.homeserver,
|
|
"user_id": resp.user_id,
|
|
"device_id": resp.device_id,
|
|
"access_token": resp.access_token,
|
|
}, indent=2))
|
|
|
|
return {"user_id": resp.user_id, "device_id": resp.device_id, "new_session": True}
|
|
|
|
async def connect_token(self, homeserver: str, token: str) -> dict:
|
|
self.homeserver = homeserver.rstrip("/")
|
|
self.access_token = token
|
|
|
|
async with aiohttp.ClientSession() as s:
|
|
async with s.get(
|
|
f"{self.homeserver}/_matrix/client/v3/account/whoami",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
) as r:
|
|
if r.status != 200:
|
|
raise Exception(f"Token ungültig (HTTP {r.status})")
|
|
data = await r.json()
|
|
self.user_id = data["user_id"]
|
|
device_id = data.get("device_id", "EXPORT")
|
|
|
|
nio_store = self.store_path / "nio_store"
|
|
nio_store.mkdir(parents=True, exist_ok=True)
|
|
|
|
cfg = AsyncClientConfig(store_sync_tokens=True, encryption_enabled=True)
|
|
self.client = AsyncClient(
|
|
self.homeserver, self.user_id,
|
|
store_path=str(nio_store), config=cfg,
|
|
)
|
|
self.client.restore_login(self.user_id, device_id, token)
|
|
|
|
self.credentials_file.write_text(json.dumps({
|
|
"homeserver": self.homeserver,
|
|
"user_id": self.user_id,
|
|
"device_id": device_id,
|
|
"access_token": token,
|
|
}, indent=2))
|
|
|
|
return {"user_id": self.user_id, "device_id": device_id, "new_session": False}
|
|
|
|
async def restore_session(self) -> dict | None:
|
|
if not self.credentials_file.exists():
|
|
return None
|
|
creds = json.loads(self.credentials_file.read_text())
|
|
self.homeserver = creds["homeserver"]
|
|
self.user_id = creds["user_id"]
|
|
self.access_token = creds["access_token"]
|
|
|
|
nio_store = self.store_path / "nio_store"
|
|
nio_store.mkdir(parents=True, exist_ok=True)
|
|
cfg = AsyncClientConfig(store_sync_tokens=True, encryption_enabled=True)
|
|
self.client = AsyncClient(
|
|
self.homeserver, self.user_id,
|
|
store_path=str(nio_store), config=cfg,
|
|
)
|
|
self.client.restore_login(creds["user_id"], creds["device_id"], creds["access_token"])
|
|
return {
|
|
"user_id": self.user_id,
|
|
"device_id": creds["device_id"],
|
|
"homeserver": self.homeserver,
|
|
"restored": True,
|
|
}
|
|
|
|
async def disconnect(self):
|
|
if self.client:
|
|
await self.client.close()
|
|
self.client = None
|
|
|
|
# ------------------------------------------------------------------
|
|
# E2E Key import
|
|
# ------------------------------------------------------------------
|
|
|
|
async def import_keys(self, key_file: str, passphrase: str) -> dict:
|
|
if not self.client:
|
|
raise Exception("Nicht verbunden")
|
|
self.client.load_store()
|
|
keys = await self.client.import_keys(key_file, passphrase)
|
|
self._keys_imported = True
|
|
return {"success": True, "message": f"Schlüssel importiert ({keys} Sessions)"}
|
|
|
|
@property
|
|
def keys_imported(self) -> bool:
|
|
return self._keys_imported
|
|
|
|
# ------------------------------------------------------------------
|
|
# Room listing
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _fetch_room_info(self, session: aiohttp.ClientSession, room_id: str, headers: dict) -> dict:
|
|
encoded = quote(room_id, safe="")
|
|
name = room_id
|
|
member_count = 0
|
|
encrypted = False
|
|
topic = ""
|
|
members: list[str] = []
|
|
|
|
async with session.get(
|
|
f"{self.homeserver}/_matrix/client/v3/rooms/{encoded}/state",
|
|
headers=headers,
|
|
) as r:
|
|
if r.status == 200:
|
|
states = await r.json()
|
|
for st in states:
|
|
t = st.get("type", "")
|
|
c = st.get("content", {})
|
|
if t == "m.room.name":
|
|
name = c.get("name", room_id)
|
|
elif t == "m.room.encryption":
|
|
encrypted = True
|
|
elif t == "m.room.topic":
|
|
topic = c.get("topic", "")
|
|
elif t == "m.room.member" and c.get("membership") == "join":
|
|
member_count += 1
|
|
if st.get("state_key") != self.user_id:
|
|
members.append(c.get("displayname", st.get("state_key", "?")))
|
|
|
|
if name == room_id and members:
|
|
name = ", ".join(members[:3])
|
|
if len(members) > 3:
|
|
name += f" +{len(members) - 3}"
|
|
|
|
return {
|
|
"room_id": room_id,
|
|
"name": name,
|
|
"member_count": member_count,
|
|
"encrypted": encrypted,
|
|
"topic": topic,
|
|
}
|
|
|
|
async def list_rooms(self) -> list[dict]:
|
|
if not self.access_token:
|
|
raise Exception("Nicht verbunden")
|
|
|
|
headers = {"Authorization": f"Bearer {self.access_token}"}
|
|
rooms = []
|
|
|
|
async with aiohttp.ClientSession() as s:
|
|
async with s.get(f"{self.homeserver}/_matrix/client/v3/joined_rooms", headers=headers) as r:
|
|
data = await r.json()
|
|
room_ids = data.get("joined_rooms", [])
|
|
|
|
for room_id in room_ids:
|
|
try:
|
|
info = await self._fetch_room_info(s, room_id, headers)
|
|
rooms.append(info)
|
|
except Exception:
|
|
pass
|
|
|
|
rooms.sort(key=lambda r: r["name"].lower())
|
|
return rooms
|
|
|
|
async def list_rooms_stream(self) -> AsyncGenerator[dict, None]:
|
|
if not self.access_token:
|
|
raise Exception("Nicht verbunden")
|
|
|
|
headers = {"Authorization": f"Bearer {self.access_token}"}
|
|
|
|
async with aiohttp.ClientSession() as s:
|
|
async with s.get(f"{self.homeserver}/_matrix/client/v3/joined_rooms", headers=headers) as r:
|
|
data = await r.json()
|
|
room_ids = data.get("joined_rooms", [])
|
|
|
|
yield {"type": "total", "count": len(room_ids)}
|
|
|
|
for room_id in room_ids:
|
|
try:
|
|
info = await self._fetch_room_info(s, room_id, headers)
|
|
yield {"type": "room", "data": info}
|
|
except Exception as e:
|
|
logger.warning("Raum-Info fehlgeschlagen für %s: %s", room_id, e)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Export
|
|
# ------------------------------------------------------------------
|
|
|
|
async def export_room(self, room_id: str) -> AsyncGenerator[dict, None]:
|
|
if not self.client:
|
|
yield _log("error", "Nicht verbunden")
|
|
return
|
|
|
|
safe_name = room_id.replace("!", "").replace(":", "_").replace(".", "_")
|
|
room_dir = self.export_path / safe_name
|
|
room_dir.mkdir(parents=True, exist_ok=True)
|
|
media_dir = room_dir / "media"
|
|
media_dir.mkdir(exist_ok=True)
|
|
|
|
yield _log("info", "Starte Export …")
|
|
|
|
# --- sync to get pagination token ---
|
|
yield _log("info", "Synchronisiere mit Server …", progress=2)
|
|
try:
|
|
sync_filter = json.dumps({
|
|
"room": {"timeline": {"limit": 1}, "state": {"lazy_load_members": True}},
|
|
"presence": {"types": []},
|
|
"account_data": {"types": []},
|
|
})
|
|
await self.client.sync(timeout=30000, sync_filter=sync_filter)
|
|
start_token = self.client.next_batch
|
|
except Exception as e:
|
|
yield _log("error", f"Sync fehlgeschlagen: {e}")
|
|
return
|
|
|
|
if not start_token:
|
|
yield _log("error", "Kein Sync-Token erhalten")
|
|
return
|
|
|
|
# --- load state for incremental ---
|
|
state_file = room_dir / ".state.json"
|
|
existing_ids: set[str] = set()
|
|
existing_events: list[dict] = []
|
|
events_file = room_dir / "events.jsonl"
|
|
|
|
if events_file.exists():
|
|
try:
|
|
for line in events_file.read_text(encoding="utf-8").splitlines():
|
|
if line.strip():
|
|
ev = json.loads(line)
|
|
existing_ids.add(ev.get("event_id", ""))
|
|
existing_events.append(ev)
|
|
yield _log("info", f"{len(existing_events)} vorhandene Events geladen (inkrementell)", progress=3)
|
|
except Exception:
|
|
existing_events = []
|
|
existing_ids = set()
|
|
|
|
# --- fetch messages ---
|
|
yield _log("info", "Lade Nachrichten …", progress=5)
|
|
|
|
raw_events: list[dict] = []
|
|
from_token = start_token
|
|
page = 0
|
|
decrypted_count = 0
|
|
encrypted_fail = 0
|
|
|
|
while True:
|
|
page += 1
|
|
try:
|
|
resp = await self.client.room_messages(room_id, start=from_token, limit=200)
|
|
except Exception as e:
|
|
yield _log("warning", f"Fehler bei Seite {page}: {e}")
|
|
break
|
|
|
|
if isinstance(resp, RoomMessagesError):
|
|
yield _log("error", f"API-Fehler: {resp.message}")
|
|
break
|
|
|
|
if not resp.chunk:
|
|
break
|
|
|
|
for event in resp.chunk:
|
|
source = getattr(event, "source", {})
|
|
eid = source.get("event_id", "")
|
|
if eid in existing_ids:
|
|
continue
|
|
|
|
if isinstance(event, MegolmEvent):
|
|
try:
|
|
decrypted = self.client.decrypt_event(event)
|
|
if decrypted:
|
|
source = decrypted.source
|
|
decrypted_count += 1
|
|
else:
|
|
encrypted_fail += 1
|
|
except Exception:
|
|
encrypted_fail += 1
|
|
|
|
raw_events.append(source)
|
|
existing_ids.add(eid)
|
|
|
|
from_token = resp.end
|
|
if page % 3 == 0:
|
|
yield _log("progress", f"Seite {page}: {len(raw_events)} neue Nachrichten", progress=min(5 + page, 40))
|
|
|
|
if not resp.end:
|
|
break
|
|
|
|
raw_events.reverse()
|
|
|
|
yield _log("info",
|
|
f"{len(raw_events)} neue Nachrichten ({decrypted_count} entschlüsselt, "
|
|
f"{encrypted_fail} nicht entschlüsselbar)", progress=42)
|
|
|
|
# --- merge with existing ---
|
|
all_events = existing_events + raw_events
|
|
all_events.sort(key=lambda e: e.get("origin_server_ts", 0))
|
|
|
|
# --- save events ---
|
|
async with aiofiles.open(events_file, "w", encoding="utf-8") as f:
|
|
for ev in all_events:
|
|
await f.write(json.dumps(ev, ensure_ascii=False) + "\n")
|
|
|
|
yield _log("info", f"Gesamt: {len(all_events)} Events", progress=45)
|
|
|
|
# --- download media ---
|
|
yield _log("info", "Lade Mediendateien …", progress=46)
|
|
|
|
media_map: dict[str, str] = {}
|
|
media_items = [e for e in all_events if _has_media(e)]
|
|
media_total = len(media_items)
|
|
media_errors = 0
|
|
media_purged = 0
|
|
|
|
async with aiohttp.ClientSession() as http_session:
|
|
for i, ev in enumerate(media_items):
|
|
eid = ev.get("event_id", "")
|
|
fname = await self._download_media(ev, media_dir, http_session)
|
|
if fname == MEDIA_PURGED:
|
|
media_purged += 1
|
|
elif fname:
|
|
media_map[eid] = fname
|
|
else:
|
|
media_errors += 1
|
|
|
|
if (i + 1) % 10 == 0 or i == media_total - 1:
|
|
yield _log("progress",
|
|
f"Medien: {i + 1}/{media_total} ({len(media_map)} OK, "
|
|
f"{media_purged} gelöscht, {media_errors} Fehler)",
|
|
progress=int(46 + 40 * (i + 1) / max(media_total, 1)))
|
|
|
|
parts = [f"{len(media_map)}/{media_total} Medien heruntergeladen"]
|
|
if media_purged:
|
|
parts.append(f"{media_purged} nicht mehr verfügbar (Server-Löschpolicy)")
|
|
if media_errors:
|
|
parts.append(f"{media_errors} Fehler")
|
|
yield _log("info", ", ".join(parts), progress=88)
|
|
|
|
# --- get room name + members ---
|
|
room_name = room_id
|
|
member_names: dict[str, str] = {}
|
|
try:
|
|
headers = {"Authorization": f"Bearer {self.access_token}"}
|
|
encoded_rid = quote(room_id, safe="")
|
|
async with aiohttp.ClientSession() as s:
|
|
async with s.get(
|
|
f"{self.homeserver}/_matrix/client/v3/rooms/{encoded_rid}/state",
|
|
headers=headers,
|
|
) as r:
|
|
if r.status == 200:
|
|
for st in await r.json():
|
|
t = st.get("type", "")
|
|
c = st.get("content", {})
|
|
if t == "m.room.name":
|
|
room_name = c.get("name", room_id)
|
|
elif t == "m.room.member" and c.get("membership") == "join":
|
|
uid = st.get("state_key", "")
|
|
member_names[uid] = c.get("displayname", uid.split(":")[0].lstrip("@"))
|
|
except Exception:
|
|
pass
|
|
|
|
if room_name == room_id and member_names:
|
|
others = [n for uid, n in member_names.items() if uid != self.user_id]
|
|
if others:
|
|
room_name = ", ".join(others[:3])
|
|
|
|
# --- generate HTML ---
|
|
yield _log("info", "Generiere HTML …", progress=90)
|
|
html_path = room_dir / "chat.html"
|
|
await self._generate_html(all_events, media_map, member_names, room_name, self.user_id or "", html_path)
|
|
|
|
# --- create ZIP ---
|
|
yield _log("info", "Erstelle ZIP-Archiv …", progress=95)
|
|
zip_result = self.create_zip(safe_name)
|
|
zip_api_path = f"/api/exports/{safe_name}/zip" if zip_result else None
|
|
|
|
# --- save state ---
|
|
state = {
|
|
"room_id": room_id,
|
|
"room_name": room_name,
|
|
"total_events": len(all_events),
|
|
"total_media": len(media_map),
|
|
"last_export": datetime.now().isoformat(),
|
|
"last_token": from_token,
|
|
}
|
|
state_file.write_text(json.dumps(state, indent=2))
|
|
|
|
rel_path = f"/exports/{safe_name}/chat.html"
|
|
yield _log("done",
|
|
f"Export abgeschlossen! {len(all_events)} Nachrichten, {len(media_map)} Medien",
|
|
progress=100, html_path=rel_path, zip_path=zip_api_path)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Media download
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _download_media(self, ev: dict, media_dir: Path,
|
|
session: aiohttp.ClientSession | None = None) -> str | None:
|
|
content = ev.get("content", {})
|
|
file_info = content.get("file", {})
|
|
encrypted = bool(file_info and file_info.get("url"))
|
|
|
|
mxc_url = file_info.get("url", "") if encrypted else content.get("url", "")
|
|
if not mxc_url or not mxc_url.startswith("mxc://"):
|
|
mxc_url = content.get("avatar_url") or ""
|
|
if not mxc_url or not mxc_url.startswith("mxc://"):
|
|
return None
|
|
encrypted = False
|
|
|
|
parts = mxc_url[6:].split("/", 1)
|
|
if len(parts) != 2:
|
|
return None
|
|
server_name, media_id = parts
|
|
|
|
mimetype = content.get("info", {}).get("mimetype", "")
|
|
ext = EXT_MAP.get(mimetype, "")
|
|
filename = f"{media_id}{ext}"
|
|
filepath = media_dir / filename
|
|
|
|
if filepath.exists():
|
|
return filename
|
|
|
|
data = None
|
|
eid = ev.get("event_id", "?")
|
|
|
|
try:
|
|
resp = await self.client.download(mxc_url)
|
|
if isinstance(resp, DownloadResponse):
|
|
data = resp.body
|
|
elif isinstance(resp, DownloadError):
|
|
logger.debug("nio download fehlgeschlagen für %s (event %s): %s", mxc_url, eid, resp.message)
|
|
except Exception as e:
|
|
logger.debug("nio download exception für %s (event %s): %s", mxc_url, eid, e)
|
|
|
|
if data is None:
|
|
headers = {"Authorization": f"Bearer {self.access_token}"}
|
|
endpoints = [
|
|
f"/_matrix/client/v1/media/download/{server_name}/{media_id}",
|
|
f"/_matrix/media/v3/download/{server_name}/{media_id}",
|
|
]
|
|
own_session = session is None
|
|
s = session or aiohttp.ClientSession()
|
|
ep_404_count = 0
|
|
try:
|
|
for ep in endpoints:
|
|
try:
|
|
async with s.get(f"{self.homeserver}{ep}", headers=headers) as r:
|
|
if r.status == 200:
|
|
data = await r.read()
|
|
break
|
|
elif r.status == 404:
|
|
ep_404_count += 1
|
|
logger.debug("HTTP 404 für %s%s (event %s)", self.homeserver, ep, eid)
|
|
else:
|
|
logger.debug("HTTP %s für %s%s (event %s)", r.status, self.homeserver, ep, eid)
|
|
except Exception as e:
|
|
logger.debug("HTTP exception für %s%s (event %s): %s", self.homeserver, ep, eid, e)
|
|
continue
|
|
finally:
|
|
if own_session:
|
|
await s.close()
|
|
|
|
if data is None and ep_404_count == len(endpoints):
|
|
logger.info("Media gelöscht (404 auf allen Endpoints): %s (event %s)", mxc_url, eid)
|
|
return MEDIA_PURGED
|
|
|
|
if data is None:
|
|
logger.warning("Media-Download fehlgeschlagen: %s (event %s)", mxc_url, eid)
|
|
return None
|
|
|
|
if encrypted and file_info:
|
|
try:
|
|
from nio.crypto.attachments import decrypt_attachment
|
|
data = decrypt_attachment(
|
|
data,
|
|
file_info["key"]["k"],
|
|
file_info["hashes"]["sha256"],
|
|
file_info["iv"],
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Entschlüsselung fehlgeschlagen für %s: %s", mxc_url, e)
|
|
return None
|
|
|
|
async with aiofiles.open(filepath, "wb") as f:
|
|
await f.write(data)
|
|
return filename
|
|
|
|
# ------------------------------------------------------------------
|
|
# Preprocessing for client-side rendering
|
|
# ------------------------------------------------------------------
|
|
|
|
def _preprocess_messages(self, events: list[dict], media_map: dict[str, str],
|
|
member_names: dict[str, str], own_user_id: str) -> list[dict]:
|
|
messages: list[dict] = []
|
|
last_date = ""
|
|
|
|
for ev in events:
|
|
ts = ev.get("origin_server_ts", 0)
|
|
dt = _ts_to_datetime(ts)
|
|
date_str = dt.strftime("%A, %d. %B %Y")
|
|
time_str = dt.strftime("%H:%M")
|
|
etype = ev.get("type", "")
|
|
content = ev.get("content", {})
|
|
|
|
if date_str != last_date:
|
|
messages.append({"t": "date", "text": date_str})
|
|
last_date = date_str
|
|
|
|
if etype == "m.room.message":
|
|
sender = ev.get("sender", "")
|
|
display = member_names.get(sender, sender.split(":")[0].lstrip("@"))
|
|
msgtype = content.get("msgtype", "m.text")
|
|
eid = ev.get("event_id", "")
|
|
mfile = media_map.get(eid, "")
|
|
|
|
msg: dict = {
|
|
"t": "msg",
|
|
"s": sender,
|
|
"n": display,
|
|
"o": sender == own_user_id,
|
|
"ts": time_str,
|
|
"mt": msgtype,
|
|
"b": content.get("body", ""),
|
|
"c": _color_for_user(sender),
|
|
}
|
|
|
|
fb = content.get("formatted_body", "")
|
|
if fb:
|
|
msg["fb"] = fb
|
|
if mfile:
|
|
msg["m"] = mfile
|
|
if ev.get("unsigned", {}).get("redacted_because"):
|
|
msg["del"] = True
|
|
if msgtype == "m.audio":
|
|
if "org.matrix.msc3245.voice" in content or "org.matrix.msc1767.audio" in content:
|
|
msg["voice"] = True
|
|
if msgtype == "m.file":
|
|
msg["fs"] = _format_size(content.get("info", {}).get("size", 0))
|
|
if msgtype == "m.location":
|
|
msg["geo"] = content.get("geo_uri", "")
|
|
|
|
messages.append(msg)
|
|
|
|
elif etype == "m.room.encrypted":
|
|
sender = ev.get("sender", "")
|
|
messages.append({
|
|
"t": "enc",
|
|
"o": sender == own_user_id,
|
|
"ts": time_str,
|
|
})
|
|
|
|
elif etype == "m.room.member":
|
|
membership = content.get("membership", "")
|
|
display = content.get("displayname", ev.get("state_key", "?"))
|
|
prev = ev.get("unsigned", {}).get("prev_content", {})
|
|
prev_membership = prev.get("membership", "")
|
|
|
|
if membership == "join" and prev_membership == "join":
|
|
old_name = prev.get("displayname") or ""
|
|
new_name = content.get("displayname") or ""
|
|
old_avatar = prev.get("avatar_url") or ""
|
|
new_avatar = content.get("avatar_url") or ""
|
|
|
|
if old_name != new_name:
|
|
if old_name and new_name:
|
|
messages.append({"t": "sys", "text": f"{old_name} heißt jetzt {new_name}"})
|
|
elif new_name:
|
|
messages.append({"t": "sys", "text": f"{ev.get('state_key', '?')} heißt jetzt {new_name}"})
|
|
|
|
if old_avatar != new_avatar:
|
|
eid = ev.get("event_id", "")
|
|
mfile = media_map.get(eid, "")
|
|
text = f"{display} hat das Profilbild geändert" if new_avatar else f"{display} hat das Profilbild entfernt"
|
|
messages.append({"t": "avatar", "text": text, "m": mfile, "ts": time_str})
|
|
|
|
elif membership == "join":
|
|
messages.append({"t": "sys", "text": f"{display} ist beigetreten"})
|
|
elif membership == "leave":
|
|
sender = ev.get("sender", "")
|
|
if sender != ev.get("state_key", ""):
|
|
messages.append({"t": "sys", "text": f"{display} wurde entfernt"})
|
|
else:
|
|
messages.append({"t": "sys", "text": f"{display} hat den Raum verlassen"})
|
|
elif membership == "invite":
|
|
messages.append({"t": "sys", "text": f"{display} wurde eingeladen"})
|
|
elif membership == "ban":
|
|
messages.append({"t": "sys", "text": f"{display} wurde gebannt"})
|
|
|
|
elif etype == "m.room.topic":
|
|
topic = content.get("topic", "")
|
|
if topic:
|
|
messages.append({"t": "sys", "text": f"Thema geändert: {topic}"})
|
|
else:
|
|
messages.append({"t": "sys", "text": "Thema entfernt"})
|
|
|
|
elif etype == "m.room.name":
|
|
rname = content.get("name", "")
|
|
if rname:
|
|
messages.append({"t": "sys", "text": f"Raumname geändert zu: {rname}"})
|
|
|
|
elif etype == "m.room.avatar":
|
|
eid = ev.get("event_id", "")
|
|
mfile = media_map.get(eid, "")
|
|
messages.append({"t": "avatar", "text": "Raumbild wurde geändert", "m": mfile, "ts": time_str})
|
|
|
|
elif etype == "m.sticker":
|
|
sender = ev.get("sender", "")
|
|
eid = ev.get("event_id", "")
|
|
mfile = media_map.get(eid, "")
|
|
if mfile:
|
|
messages.append({
|
|
"t": "sticker",
|
|
"o": sender == own_user_id,
|
|
"ts": time_str,
|
|
"b": content.get("body", "Sticker"),
|
|
"m": mfile,
|
|
})
|
|
|
|
return messages
|
|
|
|
# ------------------------------------------------------------------
|
|
# HTML generation (client-side rendered)
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _generate_html(
|
|
self,
|
|
events: list[dict],
|
|
media_map: dict[str, str],
|
|
member_names: dict[str, str],
|
|
room_name: str,
|
|
own_user_id: str,
|
|
output_path: Path,
|
|
):
|
|
messages = self._preprocess_messages(events, media_map, member_names, own_user_id)
|
|
|
|
total = len(events)
|
|
first_ts = events[0].get("origin_server_ts", 0) if events else 0
|
|
last_ts = events[-1].get("origin_server_ts", 0) if events else 0
|
|
first_date = _ts_to_datetime(first_ts).strftime("%d.%m.%Y") if first_ts else ""
|
|
last_date = _ts_to_datetime(last_ts).strftime("%d.%m.%Y") if last_ts else ""
|
|
now = datetime.now().strftime("%d.%m.%Y %H:%M")
|
|
|
|
data = {
|
|
"roomName": room_name,
|
|
"total": total,
|
|
"firstDate": first_date,
|
|
"lastDate": last_date,
|
|
"exportDate": now,
|
|
"messages": messages,
|
|
}
|
|
|
|
json_str = json.dumps(data, ensure_ascii=False).replace("</", "<\\/")
|
|
|
|
template_path = Path(__file__).parent / "templates" / "chat_export.html"
|
|
template = template_path.read_text(encoding="utf-8")
|
|
html = template.replace("__DATA__", json_str)
|
|
|
|
async with aiofiles.open(output_path, "w", encoding="utf-8") as f:
|
|
await f.write(html)
|
|
|
|
# ------------------------------------------------------------------
|
|
# ZIP creation
|
|
# ------------------------------------------------------------------
|
|
|
|
def create_zip(self, room_dir_name: str) -> Path | None:
|
|
room_dir = self.export_path / room_dir_name
|
|
if not room_dir.exists():
|
|
return None
|
|
|
|
html_file = room_dir / "chat.html"
|
|
if not html_file.exists():
|
|
return None
|
|
|
|
zip_path = room_dir / "export.zip"
|
|
|
|
with zipfile.ZipFile(zip_path, "w") as zf:
|
|
zf.write(html_file, "chat.html", compress_type=zipfile.ZIP_DEFLATED)
|
|
|
|
media_dir = room_dir / "media"
|
|
if media_dir.exists():
|
|
for fpath in sorted(media_dir.iterdir()):
|
|
if fpath.is_file():
|
|
zf.write(fpath, f"media/{fpath.name}", compress_type=zipfile.ZIP_STORED)
|
|
|
|
return zip_path
|
|
|
|
# ------------------------------------------------------------------
|
|
# Existing exports
|
|
# ------------------------------------------------------------------
|
|
|
|
def delete_export(self, room_dir_name: str) -> bool:
|
|
room_dir = self.export_path / room_dir_name
|
|
if not room_dir.exists():
|
|
return False
|
|
shutil.rmtree(room_dir)
|
|
return True
|
|
|
|
def list_exports(self) -> list[dict]:
|
|
exports = []
|
|
if not self.export_path.exists():
|
|
return exports
|
|
for d in sorted(self.export_path.iterdir()):
|
|
if d.is_dir() and (d / "chat.html").exists():
|
|
state = {}
|
|
sf = d / ".state.json"
|
|
if sf.exists():
|
|
try:
|
|
state = json.loads(sf.read_text())
|
|
except Exception:
|
|
pass
|
|
exports.append({
|
|
"dir_name": d.name,
|
|
"room_id": state.get("room_id", d.name),
|
|
"room_name": state.get("room_name", d.name),
|
|
"last_export": state.get("last_export", "?"),
|
|
"total_events": state.get("total_events", 0),
|
|
"total_media": state.get("total_media", 0),
|
|
"html_path": f"/exports/{d.name}/chat.html",
|
|
"zip_path": f"/api/exports/{d.name}/zip",
|
|
})
|
|
return exports
|
|
|
|
|
|
# ======================================================================
|
|
# Helper functions
|
|
# ======================================================================
|
|
|
|
def _log(typ: str, msg: str, progress: int = 0, **extra) -> dict:
|
|
d = {"type": typ, "message": msg, "progress": progress}
|
|
d.update(extra)
|
|
return d
|
|
|
|
|
|
def _esc(text: str) -> str:
|
|
return (text
|
|
.replace("&", "&")
|
|
.replace("<", "<")
|
|
.replace(">", ">")
|
|
.replace('"', """))
|
|
|
|
|
|
def _has_media(ev: dict) -> bool:
|
|
c = ev.get("content", {})
|
|
if c.get("url") or c.get("file", {}).get("url"):
|
|
return True
|
|
if ev.get("type") == "m.room.member" and (c.get("avatar_url") or "").startswith("mxc://"):
|
|
prev = ev.get("unsigned", {}).get("prev_content", {})
|
|
if c.get("avatar_url") != prev.get("avatar_url", ""):
|
|
return True
|
|
return False
|