eriks/erepublik-discord_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Major rewrite

Browse Source
This commit is contained in:
Ēriks K 2021-08-26 19:57:49 +03:00
parent 2b4ec22349
commit 75186c3728
9 changed files with 459 additions and 374 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
dbot/__init__.py Normal file
View File

54
dbot/base.py Normal file
View File

@ -0,0 +1,54 @@
import logging
import os
import sys
from db import DiscordDB
APP_NAME = "discord_bot"
os.chdir(os.path.abspath(os.path.dirname(sys.argv[0])))
os.makedirs("debug", exist_ok=True)
logger = logging.getLogger(APP_NAME)
logger.setLevel(logging.DEBUG)
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
file_logger = logging.FileHandler("debug/logging.log", "w")
file_logger.setLevel(logging.WARNING)
file_logger.setFormatter(formatter)
logger.addHandler(file_logger)
stream_logger = logging.StreamHandler()
stream_logger.setLevel(logging.DEBUG)
stream_logger.setFormatter(formatter)
logger.addHandler(stream_logger)
DISCORD_TOKEN = os.getenv("DISCORD_TOKEN")
DEFAULT_CHANNEL_ID = os.getenv("DEFAULT_CHANNEL_ID", 603527159109124096)
ADMIN_ID = os.getenv("ADMIN_ID", 220849530730577920)
DB_NAME = os.getenv("DB_NAME", "discord.db")
PRODUCTION = bool(os.getenv("PRODUCTION"))
DB = DiscordDB(DB_NAME)
MENTION_MAPPING = {1: "D1", 2: "D2", 3: "D3", 4: "D4", 11: "Air"}
DIVISION_MAPPING = {v: k for k, v in MENTION_MAPPING.items()}
NOTIFICATION_KINDS = (
"epic",
"events",
"empty",
)
MESSAGES = dict(
not_admin="❌ Only server administrators are allowed to enable notifications!",
not_in_pm="❌ Unable to notify in PMs!",
command_failed="❌ Command failed!",
only_registered_channels="❌ This command is only available from registered channels!",
mention_help="❌ Please provide kind, action, division and role to mention! Eg. '{command} epic set d4 @div4' or '{command} empty remove d3'",
mention_info="ℹ️ If You want for me to also add division mentions write:\n`!control mention [kind] set [division] [role_mention]`\n"
"Example: `!control mention {kind} set d4 @div4` or `!control mention {kind} set air @aviators`",
nothing_to_do="ℹ️ Nothing to do here...",
notifications_unset="✅ I won't notify about {} in this channel!",
notifications_set="✅ I will notify about {} in this channel!",
)

175
dbot/bot_commands.py Normal file
View File

