RcGcDb/src/bot.py

import logging.config
from src.config import settings
import sqlite3
import sys
import signal
from src.wiki import Wiki, process_cats, process_mwmsgs, essential_info
import asyncio, aiohttp
from src.misc import get_paths
from src.exceptions import *
from src.database import db_cursor
from collections import defaultdict
from src.queue_handler import DBHandler
from src.discord import DiscordMessage
from src.msgqueue import messagequeue
import requests

logging.config.dictConfig(settings["logging"])
logger = logging.getLogger("rcgcdb.bot")
logger.debug("Current settings: {settings}".format(settings=settings))

# Log Fail states with structure wiki_url: number of fail states
all_wikis: dict = {}
mw_msgs: dict = {}  # will have the type of id: tuple

# First populate the all_wikis list with every wiki
# Reasons for this: 1. we require amount of wikis to calculate the cooldown between requests
# 2. Easier to code

for wiki in db_cursor.execute('SELECT DISTINCT wiki FROM rcgcdw'):
	all_wikis[wiki] = Wiki()


# Start queueing logic


def calculate_delay() -> float:
	"""Calculate the delay between fetching each wiki to avoid rate limits"""
	min_delay = 60 / settings["max_requests_per_minute"]
	if (len(all_wikis) * min_delay) < settings["minimal_cooldown_per_wiki_in_sec"]:
		return settings["minimal_cooldown_per_wiki_in_sec"] / len(all_wikis)
	else:
		return min_delay


def generate_targets(wiki_url: str) -> defaultdict:
	"""To minimize the amount of requests, we generate a list of language/display mode combinations to create messages for
	this way we can send the same message to multiple webhooks which have the same wiki and settings without doing another
	request to the wiki just to duplicate the message.
	"""
	combinations = defaultdict(list)
	for webhook in db_cursor.execute('SELECT webhook, lang, display FROM rcgcdw WHERE wiki = ?', (wiki_url,)):
		combination = (webhook["lang"], webhook["display"])
		combinations[combination].append(webhook["webhook"])
	return combinations


async def wiki_scanner():
	try:
		while True:
			calc_delay = calculate_delay()
			fetch_all = db_cursor.execute('SELECT webhook, wiki, lang, display, wikiid, rcid, postid FROM rcgcdw GROUP BY wiki')
			for db_wiki in fetch_all.fetchall():
				logger.debug("Wiki {}".format(db_wiki["wiki"]))
				extended = False
				if db_wiki["wiki"] not in all_wikis:
					logger.debug("New wiki: {}".format(db_wiki["wiki"]))
					all_wikis[db_wiki["wiki"]] = Wiki()
				local_wiki = all_wikis[db_wiki["wiki"]]  # set a reference to a wiki object from memory
				if local_wiki.mw_messages is None:
					extended = True
				async with aiohttp.ClientSession(headers=settings["header"],
				                                 timeout=aiohttp.ClientTimeout(2.0)) as session:
					try:
						wiki_response = await local_wiki.fetch_wiki(extended, db_wiki["wiki"], session)
						await local_wiki.check_status(db_wiki["wiki"], wiki_response.status)
					except (WikiServerError, WikiError):
						logger.exception("Exeption when fetching the wiki")
						continue  # ignore this wiki if it throws errors
					try:
						recent_changes_resp = await wiki_response.json()
						if "error" in recent_changes_resp or "errors" in recent_changes_resp:
							error = recent_changes_resp.get("error", recent_changes_resp["errors"])
							if error["code"] == "readapidenied":
								await local_wiki.fail_add(db_wiki["wiki"], 410)
								continue
							raise WikiError
						recent_changes = recent_changes_resp['query']['recentchanges']
						recent_changes.reverse()
					except aiohttp.ContentTypeError:
						logger.exception("Wiki seems to be resulting in non-json content.")
						await local_wiki.fail_add(db_wiki["wiki"], 410)
						continue
					except:
						logger.exception("On loading json of response.")
						continue
				if extended:
					await process_mwmsgs(recent_changes_resp, local_wiki, mw_msgs)
				if db_wiki["rcid"] is None:  # new wiki, just get the last rc to not spam the channel
					if len(recent_changes) > 0:
						DBHandler.add(db_wiki["wiki"], recent_changes[-1]["rcid"])
					else:
						DBHandler.add(db_wiki["wiki"], 0)
					DBHandler.update_db()
					continue
				categorize_events = {}
				targets = generate_targets(db_wiki["wiki"])
				paths = get_paths(db_wiki["wiki"], recent_changes_resp)
				for change in recent_changes:
					await process_cats(change, local_wiki, mw_msgs, categorize_events)
				for change in recent_changes:  # Yeah, second loop since the categories require to be all loaded up
					if change["rcid"] > db_wiki["rcid"]:
						for target in targets.items():
							await essential_info(change, categorize_events, local_wiki, db_wiki, target, paths,
							                     recent_changes_resp)
				if recent_changes:
					DBHandler.add(db_wiki["wiki"], change["rcid"])
				DBHandler.update_db()
				await asyncio.sleep(delay=calc_delay)
	except asyncio.CancelledError:
		raise


async def message_sender():
	while True:
		await messagequeue.resend_msgs()


def shutdown(loop, signal=None):
	DBHandler.update_db()
	loop.stop()
	logger.info("Script has shut down due to signal {}.".format(signal))
	for task in asyncio.all_tasks(loop):
		logger.debug("Killing task {}".format(task.get_name()))
		task.cancel()
	sys.exit(0)

def global_exception_handler(loop, context):
	"""Global exception handler for asyncio, lets us know when something crashes"""
	msg = context.get("exception", context["message"])
	logger.error("Global exception handler:" + msg)
	requests.post("https://discord.com/api/webhooks/"+settings["monitoring_webhook"], data={"content": "test"})

async def main_loop():
	loop = asyncio.get_event_loop()
	try:
		signals = (signal.SIGHUP, signal.SIGTERM, signal.SIGINT)
		for s in signals:
			loop.add_signal_handler(
				s, lambda s=s: shutdown(loop, signal=s))
	except AttributeError:
		logger.info("Running on Windows huh? This complicates things")
		signals = (signal.SIGBREAK, signal.SIGTERM, signal.SIGINT)
	loop.set_exception_handler(global_exception_handler)
	try:
		task1 = asyncio.create_task(wiki_scanner())
		task2 = asyncio.create_task(message_sender())
		await task1
		await task2
	except KeyboardInterrupt:
		shutdown(loop)


asyncio.run(main_loop())