Simon/Discord-ai-chatbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modified: app.py

Browse Source
This commit is contained in:
SimolZimol
2024-10-22 14:45:24 +02:00
parent 458798b8c3
commit d212ebbcd8
1 changed files with 187 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

299
app.py
View File

@@ -9,6 +9,7 @@ import subprocess
import psutil
import mysql.connector
from datetime import datetime
from requests.exceptions import HTTPError
app = Flask(__name__)
app.secret_key = os.getenv("FLASK_SECRET_KEY")
@@ -90,20 +91,21 @@ def make_discord_session(token=None, state=None):
token=token,
state=state,
redirect_uri=DISCORD_REDIRECT_URI,
scope=["identify"]
scope=["identify", "guilds"]
)
def is_admin():
"""Überprüft, ob der Benutzer Admin-Rechte hat."""
if "discord_user" in session:
"""Überprüft, ob der Benutzer Admin-Rechte in der ausgewählten Guild hat."""
if "discord_user" in session and "selected_guild_id" in session:
user_info = session["discord_user"]
user_id = user_info["id"]
guild_id = session["selected_guild_id"]
# Überprüfe die Admin-Rechte des Benutzers
# Überprüfe die Admin-Rechte des Benutzers in der ausgewählten Guild
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT permission FROM user_data WHERE user_id = %s", (user_id,))
cursor.execute("SELECT permission FROM user_data WHERE user_id = %s AND guild_id = %s", (user_id, guild_id))
user_data = cursor.fetchone()
cursor.close()
@@ -151,11 +153,15 @@ def help_page():
def callback():
"""Verarbeitet den OAuth2-Rückruf von Discord."""
discord = make_discord_session(state=session.get("oauth_state"))
token = discord.fetch_token(
DISCORD_TOKEN_URL,
client_secret=DISCORD_CLIENT_SECRET,
authorization_response=request.url,
)
try:
token = discord.fetch_token(
DISCORD_TOKEN_URL,
client_secret=DISCORD_CLIENT_SECRET,
authorization_response=request.url,
)
except HTTPError as e:
flash("OAuth2-Authentifizierung fehlgeschlagen.", "danger")
return redirect(url_for("landing_page"))
session['oauth_token'] = token
@@ -164,63 +170,75 @@ def callback():
# Speichere die Benutzerinformationen in der Session
session['discord_user'] = user_info
# Holen der Guilds, denen der Benutzer angehört
guilds_response = discord.get("https://discord.com/api/users/@me/guilds")
guilds = guilds_response.json()
# Speichern der Guilds in der Session
session['discord_guilds'] = guilds
return redirect(url_for("select_guild"))
# Hole Benutzerrollen und andere Daten aus der Datenbank
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT permission FROM user_data WHERE user_id = %s", (user_info["id"],))
user_data = cursor.fetchone()
cursor.close()
connection.close()
@app.route("/select_guild", methods=["GET", "POST"])
def select_guild():
"""Lässt den Benutzer eine Guild auswählen, die er verwalten möchte."""
if "discord_user" in session and "discord_guilds" in session:
guilds = session["discord_guilds"]
if request.method == "POST":
selected_guild_id = request.form.get("guild_id")
# Überprüfen, ob die ausgewählte Guild in der Liste der Guilds des Benutzers ist
if any(str(guild["id"]) == selected_guild_id for guild in guilds):
session["selected_guild_id"] = int(selected_guild_id)
return redirect(url_for("admin_dashboard"))
else:
flash("Ungültige Guild ausgewählt.", "danger")
return render_template("select_guild.html", guilds=guilds)
return redirect(url_for("landing_page"))
# Weiterleiten basierend auf den Berechtigungen
if user_data and user_data["permission"] >= 8:
return redirect(url_for("admin_dashboard"))
else:
return redirect(url_for("user_dashboard"))
@app.route("/admin_dashboard")
def admin_dashboard():
"""Zeigt das Admin-Dashboard an (nur für Admins)."""
if "discord_user" in session:
"""Zeigt das Admin-Dashboard an (nur für Admins) für die ausgewählte Guild."""
if "discord_user" in session and "selected_guild_id" in session:
user_info = session["discord_user"]
user_id = user_info["id"]
guild_id = session["selected_guild_id"]
# Überprüfe, ob der Benutzer Admin-Rechte hat
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT permission FROM user_data WHERE user_id = %s", (user_id,))
cursor.execute("SELECT permission FROM user_data WHERE user_id = %s AND guild_id = %s", (user_id, guild_id))
user_data = cursor.fetchone()
cursor.close()
connection.close()
if user_data and user_data["permission"] >= 8:
return render_template("admin_dashboard.html", user_info=user_info, bot_running=bot_status())
return render_template("admin_dashboard.html", user_info=user_info, bot_running=bot_status(), guild_id=guild_id)
else:
return redirect(url_for("user_dashboard"))
return redirect(url_for("landing_page"))
@app.route("/user_dashboard")
def user_dashboard():
"""Zeigt das User-Dashboard an."""
if "discord_user" in session:
"""Zeigt das User-Dashboard an für die ausgewählte Guild."""
if "discord_user" in session and "selected_guild_id" in session:
user_info = session["discord_user"]
user_id = user_info["id"]
guild_id = session["selected_guild_id"]
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT points, permission, ban FROM user_data WHERE user_id = %s", (user_id,))
cursor.execute("SELECT points, permission, ban FROM user_data WHERE user_id = %s AND guild_id = %s", (user_id, guild_id))
user_data = cursor.fetchone()
cursor.close()
connection.close()
if user_data:
return render_template("user_dashboard.html", user_info=user_info, user_data=user_data, bot_running=bot_status())
return render_template("user_dashboard.html", user_info=user_info, user_data=user_data, bot_running=bot_status(), guild_id=guild_id)
else:
return "User data not found", 404
return redirect(url_for("landing_page"))
@@ -230,6 +248,8 @@ def logout():
"""Löscht die Benutzersitzung und meldet den Benutzer ab."""
session.pop('discord_user', None)
session.pop('oauth_token', None)
session.pop('discord_guilds', None)
session.pop('selected_guild_id', None)
return redirect(url_for('landing_page'))
@app.route("/start_bot")
@@ -237,133 +257,174 @@ def start():
if is_admin():
start_bot()
user_info = session["discord_user"]
return render_template("admin_dashboard.html", user_info=user_info, bot_running=bot_status())
guild_id = session["selected_guild_id"]
return render_template("admin_dashboard.html", user_info=user_info, bot_running=bot_status(), guild_id=guild_id)
return redirect(url_for("landing_page"))
@app.route("/stop_bot")
def stop():
if is_admin():
stop_bot()
user_info = session["discord_user"]
return render_template("admin_dashboard.html", user_info=user_info, bot_running=bot_status())
guild_id = session["selected_guild_id"]
return render_template("admin_dashboard.html", user_info=user_info, bot_running=bot_status(), guild_id=guild_id)
return redirect(url_for("landing_page"))
@app.route("/settings", methods=["GET", "POST"])
def settings():
if is_admin():
if is_admin() and "selected_guild_id" in session:
if request.method == "POST":
introduction = request.form.get("introduction")
asknotes_introduction = request.form.get("asknotes_introduction")
guild_id = session["selected_guild_id"]
# Speichern der Intros
save_text_file(INTRO_FILE, introduction)
save_text_file(ASKNOTES_INTRO_FILE, asknotes_introduction)
# Speichern der Intros in der Datenbank
connection = get_db_connection()
cursor = connection.cursor()
try:
cursor.execute("""
UPDATE guild_settings
SET introduction = %s, asknotes_introduction = %s
WHERE guild_id = %s
""", (introduction, asknotes_introduction, guild_id))
connection.commit()
flash("Einstellungen erfolgreich aktualisiert!", "success")
except Exception as e:
print(f"Error updating settings: {e}")
connection.rollback()
flash("Fehler beim Aktualisieren der Einstellungen.", "danger")
finally:
cursor.close()
connection.close()
return redirect(url_for("settings"))
# Laden der aktuellen Inhalte aus den Textdateien
introduction = load_text_file(INTRO_FILE)
asknotes_introduction = load_text_file(ASKNOTES_INTRO_FILE)
# Laden der aktuellen Einstellungen aus der Datenbank
guild_id = session["selected_guild_id"]
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT introduction, asknotes_introduction FROM guild_settings WHERE guild_id = %s", (guild_id,))
settings = cursor.fetchone()
cursor.close()
connection.close()
introduction = settings["introduction"] if settings else ""
asknotes_introduction = settings["asknotes_introduction"] if settings else ""
return render_template("settings.html", introduction=introduction, asknotes_introduction=asknotes_introduction)
return redirect(url_for("landing_page"))
@app.route("/users")
def users():
"""Zeigt eine Liste aller Benutzer an."""
if is_admin():
"""Zeigt eine Liste aller Benutzer in der ausgewählten Guild an."""
if is_admin() and "selected_guild_id" in session:
guild_id = session["selected_guild_id"]
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT user_id, permission, points, ban FROM user_data")
cursor.execute("SELECT user_id, permission, points, ban FROM user_data WHERE guild_id = %s", (guild_id,))
users = cursor.fetchall()
cursor.close()
connection.close()
return render_template("users.html", users=users)
return render_template("users.html", users=users, guild_id=guild_id)
return redirect(url_for("landing_page"))
@app.route("/ban_user/<int:user_id>")
def ban_user(user_id):
"""Banned einen Benutzer."""
if is_admin():
"""Bannt einen Benutzer in der ausgewählten Guild."""
if is_admin() and "selected_guild_id" in session:
guild_id = session["selected_guild_id"]
connection = get_db_connection()
cursor = connection.cursor()
try:
cursor.execute("UPDATE user_data SET ban = 1 WHERE user_id = %s", (user_id,))
cursor.execute("UPDATE user_data SET ban = 1 WHERE user_id = %s AND guild_id = %s", (user_id, guild_id))
connection.commit()
return redirect(url_for("users"))
flash("Benutzer erfolgreich gebannt.", "success")
except Exception as e:
print(f"Error banning user: {e}")
connection.rollback()
return redirect(url_for("users"))
flash("Fehler beim Bannen des Benutzers.", "danger")
finally:
cursor.close()
connection.close()
return redirect(url_for("landing_page"))
@app.route("/update_points/<int:user_id>", methods=["POST"])
def update_points(user_id):
"""Aktualisiert die Punkte eines Benutzers."""
if is_admin():
points_change = int(request.form["points_change"])
connection = get_db_connection()
cursor = connection.cursor()
try:
cursor.execute("UPDATE user_data SET points = points + %s WHERE user_id = %s", (points_change, user_id))
connection.commit()
return redirect(url_for("users"))
except Exception as e:
print(f"Error updating points: {e}")
connection.rollback()
return redirect(url_for("users"))
finally:
cursor.close()
connection.close()
return redirect(url_for("users"))
return redirect(url_for("landing_page"))
@app.route("/unban_user/<int:user_id>")
def unban_user(user_id):
"""Entbannt einen Benutzer."""
if is_admin():
"""Entbannt einen Benutzer in der ausgewählten Guild."""
if is_admin() and "selected_guild_id" in session:
guild_id = session["selected_guild_id"]
connection = get_db_connection()
cursor = connection.cursor()
try:
cursor.execute("UPDATE user_data SET ban = 0 WHERE user_id = %s", (user_id,))
cursor.execute("UPDATE user_data SET ban = 0 WHERE user_id = %s AND guild_id = %s", (user_id, guild_id))
connection.commit()
return redirect(url_for("users"))
flash("Benutzer erfolgreich entbannt.", "success")
except Exception as e:
print(f"Error unbanning user: {e}")
connection.rollback()
return redirect(url_for("users"))
flash("Fehler beim Entbannen des Benutzers.", "danger")
finally:
cursor.close()
connection.close()
return redirect(url_for("users"))
return redirect(url_for("landing_page"))
@app.route("/update_points/<int:user_id>", methods=["POST"])
def update_points(user_id):
"""Aktualisiert die Punkte eines Benutzers in der ausgewählten Guild."""
if is_admin() and "selected_guild_id" in session:
points_change = int(request.form["points_change"])
guild_id = session["selected_guild_id"]
connection = get_db_connection()
cursor = connection.cursor()
try:
cursor.execute("UPDATE user_data SET points = points + %s WHERE user_id = %s AND guild_id = %s", (points_change, user_id, guild_id))
connection.commit()
flash("Punkte erfolgreich aktualisiert.", "success")
except Exception as e:
print(f"Error updating points: {e}")
connection.rollback()
flash("Fehler beim Aktualisieren der Punkte.", "danger")
finally:
cursor.close()
connection.close()
return redirect(url_for("users"))
return redirect(url_for("landing_page"))
@app.route("/update_role/<int:user_id>", methods=["POST"])
def update_role(user_id):
"""Aktualisiert die Rolle (Berechtigung) eines Benutzers."""
if is_admin():
new_permission = request.form["permission"]
"""Aktualisiert die Rolle (Berechtigung) eines Benutzers in der ausgewählten Guild."""
if is_admin() and "selected_guild_id" in session:
new_permission = int(request.form["permission"])
guild_id = session["selected_guild_id"]
connection = get_db_connection()
cursor = connection.cursor()
try:
cursor.execute("UPDATE user_data SET permission = %s WHERE user_id = %s", (new_permission, user_id))
cursor.execute("UPDATE user_data SET permission = %s WHERE user_id = %s AND guild_id = %s", (new_permission, user_id, guild_id))
connection.commit()
return redirect(url_for("users"))
flash("Rolle erfolgreich aktualisiert.", "success")
except Exception as e:
print(f"Error updating role: {e}")
connection.rollback()
return redirect(url_for("users"))
flash("Fehler beim Aktualisieren der Rolle.", "danger")
finally:
cursor.close()
connection.close()
return redirect(url_for("users"))
return redirect(url_for("landing_page"))
@app.route("/logs")
@@ -394,8 +455,9 @@ def download_logs():
@app.route("/admin/giveaways", methods=["GET", "POST"])
def admin_giveaways():
"""Zeigt eine Liste aller Giveaways an und ermöglicht das Bearbeiten und Sortieren."""
if is_admin():
"""Zeigt eine Liste aller Giveaways an und ermöglicht das Bearbeiten und Sortieren in der ausgewählten Guild."""
if is_admin() and "selected_guild_id" in session:
guild_id = session["selected_guild_id"]
connection = get_db_connection() # Verbindung zur Giveaway-Datenbank
cursor = connection.cursor(dictionary=True)
@@ -403,20 +465,21 @@ def admin_giveaways():
sort_field = request.args.get("sort", "id") # Standardmäßig nach 'id' sortieren
order = request.args.get("order", "asc") # Standardmäßig aufsteigend sortieren
# Holen aller Giveaways aus der Datenbank
cursor.execute(f"SELECT * FROM giveaway_data ORDER BY {sort_field} {order}")
# Holen aller Giveaways aus der Datenbank für die ausgewählte Guild
cursor.execute(f"SELECT * FROM giveaway_data WHERE guild_id = %s ORDER BY {sort_field} {order}", (guild_id,))
giveaways = cursor.fetchall()
cursor.close()
connection.close()
return render_template("admin_giveaways.html", giveaways=giveaways, sort_field=sort_field, order=order)
return render_template("admin_giveaways.html", giveaways=giveaways, sort_field=sort_field, order=order, guild_id=guild_id)
return redirect(url_for("login"))
@app.route("/admin/giveaways/edit/<int:giveaway_id>", methods=["GET", "POST"])
def edit_giveaway(giveaway_id):
"""Bearbeitet ein spezifisches Giveaway."""
if is_admin():
"""Bearbeitet ein spezifisches Giveaway in der ausgewählten Guild."""
if is_admin() and "selected_guild_id" in session:
guild_id = session["selected_guild_id"]
connection = get_db_connection() # Verbindung zur Giveaway-Datenbank
cursor = connection.cursor(dictionary=True)
@@ -427,78 +490,90 @@ def edit_giveaway(giveaway_id):
winner_dc_id = request.form.get("winner_dc_id")
aktiv = bool(request.form.get("aktiv"))
# Update der Giveaways-Daten
# Update der Giveaways-Daten für die ausgewählte Guild
cursor.execute("""
UPDATE giveaway_data
SET platform = %s, name = %s, game_key = %s, winner_dc_id = %s, aktiv = %s
WHERE id = %s
""", (platform, name, game_key, winner_dc_id, aktiv, giveaway_id))
WHERE id = %s AND guild_id = %s
""", (platform, name, game_key, winner_dc_id, aktiv, giveaway_id, guild_id))
connection.commit()
flash("Giveaway updated successfully!", "success")
flash("Giveaway erfolgreich aktualisiert!", "success")
cursor.close()
connection.close()
return redirect(url_for("admin_giveaways"))
# Daten des spezifischen Giveaways laden
cursor.execute("SELECT * FROM giveaway_data WHERE id = %s", (giveaway_id,))
# Daten des spezifischen Giveaways laden, nur für die ausgewählte Guild
cursor.execute("SELECT * FROM giveaway_data WHERE id = %s AND guild_id = %s", (giveaway_id, guild_id))
giveaway = cursor.fetchone()
cursor.close()
connection.close()
return render_template("edit_giveaway.html", giveaway=giveaway)
return redirect(url_for("login"))
if not giveaway:
flash("Giveaway nicht gefunden oder Zugriff verweigert.", "danger")
return redirect(url_for("admin_giveaways"))
return render_template("edit_giveaway.html", giveaway=giveaway, guild_id=guild_id)
return redirect(url_for("login"))
@app.route("/user/giveaways", methods=["GET"])
def user_giveaways():
"""Zeigt dem Benutzer die Giveaways, die er gewonnen hat."""
if "discord_user" in session:
"""Zeigt dem Benutzer die Giveaways, die er in der ausgewählten Guild gewonnen hat."""
if "discord_user" in session and "selected_guild_id" in session:
user_info = session["discord_user"]
user_id = user_info["id"]
guild_id = session["selected_guild_id"]
connection = get_db_connection() # Verbindung zur Giveaway-Datenbank
cursor = connection.cursor(dictionary=True)
# Suche nach Giveaways, bei denen der eingeloggte Benutzer der Gewinner ist
# Suche nach Giveaways, bei denen der eingeloggte Benutzer der Gewinner ist, in der ausgewählten Guild
cursor.execute("""
SELECT * FROM giveaway_data WHERE winner_dc_id = %s
""", (user_id,))
SELECT * FROM giveaway_data WHERE winner_dc_id = %s AND guild_id = %s
""", (user_id, guild_id))
won_giveaways = cursor.fetchall()
cursor.close()
connection.close()
return render_template("user_giveaways.html", user_info=user_info, giveaways=won_giveaways)
return render_template("user_giveaways.html", user_info=user_info, giveaways=won_giveaways, guild_id=guild_id)
return redirect(url_for("login"))
@app.route("/user/giveaway/redeem/<uuid>", methods=["GET", "POST"])
def redeem_giveaway(uuid):
"""Erlaubt dem Benutzer, den Giveaway-Code abzurufen."""
if "discord_user" in session:
"""Erlaubt dem Benutzer, den Giveaway-Code abzurufen in der ausgewählten Guild."""
if "discord_user" in session and "selected_guild_id" in session:
user_info = session["discord_user"]
user_id = user_info["id"]
guild_id = session["selected_guild_id"]
connection = get_db_connection() # Verbindung zur Giveaway-Datenbank
cursor = connection.cursor(dictionary=True)
# Überprüfen, ob der eingeloggte Benutzer der Gewinner ist
cursor.execute("SELECT * FROM giveaway_data WHERE uuid = %s AND winner_dc_id = %s", (uuid, user_id))
# Überprüfen, ob der eingeloggte Benutzer der Gewinner ist und die Guild stimmt
cursor.execute("SELECT * FROM giveaway_data WHERE uuid = %s AND winner_dc_id = %s AND guild_id = %s", (uuid, user_id, guild_id))
giveaway = cursor.fetchone()
if giveaway:
if request.method == "POST":
# Wenn der Benutzer den Key aufdeckt, setze `aktiv` auf TRUE
cursor.execute("UPDATE giveaway_data SET aktiv = TRUE WHERE uuid = %s", (uuid,))
cursor.execute("UPDATE giveaway_data SET aktiv = TRUE WHERE uuid = %s AND guild_id = %s", (uuid, guild_id))
connection.commit()
# Key aufdecken
return render_template("redeem_giveaway.html", giveaway=giveaway, key=giveaway["game_key"])
key = giveaway["game_key"]
cursor.close()
connection.close()
return render_template("redeem_giveaway.html", giveaway=giveaway, key=key)
# Zeige die Seite mit dem Button an, um den Key aufzudecken
cursor.close()
connection.close()
return render_template("redeem_giveaway.html", giveaway=giveaway, key=None)
else:
flash("You are not the winner of this giveaway or the giveaway is no longer available.", "danger")
flash("Du bist nicht der Gewinner dieses Giveaways oder das Giveaway ist nicht mehr verfügbar.", "danger")
cursor.close()
connection.close()