Files
igncore/modules/onlinebot/raids/raid_controller.py
T

285 lines
15 KiB
Python

import textwrap
import time
from datetime import datetime, timezone
from core import command_request
from core.buddy_service import BuddyService
from core.command_alias_service import CommandAliasService
from core.command_param_types import Any, Int, Const
from core.db import DB
from core.decorators import instance, command, timerevent
from core.logger import Logger
from core.lookup.pork_service import PorkService
from core.private_channel_service import PrivateChannelService
from core.setting_service import SettingService
from core.text import Text
from core.igncore import IgnCore
from core.util import Util
from modules.core.accounting.services.account_service import AccountService
from modules.core.ban.ban_service import BanService
# noinspection SqlResolve
class Raid:
leader = {}
raid_desc = "[No description set]"
raid_time = "[unknown]"
min_level = 200
leader_rank = {}
last_spam = 0
last_action = 0
bot = ""
def __init__(self, leader, rank):
self.leader = leader
self.leader_rank = rank
self.last_action = time.time()
# noinspection SqlResolve,DuplicatedCode,PyAttributeOutsideInit
@instance()
class RaidController:
profs = {"Meta-Physicist": 16308, "Adventurer": 84203, "Engineer": 16252, "Soldier": 16237, "Keeper": 84197,
"Shade": 39290, "Fixer": 16300, "Agent": 16186, "Trader": 117993, "Doctor": 44235, "Enforcer": 100998,
"Bureaucrat": 16341, "Martial Artist": 16196, "Nano-Technician": 16283}
legal_bots = ["allianceraid", "aapf", "theallianz", "igncom", "ignraid"]
def __init__(self):
# noinspection PyTypeChecker
self.raid: Raid = None
def inject(self, registry):
self.logger = Logger(__name__)
self.bot: IgnCore = registry.get_instance("bot")
self.db: DB = registry.get_instance("db")
self.text: Text = registry.get_instance("text")
self.util: Util = registry.get_instance("util")
self.pork: PorkService = registry.get_instance("pork_service")
self.org_pork: PorkService = registry.get_instance("org_pork_service")
self.command_alias_service: CommandAliasService = registry.get_instance("command_alias_service")
self.buddy_service: BuddyService = registry.get_instance("buddy_service")
self.setting_service: SettingService = registry.get_instance("setting_service")
self.ban: BanService = registry.get_instance("ban_service")
self.private_channel_service: PrivateChannelService = registry.get_instance("private_channel_service")
self.account_service: AccountService = registry.get_instance("account_service")
@command(command="raid", params=[Const("desc"), Any("<description>")],
description="Change the title of the Raid", access_level="leader")
def raid_set_name(self, request: command_request, _, desc):
if not self.raid:
return "There's no raid running."
if request.sender.access_level["level"] <= self.raid.leader_rank:
self.raid.raid_desc = desc
self.raid.last_action = time.time()
return f"Successfully changed the raid description to <highlight>{desc}</highlight>"
return f"Error! You cant do that. Your accesslevel must be equal or higher than " \
f"<highlight>{self.raid.leader.name}</highlight>'s rank."
@command(command="raid", params=[Const("bot"), Any("&lt;botname&gt;")],
description="Change the master Raidbot", access_level="leader")
def raid_set_bot(self, request: command_request, _, bot):
if not self.raid:
return "There's no raid running."
if request.sender.access_level["level"] <= self.raid.leader_rank:
if bot in self.legal_bots:
self.raid.bot = bot
self.raid.last_action = time.time()
return f"Successfully changed the raidbot to <highlight>{bot}</highlight>"
else:
return f"The bot <highlight>{bot}</highlight> is not whitelisted, you cannot use it as a raidbot. "
return f"Error! You cant do that. Your accesslevel must be equal or higher than " \
f"<highlight>{self.raid.leader.name}</highlight>'s rank."
@command(command="raid", params=[Const("level"), Int("&lt;min_level&gt;")],
description="Change the minimal level of the Raid", access_level="leader")
def raid_set_level(self, request: command_request, _, level):
if not self.raid:
return "There's no raid running."
if request.sender.access_level["level"] <= self.raid.leader_rank:
self.raid.min_level = level
self.raid.last_action = time.time()
return f"Successfully changed the minimum raid level to <highlight>{level}</highlight>"
return f"Error! You cant do that. Your accesslevel must be equal or higher than " \
f"<highlight>{self.raid.leader.name}</highlight>'s rank."
@command(command="raid", params=[Const("time"), Any("&lt;stepping_time&gt;")],
description="Change the stepping time of the Raid", access_level="leader")
def raid_set_time(self, request: command_request, _, step):
if not self.raid:
return "There's no raid running."
if request.sender.access_level["level"] <= self.raid.leader_rank:
self.raid.raid_time = step
self.raid.last_action = time.time()
return f"Successfully changed the stepping time to <highlight>{step}</highlight>"
return f"Error! You cant do that. Your accesslevel must be equal or higher than " \
f"<highlight>{self.raid.leader.name}</highlight>'s rank."
@command(command="raid", params=[Const("start")], description="Start a raid", access_level="leader")
def raid_start(self, request: command_request, _):
if self.raid:
return "There's already a raid running."
self.raid = Raid(request.sender, request.sender.access_level["level"])
# after I saw myself sending raid invites with the wrong bot reference by accident atleast once,
# the default is being set to allianceraid again.
self.raid.bot = "allianceraid" # self.bot.get_char_name()
self.raid.last_action = time.time()
spam = f"""
________________
<yellow>
You started a raid... but I need some more details about it,
please fill in this form: {self.raid_settings(request, "")}
</yellow>
________________
"""
return spam
@command(command="raid", params=[Const("stop")], description="Stop the raid", access_level="leader")
def raid_stop(self, _, _1):
if not self.raid:
return "There's no raid running."
self.raid = None
return "You stopped the raid."
# noinspection LongLine
@command(command="raid", params=[Const("invite")], description="Send the Spam to the defined Audience",
access_level="leader")
def send_tha_spam(self, request, _):
if not self.raid:
return "There's no raid running."
def blob(label, msg):
return f"<a href=\"text://{textwrap.dedent(msg)}\">{textwrap.dedent(label)}</a>"
last = time.time()
if self.raid.last_spam + 5 * 60 > last:
return f"There was an invite in the last 5 minutes; please wait " \
f"<highlight>{self.util.time_to_readable(self.raid.last_spam + 5 * 60 - last)}</highlight> " \
f"before sending another one."
click_here = blob("click here",
f"""
<header>How to join the Raid</header>
<notice>Step 1:</notice>
- Please join {self.raid.bot}: [{self.text.make_chatcmd("Join", f"/tell {self.raid.bot} join")}]
<notice>Step 2:</notice>
- Please go LFT, to show your interest: [{self.text.make_chatcmd("Go LFT", f"/lft » <green>{self.raid.bot}</green>")}]
<notice>Step 3:</notice>
Wait at the raid starting location.
Stepping is scheduled for <highlight>{self.raid.raid_time}</highlight>
Currently it is <highlight>{(datetime.now(timezone.utc).strftime("%H:%M"))} GMT-0</highlight>
""")
spam = f"""
________________
<yellow><green>{request.sender.name}</green> has invited you to join a <notice>{self.raid.raid_desc}</notice> Raid
Type <notice>/tell {self.raid.bot} join</notice> to participate
or {click_here}</yellow>
________________
"""
subtile = f"""<yellow>[<green>{request.sender.name}</green>]: Raid Starting: <notice>{self.raid.raid_desc}</notice> -- use <notice>/tell {self.raid.bot} join</notice> to participate or {click_here}</yellow></yellow> """
# noinspection SqlAggregates
players = self.db.query("SELECT p.*, a.subtile_spam, a.raid_invite, a.raid_spam, a.main, a.disabled, a.member from online o "
"left join player p on o.char_id = p.char_id "
"left join account a on a.char_id=(select main from account where char_id=o.char_id) "
"where p.level >= ? AND ((a.raid_invite=1 or a.raid_spam = 1) "
"and a.disabled = 0 "
"and a.member != -1 "
"AND a.char_id NOT IN (SELECT char_id FROM org_bots) "
"AND o.char_id NOT IN (SELECT char_id FROM org_bots)) "
"AND o.bot=? group by o.char_id "
"ORDER BY p.profession, p.name, p.level, p.org_rank_id;",
[self.raid.min_level, self.bot.get_char_id()])
self.bot.send_mass_message(request.sender.char_id, f"Attempting to send {len(players)} invites...")
info = "Sent invites to:"
prof = ""
count = len(players)
for i in players:
if not self.account_service.simple_checks(i):
count -= 1
continue
if prof != i.profession:
info += "\n\n<img src=tdb://id:GFX_GUI_FRIENDLIST_SPLITTER>"
info += "\n" + f"<img src=rdb://{self.profs.get(i.profession)}><tab><green>{i.profession}<end>\n"
info += "<img src=tdb://id:GFX_GUI_FRIENDLIST_SPLITTER>"
prof = i.profession
if i.raid_spam == 1:
self.bot.send_mass_message(i.char_id, spam if i.subtile_spam == 0 else subtile)
# if i.raid_invite == 1:
# if self.raid.bot == self.bot.get_char_name():
# if not self.private_channel_service.in_private_channel(i.char_id):
# self.private_channel_service.invite(i.char_id)
info += f"\n - {i.name: <13} ({i.level: >3}/<green>{i.ai_level: >2}</green>) - {i.org_name}"
self.raid.last_action = time.time()
self.raid.last_spam = time.time()
self.bot.send_mass_message(request.sender.char_id,
f"Successfully sent {count} invite{'s' if count != 1 else ''}! [{self.text.format_page('More', info)}]")
@command(command="raid", params=[Const("settings")],
description="Change the title of the Raid", access_level="leader")
def raid_settings(self, request: command_request, _):
if not self.raid:
return "There's no raid running."
if request.sender.access_level["level"] <= self.raid.leader_rank:
level_rest = " -"
for i in [0, 200, 205, 210, 215, 220]:
level_rest += f" [{self.text.make_chatcmd(i, '/tell <myname> raid level ' + str(i))}]"
bots = " -"
for i in ["allianceraid", "aapf", "theallianz"]:
bots += f" [{self.text.make_chatcmd(i, '/tell <myname> raid bot ' + str(i))}]"
description = " -"
for i in ["S42", "Pandemonium", "APF's"]:
description += f" [{self.text.make_chatcmd(i, '/tell <myname> raid desc ' + str(i))}]"
step = " -"
count = 0
for i in ["19:00 GMT-0", "19:30 GMT-0", "20:00 GMT-0", "20:30 GMT-0", "21:00 GMT-0", "21:30 GMT-0"]:
if count == 2:
count = 0
step += "<br> -"
step += f" [{self.text.make_chatcmd(i, '/tell <myname> raid time ' + str(i))}]"
count += 1
# noinspection LongLine
settings = self.text.format_page("Raid Settings",
textwrap.dedent(
f"""
<header>Raid Settings</header>
<notice>Set the minimum level [<yellow>{self.raid.min_level}</yellow>]:</notice>
{level_rest}
<notice>Change the Description [<yellow>{self.raid.raid_desc}</yellow>]:</notice>
{description}
<notice>Change the Raidbot [<yellow>{self.raid.bot}</yellow>]:</notice>
{bots}
<notice>Change the raid stepping time [<yellow>{self.raid.raid_time}</yellow>]:</notice>
-> Current Time: {(datetime.now(timezone.utc).strftime("%H:%M"))} GMT-0
{step}
<notice>Send the invites!</notice>
[{self.text.make_chatcmd("send the spam!", "/tell <myname> raid invite")}]
<notice>End the raid</notice>
[{self.text.make_chatcmd("stop it", "/tell <myname> raid stop")}]
"""))
return settings
return f"Error! You cant do that. Your accesslevel must be equal or higher than " \
f"<highlight>{self.raid.leader.name}</highlight>'s rank."
@timerevent(budatime="15m", description="Stop raid if inactive")
def clear_raid(self, _, _1):
if self.raid:
if self.raid.last_action + 60 * 60 < time.time():
self.bot.send_mass_message(self.raid.leader.char_id,
f"There was no interaction for more then 60 minutes for your Raid "
f"<highlight>{self.raid.raid_desc}</highlight>; I'll stop it for you.")
self.raid = None