imcompiling
58b230afe6
I think that the git error is just happening from timing out. Trying to fix by refreshing the connection when it happens.
97 lines
3.5 KiB
Python
97 lines
3.5 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 git.exc import GitCommandError
|
|
from nextcord.ext import commands
|
|
|
|
from dakiny_bot.classes.gitea_webhook import GiteaWebhook
|
|
from dakiny_bot.DakinySettings import settings
|
|
from dakiny_bot.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}
|
|
|
|
try:
|
|
self.origin.pull()
|
|
except GitCommandError:
|
|
# I think the git connection timed out. Get a new one.
|
|
self.repo_setup()
|
|
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))
|