import asyncio import uuid from urllib.parse import quote import httpx from .base import Buttons, ChatError, MessageRef, json_body, mask_secret from .matrix_format import to_matrix BACKEND_MATRIX = "matrix" API_PREFIX = "/_matrix/client/v3" HTML_FORMAT = "org.matrix.custom.html" MSGTYPE_TEXT = "m.text" MSGTYPE_NOTICE = "m.notice" REL_REPLACE = "m.replace" EDIT_PREFIX = "* " DEFAULT_TIMEOUT = 15.0 MAX_RETRY_AFTER = 30.0 DEFAULT_RETRY_AFTER = 1.0 STATUS_TOO_MANY = 429 ERROR_MAX = 300 MAX_BODY_BYTES = 30000 TRUNCATION_MARK = "…" class MatrixError(ChatError): pass def _enc(value: str) -> str: return quote(value, safe="") def _truncate_bytes(text: str, limit: int = MAX_BODY_BYTES) -> str: encoded = text.encode("utf-8") if len(encoded) <= limit: return text mark_bytes = TRUNCATION_MARK.encode("utf-8") budget = max(limit - len(mark_bytes), 0) cut = encoded[:budget] while cut and (cut[-1] & 0xC0) == 0x80: cut = cut[:-1] return cut.decode("utf-8") + TRUNCATION_MARK def _content(text: str, silent: bool = False) -> dict: body, formatted = to_matrix(text) return {"msgtype": MSGTYPE_NOTICE if silent else MSGTYPE_TEXT, "body": _truncate_bytes(body), "format": HTML_FORMAT, "formatted_body": _truncate_bytes(formatted)} class MatrixAdapter: name = BACKEND_MATRIX def __init__(self, homeserver: str, user_id: str, token: str, default_target: str, timeout: float = DEFAULT_TIMEOUT, transport: httpx.AsyncBaseTransport | None = None, sleep=asyncio.sleep, verify: bool | str = True): self.default_target = default_target self._homeserver = homeserver.rstrip("/") self._user_id = user_id self._token = token self._sleep = sleep self._client = httpx.AsyncClient(timeout=timeout, transport=transport, verify=verify, headers={"Authorization": f"Bearer {token}"}) def _mask(self, text: str) -> str: return mask_secret(text, self._token) async def _call(self, method: str, path: str, payload: dict | None = None, params: dict | None = None) -> dict: url = f"{self._homeserver}{API_PREFIX}{path}" for attempt in range(2): try: response = await self._client.request(method, url, json=payload, params=params) except (httpx.HTTPError, httpx.InvalidURL) as exc: raise MatrixError(self._mask(f"{type(exc).__name__}: {exc}")) from None body = json_body(response) if response.status_code == STATUS_TOO_MANY and attempt == 0: retry_ms = body.get("retry_after_ms") seconds = retry_ms / 1000 if isinstance(retry_ms, (int, float)) else DEFAULT_RETRY_AFTER await self._sleep(min(float(seconds), MAX_RETRY_AFTER)) continue if response.status_code < 400: return body detail = f"{body.get('errcode', '')} {body.get('error', '')}".strip() raise MatrixError(self._mask(f"HTTP {response.status_code}: {detail}")[:ERROR_MAX]) raise MatrixError("Matrix nicht erreichbar (Rate-Limit)") async def _put_message(self, room: str, content: dict) -> str: path = f"/rooms/{_enc(room)}/send/m.room.message/{uuid.uuid4().hex}" body = await self._call("PUT", path, content) event_id = body.get("event_id") if not event_id: raise MatrixError("Antwort ohne event_id") return str(event_id) async def send(self, target: str, text: str, buttons: Buttons = (), silent: bool = False) -> MessageRef: return MessageRef(self.name, target, await self._put_message(target, _content(text, silent))) async def edit(self, ref: MessageRef, text: str, buttons: Buttons = (), silent: bool = False) -> None: new_content = _content(text, silent) content = { "msgtype": new_content["msgtype"], "body": EDIT_PREFIX + new_content["body"], "format": HTML_FORMAT, "formatted_body": EDIT_PREFIX + new_content["formatted_body"], "m.new_content": new_content, "m.relates_to": {"rel_type": REL_REPLACE, "event_id": ref.message_id}, } await self._put_message(ref.target, content) async def reply(self, room: str, text: str, html: str | None, event_id: str) -> str: content = {"msgtype": MSGTYPE_TEXT, "body": text, "m.relates_to": {"m.in_reply_to": {"event_id": event_id}}} if html: content.update({"format": HTML_FORMAT, "formatted_body": to_matrix(html)[1]}) return await self._put_message(room, content) async def sync(self, since: str | None, timeout_ms: int, filter_json: str) -> dict: params: dict = {"timeout": timeout_ms, "filter": filter_json} if since: params["since"] = since return await self._call("GET", "/sync", params=params) async def check(self) -> str: who = await self._call("GET", "/account/whoami") actual = who.get("user_id") if actual != self._user_id: raise MatrixError(f"Token gehört zu {actual or '?'}, erwartet {self._user_id}") if not self.default_target: return f"Angemeldet als {actual}" joined = (await self._call("GET", "/joined_rooms")).get("joined_rooms") or [] if self.default_target not in joined: raise MatrixError(f"Bot ist dem Raum {self.default_target} nicht beigetreten") return f"Angemeldet als {actual}, Raum beigetreten" async def aclose(self) -> None: await self._client.aclose()