ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

首页/资讯中心/资讯详情

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

发布时间：2026/8/4 9:16:48 拓冰企业网站定制
Python爬虫实战：从中国地震台网中心获取历史地震数据

1. 项目缘起：从数据需求到技术选型

最近在做一个关于区域地质风险评估的业余项目，需要用到历史地震数据。一开始，我理所当然地去找了各大科研机构的公开数据集，但要么数据格式不友好，要么时间跨度不够，要么就是更新频率太低。后来把目光转向了国内最权威、更新最及时的数据源之一——中国地震台网中心（CENC）的官方数据发布平台。这个平台提供了准实时的地震速报和历史地震目录，数据质量高，对于分析地震活动性、评估区域风险来说，是极佳的一手资料。

然而，问题来了。这些数据虽然公开，但并没有提供一个可以直接下载完整历史数据的、格式统一的API或数据包。数据通常以网页列表或详情页的形式呈现，手动复制粘贴对于大量数据来说无异于天方夜谭。这时，“爬取”就成了一个自然而然的技术选项。所谓爬取，就是编写程序模拟浏览器访问，自动地从这些网页中提取、解析并结构化我们所需的地震信息，比如发震时刻、经纬度、深度、震级、参考地点等。

这个需求非常典型，也是很多数据分析师、科研工作者或地理信息爱好者会遇到的实际问题。它不涉及复杂的反爬机制（在合理、合法、遵守robots.txt的前提下），但完整走通整个流程，却能串联起网络请求、HTML解析、数据清洗、持久化存储乃至简单的调度任务等多个实用技能点。接下来，我就把自己实现这个数据采集过程的思路、代码、踩过的坑以及一些扩展思考，详细地分享出来。目标很明确：让你看完后，能独立写出一套稳定、可靠的地震数据爬虫。

2. 目标网站分析与请求策略制定

动手写代码之前，细致的侦察工作必不可少。我们的目标是中国地震台网中心的官方地震查询页面。为了避免具体网址变动，我们将其核心特征抽象出来进行分析。

首先，打开地震目录查询页面。你会发现，数据通常以分页列表的形式展示，每页显示一定数量的地震事件。我们需要解决两个核心问题：如何获取列表页，以及如何从列表页中提取详情页链接或直接提取数据。

2.1 列表页请求分析

通过浏览器的开发者工具（F12），切换到Network（网络）选项卡，然后进行翻页操作。观察翻页时浏览器发送了哪些HTTP请求。关键点在于：

  1. 请求类型：是GET还是POST？地震数据查询通常是GET请求，参数直接体现在URL中。
  2. 请求参数：URL中通常包含查询条件，例如：
    • starttime：查询开始时间
    • endtime：查询结束时间
    • latitude,longitude,radius：地理范围限定
    • minmagnitude,maxmagnitude：震级范围
    • page或pageNum：页码参数
    • size或pageSize：每页条数

例如，一个典型的列表页URL可能长这样：http://www.xxx.cn/servlet/query?starttime=2024-01-01&endtime=2024-01-31&page=1&size=20

这里有一个非常重要的实践细节：很多网站的分页并不是简单的page参数递增。有些网站采用limit和offset参数（limit=20&offset=0， 第二页就是offset=20），有些则使用lastId（上一页最后一条数据的ID）作为游标。必须通过实际观察确定其分页机制。

2.2 数据加载方式分析

