Verum/GlitchupBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

тут всякие фиксы
Some checks failed
CI / Lint (ruff + mypy) (push) Failing after 39s
Details
CI / Run tests (push) Has been skipped
Details
CI / Docker build test (push) Successful in 17s
Details

Browse Source
This commit is contained in:
Verum
2026-03-30 18:04:49 +07:00
parent 3f176902a2
commit bdae79db58
6 changed files with 584 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
migrations/versions/20260330_0003_bot_admins.py Normal file
View File

@@ -0,0 +1,37 @@
"""bot admins
Revision ID: 20260330_0003
Revises: 20260327_0002
Create Date: 2026-03-30 11:30:00.000000
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "20260330_0003"
down_revision: str | None = "20260327_0002"
branch_labels: Sequence[str] | None = None
depends_on: Sequence[str] | None = None
def upgrade() -> None:
op.create_table(
"bot_admins",
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column("user_id", sa.BigInteger(), nullable=False),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
server_default=sa.text("now()"),
nullable=False,
),
)
op.create_index(op.f("ix_bot_admins_user_id"), "bot_admins", ["user_id"], unique=True)
def downgrade() -> None:
op.drop_index(op.f("ix_bot_admins_user_id"), table_name="bot_admins")
op.drop_table("bot_admins")

480
src/glitchup_bot/bot/handlers/commands.py
View File

@@ -1,6 +1,9 @@
import logging
from collections import OrderedDict
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from html import escape
from uuid import uuid4
from aiogram import F, Router
from aiogram.exceptions import TelegramBadRequest
@@ -18,7 +21,12 @@ from glitchup_bot.bot.keyboards import (
help_result_keyboard,
help_subscriptions_keyboard,
)
from glitchup_bot.config import settings
from glitchup_bot.services.admins import (
add_admin,
is_admin,
list_effective_admins,
remove_admin,
)
from glitchup_bot.services.digest_builder import (
build_digest,
build_project_summary,
@@ -47,6 +55,20 @@ from glitchup_bot.services.routing import (
router = Router()
logger = logging.getLogger(__name__)
MAX_PAGE_CHARS = 3000
MAX_PAGE_LINES = 18
MAX_PAGINATION_SESSIONS = 200
@dataclass(slots=True)
class PaginationSession:
pages: list[str]
back_callback: str
is_admin: bool
admin_mode: bool
PAGINATION_SESSIONS: OrderedDict[str, PaginationSession] = OrderedDict()
def _sender_id(message: Message) -> int | None:
@@ -57,8 +79,72 @@ def _callback_sender_id(callback: CallbackQuery) -> int | None:
return callback.from_user.id if callback.from_user else None
def _is_admin_user(user_id: int | None) -> bool:
return settings.is_admin(user_id)
async def _is_admin_user(user_id: int | None) -> bool:
return await is_admin(user_id)
def _paginate_text(text: str) -> list[str]:
if len(text) <= MAX_PAGE_CHARS and text.count("\n") + 1 <= MAX_PAGE_LINES:
return [text]
pages: list[str] = []
current_lines: list[str] = []
current_len = 0
for line in text.splitlines():
projected_len = current_len + len(line) + (1 if current_lines else 0)
if current_lines and (
projected_len > MAX_PAGE_CHARS or len(current_lines) >= MAX_PAGE_LINES
):
pages.append("\n".join(current_lines))
current_lines = [line]
current_len = len(line)
continue
current_lines.append(line)
current_len = projected_len
if current_lines:
pages.append("\n".join(current_lines))
return pages or [text]
def _store_pagination_session(session: PaginationSession) -> str:
token = uuid4().hex[:12]
PAGINATION_SESSIONS[token] = session
PAGINATION_SESSIONS.move_to_end(token)
while len(PAGINATION_SESSIONS) > MAX_PAGINATION_SESSIONS:
PAGINATION_SESSIONS.popitem(last=False)
return token
def _result_keyboard(
*,
back_callback: str,
is_admin: bool,
admin_mode: bool,
page_token: str | None = None,
page_index: int = 0,
total_pages: int = 1,
):
if admin_mode:
return admin_result_keyboard(
back_callback,
page_token=page_token,
page_index=page_index,
total_pages=total_pages,
)
return help_result_keyboard(
back_callback,
is_admin,
page_token=page_token,
page_index=page_index,
total_pages=total_pages,
)
def _help_text(is_admin: bool) -> str:
@@ -89,6 +175,9 @@ def _help_text(is_admin: bool) -> str:
"<b>Для админа:</b>",
"• /admin — панель управления",
"• /sync — принудительная синхронизация",
"• /admins — список администраторов",
"• /admin_add &lt;user_id&gt; или reply на сообщение",
"• /admin_del &lt;user_id&gt;",
"• /ownership — текущее распределение routing-настроек",
"• /mute_list — список mute rules",
]
@@ -163,6 +252,7 @@ def _admin_text() -> str:
"",
"• <b>Центр синхронизации</b> — запуск sync и проверка статуса",
"• <b>Сводки и мониторинг</b> — основные обзорные экраны",
"• <b>Администраторы</b> — список и управление доступом",
"• <b>Ownership и topics</b> — текущая маршрутизация",
"• <b>Mute rules</b> — правила скрытия шумных событий",
]
@@ -198,6 +288,10 @@ def _admin_guide_text() -> str:
"Через кнопки удобно смотреть состояние системы, "
"а точечные настройки меняются командами.",
"",
"• <code>/admin_add 123456</code> — добавить администратора",
"• <code>/admin_add</code> ответом на сообщение — добавить автора сообщения",
"• <code>/admin_del 123456</code> — удалить runtime-администратора",
"",
"• <code>/owner slug backend</code> — переназначить проект в группу",
"• <code>/topic backend 123</code> — сменить topic override",
"• <code>/mute_add payment.*timeout</code> — добавить mute rule",
@@ -218,8 +312,40 @@ def _sync_summary_text(summary) -> str:
)
async def _admins_text() -> str:
admins = await list_effective_admins()
if not admins:
return "Список администраторов пуст."
lines = [
"<b>Администраторы</b>",
"",
"Добавлять можно командой <code>/admin_add &lt;user_id&gt;</code> "
"или ответом на сообщение пользователя.",
"",
]
for user_id, source in admins:
source_label = source.replace("env", "из .env")
lines.append(f"• <code>{user_id}</code> — {source_label}")
return "\n".join(lines)
def _extract_target_user_id(message: Message) -> int | None:
args = message.text.split(maxsplit=1) if message.text else []
if len(args) >= 2:
raw_value = args[1].strip()
return int(raw_value) if raw_value.lstrip("-").isdigit() else None
reply = message.reply_to_message
if reply and reply.from_user:
return reply.from_user.id
return None
async def _require_admin(message: Message) -> bool:
if _is_admin_user(_sender_id(message)):
if await _is_admin_user(_sender_id(message)):
return True
await message.answer("Команда доступна только администраторам.")
@@ -227,7 +353,7 @@ async def _require_admin(message: Message) -> bool:
async def _require_admin_callback(callback: CallbackQuery) -> bool:
if _is_admin_user(_callback_sender_id(callback)):
if await _is_admin_user(_callback_sender_id(callback)):
return True
await callback.answer("Только для администраторов", show_alert=True)
@@ -305,6 +431,49 @@ async def _deliver_text(
)
async def _deliver_result(
target: Message | CallbackQuery,
text: str,
*,
back_callback: str,
is_admin: bool,
admin_mode: bool = False,
) -> None:
pages = _paginate_text(text)
if len(pages) == 1:
await _deliver_text(
target,
pages[0],
reply_markup=_result_keyboard(
back_callback=back_callback,
is_admin=is_admin,
admin_mode=admin_mode,
),
)
return
token = _store_pagination_session(
PaginationSession(
pages=pages,
back_callback=back_callback,
is_admin=is_admin,
admin_mode=admin_mode,
)
)
await _deliver_text(
target,
pages[0],
reply_markup=_result_keyboard(
back_callback=back_callback,
is_admin=is_admin,
admin_mode=admin_mode,
page_token=token,
page_index=0,
total_pages=len(pages),
),
)
async def _handle_subscription_action(
target: Message | CallbackQuery,
group_name: str,
@@ -346,9 +515,17 @@ async def _run_summary_action(
target: Message | CallbackQuery,
loader: Callable[[], Awaitable[str]],
*,
reply_markup=None,
back_callback: str,
is_admin: bool,
admin_mode: bool = False,
) -> None:
await _deliver_text(target, await loader(), reply_markup=reply_markup)
await _deliver_result(
target,
await loader(),
back_callback=back_callback,
is_admin=is_admin,
admin_mode=admin_mode,
)
async def _ownership_text() -> str:
@@ -415,7 +592,7 @@ async def _mute_rules_text() -> str:
@router.message(Command("start"))
async def cmd_start(message: Message) -> None:
is_admin = _is_admin_user(_sender_id(message))
is_admin = await _is_admin_user(_sender_id(message))
await message.answer(
_help_text(is_admin),
reply_markup=help_home_keyboard(is_admin),
@@ -425,7 +602,7 @@ async def cmd_start(message: Message) -> None:
@router.message(Command("help"))
async def cmd_help(message: Message) -> None:
is_admin = _is_admin_user(_sender_id(message))
is_admin = await _is_admin_user(_sender_id(message))
await message.answer(
_help_text(is_admin),
reply_markup=help_home_keyboard(is_admin),
@@ -449,7 +626,7 @@ async def cmd_admin(message: Message) -> None:
async def cb_help_actions(callback: CallbackQuery) -> None:
data = callback.data or ""
action = data.removeprefix("help:")
is_admin = _is_admin_user(_callback_sender_id(callback))
is_admin = await _is_admin_user(_callback_sender_id(callback))
await callback.answer()
if action == "open":
@@ -491,42 +668,48 @@ async def cb_help_actions(callback: CallbackQuery) -> None:
await _run_summary_action(
callback,
lambda: build_digest(refresh=True),
reply_markup=help_result_keyboard("help:menu:monitoring", is_admin),
back_callback="help:menu:monitoring",
is_admin=is_admin,
)
return
if action == "today":
await _run_summary_action(
callback,
lambda: build_today_summary(refresh=True),
reply_markup=help_result_keyboard("help:menu:monitoring", is_admin),
back_callback="help:menu:monitoring",
is_admin=is_admin,
)
return
if action == "top":
await _run_summary_action(
callback,
lambda: build_top_issues(refresh=True),
reply_markup=help_result_keyboard("help:menu:monitoring", is_admin),
back_callback="help:menu:monitoring",
is_admin=is_admin,
)
return
if action == "stale":
await _run_summary_action(
callback,
lambda: build_stale_issues(refresh=True),
reply_markup=help_result_keyboard("help:menu:monitoring", is_admin),
back_callback="help:menu:monitoring",
is_admin=is_admin,
)
return
if action == "releases":
await _run_summary_action(
callback,
lambda: build_release_summary(refresh=True),
reply_markup=help_result_keyboard("help:menu:releases", is_admin),
back_callback="help:menu:releases",
is_admin=is_admin,
)
return
if action == "sync_status":
await _run_summary_action(
callback,
build_sync_status,
reply_markup=help_result_keyboard("help:open", is_admin),
back_callback="help:open",
is_admin=is_admin,
)
return
if action.startswith("sub:"):
@@ -584,96 +767,169 @@ async def cb_admin_actions(callback: CallbackQuery) -> None:
reply_markup=admin_overview_keyboard(),
)
return
if action == "admins":
await _deliver_result(
callback,
await _admins_text(),
back_callback="admin:open",
is_admin=True,
admin_mode=True,
)
return
if action == "sync":
summary = await run_manual_sync()
await _show_callback_screen(
await _deliver_result(
callback,
_sync_summary_text(summary),
reply_markup=admin_result_keyboard("admin:menu:sync"),
back_callback="admin:menu:sync",
is_admin=True,
admin_mode=True,
)
return
if action == "sync_status":
await _run_summary_action(
callback,
build_sync_status,
reply_markup=admin_result_keyboard("admin:menu:sync"),
back_callback="admin:menu:sync",
is_admin=True,
admin_mode=True,
)
return
if action == "ownership":
await _show_callback_screen(
await _deliver_result(
callback,
await _ownership_text(),
reply_markup=admin_result_keyboard("admin:open"),
back_callback="admin:open",
is_admin=True,
admin_mode=True,
)
return
if action == "mute_list":
await _show_callback_screen(
await _deliver_result(
callback,
await _mute_rules_text(),
reply_markup=admin_result_keyboard("admin:open"),
back_callback="admin:open",
is_admin=True,
admin_mode=True,
)
return
if action == "releases":
await _run_summary_action(
callback,
lambda: build_release_summary(refresh=True),
reply_markup=admin_result_keyboard("admin:menu:overview"),
back_callback="admin:menu:overview",
is_admin=True,
admin_mode=True,
)
return
if action == "today":
await _run_summary_action(
callback,
lambda: build_today_summary(refresh=True),
reply_markup=admin_result_keyboard("admin:menu:overview"),
back_callback="admin:menu:overview",
is_admin=True,
admin_mode=True,
)
return
if action == "week":
await _run_summary_action(
callback,
lambda: build_digest(refresh=True),
reply_markup=admin_result_keyboard("admin:menu:overview"),
back_callback="admin:menu:overview",
is_admin=True,
admin_mode=True,
)
return
if action == "top":
await _run_summary_action(
callback,
lambda: build_top_issues(refresh=True),
reply_markup=admin_result_keyboard("admin:menu:overview"),
back_callback="admin:menu:overview",
is_admin=True,
admin_mode=True,
)
return
if action == "stale":
await _run_summary_action(
callback,
lambda: build_stale_issues(refresh=True),
reply_markup=admin_result_keyboard("admin:menu:overview"),
back_callback="admin:menu:overview",
is_admin=True,
admin_mode=True,
)
return
if action == "guide":
await _show_callback_screen(
await _deliver_result(
callback,
_admin_guide_text(),
reply_markup=admin_result_keyboard("admin:open"),
back_callback="admin:open",
is_admin=True,
admin_mode=True,
)
return
@router.callback_query(F.data.startswith("page:"))
async def cb_paginated_results(callback: CallbackQuery) -> None:
payload = (callback.data or "").split(":")
if len(payload) != 3:
await callback.answer()
return
_, token, raw_index = payload
session = PAGINATION_SESSIONS.get(token)
if session is None:
await callback.answer("Эта навигация устарела. Открой раздел заново.", show_alert=True)
return
if session.admin_mode and not await _is_admin_user(_callback_sender_id(callback)):
await callback.answer("Только для администраторов", show_alert=True)
return
if not raw_index.isdigit():
await callback.answer()
return
page_index = int(raw_index)
if page_index < 0 or page_index >= len(session.pages):
await callback.answer()
return
await callback.answer()
PAGINATION_SESSIONS.move_to_end(token)
await _show_callback_screen(
callback,
session.pages[page_index],
reply_markup=_result_keyboard(
back_callback=session.back_callback,
is_admin=session.is_admin,
admin_mode=session.admin_mode,
page_token=token,
page_index=page_index,
total_pages=len(session.pages),
),
)
@router.message(Command("week"))
async def cmd_week(message: Message) -> None:
is_admin = _is_admin_user(_sender_id(message))
await message.answer(
is_admin = await _is_admin_user(_sender_id(message))
await _deliver_result(
message,
await build_digest(refresh=True),
reply_markup=help_result_keyboard("help:menu:monitoring", is_admin),
disable_web_page_preview=True,
back_callback="help:menu:monitoring",
is_admin=is_admin,
)
@router.message(Command("today"))
async def cmd_today(message: Message) -> None:
is_admin = _is_admin_user(_sender_id(message))
await message.answer(
is_admin = await _is_admin_user(_sender_id(message))
await _deliver_result(
message,
await build_today_summary(refresh=True),
reply_markup=help_result_keyboard("help:menu:monitoring", is_admin),
disable_web_page_preview=True,
back_callback="help:menu:monitoring",
is_admin=is_admin,
)
@@ -684,41 +940,45 @@ async def cmd_project(message: Message) -> None:
await message.answer("Использование: /project &lt;slug&gt;")
return
is_admin = _is_admin_user(_sender_id(message))
await message.answer(
is_admin = await _is_admin_user(_sender_id(message))
await _deliver_result(
message,
await build_project_summary(args[1].strip(), refresh=True),
reply_markup=help_result_keyboard("help:menu:monitoring", is_admin),
disable_web_page_preview=True,
back_callback="help:menu:monitoring",
is_admin=is_admin,
)
@router.message(Command("top"))
async def cmd_top(message: Message) -> None:
is_admin = _is_admin_user(_sender_id(message))
await message.answer(
is_admin = await _is_admin_user(_sender_id(message))
await _deliver_result(
message,
await build_top_issues(refresh=True),
reply_markup=help_result_keyboard("help:menu:monitoring", is_admin),
disable_web_page_preview=True,
back_callback="help:menu:monitoring",
is_admin=is_admin,
)
@router.message(Command("stale"))
async def cmd_stale(message: Message) -> None:
is_admin = _is_admin_user(_sender_id(message))
await message.answer(
is_admin = await _is_admin_user(_sender_id(message))
await _deliver_result(
message,
await build_stale_issues(refresh=True),
reply_markup=help_result_keyboard("help:menu:monitoring", is_admin),
disable_web_page_preview=True,
back_callback="help:menu:monitoring",
is_admin=is_admin,
)
@router.message(Command("releases"))
async def cmd_releases(message: Message) -> None:
is_admin = _is_admin_user(_sender_id(message))
await message.answer(
is_admin = await _is_admin_user(_sender_id(message))
await _deliver_result(
message,
await build_release_summary(refresh=True),
reply_markup=help_result_keyboard("help:menu:releases", is_admin),
disable_web_page_preview=True,
back_callback="help:menu:releases",
is_admin=is_admin,
)
@@ -729,21 +989,23 @@ async def cmd_release(message: Message) -> None:
await message.answer("Использование: /release &lt;version&gt;")
return
is_admin = _is_admin_user(_sender_id(message))
await message.answer(
is_admin = await _is_admin_user(_sender_id(message))
await _deliver_result(
message,
await build_release_detail(args[1].strip(), refresh=True),
reply_markup=help_result_keyboard("help:menu:releases", is_admin),
disable_web_page_preview=True,
back_callback="help:menu:releases",
is_admin=is_admin,
)
@router.message(Command("sync_status"))
async def cmd_sync_status(message: Message) -> None:
is_admin = _is_admin_user(_sender_id(message))
await message.answer(
is_admin = await _is_admin_user(_sender_id(message))
await _deliver_result(
message,
await build_sync_status(),
reply_markup=help_result_keyboard("help:open", is_admin),
disable_web_page_preview=True,
back_callback="help:open",
is_admin=is_admin,
)
@@ -753,10 +1015,78 @@ async def cmd_sync(message: Message) -> None:
return
summary = await run_manual_sync()
await message.answer(
await _deliver_result(
message,
_sync_summary_text(summary),
reply_markup=admin_result_keyboard("admin:menu:sync"),
disable_web_page_preview=True,
back_callback="admin:menu:sync",
is_admin=True,
admin_mode=True,
)
@router.message(Command("admins"))
async def cmd_admins(message: Message) -> None:
if not await _require_admin(message):
return
await _deliver_result(
message,
await _admins_text(),
back_callback="admin:open",
is_admin=True,
admin_mode=True,
)
@router.message(Command("admin_add"))
async def cmd_admin_add(message: Message) -> None:
if not await _require_admin(message):
return
user_id = _extract_target_user_id(message)
if user_id is None:
await message.answer(
"Использование: /admin_add &lt;user_id&gt; или ответом на сообщение пользователя."
)
return
added = await add_admin(user_id)
text = (
f"Администратор <code>{user_id}</code> добавлен."
if added
else f"Пользователь <code>{user_id}</code> уже есть среди runtime-администраторов."
)
await _deliver_result(
message,
text,
back_callback="admin:open",
is_admin=True,
admin_mode=True,
)
@router.message(Command("admin_del"))
async def cmd_admin_del(message: Message) -> None:
if not await _require_admin(message):
return
user_id = _extract_target_user_id(message)
if user_id is None:
await message.answer("Использование: /admin_del &lt;user_id&gt;")
return
removed = await remove_admin(user_id)
text = (
f"Runtime-администратор <code>{user_id}</code> удалён."
if removed
else f"Runtime-администратор <code>{user_id}</code> не найден."
)
await _deliver_result(
message,
text,
back_callback="admin:open",
is_admin=True,
admin_mode=True,
)
@@ -767,7 +1097,7 @@ async def cmd_subscribe(message: Message) -> None:
await message.answer("Использование: /subscribe &lt;backend|frontend&gt;")
return
is_admin = _is_admin_user(_sender_id(message))
is_admin = await _is_admin_user(_sender_id(message))
await _handle_subscription_action(
message,
args[1].strip().lower(),
@@ -784,7 +1114,7 @@ async def cmd_unsubscribe(message: Message) -> None:
await message.answer("Использование: /unsubscribe &lt;backend|frontend&gt;")
return
is_admin = _is_admin_user(_sender_id(message))
is_admin = await _is_admin_user(_sender_id(message))
await _handle_subscription_action(
message,
args[1].strip().lower(),
@@ -799,10 +1129,12 @@ async def cmd_ownership(message: Message) -> None:
if not await _require_admin(message):
return
await message.answer(
await _deliver_result(
message,
await _ownership_text(),
reply_markup=admin_result_keyboard("admin:open"),
disable_web_page_preview=True,
back_callback="admin:open",
is_admin=True,
admin_mode=True,
)
@@ -897,10 +1229,12 @@ async def cmd_mute_list(message: Message) -> None:
if not await _require_admin(message):
return
await message.answer(
await _deliver_result(
message,
await _mute_rules_text(),
reply_markup=admin_result_keyboard("admin:open"),
disable_web_page_preview=True,
back_callback="admin:open",
is_admin=True,
admin_mode=True,
)

67
src/glitchup_bot/bot/keyboards.py
View File

@@ -2,6 +2,26 @@ from aiogram.types import InlineKeyboardMarkup
from aiogram.utils.keyboard import InlineKeyboardBuilder
def _add_pagination_row(
builder: InlineKeyboardBuilder,
*,
page_token: str | None,
page_index: int,
total_pages: int,
) -> None:
if not page_token or total_pages <= 1:
return
if page_index > 0:
builder.button(text="◀", callback_data=f"page:{page_token}:{page_index - 1}")
builder.button(
text=f"{page_index + 1}/{total_pages}",
callback_data=f"page:{page_token}:{page_index}",
)
if page_index + 1 < total_pages:
builder.button(text="▶", callback_data=f"page:{page_token}:{page_index + 1}")
def help_home_keyboard(is_admin: bool) -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
builder.button(text="Обзор и метрики", callback_data="help:menu:monitoring")
@@ -51,15 +71,34 @@ def help_subscriptions_keyboard(is_admin: bool) -> InlineKeyboardMarkup:
return builder.as_markup()
def help_result_keyboard(back_callback: str, is_admin: bool) -> InlineKeyboardMarkup:
def help_result_keyboard(
back_callback: str,
is_admin: bool,
*,
page_token: str | None = None,
page_index: int = 0,
total_pages: int = 1,
) -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
_add_pagination_row(
builder,
page_token=page_token,
page_index=page_index,
total_pages=total_pages,
)
builder.button(text="Назад", callback_data=back_callback)
builder.button(text="Главное меню", callback_data="help:open")
if is_admin:
builder.button(text="Админ-панель", callback_data="admin:open")
builder.adjust(2, 1)
if total_pages > 1:
builder.adjust(3, 2, 1)
else:
builder.adjust(2, 1)
else:
builder.adjust(2)
if total_pages > 1:
builder.adjust(3, 2)
else:
builder.adjust(2)
return builder.as_markup()
@@ -67,11 +106,12 @@ def admin_home_keyboard() -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
builder.button(text="Центр синхронизации", callback_data="admin:menu:sync")
builder.button(text="Сводки и мониторинг", callback_data="admin:menu:overview")
builder.button(text="Администраторы", callback_data="admin:admins")
builder.button(text="Ownership и topics", callback_data="admin:ownership")
builder.button(text="Mute rules", callback_data="admin:mute_list")
builder.button(text="Инструкция", callback_data="admin:guide")
builder.button(text="Пользовательское меню", callback_data="help:open")
builder.adjust(2, 2, 1, 1)
builder.adjust(2, 2, 2, 1)
return builder.as_markup()
@@ -97,10 +137,25 @@ def admin_overview_keyboard() -> InlineKeyboardMarkup:
return builder.as_markup()
def admin_result_keyboard(back_callback: str) -> InlineKeyboardMarkup:
def admin_result_keyboard(
back_callback: str,
*,
page_token: str | None = None,
page_index: int = 0,
total_pages: int = 1,
) -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
_add_pagination_row(
builder,
page_token=page_token,
page_index=page_index,
total_pages=total_pages,
)
builder.button(text="Назад", callback_data=back_callback)
builder.button(text="Админ-панель", callback_data="admin:open")
builder.button(text="Пользовательское меню", callback_data="help:open")
builder.adjust(2, 1)
if total_pages > 1:
builder.adjust(3, 2, 1)
else:
builder.adjust(2, 1)
return builder.as_markup()

2
src/glitchup_bot/models/__init__.py
View File

@@ -1,3 +1,4 @@
from glitchup_bot.models.admins import BotAdmin
from glitchup_bot.models.base import Base
from glitchup_bot.models.issues import IssueCache
from glitchup_bot.models.mute_rules import MuteRule
@@ -11,6 +12,7 @@ from glitchup_bot.models.sync import SyncState
__all__ = [
"Base",
"BotAdmin",
"GroupSubscriberOverride",
"GroupTopicOverride",
"IssueCache",

14
src/glitchup_bot/models/admins.py Normal file
View File

@@ -0,0 +1,14 @@
from datetime import datetime
from sqlalchemy import BigInteger, DateTime, Integer, func
from sqlalchemy.orm import Mapped, mapped_column
from glitchup_bot.models.base import Base
class BotAdmin(Base):
__tablename__ = "bot_admins"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(BigInteger, unique=True, index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

63
src/glitchup_bot/services/admins.py Normal file
View File

@@ -0,0 +1,63 @@
from sqlalchemy import select
from glitchup_bot.config import settings
from glitchup_bot.models.admins import BotAdmin
from glitchup_bot.models.database import get_session_factory
async def list_runtime_admins() -> list[BotAdmin]:
async with get_session_factory()() as session:
result = await session.execute(select(BotAdmin).order_by(BotAdmin.user_id))
return list(result.scalars().all())
async def list_effective_admins() -> list[tuple[int, str]]:
sources_by_user: dict[int, set[str]] = {}
for user_id in settings.telegram_admin_ids:
sources_by_user.setdefault(user_id, set()).add("env")
for record in await list_runtime_admins():
sources_by_user.setdefault(record.user_id, set()).add("runtime")
return [
(user_id, " + ".join(sorted(sources)))
for user_id, sources in sorted(sources_by_user.items(), key=lambda item: item[0])
]
async def is_admin(user_id: int | None) -> bool:
if user_id is None:
return False
if user_id in settings.telegram_admin_ids:
return True
runtime_admins = await list_runtime_admins()
if any(record.user_id == user_id for record in runtime_admins):
return True
return not settings.telegram_admin_ids and not runtime_admins
async def add_admin(user_id: int) -> bool:
async with get_session_factory()() as session:
result = await session.execute(select(BotAdmin).where(BotAdmin.user_id == user_id))
record = result.scalar_one_or_none()
if record is not None:
return False
session.add(BotAdmin(user_id=user_id))
await session.commit()
return True
async def remove_admin(user_id: int) -> bool:
async with get_session_factory()() as session:
result = await session.execute(select(BotAdmin).where(BotAdmin.user_id == user_id))
record = result.scalar_one_or_none()
if record is None:
return False
await session.delete(record)
await session.commit()
return True