1. 从手动刷新到批量抓取一个真实的需求场景如果你曾经在某个需要频繁查询成绩、分数或排名的场景里工作过比如教育机构的教务老师、竞赛活动的组织者或者只是帮家里几十个孩子查期末成绩的“热心家长”那你一定对“手动查分”这件事深恶痛绝。想象一下你面前摆着一张几百人的名单每个名字对应一个唯一的查询凭证可能是学号、身份证号或准考证号。你的任务是把他们每个人的分数都查出来汇总成表。你打开浏览器输入网址粘贴第一个凭证点击查询等待页面加载从一堆HTML标签里找到那个分数复制粘贴到Excel里。然后重复这个动作几百次。整个过程枯燥、耗时且极易出错——手一滑可能就输错了数字或者看串了行。这就是“批量查分爬虫”诞生的最朴素动机用程序自动化代替人工的重复劳动。它不是什么高深莫测的黑科技而是一个典型的、解决具体效率问题的自动化脚本。其核心逻辑非常简单程序模拟人的操作自动遍历一个凭证列表依次向目标查询网站发起请求从返回的页面或数据中解析出我们需要的信息分数并最终整理成结构化的格式如CSV或Excel文件。整个过程从几个小时甚至几天的手工劳动缩短到几分钟的脚本运行时间解放出来的精力可以用来做更有价值的分析和沟通工作。然而把想法变成稳定可用的工具中间隔着不少“坑”。网站的结构千差万别有的返回规整的JSON数据有的则是充满嵌套表格的HTML有的查询无需登录有的则需要维护复杂的会话状态更不用说还有验证码、请求频率限制、反爬虫机制这些“拦路虎”。一个健壮的批量查分爬虫不仅仅是会发HTTP请求那么简单它需要处理网络异常、解析各种数据格式、模拟浏览器行为甚至进行简单的智能调度。接下来我将以一个虚构但典型的“全国青少年编程大赛成绩查询”为例拆解构建这样一个爬虫的完整思路、技术选型、核心代码实现以及那些只有踩过坑才知道的注意事项。2. 项目蓝图定义爬虫的“输入”与“输出”在动手写一行代码之前我们必须像建筑师画蓝图一样明确这个爬虫的边界和能力。一个模糊的“查分”想法是无法指导开发的。我们需要将其转化为清晰的技术规格。2.1 目标网站分析与数据接口定位首先我们需要明确爬虫的工作对象。假设目标查询网站是score.competition.org其查询流程如下访问首页可能有一个登录环节本例假设查询仅需准考证号无需登录。在查询输入框中填入准考证号例如2024PROG001点击“查询”按钮。页面跳转或局部刷新展示该考生的详细信息包括姓名、准考证号、编程题得分、理论题得分、总分、排名。我们的第一个关键任务是找到数据真正的来源。打开浏览器的开发者工具F12切换到“网络”(Network)选项卡清空记录然后进行一次手动查询。观察点击“查询”按钮后浏览器发出了哪些新的网络请求。这里通常有两种情况情况AAPI接口。你会发现一个独立的请求其类型Type可能是XHR或Fetch地址可能像https://score.competition.org/api/query返回的数据是干净的JSON格式如{name:张三,total_score: 95}。这是最理想的情况数据处理起来非常简单。情况B服务端渲染页面。点击查询后浏览器会向一个新地址发起一个document类型的请求例如https://score.competition.org/result?ticket2024PROG001返回一个完整的HTML页面。我们需要的数据就混杂在HTML标签中。对于情况B我们需要分析这个HTML页面的结构。在开发者工具的“元素”(Elements)选项卡中使用检查工具CtrlShiftC点击页面上的分数定位到对应的HTML代码。它可能被包裹在如下的标签中div classscore-card span classlabel总分/span span idtotal_score classvalue95/span /div我们的爬虫需要能精准地定位并提取idtotal_score或classvalue标签内的文本。技术选型思考对于API接口使用requests库直接发起HTTP请求并解析JSON是最佳选择。对于HTML页面则需要一个HTML解析器。BeautifulSoup库因其语法简洁、容错性好而成为首选。如果页面是动态加载的即数据由JavaScript生成初始HTML中没有则可能需要Selenium或Playwright这类浏览器自动化工具来模拟真实用户操作。原则是从最简单、最轻量的方案开始尝试。优先寻找隐藏的API其次用requests BeautifulSoup最后才考虑动用重型武器Selenium。2.2 输入与输出规格设计明确了数据来源接下来定义爬虫的“食谱”和“成品”。输入Input一个包含所有待查询准考证号的文本文件ticket_numbers.txt每行一个号码。这是最通用和易于维护的方式。也可以支持从Excel或CSV读取但核心是程序要能遍历一个列表。2024PROG001 2024PROG002 2024PROG003 ...输出Output一个结构化的数据文件包含所有查询到的字段。最常用的格式是CSV逗号分隔值因为它可以被Excel、Numbers、WPS以及任何数据分析工具直接打开。我们的输出文件results.csv应该包含如下表头和数据姓名,准考证号,编程题得分,理论题得分,总分,排名 张三,2024PROG001,50,45,95,12 李四,2024PROG002,48,40,88,25 ...如果某位考生查询失败如准考证号无效、网络超时我们也应该在结果中明确标记例如在“姓名”栏填入查询失败或Error并在日志中记录具体原因便于后续核对。这个清晰的蓝图确保了我们的开发工作始终聚焦不会在过程中迷失方向。3. 核心引擎构建健壮的请求与解析模块蓝图已定现在开始建造爬虫的核心引擎。这个引擎必须稳定、可靠能够优雅地处理各种异常情况。3.1 使用Requests库处理HTTP请求Requests库是Python中进行HTTP通信的“瑞士军刀”。我们的核心查询功能将围绕它构建。首先根据之前的分析构造请求。如果是API接口import requests def query_by_api(ticket_number): url https://score.competition.org/api/query # 通常API会使用POST方法参数在JSON或表单中 payload {ticket: ticket_number} headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, # 模拟浏览器 Content-Type: application/json } try: response requests.post(url, jsonpayload, headersheaders, timeout10) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 # 假设返回的是JSON data response.json() return data except requests.exceptions.Timeout: print(f查询 {ticket_number} 超时) return None except requests.exceptions.HTTPError as e: print(f查询 {ticket_number} HTTP错误: {e}) return None except requests.exceptions.RequestException as e: print(f查询 {ticket_number} 请求异常: {e}) return None如果是HTML页面def query_by_html(ticket_number): url fhttps://score.competition.org/result?ticket{ticket_number} headers {User-Agent: ...} # 同上 try: response requests.get(url, headersheaders, timeout10) response.raise_for_status() html_content response.text return html_content except requests.exceptions.RequestException as e: print(f获取页面 {ticket_number} 失败: {e}) return None关键点1异常处理。网络请求充满不确定性超时、连接错误、目标服务器返回404或500错误都是家常便饭。使用try...except块并捕获requests.exceptions中的各种异常是保证程序不会因单个请求失败而崩溃的关键。返回None或一个特定的错误标识便于上层逻辑处理。关键点2请求头Headers。设置一个合理的User-Agent是基本礼仪可以避免被一些简单的反爬策略直接拒绝。有时还需要添加Referer、Accept-Language等头信息来模拟得更像真实浏览器。关键点3超时设置。timeout10表示连接和读取的超时时间均为10秒。没有超时设置的请求可能会永远挂起。3.2 使用BeautifulSoup进行HTML解析拿到HTML内容后就需要BeautifulSoup来“淘金”了。from bs4 import BeautifulSoup def parse_html(html_content): if not html_content: return None soup BeautifulSoup(html_content, html.parser) # 根据之前分析的结构定位元素 # 方法1通过ID查找最精确 total_score_elem soup.find(idtotal_score) # 方法2通过CSS选择器查找 name_elem soup.select_one(.score-card .name) # 查找class为score-card下的class为name的元素 # 方法3通过标签和属性组合查找 rank_elem soup.find(span, class_rank-value) # 提取文本并清洗 data {} if total_score_elem: data[total_score] total_score_elem.get_text(stripTrue) # stripTrue去除首尾空白 if name_elem: data[name] name_elem.get_text(stripTrue) # ... 提取其他字段 # 处理可能缺失的字段 if not data: print(警告未能从页面中解析出任何数据请检查页面结构或选择器。) return None return data解析经验谈网页结构并非一成不变。今天能用的选择器明天网站改版可能就失效了。因此选择器的健壮性优先使用id因为它在HTML中应该是唯一的。其次使用具有明确语义的class或>import time import random def query_with_delay(ticket_number): # ... 执行查询和解析 ... time.sleep(random.uniform(1, 3)) # 等待1到3秒之间的一个随机时间 return data将sleep放在整个查询函数完成后可以确保无论本次查询成功与否都会等待一段时间再进行下一次。随机化间隔时间使得爬虫的行为模式更不易被识别。使用会话Session如果查询涉及多个步骤如先登录使用requests.Session()可以自动管理Cookies保持登录状态比单独请求更高效、更接近真实浏览器。session requests.Session() # 首先登录 login_data {username:..., password:...} session.post(login_url, datalogin_data) # 后续查询都使用同一个session response session.get(query_url)代理IPProxies在极端情况下如果IP被封锁可能需要使用代理IP池来轮换请求源。但对于一般的批量查分控制好请求频率通常就足够了。引入代理会增加复杂度和成本如果使用付费代理非必要不添加。注意务必尊重网站的robots.txt协议并确认你的爬取行为不违反网站的服务条款。批量查分用于个人或内部授权的数据整理通常是可接受的但用于大规模抓取公开数据或商业用途则可能涉及法律风险。4. 流程编排让爬虫自动运行与处理数据核心引擎部件已经就位现在需要一条“流水线”把它们组装起来并处理输入和输出。4.1 主控循环读取、查询、保存这是爬虫的“大脑”负责协调整个流程。import csv from typing import List, Dict def main(input_file: str, output_file: str): # 1. 读取输入文件 ticket_numbers [] try: with open(input_file, r, encodingutf-8) as f: ticket_numbers [line.strip() for line in f if line.strip()] except FileNotFoundError: print(f错误输入文件 {input_file} 不存在。) return if not ticket_numbers: print(警告输入文件为空。) return # 2. 准备输出 all_results [] fieldnames [姓名, 准考证号, 编程题得分, 理论题得分, 总分, 排名] # 3. 遍历查询 for idx, ticket in enumerate(ticket_numbers, 1): print(f正在处理第 {idx}/{len(ticket_numbers)} 个: {ticket}) # 调用查询函数 # 这里以HTML查询为例 html query_by_html(ticket) if html is None: # 查询失败记录错误信息 all_results.append({ 姓名: 查询失败, 准考证号: ticket, 编程题得分: , 理论题得分: , 总分: , 排名: }) continue # 继续下一个 data parse_html(html) if data is None: # 解析失败 all_results.append({ 姓名: 解析失败, 准考证号: ticket, 编程题得分: , 理论题得分: , 总分: , 排名: }) continue # 将解析后的数据按字段名整理 result_row { 姓名: data.get(name, ), 准考证号: ticket, 编程题得分: data.get(program_score, ), 理论题得分: data.get(theory_score, ), 总分: data.get(total_score, ), 排名: data.get(rank, ) } all_results.append(result_row) # 可选每处理N条就保存一次防止程序中途崩溃数据全丢 if idx % 10 0: save_intermediate_results(all_results, output_file, fieldnames) # 4. 最终保存 save_results_to_csv(all_results, output_file, fieldnames) print(f处理完成共处理 {len(ticket_numbers)} 个成功 {len([r for r in all_results if r[姓名] not in [查询失败,解析失败]])} 个。结果已保存至 {output_file}) def save_results_to_csv(results: List[Dict], filename: str, fieldnames: List[str]): with open(filename, w, newline, encodingutf-8-sig) as csvfile: # utf-8-sig让Excel正确识别中文 writer csv.DictWriter(csvfile, fieldnamesfieldnames) writer.writeheader() writer.writerows(results)这个主循环清晰地体现了“读取-处理-收集”的流程。enumerate函数提供了当前序号便于进度展示。错误处理被集成在流程中确保单个失败不会影响整体。4.2 数据持久化与中间状态保存将结果保存为CSV是最常见的选择。使用Python内置的csv模块特别是csv.DictWriter可以非常方便地将字典列表写入文件并自动处理字段名表头。一个重要的实践增量保存。在循环中每处理完一定数量比如10个的查询就将当前结果列表all_results写入文件一次。这样即使程序在运行到第150个时因为网络波动或意外而崩溃你之前处理的149个结果也已经保存在文件里了只需修改输入文件移除已成功查询的准考证号从断点重新运行即可避免了前功尽弃。def save_intermediate_results(results, filename, fieldnames): 保存中间结果覆盖原文件 with open(filename, w, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() writer.writerows(results)4.3 日志记录让运行过程有迹可循print语句在简单脚本中够用但对于一个可能运行数小时的批量任务一个正式的日志系统至关重要。它可以帮助你事后分析哪些成功了、哪些失败了、失败的原因是什么。import logging # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(batch_query.log, encodingutf-8), logging.StreamHandler() # 同时在控制台输出 ] ) # 在代码中使用 logging.info(f开始处理准考证号: {ticket}) try: # ... 查询逻辑 ... logging.info(f准考证号 {ticket} 查询成功分数: {score}) except requests.exceptions.Timeout as e: logging.error(f准考证号 {ticket} 请求超时: {e}) except Exception as e: logging.exception(f处理准考证号 {ticket} 时发生未知错误) # 会记录完整的异常堆栈日志文件batch_query.log会成为你排查问题的第一手资料。5. 实战部署与进阶优化一个在本地IDE里能跑的脚本和一个能在各种环境下稳定运行的“工具”还有一段距离。我们需要考虑它的易用性和鲁棒性。5.1 封装为命令行工具让用户通过命令行参数来指定输入输出文件比直接修改脚本里的硬编码路径要友好得多。可以使用Python标准库argparse。import argparse def setup_argparse(): parser argparse.ArgumentParser(description批量查分爬虫) parser.add_argument(-i, --input, requiredTrue, help输入文件路径包含准考证号每行一个) parser.add_argument(-o, --output, defaultresults.csv, help输出CSV文件路径 (默认: results.csv)) parser.add_argument(-d, --delay, typefloat, default2.0, help请求间延迟秒数 (默认: 2)) parser.add_argument(--mode, choices[api, html], defaulthtml, help查询模式api或html (默认: html)) return parser.parse_args() if __name__ __main__: args setup_argparse() print(f启动参数: 输入{args.input}, 输出{args.output}, 延迟{args.delay}秒, 模式{args.mode}) # 将args传递给主函数 main(args.input, args.output, args.delay, args.mode)这样用户就可以在终端中运行python score_spider.py -i tickets.txt -o final_scores.csv -d 1.5。5.2 错误重试机制网络请求偶尔失败是正常的。增加一个简单的重试逻辑可以大幅提高最终的成功率。from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def robust_query(ticket_number): 带有重试机制的查询函数 return query_by_html(ticket_number) # 或者 query_by_api这里使用了tenacity库它提供了非常优雅的重试装饰器。上面的配置表示最多重试3次重试等待时间呈指数增长第一次等4秒第二次等8秒...最多10秒。如果3次都失败则抛出最后一次的异常。你可以将这个装饰器应用到你的核心查询函数上。5.3 并发查询以提升速度高级当需要查询的名单有成千上万条时即使每次延迟2秒总时间也会非常长。此时可以考虑并发请求。但必须极度谨慎因为过高的并发请求会被视为攻击导致IP被永久封禁。一个折中的方案是使用有限并发例如同时只发起3-5个请求。可以使用concurrent.futures库的ThreadPoolExecutor。from concurrent.futures import ThreadPoolExecutor, as_completed def batch_query_concurrent(ticket_list, max_workers3): 使用线程池进行并发查询 results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务 future_to_ticket {executor.submit(robust_query, ticket): ticket for ticket in ticket_list} for future in as_completed(future_to_ticket): ticket future_to_ticket[future] try: data future.result(timeout15) # 每个future单独设置超时 if data: results.append(process_data(data, ticket)) logging.info(f{ticket} 完成) else: logging.warning(f{ticket} 查询返回空) results.append(create_error_row(ticket, 查询无结果)) except Exception as exc: logging.error(f{ticket} 生成异常: {exc}) results.append(create_error_row(ticket, str(exc))) return results重要警告使用并发时务必确保目标服务器能够承受这样的压力并且你明确了解其服务条款。将max_workers设置为一个很小的数字如3并确保总的请求频率max_workers / 平均单次请求耗时远低于手动操作的速度。同时并发下的错误处理和日志记录会更复杂。6. 避坑指南那些我踩过的“坑”与应对策略纸上得来终觉浅绝知此事要躬行。下面分享几个在开发和使用此类爬虫过程中容易遇到的问题和解决方案。6.1 网页结构变动导致解析失败这是最常见的问题。今天还能跑的脚本明天可能就因为网站前端改版而全军覆没。应对策略选择器降级尽量使用更稳定、更不易变的选择器。优先级idname属性 具有特定语义的class如user-score 标签层级关系。避免使用与样式强相关的class如col-md-6。多层解析与模糊匹配如果直接定位目标元素失败可以尝试先定位其父级容器通常结构更稳定再在容器内搜索。card soup.find(div, class_result-card) # 假设这个卡片容器稳定 if card: name card.find(span, class_name) # 在容器内找 score card.find(stringre.compile(r总分\s*(\d))) # 使用正则模糊匹配文本建立监控与告警对于需要长期运行的爬虫可以添加一个“健康检查”机制。例如在每次运行时先用一个已知的、有效的测试准考证号跑一遍流程检查是否能成功解析出预期格式的数据。如果失败则立即通过邮件或消息通知负责人而不是等到处理完几万条数据后才发现。6.2 遭遇验证码或动态加载有些网站在频繁查询或检测到非浏览器行为时会弹出验证码或者关键数据是通过JavaScript异步加载的初始HTML中没有。应对策略降低请求频率这是预防触发验证码最有效的方法。将延迟时间调大并加入随机性。切换至浏览器自动化工具当requestsBeautifulSoup组合失效时Selenium或Playwright是终极方案。它们可以控制真实的浏览器如Chrome、Firefox加载页面执行JavaScript甚至手动处理验证码虽然全自动破解验证码通常不道德且可能违法。使用它们代价是速度极慢、资源消耗大仅作为最后手段。from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC driver webdriver.Chrome() driver.get(query_url) # 等待特定元素加载出来 score_element WebDriverWait(driver, 10).until( EC.presence_of_element_located((By.ID, total_score)) ) score score_element.text driver.quit()寻找隐藏的移动端API有时网站的移动端页面或APP使用的API接口验证更简单。可以尝试抓包分析手机APP的流量或许能找到更友好的数据接口。6.3 数据格式不一致与清洗解析出来的文本数据可能是“95分”、“总分95”、“95.0”、“九十五”需要统一清洗为数字。应对策略编写健壮的清洗函数def clean_score(score_text): if not score_text: return None # 移除中文、空格等非数字字符保留小数点 import re numbers re.findall(r\d\.?\d*, score_text) if numbers: try: return float(numbers[0]) except ValueError: return None return None保存原始数据在最终输出的CSV中除了清洗后的规范数据列可以额外增加一列“原始数据”或“备注”用于保存从网页中直接抓取到的原始文本。这在后续数据核对和问题排查时非常有用。6.4 网络环境与依赖管理你的脚本在你自己电脑上运行良好但同事或另一台服务器上却报错可能是缺少库或网络代理问题。应对策略使用requirements.txt在项目根目录创建此文件列出所有依赖库及其版本。requests2.31.0 beautifulsoup44.12.2 pandas2.1.4他人可以通过pip install -r requirements.txt一键安装所有依赖。考虑网络代理如果运行脚本的环境需要通过公司代理访问外网需要在代码中或系统环境变量里配置代理。requests库支持通过proxies参数设置。proxies { http: http://your-proxy:port, https: http://your-proxy:port, } response requests.get(url, proxiesproxies)容器化部署对于更复杂的部署场景可以考虑使用Docker。将爬虫脚本及其运行环境打包成一个镜像可以在任何安装了Docker的机器上以完全相同的方式运行彻底解决“在我机器上好好的”这类问题。构建一个可靠的批量查分爬虫技术实现只是其中一环更重要的是对目标系统的分析、对异常情况的预见、以及对效率与风险之间的平衡把握。它更像是一个细心、耐心、有礼貌的数字化助手其价值不在于用了多炫酷的技术而在于真正可靠地完成了那份枯燥却必要的工作。从手动复制粘贴到自动化脚本这不仅仅是效率的提升更是工作思维的转变。当你下次再面对重复性数据任务时第一反应不再是“动手”而是“能不能写个脚本”那么你就已经掌握了这个时代最重要的技能之一。