@ -0,0 +1,175 @@
import sys
from discord import Embed
from discord.enums import ChannelType
from discord.ext import commands
from dbot.base import ADMIN_ID, DB, DIVISION_MAPPING, MESSAGES, NOTIFICATION_KINDS, logger
from dbot.utils import check_battles, get_battle_page
__all__ = ["bot"]
bot = commands.Bot(command_prefix="!")
async def control_register(ctx, *args):
if " ".join(args) == "From AF With Love!":
DB.update_member(ctx.author.id, str(ctx.author), True)
return await ctx.send("✅ You have been registered and are allowed to issue commands privately! 🥳")
return await ctx.send(MESSAGES["command_failed"])
async def control_notify(ctx, kind):
if kind == "epic":
if DB.add_notification_channel(ctx.guild.id, ctx.channel.id, kind):
return await ctx.send(MESSAGES["notifications_set"].format("epic battles"))
elif kind == "events":
if DB.add_notification_channel(ctx.guild.id, ctx.channel.id, kind):
return await ctx.send(MESSAGES["notifications_set"].format("eLatvia's events"))
elif kind == "empty":
if DB.add_notification_channel(ctx.guild.id, ctx.channel.id, kind):
return await ctx.send(MESSAGES["notifications_set"].format("empty medals"))
return await ctx.send(MESSAGES["nothing_to_do"])
async def control_unnotify(ctx, kind):
if DB.remove_kind_notification_channel(kind, ctx.channel.id):
if kind == "epic":
return await ctx.send(MESSAGES["notifications_unset"].format("epic battles"))
if kind == "events":
return await ctx.send(MESSAGES["notifications_unset"].format("eLatvia's notifications"))
if kind == "empty":
return await ctx.send(MESSAGES["notifications_unset"].format("empty medals"))
return await ctx.send(MESSAGES["command_failed"])
async def control_mention_set(ctx, kind: str, division: str, role: str):
for guild_role in ctx.guild.roles:
if guild_role.mention == role:
if not guild_role.mentionable:
return await ctx.send(f"❌ Unable to use {role=}, because this role is not globally mentionable!")
DB.add_role_mapping_entry(kind, ctx.channel.id, DIVISION_MAPPING[division], guild_role.id)
return await ctx.send(f"✅ Success! For {division} epics I will mention {guild_role.mention}")
return await ctx.send(MESSAGES["command_failed"])
async def control_mention_remove(ctx, kind: str, division: str):
if DB.remove_role_mapping(kind, ctx.channel.id, DIVISION_MAPPING[division]):
return await ctx.send(f"✅ I won't mention here any role about {division} events!")
return await ctx.send(MESSAGES["nothing_to_do"])
@bot.event
async def on_ready():
logger.info("Bot loaded")
# print(bot.user.name)
# print(bot.user.id)
logger.info("------")
@bot.command()
async def empty(ctx, division, minutes: int = 60):
if not ctx.channel.id == 603527159109124096 or not DB.get_member(ctx.message.author.id).get("pm_is_allowed"):
return await ctx.send("Currently unavailable!")
try:
div = int(division)
except ValueError:
try:
div = dict(D1=1, D2=3, D3=3, D4=4, Air=11)[division.title()]
except (AttributeError, KeyError):
return await ctx.send("First argument must be a value from: 1, d1, 2, d2, 3, d3, 4, d4, 11, air!")
s_div = {1: "D1", 2: "D2", 3: "D3", 4: "D4", 11: "Air"}[div]
embed = Embed(
title=f"Possibly empty {s_div} medals",
description="'Empty' medals are being guessed based on the division wall. Expect false-positives!",
)
for kind, div_div, data in check_battles(get_battle_page().get("battles")):
if kind == "empty" and div_div == div and data["round_time_s"] >= minutes * 60:
embed.add_field(
name=f"**Battle for {data['region']} {' '.join(data['sides'])}**",
value=f"[R{data['zone_id']} | Time {data['round_time']}]({data['url']})",
)
if len(embed.fields) >= 10:
return await ctx.send(embed=embed)
if embed.fields:
return await ctx.send(embed=embed)
else:
return await ctx.send(f"No empty {s_div} medals found")
@empty.error
async def division_error(ctx, error):
if isinstance(error, (commands.BadArgument, commands.MissingRequiredArgument)):
return await ctx.send("❌ Division is mandatory, eg, `!empty [1,2,3,4,11, d1,d2,d3,d4,air, D1,D2,D3,D4,Air] [1-120]`")
logger.exception(error, exc_info=error)
await ctx.send("❌ Something went wrong! 😔")
@bot.command()
async def control(ctx: commands.Context, command: str, *args):
if not DB.get_member(ctx.author.id):
DB.add_member(ctx.author.id, str(ctx.author))
if command == "register":
return await control_register(ctx, *args)
if command in ["notify", "unnotify"]:
if ctx.channel.type == ChannelType.private:
return await ctx.send(MESSAGES["not_in_pm"])
if not ctx.author.guild_permissions.administrator:
return await ctx.send(MESSAGES["not_admin"])
if not args:
return await ctx.send(
f"❌ Please provide what kind of notifications You would like to {'en' if command == 'notify' else 'dis'}able! Currently available: {', '.join(NOTIFICATION_KINDS)}"
)
kind = str(args[0]).lower()
if kind not in NOTIFICATION_KINDS:
return await ctx.send(f'❌ Notification {kind=} is unknown! Currently available: {", ".join(NOTIFICATION_KINDS)}')
if command == "notify":
return await control_notify(ctx, kind)
if command == "unnotify":
return await control_unnotify(ctx, kind)
if command == "mention":
if ctx.channel.type == ChannelType.private:
return await ctx.send(MESSAGES["not_in_pm"])
if not ctx.author.guild_permissions.administrator:
return await ctx.send(MESSAGES["not_admin"])
if not args or not 3 <= len(args) <= 4:
return await ctx.send(MESSAGES["mention_help"].format(command=command))
try:
kind, action, division, *role = args
if role:
role = role[0]
kind = str(kind).lower()
if ctx.channel.id not in DB.get_kind_notification_channel_ids(kind):
return await ctx.send(MESSAGES["only_registered_channels"])
if kind not in ("epic", "empty"):
return await ctx.send(f"❌ {kind=} doesn't support division mentioning!")
if action not in ("set", "remove"):
return await ctx.send(MESSAGES["mention_help"].format(command=command))
action = str(action).lower()
division = str(division).title()
if division not in DIVISION_MAPPING:
await ctx.send(f"❌ Unknown {division=}! Available divisions: {', '.join(d.title() for d in DIVISION_MAPPING.keys())}")
return await ctx.send(MESSAGES["mention_info"].format(kind=kind))
if action == "set":
return await control_mention_set(ctx, kind, division, role)
if action == "remove":
return await control_mention_remove(ctx, kind, division)
except Exception as e:
logger.warning(str(e), exc_info=e, stacklevel=3)
return await ctx.send(MESSAGES["mention_help"].format(command=command))
if command == "exit":
if ctx.author.id == ADMIN_ID:
await ctx.send(f"{ctx.author.mention} Bye!")
sys.exit(0)
return await ctx.send(f"❌ Unknown {command=}!")
@empty.error
async def control_error(ctx, error):
logger.exception(error, exc_info=error)
return await ctx.send(MESSAGES["command_failed"])

99
dbot/constants.py
View File

