From c78a671463d9eed81f938fb1281e110f540d8373 Mon Sep 17 00:00:00 2001 From: SimolZimol <70102430+SimolZimol@users.noreply.github.com> Date: Sun, 24 Aug 2025 01:06:14 +0200 Subject: [PATCH] modified: app.py --- app.py | 143 +++++++++------------------------------------------------ 1 file changed, 21 insertions(+), 122 deletions(-) diff --git a/app.py b/app.py index fb33e1b..6738a60 100644 --- a/app.py +++ b/app.py @@ -85,78 +85,15 @@ def stop_bot(): else: print("Bot läuft nicht.") -def test_db_connection(): - """Testet die Datenbankverbindung.""" - try: - connection = get_db_connection() - if connection.is_connected(): - print("Datenbankverbindung erfolgreich!") - connection.close() - return True - except Exception as err: - print(f"Datenbankverbindungstest fehlgeschlagen: {err}") - return False - return False - -def safe_db_operation(operation_func, *args, **kwargs): - """Führt eine Datenbankoperation sicher aus mit Fehlerbehandlung.""" - try: - connection = get_db_connection() - cursor = connection.cursor(dictionary=True) - - result = operation_func(cursor, *args, **kwargs) - - cursor.close() - connection.close() - - return result - except mysql.connector.Error as err: - print(f"Datenbankfehler: {err}") - return None - except Exception as err: - print(f"Unerwarteter Fehler bei Datenbankoperation: {err}") - return None - def get_db_connection(): - """Stellt eine Verbindung zur MySQL-Datenbank her mit Retry-Mechanismus.""" - import time - max_retries = 3 - retry_delay = 1 # Sekunden - - for attempt in range(max_retries): - try: - connection = mysql.connector.connect( - host=DB_HOST, - port=DB_PORT, - user=DB_USER, - password=DB_PASS, - database=DB_NAME, - autocommit=True, - connection_timeout=10, - reconnect=True, - charset='utf8mb4', - collation='utf8mb4_unicode_ci' - ) - # Test the connection - if connection.is_connected(): - return connection - except mysql.connector.Error as err: - print(f"Datenbankverbindung Versuch {attempt + 1} fehlgeschlagen: {err}") - if attempt < max_retries - 1: - time.sleep(retry_delay) - retry_delay *= 2 # Exponential backoff - else: - print("Alle Datenbankverbindungsversuche fehlgeschlagen") - raise - except Exception as err: - print(f"Unerwarteter Fehler bei Datenbankverbindung: {err}") - if attempt < max_retries - 1: - time.sleep(retry_delay) - retry_delay *= 2 - else: - raise - - raise mysql.connector.Error("Konnte keine Datenbankverbindung herstellen") + """Stellt eine Verbindung zur MySQL-Datenbank her.""" + return mysql.connector.connect( + host=DB_HOST, + port=DB_PORT, + user=DB_USER, + password=DB_PASS, + database=DB_NAME + ) def token_updater(token): session['oauth_token'] = token @@ -280,33 +217,22 @@ def load_user_data(): g.bot_running = bot_status() # Lädt den Bot-Status in g # Hole die Liste der Gilden aus der Datenbank - try: - connection = get_db_connection() - cursor = connection.cursor(dictionary=True) - - # Lade die Gilden des Nutzers - user_guilds = session.get("discord_guilds", []) - user_guild_ids = [guild["id"] for guild in user_guilds] + connection = get_db_connection() + cursor = connection.cursor(dictionary=True) + + # Lade die Gilden des Nutzers + user_guilds = session.get("discord_guilds", []) + user_guild_ids = [guild["id"] for guild in user_guilds] - if user_guild_ids: - # Finde nur die Gilden, die auch in der Datenbank existieren - 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() + # Finde nur die Gilden, die auch in der Datenbank existieren + 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 - g.guilds = [guild for guild in user_guilds if int(guild["id"]) in {g["guild_id"] for g in existing_guilds}] - else: - g.guilds = [] + # 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}] - cursor.close() - connection.close() - - except mysql.connector.Error as err: - print(f"Datenbankfehler in load_user_data: {err}") - g.guilds = [] # Fallback auf leere Liste - except Exception as err: - print(f"Unerwarteter Fehler in load_user_data: {err}") - g.guilds = [] # Fallback auf leere Liste + cursor.close() + connection.close() else: # Falls der Benutzer nicht eingeloggt ist, keine Daten setzen g.user_info = None @@ -881,38 +807,11 @@ def stop(): return redirect(url_for("global_admin_dashboard")) return redirect(url_for("landing_page")) -@app.route("/health") -def health_check(): - """Health-Check-Endpunkt für die Anwendung und Datenbank.""" - health_status = { - "status": "healthy", - "timestamp": datetime.now().isoformat(), - "database": "disconnected", - "version": __version__ - } - - # Teste die Datenbankverbindung - if test_db_connection(): - health_status["database"] = "connected" - return jsonify(health_status), 200 - else: - health_status["status"] = "unhealthy" - health_status["database"] = "error" - return jsonify(health_status), 503 - if __name__ == "__main__": # Disable default Flask logging for static files app.logger.disabled = True log = logging.getLogger('werkzeug') log.disabled = True - # Test database connection on startup - print("Teste Datenbankverbindung beim Start...") - if test_db_connection(): - print("✅ Datenbankverbindung erfolgreich!") - else: - print("❌ Datenbankverbindung fehlgeschlagen! Überprüfe die Konfiguration.") - print("Die Anwendung wird trotzdem gestartet, aber Datenbankfunktionen sind nicht verfügbar.") - # Start app with minimal logging app.run(host="0.0.0.0", port=5000, debug=True)