tg_names v0.1.0 [base]
Кэш имён пользователей Telegram для редактора прав (подписчик события сообщения).
| version | date | commit | файлов |
|---|---|---|---|
| 0.1.0 | 2026-09-17 | 91610a9fc155 | 5 |
README
# tg_names
Кэш имён пользователей Telegram для редактора прав.
Тип: Модуль. Категория: `base`. Зависимостей нет.
## Описание
Тихий помощник без команд. Модуль слушает шину сообщений и запоминает, как зовут каждого писавшего: имя пользователя и отображаемое имя. Кэш ограничен по размеру, старые записи вытесняются новыми.
Благодаря ему редактор прав и другие экраны показывают людей по-человечески — именем или ником, а не голым числовым id. Модуль ничего не требует и ни во что не вмешивается.
## Возможности
- `resolve {uid: str}` → отображаемое имя из кеша (`@username` или имя)
## События шины
- Подписано: `transport_telegram.message`
## Конфигурация (`config.schema.json`)
- Опциональные: `MAX_ENTRIES` = `5000`
Манифест
{
"name": "tg_names",
"version": "0.1.0",
"category": "base",
"description": "Кэш имён пользователей Telegram для редактора прав (подписчик события сообщения).",
"requires": [],
"rights": [],
"api_methods": [],
"web_routes": [],
"commit": "91610a9fc1556c139f02139ef0fd9a4504251dcf",
"updated": "2026-09-17 12:26:56 +0300",
"integrity": "sha256:a9f0ca47224f8ed712f6e70086ea2609b0df07f1c25e9b709453636582e1baf6",
"files": "[5 файлов — см. вкладки ниже]",
"note": "",
"wdesc": "",
"wbody": "",
"_extra": {
"api": "",
"events_emitted": "[]",
"events_subscribed": "",
"config_schema": "config.schema.json",
"env_example": ".env.example"
}
} Файлы и исходники
Дерево файлов
- · корень
- 1.2 КБ
- 0.5 КБ
- 0.1 КБ
- 0.6 КБ
- 2.1 КБ
Предпросмотр
Выберите файл в дереве выше — код откроется здесь.
Все исходники (.py) одним списком
api.py 0.5 КБ
"""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)
module.py 2.1 КБ
"""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", {}))}