shop-old/core/python/setup_database.py
2026-04-20 01:03:43 +02:00

123 lines
4.2 KiB
Python

#!/usr/bin/env python3
"""
🗄️ Database Setup Utility für Problem-FAQ System
ACHTUNG: Nur ausführen wenn DB-Zugang verfügbar!
"""
import mysql.connector
import json
import sys
from config import DB_CONFIG
def create_problem_database():
"""Erstellt die Problem-Datenbank und Tabellen"""
try:
# Verbindung ohne spezifische Datenbank
temp_config = DB_CONFIG.copy()
temp_config.pop('database', None)
connection = mysql.connector.connect(**temp_config)
cursor = connection.cursor()
# Datenbank erstellen
cursor.execute(f"CREATE DATABASE IF NOT EXISTS {DB_CONFIG['database']} CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
cursor.execute(f"USE {DB_CONFIG['database']}")
print(f"✅ Datenbank '{DB_CONFIG['database']}' erstellt/ausgewählt")
# SQL-Script ausführen
with open('/var/www/vhosts/intelectra.de/httpdocs/core/python/create_problem_database.sql', 'r', encoding='utf-8') as f:
sql_script = f.read()
# Script in einzelne Statements aufteilen
statements = [stmt.strip() for stmt in sql_script.split(';') if stmt.strip()]
for statement in statements:
if statement.upper().startswith(('CREATE', 'INSERT')):
try:
cursor.execute(statement)
print(f"✅ SQL Statement ausgeführt: {statement[:50]}...")
except mysql.connector.Error as err:
print(f"⚠️ Warning: {err} für Statement: {statement[:50]}...")
connection.commit()
cursor.close()
connection.close()
print("\n🎉 Database Setup erfolgreich abgeschlossen!")
return True
except mysql.connector.Error as err:
print(f"❌ Database Error: {err}")
return False
except Exception as e:
print(f"💥 Setup Error: {str(e)}")
return False
def test_database_connection():
"""Testet die Datenbankverbindung"""
try:
connection = mysql.connector.connect(**DB_CONFIG)
cursor = connection.cursor()
# Test-Query
cursor.execute("SELECT COUNT(*) FROM problem_solutions WHERE active = TRUE")
count = cursor.fetchone()[0]
cursor.close()
connection.close()
print(f"✅ Database Connection OK - {count} aktive Probleme gefunden")
return True
except Exception as e:
print(f"❌ Database Connection Failed: {str(e)}")
return False
def show_sample_problems():
"""Zeigt Beispiel-Probleme aus der Datenbank"""
try:
connection = mysql.connector.connect(**DB_CONFIG)
cursor = connection.cursor(dictionary=True)
cursor.execute("""
SELECT problem_title, problem_category, urgency_level, frequency_score
FROM problem_solutions
WHERE active = TRUE
ORDER BY frequency_score DESC
LIMIT 5
""")
problems = cursor.fetchall()
print("\n📊 Top 5 Probleme in der Datenbank:")
print("-" * 50)
for i, problem in enumerate(problems, 1):
print(f"{i}. {problem['problem_title']}")
print(f" Kategorie: {problem['problem_category']}")
print(f" Urgency: {problem['urgency_level']}")
print(f" Häufigkeit: {problem['frequency_score']}%")
print()
cursor.close()
connection.close()
except Exception as e:
print(f"❌ Error showing problems: {str(e)}")
if __name__ == "__main__":
print("🗄️ Problem-FAQ Database Setup")
print("=" * 40)
if len(sys.argv) > 1 and sys.argv[1] == "--create":
print("🔨 Creating database and tables...")
if create_problem_database():
test_database_connection()
show_sample_problems()
else:
print("🧪 Testing existing database connection...")
if test_database_connection():
show_sample_problems()
else:
print("\n💡 Tipp: Verwende --create um die Datenbank zu erstellen")
print(" Beispiel: python3 setup_database.py --create")