85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
"""Доступ к данным пользователей."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from common.db.pool import get_pool
|
||
|
||
|
||
async def add_user(id_user: int, username: str, first_name: str, last_name: str) -> None:
|
||
"""Создать/обновить пользователя (счётчик сообщений ведёт аналитика)."""
|
||
async with get_pool().connection() as conn, conn.cursor() as cur:
|
||
await cur.execute(
|
||
"CALL bot.add_user(%s, %s, %s, %s)",
|
||
(id_user, username, first_name, last_name),
|
||
)
|
||
|
||
|
||
async def add_nick(id_user: int, nick: str) -> None:
|
||
"""Установить ник пользователя."""
|
||
async with get_pool().connection() as conn, conn.cursor() as cur:
|
||
await cur.execute("CALL bot.add_nick(%s, %s)", (id_user, nick))
|
||
|
||
|
||
async def set_nick_with_history(
|
||
id_user: int,
|
||
nick: str,
|
||
*,
|
||
editor_id: int | None = None,
|
||
source: str = "user",
|
||
) -> str | None:
|
||
"""Сменить ник и записать изменение в ``user_changes``.
|
||
|
||
Возвращает старый ник (``None``, если записи не было). Используется командой
|
||
``/ник``, чтобы смена ника пользователем была видна в админке.
|
||
"""
|
||
async with get_pool().connection() as conn:
|
||
cur = await conn.execute("SELECT nick FROM bot.de_users WHERE id_user = %s FOR UPDATE", (id_user,))
|
||
row = await cur.fetchone()
|
||
|
||
if row is None:
|
||
await conn.execute("INSERT INTO bot.de_users (id_user, nick) VALUES (%s, %s)", (id_user, nick))
|
||
old_nick = None
|
||
else:
|
||
old_nick = row["nick"]
|
||
if old_nick == nick:
|
||
return old_nick
|
||
await conn.execute("UPDATE bot.de_users SET nick = %s WHERE id_user = %s", (nick, id_user))
|
||
|
||
await conn.execute(
|
||
"""
|
||
INSERT INTO bot.user_changes (id_user, field, old_value, new_value, editor_id, source)
|
||
VALUES (%s, 'nick', %s, %s, %s, %s)
|
||
""",
|
||
(id_user, old_nick, nick, editor_id, source),
|
||
)
|
||
return old_nick
|
||
|
||
|
||
async def get_nick_list() -> list[dict[str, Any]]:
|
||
"""Список всех игроков (username, nick), отсортированный по нику."""
|
||
async with get_pool().connection() as conn, conn.cursor() as cur:
|
||
await cur.execute("SELECT username, nick FROM bot.get_nick_list()")
|
||
return await cur.fetchall()
|
||
|
||
|
||
async def get_nick_list_subscribe() -> list[dict[str, Any]]:
|
||
"""Список подписанных на рассылки (username, nick)."""
|
||
async with get_pool().connection() as conn, conn.cursor() as cur:
|
||
await cur.execute("SELECT username, nick FROM bot.get_nick_list_subscribe()")
|
||
return await cur.fetchall()
|
||
|
||
|
||
async def get_info_by_id(id_user: int) -> dict[str, Any] | None:
|
||
"""Сводная информация о пользователе."""
|
||
async with get_pool().connection() as conn, conn.cursor() as cur:
|
||
await cur.execute("SELECT * FROM bot.get_info_by_id(%s)", (id_user,))
|
||
return await cur.fetchone()
|
||
|
||
|
||
async def del_user_by_id(id_user: int) -> None:
|
||
"""Удалить пользователя (при выходе из чата)."""
|
||
async with get_pool().connection() as conn, conn.cursor() as cur:
|
||
await cur.execute("CALL bot.del_user_by_id(%s)", (id_user,))
|