import time import logging from enum import Enum from velascobot2.utils import ChatData, GraphicType, Response from velascobot2.parrot import Parrot logger = logging.getLogger(__name__) class ReplyMode(Enum): FIXED = 1 CHANCE = 2 class Messenger: def __init__( self, bot_name: str, bosses: list[str] = [], nicknames: list[str] = [], reply: float = 0.1, repeat: float = 0.05, wake_up: bool = False, mode: ReplyMode = ReplyMode.FIXED, mute_time: int = 60, max_len: int = 50, min_period: int = 1, max_period: int = 100000, ): # List of nicknames (other than the username) that the bot can react to self.names = nicknames # Mute time for Telegram network errors self.mute_time = mute_time # Last mute timestamp self.mute_timer = None # The bot's username, "@" included. Updated on Bot creation self.username = "@" + bot_name # Wakeup flag that determines if it should send a wakeup message to # stored groupchats self.wake_up = wake_up # Chance of sending messages as replies self.reply = reply # Chance of sending 2 messages in a row self.repeat = repeat # Bot bosses ID self.bosses = bosses # Max word length for a message self.max_len = max_len # Whether a message is guaranteed after period_messages, or if it's a # random chance self.mode = mode # minimum value allowed for message_period self.min_period = min_period # maximum value allowed for message_period self.max_period = max_period self._parrot = Parrot(bot_name) def get_all_chats(self) -> list[ChatData]: return self._parrot.get_all_chats() def mentioned(self, text: str) -> bool: """Returns True if the bot's username or any of its nicknames is mentioned in the text, as long as it's not some other user's username""" if self.username in text: return True for name in self.names: if name in text and "@{}".format(name) not in text: return True return False def is_mute(self) -> bool: """Returns True if not enough time has passed since the last time the bot got muted.""" current_time = int(time.perf_counter()) return ( self.mute_timer is not None and (current_time - self.mute_timer) < self.mute_time ) def give_response( self, chat_id: str ) -> tuple[Response, str | tuple[GraphicType, str]]: response = self._parrot.choose_response_type(chat_id) if response == Response.TEXT: return response, self._parrot.give_text(chat_id) elif response == Response.GRAPHIC: return response, self._parrot.give_graphic(chat_id) raise Exception("Response type not implemented") def set_mute_timer(self, counter: int): self.mute_timer = counter def process_text(self, chat_id: str, text: str): logger.debug(f"{chat_id} - Processing text: {text}") for name in self.names: if name.casefold() in text.casefold() and len(text.split()) <= 3: # If it's less than 3 words and one of the bot's names is in # the message, ignore it as it's most probably just a summon return self._parrot.parse_text(chat_id, text) def process_graphic( self, chat_id: str, graph_type: GraphicType, graph_id: str ): logger.debug(f"{chat_id} - Processing graphic {graph_type}: {graph_id}") self._parrot.parse_graphic(chat_id, graph_type, graph_id) def get_chat_data(self, chat_id: str) -> ChatData | None: return self._parrot.get_chat_data(chat_id) def ensure_chat_exists(self, chat_id: str, chat_type: str, chat_name: str): self._parrot.ensure_chat_exists(chat_id, chat_type, chat_name) def update_chat_name(self, chat_id: str, chat_name: str): self._parrot.update_chat_name(chat_id, chat_name) def update_message_period(self, chat_id: str, period: int): if period < self.min_period or period > self.max_period: raise Exception( f"The period must be between {self.min_period} and" f" {self.max_period}, but got {period}" ) self._parrot.update_chat_message_period(chat_id, period) def update_answer_chance(self, chat_id: str, chance: float): if chance < 0 or chance > 1: raise Exception( f"The period must be between 0 and 1, but got {chance}" ) self._parrot.update_chat_answer_chance(chat_id, chance) def update_set_restricted(self, chat_id: str, restricted: bool): self._parrot.update_chat_set_restricted(chat_id, restricted) def update_set_silenced(self, chat_id: str, silenced: bool): self._parrot.update_chat_set_silenced(chat_id, silenced) def get_processed_messages(self, chat_id: str) -> int: return self._parrot.get_processed_messages(chat_id)