92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
"""File contains logic for creating a webhook listener, pulling from git, and updating all messages."""
|
|
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
import uvicorn
|
|
from fastapi import FastAPI
|
|
from git import Repo
|
|
from nextcord.ext import commands
|
|
|
|
from classes.gitea_webhook import GiteaWebhook
|
|
from DakinySettings import settings
|
|
from utils import REPO_DIR, edit_clan_council_info, update_clan_messages
|
|
|
|
if TYPE_CHECKING:
|
|
import nextcord
|
|
|
|
app = FastAPI()
|
|
REPO_URL = "https://gitea.cheezers.org/imcompiling/dakiny-messages.git"
|
|
|
|
|
|
class ClanMessages(commands.Cog, name="Clan Messages"):
|
|
"""
|
|
Cog to be loaded into the bot.
|
|
|
|
Used to be able to have a fastapi post listener and access to the bot at
|
|
the same time.
|
|
"""
|
|
|
|
def __init__(self, bot: commands.Bot):
|
|
self.bot = bot
|
|
self.repo_setup()
|
|
|
|
@commands.Cog.listener()
|
|
async def on_ready(self) -> None:
|
|
"""Triggers when the bot is in ready state. Adds the webhook listener to FastAPI and serves."""
|
|
app.add_api_route("/messages/", self.update_messages, methods=["POST"])
|
|
# Bind to all interfaces, don't know the docker interface ahead of time.
|
|
config = uvicorn.Config(app=app, host="0.0.0.0", port=5000, log_level="info") # noqa: S104
|
|
server = uvicorn.Server(config)
|
|
await server.serve()
|
|
|
|
async def update_messages(self, hook_info: GiteaWebhook) -> dict:
|
|
"""
|
|
Triggered by webhook.
|
|
|
|
Pulls from git and then calls method to update all the discord messages as described by the
|
|
files in the git repo.
|
|
"""
|
|
# Get all modified/added files
|
|
modified_files: list[str] = [
|
|
file for files in [commit.modified + commit.added for commit in hook_info.commits] for file in files
|
|
]
|
|
|
|
template_updates = {file.split("/")[1] for file in modified_files if "message_templates" in file}
|
|
channel_updates = {file.split("/")[1] for file in modified_files if file.split("/")[1] not in template_updates}
|
|
|
|
self.origin.pull()
|
|
guild: nextcord.Guild = self.bot.get_guild(settings.params.guild_id)
|
|
|
|
await update_clan_messages(guild, channel_updates)
|
|
|
|
# if the council members template has changed, update the message
|
|
if "council_members.txt" in template_updates:
|
|
await edit_clan_council_info(self.bot.get_guild(settings.params.guild_id))
|
|
|
|
return {"message": "Successfully updated all discord messages."}
|
|
|
|
def repo_setup(self) -> None:
|
|
"""Checks if the repo exists and creates repo object, clones if it does not."""
|
|
# check if repo has been cloned. open repo object
|
|
if Path(REPO_DIR).is_dir():
|
|
self.repo = Repo(REPO_DIR)
|
|
else:
|
|
self.repo = Repo.clone_from(REPO_URL, REPO_DIR)
|
|
|
|
self.origin = self.repo.remote("origin")
|
|
|
|
if settings.params.git_branch == "test":
|
|
self.repo.create_head("test", self.origin.refs.test).set_tracking_branch(self.origin.refs.test).checkout()
|
|
else:
|
|
self.repo.create_head("main", self.origin.refs.main).set_tracking_branch(self.origin.refs.main).checkout()
|
|
|
|
self.origin.pull()
|
|
|
|
|
|
# This function will be called when this extension is loaded.
|
|
# It is necessary to add these functions to the bot.
|
|
def setup(bot: commands.Bot) -> None:
|
|
"""Adds function to bot."""
|
|
bot.add_cog(ClanMessages(bot))
|