239 lines
8.4 KiB
Python
239 lines
8.4 KiB
Python
"""
|
|
懂球帝比赛实时详情采集脚本
|
|
- 每5秒轮询 la_match 表，找出已到开赛时间但未结束的比赛
|
|
- 异步并发采集每场比赛的 liveDetail 页面
|
|
- 解析 __NUXT__ 数据，更新 la_match 比分/状态，写入 la_match_data
|
|
- 每个异步任务间隔5-10秒随机
|
|
- 比赛结束后自动停止该场采集
|
|
"""
|
|
import asyncio
|
|
import random
|
|
import signal
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
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.core.config import get_config
|
|
from src.storage.database import Database
|
|
from src.parser.nuxt_parser import parse_nuxt_match_sample
|
|
|
|
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}
|
|
|
|
# 正在采集中的 match_id 集合，防止重复调度
|
|
_active_tasks: dict = {} # match_id -> asyncio.Task
|
|
|
|
# 全局停止信号
|
|
_stop_event = asyncio.Event()
|
|
|
|
|
|
def fetch_live_detail(match_id: int) -> str:
|
|
"""同步请求 liveDetail 页面（在线程池中执行）"""
|
|
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
|
|
|
|
|
|
async def crawl_single_match(match_id: int, match_info: dict, db: Database):
|
|
"""采集单场比赛的实时数据，循环直到比赛结束"""
|
|
home = match_info.get("home_team", "")
|
|
away = match_info.get("away_team", "")
|
|
league = match_info.get("league_name", "")
|
|
logger.info(f"[开始采集] match_id={match_id} {league} {home} vs {away}")
|
|
|
|
consecutive_errors = 0
|
|
max_errors = 5
|
|
|
|
while not _stop_event.is_set():
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
html = await loop.run_in_executor(None, fetch_live_detail, match_id)
|
|
|
|
match_sample = parse_nuxt_match_sample(html)
|
|
if not match_sample:
|
|
logger.warning(f"[{match_id}] 解析 __NUXT__ 失败，跳过本轮")
|
|
consecutive_errors += 1
|
|
if consecutive_errors >= max_errors:
|
|
logger.error(f"[{match_id}] 连续 {max_errors} 次失败，停止采集")
|
|
break
|
|
await asyncio.sleep(random.uniform(5, 10))
|
|
continue
|
|
|
|
consecutive_errors = 0
|
|
|
|
# 提取关键数据
|
|
status_str = str(match_sample.get("status", ""))
|
|
status_int = STATUS_MAP.get(status_str, 0)
|
|
fs_a = match_sample.get("fs_A", "")
|
|
fs_b = match_sample.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(match_sample.get("hts_A", ""))
|
|
hts_b = str(match_sample.get("hts_B", ""))
|
|
half_score = f"{hts_a}-{hts_b}" if hts_a and hts_b and hts_a != "" and hts_b != "" else ""
|
|
minute = str(match_sample.get("minute", ""))
|
|
minute_period = str(match_sample.get("minute_period", ""))
|
|
current_minute = minute_period if minute_period else minute
|
|
|
|
logger.info(
|
|
f"[{match_id}] {league} {home} {home_score}-{away_score} {away} | "
|
|
f"status={status_str} minute={current_minute}"
|
|
)
|
|
|
|
# 更新 la_match
|
|
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=current_minute,
|
|
)
|
|
|
|
# 写入 la_match_data
|
|
await db.upsert_match_data(match_sample)
|
|
|
|
# 比赛结束
|
|
if status_str == "Played":
|
|
logger.info(f"[{match_id}] 比赛已结束: {home} {home_score}-{away_score} {away}")
|
|
break
|
|
|
|
except asyncio.CancelledError:
|
|
logger.info(f"[{match_id}] 任务被取消")
|
|
break
|
|
except Exception as e:
|
|
consecutive_errors += 1
|
|
logger.error(f"[{match_id}] 采集异常: {e}")
|
|
if consecutive_errors >= max_errors:
|
|
logger.error(f"[{match_id}] 连续 {max_errors} 次异常，停止采集")
|
|
break
|
|
|
|
# 随机间隔 5-10 秒
|
|
delay = random.uniform(5, 10)
|
|
try:
|
|
await asyncio.wait_for(_stop_event.wait(), timeout=delay)
|
|
break # stop_event 被设置，退出
|
|
except asyncio.TimeoutError:
|
|
pass # 正常超时，继续下一轮
|
|
|
|
logger.info(f"[结束采集] match_id={match_id} {league} {home} vs {away}")
|
|
|
|
|
|
async def poll_loop(db: Database, interval: int = 5):
|
|
"""主轮询循环：每 interval 秒检查一次需要采集的比赛"""
|
|
logger.info(f"启动轮询循环，间隔 {interval} 秒")
|
|
|
|
while not _stop_event.is_set():
|
|
try:
|
|
now_ts = int(time.time())
|
|
matches = await db.get_live_matches(now_ts)
|
|
|
|
# 清理已完成的任务
|
|
done_ids = [mid for mid, task in _active_tasks.items() if task.done()]
|
|
for mid in done_ids:
|
|
del _active_tasks[mid]
|
|
|
|
if matches:
|
|
new_count = 0
|
|
for m in matches:
|
|
mid = int(m["match_id"])
|
|
if mid in _active_tasks:
|
|
continue # 已在采集中
|
|
task = asyncio.create_task(crawl_single_match(mid, m, db))
|
|
_active_tasks[mid] = task
|
|
new_count += 1
|
|
# 每启动一个任务间隔 1-2 秒，避免瞬间并发太多
|
|
if new_count > 0:
|
|
await asyncio.sleep(random.uniform(1, 2))
|
|
|
|
if new_count > 0:
|
|
logger.info(f"新启动 {new_count} 个采集任务，当前活跃: {len(_active_tasks)}")
|
|
else:
|
|
active_count = len(_active_tasks)
|
|
if active_count > 0:
|
|
logger.debug(f"无新比赛，当前活跃任务: {active_count}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"轮询异常: {e}")
|
|
|
|
# 等待 interval 秒或停止信号
|
|
try:
|
|
await asyncio.wait_for(_stop_event.wait(), timeout=interval)
|
|
break
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
|
|
# 停止所有活跃任务
|
|
logger.info(f"停止所有活跃任务 ({len(_active_tasks)} 个)...")
|
|
for mid, task in _active_tasks.items():
|
|
task.cancel()
|
|
if _active_tasks:
|
|
await asyncio.gather(*_active_tasks.values(), return_exceptions=True)
|
|
_active_tasks.clear()
|
|
|
|
|
|
async def run(interval: int = 5):
|
|
db = Database()
|
|
try:
|
|
await db.connect()
|
|
logger.info("数据库连接成功")
|
|
await poll_loop(db, interval)
|
|
finally:
|
|
await db.close()
|
|
logger.info("数据库连接已关闭")
|
|
|
|
|
|
def main():
|
|
import argparse
|
|
parser = argparse.ArgumentParser(description="懂球帝比赛实时详情采集")
|
|
parser.add_argument("--interval", type=int, default=5, help="轮询间隔秒数 (默认5)")
|
|
args = parser.parse_args()
|
|
|
|
logger.info("=" * 60)
|
|
logger.info("懂球帝比赛实时详情采集 (liveDetail)")
|
|
logger.info(f"轮询间隔: {args.interval} 秒")
|
|
logger.info("按 Ctrl+C 停止")
|
|
logger.info("=" * 60)
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
def handle_signal():
|
|
logger.info("收到停止信号...")
|
|
_stop_event.set()
|
|
|
|
try:
|
|
loop.add_signal_handler(signal.SIGINT, handle_signal)
|
|
loop.add_signal_handler(signal.SIGTERM, handle_signal)
|
|
except NotImplementedError:
|
|
pass # Windows 不支持 add_signal_handler
|
|
|
|
try:
|
|
loop.run_until_complete(run(interval=args.interval))
|
|
except KeyboardInterrupt:
|
|
logger.info("收到 Ctrl+C，正在停止...")
|
|
_stop_event.set()
|
|
loop.run_until_complete(asyncio.sleep(1))
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|