ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

Python实战:田径赛事成绩数据采集、清洗与可视化全流程

Python实战:田径赛事成绩数据采集、清洗与可视化全流程 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_retry3, timeout20): 抓取HTML页面带重试机制 for attempt in range(1, max_retry 1): try: resp requests.get(url, headersHEADERS, timeouttimeout) if resp.status_code 200: resp.encoding resp.apparent_encoding return resp.text else: log_warning(fHTTP {resp.status_code}, url: {url}) except requests.Timeout: log_warning(fTimeout on attempt {attempt}, url: {url}) except requests.ConnectionError: log_warning(fConnectionError 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 个线程配合一个队列来管理 URLimport 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(fFailed to fetch: {key}) parsed parse_result_page(html, key) return parsed with ThreadPoolExecutor(max_workers4) 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(methodmin) # 所有选手按成绩排序 df_sorted df.sort_values(by[result_seconds], ascendingTrue).reset_index(dropTrue) # 取每组前三名 剩余最好成绩若干位具体规则按赛事规程 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, indexFalse, encodingutf-8-sig) df.to_json(./output/qualification_results.json, orientrecords, force_asciiFalse, indent2)导出 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(titleTrack 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 秒。定期保存断点程序中断后重新启动时从断点继续而不是全量重跑。如果需要定时自动采集可以用系统 cronLinux/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 打开乱码编码问题用文本编辑器查看文件编码使用encodingutf-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 做选手成绩趋势图和比赛分段成绩对比图。增加成绩预测模型基于历史数据预测选手下一场表现。把任务调度做成定时增量更新比赛日结束后自动生成战报数据包。如果目标是做内容发布这一步做完数据就可以直接转换成图表和结构化摘要供编辑使用。如果目标是做数据平台建议先把数据库表结构和管理后台设计好再逐步接入更多赛事数据。
返回列表