""" Fetches England Virtual Football League data from Virtustec via WebSocket. Returns upcoming fixtures and historical results for playoff 41104. """ import asyncio import aiohttp import json import uuid import time import logging from datetime import datetime, timezone from typing import Optional logger = logging.getLogger(__name__) ENGLAND_PLAYLIST_ID = 41104 MIN_AHEAD_SECONDS = 180 # skip blocks starting in less than 3 minutes LOOKAHEAD_BLOCKS = 5 # how many upcoming blocks to fetch when searching WS_URL = 'wss://virtual-proxy.virtustec.com/vs' WS_BASE = {'basePath': '/api/client/v0.1', 'host': 'wss://virtual-proxy.virtustec.com'} SPORTYBET_LOGIN_URL = ( 'https://www.sportybet.com/api/ng/virtual/goldenrace-account-v4/login?isV3Imitation=true' ) async def _get_hw_id() -> str: headers = { 'Platform': 'PC', 'DeviceId': str(uuid.uuid4()), 'OperId': '2', 'countryCode': 'ng', 'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36', 'Referer': 'https://www.sportybet.com/ng/virtual/', 'Origin': 'https://www.sportybet.com', } async with aiohttp.ClientSession() as session: async with session.get(SPORTYBET_LOGIN_URL, headers=headers) as r: data = await r.json() return data['data']['anonHash'] def _parse_matches(blocks: list) -> list: """Parse event blocks into match dicts.""" matches = [] for block in blocks: block_id = block.get('eBlockId') status = block.get('serverStatus', 'UNKNOWN') block_data = block.get('data', {}) league_id = block_data.get('champId') match_day = block_data.get('matchDay') for ev in block.get('events', []): data = ev.get('data', ev) parts = data.get('participants', []) if len(parts) < 2: continue home = parts[0] away = parts[1] odds = data.get('oddValues', []) perf = data.get('stats', {}).get('teamToTeam', {}).get('performance', {}) g = perf.get('g', []) # general xG [home, away] gh = perf.get('gh', []) # xG when playing at home [home, away] ga = perf.get('ga', []) # xG when playing away [home, away] h2h = data.get('stats', {}).get('teamToTeam', {}).get('headToHead', []) # venue-adjusted total xG: home team at home + away team away home_xg_venue = float(gh[0]) if gh else (float(g[0]) if g else None) away_xg_venue = float(ga[1]) if len(ga) > 1 else (float(g[1]) if len(g) > 1 else None) total_xg = round(home_xg_venue + away_xg_venue, 3) if (home_xg_venue and away_xg_venue) else None # H2H: last 10 scores → compute O/U stats h2h_over = sum(1 for s in h2h if len(s) == 2 and (int(s[0]) + int(s[1])) > 2) h2h_total = len(h2h) match = { 'block_id': block_id, 'league_id': league_id, 'match_day': match_day, 'status': status, 'home': home.get('name', ''), 'away': away.get('name', ''), 'home_stars': home.get('stars', 0.0), 'away_stars': away.get('stars', 0.0), 'odds_1': float(odds[0]) if len(odds) > 0 else None, # Home win 'odds_2': float(odds[1]) if len(odds) > 1 else None, # Away win 'odds_x': float(odds[2]) if len(odds) > 2 else None, # Draw 'odds_under25': float(odds[56]) if len(odds) > 56 else None, # Under 2.5 'odds_over25': float(odds[57]) if len(odds) > 57 else None, # Over 2.5 'home_xg': float(g[0]) if g else None, 'away_xg': float(g[1]) if len(g) > 1 else None, 'home_xg_h': home_xg_venue, 'away_xg_a': away_xg_venue, 'total_xg': total_xg, 'h2h_over': h2h_over, 'h2h_total': h2h_total, } # If resolved, extract result from wonMarkets result = ev.get('result', {}) if result: won = result.get('wonMarkets', []) half_won = result.get('data', {}).get('halfWonMarkets', []) # Determine match result if 'Home' in won: match['result'] = 'H' elif 'Draw' in won: match['result'] = 'D' elif 'Away' in won: match['result'] = 'A' # Determine score from wonMarkets like _X_Y for m in won: if m.startswith('_') and '_' in m[1:]: parts_s = m[1:].split('_') if len(parts_s) == 2 and parts_s[0].isdigit() and parts_s[1].isdigit(): match['home_score'] = int(parts_s[0]) match['away_score'] = int(parts_s[1]) break # Determine O/U 2.5 if 'over_2_5' in won: match['ou25'] = 'O' elif 'under_2_5' in won: match['ou25'] = 'U' matches.append(match) return matches async def fetch_england_data( history_rounds: int = 5, min_ahead_seconds: int = MIN_AHEAD_SECONDS, ) -> dict: """ Fetches upcoming and historical England virtual league data. Scans ahead up to LOOKAHEAD_BLOCKS rounds to find the first block that starts at least min_ahead_seconds from now, then falls through to later weeks if needed. Returns {'upcoming': [...], 'history': [...], 'seconds_ahead': int} """ hw_id = await _get_hw_id() ws_headers = { 'Origin': 'https://virtual-games.virtustec.com', 'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36', } async with aiohttp.ClientSession() as session: async with session.ws_connect(WS_URL, headers=ws_headers) as ws: xs = 0 async def send_recv(resource: str, query: dict, headers_extra: Optional[dict] = None) -> dict: nonlocal xs h = {'Content-Type': 'application/json'} if headers_extra: h.update(headers_extra) msg = { 'type': 'REQUEST', 'xs': xs, 'ts': int(time.time() * 1000), 'req': {'method': 'GET', 'query': query, 'headers': h, 'resource': resource, **WS_BASE}, } xs += 1 await ws.send_str(json.dumps(msg)) resp = await asyncio.wait_for(ws.receive(), timeout=10.0) return json.loads(resp.data) # Login r = await send_recv( '/session/loginHwId', {'hwId': hw_id, 'profiles': 'WEB', 'tags': 'WEB'}, ) body = r['res']['body'] client_id = body['clientId'] calc_id = body['calculationId'] ch = {'Content-Type': 'application/json', 'clientId': client_id} now_unix = int(time.time()) now_utc = datetime.now(timezone.utc) # Fetch several upcoming blocks and pick the first one with enough lead time r_up = await send_recv( '/eventBlocks/event/data', { 'contentType': 'PLAYLIST', 'contentId': ENGLAND_PLAYLIST_ID, 'offset': None, 'eventTime': now_unix, 'n': LOOKAHEAD_BLOCKS, 'profile': 'WEB', 'calculationId': calc_id, }, ch, ) all_upcoming_blocks = r_up['res'].get('body', []) if not isinstance(all_upcoming_blocks, list): all_upcoming_blocks = [] # Find the first block that starts far enough in the future chosen_block = None seconds_ahead = 0 for block in all_upcoming_blocks: et_str = block.get('eventTime', '') if not et_str: continue et_dt = datetime.fromisoformat(et_str.replace('Z', '+00:00')) secs = (et_dt - now_utc).total_seconds() if secs >= min_ahead_seconds: chosen_block = block seconds_ahead = int(secs) logger.info( f'Using block {block.get("eBlockId")} starting in {secs/60:.1f} min ' f'({et_str})' ) break else: logger.info( f'Skipping block {block.get("eBlockId")} — only {secs/60:.1f} min away' ) upcoming_blocks = [chosen_block] if chosen_block else [] # Historical results r_hist = await send_recv( '/eventBlocks/event/result', { 'contentType': 'PLAYLIST', 'contentId': ENGLAND_PLAYLIST_ID, 'offset': None, 'eventTime': now_unix, 'n': -history_rounds, 'profile': 'WEB', 'calculationId': calc_id, }, ch, ) history_blocks = r_hist['res'].get('body', []) upcoming = _parse_matches(upcoming_blocks) history = _parse_matches(history_blocks if isinstance(history_blocks, list) else []) return {'upcoming': upcoming, 'history': history, 'seconds_ahead': seconds_ahead} def get_england_data(history_rounds: int = 5) -> dict: """Synchronous wrapper for fetch_england_data.""" return asyncio.run(fetch_england_data(history_rounds=history_rounds)) if __name__ == '__main__': import pprint data = get_england_data() print('=== Upcoming ===') pprint.pprint(data['upcoming'][:5]) print('\n=== History ===') pprint.pprint(data['history'][:5])