103 lines
4.1 KiB
Python
103 lines
4.1 KiB
Python
"""Allows lichess_bot to send messages to the chat."""
|
|
from __future__ import annotations
|
|
import logging
|
|
from lib import model
|
|
from lib.engine_wrapper import EngineWrapper
|
|
from lib.lichess import Lichess
|
|
from collections.abc import Sequence
|
|
from lib.timer import seconds
|
|
MULTIPROCESSING_LIST_TYPE = Sequence[model.Challenge]
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Conversation:
|
|
"""Enables the bot to communicate with its opponent and the spectators."""
|
|
|
|
def __init__(self, game: model.Game, engine: EngineWrapper, li: Lichess, version: str,
|
|
challenge_queue: MULTIPROCESSING_LIST_TYPE) -> None:
|
|
"""
|
|
Communication between lichess_bot and the game chats.
|
|
|
|
:param game: The game that the bot will send messages to.
|
|
:param engine: The engine playing the game.
|
|
:param li: A class that is used for communication with lichess.
|
|
:param version: The lichess_bot version.
|
|
:param challenge_queue: The active challenges the bot has.
|
|
"""
|
|
self.game = game
|
|
self.engine = engine
|
|
self.li = li
|
|
self.version = version
|
|
self.challengers = challenge_queue
|
|
|
|
command_prefix = "!"
|
|
|
|
def react(self, line: ChatLine) -> None:
|
|
"""
|
|
React to a received message.
|
|
|
|
:param line: Information about the message.
|
|
"""
|
|
logger.info(f'*** {self.game.url()} [{line.room}] {line.username}: {line.text}')
|
|
if line.text[0] == self.command_prefix:
|
|
self.command(line, line.text[1:].lower())
|
|
|
|
def command(self, line: ChatLine, cmd: str) -> None:
|
|
"""
|
|
Reacts to the specific commands in the chat.
|
|
|
|
:param line: Information about the message.
|
|
:param cmd: The command to react to.
|
|
"""
|
|
from_self = line.username == self.game.username
|
|
if cmd == "commands" or cmd == "help":
|
|
self.send_reply(line, "Supported commands: !wait (wait a minute for my first move), !name, !howto, !eval, !queue")
|
|
elif cmd == "wait" and self.game.is_abortable():
|
|
self.game.ping(seconds(60), seconds(120), seconds(120))
|
|
self.send_reply(line, "Waiting 60 seconds...")
|
|
elif cmd == "name":
|
|
name = self.game.me.name
|
|
self.send_reply(line, f"{name} running {self.engine.name()} (lichess_bot v{self.version})")
|
|
elif cmd == "howto":
|
|
self.send_reply(line, "How to run: Check out 'Lichess Bot API'")
|
|
elif cmd == "eval" and (from_self or line.room == "spectator"):
|
|
stats = self.engine.get_stats(for_chat=True)
|
|
self.send_reply(line, ", ".join(stats))
|
|
elif cmd == "eval":
|
|
self.send_reply(line, "I don't tell that to my opponent, sorry.")
|
|
elif cmd == "queue":
|
|
if self.challengers:
|
|
challengers = ", ".join([f"@{challenger.challenger.name}" for challenger in reversed(self.challengers)])
|
|
self.send_reply(line, f"Challenge queue: {challengers}")
|
|
else:
|
|
self.send_reply(line, "No challenges queued.")
|
|
|
|
def send_reply(self, line: ChatLine, reply: str) -> None:
|
|
"""
|
|
Send the reply to the chat.
|
|
|
|
:param line: Information about the original message that we reply to.
|
|
:param reply: The reply to send.
|
|
"""
|
|
logger.info(f'*** {self.game.url()} [{line.room}] {self.game.username}: {reply}')
|
|
self.li.chat(self.game.id, line.room, reply)
|
|
|
|
def send_message(self, room: str, message: str) -> None:
|
|
"""Send the message to the chat."""
|
|
if message:
|
|
self.send_reply(ChatLine({"room": room, "username": "", "text": ""}), message)
|
|
|
|
|
|
class ChatLine:
|
|
"""Information about the message."""
|
|
|
|
def __init__(self, message_info: dict[str, str]) -> None:
|
|
"""Information about the message."""
|
|
self.room = message_info["room"]
|
|
"""Whether the message was sent in the chat room or in the spectator room."""
|
|
self.username = message_info["username"]
|
|
"""The username of the account that sent the message."""
|
|
self.text = message_info["text"]
|
|
"""The message sent."""
|