#!/usr/bin/env python3 from __future__ import annotations import ast import html import json import re import time from decimal import Decimal, InvalidOperation from typing import Any, Dict, Iterable, List, Optional, Tuple from urllib.parse import urljoin import requests DEFAULT_TITAN_CUP_URLS = [ "https://zq.titan007.com/cn/CupMatch/75.html", ] DEFAULT_TITAN_COMPANY_ID = 3 DEFAULT_TITAN_EURO_COMPANY_ID = 545 DEFAULT_TITAN_COMPANY_NAME = "Crown" DEFAULT_USER_AGENT = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/137.0.0.0 Safari/537.36" ) def _limit(value: Any, size: int) -> str: return re.sub(r"\s+", " ", str(value or "")).strip()[:size] def _safe_int(value: Any, default: int = 0) -> int: try: return int(str(value).strip()) except (TypeError, ValueError): return default def _number_text(value: Any, *, absolute: bool = False) -> str: if value is None or value == "": return "" try: number = Decimal(str(value)) if absolute: number = abs(number) text = format(number.normalize(), "f") return "0" if text in {"-0", ""} else text except (InvalidOperation, ValueError): return _limit(value, 32) def _extract_array_expression(script: str, start_at: int) -> Tuple[str, int]: start = script.find("[", start_at) if start < 0: raise ValueError("JavaScript array start not found") depth = 0 quote = "" escaped = False for index in range(start, len(script)): char = script[index] if quote: if escaped: escaped = False elif char == "\\": escaped = True elif char == quote: quote = "" continue if char in {"'", '"'}: quote = char elif char == "[": depth += 1 elif char == "]": depth -= 1 if depth == 0: return script[start : index + 1], index + 1 raise ValueError("JavaScript array end not found") def _fill_sparse_array_holes(expression: str) -> str: output: List[str] = [] quote = "" escaped = False length = len(expression) for index, char in enumerate(expression): output.append(char) if quote: if escaped: escaped = False elif char == "\\": escaped = True elif char == quote: quote = "" continue if char in {"'", '"'}: quote = char continue if char not in {"[", ","}: continue next_index = index + 1 while next_index < length and expression[next_index].isspace(): next_index += 1 if next_index < length and expression[next_index] == ",": output.append("None") return "".join(output) def _parse_js_array(expression: str) -> List[Any]: normalized = _fill_sparse_array_holes(expression) value = ast.literal_eval(normalized) if not isinstance(value, list): raise ValueError("JavaScript assignment is not an array") return value def _parse_named_array(script: str, marker: str) -> List[Any]: marker_index = script.find(marker) if marker_index < 0: raise ValueError(f"JavaScript marker not found: {marker}") expression, _ = _extract_array_expression(script, marker_index + len(marker)) return _parse_js_array(expression) def _is_match_row(value: Any) -> bool: if not isinstance(value, list) or len(value) < 6: return False if _safe_int(value[0]) <= 0: return False return isinstance(value[3], str) and bool(re.match(r"^\d{4}-\d{2}-\d{2}\s", value[3])) def _flatten_match_rows(value: Any) -> Iterable[List[Any]]: if _is_match_row(value): yield value return if isinstance(value, list): for item in value: yield from _flatten_match_rows(item) def parse_titan_schedule_script(script: str) -> Dict[str, Any]: script = (script or "").lstrip("\ufeff") cup = _parse_named_array(script, "var arrCup =") teams = _parse_named_array(script, "var arrTeam =") team_map = { str(row[0]): str(row[1]).strip() for row in teams if isinstance(row, list) and len(row) >= 2 } matches: Dict[str, Dict[str, Any]] = {} assignment_pattern = re.compile(r'jh\["(G[^"]+)"\]\s*=') for assignment in assignment_pattern.finditer(script): expression, _ = _extract_array_expression(script, assignment.end()) event_key = assignment.group(1) for row in _flatten_match_rows(_parse_js_array(expression)): match_id = str(row[0]) home_team_id = str(row[4] or "") away_team_id = str(row[5] or "") matches[match_id] = { "source_match_id": match_id, "source_event_id": event_key, "match_time_text": str(row[3] or "").strip(), "home_team_id": home_team_id, "away_team_id": away_team_id, "home_team": team_map.get(home_team_id, home_team_id), "away_team": team_map.get(away_team_id, away_team_id), "state": row[2] if len(row) > 2 else 0, "schedule": row, } last_update_match = re.search(r"var\s+lastUpdateTime\s*=\s*'([^']*)'", script) return { "source_league_id": str(cup[0] if cup else ""), "league_name": str(cup[1] if len(cup) > 1 else "球探赛事").strip(), "season": str(cup[7] if len(cup) > 7 else "").strip(), "last_update_time": last_update_match.group(1) if last_update_match else "", "matches": matches, } def parse_titan_odds_script(script: str) -> Dict[str, List[List[Any]]]: script = (script or "").lstrip("\ufeff") odds: Dict[str, List[List[Any]]] = {} assignment_pattern = re.compile(r'oddsData\["([OLT]_\d+)"\]\s*=') for assignment in assignment_pattern.finditer(script): expression, _ = _extract_array_expression(script, assignment.end()) value = json.loads(expression) if isinstance(value, list): odds[assignment.group(1)] = value return odds def _find_company_row(rows: List[List[Any]], company_id: int) -> Optional[List[Any]]: for row in rows: if isinstance(row, list) and row and _safe_int(row[0], -1) == company_id: return row return None def _select_company_id(market_rows: Dict[str, List[List[Any]]], preferred_id: int) -> int: coverage: Dict[int, int] = {} for rows in market_rows.values(): seen = set() for row in rows: if not isinstance(row, list) or not row: continue company_id = _safe_int(row[0], -1) if company_id < 0 or company_id in seen: continue seen.add(company_id) coverage[company_id] = coverage.get(company_id, 0) + 1 if preferred_id in coverage: return preferred_id if not coverage: return preferred_id return sorted(coverage, key=lambda company_id: (-coverage[company_id], company_id))[0] def build_titan_odds_items( schedule_data: Dict[str, Any], odds_data: Dict[str, List[List[Any]]], *, source_url: str, company_id: int = DEFAULT_TITAN_COMPANY_ID, euro_company_id: int = DEFAULT_TITAN_EURO_COMPANY_ID, company_name: str = DEFAULT_TITAN_COMPANY_NAME, show_type: str = "early", ) -> List[Dict[str, Any]]: items: List[Dict[str, Any]] = [] matches = schedule_data.get("matches") or {} for match_id, match in matches.items(): market_rows = { "moneyline": odds_data.get(f"O_{match_id}", []), "handicap": odds_data.get(f"L_{match_id}", []), "total": odds_data.get(f"T_{match_id}", []), } selected_company_id = _select_company_id( { "handicap": market_rows["handicap"], "total": market_rows["total"], }, company_id, ) selected_euro_company_id = _select_company_id( {"moneyline": market_rows["moneyline"]}, euro_company_id, ) moneyline = _find_company_row( market_rows["moneyline"], selected_euro_company_id, ) handicap = _find_company_row(market_rows["handicap"], selected_company_id) total = _find_company_row(market_rows["total"], selected_company_id) if not any((moneyline, handicap, total)): continue handicap_value = handicap[2] if handicap and len(handicap) > 2 else "" handicap_number: Optional[Decimal] = None try: handicap_number = Decimal(str(handicap_value)) except (InvalidOperation, ValueError): pass strong_side = "" if handicap_number is not None: if handicap_number > 0: strong_side = "H" elif handicap_number < 0: strong_side = "C" company_ids = { _safe_int(row[0], -1) for rows in market_rows.values() for row in rows if isinstance(row, list) and row } company_ids.discard(-1) markets = { "handicap": { "line": _number_text(handicap_value, absolute=True), "home": _number_text(handicap[1]) if handicap and len(handicap) > 1 else "", "away": _number_text(handicap[3]) if handicap and len(handicap) > 3 else "", "strong": strong_side, }, "total": { "over_line": _number_text(total[2]) if total and len(total) > 2 else "", "under_line": _number_text(total[2]) if total and len(total) > 2 else "", "over": _number_text(total[1]) if total and len(total) > 1 else "", "under": _number_text(total[3]) if total and len(total) > 3 else "", }, "moneyline": { "home": _number_text(moneyline[1]) if moneyline and len(moneyline) > 1 else "", "draw": _number_text(moneyline[2]) if moneyline and len(moneyline) > 2 else "", "away": _number_text(moneyline[3]) if moneyline and len(moneyline) > 3 else "", }, "first_half": { "handicap_line": "", "handicap_home": "", "handicap_away": "", "total_over_line": "", "total_under_line": "", "total_over": "", "total_under": "", "home_win": "", "away_win": "", "draw": "", }, "both_teams_to_score": {"yes": "", "no": ""}, "kickoff": {"home": "", "away": ""}, } selected_company_name = company_name if selected_company_id == company_id else str(selected_company_id) raw_data = { "schedule": match.get("schedule") or [], "odds": { "moneyline": market_rows["moneyline"], "handicap": market_rows["handicap"], "total": market_rows["total"], }, "selected_company_id": selected_company_id, "selected_euro_company_id": selected_euro_company_id, "selected_company_name": selected_company_name, "last_update_time": schedule_data.get("last_update_time") or "", } items.append( { "source_site": "titan007", "source_url": _limit(source_url, 255), "source_match_id": _limit(match_id, 64), "source_parent_id": "", "source_league_id": _limit(schedule_data.get("source_league_id"), 64), "source_event_id": _limit(match.get("source_event_id"), 64), "sport_type": "FT", "show_type": _limit(show_type, 20), "odds_type": "H", "league_name": _limit(schedule_data.get("league_name"), 150), "match_time_text": _limit(match.get("match_time_text"), 50), "home_team_id": _limit(match.get("home_team_id"), 64), "away_team_id": _limit(match.get("away_team_id"), 64), "home_team": _limit(match.get("home_team"), 150), "away_team": _limit(match.get("away_team"), 150), "strong_side": strong_side, "more_count": len(company_ids), "is_running": 0, "handicap_line": _limit(markets["handicap"]["line"], 32), "handicap_home_odds": _limit(markets["handicap"]["home"], 32), "handicap_away_odds": _limit(markets["handicap"]["away"], 32), "total_line": _limit(markets["total"]["over_line"], 32), "total_over_odds": _limit(markets["total"]["over"], 32), "total_under_odds": _limit(markets["total"]["under"], 32), "home_win_odds": _limit(markets["moneyline"]["home"], 32), "draw_odds": _limit(markets["moneyline"]["draw"], 32), "away_win_odds": _limit(markets["moneyline"]["away"], 32), "half_handicap_line": "", "half_home_win_odds": "", "half_draw_odds": "", "half_away_win_odds": "", "raw_data": raw_data, "markets": markets, } ) return items class TitanOddsClient: def __init__( self, *, user_agent: str = DEFAULT_USER_AGENT, timeout: int = 20, verify_tls: bool = True, company_id: int = DEFAULT_TITAN_COMPANY_ID, euro_company_id: int = DEFAULT_TITAN_EURO_COMPANY_ID, company_name: str = DEFAULT_TITAN_COMPANY_NAME, show_type: str = "early", request_delay: float = 0.2, ) -> None: self.timeout = int(timeout) self.verify_tls = bool(verify_tls) self.company_id = int(company_id) self.euro_company_id = int(euro_company_id) self.company_name = company_name self.show_type = show_type self.request_delay = float(request_delay) self.session = requests.Session() self.session.verify = self.verify_tls self.session.headers.update( { "User-Agent": user_agent, "Accept": "text/html,application/xhtml+xml,application/javascript,*/*;q=0.8", } ) def fetch_cup(self, cup_url: str) -> List[Dict[str, Any]]: page_response = self.session.get(cup_url, timeout=self.timeout) page_response.raise_for_status() page_html = page_response.text or "" season_match = re.search(r"selectSeason\s*=\s*'([^']+)'", page_html) league_match = re.search(r"SclassID\s*=\s*(\d+)", page_html) script_match = re.search( r"""src=["']([^"']*/jsData/matchResult/[^"']+\.js(?:\?[^"']*)?)["']""", page_html, re.IGNORECASE, ) if not season_match or not league_match or not script_match: raise ValueError("球探杯赛页面缺少赛季、联赛或赛程脚本") season = season_match.group(1) league_id = league_match.group(1) schedule_url = urljoin(page_response.url, html.unescape(script_match.group(1))) schedule_response = self.session.get( schedule_url, headers={"Referer": page_response.url}, timeout=self.timeout, ) schedule_response.raise_for_status() schedule_data = parse_titan_schedule_script(schedule_response.text) if self.request_delay > 0: time.sleep(self.request_delay) odds_url = urljoin(page_response.url, "/League/LeagueOddsAjax") odds_response = self.session.get( odds_url, params={ "sclassId": league_id, "subSclassId": "", "matchSeason": season, "round": 1, }, headers={ "Referer": page_response.url, "X-Requested-With": "XMLHttpRequest", }, timeout=self.timeout, ) odds_response.raise_for_status() odds_data = parse_titan_odds_script(odds_response.text) return build_titan_odds_items( schedule_data, odds_data, source_url=page_response.url, company_id=self.company_id, euro_company_id=self.euro_company_id, company_name=self.company_name, show_type=self.show_type, )