#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ C&C Match Poller - ECHTE FUNKTIONIERENDE TDRA API Verwendet die echte TDRA V2 API: https://cncapi.kanedaspring.de/V2/ """ 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.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('match_poller.log'), logging.StreamHandler() ] ) class TDRAWorkingPoller: def __init__(self, db_config: Dict): self.db_config = db_config self.session = requests.Session() # Echte TDRA V2 API - DIE FUNKTIONIERENDE! self.tdra_api = { 'base': 'https://cncapi.kanedaspring.de/V2', 'lobbychecker': 'https://cncapi.kanedaspring.de/V2/lobbychecker_check', 'matches': 'https://cncapi.kanedaspring.de/V2/matches', 'players': 'https://cncapi.kanedaspring.de/V2/players', 'ladder': 'https://cncapi.kanedaspring.de/V2/ladderboard', 'results': 'https://cncapi.kanedaspring.de/V2/results' } # Wichtig: Origin Header für CORS! self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36', 'Accept': '*/*', 'Accept-Language': 'de-DE,de;q=0.9,en-US;q=0.8,en;q=0.7', 'Accept-Encoding': 'gzip, deflate, br, zstd', 'Connection': 'keep-alive', 'Origin': 'https://cnc.kanedaspring.de', 'Referer': 'https://cnc.kanedaspring.de/', 'Sec-Ch-Ua': '"Google Chrome";v="143", "Chromium";v="143", "Not A(Brand";v="24"', 'Sec-Ch-Ua-Mobile': '?0', 'Sec-Ch-Ua-Platform': '"Windows"', 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-site' }) 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 test_tdra_connection(self) -> bool: """Testet Verbindung zur TDRA API""" try: logging.info("Teste TDRA V2 API Verbindung...") # Zuerst lobbychecker testen (bekannt zu funktionieren) response = self.session.get(self.tdra_api['lobbychecker'], timeout=10) response.raise_for_status() data = response.json() logging.info(f"✅ TDRA V2 API erreichbar: {response.status_code}") logging.info(f"✅ Lobbychecker Response: {data}") return True except requests.exceptions.RequestException as e: logging.error(f"❌ TDRA V2 API nicht erreichbar: {e}") return False def discover_tdra_endpoints(self) -> Dict[str, str]: """Entdeckt alle verfügbaren TDRA Endpunkte""" discovered = {} # Mögliche Endpunkte basierend auf Vue.js Funktionen possible_endpoints = [ '/matches', '/matches/recent', '/results', '/results/recent', '/players', '/players/rank', '/ladderboard', '/ladder', '/stats', '/games', '/rankings' ] for endpoint in possible_endpoints: try: url = f"{self.tdra_api['base']}{endpoint}" logging.info(f"Teste Endpoint: {url}") response = self.session.get(url, timeout=5) if response.status_code == 200: data = response.json() if isinstance(data, list) or (isinstance(data, dict) and data): discovered[endpoint] = url logging.info(f"✅ Endpoint gefunden: {endpoint} ({len(data) if isinstance(data, list) else 1} Einträge)") elif response.status_code == 404: logging.debug(f"⚠️ Endpoint nicht gefunden: {endpoint}") else: logging.warning(f"⚠️ Endpoint {endpoint}: {response.status_code}") except Exception as e: logging.debug(f"Fehler bei {endpoint}: {e}") return discovered def get_tdra_matches(self, discovered_endpoints: Dict[str, str], limit: int = 50) -> List[Dict]: """Holt Matches von allen verfügbaren Endpoints""" all_matches = [] # Versuche verschiedene Endpoints match_endpoints = [ '/matches', '/matches/recent', '/results', '/results/recent' ] for endpoint in match_endpoints: if endpoint in discovered_endpoints: try: url = discovered_endpoints[endpoint] params = {'limit': limit} if '?' not in url else {} logging.info(f"Hole Matches von: {url}") response = self.session.get(url, params=params, timeout=15) response.raise_for_status() data = response.json() if isinstance(data, list): matches = data elif isinstance(data, dict) and 'data' in data: matches = data['data'] elif isinstance(data, dict) and 'matches' in data: matches = data['matches'] else: matches = [data] if data else [] logging.info(f"✅ {len(matches)} Matches von {endpoint}") all_matches.extend(matches) # Wenn wir Matches haben, nicht weiter suchen if matches: break except Exception as e: logging.warning(f"Fehler bei {endpoint}: {e}") return all_matches[:limit] def convert_tdra_match(self, tdra_match: Dict) -> Dict: """Konvertiert TDRA Match zu unserem Format""" try: # TDRA Match-Struktur anpassen match_id = tdra_match.get('matchid', tdra_match.get('id', f"tdra_{int(time.time())}")) # Spieler-Daten players = [] tdra_players = tdra_match.get('players', []) # Verschiedene mögliche Spieler-Strukturen if not tdra_players: if 'player1' in tdra_match and 'player2' in tdra_match: tdra_players = [tdra_match.get('player1', {}), tdra_match.get('player2', {})] elif 'teams' in tdra_match: for team in tdra_match.get('teams', []): for player in team.get('players', []): tdra_players.append(player) for i, player_data in enumerate(tdra_players): player = { 'player_id': str(player_data.get('id', player_data.get('player_id', f'unknown_{i}'))), 'name': player_data.get('name', player_data.get('player_name', f'Player_{i+1}')), 'faction': player_data.get('faction', player_data.get('team', 'Unknown')), 'result': 'win' if player_data.get('won', player_data.get('winner', False)) else 'loss', 'final_score': player_data.get('score', player_data.get('points', 0)), 'tdra_elo': player_data.get('elo', player_data.get('rating', 1000)), 'tdra_skill': player_data.get('skill', 1000) } players.append(player) # Match-Daten match = { 'match_id': f"tdra_{match_id}", 'map_name': tdra_match.get('map', tdra_match.get('mapname', tdra_match.get('map_name', 'Unknown Map'))), 'duration_seconds': tdra_match.get('duration', tdra_match.get('time', tdra_match.get('duration_seconds', 0))), 'started_at': self._parse_tdra_timestamp(tdra_match.get('starttime', tdra_match.get('date', tdra_match.get('started_at')))), 'ended_at': self._parse_tdra_timestamp(tdra_match.get('endtime', tdra_match.get('ended_at'))), 'match_type': tdra_match.get('type', tdra_match.get('gametype', tdra_match.get('match_type', 'ranked'))), 'has_replay': tdra_match.get('replay', tdra_match.get('has_replay', False)), 'replay_url': tdra_match.get('replay_url', ''), 'host': tdra_match.get('host', tdra_match.get('hostname', 'Unknown')), 'players': players, 'raw_tdra_data': json.dumps(tdra_match), 'api_source': 'tdra_v2_working' } return match except Exception as e: logging.error(f"Fehler bei TDRA Match-Konvertierung: {e}") return {} def _parse_tdra_timestamp(self, timestamp) -> datetime: """Parst TDRA Timestamp""" try: if not timestamp: return datetime.now() if isinstance(timestamp, (int, float)): return datetime.fromtimestamp(timestamp) if isinstance(timestamp, str): # Unix Timestamp als String if timestamp.isdigit(): return datetime.fromtimestamp(int(timestamp)) # ISO Format formats = [ '%Y-%m-%dT%H:%M:%S.%fZ', '%Y-%m-%dT%H:%M:%SZ', '%Y-%m-%d %H:%M:%S' ] for fmt in formats: try: return datetime.strptime(timestamp, fmt) except ValueError: continue return datetime.now() except Exception as e: logging.warning(f"TDRA Timestamp Fehler: {e}") return datetime.now() def save_tdra_match(self, conn, match: Dict) -> bool: """Speichert TDRA 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 # Match einfügen cursor.execute(""" 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()) """, ( 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'), match.get('api_source'), match.get('raw_tdra_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"✅ TDRA-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_tdra_working_polling(self): """Führt echten TDRA V2 Polling durch""" logging.info("=== ECHTER TDRA V2 API POLLING ===") # Verbindung testen if not self.test_tdra_connection(): logging.error("TDRA V2 API nicht erreichbar - breche ab") return # Endpunkte entdecken logging.info("Entdecke TDRA Endpunkte...") discovered_endpoints = self.discover_tdra_endpoints() if not discovered_endpoints: logging.warning("Keine TDRA Endpunkte gefunden") return logging.info(f"Gefundene Endpunkte: {list(discovered_endpoints.keys())}") # Matches abrufen matches = self.get_tdra_matches(discovered_endpoints, 50) if not matches: logging.warning("Keine TDRA Matches erhalten") return # Verbindung zur DB conn = self.connect_db() if not conn: return try: total_saved = 0 for tdra_match in matches: # Konvertieren match = self.convert_tdra_match(tdra_match) if not match: continue # Speichern if self.save_tdra_match(conn, match): total_saved += 1 # Kurze Pause time.sleep(0.5) logging.info(f"✅ TDRA V2 Polling abgeschlossen: {total_saved} neue Matches") except Exception as e: logging.error(f"Fehler im TDRA V2 Polling: {e}") finally: conn.close() def main(): """Hauptfunktion""" db_config = { 'host': 'localhost', 'database': 'NexusControl', 'user': 'root', 'password': 'HOk~k5$!0q', 'charset': 'utf8mb4' } poller = TDRAWorkingPoller(db_config) print("=== ECHTER TDRA V2 API POLLER ===") print("✅ API: https://cncapi.kanedaspring.de/V2/") print("� Genau wie auf cnc.kanedaspring.de!") print() poller.run_tdra_working_polling() if __name__ == "__main__": main()