ElebobaandClaude Opus 5.5
55286495cb
CI / ruff · mypy · pytest (pull_request) Successful in 1m11s
- docker-socket-proxy в Compose (только GET /containers/*), сокет только у него - app/runtime/docker.py: read-only клиент; видны только контейнеры с меткой tgit.managed=true, чужие id отклоняются до запроса - app/runtime/checks.py: running, health, рестарты за окно, CPU/RAM, ошибки в логе, HTTP-проба; у каждой проверки свой таймаут, отчёт всегда полный - app/runtime/snapshots.py: снапшоты раз в 15 с в container_snapshots - бот: список с 🟢🟡🔴⚪, карточка, «🩺 Запустить проверки», «🔄 Обновить» - деплой ставит метки tgit.sha / tgit.deployed_at на контейнер бота - имя джоба CD без неотрисованного выражения Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
66 lines
3.0 KiB
Python
66 lines
3.0 KiB
Python
"""/start (incl. invite deep links), /help, /menu, main menu buttons and in-section navigation."""
|
|
|
|
from aiogram import F, Router
|
|
from aiogram.filters import Command, CommandObject, CommandStart
|
|
from aiogram.fsm.context import FSMContext
|
|
from aiogram.types import CallbackQuery, Message
|
|
|
|
from app.access.service import AccessService, RedeemStatus
|
|
from app.bot.menu import Nav, NavAction, Section, main_menu, section_by_button, section_nav
|
|
from app.bot.middlewares.access import tg_user
|
|
from app.db.models import Role, User
|
|
from app.texts import Texts
|
|
|
|
|
|
def build_router(texts: Texts, access: AccessService) -> Router:
|
|
router = Router(name="common")
|
|
sections = section_by_button(texts)
|
|
section_text = {
|
|
Section.LOGS: texts.section.logs,
|
|
Section.ALERTS: texts.section.alerts,
|
|
}
|
|
|
|
@router.message(CommandStart())
|
|
async def start(message: Message, command: CommandObject, state: FSMContext, member: User | None) -> None:
|
|
if member is None:
|
|
# Only an outsider with an invite token gets here (AccessMiddleware); a member's /start <x> is a plain start
|
|
if message.from_user is None or not command.args:
|
|
return
|
|
status = await access.redeem(command.args, tg_user(message.from_user))
|
|
if status is RedeemStatus.THROTTLED:
|
|
return # brute force: stay silent (SPEC A-6)
|
|
if status is RedeemStatus.REJECTED:
|
|
await message.answer(texts.start.invite_rejected)
|
|
return
|
|
member = await access.get_member(message.from_user.id)
|
|
if member is None:
|
|
return
|
|
await state.clear()
|
|
await message.answer(texts.start.welcome, reply_markup=main_menu(texts, is_admin=member.role is Role.ADMIN))
|
|
|
|
@router.message(Command("menu"))
|
|
async def menu(message: Message, state: FSMContext, member: User) -> None:
|
|
await state.clear()
|
|
await message.answer(texts.menu.placeholder, reply_markup=main_menu(texts, is_admin=member.role is Role.ADMIN))
|
|
|
|
@router.message(Command("help"))
|
|
async def help_(message: Message) -> None:
|
|
await message.answer(texts.help.text)
|
|
|
|
# Sections with their own routers (members, commit) are matched there first
|
|
@router.message(F.text.in_({k for k, v in sections.items() if v in section_text}))
|
|
async def open_section(message: Message, state: FSMContext) -> None:
|
|
await state.clear()
|
|
await message.answer(section_text[sections[message.text or ""]], reply_markup=section_nav(texts))
|
|
|
|
@router.callback_query(Nav.filter())
|
|
async def navigate(query: CallbackQuery, callback_data: Nav, state: FSMContext) -> None:
|
|
# At a section's root "Back" and "Cancel" both close it; nested screens will override "Back" per section.
|
|
await state.clear()
|
|
text = texts.section.cancelled if callback_data.action is NavAction.CANCEL else texts.menu.placeholder
|
|
if isinstance(query.message, Message):
|
|
await query.message.edit_text(text)
|
|
await query.answer()
|
|
|
|
return router
|