@ -1,9 +1,9 @@
import re
from typing import NamedTuple
from typing import Any, Dict, List, NamedTuple, TypedDict
from erepublik.constants import COUNTRIES
__all__ = ["events", COUNTRIES, "FLAGS", "UTF_FLAG"]
__all__ = ["events", "COUNTRIES", "UTF_FLAG", "DivisionData"]
region = r"[\w\(\)\-& ']+"
country = r"(Resistance force of )?[\w\(\)\- ]+"
@ -191,9 +191,9 @@ events = [
"Res Concession",
re.compile(
rf"A Resource Concession law to //www.erepublik.com<b>(?P<target>{country})</b> "
rf'<a href="(?P<link>((https?:)?//www\.erepublik\.com)?/en/main/law/(?P<source>{country}/\d+))">has been (?P<result>.*?)</a>'
rf'<a href="(?P<link>((https?:)?//www\.erepublik\.com)?/en/main/law/(?P<source>{country})/\d+)">has been (?P<result>.*?)</a>'
),
"Resource Concession law between {current_country} and {target} has been {result}",
"Resource Concession law between {source} and {target} has been {result}",
),
EventKind(
"cp_impeachment",
@ -258,13 +258,13 @@ events = [
EventKind(
"new_welcome_message_proposed",
"New Welcome message has been proposed",
re.compile(r"President of (?P<country>{country}) proposed a new welcome message for new citizens"),
re.compile(rf"President of (?P<country>{country}) proposed a new welcome message for new citizens"),
"{country} proposed new welcome message!",
),
EventKind(
"new_welcome_message_approved",
"New Welcome message has been approved",
re.compile(r"(?P<country>{country}) now has a new welcoming message for new citizens"),
re.compile(rf"(?P<country>{country}) now has a new welcoming message for new citizens"),
"{country} approved new welcome message!",
),
]
@ -345,79 +345,14 @@ UTF_FLAG = {
170: "🇳🇬",
171: "🇨🇺",
}
FLAGS = {
1: "flag_ro",
9: "flag_br",
10: "flag_it",
11: "flag_fr",
12: "flag_de",
13: "flag_hu",
14: "flag_cn",
15: "flag_es",
23: "flag_ca",
24: "flag_us",
26: "flag_mx",
27: "flag_ar",
28: "flag_ve",
29: "flag_gb",
30: "flag_ch",
31: "flag_nl",
32: "flag_be",
33: "flag_at",
34: "flag_cz",
35: "flag_pl",
36: "flag_sk",
37: "flag_no",
38: "flag_se",
39: "flag_fi",
40: "flag_ua",
41: "flag_ru",
42: "flag_bg",
43: "flag_tr",
44: "flag_gr",
45: "flag_jp",
47: "flag_kr",
48: "flag_in",
49: "flag_id",
50: "flag_au",
51: "flag_za",
52: "flag_md",
53: "flag_pt",
54: "flag_ie",
55: "flag_de",
56: "flag_ir",
57: "flag_pk",
58: "flag_il",
59: "flag_th",
61: "flag_si",
63: "flag_hr",
64: "flag_cl",
65: "flag_rs",
66: "flag_my",
67: "flag_ph",
68: "flag_sg",
69: "flag_ba",
70: "flag_ee",
71: "flag_lv",
72: "flag_lt",
73: "flag_kp",
74: "flag_uy",
75: "flag_py",
76: "flag_bo",
77: "flag_pe",
78: "flag_co",
79: "flag_mk",
80: "flag_me",
81: "flag_tw",
82: "flag_cy",
83: "flag_by",
84: "flag_nz",
164: "flag_sa",
165: "flag_eg",
166: "flag_ae",
167: "flag_al",
168: "flag_ge",
169: "flag_am",
170: "flag_ng",
171: "flag_cu",
}
class DivisionData(TypedDict):
region: str
round_time: str
round_time_s: int
sides: List[str]
url: str
zone_id: int
div_id: int
extra: Dict[str, Any]

174
dbot/db.py
View File

@ -1,51 +1,51 @@
from typing import Dict, Optional, Union, List, Tuple
import logging
from typing import Dict, List, Optional, Union
from sqlite_utils import Database
from sqlite_utils.db import NotFoundError
logger = logging.getLogger(__name__)
class DiscordDB:
_name: str
_db: Database
def __init__(self, db_name: str = ""):
self._name = db_name
if not self._name:
self._db = Database(memory=True)
else:
self._db = Database(self._name)
if "member" not in self._db.table_names():
self._db.create_table("member", {"id": int, "name": str}, pk="id", not_null={"id", "name"})
self._db = Database(db_name) if db_name else Database(memory=True)
if "player" not in self._db.table_names():
self._db.create_table("player", {"id": int, "name": str}, pk="id", not_null={"id", "name"})
if "epic" not in self._db.table_names():
self._db.create_table("epic", {"id": int, "fake": bool}, pk="id", not_null={"id"}, defaults={"fake": False})
if "rss_feed" not in self._db.table_names():
self._db.create_table("rss_feed", {"id": int, "timestamp": float}, pk="id", not_null={"id", "timestamp"})
if "notification_channel" not in self._db.table_names():
self._db.create_table(
"notification_channel", {"id": int, "guild_id": int, "channel_id": int, "kind": str}, pk="id", not_null={"id", "guild_id", "channel_id"}, defaults={"kind": "epic"}
)
self._db["notification_channel"].create_index(("guild_id", "channel_id", "kind"), unique=True)
self._db.create_table("role_mapping", {"id": int, "channel_id": int, "division": int, "role_id": int}, pk="id", not_null={"id", "channel_id", "division", "role_id"})
self._db["role_mapping"].add_foreign_key("channel_id", "notification_channel", "channel_id")
self._db["role_mapping"].create_index(("channel_id", "division"), unique=True)
self._db.vacuum()
self.initialize()
self.member = self._db.table("member")
self.player = self._db.table("player")
self.epic = self._db.table("epic")
self.division = self._db.table("division")
self.rss_feed = self._db.table("rss_feed")
self.channel = self._db.table("notification_channel")
self.channel = self._db.table("channel")
self.role_mapping = self._db.table("role_mapping")
def initialize(self):
hard_tables = ["member", "player", "channel", "role_mapping"]
db_tables = self._db.table_names()
if "member" not in db_tables:
self._db.create_table("member", {"name": str, "pm_is_allowed": bool}, pk="id", not_null={"name", "pm_is_allowed"}, defaults={"pm_is_allowed": False})
if "player" not in db_tables:
self._db.create_table("player", {"name": str}, pk="id", not_null={"name"})
if "notification_channel" in db_tables or "channel" not in db_tables:
self._db.create_table("channel", {"guild_id": int, "channel_id": int, "kind": str}, pk="id", not_null={"guild_id", "channel_id", "kind"}, defaults={"kind": "epic"})
self._db["channel"].create_index(("guild_id", "channel_id", "kind"), unique=True)
for row in self._db.table("notification_channel").rows:
self._db["channel"].insert(**row)
if "role_mapping" not in db_tables:
self._db.create_table("role_mapping", {"channel_id": int, "division": int, "role_id": int}, pk="id", not_null={"channel_id", "division", "role_id"})
self._db["role_mapping"].add_foreign_key("channel_id", "channel", "id")
self._db["role_mapping"].create_index(("channel_id", "division"), unique=True)
for table in self._db.table_names():
if table not in hard_tables:
self._db.table(table).drop(ignore=True)
self._db.create_table("division", {"division_id": int, "epic": bool, "empty": bool}, pk="id", defaults={"epic": False, "empty": False}, not_null={"division_id"})
self._db.create_table("rss_feed", {"timestamp": float}, pk="id", not_null={"timestamp"})
self._db.vacuum()
# Player methods
def get_player(self, pid: int) -> Optional[Dict[str, Union[int, str]]]:
@ -87,7 +87,7 @@ class DiscordDB:
# Member methods
def get_member(self, member_id) -> Dict[str, Union[int, str]]:
def get_member(self, member_id) -> Optional[Dict[str, Union[int, str]]]:
"""Get discord member
:param member_id: int Discord Member ID
@ -98,56 +98,85 @@ class DiscordDB:
try:
return self.member.get(member_id)
except NotFoundError:
raise NotFoundError("Member with given id not found")
return
def add_member(self, id: int, name: str) -> Dict[str, Union[int, str]]:
def add_member(self, id: int, name: str, pm_is_allowed: bool = False) -> Dict[str, Union[int, str]]:
"""Add discord member.
:param id: int Discord member ID
:param name: Discord member Name
:param pm_is_allowed: Allow discord member to contact bot through PMs
"""
try:
self.member.insert({"id": id, "name": name})
self.member.insert({"id": id, "name": name, "pm_is_allowed": pm_is_allowed})
finally:
return self.member.get(id)
def update_member(self, member_id: int, name: str) -> bool:
def update_member(self, member_id: int, name: str, pm_is_allowed: bool = None) -> bool:
"""Update discord member"s record
:param member_id: Discord Mention ID
:type member_id: int Discord Mention ID
:type member_id: int
:param name: Discord user name
:type name: str Discord user name
:type name: str
:param pm_is_allowed: Is discord user allowed to interact through PMs
:type pm_is_allowed: Optional[bool]
:return: bool
"""
try:
member = self.get_member(member_id)
except NotFoundError:
member = self.add_member(member_id, name)
self.member.update(member["id"], {"name": name})
member = self.get_member(member_id)
if member:
if pm_is_allowed is None:
pm_is_allowed = self.member.get(member_id).get("pm_is_allowed")
self.member.update(member["id"], {"name": name, "pm_is_allowed": pm_is_allowed})
return True
self.add_member(member_id, name)
return True
# Epic Methods
def get_epic(self, division_id: int) -> Optional[Dict[str, Union[int, str]]]:
def check_epic(self, division_id: int) -> bool:
"""Check if epic has been registered in the division
:param division_id: int Division ID
:return: bool
"""
try:
return bool(next(self.division.rows_where("division_id = ? and epic = ?", (division_id, True))))
except StopIteration:
return False
def add_epic(self, division_id: int) -> bool:
"""Register epic in division.
:param division_id: int Epic division ID
:return: bool Epic division added
"""
if not self.check_epic(division_id):
self.division.insert({"division_id": division_id, "epic": True})
return True
return False
# Epic Methods
def check_empty_medal(self, division_id: int) -> bool:
"""Get Epic division
:param division_id: int Division ID
:return: division id
"""
try:
return self.epic.get(division_id)
except NotFoundError:
return None
return bool(next(self.division.rows_where("division_id = ? and empty = ?", (division_id, True))))
except StopIteration:
return False
def add_epic(self, division_id: int) -> bool:
def add_empty_medal(self, division_id: int) -> bool:
"""Add Epic division.
:param division_id: int Epic division ID
:return: bool Epic division added
"""
if not self.get_epic(division_id):
self.epic.insert({"id": division_id})
if not self.check_empty_medal(division_id):
self.division.insert({"division_id": division_id, "empty": True})
return True
return False
@ -175,7 +204,7 @@ class DiscordDB:
else:
self.rss_feed.insert({"id": country_id, "timestamp": timestamp})
# RSS Event Methods
# Notification methods
def add_notification_channel(self, guild_id: int, channel_id: int, kind: str) -> bool:
if channel_id in self.get_kind_notification_channel_ids(kind):
@ -185,31 +214,50 @@ class DiscordDB:
def get_kind_notification_channel_ids(self, kind: str) -> List[int]:
channels = [row["channel_id"] for row in self.channel.rows_where("kind = ?", [kind])]
logger.info(f"Found {len(channels)} channels for {kind} kind: {channels}")
return channels
def get_notification_channel_id(self, kind: str, *, guild_id: int = None, channel_id: int = None) -> Optional[int]:
if guild_id is None and channel_id is None:
raise RuntimeError("Must provide either guild_id or channel_id!")
for row in self.channel.rows_where(f"kind = ? and {'guild_id' if guild_id is not None else 'channel_id'} = ?", [kind, guild_id or channel_id]):
return row["id"]
def remove_kind_notification_channel(self, kind, channel_id) -> bool:
if channel_id in self.get_kind_notification_channel_ids(kind):
logger.warning(f"removing channel with id {channel_id}")
self.remove_all_channel_role_mappings(channel_id, kind)
self.channel.delete_where("kind = ? and channel_id = ?", (kind, channel_id))
self.remove_role_mappings(channel_id)
return True
return False
def remove_role_mappings(self, channel_id: int):
return self.role_mapping.delete_where("channel_id = ?", (channel_id,))
# Role mapping methods
def add_role_mapping_entry(self, channel_id: int, division: int, role_id: int) -> bool:
def add_role_mapping_entry(self, kind: str, channel_id: int, division: int, role_id: int) -> bool:
ch_id = self.get_notification_channel_id(kind, channel_id=channel_id)
if division not in (1, 2, 3, 4, 11):
return False
try:
row = next(self.role_mapping.rows_where("channel_id = ? and division = ?", [channel_id, division]))
self.role_mapping.update(row["id"], {"channel_id": channel_id, "division": division, "role_id": role_id})
row = next(self.role_mapping.rows_where("channel_id = ? and division = ?", [ch_id, division]))
self.role_mapping.update(row["id"], {"channel_id": ch_id, "division": division, "role_id": role_id})
except StopIteration:
self.role_mapping.insert({"channel_id": channel_id, "division": division, "role_id": role_id})
self.role_mapping.insert({"channel_id": ch_id, "division": division, "role_id": role_id})
return True
def get_role_id_for_channel_division(self, channel_id: int, division: int) -> Optional[int]:
rows = self.role_mapping.rows_where("channel_id = ? and division = ?", (channel_id, division))
def remove_all_channel_role_mappings(self, channel_id: int, kind: str):
ch_id = self.get_notification_channel_id(kind, channel_id=channel_id)
for d in (1, 2, 3, 4, 11):
self.remove_role_mapping(kind, ch_id, d)
def remove_role_mapping(self, kind: str, channel_id: int, division_id: int) -> bool:
try:
ch_id = self.get_notification_channel_id(kind, channel_id=channel_id)
row = next(self.role_mapping.rows_where("channel_id = ? and division = ? ", (ch_id, division_id)))
self.role_mapping.delete(row["id"])
return True
except StopIteration:
return False
def get_role_id_for_channel_division(self, *, kind: str, channel_id: int, division: int) -> Optional[int]:
ch_id = self.get_notification_channel_id(kind, channel_id=channel_id)
rows = self.role_mapping.rows_where("channel_id = ? and division = ?", (ch_id, division))
for row in rows:
return row["role_id"]

212
dbot/discord_bot.py
View File

@ -1,47 +1,18 @@
import asyncio
import datetime
import logging
import os
import sys
import time
from json import JSONDecodeError
import discord
import feedparser
import pytz
import requests
from discord.ext import commands
from constants import events
from erepublik.constants import COUNTRIES
from constants import events
from db import DiscordDB
from dbot.utils import timestamp, check_battles
APP_NAME = "discord_bot"
os.chdir(os.path.abspath(os.path.dirname(sys.argv[0])))
os.makedirs("debug", exist_ok=True)
logger = logging.getLogger(APP_NAME)
logger.setLevel(logging.DEBUG)
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
file_logger = logging.FileHandler("debug/logging.log", "w")
file_logger.setLevel(logging.WARNING)
file_logger.setFormatter(formatter)
logger.addHandler(file_logger)
stream_logger = logging.StreamHandler()
stream_logger.setLevel(logging.DEBUG)
stream_logger.setFormatter(formatter)
logger.addHandler(stream_logger)
DISCORD_TOKEN = os.getenv("DISCORD_TOKEN")
DEFAULT_CHANNEL_ID = os.getenv("DEFAULT_CHANNEL_ID", 603527159109124096)
ADMIN_ID = os.getenv("ADMIN_ID", 220849530730577920)
DB_NAME = os.getenv("DB_NAME", "discord.db")
PRODUCTION = bool(os.getenv("PRODUCTION"))
DB = DiscordDB(DB_NAME)
from dbot.base import ADMIN_ID, DB, DB_NAME, DEFAULT_CHANNEL_ID, DISCORD_TOKEN, PRODUCTION, logger
from dbot.bot_commands import bot
from dbot.utils import check_battles, get_battle_page, timestamp
if PRODUCTION:
logger.warning("Production mode enabled!")
@ -53,37 +24,18 @@ if PRODUCTION:
logger.debug(f"Active configs:\nDISCORD_TOKEN='{DISCORD_TOKEN}'\nDEFAULT_CHANNEL_ID='{DEFAULT_CHANNEL_ID}'\nADMIN_ID='{ADMIN_ID}'\nDB_NAME='{DB_NAME}'")
MENTION_MAPPING = {1: "D1", 2: "D2", 3: "D3", 4: "D4", 11: "Air"}
__last_battle_response = None
__last_battle_update_timestamp = 0
def get_battle_page():
global __last_battle_update_timestamp, __last_battle_response
if int(datetime.datetime.now().timestamp()) >= __last_battle_update_timestamp + 60:
dt = datetime.datetime.now()
r = requests.get("https://www.erepublik.com/en/military/campaignsJson/list")
try:
__last_battle_response = r.json()
except JSONDecodeError:
logger.warning("Received non json response from erep.lv/battles.json!")
return get_battle_page()
__last_battle_update_timestamp = __last_battle_response.get("last_updated", int(dt.timestamp()))
return __last_battle_response
class MyClient(discord.Client):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# create the background task and run it in the background
self.last_event_timestamp = timestamp()
self.bg_task = self.loop.create_task(self.report_battle_events())
self.bg_rss_task = self.loop.create_task(self.report_rss_events())
async def on_ready(self):
logger.info("Client running")
logger.info("------")
self.bg_task = self.loop.create_task(self.report_battle_events())
# self.bg_rss_task = self.loop.create_task(self.report_rss_events())
async def on_error(self, event_method, *args, **kwargs):
logger.warning(f"Ignoring exception in {event_method}")
@ -126,6 +78,7 @@ class MyClient(discord.Client):
__link = values["link"]
entry_link = __link if __link.startswith("http") else f"https://www.erepublik.com{__link}"
logger.debug(kind.format.format(**dict(match.groupdict(), **{"current_country": country.name})))
logger.debug(entry_link)
is_latvia = country.id == 71
has_latvia = any("Latvia" in v for v in values.values())
if is_latvia or has_latvia:
@ -149,8 +102,6 @@ class MyClient(discord.Client):
logger.debug(f"Message sent: {text}")
for channel_id in DB.get_kind_notification_channel_ids("events"):
await self.get_channel(channel_id).send(embed=embed)
await asyncio.sleep((timestamp() // 300 + 1) * 300 - timestamp())
except Exception as e:
logger.error("eRepublik event reader ran into a problem!", exc_info=e)
try:
@ -158,7 +109,8 @@ class MyClient(discord.Client):
f.write(feed_response.text)
except (NameError, AttributeError):
logger.error("There was no Response object!", exc_info=e)
await asyncio.sleep(10)
finally:
await asyncio.sleep((timestamp() // 300 + 1) * 300 - timestamp())
async def report_battle_events(self):
await self.wait_until_ready()
@ -176,34 +128,35 @@ class MyClient(discord.Client):
2: discord.Embed(title="Possibly empty **__last-minute__ D2** medals", description=desc),
3: discord.Embed(title="Possibly empty **__last-minute__ D3** medals", description=desc),
4: discord.Embed(title="Possibly empty **__last-minute__ D4** medals", description=desc),
11: discord.Embed(title="Possibly empty **__last-minute__ Air** medals", description=desc)
11: discord.Embed(title="Possibly empty **__last-minute__ Air** medals", description=desc),
}
for kind, div, data in check_battles(r.get('battles')):
if kind == 'epic' and not DB.get_epic(data['div_id']):
embed = discord.Embed.from_dict(dict(
title=" ".join(data['extra']["intensity_scale"].split("_")).title(),
for kind, div, data in check_battles(r.get("battles")):
if kind == "epic" and not DB.check_epic(data["div_id"]):
embed_data = dict(
title=" ".join(data["extra"]["intensity_scale"].split("_")).title(),
url=data["url"],
description=f"Epic battle {' vs '.join(data['sides'])}!\nBattle for {data['region']}, Round {data['zone_id']}",
footer=f"Round time {data['round_time']}"
))
logger.debug(f"{embed.title=}, {embed.description=}, {embed.url=}, {embed.footer=}")
footer=f"Round time {data['round_time']}",
)
embed = discord.Embed.from_dict(embed_data)
logger.debug(f"{embed_data=}")
for channel_id in DB.get_kind_notification_channel_ids("epic"):
if role_id := DB.get_role_id_for_channel_division(channel_id, division=div):
await self.get_channel(channel_id).send(f"<@&{role_id}>", embed=embed)
if role_id := DB.get_role_id_for_channel_division(kind="epic", channel_id=channel_id, division=div):
await self.get_channel(channel_id).send(f"<@&{role_id}> epic battle detected!", embed=embed)
else:
await self.get_channel(channel_id).send(embed=embed)
DB.add_epic(data['div_id'])
DB.add_epic(data["div_id"])
if kind == 'empty' and data['round_time_s'] >= 87 * 60:
if kind == "empty" and data["round_time_s"] >= 85 * 60 and not DB.check_empty_medal(data["div_id"]):
empty_divisions[div].add_field(
name=f"**Battle for {data['region']} {' '.join(data['sides'])}**",
value=f"[R{data['zone_id']} | Time {data['round_time']}]({data['url']})"
name=f"**Battle for {data['region']} {' '.join(data['sides'])}**", value=f"[R{data['zone_id']} | Time {data['round_time']}]({data['url']})"
)
DB.add_empty_medal(data["div_id"])
for d, e in empty_divisions.items():
if e.fields:
for channel_id in DB.get_kind_notification_channel_ids("empty"):
if role_id := DB.get_role_id_for_channel_division(channel_id, division=d):
await self.get_channel(channel_id).send(f"<@&{role_id}>", embed=e)
if role_id := DB.get_role_id_for_channel_division(kind="empty", channel_id=channel_id, division=d):
await self.get_channel(channel_id).send(f"<@&{role_id}> empty medals in late rounds!", embed=e)
else:
await self.get_channel(channel_id).send(embed=e)
sleep_seconds = r.get("last_updated") + 60 - timestamp()
@ -221,119 +174,6 @@ class MyClient(discord.Client):
loop = asyncio.get_event_loop()
client = MyClient()
bot = commands.Bot(command_prefix="!")
@bot.event
async def on_ready():
logger.info("Bot loaded")
# print(bot.user.name)
# print(bot.user.id)
logger.info("------")
@bot.command()
async def unnotify(ctx, kind: str):
if ctx.author.guild_permissions.administrator:
channel_id = ctx.channel.id
if DB.remove_kind_notification_channel(kind, channel_id):
return await ctx.send(f"I wont notify about {kind} in this channel!")
return await ctx.send("Nothing to do...")
@bot.command()
async def notify(ctx, kind: str):
if ctx.author.guild_permissions.administrator:
guild_id = ctx.guild.id
channel_id = ctx.channel.id
if kind == "epic":
if DB.add_notification_channel(guild_id, channel_id, kind):
await ctx.send("I will notify about epics in this channel!")
await ctx.send(
"If You want for me to also add division mentions write:\n"
"`!set_division d1 @role_to_mention`\n"
"`!set_division d2 @role_to_mention`\n"
"`!set_division d3 @role_to_mention`\n"
"`!set_division d4 @role_to_mention`\n"
"`!set_division air @role_to_mention`"
)
elif kind == "events":
DB.add_notification_channel(guild_id, channel_id, kind)
await ctx.send("I will notify about eLatvia's events in this channel!")
elif kind == "empty":
DB.add_notification_channel(guild_id, channel_id, kind)
await ctx.send("I will notify about empty medals in this channel!")
else:
await ctx.send(f"Unknown {kind=}")
else:
return await ctx.send("This command is only available for server administrators")
@bot.command()
async def set_division(ctx, division: str, role_mention):
if not ctx.author.guild_permissions.administrator:
return await ctx.send("This command is only available for server administrators")
if ctx.channel.id not in DB.get_kind_notification_channel_ids("epic"):
return await ctx.send("This command is only available from registered channels!")
div_map = dict(D1=1, D2=3, D3=3, D4=4, Air=11)
if division.title() not in div_map:
return await ctx.send(f"Unknown {division=}! Available divisions {', '.join(div_map.keys())}")
for role in ctx.guild.roles:
if role.mention == role_mention:
DB.add_role_mapping_entry(ctx.channel.id, div_map[division.title()], role.id)
return await ctx.send(f"Success! For {division.title()} epics I will mention <@&{role.id}>")
else:
await ctx.send(f"Unable to find the role You mentioned...")
@bot.command()
async def exit(ctx):
if ctx.author.id == ADMIN_ID:
await ctx.send(f"{ctx.author.mention} Bye!")
sys.exit(0)
else:
return await ctx.send(f"Labs mēģinājums! Mani nogalināt var tikai <@{ADMIN_ID}>")
@bot.command()
async def empty(ctx, division):
if not ctx.channel.id == 603527159109124096:
return await ctx.send("Currently unavailable!")
try:
div = int(division)
except ValueError:
try:
div = dict(D1=1, D2=3, D3=3, D4=4, Air=11)[division.title()]
except (AttributeError, KeyError) as e:
await ctx.send(f"First argument must be a value from: 1, d1, 2, d2, 3, d3, 4, d4, 11, air!")
return
s_div = {1: "D1", 2: "D2", 3: "D3", 4: "D4", 11: "Air"}[div]
embed = discord.Embed(
title=f"Possibly empty {s_div} medals",
description=f"'Empty' medals are being guessed based on the division wall. Expect false-positives!",
)
for kind, div_div, data in check_battles(get_battle_page().get('battles')):
if kind == 'empty' and div_div == div:
embed.add_field(
name=f"**Battle for {data['region']} {' '.join(data['sides'])}**",
value=f"[R{data['zone_id']} | Time {data['round_time']}]({data['url']})",
)
if len(embed.fields) >= 10:
return await ctx.send(embed=embed)
if embed.fields:
return await ctx.send(embed=embed)
else:
return await ctx.send(f"No empty {s_div} medals found")
@empty.error
async def division_error(ctx, error):
if isinstance(error, (commands.BadArgument, commands.MissingRequiredArgument)):
await ctx.send('Division is mandatory, eg, `!empty [1,2,3,4,11, d1,d2,d3,d4,air, D1,D2,D3,D4,Air] [1-120]`')
else:
await ctx.send('Something went wrong! 😔')
logger.exception(error, exc_info=error)
def main():

53
dbot/tests.py
View File

@ -1,7 +1,5 @@
import unittest
import re
from sqlite_utils.db import NotFoundError
import unittest
from dbot import constants, db
@ -11,16 +9,17 @@ class TestDatabase(unittest.TestCase):
self.db = db.DiscordDB()
def test_member(self):
member = {"id": 1200, "name": "username"}
self.db.add_member(**member)
member = {"id": 1200, "name": "username", "pm_is_allowed": False}
self.assertTrue(self.db.add_member(**member))
self.assertEqual(self.db.add_member(**member), member)
self.assertRaises(NotFoundError, self.db.get_member, member_id=100)
self.assertFalse(self.db.get_member(100))
self.assertEqual(self.db.get_member(member_id=member["id"]), member)
member.update(name="Success")
self.assertTrue(self.db.update_member(member["id"], member["name"]))
self.assertEqual(self.db.get_member(member_id=member["id"]), member)
self.assertTrue(self.db.update_member(100, member["name"]))
def test_player(self):
player = {"id": 1, "name": "plato"}
@ -35,31 +34,41 @@ class TestDatabase(unittest.TestCase):
self.assertEqual(self.db.get_player(player["id"]), player)
def test_epic(self):
self.assertFalse(self.db.get_epic(123456))
self.assertFalse(self.db.check_epic(123456))
self.assertTrue(self.db.add_epic(123456))
self.assertFalse(self.db.add_epic(123456))
self.assertTrue(self.db.get_epic(123456))
self.assertTrue(self.db.check_epic(123456))
def test_empty(self):
self.assertFalse(self.db.check_empty_medal(123456))
self.assertTrue(self.db.add_empty_medal(123456))
self.assertFalse(self.db.add_empty_medal(123456))
self.assertTrue(self.db.check_empty_medal(123456))
def test_rss_feed(self):
self.assertEqual(self.db.get_rss_feed_timestamp(71), 0.0)
self.db.set_rss_feed_timestamp(71, 16000000)
self.assertEqual(self.db.get_rss_feed_timestamp(71), 16000000.0)
self.db.set_rss_feed_timestamp(71, 16000001)
self.assertEqual(self.db.get_rss_feed_timestamp(71), 16000001.0)
def test_channels(self):
self.assertTrue(self.db.add_notification_channel(13, 16, "epic"))
self.assertFalse(self.db.add_notification_channel(13, 16, "epic"))
self.assertListEqual(self.db.get_kind_notification_channel_ids("epic"), [16])
self.assertFalse(self.db.add_role_mapping_entry(16, 5, 160003))
self.assertTrue(self.db.add_role_mapping_entry(16, 3, 160003))
self.assertTrue(self.db.add_role_mapping_entry(16, 4, 160003))
self.assertTrue(self.db.add_role_mapping_entry(16, 4, 160004))
self.assertEqual(self.db.get_role_id_for_channel_division(16, 3), 160003)
self.assertEqual(self.db.get_role_id_for_channel_division(16, 4), 160004)
self.assertTrue(self.db.remove_kind_notification_channel("epic", 16))
self.assertFalse(self.db.remove_kind_notification_channel("epic", 16))
self.assertFalse(self.db.get_role_id_for_channel_division(16, 3))
self.assertFalse(self.db.get_role_id_for_channel_division(16, 4))
self.assertFalse(self.db.get_role_id_for_channel_division(16, 5))
kind = "epic"
self.assertTrue(self.db.add_notification_channel(13, 16, kind))
self.assertFalse(self.db.add_notification_channel(13, 16, kind))
self.assertListEqual(self.db.get_kind_notification_channel_ids(kind), [16])
self.assertFalse(self.db.add_role_mapping_entry(kind, 16, 5, 160003))
self.assertTrue(self.db.add_role_mapping_entry(kind, 16, 3, 160003))
self.assertTrue(self.db.add_role_mapping_entry(kind, 16, 4, 160003))
self.assertTrue(self.db.add_role_mapping_entry(kind, 16, 4, 160004))
self.assertEqual(self.db.get_role_id_for_channel_division(kind=kind, channel_id=16, division=3), 160003)
self.assertEqual(self.db.get_role_id_for_channel_division(kind=kind, channel_id=16, division=4), 160004)
self.assertTrue(self.db.remove_role_mapping(kind, 16, 3))
self.assertTrue(self.db.remove_kind_notification_channel(kind, 16))
self.assertFalse(self.db.remove_kind_notification_channel(kind, 16))
self.assertFalse(self.db.get_role_id_for_channel_division(kind=kind, channel_id=16, division=4))
self.assertFalse(self.db.get_role_id_for_channel_division(kind=kind, channel_id=16, division=5))
self.assertRaises(RuntimeError, self.db.get_notification_channel_id, "non-existant")
class TestRegexes(unittest.TestCase):

62
dbot/utils.py
View File

@ -1,12 +1,19 @@
import datetime
from json import JSONDecodeError
from operator import itemgetter
from typing import Union, Dict, Any, Generator, Tuple, List
from typing import Any, Dict, Generator, Tuple, Union
from dbot.constants import UTF_FLAG
import requests as requests
from dbot.base import logger
from dbot.constants import UTF_FLAG, DivisionData
LAST_BATTLE_RESPONSE = None
LAST_BATTLE_UPDATE_TIMESTAMP = 0
def timestamp_to_datetime(timestamp: int) -> datetime.datetime:
return datetime.datetime.fromtimestamp(timestamp)
def timestamp_to_datetime(ts: int) -> datetime.datetime:
return datetime.datetime.fromtimestamp(ts)
def timestamp() -> int:
@ -21,7 +28,7 @@ def s_to_human(seconds: Union[int, float]) -> str:
return f"{h:01d}:{m:02d}:{s:02d}"
def check_battles(battle_json: Dict[str, Dict[str, Any]]) -> Generator[Tuple[str, int, Dict[str, Union[str, List[str], int, Dict[str, Union[str, int]]]]], None, None]:
def check_battles(battle_json: Dict[str, Dict[str, Any]]) -> Generator[Tuple[str, int, DivisionData], None, None]:
for battle in sorted(battle_json.values(), key=itemgetter("start")):
if battle["start"] > timestamp():
continue
@ -29,38 +36,51 @@ def check_battles(battle_json: Dict[str, Dict[str, Any]]) -> Generator[Tuple[str
invader_flag = UTF_FLAG[battle["inv"]["id"]]
defender_flag = UTF_FLAG[battle["def"]["id"]]
for div in battle["div"].values():
if div['end']:
if div["end"]:
continue
division = div['div']
division = div["div"]
dom = div["wall"]["dom"]
epic = div["epic"]
division_meta_data = dict(
division_meta_data = DivisionData(
region=region_name,
round_time=s_to_human(timestamp() - battle["start"]),
round_time_s=int(timestamp() - battle["start"]),
sides=[],
url=f"https://www.erepublik.com/en/military/battlefield/{battle['id']}",
zone_id=battle["zone_id"],
div_id=div['id'],
extra={}
div_id=div["id"],
extra={},
)
if dom == 50:
division_meta_data.update(sides=[invader_flag, defender_flag])
yield 'empty', division, division_meta_data
division_meta_data['sides'].clear()
yield "empty", division, division_meta_data
division_meta_data["sides"].clear()
if dom == 100:
division_meta_data.update(sides=[invader_flag if battle["def"]["id"] == div["wall"]["for"] else defender_flag])
yield 'empty', division, division_meta_data
division_meta_data['sides'].clear()
yield "empty", division, division_meta_data
division_meta_data["sides"].clear()
if epic > 1:
division_meta_data.update(sides=[invader_flag, defender_flag])
division_meta_data['extra'].update(intensity_scale=div['intensity_scale'],
epic_type=epic)
yield 'epic', division, division_meta_data
division_meta_data['sides'].clear()
division_meta_data['extra'].clear()
division_meta_data["extra"].update(intensity_scale=div["intensity_scale"], epic_type=epic)
yield "epic", division, division_meta_data
division_meta_data["sides"].clear()
division_meta_data["extra"].clear()
if dom >= 66.8:
division_meta_data.update(sides=[invader_flag if battle["def"]["id"] == div["wall"]["for"] else defender_flag])
yield 'steal', division, division_meta_data
division_meta_data['sides'].clear()
yield "steal", division, division_meta_data
division_meta_data["sides"].clear()
return
def get_battle_page():
global LAST_BATTLE_UPDATE_TIMESTAMP, LAST_BATTLE_RESPONSE
if int(datetime.datetime.now().timestamp()) >= LAST_BATTLE_UPDATE_TIMESTAMP + 60:
dt = datetime.datetime.now()
r = requests.get("https://www.erepublik.com/en/military/campaignsJson/list")
try:
LAST_BATTLE_RESPONSE = r.json()
except JSONDecodeError:
logger.warning("Received non json response from erep.lv/battles.json!")
return get_battle_page()
LAST_BATTLE_UPDATE_TIMESTAMP = LAST_BATTLE_RESPONSE.get("last_updated", int(dt.timestamp()))
return LAST_BATTLE_RESPONSE

4
lint.sh Normal file
View File

@ -0,0 +1,4 @@
#!/usr/bin/env sh
isort dbot
black dbot
flake8 dbot