Python实战:田径赛事成绩数据采集、清洗与可视化全流程
2026/9/6 4:04:02 网站建设 项目流程

2026年全国田径锦标赛上,男子田径选手刘凯跑出46.22秒的成绩晋级决赛。很多技术圈读者可能第一反应是:这个成绩到底是什么水平?背后的数据分析怎么做的?如果我要做一套田径赛事成绩采集、清洗、可视化分析的流程,应该从哪里入手?

这次我们不聊“精神”“意志”这些抽象话题,直接从一个具体的赛事成绩出发,用Python把体育数据的采集、解析、入库、查询、可视化完整串一遍。文章会涉及requests数据请求、Pandas数据清洗、SQLite存储、ECharts/Pyecharts可视化,以及多线程批量爬虫的踩坑记录。如果你是做数据分析、爬虫开发或者体育数据平台相关工作,这篇文章可以直接当一份实战手册收藏。

1. 核心能力速览

能力项说明
数据源体育赛事公开成绩页面、赛事官网公告、第三方体育数据平台
技术栈Python 3.8+、requests、Pandas、SQLite、Pyecharts
核心功能赛事成绩抓取、成绩清洗、选手排名、分段对比、可视化输出
支持批量支持多线程批量请求,可按比赛项目、轮次、组别分批抓取
输出格式CSV、JSON、SQLite 数据库、HTML 图表
适用场景体育新闻报道辅助、赛事数据平台搭建、训练数据分析、成绩趋势研究
硬件要求普通办公电脑即可,无 GPU 需求,内存建议 8G 以上

这个方案不依赖高性能显卡,开发调试和日常跑数据用普通笔记本就能完成。核心难点不在算力,而在数据源的稳定性、字段清理和批量请求的并发控制。

2. 适用场景与使用边界

体育数据分析适合这几类场景:

第一类是内容生产场景。自媒体、体育编辑需要快速从官方成绩单中提取名次、成绩、选手信息,生成战报或数据长图。用程序替代手工复制粘贴,可以把发布时效从分钟级缩短到秒级。

第二类是训练辅助场景。教练员可以收集一个选手在不同比赛中的成绩,制作时间趋势图、分段配速图,观察竞技状态的波动。

第三类是赛事数据平台建设。如果要做一个小型成绩查询系统,把历届比赛数据入库,这套流程就能直接复用。

但这里要提醒几个边界:

  • 数据采集必须遵守目标网站的 robots 协议和版权要求。公开的成绩数据一般可以合理使用,但转载、商用前要确认授权。
  • 个人隐私信息(如运动员身份证号、联系方式)绝不能采集和公开。
  • 请求频率要控制,避免对赛事官网造成访问压力。批量采集时建议加随机延时和重试机制。
  • 46.22 秒这个成绩的具体项目归属,以官方成绩公告为准。做数据分析时,成绩字段必须附带项目、组别、轮次、风速等上下文信息,否则跨项目比较没有意义。

从数据角度说,体育成绩数据的最大特点是不均一。田赛和径赛的计量单位不同,不同项目的成绩格式不同(秒、分:秒、米、厘米),同一选手不同轮次的数据结构也可能变化。清洗环节必须针对项目规则做定制化处理。

3. 环境准备与前置条件

先说操作系统,Windows 10/11、macOS、Linux 都可以,文章里的代码不依赖特定系统。

然后是 Python 环境。建议直接用 Anaconda 创建独立虚拟环境,避免依赖冲突。文章代码基于 Python 3.9 测试,但 3.8 以上的版本基本都能跑。

需要安装的核心库:

pip install requests pandas sqlite3-utils pyecharts beautifulsoup4 lxml

如果只是做轻量级数据获取,requests + BeautifulSoup 就够了。要做数据处理和图表输出,再补 Pandas 和 Pyecharts。

建议目录结构:

track_field_data/ ├── data/ # 原始数据存放目录 │ ├── raw/ # 抓取到的原始响应 │ └── processed/ # 清洗后的结构化数据 ├── output/ # 图表和导出文件 ├── logs/ # 运行日志 ├── crawler.py # 爬虫脚本 ├── parser.py # 数据解析脚本 ├── analyzer.py # 数据分析与可视化脚本 └── config.py # 公共配置

提前检查两件事:

第一,网络环境能否正常访问目标数据源。有些官网响应较慢,超时时间建议设置为 15 到 30 秒。

