160 lines
5.0 KiB
Python
160 lines
5.0 KiB
Python
"""
|
|
获取单场比赛详情
|
|
用法: python scripts/fetch_match_detail.py <match_id> [--save]
|
|
--save 将数据写入 la_match_data 并更新 la_match
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from curl_cffi import requests as curl_requests
|
|
from loguru import logger
|
|
|
|
from src.parser.nuxt_parser import parse_nuxt_match_sample
|
|
from src.storage.database import Database
|
|
|
|
LIVE_DETAIL_URL = "https://www.dongqiudi.com/liveDetail/{match_id}"
|
|
|
|
HEADERS = {
|
|
"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",
|
|
"Referer": "https://www.dongqiudi.com/",
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
|
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
|
|
}
|
|
|
|
STATUS_MAP = {"Fixture": 0, "Playing": 1, "Played": 2}
|
|
|
|
|
|
def fetch_html(match_id: int) -> str:
|
|
url = LIVE_DETAIL_URL.format(match_id=match_id)
|
|
resp = curl_requests.get(url, headers=HEADERS, impersonate="chrome", timeout=20)
|
|
resp.raise_for_status()
|
|
return resp.text
|
|
|
|
|
|
def display(ms: dict):
|
|
"""格式化输出比赛详情"""
|
|
status = ms.get("status", "")
|
|
fs_a = ms.get("fs_A", "")
|
|
fs_b = ms.get("fs_B", "")
|
|
hts_a = ms.get("hts_A", "")
|
|
hts_b = ms.get("hts_B", "")
|
|
|
|
print()
|
|
print("=" * 60)
|
|
print(f" 比赛ID: {ms.get('match_id')}")
|
|
print(f" 赛事: {ms.get('competition_name', '')} {ms.get('match_title', '')}")
|
|
print(f" 轮次: {ms.get('gameweek', '') or ms.get('round_name', '') or '-'}")
|
|
print("-" * 60)
|
|
print(f" 主队: {ms.get('team_A_name', '')} (ID: {ms.get('team_A_id', '')})")
|
|
print(f" 客队: {ms.get('team_B_name', '')} (ID: {ms.get('team_B_id', '')})")
|
|
print("-" * 60)
|
|
|
|
if status == "Played":
|
|
print(f" 比分: {fs_a} - {fs_b} (全场)")
|
|
if hts_a and hts_b:
|
|
print(f" 半场: {hts_a} - {hts_b}")
|
|
print(f" 状态: 已结束")
|
|
elif status == "Playing":
|
|
minute = ms.get("minute", "")
|
|
period = ms.get("minute_period", "")
|
|
print(f" 比分: {fs_a or 0} - {fs_b or 0} (进行中)")
|
|
print(f" 进行: {period} {minute}")
|
|
else:
|
|
print(f" 状态: 未开始")
|
|
|
|
print(f" 开赛时间: {ms.get('start_play', '')}")
|
|
print(f" 类型: {ms.get('cmp_type', '')}")
|
|
|
|
# 赔率
|
|
home_odds = ms.get("home", "")
|
|
draw_odds = ms.get("draw", "")
|
|
away_odds = ms.get("away", "")
|
|
if home_odds or draw_odds or away_odds:
|
|
print(f" 赔率: 主{home_odds} 平{draw_odds} 客{away_odds}")
|
|
|
|
# 直播源
|
|
tv = ms.get("livingTv", "")
|
|
if tv:
|
|
print(f" 直播: {tv}")
|
|
|
|
print("=" * 60)
|
|
|
|
# 额外信息
|
|
ht_info = ms.get("ht_info", "")
|
|
score_info = ms.get("score_info", "")
|
|
if ht_info:
|
|
print(f" 半场信息: {ht_info}")
|
|
if score_info:
|
|
print(f" 比分信息: {score_info}")
|
|
|
|
# Logo URLs
|
|
print(f"\n 赛事Logo: {ms.get('competition_bk_logo', '')}")
|
|
print(f" 主队Logo: {ms.get('team_A_logo', '')}")
|
|
print(f" 客队Logo: {ms.get('team_B_logo', '')}")
|
|
print()
|
|
|
|
|
|
async def save_to_db(ms: dict):
|
|
"""将数据写入数据库"""
|
|
db = Database()
|
|
try:
|
|
match_id = int(ms.get("match_id", 0))
|
|
status_str = str(ms.get("status", ""))
|
|
status_int = STATUS_MAP.get(status_str, 0)
|
|
fs_a = ms.get("fs_A", "")
|
|
fs_b = ms.get("fs_B", "")
|
|
home_score = int(fs_a) if fs_a and str(fs_a).isdigit() else 0
|
|
away_score = int(fs_b) if fs_b and str(fs_b).isdigit() else 0
|
|
hts_a = str(ms.get("hts_A", ""))
|
|
hts_b = str(ms.get("hts_B", ""))
|
|
half_score = f"{hts_a}-{hts_b}" if hts_a and hts_b else ""
|
|
minute_period = str(ms.get("minute_period", ""))
|
|
|
|
await db.update_match_live(
|
|
match_id=match_id,
|
|
status=status_int,
|
|
home_score=home_score,
|
|
away_score=away_score,
|
|
half_score=half_score,
|
|
current_minute=minute_period,
|
|
)
|
|
await db.upsert_match_data(ms)
|
|
logger.info(f"已保存到数据库: match_id={match_id}")
|
|
finally:
|
|
await db.close()
|
|
|
|
|
|
def main():
|
|
import argparse
|
|
parser = argparse.ArgumentParser(description="获取单场比赛详情")
|
|
parser.add_argument("match_id", type=int, help="懂球帝 match_id")
|
|
parser.add_argument("--save", action="store_true", help="写入数据库")
|
|
parser.add_argument("--json", action="store_true", help="输出原始 JSON")
|
|
args = parser.parse_args()
|
|
|
|
logger.info(f"获取比赛详情: match_id={args.match_id}")
|
|
|
|
html = fetch_html(args.match_id)
|
|
ms = parse_nuxt_match_sample(html)
|
|
|
|
if not ms:
|
|
logger.error("解析失败，无法提取比赛数据")
|
|
sys.exit(1)
|
|
|
|
if args.json:
|
|
print(json.dumps(ms, ensure_ascii=False, indent=2, default=str))
|
|
else:
|
|
display(ms)
|
|
|
|
if args.save:
|
|
asyncio.run(save_to_db(ms))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|