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>
57 lines
1.8 KiB
Python
57 lines
1.8 KiB
Python
"""Entry point: settings → texts → dispatcher → long polling (ADR-002)."""
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import logging
|
|
|
|
from aiogram import Bot, Dispatcher
|
|
|
|
from app import __version__
|
|
from app.bot.fsm import make_storage
|
|
from app.bot.middlewares.access import AccessMiddleware
|
|
from app.bot.routers.common import build_router
|
|
from app.config import Settings
|
|
from app.health import heartbeat
|
|
from app.logs_setup import setup_logging
|
|
from app.texts import Texts, load_texts
|
|
|
|
log = logging.getLogger("tgit")
|
|
|
|
|
|
def build_dispatcher(settings: Settings, texts: Texts) -> Dispatcher:
|
|
admin_ids = settings.access.admin_ids
|
|
|
|
async def is_member(tg_id: int) -> bool:
|
|
return tg_id in admin_ids
|
|
|
|
dp = Dispatcher(storage=make_storage())
|
|
access = AccessMiddleware(is_member, texts)
|
|
# Outer middleware: runs before filters, so nothing — not even a stale button — reaches a handler unchecked.
|
|
dp.message.outer_middleware(access)
|
|
dp.callback_query.outer_middleware(access)
|
|
dp.include_router(build_router(texts, admin_ids))
|
|
return dp
|
|
|
|
|
|
async def run(settings: Settings) -> None:
|
|
texts = load_texts(settings.runtime.config_dir)
|
|
dp = build_dispatcher(settings, texts)
|
|
rt = settings.runtime
|
|
beat = asyncio.create_task(heartbeat(rt.heartbeat_file, rt.heartbeat_interval))
|
|
bot = Bot(settings.telegram.bot_token.get_secret_value())
|
|
log.info("tgit %s starting, %d bootstrap admin(s)", __version__, len(settings.access.admin_ids))
|
|
try:
|
|
await dp.start_polling(bot, handle_signals=True)
|
|
finally:
|
|
beat.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await beat
|
|
await bot.session.close()
|
|
log.info("tgit stopped")
|
|
|
|
|
|
def main() -> None:
|
|
settings = Settings()
|
|
setup_logging(settings.runtime.log_level)
|
|
asyncio.run(run(settings))
|