forked from FOSS/ASF-Control-Bot
102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
from sqlalchemy import select
|
|
|
|
from bot.db.models import BotAccount, User
|
|
from bot.db.session import async_session
|
|
from bot.config import ADMIN_ID
|
|
import hashlib
|
|
|
|
|
|
async def list_user_bot_accounts(telegram_id: int) -> list[BotAccount]:
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(BotAccount)
|
|
.join(User)
|
|
.where(User.telegram_id == telegram_id)
|
|
.order_by(BotAccount.bot_name)
|
|
)
|
|
return list(result.scalars())
|
|
|
|
|
|
async def list_user_activation_bot_accounts(telegram_id: int) -> list[BotAccount]:
|
|
"""Возвращает личных ботов, доступных для активации ключей пользователем.
|
|
|
|
Боты, импортированные из ASF для совместимости с панелью администратора,
|
|
являются системными, а не загруженными администратором лично.
|
|
"""
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(BotAccount)
|
|
.join(User)
|
|
.where(
|
|
User.telegram_id == telegram_id,
|
|
BotAccount.source_filename != "existing-asf",
|
|
)
|
|
.order_by(BotAccount.bot_name)
|
|
)
|
|
return list(result.scalars())
|
|
|
|
|
|
async def get_owned_bot(telegram_id: int, account_id: int) -> BotAccount | None:
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(BotAccount)
|
|
.join(User)
|
|
.where(BotAccount.id == account_id, User.telegram_id == telegram_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def create_bot_account(telegram_id: int, **values) -> BotAccount:
|
|
async with async_session() as session:
|
|
user = await session.scalar(select(User).where(User.telegram_id == telegram_id))
|
|
if user is None:
|
|
user = User(telegram_id=telegram_id)
|
|
session.add(user)
|
|
await session.flush()
|
|
account = BotAccount(owner_id=user.id, **values)
|
|
session.add(account)
|
|
await session.commit()
|
|
await session.refresh(account)
|
|
return account
|
|
|
|
|
|
async def bot_name_exists(bot_name: str) -> bool:
|
|
async with async_session() as session:
|
|
return (
|
|
await session.scalar(
|
|
select(BotAccount.id).where(BotAccount.bot_name == bot_name)
|
|
)
|
|
is not None
|
|
)
|
|
|
|
|
|
async def user_hash_exists(telegram_id: int, config_sha256: str) -> bool:
|
|
async with async_session() as session:
|
|
return (
|
|
await session.scalar(
|
|
select(BotAccount.id)
|
|
.join(User)
|
|
.where(
|
|
User.telegram_id == telegram_id,
|
|
BotAccount.config_sha256 == config_sha256,
|
|
)
|
|
)
|
|
is not None
|
|
)
|
|
|
|
|
|
async def ensure_admin_compatibility(bot_names: list[str]) -> None:
|
|
"""Показывает администратору ботов ASF, созданных до добавления владельцев."""
|
|
for bot_name in bot_names:
|
|
if await bot_name_exists(bot_name):
|
|
continue
|
|
|
|
await create_bot_account(
|
|
ADMIN_ID,
|
|
bot_name=bot_name,
|
|
source_filename="existing-asf",
|
|
config_sha256=hashlib.sha256(bot_name.encode()).hexdigest(),
|
|
upload_state="attached",
|
|
is_attached=True,
|
|
)
|