RLStatsAPIServer.py

#!/usr/bin/env python3
"""
Rocket League PsyNet API Client with HTTP Stats Server
"""

import asyncio
import os
import re
import argparse
import json
import requests
import uuid
import webbrowser
import base64
import tkinter as tk
from tkinter import simpledialog, messagebox
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse
import threading
import time
import ssl
import logging
from typing import Optional, Dict, Any
from pathlib import Path

from rlapi import PsyNet, Platform, PsyNetError
from rlapi.auth import auth_player
from rlapi.client import RocketLeagueClient

# Configure logging
logging.basicConfig(
 level=logging.INFO,
 format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Configuration constants
CONFIG = {
 'envfile': '.epicenv', # Where to store your personal Epic Launcher API creds
 'epic_api_url': 'https://account-public-service-prod.ak.epicgames.com/account/api',
 'epic_dev_api_url': 'https://api.epicgames.dev',
 'client_id': '34a02cf8f4414e29b15921876da36f9a', # Epic Launcher ID
 'client_secret': 'daafbccc737745039dffe53d94fc76cf', # Epic Launcher secret (this is a publicly known secret)
 'dev_client_id': 'xyza7891p5D7s9R6Gm6moTHWGloerp7B', # Epic Dev ID
 'dev_client_secret': 'Knh18du4NVlFs+3uQ+ZPpDCVto0WYf4yXP8+OcwVt1o', # Epic Dev secret (this is a publicly known secret)
 'deployment_id': 'da32ae9c12ae40e8a112c52e1f17f3ba',
 'proxy': {}, # {'http': '127.0.0.1:8888', 'https': '127.0.0.1:8888'}
 'verify_ssl': True,
 'server_port': 9280,
 'player_name': '', # You can leave this blank, it doesn't actually get verified
 'version': '260506.26700.517210' # Placeholder - we'll dynamically figure out the version from RL's log file, or from a GitHub repo
}


class EpicAuthManager:
 """Handles Epic Games authentication flow"""
 
 def __init__(self, envfile: str, proxy: Dict = None, verify: bool = True):
 self.envfile = Path(envfile)
 self.proxy = proxy or {}
 self.verify = verify
 self.headers = {
 'Accept': '*/*',
 'Accept-Encoding': 'deflate, gzip',
 'User-Agent': 'UELauncher/16.12.1-36115220+++Portal+Release-Live Windows/10.0.22631.1.768.64bit',
 }
 
 def _make_basic_auth(self, client_id: str, client_secret: str) -> str:
 """Create Basic Auth header value with base64-encoded credentials"""
 credentials = f"{client_id}:{client_secret}"
 encoded = base64.b64encode(credentials.encode('utf-8')).decode('utf-8')
 return f'basic {encoded}'
 
 def _generate_correlation_id(self) -> str:
 return f'UE4-5dea7166457d308530e85a9b3333ff78-C9AD1AA540580FDB3CCEA5B0A22B3218-{uuid.uuid4().hex.upper()}'
 
 def _api_request(self, method: str, path: str, data: str = '', 
 auth: str = '', base_url: str = None) -> requests.Response:
 """Make authenticated API request to Epic services"""
 url = (base_url or CONFIG['epic_api_url']) + path
 headers = self.headers.copy()
 headers['X-Epic-Correlation-ID'] = self._generate_correlation_id()
 if auth:
 headers['Authorization'] = auth
 if data:
 headers['Content-Type'] = 'application/x-www-form-urlencoded'
 
 kwargs = {'headers': headers, 'proxies': self.proxy, 'verify': self.verify}
 if data:
 kwargs['data'] = data
 
 if method.lower() == 'post':
 return requests.post(url, **kwargs)
 return requests.get(url, **kwargs)
 
 def load_credentials(self) -> Dict[str, str]:
 """Load or initialize credentials from env file"""
 if self.envfile.exists():
 try:
 return json.loads(self.envfile.read_text().strip())
 except json.JSONDecodeError:
 logger.warning(f"Invalid JSON in {self.envfile}, starting fresh")
 return {}
 
 def save_credentials(self, creds: Dict[str, str]):
 """Save credentials to env file"""
 self.envfile.write_text(json.dumps(creds, indent=2))
 
 def get_authorization_code(self) -> str:
 """Open browser for user to login and retrieve auth code"""
 redirect_url = (
 f'https://www.epicgames.com/id/login?'
 f'redirectUrl=https%3A//www.epicgames.com/id/api/redirect'
 f'%3FclientId%3D{CONFIG["client_id"]}'
 f'%26responseType%3Dcode%26prompt%3Dlogin%26'
 )
 webbrowser.open(redirect_url, new=0, autoraise=True)
 
 # Create hidden Tk window for dialog
 root = tk.Tk()
 root.withdraw()
 root.attributes('-topmost', True)
 
 code = simpledialog.askstring(
 "Authorization Code", 
 "Please login in your browser, then paste the authorization code:",
 parent=root
 )
 root.destroy()
 return (code or '').strip()
 
 async def authenticate(self) -> tuple[str, str, str]:
 """
 Complete authentication flow.
 Returns: (access_token, refresh_token, account_id)
 """
 creds = self.load_credentials()
 
 # Get auth code if needed
 if not creds.get('refreshToken') and not creds.get('authCode'):
 logger.warning("No credentials found. Please authorize...")
 messagebox.showwarning(
 'Authorization Required', 
 f'Opening browser to login. Please copy the authorization code.'
 )
 code = self.get_authorization_code()
 if len(code) != 32:
 raise ValueError("Invalid authorization code (must be 32 characters)")
 creds['authCode'] = code
 self.save_credentials(creds)
 
 # Use Epic Games client credentials for the initial requests
 client_auth = self._make_basic_auth(
 CONFIG['client_id'], 
 CONFIG['client_secret']
 )
 
 # Exchange auth code for refresh token if needed
 refresh_token = creds.get('refreshToken')
 if not refresh_token and creds.get('authCode'):
 logger.info("Exchanging auth code for refresh token...")
 resp = self._api_request(
 'post', '/oauth/token',
 auth=client_auth,
 data=f'grant_type=authorization_code&code={creds["authCode"]}'
 )
 resp.raise_for_status()
 refresh_token = resp.json()['refresh_token']
 
 # Get eg1 access token
 logger.info("Obtaining eg1 access token...")
 resp = self._api_request(
 'post', '/oauth/token',
 auth=client_auth,
 data=f'grant_type=refresh_token&refresh_token={refresh_token}&token_type=eg1'
 )
 resp.raise_for_status()
 token_data = resp.json()
 
 if 'errorMessage' in token_data:
 raise RuntimeError(f"{token_data['errorCode']}: {token_data['errorMessage']}")
 
 access_token = token_data['access_token']
 refresh_token = token_data['refresh_token']
 account_id = token_data['account_id']
 
 # Update and save credentials
 creds.update({
 'authCode': '',
 'refreshToken': refresh_token,
 'accountId': account_id
 })
 self.save_credentials(creds)
 
 # Get launcher exchange code
 logger.info("Getting launcher exchange code...")
 resp = self._api_request(
 'get', '/oauth/exchange',
 auth=f'bearer {access_token}'
 )
 resp.raise_for_status()
 exchange_data = resp.json()
 
 if 'errorMessage' in exchange_data:
 raise RuntimeError(f"{exchange_data['errorCode']}: {exchange_data['errorMessage']}")
 
 # Get dev API access token
 logger.info("Obtaining dev API access token...")
 dev_basic_auth = self._make_basic_auth(
 CONFIG['dev_client_id'], 
 CONFIG['dev_client_secret']
 )
 resp = self._api_request(
 'post', '/epic/oauth/v1/token',
 data=(f'grant_type=exchange_code&deployment_id={CONFIG["deployment_id"]}'
 f'&exchange_code={exchange_data["code"]}'),
 auth=dev_basic_auth,
 base_url=CONFIG['epic_dev_api_url']
 )
 resp.raise_for_status()
 dev_tokens = resp.json()
 
 if 'errorMessage' in dev_tokens:
 raise RuntimeError(f"{dev_tokens['errorCode']}: {dev_tokens['errorMessage']}")
 
 # Save dev tokens
 creds.update({
 'accessTokenDev': dev_tokens['access_token'],
 'refreshTokenDev': dev_tokens['refresh_token'],
 'accessTokenDevExpiry': dev_tokens['expires_at']
 })
 self.save_credentials(creds)
 
 return dev_tokens['access_token'], refresh_token, account_id


class RocketLeagueStatsServer:
 """Main application class managing RL client and HTTP server"""
 
 def __init__(self, envfile: str, port: int = 9280):
 self.envfile = envfile
 self.port = port
 self.client: Optional[RocketLeagueClient] = None
 self.rpc = None
 self.loop: Optional[asyncio.AbstractEventLoop] = None
 self._reauth_lock = asyncio.Lock() # Prevent concurrent reauth attempts
 self._last_reauth_attempt = 0 # Timestamp for rate limiting
 self._reauth_cooldown = 30 # Seconds between reauth attempts
 self.auth_manager = EpicAuthManager(
 envfile, 
 proxy=CONFIG['proxy'], 
 verify=CONFIG['verify_ssl']
 )
 
 async def initialize_client(self) -> bool:
 """Initialize the Rocket League websocket client"""
 access_token, _, account_id = await self.auth_manager.authenticate()
 
 # Try to get the latest version from GitHub
 version = self._get_latest_version()
 
 # Get build ID from version string (wide-string -> CRC-32/BZIP2 -> convert to signed int)
 build_id = self._crc32_bzip2_utf16le(version)
 build_id = build_id if build_id < (1 << 31) else build_id - (1 << 32)

 psy_net = PsyNet()
 try:
 self.rpc = await auth_player(
 psy_net,
 access_token,
 account_id,
 CONFIG['player_name'],
 str(build_id),
 CONFIG['version']
 )
 except PsyNetError as e:
 print(f"Exception: {e}")
 return False
 
 logger.info(f"Local Player ID: {self.rpc.local_player_id}")
 
 # Convert to RocketLeagueClient
 self.client = RocketLeagueClient(
 ws_conn=self.rpc.ws_conn,
 local_player_id=self.rpc.local_player_id,
 psy_token=self.rpc.psy_token,
 session_id=self.rpc.session_id,
 request_id=self.rpc.request_id,
 logger=self.rpc.logger,
 )
 # Copy internal state
 for attr in ['_lock', '_pending_reqs', '_pong_event', '_event_queue', 
 '_connected', '_ping_task', '_read_task']:
 setattr(self.client, attr, getattr(self.rpc, attr))
 
 return True
 
 async def _reauthenticate_async(self) -> bool:
 """
 Re-run the full authentication flow and update client references.
 Must be called from within the event loop.
 Returns True on success, False on failure.
 """
 async with self._reauth_lock:
 # Rate limit reauth attempts
 now = time.time()
 if now - self._last_reauth_attempt < self._reauth_cooldown:
 logger.warning(f"Reauth attempt rate-limited (cooldown: {self._reauth_cooldown}s)")
 return False
 self._last_reauth_attempt = now
 
 try:
 logger.info("Starting reauthentication flow...")
 
 # Re-authenticate with Epic and get new dev access token
 access_token, _, account_id = await self.auth_manager.authenticate()

 # Try to get the latest version from GitHub
 version = self._get_latest_version()
 
 # Get build ID from version string (wide-string -> CRC-32/BZIP2 -> convert to signed int)
 build_id = self._crc32_bzip2_utf16le(version)
 build_id = build_id if build_id < (1 << 31) else build_id - (1 << 32)

 psy_net = PsyNet()
 new_rpc = await auth_player(
 psy_net,
 access_token,
 account_id,
 CONFIG['player_name'],
 str(build_id),
 CONFIG['version']
 )
 
 # Create new client instance
 new_client = RocketLeagueClient(
 ws_conn=new_rpc.ws_conn,
 local_player_id=new_rpc.local_player_id,
 psy_token=new_rpc.psy_token,
 session_id=new_rpc.session_id,
 request_id=new_rpc.request_id,
 logger=new_rpc.logger,
 )
 # Copy internal state
 for attr in ['_lock', '_pending_reqs', '_pong_event', '_event_queue', 
 '_connected', '_ping_task', '_read_task']:
 setattr(new_client, attr, getattr(new_rpc, attr))
 
 # Gracefully close old client if it exists
 if self.client:
 try:
 await self.client.close()
 except Exception as e:
 logger.warning(f"Error closing old client: {e}")
 
 # Atomically swap references
 self.rpc = new_rpc
 self.client = new_client
 
 logger.info("Reauthentication successful")
 return True
 
 except Exception as e:
 logger.exception(f"Reauthentication failed: {e}")
 return False
 
 def reauthenticate(self) -> bool:
 """
 Synchronous wrapper to trigger reauthentication from HTTP handler.
 Returns True on success, False on failure.
 """
 if not self.loop or not self.loop.is_running():
 logger.error("Cannot reauthenticate: event loop not running")
 return False
 
 # Run async reauth in the event loop and wait for result
 future = asyncio.run_coroutine_threadsafe(
 self._reauthenticate_async(), 
 self.loop
 )
 return future.result(timeout=60) # 60s timeout for reauth flow
 
 def _crc32_bzip2_utf16le(self, s: str) -> int:
 crc = 0xFFFFFFFF
 table = [(i << 24) & 0xFFFFFFFF for i in range(256)]
 for i in range(256):
 for _ in range(8):
 table[i] = (table[i] << 1) ^ 0x04C11DB7 if table[i] & 0x80000000 else table[i] << 1
 table[i] &= 0xFFFFFFFF
 for byte in s.encode('utf-16-le'):
 crc = ((crc << 8) & 0xFFFFFFFF) ^ table[((crc >> 24) ^ byte) & 0xFF]
 return crc ^ 0xFFFFFFFF
 
 def _get_latest_version(self) -> str:
 """Get the latest game version - first try reading from the RL log file, next try from a GitHub repository, or lastly fallback to the statically defined version"""
 try:
 with open(os.getenv('USERPROFILE') + '\\Documents\\My Games\\Rocket League\\TAGame\\Logs\\Launch.log', 'rb') as f:
 content = f.read().decode('latin-1')
 match = re.findall(r'GPsyonixBuildID [0-9.]+', content)[0]
 version = match.split(' ')[1]
 return version
 except Exception as e:
 print(e)
 pass
 try:
 resp = requests.get('https://raw.githubusercontent.com/smallest-cock/RLSDK/refs/heads/main/RLSDK/GameDefines.cpp')
 version = resp.content.decode().split('Psyonix Build ID: ')[1].split('\n')[0].strip()
 return version
 except:
 pass
 
 return CONFIG['version']

 def _run_async_coroutine(self, coro):
 """Safely run an async coroutine from sync context"""
 if not self.loop or not self.loop.is_running():
 raise RuntimeError("Event loop not running")
 return asyncio.run_coroutine_threadsafe(coro, self.loop).result()
 
 def get_player_stats(self, player_id: str, timeout: float = 10.0) -> Dict[str, Any]:
 """
 Get player stats with optional automatic retry after reauth.
 """
 if not self.client:
 return {'error': 'Client not initialized'}
 try:
 result = self._run_async_coroutine(
 self.client.get_player_skill(player_id, timeout=timeout)
 )
 return result
 except Exception as e:
 logger.error(f"Error fetching stats: {e}")
 return {'error': str(e)}
 
 def start_http_server(self):
 """Start the HTTP server in a separate thread"""
 server = HTTPServer(('', self.port), StatsRequestHandler)
 server.app = self # ← Attach app reference here
 
 logger.info(f"Starting HTTP server on port {self.port}")
 
 def serve():
 try:
 server.serve_forever()
 except KeyboardInterrupt:
 pass
 finally:
 server.server_close()
 
 thread = threading.Thread(target=serve, daemon=True)
 thread.start()
 return server
 
 async def run(self):
 """Main entry point"""
 initialized = await self.initialize_client()
 if initialized is False:
 return
 
 self.loop = asyncio.get_running_loop()
 
 # Start HTTP server
 self.start_http_server()
 
 # Keep alive
 try:
 while True:
 await asyncio.sleep(60)
 # Optional: add health checks or token refresh logic here
 except asyncio.CancelledError:
 logger.info("Shutting down...")
 if self.client:
 await self.client.close()


class StatsRequestHandler(BaseHTTPRequestHandler):
 """HTTP request handler for stats queries"""
 
 def log_message(self, format, *args):
 logger.info(f"{self.address_string()} - {format % args}")
 
 @property
 def app(self) -> RocketLeagueStatsServer:
 """Access the parent app instance via the server reference"""
 return self.server.app
 
 def do_POST(self):
 # Parse path to ignore query strings
 parsed_path = urlparse(self.path).path
 if parsed_path != '/stats':
 self._send_error(404, "Not found. Use POST /stats")
 return
 
 # Read and parse JSON body
 content_length = int(self.headers.get('Content-Length', 0))
 if content_length == 0:
 self._send_error(400, "Missing request body. Send JSON with 'id' field.")
 return
 
 try:
 raw_body = self.rfile.read(content_length)
 data = json.loads(raw_body.decode('utf-8'))
 except json.JSONDecodeError as e:
 self._send_error(400, f"Invalid JSON in request body: {str(e)}")
 return
 except Exception as e:
 self._send_error(400, f"Failed to read request body: {str(e)}")
 return
 
 # Extract player ID
 player_id = data.get('id')
 if not player_id:
 self._send_error(400, "Missing required 'id' field in JSON body")
 return
 
 # Fetch stats
 stats = self.app.get_player_stats(str(player_id))
 
 # Check for errors that warrant reauthentication
 error_msg = stats.get('error', '') if isinstance(stats, dict) else ''
 should_reauth = any(keyword in error_msg.lower() for keyword in [
 'duplicatelogin', 'normal closure', 'perconmaintenance', 'no close frame'
 ])
 
 try:
 if should_reauth:
 logger.warning(f"PsyNet API error detected: '{error_msg}'. Attempting reauthentication...")
 if self.app.reauthenticate():
 logger.info("Reauth succeeded, retrying request...")
 # Retry the request once after successful reauth
 stats = self.app.get_player_stats(str(player_id))
 else:
 logger.error("Reauthentication failed, returning original error")
 # Optionally add a hint to the error response
 if isinstance(stats, dict):
 stats['reauth_hint'] = 'Session may have expired. Try again in a moment.'
 
 response_body = json.dumps(stats, indent=2).encode('utf-8')
 
 self.send_response(200)
 self.send_header('Content-Type', 'application/json')
 self.send_header('Content-Length', len(response_body))
 self.send_header('Connection', 'close')
 self.end_headers()
 self.wfile.write(response_body)
 self.wfile.flush()
 
 except Exception as e:
 logger.exception("Error handling /stats request")
 self._send_error(500, f"Failed to fetch player stats: {str(e)}")
 
 def do_GET(self):
 """Keep a simple health check endpoint for monitoring"""
 parsed_path = urlparse(self.path).path
 if parsed_path == '/health':
 status = {
 'server': 'ok',
 'client_connected': bool(self.app.client and self.app.client._connected),
 'player_id': self.app.rpc.local_player_id if self.app.rpc else None
 }
 response_body = json.dumps(status).encode()
 self.send_response(200)
 self.send_header('Content-Type', 'application/json')
 self.send_header('Content-Length', len(response_body))
 self.send_header('Connection', 'close')
 self.end_headers()
 self.wfile.write(response_body)
 else:
 self._send_error(404, "Not found. Use POST /stats with JSON body containing 'id'.")
 
 def _send_error(self, code: int, message: str):
 """Helper to send consistent JSON error responses"""
 response_body = json.dumps({'error': message}).encode()
 self.send_response(code)
 self.send_header('Content-Type', 'application/json')
 self.send_header('Content-Length', len(response_body))
 self.send_header('Connection', 'close')
 self.end_headers()
 self.wfile.write(response_body)
 self.wfile.flush()


def main():
 parser = argparse.ArgumentParser(description='Rocket League Stats Server')
 parser.add_argument('-f', '--envfile', type=str, default=CONFIG['envfile'],
 help='Credential environment file')
 parser.add_argument('-p', '--port', type=int, default=CONFIG['server_port'],
 help='HTTP server port')
 args = parser.parse_args()
 
 # Suppress tkinter console on Windows
 """
 if os.name == 'nt':
 import ctypes
 ctypes.windll.user32.ShowWindow(ctypes.windll.kernel32.GetConsoleWindow(), 0)
 """
 
 app = RocketLeagueStatsServer(args.envfile, args.port)
 
 try:
 asyncio.run(app.run())
 except KeyboardInterrupt:
 logger.info("Interrupted by user")
 except Exception as e:
 logger.exception(f"Fatal error: {e}")
 if os.name == 'nt':
 messagebox.showerror("Error", str(e))


if __name__ == '__main__':
 main()