第二,磁盘空间。纯文本数据占不了多少空间,但如果要保存大量 HTML 原始页面,建议预留至少 5GB 空间。

数据库方面,SQLite 足够应付小型赛事数据平台,不需要单独安装数据库服务。

4. 采集方案设计与启动方式

4.1 数据源分析

先明确需求:我们要获取的是类似“刘凯 46.22 秒晋级决赛”这样的赛事成绩数据。一个完整的成绩记录通常包含以下字段:

{ "rank": 1, "athlete": "刘凯", "team": "单位/省份", "result": "46.22", "unit": "秒", "round": "预赛", "group": "第3组", "wind": 0.0, "qualified": "Q" }

实际赛事页面的 HTML 结构可能复杂,但核心思路是固定的:找到成绩表格所在节点,逐行解析单元格,按字段映射关系入库。

4.2 请求封装

写一个通用的请求函数,统一处理 header、超时、重试和日志:

import requests import time import random from datetime import datetime HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept": "text/html,application/json" } def fetch_html(url, max_retry=3, timeout=20): """抓取HTML页面,带重试机制""" for attempt in range(1, max_retry + 1): try: resp = requests.get(url, headers=HEADERS, timeout=timeout) if resp.status_code == 200: resp.encoding = resp.apparent_encoding return resp.text else: log_warning(f"HTTP {resp.status_code}, url: {url}") except requests.Timeout: log_warning(f"Timeout on attempt {attempt}, url: {url}") except requests.ConnectionError: log_warning(f"ConnectionError on attempt {attempt}, url: {url}") time.sleep(random.uniform(1.5, 3.5)) return None def log_warning(msg): print(f"[{datetime.now():%H:%M:%S}] WARN - {msg}")

每次请求后加随机延时,防止频率过高。延时控制在 1.5 到 3.5 秒之间,单线程跑一天大约可以请求 25000 个页面,对中小型赛事完全够用。

4.3 多线程批量采集

如果要同时抓取多个比赛项目的成绩,可以用 ThreadPoolExecutor。但要注意控制并发数量,建议 4 到 8 个线程,配合一个队列来管理 URL:

import csv import threading from concurrent.futures import ThreadPoolExecutor, as_completed url_queue = [ ("m_400_pre_group3", "https://example.com/results/m400/pre/group3"), ("m_400_semi_group1", "https://example.com/results/m400/semi/group1"), # 其他比赛组别URL ] results = [] lock = threading.Lock() failed_urls = [] def process_url(item): key, url = item html = fetch_html(url) if html is None: raise RuntimeError(f"Failed to fetch: {key}") parsed = parse_result_page(html, key) return parsed with ThreadPoolExecutor(max_workers=4) as executor: future_map = {executor.submit(process_url, item): item for item in url_queue} for future in as_completed(future_map): item = future_map[future] try: data = future.result() with lock: results.extend(data) except Exception as e: with lock: failed_urls.append((item[0], str(e)))

失败的 URL 必须单独记录,方便后续重跑。因为体育比赛页面多为动态加载,如果 HTML 中拿不到表格数据,需要检查页面是否使用 AJAX 接口,这种情况下直接请求后端 JSON 接口效率更高。

5. 数据解析与清洗

原始 HTML 解析只是一个中间环节,核心工作是数据清洗。这里以 Pandas 处理为例。

import pandas as pd # 假设 results 是解析后的字典列表 df = pd.DataFrame(results) print("原始数据行数:", len(df)) print("缺失值统计:") print(df.isnull().sum())

典型的清理步骤包括:

成绩字段统一。径赛成绩可能有“46.22”“1:49.88”“13.35”等格式,需要统一转换为秒或保留原始字符串增加一个“成绩秒”数值列:

def convert_result_to_seconds(value): """将 46.22 或 1:49.88 转换为秒""" if isinstance(value, str) and ":" in value: parts = value.split(":") minutes = int(parts[0]) seconds = float(parts[1]) return round(minutes * 60 + seconds, 2) try: return float(value) except (TypeError, ValueError): return None df["result_seconds"] = df["result"].apply(convert_result_to_seconds)

分组排序。预赛分多组进行,每组独立排名,但所有组别选手要混合比较成绩选出晋级名单。这个逻辑用 Pandas 可以快速完成:

# 组内排名 df["rank_in_group"] = df.groupby("group")["result_seconds"].rank(method="min") # 所有选手按成绩排序 df_sorted = df.sort_values(by=["result_seconds"], ascending=True).reset_index(drop=True) # 取每组前三名 + 剩余最好成绩若干位,具体规则按赛事规程 qualified_by_group = df_sorted.groupby("group").head(2) remaining_quota = 2 # 举例:剩余最好成绩递补2人 best_remaining = ( df_sorted[~df_sorted.index.isin(qualified_by_group.index)] .head(remaining_quota) ) df_qualified = pd.concat([qualified_by_group, best_remaining]).sort_values("result_seconds")

风速等非成绩字段处理。田径比赛跨项比较时,风速是重要参考,超过一定限制的成绩不参与排名或纪录认定。字段保持浮点型,无记录时填 None。

清洗后的数据导出为标准格式:

df.to_csv("./output/qualification_results.csv", index=False, encoding="utf-8-sig") df.to_json("./output/qualification_results.json", orient="records", force_ascii=False, indent=2)

导出 CSV 用utf-8-sig编码,Excel 打开才不会出现中文乱码。

6. 接口 API 与批量任务设计

如果要把这套数据能力接入自建系统,建议把清洗结果封装成一个轻量 API 服务。可以使用 FastAPI 快速实现。

# api_server.py import sqlite3 from fastapi import FastAPI, Query from pydantic import BaseModel app = FastAPI(title="Track Field Result API") def get_db_conn(): conn = sqlite3.connect("./data/results.db") conn.row_factory = sqlite3.Row return conn @app.get("/results/athlete") def get_athlete_results(name: str = Query(..., description="运动员姓名")): conn = get_db_conn() rows = conn.execute( "SELECT * FROM athlete_results WHERE athlete = ? ORDER BY date", (name,) ).fetchall() conn.close() return [dict(row) for row in rows] @app.get("/results/event") def get_event_results(event: str, round: str = None): conn = get_db_conn() if round: rows = conn.execute( "SELECT * FROM athlete_results WHERE event = ? AND round = ? ORDER BY result_seconds", (event, round), ).fetchall() else: rows = conn.execute( "SELECT * FROM athlete_results WHERE event = ? ORDER BY result_seconds", (event,), ).fetchall() conn.close() return [dict(row) for row in rows]

启动方式:

uvicorn api_server:app --host 127.0.0.1 --port 8000

访问示例:

curl "http://127.0.0.1:8000/results/athlete?name=刘凯"

返回 JSON:

[ { "athlete": "刘凯", "event": "m_400", "round": "预赛第3组", "result": "46.22", "result_seconds": 46.22, "rank": 1, "qualified": "Q" } ]

批量任务的工程化建议:

  • 把待抓取的任务列表存到 JSON 或 CSV 文件中,程序从文件读任务,而不是硬编码在脚本里。
  • 每个任务记录状态:pending、running、success、failed。
  • 失败任务最多重试 3 次,重试间隔递增(5 秒、15 秒、30 秒)。
  • 定期保存断点,程序中断后重新启动时从断点继续,而不是全量重跑。

如果需要定时自动采集,可以用系统 cron(Linux/macOS)或任务计划程序(Windows),每天凌晨跑一次增量更新。

7. 资源占用与性能观察

这套数据处理方案不吃显卡,资源占用主要看四个方面。

内存占用。抓取 1000 个比赛页面,每个页面 HTML 约 100 到 300KB,一次性全部读入内存大约占 200 到 400MB。建议边抓边解析,只把结构化结果保留在内存中,避免把整个 HTML 堆在一起。

并发请求性能。单线程串行请求,每次 2 秒延时,一分钟大约能处理 30 个页面。4 线程并发,一分钟可以处理 80 到 110 个页面。如果目标网站性能好,可以把线程数提高到 8,但不要盲目加大,否则很容易触发反爬策略或给目标服务器造成压力。

磁盘占用。纯文本存储,一万条成绩记录加关联信息,SQLite 数据库文件通常只有几十 MB。真正占用空间的是原始 HTML 备份,如果不需要回溯分析,不推荐长期保存原始页面。

CPU 占用。解析 HTML 时 BeautifulSoup 比较吃 CPU,但量级很低。大规模清洗任务如果用 Pandas,注意apply函数不要滥用,能向量化操作就向量化操作。

如果要观察实时进程状态,可以用nvidia-smi看 GPU,但这个方案不需要。普通任务管理器观察内存和网络占用即可。如果跑批量抓取速度变慢,优先检查网络连接质量,而不是本地 CPU 性能。

8. 常见问题与排查方法

问题现象可能原因排查方式解决方案
抓取到空白页面页面为动态渲染,HTML 中没有数据查看响应内容,检查是否有 JS 渲染改用 AJAX 接口或 Selenium
请求被拒绝 403/429请求频率过高或缺少头部信息查看响应码和响应头增加延时、更换 User-Agent、使用代理池
中文乱码编码识别错误打印resp.apparent_encoding手动设置resp.encoding = "utf-8"
成绩格式解析失败字段含单位、文本说明检查原始字段样例编写正则处理异常值
SQLite 数据库锁死多线程同时写入查看报错信息写入统一走单线程或使用队列
CSV 打开乱码编码问题用文本编辑器查看文件编码使用encoding="utf-8-sig"导出
API 服务启动失败端口被占用检查 8000 端口更换端口:--port 8001
批量任务中途失败网络波动或页面结构调整查看任务日志重启时从失败断点继续执行

最大概率遇到的坑是页面结构变化。赛事官网改版后,之前的 CSS 选择器全失效,解析函数直接返回空列表。这个没有一劳永逸的解决方案,只能做好日志监控,每次抓取结束后检查有效数据行数,低于阈值就自动告警。

批量任务部分,建议每批次结束输出如下格式摘要:

批次:m400_pre 任务数:72 成功:70 失败:2 有效记录:840 耗时:4分38秒

这样问题一出现就能快速定位到具体批次。

9. 最佳实践与使用建议

第一,第一次跑通全流程时,先用一个组别的数据调试,不要一上来就全量抓取。确认数据解析、清洗、导出都没有问题后,再扩大任务范围。

第二,数据字段设计要留扩展空间。除了成绩字段,必须记录比赛日期、比赛地点、项目编号、轮次、组别、风速、温度这些上下文信息。体育数据的价值往往体现在纵向对比中,缺少元信息的成绩记录后续很难用起来。

第三,原始响应建议压缩存储,按日期和比赛批次命名。虽然不一定每次都会回溯原始 HTML,但一旦出现解析规则调整,原始数据可以帮你快速验证新旧解析逻辑的差异。

第四,清洗逻辑单独写成一个模块,做好注释。因为清洗规则的调整频次远高于爬虫规则,把两者解耦可以大幅降低维护成本。

第五,涉及成绩排名时,务必阅读赛事官方规程。不同赛事的晋级规则不同,有的是每组 N 名 + 最好成绩递补,有的是直接取总排名前 N 名。代码里的晋级逻辑必须可配置,不能写死。

第六,数据发布前必须人工复核。程序可以告诉你“刘凯 46.22 秒”在数据表中排第几名,但项目的真实性、选手资格、成绩有效性,都需要和官方公告交叉核对。

10. 总结与下一步

这套“赛事数据采集 + 清洗 + 查询 + 可视化”方案,核心价值在于把一段零散的成绩信息(比如“刘凯 46.22 秒晋级决赛”)扩展成可查询、可对比、可追溯的结构化数据。整个流程不需要高端硬件,一台普通电脑就能跑通,适合体育编辑、数据分析师和赛事平台开发者直接复用。

最先应该验证的功能是成绩解析和晋级判定逻辑。找一组历史成绩数据,手动算一遍排名,再和程序输出对比,确认无误后再接入批量任务。

最容易踩的坑有三个:页面结构变化导致解析失效、成绩格式不统一导致清洗报错、并发请求频率过高导致 IP 被限制。这三个问题分别用日志监控、正则容错和请求限速解决。

后续可以继续扩展的方向:

  • 用 Flask/FastAPI 做一套完整的成绩查询网站,支持按选手、项目、时间段检索。
  • 接入 ECharts 做选手成绩趋势图和比赛分段成绩对比图。
  • 增加成绩预测模型,基于历史数据预测选手下一场表现。
  • 把任务调度做成定时增量更新,比赛日结束后自动生成战报数据包。

如果目标是做内容发布,这一步做完,数据就可以直接转换成图表和结构化摘要供编辑使用。如果目标是做数据平台,建议先把数据库表结构和管理后台设计好,再逐步接入更多赛事数据。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询