""" 1xBet Nigeria scraper. Endpoints discovered via Playwright network interception on 1xbet.ng/en/line/football. League list: GET /service-api/LineFeed/GetSportsShortZip?sports=1&lng=en&country=132&partner=159 Events: GET /service-api/LineFeed/Get1x2_VZip?sports=1&champs=
  • &count=500&lng=en&mode=4&country=132&partner=159&getEmpty=true&virtualSports=true Football: fetches top MAX_LEAGUES leagues sorted by event count (~790 events). Querying all 356 leagues triggers rate limiting (406). 40 leagues at 0.3s delay is reliable. Basketball/Tennis: single call (league listing not available via this endpoint for these sports). Response field mapping: O1 / O2 → home / away team name I → event ID S → start time (Unix seconds, NOT milliseconds) L → league name E[] → main odds array G=1, T=1/2/3 → 1X2: Home / Draw / Away G=17, T=9/10 → Over/Under (P = line value, e.g. 2.5) G=101, T=401/402 → Basketball Winner: Team1 / Team2 """ import time import logging from datetime import datetime from typing import List, Tuple from scrapers.base import BaseScraper from core.models import Event, Outcome logger = logging.getLogger(__name__) # Fetch top N leagues by event count. Querying all 356 leagues triggers 406 rate limiting. # 40 leagues at 0.3s/request gives ~790 events reliably in ~15 seconds. MAX_LEAGUES = 40 BASE_URL = "https://1xbet.ng/service-api/LineFeed" LEAGUES_URL = f"{BASE_URL}/GetSportsShortZip" EVENTS_URL = f"{BASE_URL}/Get1x2_VZip" # 1xBet internal sport IDs (not Sportradar IDs) SPORT_IDS = { 'football': '1', 'basketball': '3', 'tennis': '4', } # Market group G → (market name, outcome map {T: label}) # Only the markets we care about for arb MARKETS = { 1: ('1X2', {1: 'Home', 2: 'Draw', 3: 'Away'}), 17: ('Over/Under 2.5', {9: 'Over', 10: 'Under'}), # P=2.5 filtered below 101: ('Home/Away', {401: 'Home', 402: 'Away'}), # basketball } # For tennis 1X2 market has no draw — only T=1 and T=3 TENNIS_1X2 = ('Home/Away', {1: 'Home', 3: 'Away'}) class OneXBetScraper(BaseScraper): def __init__(self): super().__init__('1xBet') self.session.headers = { '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', 'Accept': 'application/json, text/plain, */*', 'Referer': 'https://1xbet.ng/en/line/football', 'Origin': 'https://1xbet.ng', } # ── Public ──────────────────────────────────────────────────────────────── def get_events(self, sport: str) -> List[Event]: sport_id = SPORT_IDS.get(sport) if not sport_id: return [] if sport == 'football': return self._get_football_events() else: return self._get_events_single_call(sport, sport_id) # ── Football: per-league pagination ─────────────────────────────────────── def _get_football_events(self) -> List[Event]: try: leagues = self._fetch_leagues('1') except Exception as ex: logger.error(f"[1xBet] league list failed: {ex}") return [] events: List[Event] = [] per_league_blocked = False for league_id, league_name in leagues: try: raw_events = self._fetch_events('1', champs=str(league_id)) for raw in raw_events: events.extend(self._parse(raw, 'football')) time.sleep(0.3) # rate limiting — server 406s at higher request rates except Exception as ex: if '406' in str(ex): # Per-league endpoint is blocked — stop immediately, use global fallback logger.warning("[1xBet] per-league requests blocked (406), falling back to global endpoint") per_league_blocked = True break logger.warning(f"[1xBet] league {league_id} ({league_name}) error: {ex}") if per_league_blocked or (not events and leagues): try: raw_events = self._fetch_events('1') for raw in raw_events: events.extend(self._parse(raw, 'football')) except Exception as ex: logger.error(f"[1xBet] fallback also failed: {ex}") return events def _fetch_leagues(self, sport_id: str) -> List[Tuple[int, str]]: r = self.session.get( LEAGUES_URL, params={ 'sports': sport_id, 'lng': 'en', 'country': '132', 'partner': '159', 'virtualSports': 'true', 'gr': '412', }, timeout=15, ) r.raise_for_status() sport_data = r.json().get('Value', []) if not sport_data: return [] leagues = sport_data[0].get('L', []) # Sort by event count descending, take top MAX_LEAGUES to avoid rate limiting leagues_sorted = sorted(leagues, key=lambda x: x.get('GC', 0), reverse=True) return [(lg['LI'], lg.get('L', '')) for lg in leagues_sorted[:MAX_LEAGUES] if 'LI' in lg] # ── Basketball / Tennis: single call ────────────────────────────────────── def _get_events_single_call(self, sport: str, sport_id: str) -> List[Event]: events: List[Event] = [] try: raw_events = self._fetch_events(sport_id) for raw in raw_events: events.extend(self._parse(raw, sport)) except Exception as ex: logger.error(f"[1xBet] {sport} fetch error: {ex}") return events # ── Shared fetch ────────────────────────────────────────────────────────── def _fetch_events(self, sport_id: str, champs: str = None) -> list: params = { 'sports': sport_id, 'count': '500', 'lng': 'en', 'mode': '4', 'country': '132', 'partner': '159', 'getEmpty': 'true', 'virtualSports': 'true', } if champs: params['champs'] = champs r = self.session.get(EVENTS_URL, params=params, timeout=15) r.raise_for_status() return r.json().get('Value', []) # ── Parser ──────────────────────────────────────────────────────────────── def _parse(self, raw: dict, sport: str) -> List[Event]: home = (raw.get('O1') or '').strip() away = (raw.get('O2') or '').strip() if not home or not away: return [] ts = raw.get('S') # Unix seconds (not ms) try: starts_at = datetime.utcfromtimestamp(ts) if ts else None except Exception: starts_at = None event_id_base = str(raw.get('I', '')) league = raw.get('L', '') # Group E[] entries by market G markets_data: dict = {} for entry in raw.get('E', []): g = entry.get('G') if g not in MARKETS and not (sport == 'tennis' and g == 1): continue markets_data.setdefault(g, []).append(entry) result: List[Event] = [] for g, entries in markets_data.items(): if sport == 'tennis' and g == 1: market_name, outcome_map = TENNIS_1X2 else: market_name, outcome_map = MARKETS[g] # For Over/Under, only take the 2.5 line if g == 17: entries = [e for e in entries if e.get('P') == 2.5] outcomes = [] for e in entries: t = e.get('T') label = outcome_map.get(t) if label is None: continue try: odds = float(e.get('C', 0)) except (TypeError, ValueError): continue if odds > 1.0: outcomes.append(Outcome(name=label, odds=odds, bookmaker='1xBet')) # Need exactly the right number of outcomes for the market expected = len(outcome_map) if len(outcomes) == expected: result.append(Event( event_id=f"1x_{event_id_base}_{g}", bookmaker='1xBet', sport=sport, home_team=home, away_team=away, market=market_name, outcomes=outcomes, starts_at=starts_at, league=league, )) return result