""" Melbet Nigeria scraper. Platform: Custom (sport.melbet.ng) — protected by Akamai Bot Manager. Akamai blocks any request not made by a real browser (cookie seeding and ctx.request both fail because they use a different TLS/HTTP stack). Strategy: navigate the Playwright browser to each sport/competition page and intercept the API responses that the page's own JavaScript triggers naturally. Those requests carry valid Akamai sensor_data headers and always return 200. Browser worker pattern: - One long-lived daemon thread owns the Playwright browser (avoids asyncio conflicts). - Callers push (nav_url, fragment, Future) tasks onto _task_queue. - Worker navigates to nav_url, captures the first response whose URL contains fragment, XOR-decodes the body, and resolves the Future. Coverage: Football — /en/sport/football page → gettopeventslist (top ~32 events, all markets) + each competition page → GetEventsList (1X2 per competition) Basketball — /en/sport/basketball page → gettopeventslist Tennis — /en/sport/tennis page → gettopeventslist API base: https://sport.melbet.ng/50cf6697-9dfb-4591-bb33-b36fc4245385 StakeType IDs (as requested by the page's own JS): 1 → 1X2 football (SN: W1=Home, X=Draw, W2=Away) 702 → H/A moneyline basketball/tennis -3 → Over/Under main line 26 → BTTS football 10 → Double Chance football (candidate — verify via debug logs) 6 → Draw No Bet football (candidate — verify via debug logs) Event fields: HT/AT=teams D=ISO datetime UTC CN=league Id=event ID """ import re import time import json import logging import queue import threading import concurrent.futures from datetime import datetime from typing import List, Optional from scrapers.base import BaseScraper from core.models import Event, Outcome logger = logging.getLogger(__name__) API_UUID = '50cf6697-9dfb-4591-bb33-b36fc4245385' BASE_URL = f'https://sport.melbet.ng/{API_UUID}' PARTNER = 3000060 LANG_ID = 2 COUNTRY = 'NG' CHUNK_SIZE = 8192 # Browser worker state _task_queue = queue.Queue() _browser_ready = False _browser_lock = threading.Lock() SPORT_IDS = { 'football': 1, 'tennis': 3, 'basketball': 4, } MARKET_NAMES = { 1: '1X2', 702: 'Home/Away', -3: 'Over/Under 2.5', 26: 'BTTS', # DC and DNB IDs are platform-specific; confirmed when they appear in logs. # Common candidates tried: 10=DC, 6=DNB (standard on many platforms). 10: 'Double Chance', 6: 'Draw No Bet', } # Sport pages the browser navigates to; the page's JS triggers the API calls we intercept SPORT_NAV_URLS = { 'football': 'https://sport.melbet.ng/en/sport/football', 'basketball': 'https://sport.melbet.ng/en/sport/basketball', 'tennis': 'https://sport.melbet.ng/en/sport/tennis', } FOOTBALL_COMP_IDS = { 4520: 'Turkey. Super Lig', 4567: 'Spain. Segunda Division', 4486: 'Spain. La Liga', 4536: 'Scotland. Premier League', 5016: 'Russia. Premier League', 4565: 'Portugal. Primeira League', 4535: 'Netherlands. Eredivisie', 4484: 'Italy. Serie A', 4504: 'Italy. Serie B', 4544: 'Germany. Bundesliga 2', 4261: 'Germany. Bundesliga', 4523: 'France. Ligue 2', 4610: 'France. Ligue 1', 4485: 'England. Premier League', 4604: 'England. Championship', 5566: 'Brazil. Serie A', 4550: 'Belgium. Jupiler Pro League', 4564: 'Argentina. Primera Division', } def _decode(raw: bytes): """XOR-decode Melbet's chunked obfuscated response → Python list.""" if not raw or len(raw) < 2: return None key = raw[1] ^ 0x5b parts = [] for offset in range(0, len(raw), CHUNK_SIZE): chunk = raw[offset:offset + CHUNK_SIZE] parts.append(bytes(b ^ key for b in chunk[1:])) decoded = b''.join(parts).decode('utf-8', errors='replace') end = max(decoded.rfind(']'), decoded.rfind('}')) if end < 0: return None try: return json.loads(decoded[:end + 1]) except json.JSONDecodeError: return None def _comp_nav_url(comp_id: int, comp_name: str) -> str: """Build the Melbet competition page URL from ID + name.""" slug = comp_name.lower() slug = re.sub(r'[^a-z0-9]+', '-', slug).strip('-') return f'https://sport.melbet.ng/en/sport/football/{comp_id}-{slug}/' def _navigate_and_capture(page, nav_url: str, fragment: str) -> Optional[bytes]: """Navigate to nav_url and return the raw body of the first response whose URL contains fragment. Navigates via about:blank first to bust SPA caching.""" captured = [] seen_api = [] # diagnostic: API URLs actually seen done = threading.Event() def on_resp(resp): url = resp.url if API_UUID in url: seen_api.append(url) if fragment in url and not done.is_set(): try: captured.append(resp.body()) except Exception: pass done.set() page.on('response', on_resp) try: # Navigate to blank first so the target URL triggers a full reload (not SPA cache) page.goto('about:blank', wait_until='load', timeout=5_000) except Exception: pass try: page.goto(nav_url, wait_until='domcontentloaded', timeout=30_000) done.wait(timeout=20) except Exception: pass finally: try: page.remove_listener('response', on_resp) except Exception: pass if not captured: if seen_api: logger.warning('[Melbet] fragment "%s" not matched. Actual API URLs: %s', fragment, [u.split('?')[0].split('/')[-1] for u in seen_api[:5]]) else: logger.warning('[Melbet] no API calls seen navigating to %s', nav_url) return captured[0] if captured else None def _browser_worker(): """Long-lived daemon thread that owns the Playwright browser. Processes (nav_url, fragment, Future) tasks from _task_queue.""" global _browser_ready try: from playwright.sync_api import sync_playwright pw = sync_playwright().start() browser = pw.chromium.launch( headless=True, args=['--no-sandbox', '--disable-blink-features=AutomationControlled'], ) ctx = browser.new_context( locale='en-US', user_agent=( 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ' '(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' ), viewport={'width': 1920, 'height': 1080}, ) page = ctx.new_page() try: from playwright_stealth import stealth_sync stealth_sync(page) logger.info('[Melbet] stealth mode applied') except ImportError: # Fallback: manually patch the most common Akamai detection signals page.add_init_script( "Object.defineProperty(navigator, 'webdriver', {get: () => undefined});" "window.chrome = {runtime: {}};" ) logger.warning('[Melbet] playwright-stealth not installed, using basic patches') # Navigate to homepage (not football page) so Akamai session is established # but the first football task still gets a fresh uncached load. page.goto('https://sport.melbet.ng/', wait_until='domcontentloaded', timeout=30_000) time.sleep(4) _browser_ready = True logger.info('[Melbet] browser ready') while True: task = _task_queue.get() if task is None: break nav_url, fragment, fut = task try: raw = _navigate_and_capture(page, nav_url, fragment) if raw is None: fut.set_exception(RuntimeError(f'no response captured from {nav_url}')) else: data = _decode(raw) fut.set_result(data if isinstance(data, list) else []) except Exception as ex: try: fut.set_exception(ex) except Exception: pass browser.close() pw.stop() except Exception as ex: logger.error(f'[Melbet] browser worker failed: {ex}') _browser_ready = False class MelbetScraper(BaseScraper): def __init__(self): super().__init__('Melbet') # ── Playwright lifecycle ─────────────────────────────────────────────────── def _ensure_browser(self) -> bool: global _browser_ready if _browser_ready: return True with _browser_lock: if _browser_ready: return True t = threading.Thread(target=_browser_worker, daemon=True, name='melbet-browser') t.start() for _ in range(500): # wait up to 50s if _browser_ready: return True time.sleep(0.1) logger.error('[Melbet] browser did not become ready in time') return False def _close_browser(self): global _browser_ready _browser_ready = False _task_queue.put(None) # ── Public ──────────────────────────────────────────────────────────────── def get_events(self, sport: str) -> List[Event]: if sport not in SPORT_IDS: return [] if not self._ensure_browser(): return [] try: return self._fetch_sport(sport) except Exception as ex: logger.error(f'[Melbet] {sport} error: {ex}') self._close_browser() return [] # ── Fetch ───────────────────────────────────────────────────────────────── def _fetch_sport(self, sport: str) -> List[Event]: events: List[Event] = [] sport_id = SPORT_IDS[sport] # Top events — captured from the sport's main page nav_url = SPORT_NAV_URLS[sport] fragment = 'gettopeventslist' raw_list = self._navigate_fetch(nav_url, fragment, sport) for raw in raw_list: events.extend(self._parse(raw, sport)) # Football only: per-competition events from each competition page if sport == 'football': for comp_id, comp_name in FOOTBALL_COMP_IDS.items(): comp_url = _comp_nav_url(comp_id, comp_name) fragment = f'GetEventsList' raw_list = self._navigate_fetch(comp_url, fragment, sport, warn_label=f'comp {comp_id} ({comp_name})') for raw in raw_list: events.extend(self._parse(raw, sport)) return events def _navigate_fetch(self, nav_url: str, fragment: str, sport: str, warn_label: str = None) -> list: """Push a navigation task to the browser worker and return the decoded list.""" fut = concurrent.futures.Future() _task_queue.put((nav_url, fragment, fut)) try: return fut.result(timeout=45) except Exception as ex: label = warn_label or f'{sport} {fragment}' logger.warning(f'[Melbet] {label}: {ex}') return [] # ── Parser ──────────────────────────────────────────────────────────────── def _parse(self, raw: dict, sport: str) -> List[Event]: home = (raw.get('HT') or '').strip() away = (raw.get('AT') or '').strip() if not home or not away: return [] date_str = raw.get('D', '') try: starts_at = datetime.fromisoformat( date_str.replace('Z', '+00:00') ).replace(tzinfo=None) if date_str else None except Exception: starts_at = None league = (raw.get('CN') or '').strip() event_id = raw.get('Id', '') event_url = f'https://sport.melbet.ng/en/event/{event_id}' if event_id else None result: List[Event] = [] seen_ids: set = set() for st in (raw.get('StakeTypes') or []): st_id = st.get('Id') seen_ids.add(st_id) market_name = MARKET_NAMES.get(st_id) if not market_name: continue if st_id == 1 and sport != 'football': continue if st_id in (6, 10) and sport != 'football': continue outcomes = self._parse_outcomes(st.get('Stakes') or [], st_id, sport, event_url) if not outcomes: continue expected = 3 if (st_id == 1 and sport == 'football') else 2 if len(outcomes) != expected: continue result.append(Event( event_id = f'mb_{event_id}_{st_id}', bookmaker = 'Melbet', sport = sport, home_team = home, away_team = away, market = market_name, outcomes = outcomes, starts_at = starts_at, league = league, )) # Log any StakeType IDs we don't recognise so we can add them later unknown = seen_ids - set(MARKET_NAMES) if unknown and sport == 'football': logger.debug(f'[Melbet] unknown StakeType IDs on {home} vs {away}: {sorted(unknown)}') return result def _parse_outcomes(self, stakes: list, st_id: int, sport: str, event_url: Optional[str]) -> List[Outcome]: outcomes = [] for s in stakes: try: odds = float(s.get('F', 0)) except (TypeError, ValueError): continue if odds <= 1.0: continue sn = (s.get('SN') or '').strip() if st_id == 1: if sn == 'W1': label = 'Home' elif sn == 'X': label = 'Draw' elif sn == 'W2': label = 'Away' else: continue elif st_id == 702: if sn == 'W1': label = 'Home' elif sn == 'W2': label = 'Away' else: continue elif st_id == -3: if sn == 'O': label = 'Over' elif sn == 'U': label = 'Under' else: continue elif st_id == 26: if sn == 'Yes': label = 'Yes' elif sn == 'No': label = 'No' else: continue elif st_id == 10: # Double Chance if sn in ('1X', 'W1X'): label = '1X' elif sn in ('X2', 'XW2'): label = 'X2' elif sn == '12': label = '12' else: continue elif st_id == 6: # Draw No Bet if sn in ('W1', '1'): label = 'Home' elif sn in ('W2', '2'): label = 'Away' else: continue else: continue outcomes.append(Outcome( name=label, odds=odds, bookmaker='Melbet', event_url=event_url )) return outcomes