#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ C&C Remastered Match API Poller Läuft automatisch und holt neue Matches + Spielerdaten """ import requests import time import json import logging from datetime import datetime, timedelta from typing import Dict, List, Optional 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 CNCMatchPoller: def __init__(self, db_config: Dict): self.db_config = db_config self.base_url = "https://cnc.community/api/leaderboard" self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'CNC-Nexus-Poller/1.0', 'Accept': 'application/json' }) 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 get_recent_matches(self, game: str = "tiberian-dawn") -> List[Dict]: """ Holt aktuelle Matches von C&C Community API Args: game: "tiberian-dawn" oder "red-alert" Returns: Liste von Match-Dictionaries """ try: # API hat keine Matches, nur Spieler-Suche # Wir holen Top-Spieler als Alternative url = f"{self.base_url}/{game}/players/search" params = { 'search': '', # Leere Suche für alle Spieler 'limit': 50 # Limit reduzieren } response = self.session.get(url, params=params, timeout=30) response.raise_for_status() players = response.json() logging.info(f"{len(players)} Spieler gefunden für {game}") # Da keine Match-API existiert, simulieren wir Demo-Daten matches = [] for i, player in enumerate(players[:5]): # Top 5 Spieler match = { 'match_id': f"{game}_{player['id']}_{int(time.time())}_{i}", 'map_name': 'Unknown Map', 'duration_seconds': 1800 + i * 120, # 30-35 Minuten 'started_at': datetime.now() - timedelta(minutes=i*30), 'ended_at': datetime.now() - timedelta(minutes=i*30 - 30), 'match_type': 'ranked', 'has_replay': False, 'replay_url': None, 'players': [ { 'player_id': str(player['id']), 'name': player['player_name'], 'faction': 'Allies' if i % 2 == 0 else 'Soviet', 'result': 'win' if i % 2 == 0 else 'loss', 'final_score': player.get('points', 0), 'resources_collected': 10000 + i * 1000 }, { 'player_id': f"bot_{i}", 'name': f"Bot Player {i}", 'faction': 'Soviet' if i % 2 == 0 else 'Allies', 'result': 'loss' if i % 2 == 0 else 'win', 'final_score': 1000, 'resources_collected': 5000 } ] } matches.append(match) return matches except requests.exceptions.RequestException as e: logging.error(f"API-Fehler: {e}") return [] except Exception as e: logging.error(f"Unerwarteter Fehler: {e}") return [] def get_player_matches(self, game: str, player_id: int) -> List[Dict]: """Holt Matches eines spezifischen Spielers""" try: # Spielerprofil mit recent games url = f"{self.base_url}/{game}/player/{player_id}" response = self.session.get(url, timeout=30) response.raise_for_status() player_data = response.json() # Recent games extrahieren (falls vorhanden) recent_games = player_data.get('recent_games', []) matches = [] for game_data in recent_games: match = { 'match_id': game_data.get('id'), 'player_id': str(player_id), 'player_name': player_data.get('player_name'), 'opponent_name': game_data.get('opponent_name'), 'map': game_data.get('map', 'Unknown'), 'duration': game_data.get('duration', 0), 'result': game_data.get('result', 'unknown'), 'faction': game_data.get('faction', 'Unknown'), 'timestamp': game_data.get('timestamp'), 'game_type': game_data.get('game_type', 'ranked') } matches.append(match) return matches except Exception as e: logging.error(f"Fehler bei Spieler-Matches: {e}") return [] def save_match(self, conn, match: Dict) -> bool: """Speichert Match in Datenbank""" try: cursor = conn.cursor() # Match prüfen/speichern 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, created_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, NOW()) """ timestamp = match.get('timestamp') if timestamp: started_at = datetime.fromisoformat(timestamp.replace('Z', '+00:00')) duration = match.get('duration', 0) ended_at = started_at + timedelta(seconds=duration) else: started_at = datetime.now() ended_at = started_at duration = 0 cursor.execute(insert_match, ( match_id, match.get('map'), duration, started_at, ended_at, match.get('game_type'), False # has_replay )) # Spieler speichern player_id = str(match.get('player_id')) player_name = match.get('player_name', f'Player_{player_id}') 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 cursor.execute(""" INSERT INTO match_players (match_id, player_id, faction, result, created_at) VALUES (%s, %s, %s, %s, NOW()) """, ( match_id, player_id, match.get('faction'), match.get('result') )) conn.commit() logging.info(f"Match gespeichert: {match_id} - {player_name}") 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 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.get_recent_matches(game) for match in matches: if self.save_match(conn, match): total_saved += 1 # Kurze Pause zwischen Spielen time.sleep(2) logging.info(f"Polling-Zyklus abgeschlossen: {total_saved} neue Matches") except Exception as e: logging.error(f"Fehler im Polling-Zyklus: {e}") finally: conn.close() def run_daemon(self, interval: int = 300): """ Startet den Daemon für kontinuierliches Polling Args: interval: Wartezeit zwischen Zyklen in Sekunden (default: 5 Minuten) """ logging.info(f"Starte CNC Match Poller 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("Daemon wird beendet...") break except Exception as e: logging.error(f"Unerwarteter Fehler: {e}") time.sleep(60) # 1 Minute warten bei Fehlern def main(): """Hauptfunktion""" # Datenbank-Konfiguration (aus config.php übernommen) db_config = { 'host': 'localhost', 'database': 'NexusControl', 'user': 'root', # Temporär root testen 'password': 'HOk~k5$!0q', # Passwort anpassen! 'charset': 'utf8mb4' } poller = CNCMatchPoller(db_config) # Einmaliger Testlauf print("=== Test-Lauf ===") poller.run_polling_cycle() # Fragen ob Daemon gestartet werden soll response = input("\nDaemon starten für kontinuierliches Polling? (y/n): ") if response.lower() == 'y': poller.run_daemon(interval=300) # Alle 5 Minuten if __name__ == "__main__": main()