Simon/hoi4botdc
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-10-26 01:35:24 +02:00
parent 3abe6bd10d
commit 5b91b8ebd3
1 changed files with 221 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

368
app.py
View File

@@ -4,7 +4,7 @@ import os
import asyncio import asyncio
import logging import logging
from dotenv import load_dotenv from dotenv import load_dotenv
import asyncpg import aiomysql
import json import json
from datetime import datetime from datetime import datetime
from typing import Optional, List, Dict from typing import Optional, List, Dict
@@ -25,9 +25,51 @@ DB_PASSWORD = os.getenv('DB_PASSWORD')
# Build DATABASE_URL from individual components if not provided # Build DATABASE_URL from individual components if not provided
if not DATABASE_URL and all([DB_HOST, DB_NAME, DB_USER, DB_PASSWORD]): if not DATABASE_URL and all([DB_HOST, DB_NAME, DB_USER, DB_PASSWORD]):
DATABASE_URL = f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}" DATABASE_URL = f"mysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
print(f"📝 Built DATABASE_URL from individual environment variables") print(f"📝 Built DATABASE_URL from individual environment variables")
# Parse MySQL connection details from DATABASE_URL
def parse_database_url(url):
"""Parse MySQL connection URL into components"""
if not url:
return None
# Remove mysql:// prefix
if url.startswith('mysql://'):
url = url[8:]
# Split user:pass@host:port/db
if '@' in url:
auth, host_db = url.split('@', 1)
if ':' in auth:
user, password = auth.split(':', 1)
else:
user, password = auth, ''
else:
return None
if '/' in host_db:
host_port, database = host_db.split('/', 1)
else:
return None
if ':' in host_port:
host, port = host_port.split(':', 1)
try:
port = int(port)
except ValueError:
port = 3306
else:
host, port = host_port, 3306
return {
'host': host,
'port': port,
'user': user,
'password': password,
'db': database
}
# Global database connection pool # Global database connection pool
db_pool = None db_pool = None
@@ -90,82 +132,106 @@ async def init_database():
"""Initialize database connection and create tables""" """Initialize database connection and create tables"""
global db_pool global db_pool
try: try:
db_pool = await asyncpg.create_pool(DATABASE_URL) # Parse DATABASE_URL for MySQL connection
db_config = parse_database_url(DATABASE_URL)
if not db_config:
raise ValueError("Invalid DATABASE_URL format")
print(f"🔌 Connecting to MySQL: {db_config['host']}:{db_config['port']}/{db_config['db']}")
# Create MySQL connection pool
db_pool = await aiomysql.create_pool(
host=db_config['host'],
port=db_config['port'],
user=db_config['user'],
password=db_config['password'],
db=db_config['db'],
charset='utf8mb4',
autocommit=True,
maxsize=10
)
async with db_pool.acquire() as conn: async with db_pool.acquire() as conn:
# Create players table async with conn.cursor() as cursor:
await conn.execute(''' # Create players table (MySQL syntax)
CREATE TABLE IF NOT EXISTS players ( await cursor.execute('''
id SERIAL PRIMARY KEY, CREATE TABLE IF NOT EXISTS players (
discord_id BIGINT UNIQUE NOT NULL, id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(255) NOT NULL, discord_id BIGINT UNIQUE NOT NULL,
standard_elo INTEGER DEFAULT 800, username VARCHAR(255) NOT NULL,
competitive_elo INTEGER DEFAULT 800, standard_elo INT DEFAULT 800,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, competitive_elo INT DEFAULT 800,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
) updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
''') )
''')
# Create games table
await conn.execute(''' # Create games table (MySQL syntax)
CREATE TABLE IF NOT EXISTS games ( await cursor.execute('''
id SERIAL PRIMARY KEY, CREATE TABLE IF NOT EXISTS games (
game_name VARCHAR(255) NOT NULL, id INT AUTO_INCREMENT PRIMARY KEY,
game_type VARCHAR(50) NOT NULL, game_name VARCHAR(255) NOT NULL,
status VARCHAR(50) DEFAULT 'setup', game_type VARCHAR(50) NOT NULL,
players JSONB NOT NULL DEFAULT '[]', status VARCHAR(50) DEFAULT 'setup',
winner_team VARCHAR(255), players JSON NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, winner_team VARCHAR(255),
finished_at TIMESTAMP created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
) finished_at TIMESTAMP NULL
''') )
''')
# Create game_results table for detailed match history
await conn.execute(''' # Create game_results table (MySQL syntax)
CREATE TABLE IF NOT EXISTS game_results ( await cursor.execute('''
id SERIAL PRIMARY KEY, CREATE TABLE IF NOT EXISTS game_results (
game_id INTEGER REFERENCES games(id), id INT AUTO_INCREMENT PRIMARY KEY,
discord_id BIGINT NOT NULL, game_id INT,
team_name VARCHAR(255) NOT NULL, discord_id BIGINT NOT NULL,
t_level INTEGER NOT NULL, team_name VARCHAR(255) NOT NULL,
old_elo INTEGER NOT NULL, t_level INT NOT NULL,
new_elo INTEGER NOT NULL, old_elo INT NOT NULL,
elo_change INTEGER NOT NULL, new_elo INT NOT NULL,
won BOOLEAN NOT NULL, elo_change INT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP won BOOLEAN NOT NULL,
) created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
''') FOREIGN KEY (game_id) REFERENCES games(id)
)
''')
print("✅ Database initialized successfully") print("✅ Database initialized successfully")
except Exception as e: except Exception as e:
print(f"❌ Database initialization failed: {e}") print(f"❌ Database initialization failed: {e}")
import traceback
traceback.print_exc()
async def get_or_create_player(discord_id: int, username: str) -> Dict: async def get_or_create_player(discord_id: int, username: str) -> Dict:
"""Get or create a player in the database""" """Get or create a player in the database"""
async with db_pool.acquire() as conn: async with db_pool.acquire() as conn:
# Try to get existing player async with conn.cursor(aiomysql.DictCursor) as cursor:
player = await conn.fetchrow( # Try to get existing player
"SELECT * FROM players WHERE discord_id = $1", discord_id await cursor.execute(
) "SELECT * FROM players WHERE discord_id = %s", (discord_id,)
if not player:
# Create new player
await conn.execute(
"INSERT INTO players (discord_id, username) VALUES ($1, $2)",
discord_id, username
)
player = await conn.fetchrow(
"SELECT * FROM players WHERE discord_id = $1", discord_id
)
else:
# Update username if changed
await conn.execute(
"UPDATE players SET username = $1, updated_at = CURRENT_TIMESTAMP WHERE discord_id = $2",
username, discord_id
) )
player = await cursor.fetchone()
return dict(player) if not player:
# Create new player
await cursor.execute(
"INSERT INTO players (discord_id, username) VALUES (%s, %s)",
(discord_id, username)
)
await cursor.execute(
"SELECT * FROM players WHERE discord_id = %s", (discord_id,)
)
player = await cursor.fetchone()
else:
# Update username if changed
await cursor.execute(
"UPDATE players SET username = %s, updated_at = CURRENT_TIMESTAMP WHERE discord_id = %s",
(username, discord_id)
)
return dict(player)
def calculate_elo_change(player_elo: int, opponent_avg_elo: int, won: bool, t_level: int) -> int: def calculate_elo_change(player_elo: int, opponent_avg_elo: int, won: bool, t_level: int) -> int:
"""Calculate ELO change using standard ELO formula with T-level multiplier""" """Calculate ELO change using standard ELO formula with T-level multiplier"""
@@ -240,21 +306,23 @@ async def hoi4create(ctx, game_type: str, game_name: str):
try: try:
async with db_pool.acquire() as conn: async with db_pool.acquire() as conn:
# Check if game name already exists and is active async with conn.cursor(aiomysql.DictCursor) as cursor:
existing_game = await conn.fetchrow( # Check if game name already exists and is active
"SELECT * FROM games WHERE game_name = $1 AND status = 'setup'", await cursor.execute(
game_name "SELECT * FROM games WHERE game_name = %s AND status = 'setup'",
) (game_name,)
)
if existing_game: existing_game = await cursor.fetchone()
await ctx.send(f"❌ A game with name '{game_name}' is already in setup phase!")
return if existing_game:
await ctx.send(f"❌ A game with name '{game_name}' is already in setup phase!")
# Create new game return
await conn.execute(
"INSERT INTO games (game_name, game_type, status) VALUES ($1, $2, 'setup')", # Create new game
game_name, game_type.lower() await cursor.execute(
) "INSERT INTO games (game_name, game_type, status, players) VALUES (%s, %s, 'setup', %s)",
(game_name, game_type.lower(), '[]')
)
embed = discord.Embed( embed = discord.Embed(
title="🎮 Game Created", title="🎮 Game Created",
@@ -280,43 +348,45 @@ async def hoi4setup(ctx, game_name: str, user: discord.Member, team_name: str, t
try: try:
async with db_pool.acquire() as conn: async with db_pool.acquire() as conn:
# Get the game async with conn.cursor(aiomysql.DictCursor) as cursor:
game = await conn.fetchrow( # Get the game
"SELECT * FROM games WHERE game_name = $1 AND status = 'setup'", await cursor.execute(
game_name "SELECT * FROM games WHERE game_name = %s AND status = 'setup'",
) (game_name,)
)
if not game: game = await cursor.fetchone()
await ctx.send(f"❌ No game found with name '{game_name}' in setup phase!")
return if not game:
await ctx.send(f"❌ No game found with name '{game_name}' in setup phase!")
# Get or create player
player = await get_or_create_player(user.id, user.display_name)
# Parse existing players
players = json.loads(game['players']) if game['players'] else []
# Check if player already in game
for p in players:
if p['discord_id'] == user.id:
await ctx.send(f"❌ {user.display_name} is already in this game!")
return return
# Add player to game # Get or create player
player_data = { player = await get_or_create_player(user.id, user.display_name)
'discord_id': user.id,
'username': user.display_name, # Parse existing players
'team_name': team_name, players = json.loads(game['players']) if game['players'] else []
't_level': t_level,
'current_elo': player[f"{game['game_type']}_elo"] # Check if player already in game
} for p in players:
players.append(player_data) if p['discord_id'] == user.id:
await ctx.send(f"❌ {user.display_name} is already in this game!")
# Update game return
await conn.execute(
"UPDATE games SET players = $1 WHERE id = $2", # Add player to game
json.dumps(players), game['id'] player_data = {
) 'discord_id': user.id,
'username': user.display_name,
'team_name': team_name,
't_level': t_level,
'current_elo': player[f"{game['game_type']}_elo"]
}
players.append(player_data)
# Update game
await cursor.execute(
"UPDATE games SET players = %s WHERE id = %s",
(json.dumps(players), game['id'])
)
embed = discord.Embed( embed = discord.Embed(
title="✅ Player Added", title="✅ Player Added",
@@ -339,11 +409,13 @@ async def hoi4end(ctx, game_name: str, winner_team: str):
"""End a game and calculate ELO changes""" """End a game and calculate ELO changes"""
try: try:
async with db_pool.acquire() as conn: async with db_pool.acquire() as conn:
# Get the game async with conn.cursor(aiomysql.DictCursor) as cursor:
game = await conn.fetchrow( # Get the game
"SELECT * FROM games WHERE game_name = $1 AND status = 'setup'", await cursor.execute(
game_name "SELECT * FROM games WHERE game_name = %s AND status = 'setup'",
) (game_name,)
)
game = await cursor.fetchone()
if not game: if not game:
await ctx.send(f"❌ No active game found with name '{game_name}'!") await ctx.send(f"❌ No active game found with name '{game_name}'!")
@@ -412,31 +484,31 @@ async def hoi4end(ctx, game_name: str, winner_team: str):
'won': won 'won': won
}) })
# Update player ELOs and save game results # Update player ELOs and save game results
for change in elo_changes: for change in elo_changes:
# Update player ELO # Update player ELO
elo_field = f"{game['game_type']}_elo" elo_field = f"{game['game_type']}_elo"
await conn.execute( await cursor.execute(
f"UPDATE players SET {elo_field} = $1, updated_at = CURRENT_TIMESTAMP WHERE discord_id = $2", f"UPDATE players SET {elo_field} = %s, updated_at = CURRENT_TIMESTAMP WHERE discord_id = %s",
change['new_elo'], change['discord_id'] (change['new_elo'], change['discord_id'])
) )
# Save game result
await cursor.execute(
"""INSERT INTO game_results
(game_id, discord_id, team_name, t_level, old_elo, new_elo, elo_change, won)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)""",
(game['id'], change['discord_id'], change['team_name'],
change['t_level'], change['old_elo'], change['new_elo'],
change['elo_change'], change['won'])
)
# Save game result # Mark game as finished
await conn.execute( await cursor.execute(
"""INSERT INTO game_results "UPDATE games SET status = 'finished', winner_team = %s, finished_at = CURRENT_TIMESTAMP WHERE id = %s",
(game_id, discord_id, team_name, t_level, old_elo, new_elo, elo_change, won) (winner_team, game['id'])
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)""",
game['id'], change['discord_id'], change['team_name'],
change['t_level'], change['old_elo'], change['new_elo'],
change['elo_change'], change['won']
) )
# Mark game as finished
await conn.execute(
"UPDATE games SET status = 'finished', winner_team = $1, finished_at = CURRENT_TIMESTAMP WHERE id = $2",
winner_team, game['id']
)
# Create result embed # Create result embed
embed = discord.Embed( embed = discord.Embed(
title="🏆 Game Finished!", title="🏆 Game Finished!",
@@ -501,9 +573,11 @@ async def hoi4games(ctx):
"""Show all active games""" """Show all active games"""
try: try:
async with db_pool.acquire() as conn: async with db_pool.acquire() as conn:
games = await conn.fetch( async with conn.cursor(aiomysql.DictCursor) as cursor:
"SELECT * FROM games WHERE status = 'setup' ORDER BY created_at DESC" await cursor.execute(
) "SELECT * FROM games WHERE status = 'setup' ORDER BY created_at DESC"
)
games = await cursor.fetchall()
if not games: if not games:
await ctx.send("📝 No active games found. Use `/hoi4create` to create a new game!") await ctx.send("📝 No active games found. Use `/hoi4create` to create a new game!")