添加赔率
This commit is contained in:
hajimi
@@ -0,0 +1,448 @@
|
||||
#!/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,
|
||||
)
|
||||
Reference in New Issue
Block a user