#!/usr/bin/env python3 from __future__ import annotations import base64 import json import os import re import time from dataclasses import dataclass from typing import Any, Dict, Iterable, List, Optional from urllib.parse import urljoin from xml.etree import ElementTree as ET import pymysql import requests import urllib3 DEFAULT_BASE_URLS = [ "https://205.201.0.120/", "https://mos022.com/", "https://hga039.com/", "https://mos100.com/", ] DEFAULT_SPORT_TYPES = ["FT", "BK", "ES", "TN", "VB", "BM", "TT", "BS", "SK", "OP"] DEFAULT_SHOW_TYPES = ["live", "today", "early"] SHOW_TYPE_RTYPE = { "live": "rb", "today": "r", "early": "r", } 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" ) REQUEST_DELAY_SECONDS = 0.4 class OddsGameListUnavailable(Exception): """Raised when the account/menu combination has no accessible game list.""" def _split_env_list(value: str, default: Iterable[str]) -> List[str]: if not str(value or "").strip(): return list(default) parts = re.split(r"[\s,]+", str(value).strip()) return [item.strip() for item in parts if item.strip()] def _normalize_base_url(url: str) -> str: normalized = str(url or "").strip() if not normalized: raise ValueError("empty odds base url") if not normalized.startswith(("http://", "https://")): normalized = "https://" + normalized return normalized.rstrip("/") + "/" def _bool_env(value: Any, default: bool = False) -> bool: if value is None or value == "": return default return str(value).strip().lower() in {"1", "true", "yes", "on", "y"} def _safe_int(value: Any, default: int = 0) -> int: try: return int(str(value).strip()) except (TypeError, ValueError): return default def _text(raw: Dict[str, str], key: str) -> str: return str(raw.get(key) or "").strip() def _limit(value: Any, size: int) -> str: return re.sub(r"\s+", " ", str(value or "")).strip()[:size] def _dump_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, separators=(",", ":"), default=str) def _extract_version(html: str) -> str: match = re.search(r"top\.ver\s*=\s*'([^']+)'", html or "") if not match: raise ValueError("odds site version not found") return match.group(1) def _parse_login_xml(body: str) -> Dict[str, str]: root = ET.fromstring((body or "").strip()) status = (root.findtext("status") or "").strip() msg = (root.findtext("msg") or "").strip() uid = (root.findtext("uid") or "").strip() if status != "200" or not uid: code_message = (root.findtext("code_message") or root.findtext("msg") or "login failed").strip() raise ValueError(f"odds login failed: {code_message or msg or status}") return {child.tag: (child.text or "").strip() for child in list(root)} def _skippable_game_list_reason(body: str) -> str: prefix = re.sub(r"\s+", " ", str(body or "").strip())[:120] if prefix.startswith("CheckEMNU"): return "菜单不可用" if prefix.startswith("coupon_date error"): return "早盘日期不可用" return "" def _build_markets(raw: Dict[str, str]) -> Dict[str, Any]: return { "handicap": { "line": _text(raw, "RATIO_R"), "home": _text(raw, "IOR_RH"), "away": _text(raw, "IOR_RC"), "strong": _text(raw, "STRONG"), }, "total": { "over_line": _text(raw, "RATIO_OUO"), "under_line": _text(raw, "RATIO_OUU"), "over": _text(raw, "IOR_OUC"), "under": _text(raw, "IOR_OUH"), }, "moneyline": { "home": _text(raw, "IOR_MH"), "away": _text(raw, "IOR_MC"), "draw": _text(raw, "IOR_MN"), }, "first_half": { "handicap_line": _text(raw, "RATIO_HR"), "handicap_home": _text(raw, "IOR_HRH"), "handicap_away": _text(raw, "IOR_HRC"), "total_over_line": _text(raw, "RATIO_HOUO"), "total_under_line": _text(raw, "RATIO_HOUU"), "total_over": _text(raw, "IOR_HOUC"), "total_under": _text(raw, "IOR_HOUH"), "home_win": _text(raw, "IOR_HMH"), "away_win": _text(raw, "IOR_HMC"), "draw": _text(raw, "IOR_HMN"), }, "both_teams_to_score": { "yes": _text(raw, "IOR_TSY"), "no": _text(raw, "IOR_TSN"), }, "kickoff": { "home": _text(raw, "IOR_TKH"), "away": _text(raw, "IOR_TKC"), }, } def parse_game_list_xml( body: str, *, source_site: str, source_url: str, sport_type: str, show_type: str, odds_type: str, ) -> List[Dict[str, Any]]: root = ET.fromstring((body or "").strip()) items: List[Dict[str, Any]] = [] ec_nodes = root.findall(".//ec") if not ec_nodes: ec_nodes = [root] for ec_node in ec_nodes: ec_id = str(ec_node.attrib.get("id") or "").replace("ec", "", 1) for game in ec_node.findall(".//game"): raw = {child.tag: (child.text or "").strip() for child in list(game)} source_match_id = _text(raw, "GID") or str(game.attrib.get("id") or "").strip() if not source_match_id: continue if not _text(raw, "ECID") and ec_id: raw["ECID"] = ec_id markets = _build_markets(raw) item = { "source_site": _limit(source_site, 50), "source_url": _limit(source_url, 255), "source_match_id": _limit(source_match_id, 64), "source_parent_id": _limit(_text(raw, "GIDM"), 64), "source_league_id": _limit(_text(raw, "LID"), 64), "source_event_id": _limit(_text(raw, "ECID") or ec_id, 64), "sport_type": _limit(str(sport_type).upper(), 20), "show_type": _limit(str(show_type).lower(), 20), "odds_type": _limit(str(odds_type).upper(), 20), "league_name": _limit(_text(raw, "LEAGUE"), 150), "match_time_text": _limit(_text(raw, "DATETIME"), 50), "home_team_id": _limit(_text(raw, "TEAM_H_ID"), 64), "away_team_id": _limit(_text(raw, "TEAM_C_ID"), 64), "home_team": _limit(_text(raw, "TEAM_H"), 150), "away_team": _limit(_text(raw, "TEAM_C"), 150), "strong_side": _limit(_text(raw, "STRONG"), 10), "more_count": _safe_int(_text(raw, "MORE")), "is_running": 1 if _text(raw, "IS_RB").upper() == "Y" or str(show_type).lower() == "live" else 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": _limit(markets["first_half"]["handicap_line"], 32), "half_home_win_odds": _limit(markets["first_half"]["home_win"], 32), "half_draw_odds": _limit(markets["first_half"]["draw"], 32), "half_away_win_odds": _limit(markets["first_half"]["away_win"], 32), "raw_data": raw, "markets": markets, } items.append(item) return items @dataclass class OddsLoginSession: base_url: str transform_url: str version: str uid: str ltype: str odds_type: str class HgOddsClient: def __init__( self, *, base_url: str, username: str, password: str, user_agent: str = DEFAULT_USER_AGENT, timeout: int = 20, verify_tls: bool = False, langx: str = "zh-cn", odds_type: str = "H", ) -> None: self.base_url = _normalize_base_url(base_url) self.username = username self.password = password self.user_agent = user_agent self.timeout = int(timeout) self.verify_tls = bool(verify_tls) self.langx = langx self.odds_type = odds_type self.session = requests.Session() self.session.verify = self.verify_tls self.session.headers.update( { "User-Agent": self.user_agent, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Referer": self.base_url, } ) if not self.verify_tls: urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) self.login_session: Optional[OddsLoginSession] = None def login(self) -> OddsLoginSession: bootstrap = self.session.post( self.base_url, data={"detection": "Y", "sub_doubleLogin": "", "isapp": "", "q": "", "appversion": ""}, timeout=self.timeout, ) bootstrap.raise_for_status() version = _extract_version(bootstrap.text) transform_url = urljoin(self.base_url, "transform_nl.php") + f"?ver={version}" user_agent_b64 = base64.b64encode(self.user_agent.encode("utf-8")).decode("ascii") login_response = self.session.post( transform_url, data={ "p": "chk_login", "langx": self.langx, "ver": version, "username": self.username, "password": self.password, "app": "N", "auto": "EDACFD", "userAgent": user_agent_b64, }, timeout=self.timeout, ) login_response.raise_for_status() parsed = _parse_login_xml(login_response.text) self.login_session = OddsLoginSession( base_url=self.base_url, transform_url=transform_url, version=version, uid=parsed["uid"], ltype=parsed.get("ltype") or "3", odds_type=parsed.get("odd_f_type") or self.odds_type, ) return self.login_session def fetch_game_list(self, sport_type: str, show_type: str) -> List[Dict[str, Any]]: login_session = self.login_session or self.login() normalized_show_type = str(show_type).lower() normalized_sport_type = str(sport_type).upper() now_ms = int(time.time() * 1000) response = self.session.post( login_session.transform_url, data={ "uid": login_session.uid, "ver": login_session.version, "langx": self.langx, "p": "get_game_list", "p3type": "", "date": "", "gtype": normalized_sport_type.lower(), "showtype": normalized_show_type, "rtype": SHOW_TYPE_RTYPE.get(normalized_show_type, "r"), "ltype": login_session.ltype, "filter": normalized_sport_type, "cupFantasy": "N", "sorttype": "L", "specialClick": "", "isFantasy": "N", "ts": str(now_ms), "chgSortTS": str(now_ms), }, timeout=self.timeout, ) response.raise_for_status() body = response.text or "" skip_reason = _skippable_game_list_reason(body) if skip_reason: raise OddsGameListUnavailable(skip_reason) if not body.lstrip().startswith("<"): prefix = re.sub(r"\s+", " ", body.strip())[:120] raise ValueError(f"unexpected odds response: {prefix}") return parse_game_list_xml( body, source_site="hg", source_url=login_session.base_url, sport_type=normalized_sport_type, show_type=normalized_show_type, odds_type=login_session.odds_type, ) @dataclass class MatchOddsRepository: db_config: Dict[str, Any] def __post_init__(self) -> None: self.prefix = str(self.db_config.get("prefix") or "la_") self.conn = pymysql.connect( host=self.db_config["host"], port=int(self.db_config["port"]), user=self.db_config["user"], password=self.db_config["password"], database=self.db_config["database"], charset=self.db_config.get("charset") or "utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=False, ) def close(self) -> None: self.conn.close() def ensure_table(self) -> None: sql = f""" CREATE TABLE IF NOT EXISTS `{self.prefix}match_odds` ( `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, `source_site` VARCHAR(50) NOT NULL DEFAULT '' COMMENT '赔率来源', `source_url` VARCHAR(255) NOT NULL DEFAULT '' COMMENT '来源线路', `source_match_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '来源比赛ID', `source_parent_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '来源主比赛ID', `source_league_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '来源联赛ID', `source_event_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '来源事件ID', `sport_type` VARCHAR(20) NOT NULL DEFAULT '' COMMENT '运动类型', `show_type` VARCHAR(20) NOT NULL DEFAULT '' COMMENT '盘口类型 live/today/early', `odds_type` VARCHAR(20) NOT NULL DEFAULT '' COMMENT '赔率格式', `league_name` VARCHAR(150) NOT NULL DEFAULT '' COMMENT '联赛名称', `match_time_text` VARCHAR(50) NOT NULL DEFAULT '' COMMENT '来源比赛时间文本', `home_team_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '主队来源ID', `away_team_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '客队来源ID', `home_team` VARCHAR(150) NOT NULL DEFAULT '' COMMENT '主队名称', `away_team` VARCHAR(150) NOT NULL DEFAULT '' COMMENT '客队名称', `strong_side` VARCHAR(10) NOT NULL DEFAULT '' COMMENT '让球强弱方', `more_count` INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '更多玩法数量', `is_running` TINYINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '是否滚球', `handicap_line` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '让球盘口', `handicap_home_odds` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '主队让球赔率', `handicap_away_odds` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '客队让球赔率', `total_line` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '大小球盘口', `total_over_odds` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '大球赔率', `total_under_odds` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '小球赔率', `home_win_odds` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '主胜赔率', `draw_odds` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '平局赔率', `away_win_odds` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '客胜赔率', `half_handicap_line` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '上半场让球盘口', `half_home_win_odds` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '上半场主胜赔率', `half_draw_odds` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '上半场平局赔率', `half_away_win_odds` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '上半场客胜赔率', `create_time` INT UNSIGNED NOT NULL DEFAULT 0, `raw_data` JSON NULL COMMENT '来源原始game节点', `markets_json` JSON NULL COMMENT '规范化盘口JSON', `update_time` INT UNSIGNED NOT NULL DEFAULT 0, PRIMARY KEY (`id`), UNIQUE KEY `uk_source_match` (`source_site`, `source_match_id`), KEY `idx_sport_show` (`sport_type`, `show_type`), KEY `idx_league` (`source_league_id`), KEY `idx_update_time` (`update_time`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='第三方赛事赔率表' """ with self.conn.cursor() as cur: cur.execute(sql) self.conn.commit() def upsert_latest(self, items: List[Dict[str, Any]], now_ts: Optional[int] = None) -> int: if not items: return 0 now_ts = int(now_ts or time.time()) sql = ( f"INSERT INTO `{self.prefix}match_odds` (" "`source_site`, `source_url`, `source_match_id`, `source_parent_id`, `source_league_id`, `source_event_id`, " "`sport_type`, `show_type`, `odds_type`, `league_name`, `match_time_text`, " "`home_team_id`, `away_team_id`, `home_team`, `away_team`, `strong_side`, `more_count`, `is_running`, " "`handicap_line`, `handicap_home_odds`, `handicap_away_odds`, " "`total_line`, `total_over_odds`, `total_under_odds`, " "`home_win_odds`, `draw_odds`, `away_win_odds`, " "`half_handicap_line`, `half_home_win_odds`, `half_draw_odds`, `half_away_win_odds`, " "`create_time`, `raw_data`, `markets_json`, `update_time`" ") VALUES (" + ", ".join(["%s"] * 35) + ") ON DUPLICATE KEY UPDATE " "`source_url`=VALUES(`source_url`), " "`source_parent_id`=VALUES(`source_parent_id`), " "`source_league_id`=VALUES(`source_league_id`), " "`source_event_id`=VALUES(`source_event_id`), " "`sport_type`=VALUES(`sport_type`), " "`show_type`=VALUES(`show_type`), " "`odds_type`=VALUES(`odds_type`), " "`league_name`=VALUES(`league_name`), " "`match_time_text`=VALUES(`match_time_text`), " "`home_team_id`=VALUES(`home_team_id`), " "`away_team_id`=VALUES(`away_team_id`), " "`home_team`=VALUES(`home_team`), " "`away_team`=VALUES(`away_team`), " "`strong_side`=VALUES(`strong_side`), " "`more_count`=VALUES(`more_count`), " "`is_running`=VALUES(`is_running`), " "`handicap_line`=VALUES(`handicap_line`), " "`handicap_home_odds`=VALUES(`handicap_home_odds`), " "`handicap_away_odds`=VALUES(`handicap_away_odds`), " "`total_line`=VALUES(`total_line`), " "`total_over_odds`=VALUES(`total_over_odds`), " "`total_under_odds`=VALUES(`total_under_odds`), " "`home_win_odds`=VALUES(`home_win_odds`), " "`draw_odds`=VALUES(`draw_odds`), " "`away_win_odds`=VALUES(`away_win_odds`), " "`half_handicap_line`=VALUES(`half_handicap_line`), " "`half_home_win_odds`=VALUES(`half_home_win_odds`), " "`half_draw_odds`=VALUES(`half_draw_odds`), " "`half_away_win_odds`=VALUES(`half_away_win_odds`), " "`raw_data`=VALUES(`raw_data`), " "`markets_json`=VALUES(`markets_json`), " "`update_time`=VALUES(`update_time`)" ) saved_count = 0 with self.conn.cursor() as cur: for item in items: params = ( item["source_site"], item["source_url"], item["source_match_id"], item["source_parent_id"], item["source_league_id"], item["source_event_id"], item["sport_type"], item["show_type"], item["odds_type"], item["league_name"], item["match_time_text"], item["home_team_id"], item["away_team_id"], item["home_team"], item["away_team"], item["strong_side"], int(item["more_count"]), int(item["is_running"]), item["handicap_line"], item["handicap_home_odds"], item["handicap_away_odds"], item["total_line"], item["total_over_odds"], item["total_under_odds"], item["home_win_odds"], item["draw_odds"], item["away_win_odds"], item["half_handicap_line"], item["half_home_win_odds"], item["half_draw_odds"], item["half_away_win_odds"], now_ts, _dump_json(item["raw_data"]), _dump_json(item["markets"]), now_ts, ) cur.execute(sql, params) saved_count += 1 self.conn.commit() return saved_count def _build_result( *, success: bool, candidate_count: int, saved_count: int, source_url: str = "", skipped_count: int = 0, errors: Optional[List[str]] = None, ) -> Dict[str, Any]: errors = errors or [] summary_text = ( f"第三方赛事赔率抓取完成: 来源 {source_url or '-'}, 待写入 {candidate_count} 场, " f"入库 {saved_count} 场, 跳过 {skipped_count} 项, 失败 {len(errors)} 项" ) result = { "success": bool(success), "candidate_count": int(candidate_count), "saved_count": int(saved_count), "count": int(saved_count), "source_url": source_url, "skipped_count": int(skipped_count), "failed_count": len(errors), "summary_text": summary_text, } if errors: result["error"] = "; ".join(errors[:5]) return result def run(db_config: Dict[str, Any], env: Optional[Dict[str, str]] = None) -> Dict[str, Any]: env = env if env is not None else os.environ username = (env.get("ODDS_USERNAME") or env.get("HG_ODDS_USERNAME") or "").strip() password = (env.get("ODDS_PASSWORD") or env.get("HG_ODDS_PASSWORD") or "").strip() if not username or not password: return _build_result( success=False, candidate_count=0, saved_count=0, errors=["ODDS_USERNAME/ODDS_PASSWORD 未配置"], ) base_urls = [_normalize_base_url(url) for url in _split_env_list(env.get("ODDS_BASE_URLS", ""), DEFAULT_BASE_URLS)] sport_types = [item.upper() for item in _split_env_list(env.get("ODDS_SPORT_TYPES", ""), DEFAULT_SPORT_TYPES)] show_types = [item.lower() for item in _split_env_list(env.get("ODDS_SHOW_TYPES", ""), DEFAULT_SHOW_TYPES)] timeout = _safe_int(env.get("ODDS_TIMEOUT"), 20) verify_tls = _bool_env(env.get("ODDS_VERIFY_TLS"), default=False) langx = (env.get("ODDS_LANGX") or "zh-cn").strip() odds_type = (env.get("ODDS_TYPE") or "H").strip().upper() request_delay = float(env.get("ODDS_REQUEST_DELAY_SECONDS") or REQUEST_DELAY_SECONDS) client: Optional[HgOddsClient] = None login_errors: List[str] = [] for base_url in base_urls: try: candidate = HgOddsClient( base_url=base_url, username=username, password=password, timeout=timeout, verify_tls=verify_tls, langx=langx, odds_type=odds_type, ) candidate.login() client = candidate break except Exception as exc: # noqa: BLE001 login_errors.append(f"{base_url}: {exc}") if client is None: return _build_result(success=False, candidate_count=0, saved_count=0, errors=login_errors) items: List[Dict[str, Any]] = [] fetch_errors: List[str] = [] skipped_count = 0 for sport_type in sport_types: for show_type in show_types: try: items.extend(client.fetch_game_list(sport_type, show_type)) except OddsGameListUnavailable: skipped_count += 1 except Exception as exc: # noqa: BLE001 fetch_errors.append(f"{sport_type}/{show_type}: {exc}") if request_delay > 0: time.sleep(request_delay) repo = None try: repo = MatchOddsRepository(db_config) repo.ensure_table() saved_count = repo.upsert_latest(items, now_ts=int(time.time())) except Exception as exc: # noqa: BLE001 return _build_result( success=False, candidate_count=len(items), saved_count=0, source_url=client.base_url, skipped_count=skipped_count, errors=fetch_errors + [f"database: {exc}"], ) finally: if repo is not None: repo.close() return _build_result( success=len(fetch_errors) == 0, candidate_count=len(items), saved_count=saved_count, source_url=client.base_url, skipped_count=skipped_count, errors=fetch_errors, ) if __name__ == "__main__": from src.core.config import load_config cfg = load_config().database print( json.dumps( run( { "host": cfg.host, "port": cfg.port, "user": cfg.username, "password": cfg.password, "database": cfg.database, "charset": cfg.charset, "prefix": cfg.prefix, } ), ensure_ascii=False, ) )