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 17:50:20 +02:00
parent 5ba772d558
commit 70e562266d
1 changed files with 107 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

205
app.py
View File

@@ -1,25 +1,22 @@
__version__ = "dev-0.4.6"
__all__ = ["Discordbot-chatai-webpanel (Discord)"]
__author__ = "SimolZimol"
from flask import Flask, render_template, redirect, url_for, request, session, jsonify, flash
from flask import Flask, render_template, redirect, url_for, request, session, jsonify, send_file, flash
from requests_oauthlib import OAuth2Session
import os
import subprocess
import mysql.connector
from datetime import datetime
import subprocess
app = Flask(__name__)
app.secret_key = os.getenv("FLASK_SECRET_KEY")
# DB configuration
LOG_FILE_PATH = os.path.join("logs", f"{datetime.now().strftime('%Y-%m-%d')}.log")
# Verwende Umgebungsvariablen aus Coolify für die Datenbankverbindung
DB_HOST = os.getenv("DB_HOST")
DB_PORT = os.getenv("DB_PORT")
DB_USER = os.getenv("DB_USER")
DB_PASS = os.getenv("DB_PASSWORD")
DB_NAME = os.getenv("DB_DATABASE")
# Discord OAuth2 configuration
DISCORD_CLIENT_ID = os.getenv("DISCORD_CLIENT_ID")
DISCORD_CLIENT_SECRET = os.getenv("DISCORD_CLIENT_SECRET")
DISCORD_REDIRECT_URI = os.getenv("DISCORD_REDIRECT_URI")
@@ -28,11 +25,35 @@ DISCORD_TOKEN_URL = "https://discord.com/api/oauth2/token"
DISCORD_API_URL = "https://discord.com/api/users/@me"
os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'
# Global Bot Management
bot_process = None
def bot_status():
"""Überprüft, ob der Bot läuft."""
global bot_process
if bot_process is None:
return False
return bot_process.poll() is None # None bedeutet, dass der Prozess noch läuft
def start_bot():
"""Startet den Bot."""
global bot_process
if not bot_status():
bot_process = subprocess.Popen(["python", "bot.py"], cwd=os.path.dirname(os.path.abspath(__file__)))
else:
print("Bot läuft bereits.")
def stop_bot():
"""Stoppt den Bot."""
global bot_process
if bot_process and bot_status():
bot_process.terminate()
bot_process.wait()
bot_process = None
else:
print("Bot läuft nicht.")
def get_db_connection():
"""Returns a MySQL connection."""
"""Stellt eine Verbindung zur MySQL-Datenbank her."""
return mysql.connector.connect(
host=DB_HOST,
port=DB_PORT,
@@ -50,149 +71,140 @@ def make_discord_session(token=None, state=None):
scope=["identify"]
)
def bot_status():
"""Checks if the bot is running."""
global bot_process
if bot_process is None:
return False
return bot_process.poll() is None # None means the bot is still running
def start_bot():
"""Starts the bot process."""
global bot_process
if not bot_status():
bot_process = subprocess.Popen(["python", "bot.py"])
else:
print("Bot is already running.")
def stop_bot():
"""Stops the bot process."""
global bot_process
if bot_process and bot_status():
bot_process.terminate()
bot_process.wait()
bot_process = None
else:
print("Bot is not running.")
def is_global_admin():
"""Checks if the user has global admin privileges."""
"""Überprüft, ob der Benutzer globale Admin-Rechte hat."""
if "discord_user" in session:
user_id = session["discord_user"]["id"]
user_info = session["discord_user"]
user_id = user_info["id"]
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT global_permission FROM bot_data WHERE user_id = %s", (user_id,))
user_data = cursor.fetchone()
bot_data = cursor.fetchone()
cursor.close()
connection.close()
if user_data and user_data["global_permission"] >= 8:
if bot_data and bot_data["global_permission"] >= 8:
return True
return False
def is_server_admin(guild_id):
"""Checks if the user is an admin of the specified server."""
"""Überprüft, ob der Benutzer Admin-Rechte auf einem bestimmten Server hat."""
if "discord_user" in session:
user_id = session["discord_user"]["id"]
user_info = session["discord_user"]
user_id = user_info["id"]
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
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 True
return False
@app.route("/")
def landing_page():
"""Displays the landing page."""
"""Ungeschützte Landing Page"""
return render_template("landing.html")
@app.route("/login")
def login():
"""Initiates the Discord OAuth2 login flow."""
"""Startet den Discord-OAuth2-Flow."""
discord = make_discord_session()
authorization_url, state = discord.authorization_url(DISCORD_OAUTH2_URL)
session['oauth_state'] = state
return redirect(authorization_url)
@app.route("/callback")
def callback():
"""Processes the Discord OAuth2 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)
token = discord.fetch_token(
DISCORD_TOKEN_URL,
client_secret=DISCORD_CLIENT_SECRET,
authorization_response=request.url,
)
session['oauth_token'] = token
user_info = discord.get(DISCORD_API_URL).json()
session['discord_user'] = user_info
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT global_permission FROM bot_data WHERE user_id = %s", (user_info["id"],))
global_user_data = cursor.fetchone()
bot_data = cursor.fetchone()
cursor.close()
connection.close()
if global_user_data and global_user_data["global_permission"] >= 8:
return redirect(url_for("admin_dashboard"))
if bot_data and bot_data["global_permission"] >= 8:
return redirect(url_for("global_admin_dashboard"))
else:
return redirect(url_for("server_select"))
return redirect(url_for("select_guild"))
@app.route("/server_select")
def server_select():
"""Displays a list of servers for the user to select from."""
@app.route("/select_guild")
def select_guild():
"""Lässt den Benutzer den Server auswählen, auf dem er aktiv sein möchte."""
if "discord_user" in session:
user_id = session["discord_user"]["id"]
user_info = session["discord_user"]
user_id = user_info["id"]
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT DISTINCT guild_id FROM user_data WHERE user_id = %s", (user_id,))
cursor.execute("SELECT guild_id FROM user_data WHERE user_id = %s", (user_id,))
guilds = cursor.fetchall()
cursor.close()
connection.close()
return render_template("server_select.html", guilds=guilds)
return redirect(url_for("login"))
if guilds:
return render_template("select_guild.html", guilds=guilds)
else:
return "No server found for this user.", 404
return redirect(url_for("landing_page"))
@app.route("/set_guild/<int:guild_id>")
def set_guild(guild_id):
"""Setzt den aktuellen Server (guild_id) für den Benutzer."""
session["guild_id"] = guild_id
return redirect(url_for("server_dashboard", guild_id=guild_id))
@app.route("/global_admin_dashboard")
def global_admin_dashboard():
"""Globales Admin-Dashboard für globale Administratoren."""
if is_global_admin():
user_info = session["discord_user"]
return render_template("global_admin_dashboard.html", user_info=user_info, bot_running=bot_status())
return redirect(url_for("landing_page"))
@app.route("/server_dashboard/<int:guild_id>")
def server_dashboard(guild_id):
"""Displays the server dashboard for the selected server."""
if "discord_user" in session:
if is_server_admin(guild_id):
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT * FROM user_data WHERE guild_id = %s", (guild_id,))
user_data = cursor.fetchall()
cursor.close()
connection.close()
return render_template("server_dashboard.html", guild_id=guild_id, user_data=user_data)
return "Access denied", 403
return redirect(url_for("login"))
@app.route("/admin_dashboard")
def admin_dashboard():
"""Displays the global admin dashboard."""
if "discord_user" in session and is_global_admin():
return render_template("admin_dashboard.html", bot_running=bot_status())
return redirect(url_for("login"))
"""Server-spezifisches Dashboard für den aktuellen Server (guild_id)."""
if is_server_admin(guild_id):
user_info = session["discord_user"]
return render_template("server_dashboard.html", user_info=user_info, guild_id=guild_id, bot_running=bot_status())
return redirect(url_for("landing_page"))
@app.route("/start_bot")
def start():
"""Starts the bot (global admin only)."""
if is_global_admin():
start_bot()
return redirect(url_for("admin_dashboard"))
return redirect(url_for("login"))
return redirect(url_for("global_admin_dashboard"))
return redirect(url_for("landing_page"))
@app.route("/stop_bot")
def stop():
"""Stops the bot (global admin only)."""
if is_global_admin():
stop_bot()
return redirect(url_for("admin_dashboard"))
return redirect(url_for("login"))
return redirect(url_for("global_admin_dashboard"))
return redirect(url_for("landing_page"))
@app.route("/giveaways/<int:guild_id>")
def view_giveaways(guild_id):
"""Displays the giveaways for the specified guild."""
@app.route("/server_admin/giveaways/<int:guild_id>")
def server_admin_giveaways(guild_id):
"""Zeigt eine Liste der Giveaways für einen bestimmten Server an."""
if is_server_admin(guild_id):
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
@@ -200,26 +212,23 @@ def view_giveaways(guild_id):
giveaways = cursor.fetchall()
cursor.close()
connection.close()
return render_template("giveaways.html", giveaways=giveaways, guild_id=guild_id)
return redirect(url_for("login"))
return render_template("server_admin_giveaways.html", giveaways=giveaways, guild_id=guild_id)
return redirect(url_for("landing_page"))
@app.route("/user_dashboard/<int:guild_id>")
def user_dashboard(guild_id):
"""Displays the user dashboard for the selected guild."""
if "discord_user" in session:
user_id = session["discord_user"]["id"]
@app.route("/users")
def users():
"""Zeigt die Liste der Benutzer an."""
if is_global_admin():
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT * FROM user_data WHERE user_id = %s AND guild_id = %s", (user_id, guild_id))
user_data = cursor.fetchone()
cursor.execute("SELECT user_id, permission, points, ban FROM user_data")
users = cursor.fetchall()
cursor.close()
connection.close()
return render_template("users.html", users=users)
return redirect(url_for("landing_page"))
if user_data:
return render_template("user_dashboard.html", user_data=user_data)
else:
return "No data found", 404
return redirect(url_for("login"))
# Weitere Routen für Server-spezifische Funktionen, Punkteverwaltung, Giveaways etc.
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=True)