96 lines
3.1 KiB
Python
96 lines
3.1 KiB
Python
import os
|
|
|
|
import uvicorn
|
|
from fastapi import FastAPI
|
|
from git import Repo
|
|
from nextcord.ext import commands
|
|
|
|
from config import DakinyConfig
|
|
|
|
app = FastAPI()
|
|
REPO_URL = "https://gitea.cheezers.org/imcompiling/dakiny-messages.git"
|
|
REPO_DIR = "./dakiny-messages"
|
|
MESSAGES_DIR = f"{REPO_DIR}/messages"
|
|
|
|
|
|
class ClanMessages(commands.Cog, name="Clan Messages"):
|
|
def __init__(self, bot: commands.Bot):
|
|
self.bot = bot
|
|
self.config: DakinyConfig = self.bot.config
|
|
self.repo_setup()
|
|
|
|
@commands.Cog.listener()
|
|
async def on_ready(self):
|
|
"""
|
|
Triggers when the bot is in ready state. Adds the webhook listener to FastAPI and serves.
|
|
"""
|
|
await self.update_messages()
|
|
app.add_api_route("/message/", self.hello, methods=["POST"])
|
|
config = uvicorn.Config(app=app, port=5000, log_level="info")
|
|
server = uvicorn.Server(config)
|
|
await server.serve()
|
|
|
|
async def hello(self):
|
|
await self.update_messages()
|
|
return
|
|
|
|
async def update_messages(self):
|
|
"""
|
|
_summary_
|
|
"""
|
|
|
|
self.origin.pull()
|
|
|
|
# for ever directory, sort the files by number then post
|
|
# posting involves checking if there is a message for that particular file then editing it or creating a
|
|
# new one. the message could be deleted, which would cause a new message to be sent and database updated.
|
|
dirs = [
|
|
dir_name
|
|
for dir_name in os.listdir(MESSAGES_DIR)
|
|
if os.path.isdir(f"{MESSAGES_DIR}/{dir_name}")
|
|
]
|
|
for dir_name in dirs:
|
|
dir_path = f"{MESSAGES_DIR}/{dir_name}"
|
|
sorted_files = sorted(
|
|
os.listdir(dir_path), key=lambda filename: int(filename.split(".")[0])
|
|
)
|
|
for file_name in sorted_files:
|
|
file_path = f"{dir_path}/{file_name}"
|
|
with open(file_path, "r", encoding="utf8") as reader:
|
|
print(reader.read(2000))
|
|
|
|
# get the channel from dir_name
|
|
# if doesn't exist, make one
|
|
# write or edit message
|
|
# save message id if written
|
|
|
|
def repo_setup(self):
|
|
"""
|
|
Checks if the repo exists and creates repo object, clones if it does not.
|
|
"""
|
|
# check if repo has been cloned. open repo object
|
|
if os.path.isdir(REPO_DIR):
|
|
self.repo = Repo(REPO_DIR)
|
|
else:
|
|
self.repo = Repo.clone_from(REPO_URL, REPO_DIR)
|
|
|
|
self.origin = self.repo.remote("origin")
|
|
|
|
if self.config.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):
|
|
"""Adds function to bot"""
|
|
bot.add_cog(ClanMessages(bot))
|