313 lines
14 KiB
Python
313 lines
14 KiB
Python
import json
|
|
import os
|
|
import sqlite3
|
|
import pickle
|
|
|
|
from action import Action, ActionInteraction
|
|
|
|
|
|
class MessageDB():
|
|
def add_application_msg(msg_id: str, author_id: str, guild_id: str, app_name: str) -> None:
|
|
data = (msg_id, author_id, guild_id, app_name)
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("INSERT INTO app_msg_db VALUES (?, ?, ?, ?)", data)
|
|
con.commit()
|
|
|
|
def get_application_msg(msg_id: str) -> tuple[str, str]:
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute(f"SELECT user_id, guild_id, app_name FROM app_msg_db WHERE msg_id={msg_id}")
|
|
user_id, guild_id, app_name = cur.fetchone()
|
|
return user_id, guild_id, app_name
|
|
|
|
def remove_application_msg(msg_id: str) -> None:
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute(f"DELETE FROM app_msg_db WHERE msg_id={msg_id}")
|
|
con.commit()
|
|
|
|
|
|
class StartButtonDB():
|
|
def add_start_msg(msg_id: str, app_name: str, guild_id: str) -> None:
|
|
data = (msg_id, app_name, guild_id)
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("INSERT INTO app_start_db VALUES (?, ?, ?)", data)
|
|
con.commit()
|
|
|
|
def get_start_msg(msg_id: str) -> tuple[str, str]:
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT app_name, guild_id FROM app_start_db WHERE msg_id=?", (str(msg_id), ))
|
|
app_name, guild_id = cur.fetchone()
|
|
return app_name, guild_id
|
|
|
|
def remove_start_msg(msg_id: str) -> None:
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute(f"DELETE FROM app_start_db WHERE msg_id={msg_id}")
|
|
con.commit()
|
|
|
|
class GuildAppDB():
|
|
def create_guild(guild_id: str, guild_name: str) -> None:
|
|
applications = {}
|
|
application_blob = pickle.dumps(applications)
|
|
data = guild_id, guild_name, application_blob
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("INSERT INTO app_guildapp_db VALUES (?, ?, ?)", data)
|
|
con.commit()
|
|
|
|
def remove_guild(guild_id: str) -> None:
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("DELETE FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
con.commit()
|
|
|
|
def get_all_guilds() -> list:
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT guild_id FROM app_guildapp_db")
|
|
guilds = cur.fetchall()
|
|
guilds = [guild[0] for guild in guilds]
|
|
return guilds
|
|
|
|
def add_application_entry(guild_id: str, application_name: str) -> str:
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name not in applications.keys():
|
|
applications[application_name] = {
|
|
"app_id": "",
|
|
"resp_channel": "",
|
|
"questions": [],
|
|
"actions": []
|
|
}
|
|
application_blob2 = pickle.dumps(applications)
|
|
cur.execute("UPDATE app_guildapp_db SET applications_blob = (?) WHERE guild_id= (?)", (application_blob2, guild_id))
|
|
con.commit()
|
|
return "success"
|
|
else:
|
|
return "error on add application entry: application exists"
|
|
|
|
def remove_application_entry(guild_id: str, application_name: str) -> str:
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name in applications.keys():
|
|
applications.pop(application_name)
|
|
application_blob2 = pickle.dumps(applications)
|
|
cur.execute("UPDATE app_guildapp_db SET applications_blob = (?) WHERE guild_id= (?)", (application_blob2, guild_id))
|
|
con.commit()
|
|
return "success"
|
|
else:
|
|
return "error on remove application entry: application not found"
|
|
|
|
def get_application_entry(guild_id: str, application_name: str):
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name in applications.keys():
|
|
return applications[application_name]
|
|
|
|
def get_applications(guild_id: str):
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
return list(applications.keys())
|
|
|
|
def set_response_channel(guild_id: str, application_name: str, channel_id: str) -> str:
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name in applications.keys():
|
|
applications[application_name]["resp_channel"] = channel_id
|
|
application_blob2 = pickle.dumps(applications)
|
|
cur.execute("UPDATE app_guildapp_db SET applications_blob = (?) WHERE guild_id= (?)", (application_blob2, guild_id))
|
|
con.commit()
|
|
return "success"
|
|
else:
|
|
return "error on set response channel: application not found"
|
|
|
|
def get_response_channel(guild_id: str, application_name: str) -> str:
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name in applications.keys():
|
|
return applications[application_name]["resp_channel"]
|
|
|
|
def add_question(guild_id: str, application_name: str, question: str) -> str:
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name in applications.keys():
|
|
question_index = int(len(applications[application_name]["questions"]))
|
|
applications[application_name]["questions"].append(question)
|
|
application_blob2 = pickle.dumps(applications)
|
|
cur.execute("UPDATE app_guildapp_db SET applications_blob = (?) WHERE guild_id= (?)", (application_blob2, guild_id))
|
|
con.commit()
|
|
return "success"
|
|
else:
|
|
return "error on add question: application not found"
|
|
|
|
def get_questions(guild_id: str, application_name: str):
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name in applications.keys():
|
|
questions = applications[application_name]["questions"]
|
|
return questions, len(questions)
|
|
else:
|
|
return "error on get questions: application not found", ""
|
|
|
|
def edit_question(guild_id: str, application_name: str, question_index: int, new_question: str):
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name in applications.keys():
|
|
questions = applications[application_name]["questions"]
|
|
if question_index <= len(questions):
|
|
questions[question_index-1] = new_question
|
|
applications[application_name]["questions"] = questions
|
|
application_blob2 = pickle.dumps(applications)
|
|
cur.execute("UPDATE app_guildapp_db SET applications_blob = (?) WHERE guild_id= (?)", (application_blob2, guild_id))
|
|
con.commit()
|
|
return "success"
|
|
else:
|
|
return "error on edit question: question index not found"
|
|
else:
|
|
return "error on edit question: application not found"
|
|
|
|
def move_question(guild_id: str, application_name: str, init_que_index: int, fin_que_index: int):
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name in applications.keys():
|
|
questions = applications[application_name]["questions"]
|
|
if init_que_index <= len(questions) and fin_que_index <= len(questions):
|
|
if init_que_index > fin_que_index:
|
|
questions.insert(fin_que_index-1, questions[init_que_index-1])
|
|
questions.pop(init_que_index)
|
|
elif init_que_index < fin_que_index:
|
|
questions.insert(fin_que_index, questions[init_que_index-1])
|
|
questions.pop(init_que_index-1)
|
|
else:
|
|
return "error on move question: init and fin index equal"
|
|
applications[application_name]["questions"] = questions
|
|
application_blob2 = pickle.dumps(applications)
|
|
cur.execute("UPDATE app_guildapp_db SET applications_blob = (?) WHERE guild_id= (?)", (application_blob2, guild_id))
|
|
con.commit()
|
|
return "success"
|
|
else:
|
|
return "error on move question: question index not found"
|
|
else:
|
|
return "error on move question: application not found"
|
|
|
|
def remove_question(guild_id: str, application_name: str, question_index: int):
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name in applications.keys():
|
|
questions = applications[application_name]["questions"]
|
|
if question_index <= len(questions):
|
|
questions.pop(question_index-1)
|
|
applications[application_name]["questions"] = questions
|
|
application_blob2 = pickle.dumps(applications)
|
|
cur.execute("UPDATE app_guildapp_db SET applications_blob = (?) WHERE guild_id= (?)", (application_blob2, guild_id))
|
|
con.commit()
|
|
return "success"
|
|
else:
|
|
return "error on remove question: question index not found"
|
|
else:
|
|
return "error on remove question: application not found"
|
|
|
|
|
|
|
|
|
|
def add_action(guild_id: str, application_name: str, action: dict) -> str:
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name in applications.keys():
|
|
applications[application_name]["actions"].append(action)
|
|
application_blob2 = pickle.dumps(applications)
|
|
cur.execute("UPDATE app_guildapp_db SET applications_blob = (?) WHERE guild_id= (?)", (application_blob2, guild_id))
|
|
con.commit()
|
|
return "success"
|
|
else:
|
|
return "error on add action: application not found"
|
|
|
|
def get_actions(guild_id: str, application_name: str, action_type: ActionInteraction):
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name in applications.keys():
|
|
actions = applications[application_name]["actions"]
|
|
actret = []
|
|
for i in actions:
|
|
if i["result"] == action_type:
|
|
actret.append(i)
|
|
return actret
|
|
else:
|
|
return "error on get actions: application not found"
|
|
|
|
def remove_action(guild_id: str, application_name: str, action_type: ActionInteraction, action_index: int):
|
|
con = sqlite3.connect("applications.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT applications_blob FROM app_guildapp_db WHERE guild_id=(?)", (guild_id, ))
|
|
application_blob = cur.fetchone()
|
|
applications = pickle.loads(application_blob[0])
|
|
if application_name in applications.keys():
|
|
actions = applications[application_name]["actions"]
|
|
actedit = []
|
|
actnoedit = []
|
|
for i in actions:
|
|
if i["result"] == action_type:
|
|
actedit.append(i)
|
|
else:
|
|
actnoedit.append(i)
|
|
actedit
|
|
if action_index <= len(actedit):
|
|
actedit.pop(action_index-1)
|
|
|
|
for x in actedit:
|
|
actnoedit.append(x)
|
|
|
|
applications[application_name]["actions"] = actnoedit
|
|
application_blob2 = pickle.dumps(applications)
|
|
cur.execute("UPDATE app_guildapp_db SET applications_blob = (?) WHERE guild_id= (?)", (application_blob2, guild_id))
|
|
con.commit()
|
|
return "success"
|
|
else:
|
|
return "error on remove action: action index not found"
|
|
else:
|
|
return "error on remove action: application not found"
|
|
|