接下来，要看数据是服务端渲染（SSR）还是客户端动态加载。

  • 服务端渲染：刷新页面或提交查询后，返回的HTML源代码中直接包含了表格数据。你可以在翻页后，查看页面源代码（Ctrl+U），如果能直接搜索到地震震级、地点等文本，就是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, */*; q=0.01', # 如果请求JSON API 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.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），尊重网站的服务器负载。这意味着：

  1. 添加延迟：在请求之间使用time.sleep()，例如time.sleep(1)表示间隔1秒。对于历史数据，可以间隔更长，如2-3秒。避免高频请求对服务器造成压力。
  2. 明确数据用途：确保你的数据采集行为仅用于个人学习、研究或非商业用途。切勿将大量爬取的数据用于商业盈利或对网站进行恶意攻击。
  3. 识别并遵守API限制：如果网站提供了公开API，优先使用API。如果没有，你的爬取行为应尽可能模拟人类浏览器的行为。

3. 核心爬取流程与代码实现

假设我们经过分析，确定目标网站的数据是通过一个GET请求的JSON API返回的，分页参数是page和size。下面我们来构建完整的爬虫。

3.1 环境准备与依赖安装

我们需要requests库来发送HTTP请求，pandas库来处理和保存数据。如果数据在HTML中，还需要BeautifulSoup4。

pip install requests pandas beautifulsoup4

3.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, params=params, headers=headers, timeout=10) 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 None

3.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_page

3.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, params=params, headers=headers) 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(f"API返回错误状态: {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_index=True) 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_url=BASE_API_URL, query_params=QUERY_PARAMS, headers=HEADERS, start_page=1, max_pages=10 ) # 查看数据 if not earthquake_df.empty: print(earthquake_df.head()) print(earthquake_df.info()) # 保存数据到CSV文件 output_file = f"earthquake_data_{QUERY_PARAMS['starttime'][:10]}_to_{QUERY_PARAMS['endtime'][:10]}.csv" earthquake_df.to_csv(output_file, index=False, encoding='utf-8-sig') print(f"数据已保存至: {output_file}")

4. 数据清洗、存储与后续处理

爬取到的原始数据往往不能直接使用，需要进行清洗和格式化。

4.1 常见的数据清洗任务

  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', errors='coerce') # 如果原始时间包含时区，如 '2024-01-15T08:30:25+08:00' # earthquake_df['time_parsed'] = pd.to_datetime(earthquake_df['time_utc'], utc=True)

  2. 处理缺失值与异常值：检查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)]

  3. 坐标验证：确保经纬度在合理范围内（纬度[-90, 90]，经度[-180, 180]）。

  4. 去重：基于唯一标识（如event_id）或时间、位置、震级组合进行去重。

    earthquake_df_cleaned = earthquake_df_cleaned.drop_duplicates(subset=['event_id'], keep='first') # 如果没有ID，可以用时空震级复合去重（精度要求高） # earthquake_df_cleaned = earthquake_df_cleaned.drop_duplicates(subset=['time_parsed', 'latitude', 'longitude', 'magnitude'], keep='first')

4.2 数据存储方案

除了保存为CSV，根据数据量和使用场景，可以考虑其他存储方式：

  • SQLite数据库：适合中小型数据集，无需安装数据库服务器，便于管理和查询。
    import sqlite3 conn = sqlite3.connect('earthquake.db') earthquake_df_cleaned.to_sql('earthquakes', conn, if_exists='replace', index=False) conn.close()
  • PostgreSQL/MySQL with PostGIS：如果数据量巨大（数十万条以上），且需要进行复杂的地理空间查询（如“查找某点100公里内的所有地震”），专业的空间数据库是更好的选择。
  • Parquet/Feather格式：比CSV读写更快，更节省磁盘空间，适合作为数据分析的中间格式。

4.3 数据更新与增量爬取

对于需要持续更新的场景（如每天爬取最新地震），增量爬取是关键。策略如下：

  1. 基于最新时间：每次爬取时，starttime设置为上次爬取到的最大时间，endtime设置为当前时间。注意处理时间边界重叠问题。
  2. 基于事件ID：如果API返回的事件ID是递增的，可以记录上次爬取的最大ID，下次请求时添加min_id参数。
  3. 状态记录：将每次爬取的元信息（如最后爬取时间、最后事件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 数据时间时区混乱

  • 现象：爬取到的北京时间（UTC+8）和UTC时间混在一起，导致时间序列分析出错。
  • 解决：
    • 统一时区：在数据清洗阶段，将所有时间字段明确转换为datetime对象，并统一到一个时区（推荐UTC）。Pandas的pd.to_datetime(..., utc=True)可以很好地处理带时区信息的字符串。
    • 标注时区：在数据表中新增一列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( level=logging.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提供了项目结构、异步请求、中间件、管道等一套完整机制，但学习曲线比requests+BeautifulSoup要陡峭一些。对于中国地震台网这种相对简单的API爬取，我们目前的脚本已经足够高效和灵活。

6.4 数据质量监控

定期运行爬虫时，可以加入一些简单的数据质量检查：

  • 数据量检查：本次爬取的数据条数是否在历史合理范围内？（例如，某天突然为零条，可能是爬虫失效或网站改版）。
  • 字段完整性检查：关键字段（震级、经纬度）的缺失率是否异常升高？
  • 数值范围检查：震级、深度是否出现了历史极值外的异常数据？

可以将这些检查结果也记录到日志中，甚至设置报警（如通过邮件），以便及时发现问题。

整个流程走下来，你会发现爬取公开数据本身的技术难度可能并不高，但构建一个稳健、可维护、有礼貌的数据采集系统，却需要考虑到很多细节。从分析网站结构开始，到写出第一行代码，再到处理各种边界情况和异常，最后思考如何工程化，每一步都是对实际问题解决能力的锻炼。希望这份详细的指南能帮你避开我踩过的那些坑，更顺畅地获取到你所需的地震数据，为后续的分析工作打下坚实的基础。

网站建设 企业官网 运营推广
返回列表