Python爬虫实战：从中国地震台网中心获取历史地震数据

1. 项目缘起从数据需求到技术选型最近在做一个关于区域地质风险评估的业余项目需要用到历史地震数据。一开始我理所当然地去找了各大科研机构的公开数据集但要么数据格式不友好要么时间跨度不够要么就是更新频率太低。后来把目光转向了国内最权威、更新最及时的数据源之一——中国地震台网中心CENC的官方数据发布平台。这个平台提供了准实时的地震速报和历史地震目录数据质量高对于分析地震活动性、评估区域风险来说是极佳的一手资料。然而问题来了。这些数据虽然公开但并没有提供一个可以直接下载完整历史数据的、格式统一的API或数据包。数据通常以网页列表或详情页的形式呈现手动复制粘贴对于大量数据来说无异于天方夜谭。这时“爬取”就成了一个自然而然的技术选项。所谓爬取就是编写程序模拟浏览器访问自动地从这些网页中提取、解析并结构化我们所需的地震信息比如发震时刻、经纬度、深度、震级、参考地点等。这个需求非常典型也是很多数据分析师、科研工作者或地理信息爱好者会遇到的实际问题。它不涉及复杂的反爬机制在合理、合法、遵守robots.txt的前提下但完整走通整个流程却能串联起网络请求、HTML解析、数据清洗、持久化存储乃至简单的调度任务等多个实用技能点。接下来我就把自己实现这个数据采集过程的思路、代码、踩过的坑以及一些扩展思考详细地分享出来。目标很明确让你看完后能独立写出一套稳定、可靠的地震数据爬虫。2. 目标网站分析与请求策略制定动手写代码之前细致的侦察工作必不可少。我们的目标是中国地震台网中心的官方地震查询页面。为了避免具体网址变动我们将其核心特征抽象出来进行分析。首先打开地震目录查询页面。你会发现数据通常以分页列表的形式展示每页显示一定数量的地震事件。我们需要解决两个核心问题如何获取列表页以及如何从列表页中提取详情页链接或直接提取数据。2.1 列表页请求分析通过浏览器的开发者工具F12切换到Network网络选项卡然后进行翻页操作。观察翻页时浏览器发送了哪些HTTP请求。关键点在于请求类型是GET还是POST地震数据查询通常是GET请求参数直接体现在URL中。请求参数URL中通常包含查询条件例如starttime查询开始时间endtime查询结束时间latitude,longitude,radius地理范围限定minmagnitude,maxmagnitude震级范围page或pageNum页码参数size或pageSize每页条数例如一个典型的列表页URL可能长这样http://www.xxx.cn/servlet/query?starttime2024-01-01endtime2024-01-31page1size20这里有一个非常重要的实践细节很多网站的分页并不是简单的page参数递增。有些网站采用limit和offset参数limit20offset0 第二页就是offset20有些则使用lastId上一页最后一条数据的ID作为游标。必须通过实际观察确定其分页机制。2.2 数据加载方式分析接下来要看数据是服务端渲染SSR还是客户端动态加载。服务端渲染刷新页面或提交查询后返回的HTML源代码中直接包含了表格数据。你可以在翻页后查看页面源代码CtrlU如果能直接搜索到地震震级、地点等文本就是SSR。这是最简单的情况直接用requests库获取HTML再用BeautifulSoup或lxml解析即可。客户端动态加载Ajax翻页时页面主体没有刷新但数据更新了。在Network选项卡中你会看到一个新的XHR或Fetch请求其响应通常是JSON格式。这种情况更友好因为数据已经是结构化的JSON直接解析这个JSON响应即可无需解析HTML。你需要找到这个API接口的URL和参数。以我的经验许多数据查询平台为了前后端分离和用户体验会采用Ajax加载数据。因此优先在Network中筛选XHR/Fetch请求寻找包含地震列表数据的JSON响应。2.3 请求头与反爬策略考量即使是一个相对开放的公开数据网站添加一些基本的请求头也是良好实践和规避最低限度反爬的措施。import requests headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Accept: application/json, text/javascript, */*; q0.01, # 如果请求JSON API Accept-Language: zh-CN,zh;q0.9,en;q0.8, Referer: http://www.xxx.cn/query.html, # 填写查询页面的URL }User-Agent模拟真实浏览器这是最基本的。Accept告诉服务器客户端希望接收的数据类型。如果是Ajax请求通常期望JSON。Referer表示请求的来源页面有些网站会校验。关于Cookie/Session对于简单的公开数据查询通常不需要登录因此一般不需要处理复杂的会话。但如果查询需要先通过一个表单提交可能会设置一个简单的会话Cookie这时使用requests.Session()对象可以自动保持Cookie。2.4 频率限制与道德规范在编写爬虫时必须遵守robots.txt协议通常访问网站域名/robots.txt尊重网站的服务器负载。这意味着添加延迟在请求之间使用time.sleep()例如time.sleep(1)表示间隔1秒。对于历史数据可以间隔更长如2-3秒。避免高频请求对服务器造成压力。明确数据用途确保你的数据采集行为仅用于个人学习、研究或非商业用途。切勿将大量爬取的数据用于商业盈利或对网站进行恶意攻击。识别并遵守API限制如果网站提供了公开API优先使用API。如果没有你的爬取行为应尽可能模拟人类浏览器的行为。3. 核心爬取流程与代码实现假设我们经过分析确定目标网站的数据是通过一个GET请求的JSON API返回的分页参数是page和size。下面我们来构建完整的爬虫。3.1 环境准备与依赖安装我们需要requests库来发送HTTP请求pandas库来处理和保存数据。如果数据在HTML中还需要BeautifulSoup4。pip install requests pandas beautifulsoup43.2 构建基础请求函数首先写一个健壮的请求函数包含错误处理和重试机制。import requests import time import pandas as pd from typing import Optional, Dict, Any def make_request(url: str, params: Optional[Dict] None, headers: Optional[Dict] None, max_retries: int 3) - Optional[requests.Response]: 发送HTTP请求包含重试机制。 Args: url: 请求的URL params: 查询参数字典 headers: 请求头字典 max_retries: 最大重试次数 Returns: requests.Response对象如果失败则返回None retries 0 while retries max_retries: try: resp requests.get(url, paramsparams, headersheaders, timeout10) resp.raise_for_status() # 如果状态码不是200抛出HTTPError异常 # 简单检查返回内容是否为有效JSON如果是API if application/json in resp.headers.get(Content-Type, ): resp.json() # 尝试解析如果无效JSON会抛出异常 return resp except requests.exceptions.RequestException as e: print(f请求失败 (尝试 {retries 1}/{max_retries}): {e}) retries 1 if retries max_retries: wait_time 2 ** retries # 指数退避策略 print(f等待 {wait_time} 秒后重试...) time.sleep(wait_time) else: print(f已达到最大重试次数放弃请求: {url}) return None except ValueError as e: # 捕获JSON解析错误 print(f响应内容不是有效的JSON: {e}) # 可以考虑返回resp但标记内容异常这里直接返回None return None3.3 解析单页数据并封装成函数假设API返回的JSON结构如下{ success: true, data: { list: [ { id: 12345, time: 2024-01-15 08:30:25, latitude: 39.91, longitude: 116.40, depth: 10, magnitude: 3.2, location: 北京海淀区 }, // ... 更多地震事件 ], total: 150, page: 1, size: 20 } }我们编写解析单页数据的函数def parse_single_page(json_data: Dict[str, Any]) - pd.DataFrame: 解析单页JSON数据返回一个pandas DataFrame。 Args: json_data: 从API响应中解析出的字典 Returns: 包含本页地震数据的DataFrame earthquakes [] # 根据实际JSON结构定位数据列表 # 这里假设数据在 data[list] 路径下 event_list json_data.get(data, {}).get(list, []) if not event_list: print(当前页无数据。) return pd.DataFrame() for event in event_list: earthquake { event_id: event.get(id), time_utc: event.get(time), # 注意时间格式和时区 latitude: event.get(latitude), longitude: event.get(longitude), depth_km: event.get(depth), magnitude: event.get(magnitude), location: event.get(location), # 可以根据需要添加更多字段如震源类型、数据来源等 } earthquakes.append(earthquake) df_page pd.DataFrame(earthquakes) return df_page3.4 实现分页循环与主爬取逻辑现在将分页逻辑整合起来。我们需要计算总页数并循环请求每一页。def crawl_earthquake_data(base_url: str, query_params: Dict[str, Any], headers: Dict[str, str], start_page: int 1, max_pages: int None) - pd.DataFrame: 主爬取函数负责分页请求和数据拼接。 Args: base_url: API的基础URL不包含分页参数 query_params: 固定的查询参数如时间、震级范围 headers: 请求头 start_page: 起始页码 max_pages: 最大爬取页数为None则爬取所有页 Returns: 包含所有爬取数据的DataFrame all_data [] current_page start_page total_pages_estimated None while True: # 1. 构造当前页的请求参数 params query_params.copy() params[page] current_page # params[size] 20 # 如果每页条数可配置 print(f正在爬取第 {current_page} 页...) # 2. 发送请求 resp make_request(base_url, paramsparams, headersheaders) if resp is None: print(f第 {current_page} 页请求失败跳过。) break # 3. 解析JSON try: json_data resp.json() except ValueError as e: print(f第 {current_page} 页响应JSON解析失败: {e}) break # 4. 检查API返回状态根据实际结构调整 if not json_data.get(success, True): print(fAPI返回错误状态: {json_data}) break # 5. 解析数据 df_page parse_single_page(json_data) if df_page.empty: print(f第 {current_page} 页无数据可能已到末页。) break # 6. 保存本页数据 all_data.append(df_page) print(f第 {current_page} 页爬取成功获得 {len(df_page)} 条记录。) # 7. 判断是否继续翻页 # 方法A根据返回的数据条数判断。如果返回条数小于每页大小可能是最后一页。 page_size params.get(size, 20) # 假设每页20条 if len(df_page) page_size: print(f当前页数据不足 {page_size} 条视为最后一页。) break # 方法B根据API返回的总页数或总条数判断更可靠 # total json_data.get(data, {}).get(total, 0) # total_pages (total page_size - 1) // page_size # 向上取整 # if current_page total_pages: # break # 8. 更新页码准备下一次请求 current_page 1 # 9. 检查最大页数限制 if max_pages is not None and current_page start_page max_pages - 1: print(f已达到设定的最大爬取页数 {max_pages}。) break # 10. 礼貌性延迟避免请求过快 time.sleep(1.5) # 合并所有数据 if all_data: final_df pd.concat(all_data, ignore_indexTrue) print(f所有数据爬取完成总计 {len(final_df)} 条记录。) return final_df else: print(未爬取到任何数据。) return pd.DataFrame()3.5 配置参数并执行爬取最后我们配置具体的参数并运行爬虫。if __name__ __main__: # 配置请求参数以下为示例需根据实际网站调整 BASE_API_URL http://www.xxx.cn/api/earthquake/list # 替换为真实API地址 HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ..., Accept: application/json, Referer: http://www.xxx.cn/query, } # 查询条件例如爬取2024年1月的数据 QUERY_PARAMS { starttime: 2024-01-01 00:00:00, endtime: 2024-01-31 23:59:59, minmagnitude: 1.0, # 最小震级 # maxmagnitude: 9.0, # latitude: 40.0, # longitude: 116.0, # radius: 500, size: 50, # 每页50条如果API支持 } # 开始爬取最多爬10页防止意外无限循环 earthquake_df crawl_earthquake_data( base_urlBASE_API_URL, query_paramsQUERY_PARAMS, headersHEADERS, start_page1, max_pages10 ) # 查看数据 if not earthquake_df.empty: print(earthquake_df.head()) print(earthquake_df.info()) # 保存数据到CSV文件 output_file fearthquake_data_{QUERY_PARAMS[starttime][:10]}_to_{QUERY_PARAMS[endtime][:10]}.csv earthquake_df.to_csv(output_file, indexFalse, encodingutf-8-sig) print(f数据已保存至: {output_file})4. 数据清洗、存储与后续处理爬取到的原始数据往往不能直接使用需要进行清洗和格式化。4.1 常见的数据清洗任务时间格式标准化API返回的时间字符串可能包含时区信息如08:00我们需要将其转换为Python的datetime对象并统一为UTC时间或本地时间便于后续时间序列分析。import pandas as pd # 假设原始时间列名为 time_utc格式为 2024-01-15 08:30:25 earthquake_df[time_parsed] pd.to_datetime(earthquake_df[time_utc], format%Y-%m-%d %H:%M:%S, errorscoerce) # 如果原始时间包含时区如 2024-01-15T08:30:2508:00 # earthquake_df[time_parsed] pd.to_datetime(earthquake_df[time_utc], utcTrue)处理缺失值与异常值检查depth_km深度、magnitude震级等数值字段是否存在NaN或明显不合理的值如深度为0或负数震级超过合理范围。# 检查缺失值 print(earthquake_df.isnull().sum()) # 处理缺失值可以删除或用中位数/均值填充需谨慎对于震级一般不填充 earthquake_df_cleaned earthquake_df.dropna(subset[magnitude, latitude, longitude]) # 检查异常值 print(earthquake_df_cleaned[magnitude].describe()) # 假设震级合理范围是 0.0 ~ 9.5 earthquake_df_cleaned earthquake_df_cleaned[(earthquake_df_cleaned[magnitude] 0.0) (earthquake_df_cleaned[magnitude] 9.5)]坐标验证确保经纬度在合理范围内纬度[-90, 90]经度[-180, 180]。去重基于唯一标识如event_id或时间、位置、震级组合进行去重。earthquake_df_cleaned earthquake_df_cleaned.drop_duplicates(subset[event_id], keepfirst) # 如果没有ID可以用时空震级复合去重精度要求高 # earthquake_df_cleaned earthquake_df_cleaned.drop_duplicates(subset[time_parsed, latitude, longitude, magnitude], keepfirst)4.2 数据存储方案除了保存为CSV根据数据量和使用场景可以考虑其他存储方式SQLite数据库适合中小型数据集无需安装数据库服务器便于管理和查询。import sqlite3 conn sqlite3.connect(earthquake.db) earthquake_df_cleaned.to_sql(earthquakes, conn, if_existsreplace, indexFalse) conn.close()PostgreSQL/MySQL with PostGIS如果数据量巨大数十万条以上且需要进行复杂的地理空间查询如“查找某点100公里内的所有地震”专业的空间数据库是更好的选择。Parquet/Feather格式比CSV读写更快更节省磁盘空间适合作为数据分析的中间格式。4.3 数据更新与增量爬取对于需要持续更新的场景如每天爬取最新地震增量爬取是关键。策略如下基于最新时间每次爬取时starttime设置为上次爬取到的最大时间endtime设置为当前时间。注意处理时间边界重叠问题。基于事件ID如果API返回的事件ID是递增的可以记录上次爬取的最大ID下次请求时添加min_id参数。状态记录将每次爬取的元信息如最后爬取时间、最后事件ID、成功状态记录在一个小文件或数据库表中便于下次启动时读取。5. 实战中遇到的坑与解决方案在实际操作中不可能一帆风顺。下面分享几个我踩过的坑和解决办法。5.1 请求被阻断或返回非预期内容现象程序运行一段时间后返回的不再是JSON数据而是验证码页面、错误提示或空数据。可能原因与解决请求频率过高这是最常见的原因。立即增加请求间隔time.sleep()并考虑使用随机延迟如time.sleep(random.uniform(2, 5))来模拟更自然的人类行为。请求头不完整检查是否缺少必要的Headers如Accept、Accept-Language、Referer甚至Cookie。用浏览器正常访问一次复制完整的请求头信息。IP被暂时限制如果使用了代理池切换IP。如果是个人IP只能等待一段时间如半小时或几小时再试。务必遵守爬虫道德不要试图绕过严重的反爬措施。API参数格式或值错误仔细核对每个参数的名字和格式。例如时间格式可能是YYYY-MM-DD也可能是YYYYMMDD或时间戳。震级参数名可能是mag而不是magnitude。5.2 数据解析错误或字段缺失现象JSON解析失败或者解析出的字典中找不到预期的字段。解决加强异常处理在parse_single_page函数中对每个字段的获取使用.get()方法并提供默认值如.get(magnitude, None)避免因单个事件字段缺失导致整个解析失败。打印调试在解析失败时将原始的JSON响应片段或整个结构打印出来仔细检查其实际层级和字段名。网站结构可能发生微调。编写适配器如果字段名变了例如从mag变成了magnitude可以写一个字段映射字典来兼容。5.3 分页逻辑失效陷入死循环现象爬虫一直请求下一页但返回的数据可能是重复的或始终是最后一页的数据。解决采用更可靠的分页终止条件优先使用API返回的total总条数和pageSize每页大小来计算总页数以此作为循环终止条件。代码中已给出示例。记录已爬取的ID在内存或文件中记录已处理的事件ID如果新一页的数据ID都已存在则终止爬取。设置绝对最大页数像示例代码中的max_pages参数作为一个安全阀防止因逻辑错误导致的无限请求。5.4 数据时间时区混乱现象爬取到的北京时间UTC8和UTC时间混在一起导致时间序列分析出错。解决统一时区在数据清洗阶段将所有时间字段明确转换为datetime对象并统一到一个时区推荐UTC。Pandas的pd.to_datetime(..., utcTrue)可以很好地处理带时区信息的字符串。标注时区在数据表中新增一列timezone记录原始数据的时区信息。6. 进阶思考从爬虫到数据管道一个健壮的数据采集任务不应只是一个脚本而应该是一个可维护、可监控的管道。6.1 模块化与配置化将爬虫代码拆分成独立模块config.py存放URL、请求头、查询参数、数据库连接字符串等配置。requester.py封装make_request等网络请求相关函数。parser.py封装parse_single_page等数据解析函数。storage.py封装数据保存到CSV、数据库的函数。scheduler.py负责调度和运行主流程处理异常和日志。这样使得代码更清晰也便于单元测试。6.2 添加日志记录使用Python内置的logging模块替代print语句可以输出不同级别DEBUG, INFO, WARNING, ERROR的日志到文件和控制台方便事后排查问题。import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(earthquake_crawler.log), logging.StreamHandler() ] ) logger logging.getLogger(__name__) # 使用时 logger.info(f正在爬取第 {current_page} 页...) logger.error(f请求失败: {e})6.3 考虑使用Scrapy框架如果项目复杂度增加例如需要爬取多个不同结构的数据源、处理更复杂的反爬、或需要极高的并发性能那么使用专业的爬虫框架如Scrapy是更好的选择。Scrapy提供了项目结构、异步请求、中间件、管道等一套完整机制但学习曲线比requestsBeautifulSoup要陡峭一些。对于中国地震台网这种相对简单的API爬取我们目前的脚本已经足够高效和灵活。6.4 数据质量监控定期运行爬虫时可以加入一些简单的数据质量检查数据量检查本次爬取的数据条数是否在历史合理范围内例如某天突然为零条可能是爬虫失效或网站改版。字段完整性检查关键字段震级、经纬度的缺失率是否异常升高数值范围检查震级、深度是否出现了历史极值外的异常数据可以将这些检查结果也记录到日志中甚至设置报警如通过邮件以便及时发现问题。整个流程走下来你会发现爬取公开数据本身的技术难度可能并不高但构建一个稳健、可维护、有礼貌的数据采集系统却需要考虑到很多细节。从分析网站结构开始到写出第一行代码再到处理各种边界情况和异常最后思考如何工程化每一步都是对实际问题解决能力的锻炼。希望这份详细的指南能帮你避开我踩过的那些坑更顺畅地获取到你所需的地震数据为后续的分析工作打下坚实的基础。