如何用Python批量爬取GBGB网站指定赛道的赛狗赛事结果？

针对你手动输入ID爬取效率低的问题，利用已知的单日期查询API，我们可以实现指定日期范围自动爬取+无数据量限制的方案，核心思路是遍历日期范围、自动处理分页、筛选目标赛道数据，具体实现如下：

核心实现步骤 #

• 生成起始到结束日期之间的所有日期字符串
• 针对每个日期，根据API分页信息循环请求，获取当日全量数据
• 过滤出目标赛道的赛事记录
• 将所有有效数据持久化为NDJSON格式

完整代码实现 #

import requests
from datetime import datetime, timedelta
import ndjson
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session():
    # 创建带重试机制的session，提升爬取稳定性
    session = requests.Session()
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    # 模拟浏览器请求头，避免被拦截
    session.headers.update({
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36"
    })
    return session

def get_dates_in_range(start_date_str, end_date_str):
    # 生成日期范围内的所有日期（格式YYYY-MM-DD）
    start_date = datetime.strptime(start_date_str, "%Y-%m-%d")
    end_date = datetime.strptime(end_date_str, "%Y-%m-%d")
    date_list = []
    current_date = start_date
    while current_date <= end_date:
        date_list.append(current_date.strftime("%Y-%m-%d"))
        current_date += timedelta(days=1)
    return date_list

def fetch_daily_results(session, target_track, date_str):
    base_url = "https://api.gbgb.org.uk/api/results"
    items_per_page = 200
    page = 1
    daily_results = []
    
    while True:
        params = {
            "page": page,
            "itemsPerPage": items_per_page,
            "date": date_str,
            "race_type": "race"
        }
        try:
            response = session.get(base_url, params=params)
            response.raise_for_status()
            data = response.json()
            
            # 筛选目标赛道的赛事数据
            track_results = [item for item in data["items"] if item["track"]["name"] == target_track]
            daily_results.extend(track_results)
            
            # 判断是否还有下一页数据
            total_pages = data["pagination"]["totalPages"]
            if page >= total_pages:
                break
            page += 1
        except Exception as e:
            print(f"日期{date_str}第{page}页数据获取失败: {str(e)}")
            break
    return daily_results

if __name__ == "__main__":
    # 配置参数，按需修改
    START_DATE = "2023-07-01"
    END_DATE = "2023-07-21"
    TARGET_TRACK = "Hove"
    OUTPUT_FILE = "Hove_results.json"
    
    session = create_session()
    all_results = []
    
    # 遍历日期范围爬取数据
    date_list = get_dates_in_range(START_DATE, END_DATE)
    for date in date_list:
        print(f"正在处理日期: {date}")
        daily_data = fetch_daily_results(session, TARGET_TRACK, date)
        all_results.extend(daily_data)
        print(f"日期{date}获取到{len(daily_data)}条Hove赛道数据")
    
    # 保存为NDJSON格式
    with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
        ndjson.dump(all_results, f)
    
    print(f"爬取完成，共获取{len(all_results)}条数据，已保存至{OUTPUT_FILE}")

关键优化说明 #

• 带重试的Session：处理临时网络故障、服务器异常，避免单次失败导致爬取中断
• 自动分页处理：根据API返回的分页信息，自动遍历所有页面，突破单页200条的数据限制
• 赛道精准筛选：在获取当日全量数据后过滤目标赛道，避免无效数据占用内存
• 异常捕获：针对请求过程中的错误做捕获处理，保证程序稳定运行

内容的提问来源于stack exchange，提问作者Rv02