from __future__ import annotations import logging import os import subprocess import sys import time from typing import Optional, TYPE_CHECKING from queue import Empty import humanize from crontab import CronTab from bot.commands.command import Command from bot.config.models import CronEntryModel from bot.player.enums import State from bot import app_vars, errors if TYPE_CHECKING: from bot.TeamTalk.structs import User class BlockCommandCommand(Command): @property def help(self) -> str: return self.translator.translate( "+/-COMMAND Block or unblock command. +COMMAND add command to the block list. -COMMAND remove from it. Without a command shows the block list." ) def __call__(self, arg: str, user: User) -> Optional[str]: arg = arg.lower() if len(arg) >= 1 and arg[1:] not in self.command_processor.commands_dict: raise errors.InvalidArgumentError() if not arg: return ( ", ".join(self.config.general.blocked_commands) if self.config.general.blocked_commands else self.translator.translate("the list is empty") ) if arg[0] == "+": if arg[1::] not in self.config.general.blocked_commands: self.config.general.blocked_commands.append(arg[1::]) self.run_async( self.ttclient.send_message, self.translator.translate( "command \"{command}\" was blocked." ).format(command=arg[1::]), user, ) self.config_manager.save() else: self.translator.translate("This command is already added") elif arg[0] == "-": if arg[1::] in self.config.general.blocked_commands: del self.config.general.blocked_commands[ self.config.general.blocked_commands.index(arg[1::]) ] self.run_async( self.ttclient.send_message, self.translator.translate( "command \"{command}\" was unblocked." ).format(command=arg[1::]), user, ) self.config_manager.save() else: return self.translator.translate("This command is not blocked") else: raise errors.InvalidArgumentError() class ChangeGenderCommand(Command): @property def help(self) -> str: return self.translator.translate( "GENDER Changes bot's gender. n neutral, m male, f female" ) def __call__(self, arg: str, user: User) -> Optional[str]: try: self.ttclient.change_gender(arg) self.config.teamtalk.gender = arg except KeyError: raise errors.InvalidArgumentError() self.config_manager.save() class ChangeLanguageCommand(Command): @property def help(self) -> str: return self.translator.translate("LANGUAGE change player language") def __call__(self, arg: str, user: User) -> Optional[str]: if arg: try: self.translator.set_locale(arg) self.config.general.language = arg self.ttclient.change_status_text("") self.run_async( self.ttclient.send_message, self.translator.translate( "the language has been changed to {language}." ).format(language=arg), user, ) self.config_manager.save() except errors.LocaleNotFoundError: return self.translator.translate("Incorrect language") else: return self.translator.translate( "Current language: {current_language}. Available languages: {available_languages}." ).format( current_language=self.translator.get_locale(), available_languages=", ".join(self.translator.get_locales()), ) class ChangeNicknameCommand(Command): @property def help(self) -> str: return self.translator.translate("NICKNAME Changes bot's nickname") def __call__(self, arg: str, user: User) -> Optional[str]: if arg: self.ttclient.change_nickname(arg) self.config.teamtalk.nickname = arg self.run_async( self.ttclient.send_message, self.translator.translate( "nickname change to {newname}." ).format(newname=arg), user, ) else: self.ttclient.change_nickname(app_vars.client_name) self.config.teamtalk.nickname = app_vars.client_name self.run_async( self.ttclient.send_message, self.translator.translate( "nickname set to default client ({newname})." ).format(newname=app_vars.client_name), user, ) self.config_manager.save() class ClearCacheCommand(Command): @property def help(self) -> str: return self.translator.translate( "r/f Clears bot's cache. r clears recents, f clears favorites, without an option clears the entire cache" ) def __call__(self, arg: str, user: User) -> Optional[str]: if not arg: self.cache.recents.clear() self.cache.favorites.clear() self.cache_manager.save() return self.translator.translate("Cache cleared") elif arg == "r": self.cache.recents.clear() self.cache_manager.save() return self.translator.translate("Recents cleared") elif arg == "f": self.cache.favorites.clear() self.cache_manager.save() return self.translator.translate("Favorites cleared") class ShowDefaultServerCommand(Command): @property def help(self) -> str: return self.translator.translate("Read config file and show default Server, Port, Username, Password, Channel, Channel password.") def __call__(self, arg: str, user: User) -> Optional[str]: return self.translator.translate( "Server: {hostname}\n" "TCP Port: {tcpport}\n" "UDP Port: {udpport}\n" "Username: {username}\n" "Password: {password}\n" "Channel: {channel}\n" "Channel password: {channel_password}" ).format( hostname=self.config.teamtalk.hostname, tcpport=self.config.teamtalk.tcp_port, udpport=self.config.teamtalk.udp_port, username=self.config.teamtalk.username, password=self.config.teamtalk.password, channel=self.config.teamtalk.channel, channel_password=self.config.teamtalk.channel_password, ) class DefaultServerCommand(Command): @property def help(self): return self.translator.translate( 'Change default Server information.' ) def __call__(self, arg: str, user: User) -> Optional[str]: if arg: self.config.teamtalk.hostname = arg self.run_async( self.ttclient.send_message, self.translator.translate( "Default Server change to {newhostname}." ).format(newhostname=arg), user, ) self.config_manager.save() else: return self.translator.translate("Default server can not be blank, Please specify default Server information!") class DefaultTCPPortCommand(Command): @property def help(self): return self.translator.translate( 'Change default TCP port information.' ) def __call__(self, arg: str, user: User) -> Optional[str]: if arg: try: tcpport = int(arg) self.config.teamtalk.tcp_port = int(arg) self.run_async( self.ttclient.send_message, self.translator.translate( "Default TCP port change to {newtcp_port}." ).format(newtcp_port=int(arg)), user, ) self.config_manager.save() except ValueError: raise errors.InvalidArgumentError else: return self.translator.translate("Default TCP port can not be blank, Please specify default TCP port!") class DefaultUDPPortCommand(Command): @property def help(self): return self.translator.translate( 'Change default UDP port information.' ) def __call__(self, arg: str, user: User) -> Optional[str]: if arg: try: udpport = int(arg) self.config.teamtalk.udp_port = int(arg) self.run_async( self.ttclient.send_message, self.translator.translate( "Default TCP port change to {newudp_port}." ).format(newudp_port=int(arg)), user, ) self.config_manager.save() except ValueError: raise errors.InvalidArgumentError else: return self.translator.translate("Default UDP port can not be blank, Please specify default UDP port!") class DefaultUsernameCommand(Command): @property def help(self): return self.translator.translate( 'Change default Username information. Without any arguments, set default Username to blank.' ) def __call__(self, arg: str, user: User) -> Optional[str]: if arg: self.config.teamtalk.username = arg self.run_async( self.ttclient.send_message, self.translator.translate( "Default Username change to {newusername}." ).format(newusername=arg), user, ) self.config_manager.save() else: self.config.teamtalk.username = "" self.config_manager.save() return self.translator.translate("Default username set to blank.") class DefaultPasswordCommand(Command): @property def help(self): return self.translator.translate( 'Change default Password information. Without any arguments, set default password to blank.' ) def __call__(self, arg: str, user: User) -> Optional[str]: if arg: self.config.teamtalk.password = arg self.run_async( self.ttclient.send_message, self.translator.translate( "Default Password change to {newpassword}." ).format(newpassword=arg), user, ) self.config_manager.save() else: self.config.teamtalk.password = "" self.config_manager.save() return self.translator.translate("Default Password set to blank.") class DefaultChannelCommand(Command): @property def help(self): return self.translator.translate( 'Change default Channel information. Without any arguments, set default Channel to current channel.' ) def __call__(self, arg: str, user: User) -> Optional[str]: if arg: self.config.teamtalk.channel = arg self.run_async( self.ttclient.send_message, self.translator.translate( "Default Channel change to {newchannel}, if your channel has a password, please set it manually!" ).format(newchannel=arg), user, ) self.config_manager.save() else: channel_id = str(self.ttclient.channel.id) self.config.teamtalk.channel = channel_id self.config_manager.save() return self.translator.translate("Default Channel set to current channel, if your channel has a password, please set it manually!") class DefaultChannelPasswordCommand(Command): @property def help(self): return self.translator.translate( 'Change default Channel password information. Without any arguments, set default Channel password to blank..' ) def __call__(self, arg: str, user: User) -> Optional[str]: if arg: self.config.teamtalk.channel_password = arg self.run_async( self.ttclient.send_message, self.translator.translate( "Default Channel password change to {newchannel_password}." ).format(newchannel_password=arg), user, ) self.config_manager.save() else: self.config.teamtalk.channel_password = "" self.config_manager.save() return self.translator.translate("Default Channel password set to blank.") class GetChannelIDCommand(Command): @property def help(self) -> str: return self.translator.translate("Get current channel ID, will be useful for several things.") def __call__(self, arg: str, user: User) -> Optional[str]: return str(self.ttclient.channel.id) class JoinChannelCommand(Command): @property def help(self) -> str: return self.translator.translate( 'Join channel. first argument is channel name or id, second argument is password, split argument " | ", if password is undefined, don\'t type second argument' ) def __call__(self, arg: str, user: User) -> Optional[str]: args = self.command_processor.split_arg(arg) if not arg: channel = self.config.teamtalk.channel password = self.config.teamtalk.channel_password elif len(args) == 2: channel = args[0] password = args[1] else: channel = arg password = "" if isinstance(channel, str) and channel.isdigit(): channel = int(channel) try: cmd = self.ttclient.join_channel(channel, password) except ValueError: return self.translator.translate("This channel does not exist") while True: try: event = self.ttclient.event_success_queue.get_nowait() if event.source == cmd: break else: self.ttclient.event_success_queue.put(event) except Empty: pass try: error = self.ttclient.errors_queue.get_nowait() if error.command_id == cmd: return self.translator.translate( "Error joining channel: {error}".format(error=error.message) ) else: self.ttclient.errors_queue.put(error) except Empty: pass time.sleep(app_vars.loop_timeout) """ class CreateChannelCommand(Command): @property def help(self) -> str: return self.translator.translate( 'Create channel. first argument is channel name, second argument is parent channel ID, third argument is the channel topic, fourth argument is the channel password. Split arguments " | ", if password or topic is undefined, don\'t type fourth or third argument.' ) def __call__(self, arg: str, user: User) -> Optional[str]: args = self.command_processor.split_arg(arg) if len(args) == 4: channel_name = args[0] parent_channel_id = int(args[1]) channel_topic = args[2] channel_password = args[3] elif len(args) == 3: channel_name = args[0] parent_channel_id = int(args[1]) channel_topic = args[2] channel_password = "" elif len(args) == 2: channel_name = args[0] parent_channel_id = int(args[1]) channel_topic = "" channel_password = "" else: return self.help """ """ class TaskSchedulerCommand(Command): @property def help(self) -> str: return self.translator.translate("Task scheduler") def __call__(self, arg: str, user: User) -> Optional[str]: if arg[0] == "+": self._add(arg[1::]) def _add(self, arg: str) -> None: args = arg.split("|") timestamp = self._get_timestamp(args[0]) task = [] for arg in args[1::]: try: command, arg = self.parse_command(message.text) if self.check_access(message.user, command): command = self.get_command(command, message.user) task.append((command, arg)) except errors.AccessDeniedError as e: return e except (errors.ParseCommandError, errors.UnknownCommandError): return self.translator.translate("Unknown command. Send \"start\" for help.") except errors.InvalidArgumentError: return self.help(command, message.user) if timestamp in self.module_manager.task_scheduler.tasks: self.module_manager.task_scheduler[timestamp].append(task) else: self.module_manager.task_scheduler.tasks[timestamp] = [task,] def _get_timestamp(self, t): return int(datetime.combine(datetime.today(), datetime.strptime(t, self.config["general"]["time_format"]).time()).timestamp()) """ class VoiceTransmissionCommand(Command): @property def help(self) -> str: return self.translator.translate("Enables or disables voice transmission") def __call__(self, arg: str, user: User) -> Optional[str]: if not self.ttclient.is_voice_transmission_enabled: self.ttclient.enable_voice_transmission() if self.player.state == State.Stopped: self.ttclient.change_status_text( self.translator.translate("Voice transmission enabled") ) return self.translator.translate("Voice transmission enabled") else: self.ttclient.disable_voice_transmission() if self.player.state == State.Stopped: self.ttclient.change_status_text("") return self.translator.translate("Voice transmission disabled") class LockCommand(Command): @property def help(self) -> str: return self.translator.translate("Locks or unlocks the bot") def __call__(self, arg: str, user: User) -> Optional[str]: self.command_processor.locked = not self.command_processor.locked if self.command_processor.locked: self.config.general.bot_lock = True self.run_async( self.ttclient.send_message, self.translator.translate("Bot is now locked by administrator."), user, ) else: self.config.general.bot_lock = False self.run_async( self.ttclient.send_message, self.translator.translate("Bot is now unlocked by administrator."), user, ) self.config_manager.save() return None def on_startup(self) -> None: self.command_processor.locked = self.config.general.bot_lock class ChangeStatusCommand(Command): @property def help(self) -> str: return self.translator.translate("STATUS Changes bot's status") def __call__(self, arg: str, user: User) -> Optional[str]: if arg: self.ttclient.change_status_text(arg) self.config.teamtalk.status = self.ttclient.status self.run_async( self.ttclient.send_message, self.translator.translate("Status changed."), user, ) else: self.config.teamtalk.status = "" self.ttclient.change_status_text("") self.run_async( self.ttclient.send_message, self.translator.translate("Status change to default."), user, ) self.config_manager.save() class EventHandlingCommand(Command): @property def help(self) -> str: return self.translator.translate("Enables or disables event handling") def __call__(self, arg: str, user: User) -> Optional[str]: self.config.teamtalk.event_handling.load_event_handlers = ( not self.config.teamtalk.event_handling.load_event_handlers ) return ( self.translator.translate("Event handling is enabled") if self.config.teamtalk.event_handling.load_event_handlers else self.translator.translate("Event handling is disabled") ) class SendChannelMessages(Command): @property def help(self) -> str: return self.translator.translate( "Send your message to current channel. Without any argument, turn or off Channel messages." ) def __call__(self, arg: str, user: User) -> Optional[str]: if arg: if self.config.general.send_channel_messages: self.ttclient.send_message( self .translator.translate("{message}.\nSender: {nickname}.").format( message=arg, nickname=user.nickname), type=2 ) elif not self.config.general.send_channel_messages: return self.translator.translate( "Please enable channel messages first" ) else: self.config.general.send_channel_messages = (not self.config.general.send_channel_messages) return (self.translator.translate("Channel messages enabled") if self.config.general.send_channel_messages else self.translator.translate("Channel messages disabled") ) self.config_manager.save() class SendBroadcastMessages(Command): @property def help(self) -> str: return self.translator.translate( "Send broadcast message to all users. Without any argument, turn on or off Broadcast messages" ) def __call__(self, arg: str, user: User) -> Optional[str]: if arg: if self.config.general.send_broadcast_messages: self.ttclient.send_message( self .translator.translate("{message}.\nSender: {nickname}.").format( message=arg, nickname=user.nickname), type=3 ) elif not self.config.general.send_broadcast_messages: return self.translator.translate( "Please enable broadcast messages first" ) else: self.config.general.send_broadcast_messages = (not self.config.general.send_broadcast_messages) return (self.translator.translate("Broadcast messages enabled") if self.config.general.send_broadcast_messages else self.translator.translate("Broadcast messages disabled") ) self.config_manager.save() class SaveConfigCommand(Command): @property def help(self) -> str: return self.translator.translate("Saves bot's configuration") def __call__(self, arg: str, user: User) -> Optional[str]: self.config_manager.save() return self.translator.translate("Configuration saved") class AdminUsersCommand(Command): @property def help(self) -> str: return self.translator.translate( "+/-USERNAME Manage the list of administrators. +USERNAME add a user. -USERNAME remove it. Without an option show the list." ) def __call__(self, arg: str, user: User) -> Optional[str]: if arg: if arg[0] == "+": self.config.teamtalk.users.admins.append(arg[1::]) self.run_async( self.ttclient.send_message, self.translator.translate( "{username} is now admin in this player." ).format(username=arg[1::]), user, ) self.config_manager.save() elif arg[0] == "-": try: del self.config.teamtalk.users.admins[ self.config.teamtalk.users.admins.index(arg[1::]) ] self.run_async( self.ttclient.send_message, self.translator.translate( "{username} no longer admin in this player." ).format(username=arg[1::]), user, ) self.config_manager.save() except ValueError: return self.translator.translate( "this user is not in the admin list" ) else: admins = self.config.teamtalk.users.admins.copy() if len(admins) > 0: if "" in admins: admins[admins.index("")] = "" return ", ".join(admins) else: return self.translator.translate("the list is empty") class BannedUsersCommand(Command): @property def help(self) -> str: return self.translator.translate( "+/-USERNAME Manage the list of banned users. +USERNAME add a user. -USERNAME remove it. Without an option show the list." ) def __call__(self, arg: str, user: User) -> Optional[str]: if arg: if arg[0] == "+": self.config.teamtalk.users.banned_users.append(arg[1::]) self.run_async( self.ttclient.send_message, self.translator.translate( "{username} now banned" ).format(username=arg[1::], nickname=user.nickname), user, ) self.config_manager.save() elif arg[0] == "-": try: del self.config.teamtalk.users.banned_users[ self.config.teamtalk.users.banned_users.index(arg[1::]) ] self.run_async( self.ttclient.send_message, self.translator.translate( "{username} now unbanned." ).format(username=arg[1::]), user, ) self.config_manager.save() except ValueError: return self.translator.translate("this user is not banned") else: banned_users = self.config.teamtalk.users.banned_users.copy() if len(banned_users) > 0: if "" in banned_users: banned_users[banned_users.index("")] = "" return ", ".join(banned_users) else: return self.translator.translate("the list is empty") class QuitCommand(Command): @property def help(self) -> str: return self.translator.translate("Quits the bot") def __call__(self, arg: str, user: User) -> Optional[str]: self.config_manager.save() if self.player.state != State.Stopped: volume = int(0) if 0 <= volume <= self.config.player.max_volume: self.player.set_volume(int(0)) self._bot.close() class RestartCommand(Command): @property def help(self) -> str: return self.translator.translate("Restarts the bot") def __call__(self, arg: str, user: User) -> Optional[str]: self.config_manager.save() if self.player.state != State.Stopped: volume = int(0) if 0 <= volume <= self.config.player.max_volume: self.player.set_volume(int(0)) self._bot.close() args = sys.argv if sys.platform == "win32": subprocess.run([sys.executable] + args) else: args.insert(0, sys.executable) os.execv(sys.executable, args) class SchedulerCommand(Command): @property def help(self) -> str: return self.translator.translate( "Controls the cron scheduler, allowing admins to toggle it and add / remove scheduled commands" ) def __call__(self, arg: str, user: User) -> Optional[str]: if arg == "toggle" or arg == "t": return self.toggle() elif arg.startswith("add "): # pass in the string after our command: add_argstr = arg[4:] return self.add(add_argstr) elif arg.startswith("rm "): return self.remove(arg[3:]) elif arg.strip() == "list" or arg.strip() == "ls": return self.list_tasks() elif arg.strip() in ["help", "?", "h"]: return """Unknown cr subcommand. Options: toggle (shorten to t) - turn the scheduler on or off. add - add a cron task. Format is cron expression|command with arguments. Remove - remove a task. list or ls - list all scheduled tasks.""" else: s: str = ( self.translator.translate( "Scheduled tasks are enabled (disable with 'cr toggle')" ) if self.config.schedule.enabled else self.translator.translate( "Scheduled tasks are disabled; enable with 'cr toggle'" ) ) s += "\n" + self.list_tasks() s += self.translator.translate( "Commands: \ncr add cron expression|command\ncr rm #\ncr ls\ncr toggle" ) return s def toggle(self): self.config.schedule.enabled = not self.config.schedule.enabled if self.config.schedule.enabled: self.reparse_patterns() return ( self.translator.translate("Scheduler enabled.") if self.config.schedule.enabled else self.translator.translate("Scheduler disabled") ) def reparse_patterns(self): self._bot.cron_patterns = [] for entry in self.config.schedule.patterns: logging.debug( f"Parsing cron pattern '{entry.pattern}' and appending to list" ) e = CronTab(entry.pattern) self._bot.cron_patterns.append((e, entry)) def add(self, argstr: str) -> str: # Our arg should be a cron expression, | and the command. help_text = self.translator.translate( "Incorrect format. Cron expression | command you want to run with arguments after" ) if "|" not in argstr: return help_text args = argstr.split("|") if len(args) != 2: return help_text cronexpr = args[0].strip() cmd = args[1].strip() try: ct: CronTab = CronTab(cronexpr) entry = CronEntryModel(pattern=cronexpr, command=cmd) self.config.schedule.patterns.append(entry) self.reparse_patterns() return self.translator.translate("Task scheduled.") except ValueError: return self.translator.translate( "Not a valid cron expression. Pleaes use cr expression-help for more details." ) def remove(self, arg: str) -> str: if arg == "": return self.list_tasks() try: task_index = int(arg) task_index -= 1 if task_index > len(self.config.schedule.patterns) or task_index < 0: return self.translator.translate( "Task number out of range - should be 1 to {}".format(len(l)) ) task = self.config.schedule.patterns.pop(task_index) self.reparse_patterns() return self.translator.translate( f"Removed task #{task_index+1}, {task.pattern}: {task.command}" ) except ValueError: return self.translator.translate("Invalid task number") def list_tasks(self) -> str: if len(self.config.schedule.patterns) == 0: return self.translator.translate("There are no scheduled tasks") lines: list[str] = [] for idx, task in enumerate(self.config.schedule.patterns): lines.append("Task #{}".format(idx + 1)) lines.append(" Cron pattern: {}".format(task.pattern)) lines.append(" command: {}".format(task.command)) c = CronTab(task.pattern) lines.append( "Pattern ran: {} ago;".format( humanize.precisedelta(c.previous(default_utc=False)) ) ) lines.append( "Next run in {}.".format( humanize.precisedelta(c.next(default_utc=False)) ) ) s: str = "" for l in lines: s += l + "\n" return s return lines