"""In-process boti: DB ucty botov a ich rozhodovacie "mozgy" z rl/players.py. Bot je normalny hrac na sedadle -- ma riadok v tabulke `players` (aby historia, standings a restore fungovali bez zmeny), ale ziadny socket. Tahovu slucku botov ma api/__init__.py (_run_bot_turns); tu je len to, co potrebuje DB a rl vrstvu. Bezpecnost botich uctov: totp_secret je nahodny a nikde sa neuklada v citatelnej podobe, totp_last_step sa nastavi na aktualny krok -- ucet tym padom NIE JE "nedokoncena registracia" (viz auth._is_unconfirmed), takze register_account ho odmietne prepisat a login bez secretu neprejde. """ import os from random import Random import pyotp from sqlalchemy import select from api import auth from db import crypto from db.db import async_session from db.models import Player from rl.players import HeuristicPlayer, RandomPlayer from rl.pure_net import DEFAULT_WEIGHTS_PATH, PureNet, PureNeuralPlayer # Prefix je konvencia na rozpoznanie bota (aj po restarte servera, kedy sa # sedadla obnovuju z DB len ako (player_id, username)). BOT_PREFIX = "bot:" DEFAULT_KIND = "heuristic" # Natrenovana siet -- vahy (rl/weights/, export z rl/export.py) sa nacitaju # raz a zdielaju medzi botmi (PureNet je bezstavovy, len cita). _pure_net: PureNet | None = None def _neural_brain() -> PureNeuralPlayer: global _pure_net if _pure_net is None: _pure_net = PureNet.load() return PureNeuralPlayer(_pure_net) def neural_available() -> bool: return os.path.exists(DEFAULT_WEIGHTS_PATH) _BRAINS = { "heuristic": lambda: HeuristicPlayer(Random()), "random": lambda: RandomPlayer(Random()), "neural": _neural_brain, } BOT_KINDS = tuple(_BRAINS) def available_kinds() -> tuple: """Druhy botov ponuknutelne na tomto serveri (neural len s vahami).""" return tuple(k for k in _BRAINS if k != "neural" or neural_available()) def is_bot_username(username: str) -> bool: return bool(username) and username.startswith(BOT_PREFIX) def kind_of(username: str) -> str: """'bot:heuristic-2' -> 'heuristic'; neznamy druh padne na DEFAULT_KIND.""" body = username[len(BOT_PREFIX):] kind = body.rsplit("-", 1)[0] return kind if kind in _BRAINS else DEFAULT_KIND def make_brain(username: str): """Rozhodovaci objekt (guess/play rozhranie z rl/players.py) pre bota. Neural bez suboru vah (napr. restore hry na serveri bez exportu) padne na heuristiku -- sedadlo hra dalej, len inym mozgom. """ kind = kind_of(username) if kind == "neural" and not neural_available(): kind = DEFAULT_KIND return _BRAINS[kind]() def _suffix_number(username: str) -> int: try: return int(username.rsplit("-", 1)[1]) except (IndexError, ValueError): return 0 async def ensure_bot_account(kind: str, exclude_ids: set) -> dict: """Najde alebo zalozi boti ucet daneho druhu; vrati {player_id, username}. `exclude_ids` su ucty uz obsadene v danej hre -- kazde sedadlo potrebuje INY ucet (Game.playerN_id aj unikat v Guess predpokladaju 4 rozne ID). Ucty sa cisluju bot:-1, -2, ... a recykluju sa medzi hrami. """ prefix = f"{BOT_PREFIX}{kind}-" async with async_session() as session: rows = ( await session.scalars( select(Player).where(Player.username.like(prefix + "%")) ) ).all() for player in sorted(rows, key=lambda p: _suffix_number(p.username)): if player.id not in exclude_ids: return {"player_id": player.id, "username": player.username} number = 1 + max((_suffix_number(p.username) for p in rows), default=0) username = f"{prefix}{number}" player = Player( username=username, # nahodny secret, ktory sa zahodi -- nikto sa zan neprihlasi totp_secret=crypto.encrypt(pyotp.random_base32()), # nenulovy last_step = ucet sa netvari ako nedokoncena registracia totp_last_step=auth._current_step(), ) session.add(player) await session.commit() return {"player_id": player.id, "username": player.username}