sbnews/docker/crawler/scripts/crawl_matches.py

"""
懂球帝比赛数据采集脚本
- 通过 /api/data/tab/new/important 接口获取比赛记录
- 按周循环翻页 (start + 7天)
- 按天匹配，已有数据跳过
- 每天执行一次
"""
import asyncio
import sys
import time
from datetime import datetime, timedelta
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

API_URL = "https://www.dongqiudi.com/api/data/tab/new/important"

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": "application/json, text/plain, */*",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}


def fetch_page(start_str: str) -> dict:
    params = {
        "start": f"{start_str}next",
        "init": "1",
        "platform": "www",
    }
    resp = curl_requests.get(
        API_URL, params=params, headers=HEADERS,
        impersonate="chrome", timeout=20,
    )
    resp.raise_for_status()
    return resp.json()


def collect_range(start_date: datetime, end_date: datetime) -> list:
    all_items = []
    current = start_date
    page = 0

    while current < end_date:
        page += 1
        start_str = current.strftime("%Y-%m-%d") + " 00:00:00"
        logger.info(f"[第{page}页] 请求: start={start_str}")

        try:
            data = fetch_page(start_str)
        except Exception as e:
            logger.error(f"请求失败: {e}")
            current += timedelta(days=7)
            time.sleep(2)
            continue

        items = data.get("list", [])
        logger.info(f"  返回 {len(items)} 条记录")

        for item in items:
            all_items.append(item)

        current += timedelta(days=7)
        time.sleep(3)

    return all_items


async def run(start_date: datetime, end_date: datetime):
    cfg = get_config()
    db = Database()

    try:
        existing_ids = await db.get_existing_match_ids()
        logger.info(f"数据库已有 {len(existing_ids)} 条比赛记录(有match_id)")

        days = (end_date - start_date).days
        logger.info(f"采集范围: {start_date.strftime('%Y-%m-%d')} ~ {end_date.strftime('%Y-%m-%d')} ({days}天)")

        all_items = collect_range(start_date, end_date)

        unique_items = {}
        for item in all_items:
            mid = int(item["match_id"])
            if mid not in unique_items:
                unique_items[mid] = item

        logger.info(f"去重后共 {len(unique_items)} 条唯一比赛")

        new_items = [item for mid, item in unique_items.items() if mid not in existing_ids]
        logger.info(f"新增比赛: {new_items.__len__()} 条 (跳过已有 {len(unique_items) - len(new_items)} 条)")

        if not new_items:
            logger.info("没有新比赛需要入库")
            return

        fixture_items = [i for i in new_items if i.get("status") == "Fixture"]
        played_items = [i for i in new_items if i.get("status") == "Played"]
        logger.info(f"  未开始: {len(fixture_items)} 条, 已结束: {len(played_items)} 条")

        count = await db.upsert_la_match(new_items)
        logger.info(f"✅ 入库完成: {count} 条比赛记录")

        await db.log_crawl(
            crawl_type="match_important",
            target=API_URL,
            success=True,
            elapsed=0,
            record_count=count,
        )

    except Exception as e:
        logger.error(f"采集失败: {e}", exc_info=True)
        try:
            await db.log_crawl(
                crawl_type="match_important",
                target=API_URL,
                success=False,
                elapsed=0,
                error=str(e),
            )
        except Exception:
            pass
        raise
    finally:
        await db.close()


def parse_date(s: str) -> datetime:
    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y%m%d"):
        try:
            return datetime.strptime(s, fmt)
        except ValueError:
            continue
    raise ValueError(f"无法解析日期: {s}，支持格式: YYYY-MM-DD / YYYY/MM/DD / YYYYMMDD")


def main():
    import argparse
    parser = argparse.ArgumentParser(
        description="懂球帝比赛数据采集",
        epilog="示例:\n"
               "  python crawl_matches.py --weeks 4          # 从今天起采集4周\n"
               "  python crawl_matches.py --start 2026-01-01 --end 2026-02-01  # 指定时间段\n"
               "  python crawl_matches.py --start 2026-01-01  # 从指定日期到今天\n",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--weeks", type=int, default=None, help="从今天起采集未来N周 (默认4)")
    parser.add_argument("--start", type=str, default=None, help="起始日期 (YYYY-MM-DD)")
    parser.add_argument("--end", type=str, default=None, help="结束日期 (YYYY-MM-DD)，不指定则到今天")
    args = parser.parse_args()

    today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)

    if args.start:
        start_date = parse_date(args.start)
        if args.end:
            end_date = parse_date(args.end)
        else:
            end_date = today + timedelta(days=1)
    else:
        weeks = args.weeks or 4
        start_date = today
        end_date = today + timedelta(weeks=weeks)

    if start_date >= end_date:
        logger.error(f"起始日期 {start_date.strftime('%Y-%m-%d')} >= 结束日期 {end_date.strftime('%Y-%m-%d')}")
        sys.exit(1)

    logger.info("=" * 60)
    logger.info("懂球帝比赛数据采集 (important)")
    logger.info(f"{start_date.strftime('%Y-%m-%d')} ~ {end_date.strftime('%Y-%m-%d')}")
    logger.info("=" * 60)

    asyncio.run(run(start_date, end_date))


if __name__ == "__main__":
    main()