#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ C&C Remastered Match API Poller - ECHTE LOBBY API Holt echte Matches von der C&C Lobby API """ import requests import time import json import logging 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 CNCMatchPollerLobbyAPI: def __init__(self, db_config: Dict): self.db_config = db_config self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'NexusControl-CNC-Poller/1.0', 'Accept': 'application/json' }) # C&C Lobby API URLs self.lobby_apis = { 'tiberian-dawn': 'https://lobby.cncnet.org/tiberian-dawn/matches', 'red-alert': 'https://lobby.cncnet.org/red-alert/matches', 'yuris-revenge': 'https://lobby.cncnet.org/yuris-revenge/matches' } 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_lobby_matches(self, game: str) -> List[Dict]: """Holt echte Matches von der C&C Lobby API""" try: if game not in self.lobby_apis: logging.error(f"Unbekanntes Spiel: {game}") return [] url = self.lobby_apis[game] logging.info(f"Hole Matches von {url}") response = self.session.get(url, timeout=30) response.raise_for_status() matches_data = response.json() logging.info(f"{len(matches_data)} Matches von Lobby API für {game} erhalten") # Matches konvertieren converted_matches = [] for i, match in enumerate(matches_data): try: # Lobby API Struktur anpassen converted_match = { 'match_id': f"lobby_{game}_{match.get('id', f'unknown_{i}')}", 'map_name': match.get('map', 'Unknown Map'), 'duration_seconds': match.get('duration', 0), 'started_at': self._parse_timestamp(match.get('started', '')), 'ended_at': self._parse_timestamp(match.get('finished', '')), 'match_type': 'ranked', # Lobby Matches sind meist Ranked 'has_replay': True, # Lobby hat immer Replays 'replay_url': match.get('replay', ''), 'players': self._convert_lobby_players(match.get('players', [])), 'raw_data': json.dumps(match) # Originaldaten speichern } converted_matches.append(converted_match) except Exception as e: logging.warning(f"Fehler bei Match-Konvertierung: {e}") continue return converted_matches except requests.exceptions.RequestException as e: logging.error(f"Lobby API Fehler: {e}") return [] except json.JSONDecodeError as e: logging.error(f"JSON Fehler: {e}") return [] except Exception as e: logging.error(f"Unerwarteter Fehler: {e}") return [] def _parse_timestamp(self, timestamp_str: str) -> datetime: """Parst Timestamp von Lobby API""" try: if not timestamp_str: return datetime.now() # Verschiedene Timestamp-Formate versuchen formats = [ '%Y-%m-%dT%H:%M:%S.%fZ', '%Y-%m-%dT%H:%M:%SZ', '%Y-%m-%d %H:%M:%S', '%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%dT%H:%M:%S' ] for fmt in formats: try: return datetime.strptime(timestamp_str, fmt) except ValueError: continue # Wenn nichts funktioniert, aktuelle Zeit zurückgeben logging.warning(f"Konnte Timestamp nicht parsen: {timestamp_str}") return datetime.now() except Exception as e: logging.warning(f"Timestamp Fehler: {e}") return datetime.now() def _convert_lobby_players(self, lobby_players: List) -> List[Dict]: """Konvertiert Lobby Spieler zu unserem Format""" converted_players = [] for player in lobby_players: try: converted_player = { 'player_id': str(player.get('id', 'unknown')), 'name': player.get('name', 'Unknown Player'), 'faction': player.get('side', 'Unknown'), 'result': 'win' if player.get('won', False) else 'loss', 'final_score': player.get('score', 0), 'resources_collected': player.get('harvested', 0) } converted_players.append(converted_player) except Exception as e: logging.warning(f"Fehler bei Spieler-Konvertierung: {e}") continue return converted_players def get_all_real_players(self) -> List[Dict]: """Holt alle echten NexusControl Spieler für Namensabgleich""" conn = self.connect_db() if not conn: return [] try: cursor = conn.cursor(dictionary=True) cursor.execute(""" SELECT SpielerID as player_id, Name as name FROM Spieler WHERE is_active = 1 AND Name IS NOT NULL AND Name != '' ORDER BY Name """) players = cursor.fetchall() return players except Error as e: logging.error(f"Fehler beim Laden der Spieler: {e}") return [] finally: cursor.close() conn.close() def save_match(self, conn, match: Dict) -> bool: """Speichert echtes Lobby 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, raw_data, created_at ) VALUES (%s, %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'), 'cnc_lobby', match.get('raw_data') )) # 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 # In players Tabelle speichern 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"Lobby-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 Lobby API Polling-Zyklus...") conn = self.connect_db() if not conn: return try: # Alle Spiele von der Lobby API abfragen total_saved = 0 for game, url in self.lobby_apis.items(): matches = self.get_lobby_matches(game) for match in matches: if self.save_match(conn, match): total_saved += 1 # Kurze Pause zwischen API-Aufrufen time.sleep(2) logging.info(f"Lobby API Polling-Zyklus abgeschlossen: {total_saved} neue Matches") except Exception as e: logging.error(f"Fehler im Lobby API Polling-Zyklus: {e}") finally: conn.close() def main(): """Hauptfunktion""" db_config = { 'host': 'localhost', 'database': 'NexusControl', 'user': 'root', 'password': 'HOk~k5$!0q', 'charset': 'utf8mb4' } poller = CNCMatchPollerLobbyAPI(db_config) print("=== C&C Lobby API Test-Lauf ===") poller.run_polling_cycle() if __name__ == "__main__": main()