import random
import textwrap
import time
from datetime import datetime, timezone
from queue import Queue
from core.buddy_service import BuddyService
from core.chat_blob import ChatBlob
from core.command_param_types import Options, Int, Const, Any
from core.db import DB
from core.decorators import instance, event, command
from core.dict_object import DictObject
from core.job_scheduler import JobScheduler
from core.logger import Logger
from core.lookup.pork_service import PorkService
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.preference_controller import PreferenceController
from modules.core.accounting.services.account_service import AccountService
from modules.onlinebot.online.org_alias_controller import OrgAliasController
from modules.standard.news.weekly_controller import WeeklyController
from modules.standard.news.worldboss_controller import WorldBossController
@instance()
class NewsController:
queue = Queue()
# taken from https://app2brain.com/de/sprachen-lernen/woerter-saetze/hallo/
greetings = ["Welcome", "Greetings", "Hey", "Namaste", "Ahoy", "Salutations", "Hola", "Ahoj", "Hallo", "Hello",
"Hei", "Salut", "Helló", "Halló", "Ciao", "Olá", "Ahoj", "¡Hola"]
layout = \
"""
CURRENT TIME
hh:mm - DD.MM.YYYY
{time}
ACCOUNT
Your main is: {main}
You have {alts_show} alts registered
{discord}
Preferences: {prefs}
NEWS
{news}
TIMERS
{timers}
POPULAR COMMANDS
{commands}
"""
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.account_service: AccountService = registry.get_instance("account_service")
self.pork: PorkService = registry.get_instance("pork_service")
self.util: Util = registry.get_instance("util")
self.buddy_service: BuddyService = registry.get_instance("buddy_service")
self.setting_service: SettingService = registry.get_instance("setting_service")
self.job_schedule: JobScheduler = registry.get_instance("job_scheduler")
self.preferences: PreferenceController = registry.get_instance("preference_controller")
self.alias_controller: OrgAliasController = registry.get_instance("org_alias_controller")
self.weekly: WeeklyController = registry.get_instance("weekly_controller")
self.worldboss: WorldBossController = registry.get_instance("world_boss_controller")
def pre_start(self):
self.db.exec(
"CREATE TABLE IF NOT EXISTS news("
"id int primary key auto_increment, "
"text text, "
"added_by int not null, "
"added_at timestamp, "
"headline int not null default 0)")
def start(self):
self.commands = f" [{self.text.make_chatcmd('raids', '/tell raids')}] " \
f"[{self.text.make_chatcmd('online', '/tell online')}]\n" \
f" [{self.text.make_chatcmd('rules', '/tell rules')}] " \
f"[{self.text.make_chatcmd('about', '/tell about')}]\n" \
f" [{self.text.make_chatcmd('orgs', '/tell orgs')}] " \
f"[{self.text.make_chatcmd('admins', '/tell admins')}]\n"
@event("connect", "prepare info msg")
def prepare_info(self, _, _1):
self.INFO = ChatBlob("Welcome", f"You're getting this message,\n"
f"because you recently joined the clan alliance\n"
f"The New Alliance or in short "
f"TNA.\n"
f"I'm , a bot with the task to "
f"automate a few workflows around here,\n"
f"while making the life of everyone a bit easier.\n"
f"\n"
f"You can see a list of all my orgs by using "
f"{self.text.make_tellcmd('!orgs', 'orgs')}.\n"
f"Whenever you use {self.text.make_tellcmd('!online', 'online')} "
f"I'll show you who's currently around.\n"
f"We've got regular raids, you can view them by using "
f"{self.text.make_tellcmd('!raids', 'raids')}\n"
f"If you'd like to stay out, thats fine. Just turn all settings off here: "
f"{self.text.make_tellcmd('here', 'prefs')}\n"
f"\n"
f"If you encounter any issues, my administrators will help you. just ask them, "
f"you can find them all here: {self.text.make_tellcmd('!admins', 'admins')}\n"
f"\n"
f"Thats it already. "
f"Thank you for taking your time to read this note. See you ingame!"
f"")
self.INFO = f"\n________________\n\n" \
f" No need to PANIC!\n" \
f" [{self.text.paginate_single(self.INFO)}]\n" \
f" to our alliance!\n" \
f"________________"
@event(event_type="member_logon", description="Send news to players logging in")
def logon_event(self, _, data):
if not self.bot.is_ready():
return
if "member" in self.buddy_service.get_buddy(data.packet.char_id)["types"]:
account = data.account
# Apply standard checks. (User Banned, Account disabled, ...)
if not self.account_service.simple_checks(data.account):
return
if self.db.query_single("SELECT * from org_bots where char_id=?", [data.packet.char_id]):
return
if account.news_spam == 1:
if account.discord_joined == 0:
discord = f"Your Account is not connected to Discord " \
f"[{self.text.make_chatcmd('Join', '/tell discord invite')}]"
else:
discord = f"Your Account is connected to Discord as " \
f"[{self.alias_controller.get_alias(account.org_id)}] " \
f"{account.name}"
user = self.pork.get_character_info(data.packet.char_id)
self.job_schedule.delayed_job(self.send_news, 15, user, self.account_service.get_alts(account.main),
discord,
self.preferences.get_pref_view_small(account))
# This one is kinda redudant now that the simple checks get done above, will remove it in the future.
if account.last_seen == 0 and self.setting_service.get_value('is_alliance_bot') == "1":
self.bot.send_mass_message(data.packet.char_id, self.INFO)
@command(command="raids", params=[], description="Show the Raids", access_level="member")
def show_raids(self, _):
with open("data/latest_raids.txt", "r") as f:
return self.text.format_page("Raidschedule", f.read())
def get_timers(self):
events = [self.weekly.get_next_bs(),
self.weekly.get_next_ai(),
self.weekly.get_next_dio()]
next_event = sorted(events, key=lambda timer_event: timer_event.start)[0]
name = self.bot.get_char_name().upper()
out = f" {self.weekly.get_long_name(next_event.type)} - "
if next_event.is_running():
out += f"[Ends in {self.util.time_to_readable(next_event.end - time.time())}] " \
f"[by {name}]\n"
else:
out += f"[Starts in {self.util.time_to_readable(next_event.start - time.time())}] " \
f"[by {name}]\n"
for timer in self.worldboss.timer_data:
msg = self.worldboss.show_user(timer)
if msg != "No timers cached; please try again later.":
out += " " + msg + "\n"
return out
def get_next_event(self):
events = [self.weekly.get_next_bs(),
self.weekly.get_next_ai(),
self.weekly.get_next_dio()]
return sorted(events, key=lambda timer: timer.start)[0]
def send_news(self, _, sender, alts, discord, prefs, auto=True):
if auto:
if self.buddy_service.get_buddy(sender.char_id)["online"] == 0:
return
timers = self.get_timers()
next_event = self.get_next_event()
news = self.layout.format(time=datetime.now(timezone.utc).strftime("%H:%M - %d.%m.%Y") + " [UTC-0]",
main=alts[0].name,
alts_show=self.text.make_chatcmd(len(alts), "/tell alts"),
news=self.get_news(),
timers=timers,
commands=self.commands,
prefs=prefs,
discord=discord)
name = self.weekly.get_long_name(next_event.type)
blob = f"{random.choice(self.greetings)} {sender.name} :: " \
f"{self.text.format_page('Your News', textwrap.dedent(news))} " \
f"{f':: {name} running' if next_event.is_running() else ''}"
self.bot.send_mass_message(sender.char_id, blob)
#####################
# News Management #
#####################
@command(command="news", params=[], description="Show the news", access_level="member")
def show_news(self, sender):
user = self.db.query_single("SELECT * FROM player where char_id=?", [sender.sender.char_id])
alts = self.account_service.get_alts(sender.sender.char_id)
if not alts:
discord = f"Your Account is not connected to Discord " \
f"[{self.text.make_chatcmd('Join', '/tell discord invite')}]"
alts = [self.account_service.get_main(sender.sender.char_id)]
elif alts[0].discord_joined == 0:
discord = f"Your Account is not connected to Discord " \
f"[{self.text.make_chatcmd('Join', '/tell discord invite')}]"
else:
prefix = ""
if self.setting_service.get_value('is_alliance_bot') == '1':
prefix = f"[{self.alias_controller.get_alias(alts[0].org_id)}] "
discord = f"Your Account is connected to Discord as " \
f"{prefix}{alts[0].name}"
prefs = self.preferences.get_pref_view_small(alts[0]) if alts else ""
self.send_news(0, user, alts, discord, prefs, False)
@command(command="news", params=[Options(["pin", "unpin"]), Int("ID")], description="Set news as headlines",
access_level="moderator", sub_command="moderate")
def news_pin(self, _, option, news_id):
if option == "pin":
row = self.db.exec("UPDATE news set headline=1 where id =?", [news_id])
if row == 0:
return f"There's no entry with the ID {news_id}, or it is already pinned."
elif row == 1:
return f"The entry with the ID {news_id} has been pinned."
else:
return "Multiple entries have been pinned... SHIT!"
elif option == "unpin":
row = self.db.exec("UPDATE news set headline=0 where id =?", [news_id])
if row == 0:
return f"There's no entry with the ID {news_id}, or it is not pinned."
elif row == 1:
return f"The entry with the ID {news_id} has been unpinned."
else:
return "Multiple entries have been unpinned... SHIT!"
@command(command="news", params=[Options(["remove", "delete", "rem", "del"]), Int("ID")],
description="Set news as headlines", access_level="moderator", sub_command="moderate")
def news_del(self, _, _1, news_id):
row = self.db.exec("DELETE FROM news where id =?", [news_id])
if row == 0:
return f"There's no entry with the ID {news_id}."
elif row == 1:
return f"The entry with the ID {news_id} has been removed."
else:
return "Multiple entries have been deleted... SHIT!"
@command(command="news", params=[Const("add"), Any("text")], description="Moderate the news",
access_level="moderator", sub_command="moderate")
def news_add(self, sender, _, text):
self.add_entry(text, sender.sender)
row = DictObject(
{"added_at": datetime.utcfromtimestamp(time.time()), "added_by": sender.sender.char_id, "headline": 0,
"id": self.db.query_single("SELECT id from news order by id desc").id, "text": text})
return ChatBlob(f"News entry with ID {row.id}",
self.format_entry(row, sender.sender, sender.sender.access_level['level'] <= 30))
@command(command="news", params=[Options(["detail", "info", "more"]), Int("ID", is_optional=True)],
description="Moderate the news", access_level="member")
def news_detail(self, sender, _, news_id):
if news_id:
row = self.db.query_single("SELECT * from news where id=?", [news_id])
if row:
return ChatBlob(f"News entry with ID {news_id}",
self.format_entry(row, sender.sender, sender.sender.access_level['level'] <= 30))
else:
return f"There's no entry with the ID {news_id}."
else:
return ChatBlob("The latest news", self.get_news(10000, sender.sender))
def get_news(self, limit=500, receiver=None):
data = self.db.query("SELECT * from news order by ID desc")
normal = ""
headline = ""
for entry in data:
if entry.headline == 1:
if len(self.text.strip_html_tags(headline) or []) > limit / 2:
continue
headline += self.format_entry(entry, receiver)
else:
if len(self.text.strip_html_tags(normal) or []) > limit / 2:
continue
normal += self.format_entry(entry, receiver)
if len(self.text.strip_html_tags(headline + normal) or []) > limit:
normal += f" There seem to be more news,
but they're too long to get shown here. " \
f"[{self.text.make_tellcmd('Display', 'news detail')}]
"
return headline + "____________________________
" + normal if headline != "" else normal
# noinspection LongLine
def format_entry(self, entry, receiver=None, mod=False):
name = self.bot.get_char_name().upper()
base = f"""On {entry.added_at.strftime("%d.%m.%Y - %H:%M:%S")} UTC by {self.bot.character_service.resolve_char_to_name(entry.added_by) if entry.added_by != 0 else f"{name}"} (ID {entry.id}):
{entry.text}
"""
if mod:
base += f"""
- Moderate this entry: [{self.text.make_tellcmd('PIN' if entry.headline == 0 else 'UNPIN', f'news {"pin" if entry.headline == 0 else "unpin"} {entry.id}')}] [{self.text.make_tellcmd('DELETE', f'news delete {entry.id}')}]"""
return base
def add_entry(self, text, sender):
self.db.exec("INSERT INTO news(text, added_by, added_at) VALUES(?, ?, ?)",
[text, sender.char_id, datetime.utcfromtimestamp(time.time())])