123 lines
6.5 KiB
Python
123 lines
6.5 KiB
Python
import time
|
|
import typing
|
|
|
|
import hjson
|
|
|
|
from core.aochat.BaseModule import BaseModule
|
|
from core.chat_blob import ChatBlob
|
|
from core.command_param_types import Any, Const, Options, Time, Character
|
|
from core.db import DB
|
|
from core.decorators import instance, command, timerevent
|
|
|
|
from core.igncore import IgnCore
|
|
|
|
if typing.TYPE_CHECKING:
|
|
from modules.core.accounting.services.account_service import AccountService
|
|
from modules.core.ban.ban_service import BanService
|
|
from core.lookup.character_service import CharacterService
|
|
from core.translation_service import TranslationService
|
|
|
|
|
|
@instance()
|
|
class BanController(BaseModule):
|
|
# noinspection DuplicatedCode
|
|
def inject(self, registry):
|
|
self.bot: IgnCore = registry.get_instance("bot")
|
|
self.db: DB = registry.get_instance("db")
|
|
self.text = registry.get_instance("text")
|
|
self.util = registry.get_instance("util")
|
|
self.account_service: AccountService = registry.get_instance("account_service")
|
|
self.ban_service: BanService = registry.get_instance("ban_service")
|
|
self.command_alias_service = registry.get_instance("command_alias_service")
|
|
self.ts: TranslationService = registry.get_instance("translation_service")
|
|
self.character_service: CharacterService = registry.get_instance("character_service")
|
|
self.getresp = self.ts.get_response
|
|
|
|
def start(self):
|
|
self.command_alias_service.add_alias("unban", "ban rem")
|
|
self.ts.register_translation("module/ban", self.load_ban_msg)
|
|
|
|
def load_ban_msg(self):
|
|
with open("modules/core/ban/ban.msg", mode="r", encoding="UTF-8") as f:
|
|
return hjson.load(f)
|
|
|
|
@command(command="ban", params=[Const("list", is_optional=True)], access_level="moderator",
|
|
description="Show the ban list")
|
|
def ban_list_cmd(self, _, _1):
|
|
# query = self.db.query("SELECT a.*, p.*, b.reason, b.finished_at, b.created_at, b.sender_char_id from account a "
|
|
# "left join player p on a.char_id = p.char_id "
|
|
# "left join ban_list b on a.char_id = b.char_id "
|
|
# "where main=a.char_id "
|
|
# "and (disabled=1 or b.ended_early < 1)"
|
|
# "and p.org_id NOT IN (SELECT org_id from ban_org_list)")
|
|
query = self.ban_service.get_ban_list()
|
|
blob = ""
|
|
for row in query:
|
|
ends = "never" if (row.finished_at or -1) == -1 else self.util.format_datetime(row.finished_at)
|
|
blob += f"<highlight>{row.name}</highlight> by " \
|
|
f"{self.character_service.resolve_char_to_name(row.sender_char_id or self.bot.get_char_id())}\n"
|
|
if row.sender_char_id != 0 and row.sender_char_id is not None:
|
|
blob += f"<tab>Added at: <highlight>{self.util.format_datetime(row.created_at)}</highlight> " \
|
|
f"Ends: <highlight>{ends}</highlight>\n"
|
|
blob += f"<tab>Reason: {row.reason or 'None given'}\n"
|
|
blob += "\n"
|
|
return ChatBlob("Banned Accounts", blob)
|
|
|
|
@command(command="ban", params=[Options(["rem", "remove"]), Character("character")], access_level="moderator",
|
|
description="Remove a character from the ban list")
|
|
def ban_remove_cmd(self, request, _1, char):
|
|
if not char.char_id:
|
|
return self.getresp("global", "char_not_found", {"char": char.name})
|
|
elif not self.ban_service.get_ban(char.char_id):
|
|
return self.getresp("module/ban", "not_banned", {"char": char.name})
|
|
else:
|
|
self.ban_service.remove_ban(char.char_id)
|
|
self.account_service.add_log(char.char_id, 'admin',
|
|
'Ban has been lifted.',
|
|
request.sender.char_id)
|
|
self.account_service.add_log(request.sender.char_id, 'admin',
|
|
f'lifted active ban for {char.name}',
|
|
self.bot.get_char_id())
|
|
return self.getresp("module/ban", "unbanned_self", {"char": char.name})
|
|
|
|
@command(command="ban",
|
|
params=[Const("add", is_optional=True), Character("character"), Time("duration", is_optional=True),
|
|
Any("reason", is_optional=True)], access_level="moderator",
|
|
description="Add a character to the ban list")
|
|
def ban_add_cmd(self, request, _, char, duration, reason):
|
|
if not char.char_id:
|
|
return self.getresp("global", "char_not_found", {"char": char.name})
|
|
elif self.ban_service.get_ban(char.char_id):
|
|
return f"<highlight>{char.name}</highlight> is already banned."
|
|
else:
|
|
if not duration:
|
|
duration = -1
|
|
if not reason:
|
|
reason = 'No reason given.'
|
|
else:
|
|
reason = f"Reason: {reason}"
|
|
self.account_service.add_log(char.char_id, 'admin', f'{"Banned permanently. " if duration < 0 else f"Banned for {self.util.time_to_readable(duration)}."} {reason}', request.sender.char_id)
|
|
self.account_service.add_log(request.sender.char_id, 'admin',
|
|
f' {char.name} has been {"banned permanently. " if duration < 0 else f"banned for {self.util.time_to_readable(duration)}."} {reason}',
|
|
request.sender.char_id)
|
|
|
|
self.ban_service.add_ban(char.char_id, request.sender.char_id, duration, reason)
|
|
request.reply(f"<highlight>{char.name}</highlight> has been added to the ban list.")
|
|
|
|
@timerevent(budatime="1h", description="lift expired temporary bans")
|
|
def fix_bans(self, _, _1):
|
|
d = self.db.query(
|
|
f"SELECT b.*, COALESCE(p1.name, b.char_id) AS name, p2.name AS sender_name FROM ban_list b "
|
|
f"LEFT JOIN player p1 ON b.char_id = p1.char_id "
|
|
f"LEFT JOIN player p2 ON b.sender_char_id = p2.char_id "
|
|
f"WHERE ended_early < 1 AND (finished_at < ? AND finished_at > 0) "
|
|
f"ORDER BY b.created_at DESC", [time.time()])
|
|
|
|
for active in d:
|
|
dur = time.time() - active.created_at
|
|
self.ban_service.remove_ban(active.char_id, True)
|
|
self.account_service.add_log(active.char_id, "admin", f"Temporary ban lifted after a duration of {self.util.time_to_readable(dur)}. {active.reason}", self.bot.get_char_id())
|
|
for still_active in self.ban_service.get_ban_list():
|
|
self.account_service.account_disable(still_active.char_id)
|
|
|