Files
igncore/modules/orgbot/alliance/alliance_relay.py
T
Minidodo 810c2c8c4d Fix for mrelay - some bots are sending malformed messages, caused by wrong usage of their modules.
Accessing an external service through the bot for gathering tower data is nolonger supported; and should be done via external scripts.
Fix for callers, and missing alias'es for loot tables.
!accounts will only show mains now.
the character order of all alt lists has been reversed: [main] high => low instead of [main] low => high
!account add <name> also marks accounts as type 0, if an account gets re-enabled. might cause strange behaviour with member-logs, if used in onlinebots.
Member type is being displayed in !account now. [Member (X)]
2021-09-19 14:09:44 +02:00

426 lines
22 KiB
Python

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.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="admin",
description="View the relay settings")
def mrelay(self, request):
def display_color(color, msg=None):
if not msg:
msg = color
if color:
return f"<font color={color}>{msg}</font>"
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 <highlight>{self.character_service.get_char_name(int(bot))}</highlight>:\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 <highlight>!mrelay abbrv {bot} &lt;your Abbreviation&gt;</highlight>]\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 <highlight>{bot.name}</highlight> does not exist."
if bot.char_id in bots.keys():
return f"There's already a relay for <highlight>{bot.name}</highlight>."
bots[bot.char_id] = ""
self.relay_bots().set_value(bots)
return f"Successfully created relay <highlight>{bot.name}</highlight>. Use <symbol>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 <highlight>{bot.name}</highlight> does not exist."
if str(bot.char_id) not in bots.keys():
return f"There's no relay for <highlight>{bot.name}</highlight>."
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 <highlight>{bot.name}</highlight>"
@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 <highlight>{bot.name}</highlight> does not exist."
if str(bot.char_id) not in bots.keys():
return f"There's no relay for <highlight>{bot.name}</highlight> registered. Please use <symbol>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 <highlight><symbol>mrelay rcmd {self.character_service.get_char_name(int(bot))} &lt;Your Prefix&gt;</highlight> 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"<highlight>{self.character_service.get_char_name(int(bot))}</highlight>: <highlight>{relay_command}</highlight>"
@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 <highlight>{bot.name}</highlight> does not exist."
if str(bot.char_id) not in bots.keys():
return f"There's no relay for <highlight>{bot.name}</highlight> registered. Please use <symbol>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 <highlight><symbol>mrelay prefix {self.character_service.get_char_name(int(bot))} &lt;Your Prefix&gt;</highlight> 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"<highlight>{self.character_service.get_char_name(int(bot))}</highlight>: <highlight>{prefix}</highlight>"
@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 <highlight>{bot.name}</highlight> does not exist."
if str(bot.char_id) not in bots.keys():
return f"There's no relay for <highlight>{bot.name}</highlight> registered. Please use <symbol>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"<highlight>{self.character_service.get_char_name(int(bot))}</highlight>: <highlight>{abbreviation}</highlight>"
@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 <highlight>{bot.name}</highlight> does not exist."
if str(bot.char_id) not in bots.keys():
return f"There's no relay for <highlight>{bot.name}</highlight> registered. Please use <symbol>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"<highlight>{self.character_service.get_char_name(int(bot))}</highlight> to: <highlight>{option}</highlight>"
@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 <highlight>{bot.name}</highlight> does not exist."
if str(bot.char_id) not in bots.keys():
return f"There's no relay for <highlight>{bot.name}</highlight> registered. Please use <symbol>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 <highlight>{option}</highlight> color to <font color={color}>{color}</font>" \
f" for relay <highlight>{self.character_service.get_char_name(int(bot))}</highlight>"
def color_template(self, bot, option, current):
def display(color, msg):
return f"<font color={color}>{msg}</font> (<a href='chatcmd:///tell <myname> mrelay color {bot} {option} {color}'>Save it</a>)"
blob = f"Current color: <font color={current.get(bot, current.get('default', '#00ff00'))}>" \
f"{current.get(bot, current.get('default', '#00ff00'))}</font>\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 = 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}")
self.message_hub_service.send_message(self.MESSAGE_SOURCE, None, 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.relay_guild_abbreviations().get_value().get("default"))
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"<font color={color}>{message}</font>"