449 lines
16 KiB
Python
449 lines
16 KiB
Python
#!/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,
|
|
)
|