tg_names v0.1.0 [base]

"""tg_names.api — resolve/display без привязки к Telegram-объектам."""


def display(rec: dict) -> str:
    un = (rec.get("username") or "").lstrip("@")
    if un:
        return "@" + un
    return " ".join(
        x for x in (rec.get("first_name"), rec.get("last_name")) if x
    ) or rec.get("name") or ""


def resolve(cache_ns, uid) -> str:
    rec = (cache_ns.get("names", {}).get(str(uid)) or {})
    return rec.get("name") or display(rec)

"""tg_names.Module — запоминает имена из события transport_telegram.message.

Свой неймспейс кеша data/cache/tg_names/names. Команд нет (модуль/reactive),
наружу — api tg_names.resolve(uid) -> отображаемое имя.
"""
import time

from core.base_module import BaseModule

from . import api as names_api


class Module(BaseModule):
    def setup(self, ctx) -> None:
        self.config = ctx.module_config("tg_names")
        self.cache = ctx.cache.ns("tg_names")
        try:
            self.max_entries = int(self.config.get("MAX_ENTRIES", "5000") or 5000)
        except ValueError:
            from core.errors import ConfigError
            raise ConfigError("tg_names: MAX_ENTRIES не число")
        ctx.api.register("tg_names", "resolve", lambda uid: names_api.resolve(self.cache, uid))
        ctx.events.subscribe("transport_telegram.message", self._on_message)

    def _on_message(self, payload):
        try:
            uid = str((payload or {}).get("user_id") or "")
            if not uid:
                return None
            rec = {
                "username": (payload.get("username") or "") or None,
                "first_name": (payload.get("first_name") or "") or None,
                "last_name": (payload.get("last_name") or "") or None,
                "updated": time.time(),
            }
        except Exception:  # noqa: BLE001
            return None
        data = self.cache.get("names", {})
        prev = data.get(uid) or {}
        if any(prev.get(k) != rec[k] for k in ("username", "first_name", "last_name")):
            rec["name"] = names_api.display(rec)
            data[uid] = rec
            if len(data) > self.max_entries:
                for k in sorted(data, key=lambda x: data[x].get("updated", 0))[:len(data) - self.max_entries]:
                    del data[k]
            self.cache.set("names", data)
        return rec.get("name") or prev.get("name")

    def health(self) -> dict:
        return {"ok": True, "module": self.name,
                "names": len(self.cache.get("names", {}))}