import re from core.aochat import server_packets from core.conn import Conn from core.db import DB from core.decorators import instance, event from core.dict_object import DictObject from core.event_service import EventService from core.igncore import IgnCore from core.logger import Logger from core.lookup.pork_service import PorkService from core.public_channel_service import PublicChannelService from core.text import Text from modules.standard.helpbot.playfield_controller import PlayfieldController @instance() class TowerEventController: TOWER_ATTACK_EVENT = "tower_attack" TOWER_VICTORY_EVENT = "tower_victory" TOWER_BATTLE_OUTCOME_ID = 42949672962 ALL_TOWERS_ID = 42949672960 # The %s organization %s just entered a state of war! # %s attacked the %s organization %s's tower in %s at location (%d,%d). ATTACK_1 = [506, 12753364] ATTACK_2 = re.compile(r"^(.+) just attacked the (clan|neutral|omni) organization (.+)'s tower in (.+) " r"at location \((\d+), (\d+)\).\n$") VICTORY_1 = re.compile(r"^Notum Wars Update: Victory to the (Clan|Neutral|Omni)s!!!$") VICTORY_2 = re.compile(r"^The (Clan|Neutral|Omni) organization (.+) attacked the (Clan|Neutral|Omni) (.+) " r"at their base in (.+). The attackers won!!$") VICTORY_3 = [506, 147506468] # 'Notum Wars Update: The %s organization %s lost their base in %s.' def __init__(self): self.logger = Logger(__name__) def inject(self, registry): self.bot: IgnCore = registry.get_instance("bot") self.db: DB = registry.get_instance("db") self.text: Text = registry.get_instance("text") self.event_service: EventService = registry.get_instance("event_service") self.pork_service: PorkService = registry.get_instance("pork_service") self.playfield_controller: PlayfieldController = registry.get_instance("playfield_controller") self.public_channel_service: PublicChannelService = registry.get_instance("public_channel_service") def pre_start(self): self.event_service.register_event_type(self.TOWER_ATTACK_EVENT) self.event_service.register_event_type(self.TOWER_VICTORY_EVENT) self.bot.register_packet_handler(server_packets.PublicChannelMessage.id, self.handle_public_channel_message) self.db.load_sql_file(self.module_dir + "/" + "tower_sites.sql", pre_optimized=True) self.db.create_view("tower_sites") @event(event_type="connect", description="Check if All Towers channel is available") def handle_connect_event(self, _, _1): if self.public_channel_service.org_id and not self.public_channel_service.get_channel_id("All Towers"): self.logger.warning("This bot is a member of an org but does not have access to 'All Towers' channel and " "therefore will not receive tower attack messages") def handle_public_channel_message(self, conn: Conn, packet: server_packets.PublicChannelMessage): if conn.id != "main": return if packet.channel_id == self.TOWER_BATTLE_OUTCOME_ID: victory = self.get_victory_event(packet) if victory: # self.logger.info("tower victory packet: %s" % str(packet)) # lookup playfield playfield_name = victory.location.playfield.long_name victory.location.playfield = self.playfield_controller.get_playfield_by_name(playfield_name) or \ DictObject() victory.location.playfield.long_name = playfield_name # print("VICTORY", victory) # VICTORY {'type': 'attack', 'winner': {'faction': 'Clan', 'org_name': 'Komodites'}, 'loser': {'faction': 'Neutral', 'org_name': 'Deez Neuts'}, 'location': {'playfield': {'id': 791, 'long_name': 'Holes in the Wall', 'short_name': 'HITW', 'dungeon': 0}}} self.event_service.fire_event(self.TOWER_VICTORY_EVENT, victory) elif packet.channel_id == self.ALL_TOWERS_ID: attack = self.get_attack_event(packet) if attack: # self.logger.info("tower attack packet: %s" % str(packet)) # ServerPacket(65): [42949672960, 0, "Kletka just attacked the clan organization Lex Noctis's tower in Athen Shire at location (2606, 2022).\n", ''], ExtendedMessage: None # lookup playfield playfield_name = attack.location.playfield.long_name attack.location.playfield = self.playfield_controller.get_playfield_by_name(playfield_name) or \ DictObject() attack.location.playfield.long_name = playfield_name # lookup attacker name = attack.attacker.name faction = attack.attacker.faction org_name = attack.attacker.org_name char_info = self.pork_service.get_character_info(name) print(attack, name, faction, org_name) print(char_info) attack.attacker = char_info or DictObject() attack.attacker.name = name attack.attacker.faction = faction or attack.attacker.get("faction", "Unknown") attack.attacker.org_name = org_name if char_info: print("ATTK", attack) self.event_service.fire_event(self.TOWER_ATTACK_EVENT, attack) else: print("WARNING", attack) def get_attack_event(self, packet: server_packets.PublicChannelMessage): # The %s organization %s just entered a state of war! # %s attacked the %s organization %s's tower in %s at location (%d,%d). ATTACK_1 = [506, 12753364] if packet.extended_message and \ [packet.extended_message.category_id, packet.extended_message.instance_id] == self.ATTACK_1: params = packet.extended_message.params return DictObject({ "attacker": { "name": params[2], "faction": params[0].capitalize(), "org_name": params[1] }, "defender": { "faction": params[3].capitalize(), "org_name": params[4] }, "location": { "playfield": { "long_name": params[5] }, "x_coord": params[6], "y_coord": params[7] } }) else: ATTACK_2 = re.compile(r"^(.+) just attacked the (clan|neutral|omni) organization (.+)'s tower in (.+) " r"at location \((\d+), (\d+)\).\n$") match = self.ATTACK_2.match(packet.message) if match: return DictObject({ "attacker": { "name": match.group(1), "faction": "", "org_name": "" }, "defender": { "faction": match.group(2).capitalize(), "org_name": match.group(3) }, "location": { "playfield": { "long_name": match.group(4) }, "x_coord": match.group(5), "y_coord": match.group(6) } }) # Unknown attack self.logger.warning("Unknown tower attack: " + str(packet)) return None def get_victory_event(self, packet: server_packets.PublicChannelMessage): # Does not contain any relevant data. match = self.VICTORY_1.match(packet.message) if match: return None VICTORY_2 = re.compile(r"^The (Clan|Neutral|Omni) organization (.+) attacked the (Clan|Neutral|Omni) (.+) " r"at their base in (.+). The attackers won!!$") match = self.VICTORY_2.match(packet.message) if match: return DictObject({ "type": "attack", "winner": { "faction": match.group(1).capitalize(), "org_name": match.group(2) }, "loser": { "faction": match.group(3).capitalize(), "org_name": match.group(4) }, "location": { "playfield": { "long_name": match.group(5) } } }) if packet.extended_message and \ [packet.extended_message.category_id, packet.extended_message.instance_id] == self.VICTORY_3: params = packet.extended_message.params return DictObject({ "type": "terminated", "winner": { "faction": "", # params[0].capitalize(), "org_name": "", # params[1] }, "loser": { "faction": params[0].capitalize(), "org_name": params[1] }, "location": { "playfield": { "long_name": params[2] } } }) # Unknown victory self.logger.warning("Unknown tower victory: " + str(packet)) return None