Minidodo
9f1da9a00d
Changed the setting registration, removed the warnings. Loot roll messages are more obvious now. Superadmins are meant to stay mostily hidden, but are being exposed in !system again.
229 lines
13 KiB
Python
229 lines
13 KiB
Python
import math
|
|
|
|
from core.aochat.BaseModule import BaseModule
|
|
from core.aochat.server_packets import BuddyAdded
|
|
from core.buddy_service import BuddyService
|
|
from core.chat_blob import ChatBlob
|
|
from core.command_param_types import Character, Options, Const, Any, NamedParameters
|
|
from core.db import DB
|
|
from core.decorators import instance, command, event
|
|
from core.dict_object import DictObject
|
|
from core.event_service import EventService
|
|
from core.job_scheduler import JobScheduler
|
|
from core.lookup.pork_service import PorkService
|
|
from core.message_hub_service import MessageHubService
|
|
from core.private_channel_service import PrivateChannelService
|
|
from core.setting_service import SettingService
|
|
from core.setting_types import TextSettingType, ColorSettingType
|
|
from core.text import Text
|
|
from core.igncore import Tyrbot
|
|
from core.util import Util
|
|
from modules.core.accounting.services.account_service import AccountService
|
|
from modules.raidbot.tower.tower_controller import TowerController
|
|
|
|
|
|
# noinspection DuplicatedCode,SqlCaseVsIf
|
|
@instance()
|
|
class TrackController(BaseModule):
|
|
PRIVATE_CHANNEL_PREFIX = "[Priv]"
|
|
PAGE_SIZE = 20
|
|
MESSAGE_SOURCE = "track_log"
|
|
|
|
def inject(self, registry):
|
|
self.bot: Tyrbot = registry.get_instance("bot")
|
|
self.util: Util = registry.get_instance("util")
|
|
self.pork: PorkService = registry.get_instance("pork_service")
|
|
self.job_scheduler: JobScheduler = registry.get_instance("job_scheduler")
|
|
self.text: Text = registry.get_instance("text")
|
|
self.event_service: EventService = registry.get_instance("event_service")
|
|
self.setting_service: SettingService = registry.get_instance("setting_service")
|
|
self.buddy_service: BuddyService = registry.get_instance("buddy_service")
|
|
self.account_service: AccountService = registry.get_instance("account_service")
|
|
self.db: DB = registry.get_instance("db")
|
|
self.priv: PrivateChannelService = registry.get_instance("private_channel_service")
|
|
self.tower: TowerController = registry.get_instance("tower_controller")
|
|
self.message_hub_service: MessageHubService = registry.get_instance("message_hub_service")
|
|
|
|
def pre_start(self):
|
|
self.event_service.register_event_type("track_logon")
|
|
self.event_service.register_event_type("track_logoff")
|
|
self.setting_service.register(self.module_name, "track_on_color", "#FF0000", ColorSettingType(),
|
|
"Color for Track logon")
|
|
self.setting_service.register(self.module_name, "track_off_color", "#00FF00", ColorSettingType(),
|
|
"Color for Track logoff")
|
|
self.setting_service.register(self.module_name, "autotrack", 'none',
|
|
TextSettingType(['omni', 'clan', 'neutral', "none"]),
|
|
"Autotrack all players initiating tower attacks towards this faction:")
|
|
self.db.exec(
|
|
"CREATE TABLE IF NOT EXISTS track("
|
|
"char_id int not null primary key, "
|
|
"initiator int not null, "
|
|
"reason varchar(255))")
|
|
self.message_hub_service.register_message_source(self.MESSAGE_SOURCE)
|
|
|
|
@event(event_type="connect", description="Autoadd tracked players on connect", is_hidden=True)
|
|
def connect_add(self, _1, _2):
|
|
query = self.db.query("SELECT * from track")
|
|
for user in query:
|
|
self.buddy_service.add_buddy(user.char_id, "track")
|
|
|
|
@event(event_type="buddy_logon", description="Fire tracker events", is_hidden=True)
|
|
def track_fire_logon(self, _, event_data):
|
|
if self.bot.is_ready():
|
|
if "track" in (self.buddy_service.get_buddy(event_data.char_id) or {'types': []})["types"]:
|
|
if type(event_data) == BuddyAdded:
|
|
self.event_service.fire_event("track_logon",
|
|
self.db.query_single("SELECT * from player where char_id=?",
|
|
[event_data.char_id]))
|
|
self.db.exec("INSERT IGNORE INTO online VALUES(?, ?, ?)",
|
|
[event_data.char_id, 'track', self.bot.get_char_id()])
|
|
else:
|
|
self.job_scheduler.delayed_job(self.track_fire_logon, 10,
|
|
DictObject({'char_id': event_data.char_id, 'repeat': True}))
|
|
|
|
@event(event_type="buddy_logoff", description="Fire tracker events", is_hidden=True)
|
|
def track_fire_logoff(self, _1, event_data):
|
|
if buddy := self.buddy_service.get_buddy(event_data.char_id):
|
|
if "track" in buddy["types"]:
|
|
self.event_service.fire_event("track_logoff", self.db.query_single("SELECT * from player where char_id=?",
|
|
[event_data.char_id]))
|
|
self.db.exec("DELETE FROM online where char_id=? and bot=?", [event_data.char_id, self.bot.get_char_id()])
|
|
|
|
@event(event_type="track_logon", description="Fire tracker events", is_hidden=True)
|
|
def track_logon(self, _1, user):
|
|
if self.bot.is_ready():
|
|
color = self.setting_service.get("track_on_color").format_text("ON")
|
|
self.send_t_warn(0, f'{color} :: {self.text.format_char_info(user)}')
|
|
|
|
@event(event_type="track_logoff", description="Fire tracker events", is_hidden=True)
|
|
def track_logoff(self, _1, user):
|
|
if self.bot.is_ready():
|
|
color = self.setting_service.get("track_off_color").format_text("OFF")
|
|
self.send_t_warn(0, f'{color} :: {self.text.format_char_info(user)}')
|
|
|
|
@event(event_type=TowerController.TOWER_ATTACK_EVENT, description="Warn on Tower attacks", is_hidden=True)
|
|
def tower_attack_event(self, _, event_data):
|
|
attacker = event_data.attacker
|
|
if event_data.defender.faction.lower() == self.setting_service.get_value("autotrack"):
|
|
if not self.buddy_service.get_buddy(attacker.char_id):
|
|
if self.buddy_service.add_buddy(attacker.char_id, "track"):
|
|
self.send_t_warn(0,
|
|
f"Now tracking: "
|
|
f"<{attacker.faction.lower()}>{attacker.name}</{attacker.faction.lower()}>")
|
|
self.db.exec("INSERT IGNORE INTO track VALUES(?, ?, ?)",
|
|
[attacker.char_id, self.bot.get_char_id(), "attacked us"])
|
|
|
|
def send_t_warn(self, _, msg):
|
|
self.message_hub_service.send_message(self.MESSAGE_SOURCE, None, "[T] " + msg, "[<cyan>T</cyan>] " + msg)
|
|
|
|
def get_tracked(self, order, online):
|
|
where = "where o.char_id IS NOT NULL" if online else ""
|
|
return self.db.query(f"SELECT p.name, p.level, p.ai_level, p.org_name, p.profession, p.faction, "
|
|
f"p2.name as initiator, "
|
|
f"CASE WHEN o.char_id IS NOT NULL THEN 1 ELSE 0 END AS online, "
|
|
f"t.reason from track t "
|
|
f"left join player p on p.char_id = t.char_id "
|
|
f"left join player p2 on p2.char_id = t.initiator "
|
|
f"left join online o on o.char_id = t.char_id {where} group by t.char_id order by {order}")
|
|
|
|
@command(command="track", params=[Const('add'), Character("character"), Any('reason')], access_level="leader",
|
|
description="Initiate tracking for the character", sub_command="add")
|
|
def track_add(self, request, _, user, reason):
|
|
if not user.char_id:
|
|
return f"Character <highlight>{user.name}</highlight> not found."
|
|
self.pork.load_character_info(user.char_id, user.name)
|
|
if acc := self.account_service.get_account(request.sender.char_id):
|
|
if "track" in (self.buddy_service.get_buddy(user.char_id) or {'types': []})["types"]:
|
|
return f"Character <highlight>{user.name}</highlight> is already being tracked."
|
|
else:
|
|
self.pork.load_character_info(user.char_id, user.name)
|
|
self.send_t_warn(0, f'Tracking initiated: {user.name} by {acc.name}')
|
|
self.buddy_service.add_buddy(user.char_id, 'track')
|
|
self.db.exec("INSERT IGNORE INTO track VALUES(?, ?, ?)",
|
|
[user.char_id, acc.main, reason])
|
|
|
|
@command(command="track", params=[Const('rem'), Character("character")], access_level="moderator",
|
|
description="Remove character from the tracking", sub_command="rem")
|
|
def track_rem(self, request, _, user):
|
|
if not user.char_id:
|
|
return f"Character <highlight>{user.name}</highlight> not found."
|
|
if acc := self.account_service.get_account(request.sender.char_id):
|
|
if self.buddy_service.remove_buddy(user.char_id, "track"):
|
|
self.db.exec("DELETE FROM track where char_id=?", [user.char_id])
|
|
self.send_t_warn(0, f'Tracking stopped: {user.name} by {acc.name}')
|
|
return f"Character <highlight>{user.name}</highlight> is nolonger being tracked."
|
|
else:
|
|
return f"Character <highlight>{user.name}</highlight> is not being tracked."
|
|
|
|
@command(command="track",
|
|
params=[Const('list', is_optional=True), Options(["org", "prof", "tl"], is_optional=True),
|
|
NamedParameters(["page"])],
|
|
access_level="member",
|
|
description="Shows tracked players")
|
|
def track_list(self, _, const, group, named_params):
|
|
page = int(named_params.page or "1")
|
|
offset = (page - 1) * self.PAGE_SIZE
|
|
if group is None:
|
|
group = 'tl'
|
|
if group == "org":
|
|
players = self.get_tracked('p.org_name, p.profession, p.level desc, p.name', False if const else True)
|
|
return self.format_page(players, "org", offset, page, f"Tracklist by Organisation ({len(players)})",
|
|
'No tracked users online', f"track {const or ''} {group or ''}")
|
|
|
|
elif group == "prof":
|
|
players = self.get_tracked('p.profession, p.level desc, p.name', False if const else True)
|
|
return self.format_page(players, "prof", offset, page, f"Tracklist by Profession ({len(players)})",
|
|
'No tracked users online', f"track {const or ''} {group or ''}")
|
|
elif group == "tl":
|
|
players = self.get_tracked('p.level desc, p.profession, p.name', False if const else True)
|
|
return self.format_page(players, "tl", offset, page, f"Tracklist by Titlelevel ({len(players)})",
|
|
'No tracked users online', f"track {const or ''} {group or ''}")
|
|
|
|
def format_row(self, user):
|
|
org = f"[<{user.faction.lower()}>{user.org_name}</{user.faction.lower()}>] " if user.org_name else ""
|
|
return f"<tab>{self.util.get_prof_icon(user.profession)} " \
|
|
f"{self.text.zfill(user.level, 220)}:<green>{self.text.zfill(user.ai_level, 220)}</green> " \
|
|
f"<{user.faction.lower()}>{user.name}</{user.faction.lower()}>" \
|
|
f" {org}init by <highlight>{user.initiator}</highlight>: {user.reason} " \
|
|
f"{'[<red>ONLINE</red>]' if user.online == 1 else ''}\n"
|
|
|
|
def format_page(self, tracked, order, offset, page, title, nullmsg, cmd):
|
|
selected = tracked[offset:offset + self.PAGE_SIZE]
|
|
count = len(selected)
|
|
pages = ""
|
|
if page > 1:
|
|
pages += "Pages: " + self.text.make_tellcmd("«« Page %d" % (page - 1), f'{cmd} --page={page - 1}')
|
|
if offset + self.PAGE_SIZE < len(tracked):
|
|
pages += f" Page {page}/{math.ceil(len(tracked) / self.PAGE_SIZE)}"
|
|
pages += " " + self.text.make_tellcmd("Page %d »»" % (page + 1), f'{cmd} --page={page + 1}')
|
|
pages += "\n"
|
|
if count == 0:
|
|
return nullmsg
|
|
else:
|
|
blob = "\n\n" + pages + "<font color=CCInfoText>"
|
|
if order == "prof":
|
|
prof = ""
|
|
for player in selected:
|
|
if player.profession != prof:
|
|
prof = player.profession
|
|
blob += f"\n<pagebreak><header>{player.profession}</header>\n"
|
|
blob += self.format_row(player)
|
|
elif order == "tl":
|
|
tl = ""
|
|
for player in selected:
|
|
titlelevel = self.util.get_title_level(player.level)
|
|
if titlelevel != tl:
|
|
tl = titlelevel
|
|
blob += f"\n<pagebreak><header>TL{titlelevel}</header>\n"
|
|
blob += self.format_row(player)
|
|
elif order == "org":
|
|
org_name = ""
|
|
for player in selected:
|
|
if player.org_name != org_name:
|
|
org_name = player.org_name
|
|
blob += f"\n<pagebreak><header>{player.org_name}</header>\n"
|
|
blob += self.format_row(player)
|
|
|
|
blob += "</font>\n" + pages
|
|
return ChatBlob(title, blob)
|