eddy/matrix_maubot_roomslister
Watch
1
0
Fork
You've already forked matrix_maubot_roomslister
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Curated room index v2: moderator-managed list, no space traversal, validation on schedule

Browse source
This commit is contained in:
Eddy de Vink 2026-09-01 23:31:21 +02:00
commit 437a39a0a0
4 changed files with 261 additions and 164 deletions
Download patch file Download diff file Expand all files Collapse all files

25
base-config.yaml
View file

@ -1,19 +1,16 @@
# De room-ID van de Space die je wilt overzien, bijv. "!abcdefghijk:jouwserver.tld"
# Let op: gebruik de interne !room-id, geen #alias - je vindt hem in Element via
# "Space instellingen" -> "Geavanceerd" -> "Interne room-ID".
space_id: "!vervang-mij:example.org"
# Rooms-overzicht: gecureerde index
#
# Deze plugin houdt GEEN ruimte-hierarchy bij (dat werd onbeheersbaar door
# cyclische subspace-grafen). In plaats daarvan beheren moderators een eigen
# lijst: `!rooms add <room-id-of-alias>`, `!rooms remove <...>`, `!rooms refresh`.
# Hoe diep de bot in geneste subspaces mag kijken.
max_depth: 5
# Hoe vaak (minuten) de bot controleert of de rooms in de index nog bestaan.
# 24 uur is prima: pas bij ~1440 minuten wordt opnieuw gevalideerd.
refresh_interval_minutes: 1440
# Hoeveel rooms er per bericht getoond worden (Matrix-events hebben een max grootte,
# dus bij ~200 rooms wordt de lijst automatisch over meerdere berichten verdeeld).
# dus bij veel rooms wordt de lijst automatisch over meerdere berichten verdeeld).
rooms_per_message: 50
# Harde bovengrens voor het totale aantal rooms dat wordt opgehaald/verstuurd.
# Voorkomt dat een enorme (gefedereerde) space-boom leidt tot duizenden berichten.
max_rooms: 300
# Hoe vaak (minuten) de bot de ruimte-hierarchy opnieuw indexeert op de
# achtergrond. 24 uur is prima: pas bij ~1440 minuten wordt opnieuw geïndexeerd.
refresh_interval_minutes: 1440
# Harde bovengrens voor het aantal rooms in de index.
max_rooms: 300

2
maubot.yaml
View file

@ -1,6 +1,6 @@
maubot: 0.1.0
id: nl.roomsoverview
version: 1.0.9
version: 2.0.0
license: MIT
modules:
- roomsoverview

BIN
roomsoverview.mbp
View file

Binary file not shown.

394
roomsoverview.py
View file

