|
|
|
|
|
|
|
|
|
from flask import Flask, request, jsonify |
|
import os |
|
import logging |
|
from waitress import serve |
|
from nacl.signing import VerifyKey |
|
from nacl.exceptions import BadSignatureError |
|
import threading |
|
import requests |
|
import time |
|
|
|
|
|
logging.basicConfig( |
|
level=logging.INFO, |
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' |
|
) |
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
PUBLIC_KEY = os.getenv('Public_Key', '').strip() |
|
APPLICATION_ID = os.getenv('Application_ID', '').strip() |
|
BOT_TOKEN = os.getenv('BOT_TOKEN', '').strip() |
|
PORT = int(os.getenv('PORT', 7860)) |
|
|
|
app = Flask(__name__) |
|
|
|
|
|
try: |
|
verify_key = VerifyKey(bytes.fromhex(PUBLIC_KEY)) if PUBLIC_KEY else None |
|
logger.info("Successfully initialized verify_key") |
|
except Exception as e: |
|
logger.error(f"Error initializing verify_key: {str(e)}") |
|
verify_key = None |
|
|
|
def verify_discord_request(): |
|
try: |
|
signature = request.headers.get('X-Signature-Ed25519') |
|
timestamp = request.headers.get('X-Signature-Timestamp') |
|
|
|
if not signature or not timestamp: |
|
return False |
|
|
|
body = request.data.decode('utf-8') |
|
verify_key.verify(f"{timestamp}{body}".encode(), bytes.fromhex(signature)) |
|
return True |
|
except Exception as e: |
|
logger.error(f"Verification error: {str(e)}") |
|
return False |
|
|
|
@app.route("/", methods=["GET"]) |
|
def health_check(): |
|
"""Hugging Face nutzt diesen Endpoint um zu prüfen ob der Space läuft""" |
|
return jsonify({ |
|
"status": "running", |
|
"message": "Discord bot is running!", |
|
"public_key_present": bool(PUBLIC_KEY), |
|
"verify_key_initialized": verify_key is not None |
|
}) |
|
|
|
@app.route("/interactions", methods=["POST"]) |
|
def interactions(): |
|
if not verify_key: |
|
logger.error("verify_key not initialized") |
|
return "Configuration error", 500 |
|
|
|
if not verify_discord_request(): |
|
return "Invalid request signature", 401 |
|
|
|
try: |
|
data = request.json |
|
|
|
|
|
if data.get("type") == 1: |
|
logger.info("Responding to ping verification") |
|
return jsonify({"type": 1}) |
|
|
|
|
|
if data.get("type") == 2: |
|
command = data.get("data", {}).get("name") |
|
logger.info(f"Received command: {command}") |
|
|
|
if command == "settings": |
|
return jsonify({ |
|
"type": 4, |
|
"data": { |
|
"content": "✅ Bot ist aktiv und verifiziert!" |
|
} |
|
}) |
|
|
|
return jsonify({"type": 1}) |
|
|
|
except Exception as e: |
|
logger.error(f"Error processing request: {str(e)}") |
|
return "Internal server error", 500 |
|
|
|
def health_check_worker(): |
|
"""Background worker der regelmäßig den Health-Check Endpoint aufruft""" |
|
while True: |
|
try: |
|
response = requests.get(f"http://localhost:{PORT}/") |
|
logger.info(f"Health check status: {response.status_code}") |
|
except Exception as e: |
|
logger.error(f"Health check failed: {str(e)}") |
|
time.sleep(30) |
|
|
|
def register_commands(): |
|
"""Registriere Befehle bei Discord""" |
|
url = f"https://discord.com/api/v10/applications/{APPLICATION_ID}/commands" |
|
headers = { |
|
"Authorization": f"Bot {BOT_TOKEN}" |
|
} |
|
json = { |
|
"name": "settings", |
|
"description": "Prüft den Status des Bots", |
|
"type": 1 |
|
} |
|
|
|
response = requests.post(url, headers=headers, json=json) |
|
if response.status_code == 200: |
|
logger.info("Command successfully registered") |
|
else: |
|
logger.error(f"Failed to register command: {response.status_code} - {response.json()}") |
|
|
|
def setup_discord_channel_and_role(guild_id): |
|
"""Erstellt einen Log-Channel und eine Rolle in der angegebenen Guild""" |
|
headers = { |
|
"Authorization": f"Bot {BOT_TOKEN}" |
|
} |
|
|
|
|
|
channel_url = f"https://discord.com/api/v10/guilds/{guild_id}/channels" |
|
channel_json = { |
|
"name": "bot-logs", |
|
"type": 0 |
|
} |
|
channel_response = requests.post(channel_url, headers=headers, json=channel_json) |
|
|
|
if channel_response.status_code == 201: |
|
logger.info("Log-Channel successfully created") |
|
else: |
|
logger.error(f"Failed to create log-channel: {channel_response.status_code} - {channel_response.json()}") |
|
|
|
|
|
role_url = f"https://discord.com/api/v10/guilds/{guild_id}/roles" |
|
role_json = { |
|
"name": "Bot Commander", |
|
"permissions": "8" |
|
} |
|
role_response = requests.post(role_url, headers=headers, json=role_json) |
|
|
|
if role_response.status_code == 201: |
|
logger.info("Role successfully created") |
|
else: |
|
logger.error(f"Failed to create role: {role_response.status_code} - {role_response.json()}") |
|
|
|
if __name__ == "__main__": |
|
logger.info(f"Starting Discord bot on port {PORT}...") |
|
|
|
|
|
health_thread = threading.Thread(target=health_check_worker, daemon=True) |
|
health_thread.start() |
|
|
|
|
|
register_commands() |
|
|
|
|
|
serve(app, host="0.0.0.0", port=PORT) |
|
|