import time from core.chat_blob import ChatBlob from core.command_param_types import Const, Int, NamedParameters, Any from core.decorators import instance, command from core.igncore import IgnCore from core.logger import Logger from core.text import Text from core.util import Util from modules.standard.helpbot.playfield_controller import PlayfieldController @instance() class TowerController: def __init__(self): self.logger = Logger(__name__) def inject(self, registry): self.bot: IgnCore = registry.get_instance("bot") self.db = registry.get_instance("db") self.text: Text = registry.get_instance("text") self.util: Util = registry.get_instance("util") self.tower: TowerController = registry.get_instance("tower_controller") self.playfield_controller: PlayfieldController = registry.get_instance("playfield_controller") self.public_channel_service = registry.get_instance("public_channel_service") @command(command="attacks", params=[Const("battle"), Int("battle_id")], access_level="member", description="Show battle info for a specific battle") def attacks_battle_cmd(self, _, _1, battle_id): battle = self.db.query_single( "SELECT b.*, p.short_name FROM tower_battle b " "LEFT JOIN playfields p ON p.id = b.playfield_id WHERE b.id = ?", [battle_id]) if not battle: return "Could not find battle with ID %d." % battle_id t = int(time.time()) attackers = self.db.query("SELECT * FROM tower_attacker WHERE tower_battle_id = ? ORDER BY created_at DESC", [battle_id]) first_activity = attackers[-1].created_at if len(attackers) > 0 else battle.last_updated blob = "" blob += self.format_battle_info(battle, t) blob += f"Duration: " \ f"{self.util.time_to_readable(battle.last_updated - first_activity)}\n\n" blob += "Attackers:\n" for row in attackers: blob += "" + self.format_attacker(row) blob += " " + self.format_timestamp(row.created_at, t) blob += "\n" return ChatBlob(f"Battle Info {battle_id}", blob) @command(command="attacks", params=[NamedParameters(["page"])], access_level="member", description="Show recent tower attacks and victories") def attacks_cmd(self, _, named_params): page = int(named_params.page or "1") page_size = 30 offset = (page - 1) * page_size data = self.get_recent_attacks(offset, page_size) t = int(time.time()) return self.display(page, data, time.time()) @command(command="attacks", params=[Any("playfield"), Any("site_number", is_optional=True), NamedParameters(["page"])], access_level="member", description="Show recent tower attacks and victories") def cmd_attacks_pf_site(self, _, pf, site, named_params): page = int(named_params.page or "1") page_size = 30 offset = (page - 1) * page_size playfield = self.playfield_controller.get_playfield_by_name_or_id(pf) if not playfield: return f"Could not find Playfield {pf}." pf = playfield.id data = self.get_recent_attacks_by_lca(offset, page_size, pf, site) blob = self.display(page, data, time.time()) if site: blob.page_postfix = f" in {playfield.short_name} on x{site}" else: blob.page_postfix = f" in {playfield.short_name}" return blob def display(self, page, data, t): blob = "" if page > 1: blob += " " + self.text.make_chatcmd(f"<< Page {page - 1:d}", self.get_chat_command(page - 1)) if len(data) > 0: blob += " Page " + str(page) blob += " " + self.text.make_chatcmd(f"Page {page + 1:d} >>", self.get_chat_command(page + 1)) blob += "\n" current_battle_id = -1 for row in data: if current_battle_id != row.battle_id: blob += "\n" current_battle_id = row.battle_id blob += self.format_battle_info(row, t) blob += self.text.make_tellcmd("More Info", f"attacks battle {row.battle_id:d}") + "\n" blob += "Attackers:\n" blob += "" + self.format_attacker(row) + "\n" blob = ChatBlob(f"Tower Attacks", blob) return blob def format_attacker(self, row): level = f"{row.att_level}/{row.att_ai_level}" if row.att_ai_level > 0 else f"{row.att_level}" org = row.att_org_name + " " if row.att_org_name else "" victor = " - Winner!" if row.is_victory else "" return f"{row.att_char_name or 'Unknown attacker'} ({level} {row.att_profession})" \ f" {org}({row.att_faction}){victor}" def format_battle_info(self, row, t): blob = "" defeated = " - Defeated!" if row.is_finished else "" blob += f"Site: {row.short_name} {row.site_number or '?'}\n" blob += f"Defender: {row.def_org_name} ({row.def_faction}){defeated}\n" blob += f"Last Activity: {self.format_timestamp(row.last_updated, t)}\n" return blob def format_timestamp(self, t, current_t): return f"{self.util.format_datetime(t)} " \ f"({self.util.time_to_readable(current_t - t)} ago)" def get_chat_command(self, page): return f"/tell attacks --page={page}" def get_recent_attacks(self, offset, page_size): return self.db.query("SELECT b.*, a.*, " "COALESCE(a.att_level, 0) AS att_level, " "COALESCE(a.att_ai_level, 0) AS att_ai_level, " "p.short_name, " "b.id AS battle_id " "FROM tower_battle b " "LEFT JOIN tower_attacker a ON a.tower_battle_id = b.id " "LEFT JOIN playfields p ON p.id = b.playfield_id " "ORDER BY b.last_updated DESC, a.created_at DESC " "LIMIT ?, ?", [offset, page_size]) def get_recent_attacks_by_lca(self, offset, page_size, playfield, site_number=None): if not site_number: return self.db.query("SELECT b.*, a.*, " "COALESCE(a.att_level, 0) AS att_level, " "COALESCE(a.att_ai_level, 0) AS att_ai_level, " "p.short_name, " "b.id AS battle_id " "FROM tower_battle b " "LEFT JOIN tower_attacker a ON a.tower_battle_id = b.id " "LEFT JOIN playfields p ON p.id = b.playfield_id " "WHERE p.id =? ORDER BY b.last_updated DESC, a.created_at DESC " "LIMIT ?, ?", [playfield, offset, page_size]) else: return self.db.query("SELECT b.*, a.*, " "COALESCE(a.att_level, 0) AS att_level, " "COALESCE(a.att_ai_level, 0) AS att_ai_level, " "p.short_name, " "b.id AS battle_id " "FROM tower_battle b " "LEFT JOIN tower_attacker a ON a.tower_battle_id = b.id " "LEFT JOIN playfields p ON p.id = b.playfield_id " "WHERE p.id = ? AND b.site_number = ? ORDER BY b.last_updated DESC, a.created_at DESC " "LIMIT ?, ?", [playfield, site_number, offset, page_size])