messageLogger/src/api.py

"""telegram-panel client API (`/api/v1/client/message-logger`).

Each token reads one master's logs (the account owning the logger groups);
the plugin holds one per master. A chat's state merges every master's answer;
a single message is looked up master by master.

Calls are blocking: run them off the UI thread (`jutil.in_background`).
"""

import re
import time

import diag
from config import API_TOKEN, HTTP_TIMEOUT_SECONDS, SERVER_URL

# A token the server refuses (403: unbound, no permission) is skipped this long.
_REFUSED_SECONDS = 600


class ApiError(Exception):
    pass


class Api:
    def __init__(self, plugin):
        self.plugin = plugin
        self._refused = {}  # token -> monotonic time to retry at

    @property
    def configured(self):
        return bool(self._base() and self.tokens())

    def _base(self):
        url = str(self.plugin.get_setting(SERVER_URL, "") or "").strip().rstrip("/")
        return url + "/api/v1/client/message-logger" if url else ""

    def tokens(self):
        raw = str(self.plugin.get_setting(API_TOKEN, "") or "")
        return [t for t in re.split(r"[\s,;]+", raw) if t]

    def _call(self, token, method, path, **kwargs):
        import requests

        if not self._base():
            raise ApiError("Server URL is not set")
        try:
            response = requests.request(
                method,
                self._base() + path,
                headers={"Authorization": "Bearer " + token},
                timeout=HTTP_TIMEOUT_SECONDS,
                **kwargs,
            )
        except requests.RequestException as exc:
            raise ApiError(f"Server unreachable: {exc}") from exc
        if response.status_code == 404:
            try:
                detail = response.json().get("detail")
            except Exception:
                detail = None
            if detail == "Not Found":  # the route itself: an older server
                raise ApiError("the server doesn't support this yet (update telegram-panel)")
            return None
        if response.status_code >= 400:
            try:
                detail = response.json().get("detail")
            except Exception:
                detail = response.text[:200]
            raise ApiError(f"{response.status_code}: {detail}")
        return response.json()

    def _each(self, method, path, **kwargs):
        """(token, result) for every usable token; a refused one is skipped."""
        for token in self.tokens():
            if time.monotonic() < self._refused.get(token, 0):
                continue
            try:
                yield token, self._call(token, method, path, **kwargs)
            except ApiError as exc:
                if str(exc).startswith(("401", "403")):
                    diag.log(f"token {token[:10]}… refused: {exc}")
                    self._refused[token] = time.monotonic() + _REFUSED_SECONDS
                    continue
                raise

    def _first(self, method, path, **kwargs):
        for _token, result in self._each(method, path, **kwargs):
            if result is not None:
                return result
        return None

    def me(self):
        """[(token prefix, /me or the error)] for every token."""
        out = []
        for token in self.tokens():
            try:
                out.append((token[:10], self._call(token, "GET", "/me")))
            except ApiError as exc:
                out.append((token[:10], exc))
        return out

    def masters(self):
        """Telegram ids of the tokens' masters (from /me), cached 10 minutes."""
        cached = getattr(self, "_masters", None)
        if cached is not None and time.monotonic() - cached[0] < 600:
            return cached[1]
        found = []
        for _prefix, me in self.me():
            if isinstance(me, dict) and me.get("master_tg_id"):
                found.append(int(me["master_tg_id"]))
        self._masters = (time.monotonic(), found)
        return found

    def sync(self, account_id, chat_id, min_id, max_id, present_ids):
        body = {
            "account_id": account_id,
            "chat_id": chat_id,
            "min_id": min_id,
            "max_id": max_id,
            "present_ids": list(present_ids)[:1000],
        }
        states, missing, any_result = {}, set(), False
        for _token, result in self._each("POST", "/sync", json=body):
            if result is None:
                continue
            any_result = True
            for state in result.get("states") or []:
                states.setdefault(int(state["message_id"]), state)
            missing.update(int(i) for i in result.get("missing") or [])
        if not any_result:
            return None
        return {"states": list(states.values()), "missing": sorted(missing)}

    def message(self, account_id, chat_id, message_id):
        return self._first(
            "GET",
            f"/messages/{chat_id}/{message_id}",
            params={"account_id": account_id},
        )

    def versions(self, account_id, chat_id, message_id):
        return self._first(
            "GET",
            f"/messages/{chat_id}/{message_id}/versions",
            params={"account_id": account_id},
        )

    def comments(self, account_id, chat_id, message_id):
        return self._first(
            "GET",
            f"/messages/{chat_id}/{message_id}/comments",
            params={"account_id": account_id},
        )

    def deletions(self, account_id, chat_id, message_ids):
        ids = sorted(set(int(i) for i in message_ids))
        marked = 0
        for start in range(0, len(ids), 500):
            body = {
                "account_id": account_id,
                "chat_id": chat_id,
                "message_ids": ids[start : start + 500],
            }
            for _token, result in self._each("POST", "/deletions", json=body):
                marked += int((result or {}).get("marked", 0))
        return marked

    def hidden(self, account_id, chat_id, message_ids, hidden):
        """Hide messages from the chat for every client (or show them again);
        True once every master took it."""
        ids = sorted(set(int(i) for i in message_ids))
        for start in range(0, len(ids), 500):
            body = {
                "account_id": account_id,
                "chat_id": chat_id,
                "message_ids": ids[start : start + 500],
                "hidden": bool(hidden),
            }
            for _token, _result in self._each("POST", "/hidden", json=body):
                pass
        return True

    # removed chats

    def chat_removed(self, account_id, chat_id, own):
        body = {"account_id": account_id, "chat_id": chat_id, "own": bool(own)}
        known = False
        for _token, result in self._each("POST", "/chats/removed", json=body):
            known = known or bool((result or {}).get("known"))
        return known

    def chat_hidden(self, account_id, chat_id, hidden):
        body = {"account_id": account_id, "chat_id": chat_id, "hidden": bool(hidden)}
        for _token, _result in self._each("POST", "/chats/hidden", json=body):
            pass

    def removed_chats(self, account_id):
        """The account's removed chats, each master's answer merged (the
        first one listing a chat wins), or None when no master answered."""
        chats, any_result = {}, False
        for _token, result in self._each(
            "GET", "/chats/removed", params={"account_id": account_id}
        ):
            if result is None:
                continue
            any_result = True
            for chat in result.get("chats") or []:
                chat["master_tg_id"] = result.get("master_tg_id")
                chats.setdefault(int(chat["chat_id"]), chat)
        return list(chats.values()) if any_result else None

    def history(self, account_id, chat_id, before=0, limit=100):
        """Logged messages of a chat older than `before`, newest first: the
        first master that has any."""
        params = {"account_id": account_id, "before": int(before), "limit": int(limit)}
        for _token, result in self._each(
            "GET", f"/chats/{chat_id}/history", params=params
        ):
            if result and result.get("messages"):
                return result
        return None

    def avatars(self, account_id, chat_id):
        """Chat photos logged for a chat, newest first, every master's."""
        out = []
        for _token, result in self._each(
            "GET", f"/chats/{chat_id}/avatars", params={"account_id": account_id}
        ):
            for avatar in (result or {}).get("avatars") or []:
                avatar["master_tg_id"] = (result or {}).get("master_tg_id")
                out.append(avatar)
        out.sort(key=lambda a: a.get("date") or "", reverse=True)
        return out