import re from core.aochat import server_packets, client_packets from core.aochat.client_packets import PrivateChannelLeave from core.chat_blob import ChatBlob from core.command_param_types import Const, Character, Options, Any from core.conn import Conn from core.decorators import instance, command, setting from core.dict_object import DictObject from core.logger import Logger from core.lookup.character_service import CharacterService from core.setting_service import SettingService from core.setting_types import DictionarySettingType from core.text import Text from core.igncore import IgnCore # noinspection DuplicatedCode @instance() class AllianceRelay: MESSAGE_SOURCE = "alliance" def __init__(self): self.logger = Logger(__name__) self.relay_channel = [] def inject(self, registry): self.bot: IgnCore = registry.get_instance("bot") self.setting_service: SettingService = registry.get_instance("setting_service") self.character_service: CharacterService = registry.get_instance("character_service") self.message_hub_service = registry.get_instance("message_hub_service") self.public_channel_service = registry.get_instance("public_channel_service") self.text: Text = registry.get_instance("text") def pre_start(self): self.message_hub_service.register_message_source(self.MESSAGE_SOURCE) def start(self): self.message_hub_service.register_message_destination(self.MESSAGE_SOURCE, self.handle_relay_hub_message, ["org_channel"], [self.MESSAGE_SOURCE]) self.bot.register_packet_handler(server_packets.PrivateChannelInvited.id, self.handle_private_channel_invite, 100) self.bot.register_packet_handler(server_packets.PrivateChannelMessage.id, self.handle_private_channel_message) @command(command="mrelay", params=[], access_level="member", description="View the relay settings", sub_command="info") def mrelay(self, request): def display_color(color, msg=None): if not msg: msg = color if color: return f"{msg}" return "UNSET" blob = "" for bot in self.relay_bots().get_value().keys(): enabled = self.relay_enabled().get_value().get(bot, self.relay_enabled().get_value().get('default', 'UNSET')) prefix = self.relay_symbols().get_value().get(bot, self.relay_symbols().get_value().get('default')) relay_cmd = self.relay_command().get_value().get(bot, self.relay_command().get_value().get('default')) base = self.relay_color_base().get_value().get(bot, self.relay_color_base().get_value().get('default', None)) sender = self.relay_color_sender().get_value().get(bot, self.relay_color_sender().get_value().get('default', None)) abbrv = self.relay_color_org().get_value().get(bot, self.relay_color_org().get_value().get('default', None)) msg = self.relay_color_msg().get_value().get(bot, self.relay_color_msg().get_value().get('default', None)) our_abbrv = self.relay_guild_abbreviations().get_value().get(bot, self.relay_guild_abbreviations().get_value().get('default', 'UNSET')) example = display_color(base, f'[{display_color(abbrv, our_abbrv)}] {display_color(sender, request.sender.name)}: {display_color(msg, "And I sent you a message")}') blob += f"Relay {self.character_service.get_char_name(int(bot))}:\n" blob += f" Enabled: {enabled} [{self.text.make_tellcmd('Toggle', f'mrelay status {bot} {not enabled}')}]\n" blob += f" Prefix: {prefix} [{self.text.make_tellcmd('Edit', f'mrelay prefix {bot}')}]\n" blob += f" Relay command: {relay_cmd} [{self.text.make_tellcmd('Edit', f'mrelay rcmd {bot}')}]\n" blob += f" Base color: {display_color(base)} [{self.text.make_tellcmd('Edit', f'mrelay color {bot} base')}]\n" blob += f" Abbrv color: {display_color(abbrv)} [{self.text.make_tellcmd('Edit', f'mrelay color {bot} org')}]\n" blob += f" Sender color: {display_color(sender)} [{self.text.make_tellcmd('Edit', f'mrelay color {bot} sender')}]\n" blob += f" Message color: {display_color(msg)} [{self.text.make_tellcmd('Edit', f'mrelay color {bot} msg')}]\n" blob += f" Abbreviation: {our_abbrv} [Use !mrelay abbrv {bot} <your Abbreviation>]\n" blob += f" Example msg: {example}" blob += "\n\n" return ChatBlob("Current Relays", blob) @command(command="mrelay", params=[Const("create"), Character("relaybot_name")], access_level="admin", description="Create a new Relay") def mrelay_create(self, request, _, bot): bots: dict = self.relay_bots().get_value() if not bot.char_id: return f"The character {bot.name} does not exist." if bot.char_id in bots.keys(): return f"There's already a relay for {bot.name}." bots[bot.char_id] = "" self.relay_bots().set_value(bots) return f"Successfully created relay {bot.name}. Use mrelay to edit it." @command(command="mrelay", params=[Const("delete"), Character("relaybot_name")], access_level="admin", description="Delete a Relay") def mrelay_delete(self, request, _, bot): bots: dict = self.relay_bots().get_value() if not bot.char_id: return f"The character {bot.name} does not exist." if str(bot.char_id) not in bots.keys(): return f"There's no relay for {bot.name}." bots.pop(str(bot.char_id)) self.bot.send_packet(PrivateChannelLeave(bot.char_id)) self.relay_bots().set_value(bots) user = str(bot.char_id) cur = self.relay_enabled().get_value() cur.pop(user, None) self.relay_enabled().set_value(cur) cur = self.relay_command().get_value() cur.pop(user, None) self.relay_command().set_value(cur) cur = self.relay_bots().get_value() cur.pop(user, None) self.relay_bots().set_value(cur) cur = self.relay_guild_abbreviations().get_value() cur.pop(user, None) self.relay_guild_abbreviations().set_value(cur) cur = self.relay_color_base().get_value() cur.pop(user, None) self.relay_color_base().set_value(cur) cur = self.relay_color_msg().get_value() cur.pop(user, None) self.relay_color_msg().set_value(cur) cur = self.relay_color_sender().get_value() cur.pop(user, None) self.relay_color_sender().set_value(cur) cur = self.relay_color_org().get_value() cur.pop(user, None) self.relay_color_msg().set_value(cur) cur = self.relay_symbols().get_value() cur.pop(user, None) self.relay_symbols().set_value(cur) cur = self.relay_symbol_methods().get_value() cur.pop(user, None) self.relay_symbol_methods().set_value(cur) return f"Successfully deleted relay {bot.name}" @command(command="mrelay", params=[Const("rcmd"), Character("relaybot_name"), Any("relay_command", is_optional=True)], access_level="admin", description="Change the relay command used in a relay") def mrelay_rcmd(self, request, _, bot, relay_command): bots: dict = self.relay_bots().get_value() if not bot.char_id: return f"The character {bot.name} does not exist." if str(bot.char_id) not in bots.keys(): return f"There's no relay for {bot.name} registered. Please use mrelay create {bot.name}" bot = str(bot.char_id) current = self.relay_command().get_value() if not relay_command: blob = f"Current relay command: {current.get(bot, current.get('default', 'UNSET'))}\n\n" blob += "Here are some presets:\n" for rcmd in ["!agrc", "!gcr", "agrc", "gcr"]: blob += f" {self.text.make_tellcmd(rcmd, f'mrelay rcmd {bot} {rcmd}')}\n" blob += "\n" blob += f"Or you can use mrelay rcmd {self.character_service.get_char_name(int(bot))} <Your Prefix> to set a custom one." return ChatBlob("Pick your relay command", blob) current[bot] = relay_command self.relay_command().set_value(current) return f"Successfully changed the relaying command for relaying messages to " \ f"{self.character_service.get_char_name(int(bot))}: {relay_command}" @command(command="mrelay", params=[Const("prefix"), Character("relaybot_name"), Any("prefix", is_optional=True)], access_level="admin", description="Change the prefix setting of the relays") def mrelay_prefix(self, request, _, bot, prefix): bots: dict = self.relay_bots().get_value() if not bot.char_id: return f"The character {bot.name} does not exist." if str(bot.char_id) not in bots.keys(): return f"There's no relay for {bot.name} registered. Please use mrelay create {bot.name}" bot = str(bot.char_id) current = self.relay_symbols().get_value() if not prefix: blob = f"Current prefix: {current.get(bot, current.get('default', 'UNSET'))}\n\n" blob += "Here are some presets:\n" for prefix in ["-", "--", "+", "#", "*", "@", "$", "$$"]: blob += f" {self.text.make_tellcmd(prefix, f'mrelay prefix {bot} {prefix}')}\n" blob += "\n" blob += f"Or you can use mrelay prefix {self.character_service.get_char_name(int(bot))} <Your Prefix> to set a custom one." return ChatBlob("Pick your prefix", blob) current[bot] = prefix self.relay_symbols().set_value(current) return f"Successfully changed the symbol for relaying messages to " \ f"{self.character_service.get_char_name(int(bot))}: {prefix}" @command(command="mrelay", params=[Const("abbrv"), Character("relaybot_name"), Any("Abbreviation")], access_level="admin", description="Change the abbreviation setting of the relays") def mrelay_abbrv(self, request, _, bot, abbreviation): bots: dict = self.relay_bots().get_value() if not bot.char_id: return f"The character {bot.name} does not exist." if str(bot.char_id) not in bots.keys(): return f"There's no relay for {bot.name} registered. Please use mrelay create {bot.name}" bot = str(bot.char_id) current = self.relay_guild_abbreviations().get_value() current[bot] = abbreviation self.relay_guild_abbreviations().set_value(current) return f"Successfully changed the Abbreviation for relaying messages to " \ f"{self.character_service.get_char_name(int(bot))}: {abbreviation}" @command(command="mrelay", params=[Const("status"), Character("relaybot_name"), Options(["on", "off", "true", "false"])], access_level="admin", description="Change the enabled status of the relays") def mrelay_status(self, request, _, bot, option): bots: dict = self.relay_bots().get_value() if not bot.char_id: return f"The character {bot.name} does not exist." if str(bot.char_id) not in bots.keys(): return f"There's no relay for {bot.name} registered. Please use mrelay create {bot.name}" bot = str(bot.char_id) current = self.relay_enabled().get_value() option = True if option.lower() in ["on", "true"] else False if current.get(bot, None) == option: return f"The relay status of {self.character_service.get_char_name(int(bot))} is already {option}" current[bot] = option self.relay_enabled().set_value(current) return f"Successfully changed the status of the relay " \ f"{self.character_service.get_char_name(int(bot))} to: {option}" @command(command="mrelay", params=[Const("color"), Character("relaybot_name"), Options(["base", "sender", "org", "msg"]), Any("color_code", is_optional=True)], access_level="admin", description="Change the color settings of the relays") def mrelay_color(self, request, _, bot, option, color): bots: dict = self.relay_bots().get_value() if not bot.char_id: return f"The character {bot.name} does not exist." if str(bot.char_id) not in bots.keys(): return f"There's no relay for {bot.name} registered. Please use mrelay create {bot.name}" bot = str(bot.char_id) if option == "base": current = self.relay_color_base().get_value() if color is None: return ChatBlob("Pick a color", self.color_template(bot, option, current)) if self.check_color(color): current[bot] = color self.relay_color_base().set_value(current) else: return f"Your color is not valid: {color}. Syntax: #000000 or #000" elif option == "sender": current = self.relay_color_sender().get_value() if color is None: return ChatBlob("Pick a color", self.color_template(bot, option, current)) if self.check_color(color): current[bot] = color self.relay_color_sender().set_value(current) else: return f"Your color is not valid: {color}. Syntax: #000000 or #000" elif option == "org": current = self.relay_color_org().get_value() if color is None: return ChatBlob("Pick a color", self.color_template(bot, option, current)) if self.check_color(color): current[bot] = color self.relay_color_org().set_value(current) else: return f"Your color is not valid: {color}. Syntax: #000000 or #000" elif option == "msg": current = self.relay_color_msg().get_value() if color is None: return ChatBlob("Pick a color", self.color_template(bot, option, current)) if self.check_color(color): current[bot] = color self.relay_color_msg().set_value(current) else: return f"Your color is not valid: {color}. Syntax: #000000 or #000" return f"Successfully set the {option} color to {color}" \ f" for relay {self.character_service.get_char_name(int(bot))}" def color_template(self, bot, option, current): def display(color, msg): return f"{msg} (Save it)" blob = f"Current color: " \ f"{current.get(bot, current.get('default', '#00ff00'))}\n\n" blob += f"{display('#FF0000', 'Red')}\n" blob += f"{display('#FFFFFF', 'White')}\n" blob += f"{display('#808080', 'Grey')}\n" blob += f"{display('#DDDDDD', 'Light Grey')}\n" blob += f"{display('#9CC6E7', 'Dark Grey')}\n" blob += f"{display('#000000', 'Black')}\n" blob += f"{display('#FFFF00', 'Yellow')}\n" blob += f"{display('#8CB5FF', 'Blue')}\n" blob += f"{display('#00BFFF', 'Deep Sky Blue')}\n" blob += f"{display('#00DE42', 'Green')}\n" blob += f"{display('#FCA712', 'Orange')}\n" blob += f"{display('#FFD700', 'Gold')}\n" blob += f"{display('#FF1493', 'Deep Pink')}\n" blob += f"{display('#EE82EE', 'Violet')}\n" blob += f"{display('#8B7355', 'Brown')}\n" blob += f"{display('#00FFFF', 'Cyan')}\n" blob += f"{display('#000080', 'Navy Blue')}\n" blob += f"{display('#FF8C00', 'Dark Orange')}\n" return blob def handle_private_channel_invite(self, conn: Conn, packet: server_packets.PrivateChannelInvited): if conn.id != "main": return if str(packet.private_channel_id) not in self.relay_bots().get_value().keys(): return if not self.relay_enabled().get_value().get(str(packet.private_channel_id), self.relay_enabled().get_value().get('default', False)): return self.bot.send_packet(client_packets.PrivateChannelJoin(packet.private_channel_id)) self.logger.info(f"Joined private channel {self.character_service.get_char_name(packet.private_channel_id)}") self.relay_channel.append(str(packet.private_channel_id)) def handle_private_channel_message(self, conn: Conn, packet: server_packets.PrivateChannelMessage): if conn.id != "main": return if packet.private_channel_id == self.bot.get_char_id(): return if packet.char_id == self.bot.get_char_id(): return if not self.relay_enabled().get_value().get(str(packet.private_channel_id), self.relay_enabled().get_value().get('default', False)): return priv = str(packet.private_channel_id) message = packet.message.lstrip() rcmd = self.relay_command().get_value().get(priv, self.relay_command().get_value().get('default', "UNSET")) if rcmd == "UNSET": self.logger.warning("##FIX ME## No relay command set!") if message[:len(rcmd)] != rcmd: return message = message.replace("\n", "
") message = re.match(f"{rcmd} \[(.+?)\] (.+?): (.+)", message) if not message: return org, name, text = message.groups() org = org.strip() name = name.strip() text = text.strip() plain = f"[{org}] {name}: {text}" org = self.format_text(self.relay_color_org().get_value().get(priv, self.relay_color_org().get_value().get("default")), org) name = self.format_text(self.relay_color_sender().get_value().get(priv, self.relay_color_sender().get_value().get("default")), name) text = self.format_text(self.relay_color_msg().get_value().get(priv, self.relay_color_msg().get_value().get("default")), text) formatted = self.format_text(self.relay_color_base().get_value().get(priv, self.relay_color_base().get_value().get("default")), f"[{org}] {name}: {text}") sender = DictObject({"char_id": packet.char_id, "name": self.character_service.get_char_name(packet.char_id)}) self.message_hub_service.send_message(self.MESSAGE_SOURCE, sender, plain, formatted) def handle_relay_hub_message(self, ctx): enabled = lambda x: self.relay_enabled().get_value().get(x, self.relay_enabled().get_value().get('default', False)) if not ctx.message: return for key in self.relay_bots().get_value().keys(): if key not in self.relay_channel: continue if not enabled(key): continue prefix = self.relay_symbols().get_value().get(key, self.relay_symbols().get_value().get('default', False)) if not ctx.message.startswith(prefix): continue message = ctx.message[len(prefix):].strip() if len(message) < 1: continue sender = ctx.sender.name abbrv = self.relay_guild_abbreviations().get_value().get(key, self.setting_service.get_value("org_name")) if ctx.source == "private_channel": abbrv += " - Guest" elif ctx.source == "public_relay": abbrv += " - DC" cmd = self.relay_command().get_value().get(key, self.relay_command().get_value().get("default")) msg = f"{cmd} [{abbrv}] {sender}: {message}" self.send_message_to_alliance(key, msg) def send_message_to_alliance(self, alliance, msg): packet = client_packets.PrivateChannelMessage(int(alliance), self.text.format_message(msg, False), "\0") self.bot.conns["main"].send_packet(packet) @setting(name="relay_symbols", value={"default": "-"}, description="Symbol for external relay") def relay_symbols(self) -> DictionarySettingType: return DictionarySettingType() @setting(name="relay_symbol_methods", value={'default': "always"}, description="Relay methods for the relays") def relay_symbol_methods(self) -> DictionarySettingType: return DictionarySettingType() @setting(name="relay_bots", value={}, description="Bots used for relaying") def relay_bots(self) -> DictionarySettingType: return DictionarySettingType() @setting(name="relay_enabled", value={"default": True}, description="Enabled status of the relays") def relay_enabled(self) -> DictionarySettingType: return DictionarySettingType() @setting(name="relay_guild_abbreviations", value={"default": "UNSET"}, description="Abbreviations used for the relays") def relay_guild_abbreviations(self) -> DictionarySettingType: return DictionarySettingType() @setting(name="relay_color_base", value={"default": "#00FF00"}, description="Base color used for the relays") def relay_color_base(self) -> DictionarySettingType: return DictionarySettingType() @setting(name="relay_color_org", value={"default": "#FFFF00"}, description="Abbrv color used for the relays") def relay_color_org(self) -> DictionarySettingType: return DictionarySettingType() @setting(name="relay_color_sender", value={"default": "#FF8C00"}, description="Sender color used for the relays") def relay_color_sender(self) -> DictionarySettingType: return DictionarySettingType() @setting(name="relay_color_msg", value={"default": "#FF8C00"}, description="Message color used for the relays") def relay_color_msg(self) -> DictionarySettingType: return DictionarySettingType() @setting(name="relay_command", value={"default": "!agcr"}, description="Relay command used for the relays") def relay_command(self) -> DictionarySettingType: return DictionarySettingType() def check_color(self, msg: str): if not msg.startswith('#'): return False if len(msg) not in [4, 7]: return False return True def format_text(self, color, message): return f"{message}"