@ -13,6 +13,7 @@ from mautrix.util.async_db import Connection, Scheme, UpgradeTable
from mautrix.util.config import BaseProxyConfig, ConfigUpdateHelper
_REQUEST_TIMEOUT = aiohttp.ClientTimeout(total=30)
MOD_POWER_LEVEL = 50
upgrade_table = UpgradeTable()
@ -39,6 +40,40 @@ async def upgrade_v1(conn: Connection, scheme: Scheme) -> None:
)
@upgrade_table.register(description="Curated room index")
async def upgrade_v2(conn: Connection, scheme: Scheme) -> None:
if scheme == Scheme.SQLITE:
await conn.execute(
"""CREATE TABLE roomsoverview_rooms (
room_id TEXT PRIMARY KEY,
display_name TEXT,
canonical_alias TEXT,
topic TEXT,
member_count INTEGER,
added_by TEXT,
added_at BIGINT NOT NULL,
last_checked BIGINT NOT NULL,
alive INTEGER NOT NULL DEFAULT 1
)"""
)
await conn.execute("DROP TABLE IF EXISTS roomsoverview_cache")
else:
await conn.execute(
"""CREATE TABLE roomsoverview_rooms (
room_id TEXT PRIMARY KEY,
display_name TEXT,
canonical_alias TEXT,
topic TEXT,
member_count INTEGER,
added_by TEXT,
added_at BIGINT NOT NULL,
last_checked BIGINT NOT NULL,
alive BOOLEAN NOT NULL DEFAULT TRUE
)"""
)
await conn.execute("DROP TABLE IF EXISTS roomsoverview_cache")
def _escape_md(text: str) -> str:
"""Escape markdown-significant characters when embedding user content."""
return (
@ -55,35 +90,31 @@ def _escape_md(text: str) -> str:
class Config(BaseProxyConfig):
def do_update(self, helper: ConfigUpdateHelper) -> None:
helper.copy("space_id")
helper.copy("max_depth")
helper.copy("refresh_interval_minutes")
helper.copy("rooms_per_message")
helper.copy("max_rooms")
helper.copy("refresh_interval_minutes")
class RoomsOverviewPlugin(Plugin):
async def start(self) -> None:
self.config.load_and_update()
space_id = str(self.config.get("space_id", "") or "")
interval_minutes = max(1, int(self.config.get("refresh_interval_minutes", 1440) or 1440))
if not space_id or space_id.startswith("!vervang-mij"):
return
self._refresh_lock = asyncio.Lock()
self._scheduled_tasks = []
# Direct eenmalig indexeren (prewarm), daarna periodiek.
interval_minutes = max(
1, int(self.config.get("refresh_interval_minutes", 1440) or 1440)
)
self._scheduled_tasks.append(
self.sched.run_later(2, self._refresh_cache(space_id))
self.sched.run_later(2, self._validate_all())
)
self._scheduled_tasks.append(
self.sched.run_periodically(
interval_minutes * 60,
partial(self._refresh_cache, space_id),
partial(self._validate_all),
run_task_in_background=True,
)
)
self.log.info(
f"start: indexering gepland elke {interval_minutes} minuten"
f"start: index-validatie gepland elke {interval_minutes} minuten"
)
async def stop(self) -> None:
@ -98,164 +129,164 @@ class RoomsOverviewPlugin(Plugin):
def get_db_upgrade_table(cls) -> UpgradeTable:
return upgrade_table
async def _fetch_hierarchy(self, space_id: str) -> Tuple[List[Dict[str, Any]], bool]:
"""Fetch the paginated space hierarchy for a room.
def _base_url(self) -> str:
return str(self.client.api.base_url).rstrip("/")
Uses GET /_matrix/client/v1/rooms/{roomId}/hierarchy, which
paginates depth-first over the whole space tree. Retries on 429
rate limits and deduplicates rooms by ID across pages. Stops
early at ``max_rooms`` to avoid dumping huge federated trees.
def _headers(self) -> Dict[str, str]:
return {"Authorization": f"Bearer {self.client.api.token}"}
Returns (rooms, complete); ``complete`` is False when the fetch
aborted early (rate limits, request errors, excessive pagination
or the ``max_rooms`` cap), meaning the returned list is partial.
async def _get_room_info(
self, room_id: str
) -> Optional[Dict[str, Any]]:
"""Haal naam/topic/alias/leden van een room op.
Retourneert None als de room niet meer bestaat (M_NOT_FOUND).
"""
rooms: List[Dict[str, Any]] = []
seen: set[str] = set()
since: Optional[str] = None
base_url = str(self.client.api.base_url).rstrip("/")
token = self.client.api.token
max_depth = max(1, int(self.config.get("max_depth", 5) or 5))
max_rooms = max(1, int(self.config.get("max_rooms", 300) or 300))
headers = {"Authorization": f"Bearer {token}"}
url = f"{base_url}/_matrix/client/v1/rooms/{quote(space_id, safe='')}/hierarchy"
pages = 0
base = self._base_url()
h = self._headers()
rid = quote(room_id, safe="")
while True:
pages += 1
if pages > 100:
self.log.warning("Hierarchy: meer dan 100 paginas, stoppen")
return rooms, False
params = {
"limit": "100",
"max_depth": str(max_depth),
"suggested_only": "false",
}
if since:
params["since"] = since
retries = 0
while True:
try:
async with self.http.get(
url, params=params, headers=headers, timeout=_REQUEST_TIMEOUT
) as resp:
if resp.status == 429:
retries += 1
try:
retry_after = float(resp.headers.get("Retry-After", "1") or "1")
except ValueError:
retry_after = 1
self.log.warning(
f"Hierarchy rate-limited (429; poging {retries}): retrying in {retry_after}s"
)
if retries > 3:
self.log.warning("Hierarchy: te veel 429s, stoppen")
return rooms, False
await asyncio.sleep(min(retry_after, 30))
continue
if resp.status != 200:
body = await resp.text()
self.log.warning(f"Hierarchy request failed ({resp.status}): {body}")
return rooms, False
data = await resp.json()
break
except asyncio.CancelledError:
raise
except Exception as e:
self.log.warning(f"Hierarchy request error: {e}")
return rooms, False
for r in data.get("rooms", []):
rid = r.get("room_id")
if rid and rid not in seen:
seen.add(rid)
rooms.append(r)
if len(rooms) >= max_rooms:
self.log.warning(
f"Hierarchy: afgekapt op {max_rooms} rooms (boom bevat meer)"
)
return rooms[:max_rooms], False
since = data.get("next_batch")
if not since:
break
return rooms, True
async def _refresh_cache(self, space_id: str) -> None:
"""Live ophalen en in de cache opslaan (gedeeld via lock, ook achtergrond)."""
async with self._refresh_lock:
entries, complete = await self._fetch_hierarchy(space_id)
rooms = [
r
for r in entries
if r.get("room_id") != space_id
and (r.get("room_type") or "") != "m.space"
]
await self._store_cache(space_id, rooms, complete)
self.log.info(f"indexering voltooid ({len(rooms)} rooms)")
async def _load_cache(
self, space_id: str
) -> Optional[Tuple[List[Dict[str, Any]], bool, int]]:
row = await self.database.fetchrow(
"SELECT rooms, complete, fetched_at FROM roomsoverview_cache "
"WHERE space_id = $1",
space_id,
)
if not row:
async def state(typ: str) -> Optional[Dict[str, Any]]:
try:
async with self.http.get(
f"{base}/_matrix/client/v3/rooms/{rid}/state/{typ}",
headers=h,
timeout=_REQUEST_TIMEOUT,
) as r:
if r.status == 200:
return await r.json()
except Exception:
pass
return None
return json.loads(row["rooms"]), bool(row["complete"]), int(row["fetched_at"])
async def _store_cache(
self, space_id: str, rooms: List[Dict[str, Any]], complete: bool
) -> None:
await self.database.execute(
"INSERT INTO roomsoverview_cache (space_id, rooms, complete, fetched_at) "
"VALUES ($1, $2, $3, $4) "
"ON CONFLICT (space_id) DO UPDATE SET rooms = $2, complete = $3, fetched_at = $4",
space_id,
json.dumps(rooms, ensure_ascii=False),
complete,
int(time.time()),
name, topic, alias = await asyncio.gather(
state("m.room.name"),
state("m.room.topic"),
state("m.room.canonical_alias"),
)
@command.new(name="rooms", help="Toon alle rooms in de geconfigureerde space")
async def handle_rooms(self, evt: MessageEvent) -> None:
space_id = str(self.config.get("space_id", "") or "")
if not space_id or space_id.startswith("!vervang-mij"):
await evt.reply(
"Er is nog geen geldige `space_id` ingesteld in de plugin-config."
members: Optional[int] = None
try:
async with self.http.get(
f"{base}/_matrix/client/v3/rooms/{rid}/joined_members",
headers=h,
timeout=_REQUEST_TIMEOUT,
) as r:
if r.status == 200:
d = await r.json()
members = len(d.get("joined", {}))
elif r.status == 404:
return None # room bestaat niet (meer)
# 403: bestaat, maar bot heeft geen toegang
except Exception:
pass
return {
"name": (name or {}).get("name", "") or "",
"topic": (topic or {}).get("topic", "") or "",
"alias": (alias or {}).get("alias", "") or "",
"members": members,
}
async def _resolve_alias(self, alias: str) -> Optional[str]:
"""Los een #alias op naar een room_id."""
a = alias.lstrip("#")
try:
async with self.http.get(
f"{self._base_url()}/_matrix/client/v3/directory/room/{quote(a, safe='')}",
headers=self._headers(),
timeout=_REQUEST_TIMEOUT,
) as r:
if r.status == 200:
return (await r.json()).get("room_id")
except Exception:
pass
return None
async def _is_mod(self, room_id: str, user_id: str) -> bool:
"""Moderator (power level >= 50) in de gegeven room."""
try:
async with self.http.get(
f"{self._base_url()}/_matrix/client/v3/rooms/{quote(room_id, safe='')}/state/m.room.power_levels",
headers=self._headers(),
timeout=_REQUEST_TIMEOUT,
) as r:
if r.status != 200:
return False
d = await r.json()
except Exception:
return False
users = d.get("users", {}) or {}
default = int(d.get("users_default", 0) or 0)
pl = int(users.get(user_id, default) or 0)
return pl >= MOD_POWER_LEVEL
async def _validate_all(self) -> Tuple[int, int]:
"""Valideer alle rooms in de index; verwijder verdwenen. (removed, total)"""
async with self._refresh_lock:
rows = await self.database.fetch(
"SELECT room_id FROM roomsoverview_rooms"
)
return
removed = 0
for row in rows:
rid = row["room_id"]
info = await self._get_room_info(rid)
if info is None:
await self.database.execute(
"DELETE FROM roomsoverview_rooms WHERE room_id = $1", rid
)
removed += 1
else:
await self.database.execute(
"UPDATE roomsoverview_rooms SET display_name = $2, "
"canonical_alias = $3, topic = $4, member_count = $5, "
"last_checked = $6 WHERE room_id = $1",
rid,
info["name"] or None,
info["alias"] or None,
info["topic"] or None,
info["members"],
int(time.time()),
)
if removed:
self.log.info(f"validatie: {removed} room(s) verwijderd (bestaan niet meer)")
return removed, len(rows)
await evt.mark_read()
cached = await self._load_cache(space_id)
if cached is None:
# Nog nooit geïndexeerd: vraag een indexering aan en wacht erop
# (lock deelt met de achtergrondtaak, dus geen dubbele fetches).
await self._refresh_cache(space_id)
cached = await self._load_cache(space_id)
rooms, complete = cached[0], cached[1] if cached else ([], True)
async def _list_rooms(self) -> List[Dict[str, Any]]:
rows = await self.database.fetch(
"SELECT room_id, display_name, canonical_alias, topic, member_count "
"FROM roomsoverview_rooms WHERE alive ORDER BY "
"COALESCE(display_name, canonical_alias, room_id) ASC"
)
return [dict(r) for r in rows]
async def _send_list(self, evt: MessageEvent) -> None:
rooms = await self._list_rooms()
if not rooms:
await evt.reply(
"Er zijn (nog) geen rooms gevonden. Controleer de space_id en of de bot "
"lid is van (of toegang heeft tot) de space."
"De room-index is nog leeg. Moderators kunnen rooms toevoegen met "
"`!rooms add <room-id-of-alias>`."
)
return
def sort_key(r: Dict[str, Any]) -> str:
return (r.get("name") or r.get("canonical_alias") or r["room_id"]).lower()
return (
r.get("display_name")
or r.get("canonical_alias")
or r["room_id"]
).lower()
rooms.sort(key=sort_key)
lines = []
for r in rooms:
name = _escape_md(r.get("name") or r.get("canonical_alias") or r["room_id"])
name = _escape_md(
r.get("display_name") or r.get("canonical_alias") or r["room_id"]
)
alias = r.get("canonical_alias")
room_id = r["room_id"]
link_target = quote(alias if alias else room_id, safe="!")
member_count = r.get("num_joined_members")
member_count = r.get("member_count")
topic = _escape_md((r.get("topic") or "").strip().replace("\n", " "))
suffix = f" ({member_count} leden)" if member_count is not None else ""
line = f"* [{name}](https://matrix.to/#/{link_target}){suffix}"
@ -269,6 +300,75 @@ class RoomsOverviewPlugin(Plugin):
for idx, chunk in enumerate(chunks, start=1):
header = f"**Rooms overzicht ({total} rooms getoond, deel {idx}/{len(chunks)}):**\n\n"
if idx == 1 and not complete:
header += "*De lijst is mogelijk onvolledig: de laatste indexering was niet compleet.*\n\n"
await evt.respond(header + "\n".join(chunk), markdown=True)
await evt.respond(header + "\n".join(chunk), markdown=True)
@command.new(name="rooms", help="Toon de room-index; subcommand: add, remove, refresh")
async def handle_rooms(self, evt: MessageEvent) -> None:
await evt.mark_read()
await self._send_list(evt)
@handle_rooms.subcommand(name="add", help="Voeg een room toe (moderator+)")
async def add_room(self, evt: MessageEvent, room: str) -> None:
if not await self._is_mod(evt.room_id, evt.sender):
await evt.reply("Geen permissie: moderator of hoger vereist.")
return
room_id = room if room.startswith("!") else await self._resolve_alias(room)
if not room_id:
await evt.reply(f"Room `{room}` kon niet gevonden worden.")
return
info = await self._get_room_info(room_id)
if info is None:
await evt.reply(f"Room `{room_id}` bestaat niet (of is niet meer bereikbaar).")
return
max_rooms = max(1, int(self.config.get("max_rooms", 300) or 300))
count = await self.database.fetchval(
"SELECT COUNT(*) FROM roomsoverview_rooms"
)
if count >= max_rooms:
await evt.reply(
f"Index zit vol ({max_rooms} rooms max). Verwijder eerst iets of verhoog max_rooms."
)
return
await self.database.execute(
"INSERT INTO roomsoverview_rooms "
"(room_id, display_name, canonical_alias, topic, member_count, added_by, added_at, last_checked) "
"VALUES ($1, $2, $3, $4, $5, $6, $7, $7) "
"ON CONFLICT (room_id) DO UPDATE SET display_name = $2, canonical_alias = $3, "
"topic = $4, member_count = $5, last_checked = $7",
room_id,
info["name"] or None,
info["alias"] or None,
info["topic"] or None,
info["members"],
evt.sender,
int(time.time()),
)
await evt.reply(
f"Room toegevoegd: **{info['name'] or room_id}** "
f"(matrix.to/#/{quote(info['alias'] or room_id, safe='!')})"
)
@handle_rooms.subcommand(name="remove", help="Verwijder een room (moderator+)")
async def remove_room(self, evt: MessageEvent, room: str) -> None:
if not await self._is_mod(evt.room_id, evt.sender):
await evt.reply("Geen permissie: moderator of hoger vereist.")
return
room_id = room if room.startswith("!") else await self._resolve_alias(room)
if not room_id:
await evt.reply(f"Room `{room}` kon niet gevonden worden.")
return
result = await self.database.execute(
"DELETE FROM roomsoverview_rooms WHERE room_id = $1", room_id
)
if result == "DELETE 1":
await evt.reply(f"Room verwijderd: `{room_id}`.")
else:
await evt.reply(f"Room `{room_id}` stond niet in de index.")
@handle_rooms.subcommand(name="refresh", help="Valideer de index nu (moderator+)")
async def refresh_index(self, evt: MessageEvent) -> None:
if not await self._is_mod(evt.room_id, evt.sender):
await evt.reply("Geen permissie: moderator of hoger.")
return
removed, total = await self._validate_all()
await evt.reply(f"Index gecontroleerd: {total - removed} rooms over, {removed} verwijderd (bestonden niet meer).")