#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ C&C Remastered Match API Poller - DEMO VERSION Generiert Demo-Daten wenn API nicht verfügbar """ import time import json import logging import random from datetime import datetime, timedelta from typing import Dict, List import mysql.connector from mysql.connector import Error # Logging konfigurieren logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('match_poller.log'), logging.StreamHandler() ] ) class CNCMatchPollerDemo: def __init__(self, db_config: Dict): self.db_config = db_config def connect_db(self): """Stellt Verbindung zur Datenbank her""" try: conn = mysql.connector.connect(**self.db_config) return conn except Error as e: logging.error(f"Datenbankverbindung fehlgeschlagen: {e}") return None def generate_demo_players(self) -> List[Dict]: """Generiert Demo-Spieler""" demo_players = [ {'id': 1001, 'name': 'AlphaCommander', 'rank': 1, 'points': 1850.5}, {'id': 1002, 'name': 'SovietGeneral', 'rank': 2, 'points': 1720.3}, {'id': 1003, 'name': 'AlliesHero', 'rank': 3, 'points': 1650.8}, {'id': 1004, 'name': 'RedAlertPro', 'rank': 4, 'points': 1590.2}, {'id': 1005, 'name': 'TiberianMaster', 'rank': 5, 'points': 1480.7}, {'id': 1006, 'name': 'TankCommander', 'rank': 6, 'points': 1420.1}, {'id': 1007, 'name': 'AirForceAce', 'rank': 7, 'points': 1380.9}, {'id': 1008, 'name': 'InfantryLeader', 'rank': 8, 'points': 1320.4}, ] return demo_players def generate_demo_matches(self, game: str) -> List[Dict]: """Generiert Demo-Matches""" players = self.generate_demo_players() matches = [] # Maps für beide Spiele if game == "tiberian-dawn": maps = ["Canyon River", "Green Acres", "Tiberium Garden", "Desert Strike"] else: # red-alert maps = ["Island Warfare", "Soviet Base", "Allies Fortress", "Winter Conflict"] # 3-5 Matches pro Spiel generieren num_matches = random.randint(3, 5) for i in range(num_matches): # Zufällige Spieler auswählen player1 = random.choice(players) player2 = random.choice([p for p in players if p['id'] != player1['id']]) # Zufällige Faktionen wählen player1_faction = random.choice(['Allies', 'Soviet']) player2_faction = 'Soviet' if player1_faction == 'Allies' else 'Allies' # Zufällige Ergebnisse wählen player1_result = random.choice(['win', 'loss']) player2_result = 'loss' if player1_result == 'win' else 'win' match = { 'match_id': f"{game}_demo_{int(time.time())}_{i}", 'map_name': random.choice(maps), 'duration_seconds': random.randint(1200, 2400), # 20-40 Minuten 'started_at': datetime.now() - timedelta(minutes=random.randint(10, 120)), 'ended_at': datetime.now() - timedelta(minutes=random.randint(5, 115)), 'match_type': random.choice(['ranked', 'custom']), 'has_replay': random.choice([True, False]), 'replay_url': f"https://replays.cnc.community/demo/{game}_{i}.rep" if random.choice([True, False]) else None, 'players': [ { 'player_id': str(player1['id']), 'name': player1['name'], 'faction': player1_faction, 'result': player1_result, 'final_score': random.randint(1000, 5000), 'resources_collected': random.randint(8000, 25000) }, { 'player_id': str(player2['id']), 'name': player2['name'], 'faction': player2_faction, 'result': player2_result, 'final_score': random.randint(1000, 5000), 'resources_collected': random.randint(8000, 25000) } ] } matches.append(match) return matches def save_match(self, conn, match: Dict) -> bool: """Speichert Match in Datenbank""" try: cursor = conn.cursor() match_id = match.get('match_id') if not match_id: return False # Prüfen ob Match schon existiert cursor.execute("SELECT match_id FROM matches WHERE match_id = %s", (match_id,)) if cursor.fetchone(): return False # Bereits vorhanden # Match einfügen insert_match = """ INSERT INTO matches ( match_id, map_name, duration_seconds, started_at, ended_at, match_type, has_replay, replay_url, api_source, created_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW()) """ cursor.execute(insert_match, ( match_id, match.get('map_name'), match.get('duration_seconds'), match.get('started_at'), match.get('ended_at'), match.get('match_type'), match.get('has_replay'), match.get('replay_url'), 'demo_mode' # API-Source )) # Alle Spieler speichern for player in match.get('players', []): player_id = player.get('player_id') player_name = player.get('name') if not player_id or not player_name: continue # Überspringen wenn Daten fehlen # In players Tabelle speichern (wegen Foreign Key!) cursor.execute(""" INSERT INTO players (player_id, name, created_at, updated_at) VALUES (%s, %s, NOW(), NOW()) ON DUPLICATE KEY UPDATE name = VALUES(name), updated_at = NOW() """, (player_id, player_name)) # Match-Zuordnung speichern cursor.execute(""" INSERT INTO match_players (match_id, player_id, faction, result, created_at) VALUES (%s, %s, %s, %s, NOW()) """, ( match_id, player_id, player.get('faction', 'Unknown'), player.get('result', 'unknown') )) conn.commit() logging.info(f"Demo-Match gespeichert: {match_id}") return True except Error as e: conn.rollback() logging.error(f"Fehler beim Speichern: {e}") return False finally: cursor.close() def run_polling_cycle(self): """Führt einen kompletten Polling-Zyklus durch""" logging.info("Starte Demo-Polling-Zyklus...") conn = self.connect_db() if not conn: return try: # Beide Spiele abfragen games = ["tiberian-dawn", "red-alert"] total_saved = 0 for game in games: matches = self.generate_demo_matches(game) logging.info(f"{len(matches)} Demo-Matches generiert für {game}") for match in matches: if self.save_match(conn, match): total_saved += 1 # Kurze Pause zwischen Spielen time.sleep(1) logging.info(f"Demo-Polling-Zyklus abgeschlossen: {total_saved} neue Matches") except Exception as e: logging.error(f"Fehler im Demo-Polling-Zyklus: {e}") finally: conn.close() def run_daemon(self, interval: int = 300): """Startet den Daemon für kontinuierliches Demo-Polling""" logging.info(f"Starte CNC Match Poller Demo Daemon (Intervall: {interval}s)") while True: try: self.run_polling_cycle() logging.info(f"Warte {interval} Sekunden bis zum nächsten Zyklus...") time.sleep(interval) except KeyboardInterrupt: logging.info("Demo Daemon wird beendet...") break except Exception as e: logging.error(f"Unerwarteter Fehler: {e}") time.sleep(60) def main(): """Hauptfunktion""" # Datenbank-Konfiguration db_config = { 'host': 'localhost', 'database': 'NexusControl', 'user': 'root', 'password': 'HOk~k5$!0q', 'charset': 'utf8mb4' } poller = CNCMatchPollerDemo(db_config) # Einmaliger Testlauf print("=== Demo Test-Lauf ===") poller.run_polling_cycle() # Fragen ob Daemon gestartet werden soll response = input("\nDemo Daemon starten für kontinuierliches Polling? (y/n): ") if response.lower() == 'y': poller.run_daemon(interval=300) # Alle 5 Minuten if __name__ == "__main__": main()