106 lines
5.0 KiB
Python
106 lines
5.0 KiB
Python
import time
|
|
|
|
from core.decorators import instance
|
|
from core.dict_object import DictObject
|
|
from core.logger import Logger
|
|
from modules.core.accounting.services.account_service import AccountService
|
|
|
|
|
|
@instance()
|
|
class BanService:
|
|
BAN_ADDED_EVENT = "ban_added"
|
|
BAN_REMOVED_EVENT = "ban_removed"
|
|
|
|
def __init__(self):
|
|
self.logger = Logger(__name__)
|
|
|
|
def inject(self, registry):
|
|
self.db = registry.get_instance("db")
|
|
self.event_service = registry.get_instance("event_service")
|
|
self.command_service = registry.get_instance("command_service")
|
|
self.account_service: AccountService = registry.get_instance("account_service")
|
|
|
|
def pre_start(self):
|
|
self.db.exec("CREATE TABLE IF NOT EXISTS ban_list (char_id INT NOT NULL, "
|
|
"sender_char_id INT NOT NULL, "
|
|
"created_at INT NOT NULL, "
|
|
"finished_at INT NOT NULL, "
|
|
"reason VARCHAR(255) NOT NULL, "
|
|
"ended_early SMALLINT NOT NULL, "
|
|
"INDEX `char_id` (`char_id`) USING BTREE, "
|
|
"INDEX `sender_char_id` (`sender_char_id`) USING BTREE)")
|
|
self.db.exec("CREATE TABLE IF NOT EXISTS ban_org_list (org_id INT NOT NULL, "
|
|
"sender_char_id INT NOT NULL, "
|
|
"created_at INT NOT NULL, "
|
|
"finished_at INT NOT NULL, "
|
|
"reason VARCHAR(255) NOT NULL, "
|
|
"ended_early SMALLINT NOT NULL, "
|
|
"INDEX `org_id` (`org_id`) USING BTREE, "
|
|
"INDEX `sender_char_id` (`sender_char_id`) USING BTREE)")
|
|
self.event_service.register_event_type(self.BAN_ADDED_EVENT)
|
|
self.event_service.register_event_type(self.BAN_REMOVED_EVENT)
|
|
|
|
def start(self):
|
|
self.command_service.register_command_pre_processor(self.check_for_banned)
|
|
|
|
def add_ban(self, char_id, sender_char_id, duration=None, reason=None):
|
|
reason = reason or ""
|
|
|
|
t = int(time.time())
|
|
if duration:
|
|
finished_at = t + duration
|
|
else:
|
|
finished_at = -1
|
|
|
|
num_rows = self.db.exec("INSERT INTO ban_list (char_id, sender_char_id, created_at, "
|
|
"finished_at, reason, ended_early) VALUES (?, ?, ?, ?, ?, 0)",
|
|
[char_id, sender_char_id, t, finished_at, reason])
|
|
|
|
if num_rows:
|
|
self.account_service.create_users([(char_id, char_id, 0, time.time(), time.time())])
|
|
self.db.exec("UPDATE account set disabled=1 where main=(SELECT main from account where char_id=?)",
|
|
[char_id])
|
|
self.event_service.fire_event(self.BAN_ADDED_EVENT, DictObject({"char_id": char_id,
|
|
"sender_char_id": sender_char_id,
|
|
"duration": duration,
|
|
"reason": reason}))
|
|
|
|
return num_rows
|
|
|
|
def remove_ban(self, char_id, tmp=False):
|
|
t = int(time.time())
|
|
if not tmp:
|
|
num_rows = self.db.exec("UPDATE ban_list SET ended_early = 1 "
|
|
"WHERE char_id = ? and ended_early = 0", [char_id])
|
|
else:
|
|
num_rows = self.db.exec("UPDATE ban_list SET ended_early = 2 "
|
|
"WHERE char_id = ? AND (finished_at < ?)", [char_id, t])
|
|
if num_rows:
|
|
self.db.exec("UPDATE account SET disabled = 0 "
|
|
"WHERE main = (SELECT main from account where char_id=?)", [char_id])
|
|
self.event_service.fire_event(self.BAN_REMOVED_EVENT, DictObject({"char_id": char_id}))
|
|
|
|
return num_rows
|
|
|
|
def get_ban(self, char_id):
|
|
return self.db.query_single("SELECT * FROM account "
|
|
"WHERE char_id = (SELECT main from account where char_id=?) and disabled=1",
|
|
[char_id])
|
|
|
|
def get_ban_list(self, full=True) -> list[DictObject]:
|
|
t = int(time.time())
|
|
return 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 > ? {'OR finished_at = -1' if full else ''}) "
|
|
f"ORDER BY b.created_at DESC", [t])
|
|
|
|
def check_for_banned(self, context):
|
|
char_id = context.char_id
|
|
if self.get_ban(char_id):
|
|
# do nothing if character is banned
|
|
self.logger.info(f"ignoring banned character {self.account_service.character_service.resolve_char_to_name(char_id, default=str(char_id))} for command '{context.message}'")
|
|
return False
|
|
else:
|
|
return True
|