""" 22bet Nigeria scraper. Same LineFeed API as BetWinner — different domain, no partner param required. League list: GET /service-api/LineFeed/GetSportsShortZip?sports=1&lng=en&country=132 Events: GET /service-api/LineFeed/Get1x2_VZip?sports=&champs=
  • &count=500&lng=en&mode=4&country=132&getEmpty=true Football: per-league concurrent requests. If per-league returns 406 (non-NG IP), falls back to global endpoint (capped at 50 events by the platform). Basketball / Tennis: single global call. Response field mapping: O1 / O2 → home / away team name I → event ID S → start time (Unix seconds) L → league name E[] → odds entries G=1, T=1/2/3 → 1X2: Home / Draw / Away G=8, T=4/5/6 → Double Chance: 1X / 12 / X2 (football only) G=2, T=7/8, P=None → Draw No Bet: Home / Away (football only) G=2, T=7/8, P=val → Asian Handicap (football only; home_p + away_p = 0) G=17, T=9/10, P=float → Over/Under: Over / Under G=101, T=401/402 → Basketball H/A: Home / Away """ import time import logging from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from typing import List, Tuple from scrapers.base import BaseScraper from core.models import Event, Outcome import config logger = logging.getLogger(__name__) MAX_LEAGUES = 40 BASE_URL = 'https://22bet.ng/service-api/LineFeed' LEAGUES_URL = f'{BASE_URL}/GetSportsShortZip' EVENTS_URL = f'{BASE_URL}/Get1x2_VZip' SPORT_IDS = { 'football': '1', 'basketball': '3', 'tennis': '4', } # G → (market_name, {T: label}) — general markets MARKETS = { 1: ('1X2', {1: 'Home', 2: 'Draw', 3: 'Away'}), 8: ('Double Chance', {4: '1X', 5: '12', 6: 'X2'}), # football only 17: ('Over/Under', {9: 'Over', 10: 'Under'}), 101: ('Home/Away', {401: 'Home', 402: 'Away'}), } # G=15 and G=62 exist in the main event but represent team-individual goal markets, # NOT HT/2H totals. Removed to prevent false arb signals. _OU_LINES = {0.5, 1.5, 2.5, 3.5, 4.5, 5.5} # G=2: DNB (P=None) or Asian Handicap (P≠None) _G2 = 2 _G2_T_MAP = {7: 'Home', 8: 'Away'} # Correct Score — only in GetGameZip, not Get1x2_VZip. # Same G/T/P encoding as 1xBet/BetWinner (shared Betmaster LineFeed). _CS_G = 136 _CS_SCORE_T = 731 # specific score; P = home_goals + away_goals/1000; None → 0-0 _CS_OTHER_T = 3786 # "Any Other Score" CS_MAX_EVENTS = 150 PARTNER = 151 _COMMON_PARAMS = dict(lng='en_GB', mode='4', country='132', getEmpty='true', partner=PARTNER, tf='3000000', tz='1', gr='243') # Browser cookies required to pass 22bet's anti-bot check. # Refresh from DevTools (Application → Cookies) when 406 errors return. COOKIES = { 'platform_type': 'desktop', 'lng': 'en', 'tzo': '1', 'SESSION': '4c104a341b4cb6cdf5ffa4bbe475318b', 'auid': 'XvGECmnx43RT7qsCAwNDAg==', 'hdt': 'lUs0DvYLNcTyzGdIPqzuwQq1f3ad9vpwjyp7HkPM%2FO61ihJAbzRjF3JALkw13c1TyqLW%2BGvFPUphhDBsUN6nQIf9wD1giy9Vz6zmu1jBkQuVhhrSJAdUHNDuruz94N7%2FPE%2BzMdJG0wWIdN%2BgKNBnmsS046a0dLt9PAnBGl5bCZ5cOX%2Fcdd1cLTRMDKaiF42xcs%2FHV0fbvidpfAio%2FEZnJrNJdVQELbbcl0DsxWlm7W%2FhljpEk4vmsBMtyzILZ5j0xWX388iwBvN1f6PjXl5bipct5ufVMAaK0RtbH4P1IVzSde2GYxDD%2FuQLG8I3wfBUSiOP8s7iQqJoOHE7I6RjXokisdT5DUCFlD2F%2BQ%2FAuunc%2B6OSoLcxLgBv8mX4TUS8BWw49DN%2BKNK%2FJWWdbfODYperr4C9nXed5js5D%2BGRFUiaVqeQPWxb63zmnwnvBtCu4SkW6cK6VhMDe%2BoZ8yKRlulKpeY8%2BYCLlmZTIscpQVcq5895N6h%2BgQg7pCo8tu3OFyMyWf%2BWiGpqPoLQuxBVZ0Wvesm%2FvitdhhINjcpitX8PkDBG3AurwklAlexArhA8WgzuQrFAAXTiTIw%2Bonq91zWYqOOAnToqRIn5iyyN3ClQdJvoDnbHT9aFoimPEEgkMuazSGrt6PPaFhYbFXUj42366zAa7JEf', 'che_g': '8e4716b3-538c-f127-07e0-04780d307b67', 'sh.session.id': 'f20da35d-14eb-44f2-8b44-12b8c89eb4eb', 'v3fr': '1', } class TwentyTwoBetScraper(BaseScraper): def __init__(self): super().__init__('22bet') self.session.headers.update({ 'User-Agent': ( 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ' 'AppleWebKit/537.36 (KHTML, like Gecko) ' 'Chrome/124.0.0.0 Safari/537.36' ), 'Accept': 'application/json, text/plain, */*', 'Accept-Language': 'en-US,en;q=0.9', 'Accept-Encoding': 'gzip, deflate, br', 'Referer': 'https://22bet.ng/en/line/football', 'Origin': 'https://22bet.ng', 'sec-fetch-dest': 'empty', 'sec-fetch-mode': 'cors', 'sec-fetch-site': 'same-origin', }) self.session.cookies.update(COOKIES) # ── 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() return self._fetch_and_parse(sport_id, sport) # ── Football: per-league concurrent ────────────────────────────────────── def _get_football_events(self) -> List[Event]: leagues = self._fetch_leagues('1') if not leagues: return self._fetch_and_parse('1', 'football') results: List[Event] = [] raw_pool: List[dict] = [] blocked = ok = 0 with ThreadPoolExecutor(max_workers=min(len(leagues), 8)) as pool: futs = {pool.submit(self._fetch_league, lid): lid for lid in leagues} for fut in as_completed(futs): events, raws, status = fut.result() if status == 406: blocked += 1 elif status == 200: ok += 1 results.extend(events) raw_pool.extend(raws) total_tried = ok + blocked if total_tried > 0 and ok == 0: logger.warning(f'[22bet] all {blocked} leagues blocked (geo), skipping global fallback') return [] if total_tried > 0 and blocked / total_tried > 0.5: logger.warning(f'[22bet] {blocked}/{total_tried} leagues blocked — falling back to global') return self._fetch_and_parse('1', 'football') results.extend(self._fetch_football_cs(raw_pool)) return results def _fetch_leagues(self, sport_id: str) -> List[int]: try: r = self.session.get(LEAGUES_URL, params={**_COMMON_PARAMS, 'sports': sport_id}, timeout=15) if r.status_code != 200: return [] sport_data = r.json().get('Value', []) if not sport_data: return [] leagues = sport_data[0].get('L', sport_data) if isinstance(sport_data[0], dict) and 'L' in sport_data[0] else sport_data leagues_sorted = sorted(leagues, key=lambda x: x.get('GC', 0), reverse=True) return [lg['I'] for lg in leagues_sorted[:MAX_LEAGUES] if 'I' in lg] except Exception as ex: logger.warning(f'[22bet] league list error: {ex}') return [] def _fetch_league(self, league_id: int) -> Tuple[List[Event], List[dict], int]: try: r = self.session.get( EVENTS_URL, params={**_COMMON_PARAMS, 'sports': '1', 'champs': str(league_id), 'count': '500'}, timeout=20, ) if r.status_code == 406: return [], [], 406 r.raise_for_status() raw_items = r.json().get('Value', []) events = self._parse(raw_items, 'football') time.sleep(0.05) return events, raw_items, 200 except Exception as ex: logger.warning(f'[22bet] league {league_id} error: {ex}') return [], [], 0 # ── Generic fetch + parse ───────────────────────────────────────────────── def _fetch_and_parse(self, sport_id: str, sport: str) -> List[Event]: try: r = self.session.get( EVENTS_URL, params={**_COMMON_PARAMS, 'sports': sport_id, 'count': '500'}, timeout=(8, 20), # (connect, read) ) r.raise_for_status() return self._parse(r.json().get('Value', []), sport) except Exception as ex: logger.error(f'[22bet] {sport} fetch error: {ex}') return [] # ── Correct Score (GetGameZip) ──────────────────────────────────────────── def _fetch_football_cs(self, raw_pool: List[dict]) -> List[Event]: """Fetch Correct Score markets via concurrent GetGameZip calls.""" seen: set = set() unique_raws: List[dict] = [] for raw in raw_pool: eid = str(raw.get('I', '')) if eid and eid not in seen: seen.add(eid) unique_raws.append(raw) if len(unique_raws) >= CS_MAX_EVENTS: break if not unique_raws: return [] cs_events: List[Event] = [] def _fetch_one(raw: dict) -> List[Event]: eid = str(raw.get('I', '')) try: r = self.session.get( f'{BASE_URL}/GetGameZip', params={**_COMMON_PARAMS, 'id': eid, 'isStatic': 'true', 'groupEvents': 'true'}, timeout=15, ) r.raise_for_status() return self._parse_cs(r.json().get('Value', {}), raw) except Exception as ex: logger.debug(f'[22bet] GetGameZip {eid} CS: {ex}') return [] with ThreadPoolExecutor(max_workers=10) as pool: for result in pool.map(_fetch_one, unique_raws): cs_events.extend(result) logger.info(f'[22bet] CS: {len(cs_events)} events from {len(unique_raws)} GetGameZip calls') return cs_events def _parse_cs(self, val: dict, raw: dict) -> List[Event]: home = (val.get('O1') or raw.get('O1') or '').strip() away = (val.get('O2') or raw.get('O2') or '').strip() if not home or not away: return [] eid = str(val.get('I') or raw.get('I') or '') league = (val.get('L') or raw.get('L') or '').strip() ts = val.get('S') or raw.get('S') try: starts_at = datetime.utcfromtimestamp(ts) if ts else None except Exception: starts_at = None event_url = f'https://22bet.ng/en/line/sport/{eid}' if eid else None outcomes: List[Outcome] = [] for e in val.get('E', []): if e.get('G') != _CS_G or e.get('CE') == 1: continue t = e.get('T') p = e.get('P') try: odds = float(e.get('C', 0)) except (TypeError, ValueError): continue if odds <= 1.0: continue if t == _CS_SCORE_T: home_g = int(p) if p is not None else 0 away_g = (round(p * 1000) - home_g * 1000) if p is not None else 0 name = f'{home_g}-{away_g}' elif t == _CS_OTHER_T and p is None: name = 'Any Other Score' else: continue outcomes.append(Outcome(name=name, odds=odds, bookmaker='22bet', event_url=event_url)) if len(outcomes) < 2: return [] return [Event( event_id = f'22b_{eid}_cs', bookmaker = '22bet', sport = 'football', home_team = home, away_team = away, market = 'Correct Score', outcomes = outcomes, starts_at = starts_at, league = league, )] # ── Parser ──────────────────────────────────────────────────────────────── def _parse(self, raw: list, sport: str) -> List[Event]: cutoff = datetime.utcnow().timestamp() + config.HOURS_AHEAD * 3600 events: List[Event] = [] for item in raw: start_ts = item.get('S') if start_ts and start_ts > cutoff: continue starts_at = datetime.utcfromtimestamp(start_ts) if start_ts else None home = item.get('O1', '').strip() away = item.get('O2', '').strip() if not home or not away: continue league = item.get('L', '') event_id = item.get('I', '') event_url = f'https://22bet.ng/en/line/sport/{event_id}' if event_id else None id_base = f'22b_{event_id}' # Bucket odds entries by G by_g: dict = {} for entry in item.get('E', []): g = entry.get('G') if g is not None: by_g.setdefault(g, []).append(entry) # ── G=2: Draw No Bet + Asian Handicap (football only) ───────── if _G2 in by_g and sport == 'football': g2 = by_g.pop(_G2) dnb_entries = [e for e in g2 if e.get('P') is None] ah_entries = [e for e in g2 if e.get('P') is not None] # Draw No Bet dnb_outcomes = [] for e in dnb_entries: label = _G2_T_MAP.get(e.get('T')) if label: try: odds = float(e.get('C', 0)) except (TypeError, ValueError): continue if odds > 1.0: dnb_outcomes.append(Outcome(name=label, odds=odds, bookmaker='22bet', event_url=event_url)) if len(dnb_outcomes) == 2: events.append(self._make_event(id_base + '_dnb', sport, home, away, 'Draw No Bet', dnb_outcomes, starts_at, league)) # Asian Handicap — pair home/away by complementary P values home_odds: dict = {} away_odds: dict = {} for e in ah_entries: label = _G2_T_MAP.get(e.get('T')) p = e.get('P') if label is None or p is None: continue try: p_val = float(p) odds = float(e.get('C', 0)) except (TypeError, ValueError): continue if odds <= 1.0: continue (home_odds if label == 'Home' else away_odds)[p_val] = odds for home_p, home_cf in home_odds.items(): if abs(home_p * 4) % 2 != 0: # skip quarter-ball lines continue away_p = -home_p away_cf = away_odds.get(away_p) or next( (v for k, v in away_odds.items() if abs(k - away_p) < 1e-9), None ) if away_cf is None: continue line = f'+{home_p:g}' if home_p > 0 else f'{home_p:g}' events.append(self._make_event( f'{id_base}_ah{line}', sport, home, away, f'Asian Handicap {line}', [ Outcome(name='Home', odds=home_cf, bookmaker='22bet', event_url=event_url), Outcome(name='Away', odds=away_cf, bookmaker='22bet', event_url=event_url), ], starts_at, league, )) # ── Standard markets ────────────────────────────────────────── for g, entries in by_g.items(): if g not in MARKETS: continue if g == 8 and sport != 'football': continue market_name, t_map = MARKETS[g] if g == 17: for lv in sorted(_OU_LINES): lv_entries = [e for e in entries if e.get('P') == lv] ou_outcomes = self._extract_outcomes(lv_entries, t_map, event_url) if len(ou_outcomes) == 2: events.append(self._make_event( f'{id_base}_ou{lv}', sport, home, away, f'Over/Under {lv:g}', ou_outcomes, starts_at, league, )) else: outcomes = self._extract_outcomes(entries, t_map, event_url) if len(outcomes) >= 2: events.append(self._make_event( f'{id_base}_{g}', sport, home, away, market_name, outcomes, starts_at, league, )) return events # ── Helpers ─────────────────────────────────────────────────────────────── def _extract_outcomes(self, entries: list, t_map: dict, event_url) -> List[Outcome]: outcomes = [] for e in entries: label = t_map.get(e.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='22bet', event_url=event_url)) return outcomes def _make_event(self, event_id, sport, home, away, market, outcomes, starts_at, league) -> Event: return Event( event_id = event_id, bookmaker = '22bet', sport = sport, home_team = home, away_team = away, market = market, outcomes = outcomes, starts_at = starts_at, league = league, )