54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
"""Репозиторий администраторов веб-панели (Блок 4.2)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from common.db.pool import get_pool
|
|
|
|
|
|
async def get_admin(telegram_id: int) -> dict[str, Any] | None:
|
|
"""Вернуть администратора по Telegram-ID (для проверки доступа)."""
|
|
async with get_pool().connection() as conn:
|
|
cur = await conn.execute(
|
|
"""
|
|
SELECT telegram_id, username, role, enabled, created_at, last_login_at
|
|
FROM bot.admin_users
|
|
WHERE telegram_id = %s
|
|
""",
|
|
(telegram_id,),
|
|
)
|
|
return await cur.fetchone()
|
|
|
|
|
|
async def ensure_admins(telegram_ids: list[int]) -> int:
|
|
"""Засеять администраторов из ADMIN_IDS (существующих не трогаем).
|
|
|
|
Возвращает количество вставленных строк.
|
|
"""
|
|
if not telegram_ids:
|
|
return 0
|
|
|
|
inserted = 0
|
|
async with get_pool().connection() as conn:
|
|
for telegram_id in dict.fromkeys(telegram_ids):
|
|
cur = await conn.execute(
|
|
"""
|
|
INSERT INTO bot.admin_users (telegram_id, role)
|
|
VALUES (%s, 'admin')
|
|
ON CONFLICT (telegram_id) DO NOTHING
|
|
""",
|
|
(telegram_id,),
|
|
)
|
|
inserted += cur.rowcount or 0
|
|
return inserted
|
|
|
|
|
|
async def touch_login(telegram_id: int) -> None:
|
|
"""Отметить успешный вход (не критично, ошибки не должны ломать логин)."""
|
|
async with get_pool().connection() as conn:
|
|
await conn.execute(
|
|
"UPDATE bot.admin_users SET last_login_at = now() WHERE telegram_id = %s",
|
|
(telegram_id,),
|
|
)
|