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
2025-08-24 01:10:44 +02:00
parent c78a671463
commit 383f28a3f0
1 changed files with 138 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
app.py
View File

@@ -11,6 +11,7 @@ import mysql.connector
from datetime import datetime, timedelta from datetime import datetime, timedelta
from flask_session import Session from flask_session import Session
import logging import logging
import time
app = Flask(__name__) app = Flask(__name__)
app.secret_key = os.getenv("FLASK_SECRET_KEY") app.secret_key = os.getenv("FLASK_SECRET_KEY")
@@ -86,14 +87,68 @@ def stop_bot():
print("Bot läuft nicht.") print("Bot läuft nicht.")
def get_db_connection(): def get_db_connection():
"""Stellt eine Verbindung zur MySQL-Datenbank her.""" """Stellt eine robuste Verbindung zur MySQL-Datenbank her mit Retry-Logik."""
return mysql.connector.connect( retry_count = 3
host=DB_HOST, retry_delay = 1
port=DB_PORT,
user=DB_USER, for attempt in range(retry_count):
password=DB_PASS, try:
database=DB_NAME connection = mysql.connector.connect(
) host=DB_HOST,
port=DB_PORT,
user=DB_USER,
password=DB_PASS,
database=DB_NAME,
autocommit=True,
pool_reset_session=True,
connect_timeout=10,
charset='utf8mb4',
collation='utf8mb4_unicode_ci',
use_unicode=True
)
# Test the connection
if connection.is_connected():
return connection
else:
connection.close()
raise mysql.connector.Error("Connection test failed")
except mysql.connector.Error as e:
print(f"Datenbankverbindung Versuch {attempt + 1} fehlgeschlagen: {e}")
if attempt < retry_count - 1:
time.sleep(retry_delay)
retry_delay *= 2 # Exponential backoff
else:
print("Alle Datenbankverbindungsversuche fehlgeschlagen")
raise
except Exception as e:
print(f"Unerwarteter Fehler bei Datenbankverbindung: {e}")
if attempt < retry_count - 1:
time.sleep(retry_delay)
retry_delay *= 2
else:
raise
def safe_close_connection(connection):
"""Schließt eine Datenbankverbindung sicher."""
try:
if connection and connection.is_connected():
connection.close()
except Exception as e:
print(f"Fehler beim Schließen der Datenbankverbindung: {e}")
class DatabaseConnection:
"""Context Manager für Datenbankverbindungen."""
def __init__(self):
self.connection = None
def __enter__(self):
self.connection = get_db_connection()
return self.connection
def __exit__(self, exc_type, exc_val, exc_tb):
safe_close_connection(self.connection)
def token_updater(token): def token_updater(token):
session['oauth_token'] = token session['oauth_token'] = token
@@ -216,23 +271,29 @@ def load_user_data():
g.is_admin = session.get("is_admin", False) g.is_admin = session.get("is_admin", False)
g.bot_running = bot_status() # Lädt den Bot-Status in g g.bot_running = bot_status() # Lädt den Bot-Status in g
# Hole die Liste der Gilden aus der Datenbank # Hole die Liste der Gilden aus der Datenbank mit robuster Verbindung
connection = get_db_connection() try:
cursor = connection.cursor(dictionary=True) with DatabaseConnection() as connection:
cursor = connection.cursor(dictionary=True)
# Lade die Gilden des Nutzers # Lade die Gilden des Nutzers
user_guilds = session.get("discord_guilds", []) user_guilds = session.get("discord_guilds", [])
user_guild_ids = [guild["id"] for guild in user_guilds] user_guild_ids = [guild["id"] for guild in user_guilds]
# Finde nur die Gilden, die auch in der Datenbank existieren if user_guild_ids:
cursor.execute("SELECT guild_id FROM guilds WHERE guild_id IN (%s)" % ','.join(['%s'] * len(user_guild_ids)), user_guild_ids) # Finde nur die Gilden, die auch in der Datenbank existieren
existing_guilds = cursor.fetchall() cursor.execute("SELECT guild_id FROM guilds WHERE guild_id IN (%s)" % ','.join(['%s'] * len(user_guild_ids)), user_guild_ids)
existing_guilds = cursor.fetchall()
# Filtere die Gilden des Nutzers basierend auf der Existenz in der Datenbank # Filtere die Gilden des Nutzers basierend auf der Existenz in der Datenbank
g.guilds = [guild for guild in user_guilds if int(guild["id"]) in {g["guild_id"] for g in existing_guilds}] g.guilds = [guild for guild in user_guilds if int(guild["id"]) in {g["guild_id"] for g in existing_guilds}]
else:
g.guilds = []
cursor.close() cursor.close()
connection.close() except Exception as e:
print(f"Fehler beim Laden der Benutzerdaten: {e}")
g.guilds = []
else: else:
# Falls der Benutzer nicht eingeloggt ist, keine Daten setzen # Falls der Benutzer nicht eingeloggt ist, keine Daten setzen
g.user_info = None g.user_info = None
@@ -482,30 +543,34 @@ def user_dashboard(guild_id):
# Debugging-Ausgaben # Debugging-Ausgaben
print(f"Accessing user_dashboard for user_id: {user_id}, guild_id: {guild_id}") print(f"Accessing user_dashboard for user_id: {user_id}, guild_id: {guild_id}")
# Hole die serverbezogenen Nutzerdaten try:
connection = get_db_connection() # Hole die serverbezogenen Nutzerdaten
cursor = connection.cursor(dictionary=True) with DatabaseConnection() as connection:
cursor = connection.cursor(dictionary=True)
# Überprüfe, ob der Benutzer Mitglied des Servers (Gilde) ist # Überprüfe, ob der Benutzer Mitglied des Servers (Gilde) ist
cursor.execute("SELECT * FROM user_data WHERE user_id = %s AND guild_id = %s", (user_id, guild_id)) cursor.execute("SELECT * FROM user_data WHERE user_id = %s AND guild_id = %s", (user_id, guild_id))
user_data = cursor.fetchone() user_data = cursor.fetchone()
# Debugging-Ausgabe für user_data # Debugging-Ausgabe für user_data
print(f"user_data for user_id {user_id} on guild_id {guild_id}: {user_data}") print(f"user_data for user_id {user_id} on guild_id {guild_id}: {user_data}")
cursor.close() cursor.close()
connection.close()
if user_data: if user_data:
# Falls `user_data` vorhanden ist, setze `g.guild_id` und `g.user_data` # Falls `user_data` vorhanden ist, setze `g.guild_id` und `g.user_data`
g.guild_id = guild_id g.guild_id = guild_id
g.user_data = user_data g.user_data = user_data
print("Access granted to user_dashboard.") print("Access granted to user_dashboard.")
return render_template("user_dashboard.html") return render_template("user_dashboard.html")
else: else:
# Debugging-Ausgabe für Fehlerfall # Debugging-Ausgabe für Fehlerfall
print(f"No access for user_id {user_id} on guild_id {guild_id}") print(f"No access for user_id {user_id} on guild_id {guild_id}")
flash("You do not have access to this server.", "danger") flash("You do not have access to this server.", "danger")
return redirect(url_for("user_landing_page"))
except Exception as e:
print(f"Fehler beim Laden des User-Dashboards: {e}")
flash("Fehler beim Laden des Dashboards. Bitte versuchen Sie es später erneut.", "error")
return redirect(url_for("user_landing_page")) return redirect(url_for("user_landing_page"))
# Falls der Benutzer nicht eingeloggt ist # Falls der Benutzer nicht eingeloggt ist
@@ -517,38 +582,42 @@ def user_dashboard(guild_id):
def leaderboard(guild_id): def leaderboard(guild_id):
"""Zeigt das Level Leaderboard für einen bestimmten Server an.""" """Zeigt das Level Leaderboard für einen bestimmten Server an."""
if "discord_user" in session: if "discord_user" in session:
connection = get_db_connection() try:
cursor = connection.cursor(dictionary=True) with DatabaseConnection() as connection:
cursor = connection.cursor(dictionary=True)
current_date = datetime.now() current_date = datetime.now()
one_month_ago = current_date - timedelta(days=30) one_month_ago = current_date - timedelta(days=30)
# Hole die Leaderboard-Daten # Hole die Leaderboard-Daten
cursor.execute(""" cursor.execute("""
SELECT nickname, profile_picture, level, xp, join_date SELECT nickname, profile_picture, level, xp, join_date
FROM user_data FROM user_data
WHERE guild_id = %s WHERE guild_id = %s
AND ban = 0 AND ban = 0
AND (leave_date IS NULL OR leave_date > %s) AND (leave_date IS NULL OR leave_date > %s)
ORDER BY level DESC, xp DESC ORDER BY level DESC, xp DESC
""", (guild_id, one_month_ago)) """, (guild_id, one_month_ago))
leaderboard_data = cursor.fetchall() leaderboard_data = cursor.fetchall()
# Hole den Server-Namen aus der guilds-Tabelle # Hole den Server-Namen aus der guilds-Tabelle
cursor.execute("SELECT name FROM guilds WHERE guild_id = %s", (guild_id,)) cursor.execute("SELECT name FROM guilds WHERE guild_id = %s", (guild_id,))
guild_name_result = cursor.fetchone() guild_name_result = cursor.fetchone()
guild_name = guild_name_result["name"] if guild_name_result else f"Server {guild_id}" guild_name = guild_name_result["name"] if guild_name_result else f"Server {guild_id}"
cursor.close() cursor.close()
connection.close()
# Übergabe von enumerate und guild_name an das Template # Übergabe von enumerate und guild_name an das Template
return render_template("leaderboard.html", return render_template("leaderboard.html",
leaderboard=leaderboard_data, leaderboard=leaderboard_data,
guild_id=guild_id, guild_id=guild_id,
guild_name=guild_name, guild_name=guild_name,
enumerate=enumerate) enumerate=enumerate)
except Exception as e:
print(f"Fehler beim Laden des Leaderboards: {e}")
flash("Fehler beim Laden des Leaderboards. Bitte versuchen Sie es später erneut.", "error")
return redirect(url_for("user_landing_page"))
return redirect(url_for("landing_page")) return redirect(url_for("landing_page"))