ElebobaandClaude Opus 5.5
0ce5285da1
- config: pydantic-settings по подсистемам, ADMIN_IDS через запятую - texts.yaml + валидация при старте (лишний/пропущенный ключ — ошибка сразу) - меню: reply-клавиатура разделов (Участники — только админам), инлайн «Назад»/«Отмена» с редактированием сообщения на месте, /start, /menu, /help - AccessMiddleware (outer) на сообщениях и кнопках: до #25 пускает только ADMIN_IDS, постороннему — одна фраза в сутки, устаревшие кнопки молча гасятся - heartbeat-файл + HEALTHCHECK в Dockerfile; JSON-логи - FSM пока в памяти (фабрика), хранилище в БД — в #5 - тесты на фейковой сессии aiogram через настоящий Dispatcher Refs #4 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
"""Access gate on every update — messages and button presses alike (SPEC A-5).
|
|
|
|
Until invites land (#25) the only members are the bootstrap admins from `ADMIN_IDS`.
|
|
"""
|
|
|
|
import time
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
|
|
from aiogram import BaseMiddleware
|
|
from aiogram.types import CallbackQuery, Message, TelegramObject
|
|
|
|
from app.texts import Texts
|
|
|
|
IsMember = Callable[[int], Awaitable[bool]]
|
|
|
|
NO_ACCESS_REPLY_EVERY = 24 * 3600.0
|
|
|
|
|
|
class AccessMiddleware(BaseMiddleware):
|
|
def __init__(self, is_member: IsMember, texts: Texts, clock: Callable[[], float] = time.monotonic) -> None:
|
|
self._is_member = is_member
|
|
self._texts = texts
|
|
self._clock = clock
|
|
# tg_id -> when we last told this outsider "invite only"; answering every message would let them spam us.
|
|
self._last_refusal: dict[int, float] = {}
|
|
|
|
async def __call__(
|
|
self,
|
|
handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
|
|
event: TelegramObject,
|
|
data: dict[str, Any],
|
|
) -> Any:
|
|
user = data.get("event_from_user")
|
|
if user is not None and await self._is_member(user.id):
|
|
return await handler(event, data)
|
|
if user is not None:
|
|
await self._refuse(event, user.id)
|
|
return None
|
|
|
|
async def _refuse(self, event: TelegramObject, tg_id: int) -> None:
|
|
if isinstance(event, CallbackQuery):
|
|
await event.answer() # stop the spinner on a stale button, say nothing
|
|
return
|
|
now = self._clock()
|
|
last = self._last_refusal.get(tg_id)
|
|
if last is not None and now - last < NO_ACCESS_REPLY_EVERY:
|
|
return
|
|
self._last_refusal[tg_id] = now
|
|
if isinstance(event, Message):
|
|
await event.answer(self._texts.start.no_access)
|