Python 实战：动态渔业资源分布图采集——基于 OBIS 时间轴图层的 API 采集、清洗、聚合与导出

㊗️本期内容已收录至专栏《Python爬虫实战》，持续完善知识体系与项目实战，建议先订阅收藏，后续查阅更方便～ ㊙️本期爬虫难度指数：⭐⭐⭐⭐☆（高级） 🉐福利： 一次订阅后，专栏内的所有文章可永久免费看，持续更新中，保底1000+(篇)硬核实战内容。

全文目录：

• • 🌟 开篇语
  • 0️⃣ 前言（Preface）
  • 1️⃣ 摘要（Abstract）
  • 2️⃣ 背景与需求（Why）
  • • 2.1 为什么要采集动态渔业资源分布图
    • 2.2 本文目标站点与采集对象
    • 2.3 目标字段清单
  • 3️⃣ 合规与注意事项（必写）
  • • 3.1 robots.txt 基本说明
    • 3.2 频率控制
    • 3.3 不采集敏感信息
    • 3.4 数据解释要谨慎
  • 4️⃣ 技术选型与整体流程（What/How）
  • • 4.1 静态、动态、API 三类采集方式
    • 4.2 动态图层采集的核心思路
    • 4.3 整体流程
    • 4.4 为什么选 requests，而不是一上来就 Playwright
  • 5️⃣ 环境准备与依赖安装（可复现）
  • • 5.1 Python 版本
    • 5.2 创建虚拟环境
    • 5.3 安装依赖
    • 5.4 推荐项目结构
    • 5.5 配置文件
  • 6️⃣ 核心实现：请求层（Fetcher）
  • • 6.1 headers 设计
    • 6.2 timeout
    • 6.3 session / cookie
    • 6.4 失败处理：重试与退避
    • 6.5 代码：`config.py`
    • 6.6 代码：`fetcher.py`
  • 7️⃣ 核心实现：解析层（Parser）
  • • 7.1 解析方式
    • 7.2 列表页如何拿详情链接
    • 7.3 详情页如何抽字段
    • 7.4 缺失字段怎么办
    • 7.5 代码：`time_utils.py`
    • 7.6 代码：`parser.py`
  • 8️⃣ 数据存储与导出（Storage）
  • • 8.1 起步选择 SQLite
    • 8.2 字段映射表
    • 8.3 去重策略
    • 8.4 代码：`storage.py`
  • 9️⃣ 运行方式与结果展示（必写）
  • • 9.1 入口文件：`run_spider.py`
    • 9.2 导出 GeoJSON：`exporter.py`
    • 9.3 生成时间轴地图：`map_builder.py`
    • 9.4 启动命令
    • 9.5 输出路径
    • 9.6 示例结果
    • 9.7 查看 SQLite
  • 🔟 常见问题与排错（强烈建议写）
  • • 10.1 403 怎么办
    • 10.2 429 怎么办
    • 10.3 HTML 抓到空壳怎么办
    • 10.4 解析报错怎么办
    • 10.5 编码/乱码如何处理
    • 10.6 数据为空怎么办
    • 10.7 时间轴不动怎么办
  • 1️⃣1️⃣ 进阶优化（可选但加分）
  • • 11.1 并发优化
    • 11.2 断点续跑
    • 11.3 日志与监控
    • 11.4 定时任务
    • 11.5 空间网格升级
    • 11.6 强度指标升级
    • 11.7 多物种批量采集
    • 11.8 与 Scrapy 集成
  • 1️⃣2️⃣ 总结与延伸阅读
  • 🌟 文末
  • • ✅ 专栏持续更新中｜建议收藏 + 订阅
    • ✅ 互动征集
    • ✅ 免责声明

🌟 开篇语

哈喽，各位小伙伴们你们好呀～我是【喵手】。 运营社区： C站 / 掘金 / 腾讯云 / 阿里云 / 华为云 / 51CTO 欢迎大家常来逛逛，一起学习，一起进步～🌟

  我长期专注 Python 爬虫工程化实战，主理专栏👉 《Python爬虫实战》：从采集策略到反爬对抗，从数据清洗到分布式调度，持续输出可复用的方法论与可落地案例。内容主打一个“能跑、能用、能扩展”，让数据价值真正做到——抓得到、洗得净、用得上。

  📌 专栏食用指南（建议收藏）

• ✅ 入门基础：环境搭建 / 请求与解析 / 数据落库
• ✅ 进阶提升：登录鉴权 / 动态渲染 / 反爬对抗
• ✅ 工程实战：异步并发 / 分布式调度 / 监控与容错
• ✅ 项目落地：数据治理 / 可视化分析 / 场景化应用

📣 专栏推广时间：如果你想系统学爬虫，而不是碎片化东拼西凑，欢迎订阅专栏👉《Python爬虫实战》👈，一次订阅后，专栏内的所有文章可永久免费阅读，持续更新中。    💕订阅后更新会优先推送，按目录学习更高效💯～

0️⃣ 前言（Preface）

这篇文章要做的事情很明确：用 Python 采集公开海洋生物分布数据，围绕“物种/资源名、海域、时间、强度指标、来源”这几个字段，把动态渔业资源分布图背后的时间轴数据抓取出来，并最终导出为 CSV、SQLite、GeoJSON 和一个可播放时间轴的 HTML 地图。

读完这篇文章，你至少能拿到三样东西。

第一，你会知道动态地图类网站的数据到底藏在哪里。很多地图页面看上去是动画、图层、时间轴，本质上往往是前端按时间、范围、物种条件请求接口，再把返回的点、网格或瓦片渲染到地图上。真正要采集的不是页面上那张“会动的图”，而是图层背后的接口参数和响应数据。

第二，你会得到一套可运行的 Python 项目骨架。它包含请求层、解析层、清洗层、存储层、导出层和可视化层，不是只贴一个半截脚本就结束。代码会围绕 OBIS 公开 API 设计，默认用月度时间窗口采集指定物种在指定海域内的 occurrence 记录，再聚合成网格强度。

第三，你会掌握动态时间轴采集的基本套路：先用浏览器开发者工具观察接口，再把时间轴拆成一个个时间片，接着按时间片分页抓取，最后把原始点数据转换成“某物种在某海域、某月份、某网格的出现强度”。

我个人很喜欢这一类案例，因为它不像普通列表页采集那样只盯着标题、链接和正文，而是要处理地理范围、时间范围、图层聚合、分页、容错和导出。它更接近真实数据工程里的采集任务，也更能看出一个爬虫项目是不是有基本的工程意识。

1️⃣ 摘要（Abstract）

本文使用 Python、requests、pandas、SQLite、GeoJSON 与 Folium，采集公开海洋生物分布接口中的物种出现记录，并按月份和空间网格聚合为动态渔业资源分布图数据，最终产出可分析的表格数据和可播放时间轴的 HTML 地图。

读完本文可以获得：

本文选择公开、无需登录的数据接口做演示，不讨论绕过登录、破解付费数据、规避访问限制等内容。技术分享的重点是合规采集、结构化整理和可复现分析。

2️⃣ 背景与需求（Why）

2.1 为什么要采集动态渔业资源分布图

渔业资源、海洋生物出现记录、海域生态观测数据，通常不是一张静态表就能讲清楚的。它们天然带有三个维度：

• 空间：资源出现在哪里，靠近哪个海域，分布密度如何；
• 时间：某个资源在哪些月份更集中，是否有季节性变化；
• 强度：某个区域内记录数量、观测频率、相对密度或其他指标高不高。

很多公开平台会把这些数据做成地图图层，让用户通过筛选条件查看分布结果。有的页面还有时间滑块，拖动滑块后，地图上的点、热力图或网格颜色会发生变化。这种页面对普通用户很友好，但对数据分析来说不够方便，因为分析人员通常需要把图层背后的数据落成表格。

比如，我们可能会问：

• 某个鱼类资源在某片海域近几年记录最密集的月份是哪几个月？
• 资源分布是否呈现向某个方向移动的趋势？
• 哪些网格区域长期保持较高出现强度？
• 不同物种在同一海域的时空分布是否有明显差异？
• 是否能把公开记录整理成后续建模、看板或预警系统的数据源？

这些问题都需要结构化数据。光看地图截图不够，必须把图层数据采集下来，清洗、聚合、存储、导出。

2.2 本文目标站点与采集对象

本文以公开海洋生物记录接口作为案例，目标不是采集某个商业系统，也不是爬取受限数据。我们选择的思路是：

• 目标平台：公开海洋生物分布数据平台；
• 目标接口：occurrence 查询接口；
• 采集对象：指定物种在指定海域、指定时间范围内的出现记录；
• 时间粒度：按月拆分；
• 空间粒度：按经纬度网格聚合；
• 输出形式：CSV、SQLite、GeoJSON、HTML 动态时间轴地图。

本文设定一个示例任务：

采集 Gadus morhua 这个物种在北大西洋部分海域中，2020 年 1 月到 2020 年 12 月之间的出现记录，并按 1 度经纬度网格聚合为月度强度指标。

这里的 Gadus morhua 是大西洋鳕的学名。实际项目里，你可以把它替换成其他物种名，也可以把一个物种扩展成多个物种批量采集。

2.3 目标字段清单

本文最终希望拿到如下字段。

核心业务字段：

空间字段：

溯源字段：

强度指标在本文中定义为：同一物种、同一海域、同一月份、同一网格内的有效 occurrence 记录数量。这个指标不是官方资源量估计，也不是渔获量，只是一个基于公开记录的相对出现强度。这个定义必须写清楚，否则容易被误解为“真实资源储量”。

3️⃣ 合规与注意事项（必写）

爬虫技术本身没有问题，但使用方式必须克制。尤其是地图图层、时空数据、科学数据平台，很多由公共机构、研究组织或志愿者共同维护，访问时要尊重对方服务能力和数据政策。

3.1 robots.txt 基本说明

robots.txt 是网站放在根目录下的一个文本文件，用来声明哪些路径允许或不建议被自动化程序访问。它不是法律合同，也不是安全边界，但它是最基本的网络礼仪。

在正式采集前，建议做三件事：

示例检查方式：

import requests

def check_robots(base_url: str) –> None:
url = base_url.rstrip("/") + "/robots.txt"
resp = requests.get(url, timeout=15)
print(resp.status_code)
print(resp.text[:2000])

if __name__ == "__main__":
check_robots("https://api.obis.org")

如果 robots.txt 或官方文档给出更具体的访问说明，应以官方说明为准。没有看到限制，不代表可以无限制并发；看到公开接口，也不代表可以把对方服务当作自己的离线数据库来高频扫。

3.2 频率控制

动态地图接口最容易被误用。很多人打开开发者工具看到一个接口，就马上写循环，把时间、经纬度、物种、分页全部组合起来高并发请求。这样做很不专业，也容易给公共服务造成压力。

本文代码采用温和策略：

• 每次请求设置 timeout；
• 使用 Session 复用连接；
• 对失败请求做有限重试；
• 对 429、503 等状态码采用退避等待；
• 默认每次请求后 sleep 一小段时间；
• 不使用攻击式并发；
• 默认分页上限可配置，避免误采超大结果集。

对于公开科学数据平台，建议先小样本验证逻辑，再扩大范围。如果需要大规模分析，应使用官方推荐的大数据导出方式，而不是用 API 慢慢翻页硬抓。

3.3 不采集敏感信息

本文采集的是公开海洋生物 occurrence 数据，不采集个人隐私、账号信息、后台数据、登录态信息，也不绕过付费墙或权限控制。

如果目标网站需要登录才能访问，或者接口需要授权 token，应按照官方授权流程使用。不要通过抓包复制他人 token，不要模拟非公开接口，不要绕过访问限制。技术上能做到，不代表应该做。

3.4 数据解释要谨慎

公开 occurrence 数据通常是“记录过某物种在某处出现”，但它并不等价于真实资源量。某个网格记录多，可能是资源丰富，也可能只是采样更频繁。某个月记录少，可能是资源少，也可能是没有调查活动。

所以本文的 intensity_index 只定义为记录数量聚合，不把它包装成真实储量、捕捞量或生态结论。数据工程可以整理材料，科学解释要交给更严谨的分析模型和领域知识。

4️⃣ 技术选型与整体流程（What/How）

4.1 静态、动态、API 三类采集方式

常见网页采集可以粗略分成三类。

第一类是静态页面。HTML 里已经包含目标数据，用 requests 抓页面，再用 BeautifulSoup、lxml、XPath 或 CSS 选择器解析即可。比如新闻标题列表、普通表格、公告详情页，大多属于这一类。

第二类是动态页面。页面初始 HTML 只是一个空壳，真正的数据由 JavaScript 在浏览器里请求接口后渲染。地图图层、时间轴动画、无限滚动列表、前端看板，经常属于这一类。如果直接 requests 抓 HTML，通常只能看到一堆 script，看不到数据。

第三类是公开 API。前端页面也是调用 API，但平台本身提供文档、参数说明和结构化响应。这种情况下应该优先使用 API，而不是模拟浏览器点来点去。API 通常更稳定，字段更规范，也更适合写成可维护的项目。

本文属于“动态地图 + API 采集”。我们不是去解析地图画布上的像素，也不是从 HTML 里抠数据，而是识别地图背后的 occurrence 查询接口，用时间窗口和空间范围构造请求，再对响应 JSON 做解析和聚合。

4.2 动态图层采集的核心思路

动态地图采集的关键不是“地图”，而是“图层请求”。

一般可以按这个流程排查：

对于本文案例，我们把时间轴拆成月度窗口：

2020-01-01 ~ 2020-01-31
2020-02-01 ~ 2020-02-29
2020-03-01 ~ 2020-03-31
…
2020-12-01 ~ 2020-12-31

每个窗口内按照分页参数抓取 occurrence 记录。拿到原始记录后，根据经纬度落到网格，再统计每个网格的记录数量。

4.3 整体流程

文字版流程如下：

配置物种、海域 WKT、多个月份
↓
请求 occurrence API
↓
按 offset / size 分页
↓
解析 JSON records
↓
过滤缺失经纬度、缺失时间、无效记录
↓
写入 SQLite 原始表
↓
按 species + sea_area + month + grid 聚合
↓
导出 CSV / GeoJSON
↓
生成带时间轴的 HTML 地图

更像工程项目的流程如下：

采集 Fetcher
– 负责请求接口
– 负责 headers、timeout、retry、rate limit
– 负责分页

解析 Parser
– 负责 JSON 字段映射
– 负责时间格式、经纬度、物种名提取
– 负责容错

清洗 Cleaner
– 去掉缺失坐标
– 去掉异常经纬度
– 统一月份
– 生成 grid_id

存储 Storage
– 保存 raw_occurrences
– 保存 resource_intensity
– 保存 request_log
– 去重

导出 Exporter
– 导出 CSV
– 导出 GeoJSON
– 生成时间轴 HTML 地图

4.4 为什么选 requests，而不是一上来就 Playwright

很多动态网页确实需要 Playwright 或 Selenium，但本文不优先选它们，原因很简单：如果已经定位到公开接口，直接请求接口更稳定、更轻量、更容易测试。

Playwright 适合这些情况：

• 页面加密很重，接口参数由前端运行时生成；
• 必须执行 JavaScript 才能拿到 token；
• 数据只存在浏览器渲染后的 DOM 中；
• 需要截图、点击、滚动、模拟复杂交互。

本文的目标是接口数据采集，所以选择：

• requests：处理 HTTP 请求；
• pandas：处理表格清洗和导出；
• SQLite：本地结构化存储；
• Folium：生成 Leaflet 地图；
• tenacity：做重试和退避；
• tqdm：展示进度；
• python-dotenv：管理配置。

这套组合不花哨，但非常实用。项目后续要升级到 Scrapy、asyncio 或 Airflow，也不会推倒重来。

5️⃣ 环境准备与依赖安装（可复现）

5.1 Python 版本

建议使用 Python 3.10 或以上版本。本文示例在 Python 3.10+ 语法下编写，主要用到 dataclass、类型注解和 pathlib。

查看版本：

python –version

建议输出类似：

Python 3.10.13

5.2 创建虚拟环境

macOS / Linux：

mkdir fisheries_dynamic_obis
cd fisheries_dynamic_obis

python -m venv .venv
source .venv/bin/activate

Windows PowerShell：

mkdir fisheries_dynamic_obis
cd fisheries_dynamic_obis

python –m venv .venv
.venv\\Scripts\\Activate.ps1

5.3 安装依赖

创建 requirements.txt：

requests==2.32.3
pandas==2.2.2
tenacity==8.5.0
python-dotenv==1.0.1
tqdm==4.66.5
folium==0.17.0
branca==0.7.2

安装：

pip install -r requirements.txt

5.4 推荐项目结构

fisheries_dynamic_obis/
├── README.md
├── requirements.txt
├── .env.example
├── data/
│ ├── raw/
│ ├── processed/
│ └── maps/
├── logs/
├── fisheries_spider/
│ ├── __init__.py
│ ├── config.py
│ ├── time_utils.py
│ ├── fetcher.py
│ ├── parser.py
│ ├── storage.py
│ ├── exporter.py
│ └── map_builder.py
└── run_spider.py

这个结构不复杂，但足够清楚。fisheries_spider 放核心代码，data 放输出结果，logs 放运行日志，根目录的 run_spider.py 作为入口文件。

5.5 配置文件

创建 .env.example：

OBIS_BASE_URL=https://api.obis.org/v3
DEFAULT_TIMEOUT=30
REQUEST_SLEEP_SECONDS=0.8
PAGE_SIZE=500
MAX_PAGES_PER_MONTH=20

复制为 .env：

cp .env.example .env

Windows 下可以手动复制，或使用：

copy .env.example .env

6️⃣ 核心实现：请求层（Fetcher）

请求层的职责是“稳定、克制、可追踪地把数据拿回来”。它不应该关心字段怎么解析，也不应该直接写数据库。请求层只处理 HTTP。

6.1 headers 设计

即使是公开 API，也建议设置清晰的 User-Agent。不要伪装成奇怪的浏览器，也不要写空 UA。比较稳妥的写法是说明项目用途和联系信息。如果是公司或团队项目，可以写团队邮箱；个人测试可以写项目名称。

本文示例：

HEADERS = {
"User-Agent": "FisheriesDynamicMapCollector/1.0 (educational data collection; contact: example@example.com)",
"Accept": "application/json,text/plain,*/*",
"Referer": "https://mapper.obis.org/",
}

Referer 不是绕过限制的工具，只是告诉服务端请求来源场景。对于官方 API，不一定必须带 Referer；保留它主要是为了贴近地图图层接口的实际请求风格。

6.2 timeout

所有请求都必须设置 timeout。没有 timeout 的爬虫，迟早会卡死在某个网络抖动上。

本文默认：

timeout=30

实际项目中可以分开设置连接超时和读取超时：

timeout=(10, 30)

6.3 session / cookie

本文案例不需要登录 cookie。我们仍然使用 requests.Session()，原因是 Session 能复用连接，也方便统一 headers、重试策略和日志。

如果目标 API 明确需要官方 token，应通过配置读取，不要把 token 写死在代码里。本文不演示绕过登录态，也不采集需要权限的数据。

6.4 失败处理：重试与退避

失败处理至少要考虑这些情况：

• 连接超时；
• 读取超时；
• 502 / 503 / 504；
• 429 访问过快；
• JSON 解析失败；
• 服务器短暂返回空结果。

重试不是无限重试。本文使用最多 3 次重试，每次退避等待。遇到 429 时应该降低频率，而不是换代理硬冲。

6.5 代码：config.py

# fisheries_spider/config.py

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path

from dotenv import load_dotenv

load_dotenv()

BASE_DIR = Path(__file__).resolve().parents[1]
DATA_DIR = BASE_DIR / "data"
RAW_DIR = DATA_DIR / "raw"
PROCESSED_DIR = DATA_DIR / "processed"
MAP_DIR = DATA_DIR / "maps"
LOG_DIR = BASE_DIR / "logs"

for folder in [DATA_DIR, RAW_DIR, PROCESSED_DIR, MAP_DIR, LOG_DIR]:
folder.mkdir(parents=True, exist_ok=True)

@dataclass(frozen=True)
class SpiderConfig:
base_url: str = os.getenv("OBIS_BASE_URL", "https://api.obis.org/v3")
timeout: int = int(os.getenv("DEFAULT_TIMEOUT", "30"))
request_sleep_seconds: float = float(os.getenv("REQUEST_SLEEP_SECONDS", "0.8"))
page_size: int = int(os.getenv("PAGE_SIZE", "500"))
max_pages_per_month: int = int(os.getenv("MAX_PAGES_PER_MONTH", "20"))

sqlite_path: Path = PROCESSED_DIR / "fisheries_dynamic.sqlite3"
raw_csv_path: Path = PROCESSED_DIR / "raw_occurrences.csv"
intensity_csv_path: Path = PROCESSED_DIR / "resource_intensity.csv"
intensity_geojson_path: Path = PROCESSED_DIR / "resource_intensity.geojson"
timeline_map_path: Path = MAP_DIR / "resource_timeline_map.html"

DEFAULT_HEADERS = {
"User-Agent": (
"FisheriesDynamicMapCollector/1.0 "
"(educational data collection; contact: example@example.com)"
),
"Accept": "application/json,text/plain,*/*",
"Referer": "https://mapper.obis.org/",
}

6.6 代码：fetcher.py

# fisheries_spider/fetcher.py

from __future__ import annotations

import logging
import time
from typing import Any, Dict, Iterator, Optional
from urllib.parse import urlencode

import requests
from requests.adapters import HTTPAdapter
from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential
from urllib3.util.retry import Retry

from fisheries_spider.config import DEFAULT_HEADERS, SpiderConfig

logger = logging.getLogger(__name__)

class FetchError(RuntimeError):
"""Raised when the fetcher fails after retries."""

class ObisFetcher:
"""
A conservative API fetcher for OBIS occurrence records.

It is intentionally simple:
– requests.Session for connection reuse
– explicit headers
– timeout
– limited retries
– sleep between requests
– offset/size pagination
"""

def __init__(self, config: SpiderConfig) –> None:
self.config = config
self.session = requests.Session()
self.session.headers.update(DEFAULT_HEADERS)
self._mount_retries()

def _mount_retries(self) –> None:
retry_strategy = Retry(
total=2,
connect=2,
read=2,
status=2,
backoff_factor=1.2,
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=frozenset(["GET"]),
raise_on_status=False,
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount("https://", adapter)
self.session.mount("http://", adapter)

def build_occurrence_url(self) –> str:
return self.config.base_url.rstrip("/") + "/occurrence"

@retry(
retry=retry_if_exception_type((requests.RequestException, ValueError)),
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
reraise=True,
)
def get_json(self, url: str, params: Dict[str, Any]) –> Dict[str, Any]:
"""
Execute GET request and parse JSON.

ValueError is included because response.json() raises it
when the body is not valid JSON.
"""
full_url = f"{url}?{urlencode(params, doseq=True)}"
logger.info("GET %s", full_url)

resp = self.session.get(url, params=params, timeout=self.config.timeout)

if resp.status_code == 429:
logger.warning("429 Too Many Requests. Slowing down before retry.")
time.sleep(max(self.config.request_sleep_seconds * 5, 5))

if resp.status_code >= 400:
raise FetchError(f"HTTP {resp.status_code}: {resp.text[:300]}")

data = resp.json()
time.sleep(self.config.request_sleep_seconds)
return data

def iter_occurrences(
self,
scientific_name: str,
start_date: str,
end_date: str,
geometry_wkt: str,
fields: Optional[str] = None,
) –> Iterator[Dict[str, Any]]:
"""
Iterate occurrence records by offset pagination.

Parameters:
scientific_name: species/resource scientific name.
start_date: YYYY-MM-DD.
end_date: YYYY-MM-DD.
geometry_wkt: WKT polygon.
fields: optional comma-separated fields.
"""
url = self.build_occurrence_url()
offset = 0
page_index = 0

while True:
if page_index >= self.config.max_pages_per_month:
logger.warning(
"Reached max_pages_per_month=%s for %s %s ~ %s",
self.config.max_pages_per_month,
scientific_name,
start_date,
end_date,
)
break

params: Dict[str, Any] = {
"scientificname": scientific_name,
"startdate": start_date,
"enddate": end_date,
"geometry": geometry_wkt,
"size": self.config.page_size,
"offset": offset,
}

if fields:
params["fields"] = fields

payload = self.get_json(url, params=params)
records = self._extract_results(payload)

if not records:
logger.info(
"No more records for %s %s ~ %s at offset=%s",
scientific_name,
start_date,
end_date,
offset,
)
break

for record in records:
record["_request_url"] = f"{url}?{urlencode(params, doseq=True)}"
yield record

got = len(records)
logger.info(
"Fetched page=%s offset=%s size=%s got=%s",
page_index,
offset,
self.config.page_size,
got,
)

if got < self.config.page_size:
break

offset += self.config.page_size
page_index += 1

@staticmethod
def _extract_results(payload: Any) –> list[Dict[str, Any]]:
"""
OBIS occurrence response is commonly a dict with a 'results' list.
This function is defensive because API wrappers or endpoints can
return slightly different shapes.
"""
if isinstance(payload, list):
return [x for x in payload if isinstance(x, dict)]

if isinstance(payload, dict):
for key in ("results", "features", "data"):
value = payload.get(key)
if isinstance(value, list):
if key == "features":
return [
item.get("properties", item)
for item in value
if isinstance(item, dict)
]
return [x for x in value if isinstance(x, dict)]

return []

请求层写到这里，基本能抗住小规模采集。注意，max_pages_per_month 是一个很有必要的保险丝。刚开始调试时，不要让程序在一个月里翻几百页。先拿小样本验证解析逻辑，再慢慢放大。

7️⃣ 核心实现：解析层（Parser）

解析层处理三件事：

7.1 解析方式

本文接口返回 JSON，所以解析方式是 JSON 字段映射，不需要 XPath、CSS 或 BeautifulSoup。

不过动态地图采集不一定都是 JSON。有些地图图层可能返回：

• GeoJSON；
• MVT 矢量瓦片；
• KML；
• CSV；
• 压缩后的二进制数据；
• 前端自定义结构。

本文为了可读性，使用 occurrence JSON。后续如果遇到 MVT，可以用 mapbox-vector-tile 解析；遇到 GeoJSON，则从 features[].properties 中拿属性字段，从 geometry.coordinates 中拿坐标。

7.2 列表页如何拿详情链接

本文不走传统列表页和详情页模式。动态地图数据通常没有“列表页详情链接”这个结构，它是直接返回记录集合。

但为了对应普通采集项目里的“列表页 → 详情页”思路，可以这样理解：

• 列表页：一次 occurrence 查询返回的 results；
• 详情链接：每条记录中的 id 或 occurrenceID；
• 详情页：如 API 支持 /occurrence/{id}，可以按 ID 再查详情；
• 本文策略：为了降低请求量，优先使用列表接口返回字段，不对每条记录再发详情请求。

这是很重要的工程取舍。很多时候，列表接口已经包含分析所需字段。为了一个可有可无的字段给每条记录补详情，会让请求量暴涨，不值得。

7.3 详情页如何抽字段

本文默认不抓详情页。如果实际需要补充详情，可以写一个 get_occurrence_detail(occurrence_id)，但要有缓存和去重，避免重复请求。

常见字段映射如下：

7.4 缺失字段怎么办

数据采集不能假设每条记录都完美。字段缺失时按下面规则处理：

• 缺失 occurrence_id：用内容 hash 作为去重补充；
• 缺失 species_name：使用请求参数里的 species 兜底；
• 缺失 event_date：尝试 date_mid / date_start；仍无则丢弃或标记 unknown；
• 缺失经纬度：不能进入空间聚合；
• 经纬度超范围：丢弃；
• dataset_id 缺失：保留为空；
• basis_of_record 缺失：保留为空；
• source 固定写入采集来源。

7.5 代码：time_utils.py

# fisheries_spider/time_utils.py

from __future__ import annotations

import calendar
from dataclasses import dataclass
from datetime import date
from typing import Iterator

@dataclass(frozen=True)
class MonthWindow:
month: str
start_date: str
end_date: str

def iter_month_windows(start_month: str, end_month: str) –> Iterator[MonthWindow]:
"""
Generate inclusive month windows.

Example:
start_month='2020-01'
end_month='2020-03'

Yields:
2020-01-01 ~ 2020-01-31
2020-02-01 ~ 2020-02-29
2020-03-01 ~ 2020-03-31
"""
sy, sm = map(int, start_month.split("-"))
ey, em = map(int, end_month.split("-"))

current_year = sy
current_month = sm

while (current_year, current_month) <= (ey, em):
last_day = calendar.monthrange(current_year, current_month)[1]
start = date(current_year, current_month, 1)
end = date(current_year, current_month, last_day)

month_text = f"{current_year:04d}–{current_month:02d}"
yield MonthWindow(
month=month_text,
start_date=start.isoformat(),
end_date=end.isoformat(),
)

current_month += 1
if current_month > 12:
current_month = 1
current_year += 1

7.6 代码：parser.py

# fisheries_spider/parser.py

from __future__ import annotations

import hashlib
import json
import math
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from typing import Any, Dict, Optional

@dataclass
class RawOccurrence:
occurrence_id: str
species_name: str
sea_area: str
time_window: str
event_date: Optional[str]
longitude: float
latitude: float
dataset_id: Optional[str]
basis_of_record: Optional[str]
source: str
request_url: str
content_hash: str
fetched_at: str

@dataclass
class IntensityRecord:
species_name: str
sea_area: str
time_window: str
grid_id: str
grid_lon: int
grid_lat: int
intensity_index: int
source: str

def now_utc_iso() –> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()

def stable_hash(obj: Dict[str, Any]) –> str:
raw = json.dumps(obj, ensure_ascii=False, sort_keys=True, default=str)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()

def pick_first(record: Dict[str, Any], keys: list[str]) –> Any:
for key in keys:
value = record.get(key)
if value not in (None, ""):
return value
return None

def parse_float(value: Any) –> Optional[float]:
if value in (None, ""):
return None
try:
result = float(value)
except (TypeError, ValueError):
return None

if math.isnan(result) or math.isinf(result):
return None

return result

def valid_lon_lat(lon: float, lat: float) –> bool:
return –180 <= lon <= 180 and –90 <= lat <= 90

def parse_event_date(record: Dict[str, Any]) –> Optional[str]:
"""
Prefer eventDate. Fall back to date_mid/date_start if present.

date_mid/date_start may be unix timestamps in some datasets. This parser
is intentionally conservative: it keeps readable date strings and ignores
numeric timestamps unless you extend it.
"""
value = pick_first(record, ["eventDate", "event_date", "date", "date_mid", "date_start"])
if value is None:
return None

if isinstance(value, str):
return value[:30]

return str(value)

def grid_id_from_lon_lat(lon: float, lat: float, grid_size: float = 1.0) –> tuple[str, int, int]:
"""
Convert lon/lat to a simple degree grid.

For example:
lon=-63.25, lat=44.71, grid_size=1
-> grid_lon=-64, grid_lat=44

This is simple and explainable. For production analysis, consider geohash,
H3, S2, or official statistical areas.
"""
grid_lon = math.floor(lon / grid_size) * int(grid_size)
grid_lat = math.floor(lat / grid_size) * int(grid_size)

grid_id = f"lon_{grid_lon}_lat_{grid_lat}"
return grid_id, int(grid_lon), int(grid_lat)

def parse_occurrence(
record: Dict[str, Any],
fallback_species: str,
sea_area: str,
time_window: str,
source: str = "OBIS API",
) –> Optional[RawOccurrence]:
lon = parse_float(pick_first(record, ["decimalLongitude", "longitude", "lon", "x"]))
lat = parse_float(pick_first(record, ["decimalLatitude", "latitude", "lat", "y"]))

if lon is None or lat is None:
return None

if not valid_lon_lat(lon, lat):
return None

content_hash = stable_hash(record)
occurrence_id = str(
pick_first(record, ["id", "occurrenceID", "occurrence_id", "record_id"])
or content_hash
)

species_name = str(
pick_first(record, ["scientificName", "species", "acceptedNameUsage", "taxon"])
or fallback_species
)

event_date = parse_event_date(record)

dataset_id = pick_first(record, ["dataset_id", "datasetID", "resource_id"])
basis_of_record = pick_first(record, ["basisOfRecord", "basis_of_record"])

request_url = str(record.get("_request_url") or "")

return RawOccurrence(
occurrence_id=occurrence_id,
species_name=species_name,
sea_area=sea_area,
time_window=time_window,
event_date=event_date,
longitude=lon,
latitude=lat,
dataset_id=str(dataset_id) if dataset_id is not None else None,
basis_of_record=str(basis_of_record) if basis_of_record is not None else None,
source=source,
request_url=request_url,
content_hash=content_hash,
fetched_at=now_utc_iso(),
)

def raw_to_dict(item: RawOccurrence) –> Dict[str, Any]:
return asdict(item)

def intensity_to_dict(item: IntensityRecord) –> Dict[str, Any]:
return asdict(item)

这个解析器刻意写得保守。它不会因为某个字段缺失就让整个程序崩掉，也不会把没有经纬度的数据硬塞进空间聚合。动态地图采集最怕的不是“抓不到”，而是“抓到了脏数据还不知道”。

8️⃣ 数据存储与导出（Storage）

8.1 起步选择 SQLite

本文选择 SQLite 作为本地存储。它的优点是：

• 不需要部署数据库服务；
• 一个文件就能保存结果；
• 支持 SQL 查询；
• 适合中小规模采集；
• 后续迁移到 MySQL/PostgreSQL 也不难。

如果数据量很大，或者需要多人共享，建议换 PostgreSQL + PostGIS。空间分析多的项目，PostGIS 会舒服很多。

8.2 字段映射表

原始表 raw_occurrences：

聚合表 resource_intensity：

请求日志表 request_log：

8.3 去重策略

本文采用两层去重：

第一层：occurrence_id 唯一。如果接口返回稳定 ID，就用它作为主键。

第二层：content_hash。如果没有 ID，就对原始 JSON 做 sha256，生成一个稳定 hash，作为兜底 ID 或辅助去重字段。

为什么不只用 URL 去重？因为动态 API 同一个 URL 可能返回多条记录，而一条记录也可能在不同时间窗口或不同查询条件里被命中。URL 去重适合页面级采集，不适合 occurrence 记录级采集。

8.4 代码：storage.py

# fisheries_spider/storage.py

from __future__ import annotations

import sqlite3
from collections import Counter
from pathlib import Path
from typing import Iterable

import pandas as pd

from fisheries_spider.parser import (
IntensityRecord,
RawOccurrence,
grid_id_from_lon_lat,
intensity_to_dict,
raw_to_dict,
)

class SQLiteStorage:
def __init__(self, db_path: Path) –> None:
self.db_path = db_path
self.conn = sqlite3.connect(str(db_path))
self.conn.row_factory = sqlite3.Row
self.init_db()

def init_db(self) –> None:
cur = self.conn.cursor()

cur.execute(
"""
CREATE TABLE IF NOT EXISTS raw_occurrences (
occurrence_id TEXT PRIMARY KEY,
species_name TEXT NOT NULL,
sea_area TEXT NOT NULL,
time_window TEXT NOT NULL,
event_date TEXT,
longitude REAL NOT NULL,
latitude REAL NOT NULL,
dataset_id TEXT,
basis_of_record TEXT,
source TEXT NOT NULL,
request_url TEXT,
content_hash TEXT NOT NULL,
fetched_at TEXT NOT NULL
)
"""
)

cur.execute(
"""
CREATE INDEX IF NOT EXISTS idx_raw_species_time
ON raw_occurrences (species_name, time_window)
"""
)

cur.execute(
"""
CREATE INDEX IF NOT EXISTS idx_raw_geo
ON raw_occurrences (longitude, latitude)
"""
)

cur.execute(
"""
CREATE TABLE IF NOT EXISTS resource_intensity (
species_name TEXT NOT NULL,
sea_area TEXT NOT NULL,
time_window TEXT NOT NULL,
grid_id TEXT NOT NULL,
grid_lon INTEGER NOT NULL,
grid_lat INTEGER NOT NULL,
intensity_index INTEGER NOT NULL,
source TEXT NOT NULL,
PRIMARY KEY (species_name, sea_area, time_window, grid_id)
)
"""
)

cur.execute(
"""
CREATE TABLE IF NOT EXISTS request_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
species_name TEXT NOT NULL,
time_window TEXT NOT NULL,
start_date TEXT NOT NULL,
end_date TEXT NOT NULL,
sea_area TEXT NOT NULL,
records_saved INTEGER NOT NULL,
created_at TEXT NOT NULL
)
"""
)

self.conn.commit()

def insert_raw_occurrences(self, rows: Iterable[RawOccurrence]) –> int:
sql = """
INSERT OR IGNORE INTO raw_occurrences (
occurrence_id,
species_name,
sea_area,
time_window,
event_date,
longitude,
latitude,
dataset_id,
basis_of_record,
source,
request_url,
content_hash,
fetched_at
) VALUES (
:occurrence_id,
:species_name,
:sea_area,
:time_window,
:event_date,
:longitude,
:latitude,
:dataset_id,
:basis_of_record,
:source,
:request_url,
:content_hash,
:fetched_at
)
"""

items = [raw_to_dict(row) for row in rows]
if not items:
return 0

before = self.conn.total_changes
self.conn.executemany(sql, items)
self.conn.commit()
after = self.conn.total_changes
return after – before

def log_request(
self,
species_name: str,
time_window: str,
start_date: str,
end_date: str,
sea_area: str,
records_saved: int,
created_at: str,
) –> None:
self.conn.execute(
"""
INSERT INTO request_log (
species_name,
time_window,
start_date,
end_date,
sea_area,
records_saved,
created_at
) VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(
species_name,
time_window,
start_date,
end_date,
sea_area,
records_saved,
created_at,
),
)
self.conn.commit()

def rebuild_intensity(self, grid_size: float = 1.0) –> int:
"""
Rebuild intensity table from raw_occurrences.
intensity_index = count of raw occurrence records in each grid/month/species.
"""
df = pd.read_sql_query("SELECT * FROM raw_occurrences", self.conn)

self.conn.execute("DELETE FROM resource_intensity")
self.conn.commit()

if df.empty:
return 0

counter: Counter[tuple[str, str, str, str, int, int, str]] = Counter()

for row in df.itertuples(index=False):
grid_id, grid_lon, grid_lat = grid_id_from_lon_lat(
float(row.longitude),
float(row.latitude),
grid_size=grid_size,
)
key = (
row.species_name,
row.sea_area,
row.time_window,
grid_id,
grid_lon,
grid_lat,
row.source,
)
counter[key] += 1

intensity_rows = [
IntensityRecord(
species_name=species_name,
sea_area=sea_area,
time_window=time_window,
grid_id=grid_id,
grid_lon=grid_lon,
grid_lat=grid_lat,
intensity_index=count,
source=source,
)
for (
species_name,
sea_area,
time_window,
grid_id,
grid_lon,
grid_lat,
source,
), count in counter.items()
]

sql = """
INSERT OR REPLACE INTO resource_intensity (
species_name,
sea_area,
time_window,
grid_id,
grid_lon,
grid_lat,
intensity_index,
source
) VALUES (
:species_name,
:sea_area,
:time_window,
:grid_id,
:grid_lon,
:grid_lat,
:intensity_index,
:source
)
"""

items = [intensity_to_dict(row) for row in intensity_rows]
self.conn.executemany(sql, items)
self.conn.commit()

return len(items)

def export_raw_csv(self, output_path: Path) –> None:
df = pd.read_sql_query("SELECT * FROM raw_occurrences", self.conn)
df.to_csv(output_path, index=False, encoding="utf-8-sig")

def export_intensity_csv(self, output_path: Path) –> None:
df = pd.read_sql_query(
"""
SELECT
species_name,
sea_area,
time_window,
grid_id,
grid_lon,
grid_lat,
intensity_index,
source
FROM resource_intensity
ORDER BY species_name, time_window, intensity_index DESC
""",
self.conn,
)
df.to_csv(output_path, index=False, encoding="utf-8-sig")

def read_intensity_df(self) –> pd.DataFrame:
return pd.read_sql_query("SELECT * FROM resource_intensity", self.conn)

def close(self) –> None:
self.conn.close()

这部分代码有一个细节：聚合表不是边抓边算，而是在原始表写入后统一重建。这样调试更方便。你修改网格大小或强度定义后，可以直接重跑聚合，不需要重新请求接口。

9️⃣ 运行方式与结果展示（必写）

9.1 入口文件：run_spider.py

# run_spider.py

from __future__ import annotations

import logging
from typing import List

from tqdm import tqdm

from fisheries_spider.config import SpiderConfig
from fisheries_spider.fetcher import ObisFetcher
from fisheries_spider.parser import now_utc_iso, parse_occurrence
from fisheries_spider.storage import SQLiteStorage
from fisheries_spider.time_utils import iter_month_windows
from fisheries_spider.exporter import export_intensity_geojson
from fisheries_spider.map_builder import build_timeline_map

logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)

def main() –> None:
config = SpiderConfig()

# 示例物种。实际项目可扩展成多个物种。
species_list: List[str] = [
"Gadus morhua",
]

# 示例海域：北大西洋部分区域。
# WKT 坐标顺序是 lon lat，不是 lat lon。
sea_area = "North Atlantic Demo Area"
geometry_wkt = "POLYGON((-75 35,-75 55,-45 55,-45 35,-75 35))"

start_month = "2020-01"
end_month = "2020-12"

fetcher = ObisFetcher(config)
storage = SQLiteStorage(config.sqlite_path)

# 建议限制字段，减少响应体积。
# 注意：如果你后续要更多字段，可以在这里增加。
fields = ",".join(
[
"id",
"scientificName",
"eventDate",
"decimalLongitude",
"decimalLatitude",
"dataset_id",
"basisOfRecord",
]
)

try:
for species_name in species_list:
windows = list(iter_month_windows(start_month, end_month))

for window in tqdm(windows, desc=f"Collecting {species_name}"):
parsed_rows = []

for record in fetcher.iter_occurrences(
scientific_name=species_name,
start_date=window.start_date,
end_date=window.end_date,
geometry_wkt=geometry_wkt,
fields=fields,
):
parsed = parse_occurrence(
record=record,
fallback_species=species_name,
sea_area=sea_area,
time_window=window.month,
source="OBIS API",
)
if parsed is not None:
parsed_rows.append(parsed)

saved = storage.insert_raw_occurrences(parsed_rows)
storage.log_request(
species_name=species_name,
time_window=window.month,
start_date=window.start_date,
end_date=window.end_date,
sea_area=sea_area,
records_saved=saved,
created_at=now_utc_iso(),
)

intensity_count = storage.rebuild_intensity(grid_size=1.0)
logging.info("Rebuilt intensity table. rows=%s", intensity_count)

storage.export_raw_csv(config.raw_csv_path)
storage.export_intensity_csv(config.intensity_csv_path)
export_intensity_geojson(storage, config.intensity_geojson_path)
build_timeline_map(
geojson_path=config.intensity_geojson_path,
output_html_path=config.timeline_map_path,
)

logging.info("Raw CSV: %s", config.raw_csv_path)
logging.info("Intensity CSV: %s", config.intensity_csv_path)
logging.info("GeoJSON: %s", config.intensity_geojson_path)
logging.info("Timeline map: %s", config.timeline_map_path)

finally:
storage.close()

if __name__ == "__main__":
main()

9.2 导出 GeoJSON：exporter.py

# fisheries_spider/exporter.py

from __future__ import annotations

import json
from pathlib import Path
from typing import Dict, List

from fisheries_spider.storage import SQLiteStorage

def grid_polygon(grid_lon: int, grid_lat: int, size: float = 1.0) –> list:
"""
Build a simple polygon for a lon/lat grid cell.

Coordinates follow GeoJSON order: [lon, lat].
"""
lon0 = grid_lon
lat0 = grid_lat
lon1 = grid_lon + size
lat1 = grid_lat + size

return [
[lon0, lat0],
[lon1, lat0],
[lon1, lat1],
[lon0, lat1],
[lon0, lat0],
]

def export_intensity_geojson(
storage: SQLiteStorage,
output_path: Path,
grid_size: float = 1.0,
) –> None:
df = storage.read_intensity_df()

features: List[Dict] = []

if not df.empty:
max_intensity = int(df["intensity_index"].max())
else:
max_intensity = 0

for row in df.itertuples(index=False):
intensity = int(row.intensity_index)

feature = {
"type": "Feature",
"geometry": {
"type": "Polygon",
"coordinates": [
grid_polygon(
grid_lon=int(row.grid_lon),
grid_lat=int(row.grid_lat),
size=grid_size,
)
],
},
"properties": {
"species_name": row.species_name,
"sea_area": row.sea_area,
"time_window": row.time_window,
"time": f"{row.time_window}-01",
"grid_id": row.grid_id,
"grid_lon": int(row.grid_lon),
"grid_lat": int(row.grid_lat),
"intensity_index": intensity,
"max_intensity_in_export": max_intensity,
"source": row.source,
"popup": (
f"Species: {row.species_name}<br>"
f"Sea area: {row.sea_area}<br>"
f"Month: {row.time_window}<br>"
f"Grid: {row.grid_id}<br>"
f"Intensity: {intensity}<br>"
f"Source: {row.source}"
),
},
}
features.append(feature)

geojson = {
"type": "FeatureCollection",
"features": features,
}

output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(
json.dumps(geojson, ensure_ascii=False, indent=2),
encoding="utf-8",
)

9.3 生成时间轴地图：map_builder.py

# fisheries_spider/map_builder.py

from __future__ import annotations

import json
from pathlib import Path

import folium
from folium.plugins import TimestampedGeoJson

def style_function(feature: dict) –> dict:
props = feature.get("properties", {})
intensity = props.get("intensity_index", 0) or 0
max_intensity = props.get("max_intensity_in_export", 1) or 1

ratio = min(float(intensity) / float(max_intensity), 1.0) if max_intensity else 0

# Folium style_function can use explicit colors.
# This is map styling, not chart styling.
if ratio >= 0.75:
fill_color = "#800026"
elif ratio >= 0.5:
fill_color = "#BD0026"
elif ratio >= 0.25:
fill_color = "#E31A1C"
elif ratio > 0:
fill_color = "#FC4E2A"
else:
fill_color = "#FFEDA0"

return {
"fillColor": fill_color,
"color": "#333333",
"weight": 0.5,
"fillOpacity": 0.55,
}

def build_timeline_map(geojson_path: Path, output_html_path: Path) –> None:
data = json.loads(geojson_path.read_text(encoding="utf-8"))

fmap = folium.Map(
location=[45, –60],
zoom_start=4,
tiles="CartoDB positron",
)

TimestampedGeoJson(
data=data,
period="P1M",
duration="P1M",
transition_time=400,
auto_play=False,
loop=False,
add_last_point=False,
time_slider_drag_update=True,
style=style_function,
).add_to(fmap)

folium.LayerControl().add_to(fmap)

output_html_path.parent.mkdir(parents=True, exist_ok=True)
fmap.save(str(output_html_path))

这个 HTML 地图不是截图，而是一个可交互页面。打开后可以拖动时间轴，查看不同月份的网格强度变化。这个效果和很多动态分布图的基本交互是一致的。

9.4 启动命令

在项目根目录运行：

python run_spider.py

如果成功，会看到类似日志：

Collecting Gadus morhua: 100%|████████████████████| 12/12 [00:42<00:00, 3.51s/it]
2026-06-12 10:30:12 | INFO | root | Rebuilt intensity table. rows=86
2026-06-12 10:30:12 | INFO | root | Raw CSV: data/processed/raw_occurrences.csv
2026-06-12 10:30:12 | INFO | root | Intensity CSV: data/processed/resource_intensity.csv
2026-06-12 10:30:12 | INFO | root | GeoJSON: data/processed/resource_intensity.geojson
2026-06-12 10:30:12 | INFO | root | Timeline map: data/maps/resource_timeline_map.html

9.5 输出路径

输出文件如下：

data/processed/fisheries_dynamic.sqlite3
data/processed/raw_occurrences.csv
data/processed/resource_intensity.csv
data/processed/resource_intensity.geojson
data/maps/resource_timeline_map.html

SQLite 中包含三张表：

SELECT name FROM sqlite_master WHERE type='table';

预期结果：

raw_occurrences
resource_intensity
request_log

9.6 示例结果

resource_intensity.csv 示例：

species_name,sea_area,time_window,grid_id,grid_lon,grid_lat,intensity_index,source
Gadus morhua,North Atlantic Demo Area,2020-01,lon_-63_lat_44,-63,44,12,OBIS API
Gadus morhua,North Atlantic Demo Area,2020-01,lon_-62_lat_45,-62,45,7,OBIS API
Gadus morhua,North Atlantic Demo Area,2020-02,lon_-64_lat_46,-64,46,15,OBIS API
Gadus morhua,North Atlantic Demo Area,2020-03,lon_-61_lat_43,-61,43,9,OBIS API
Gadus morhua,North Atlantic Demo Area,2020-04,lon_-58_lat_47,-58,47,4,OBIS API

这几行表达的意思是：在指定海域内，某物种在某个月份、某个经纬度网格中有多少条有效 occurrence 记录。intensity_index 越高，说明该网格在该月的记录越多。再次强调，它是记录强度，不是资源储量。

9.7 查看 SQLite

可以用 Python 快速检查：

import sqlite3
import pandas as pd

conn = sqlite3.connect("data/processed/fisheries_dynamic.sqlite3")

df = pd.read_sql_query(
"""
SELECT *
FROM resource_intensity
ORDER BY time_window, intensity_index DESC
LIMIT 10
""",
conn,
)

print(df)
conn.close()

也可以用 sqlite3 命令行：

sqlite3 data/processed/fisheries_dynamic.sqlite3

进入后执行：

.headers on
.mode column

SELECT species_name, time_window, grid_id, intensity_index
FROM resource_intensity
ORDER BY intensity_index DESC
LIMIT 10;

🔟 常见问题与排错（强烈建议写）

10.1 403 怎么办

403 通常表示服务器拒绝访问。可能原因包括：

• 请求路径不允许；
• 缺少必要 headers；
• 接口不对外开放；
• 访问方式不符合平台规则；
• IP 被临时限制。

处理思路：

不要把 403 简单理解为“缺代理”。很多 403 是规则问题，不是网络问题。

10.2 429 怎么办

429 表示请求过快。正确处理方式是降速。

本文代码里已经对 429 做了退避等待。还可以继续优化：

• 增大 REQUEST_SLEEP_SECONDS；
• 降低 PAGE_SIZE；
• 减少物种数量；
• 减少时间范围；
• 避开高峰时段；
• 使用官方推荐的大数据导出方式。

不建议遇到 429 就换代理池。公共数据平台不是靶场，采集要有边界。

10.3 HTML 抓到空壳怎么办

动态页面直接 requests 抓 HTML，经常只能看到：

<div id="app"></div>
<script src="/assets/index.js"></script>

这说明数据不是在 HTML 里，而是由前端 JavaScript 请求接口后渲染。

解决方法：

如果接口参数由前端复杂生成，再考虑 Playwright。但只要能定位到清晰 API，就不要用浏览器自动化硬跑。

10.4 解析报错怎么办

常见解析错误包括：

• 字段名变化；
• 返回结构变化；
• 某些记录缺少经纬度；
• 某些记录 eventDate 为空；
• 响应不是 JSON；
• 接口返回错误信息但状态码仍是 200。

处理方式：

• 解析函数要用 dict.get()；
• 用 pick_first() 支持多个候选字段；
• 缺少关键字段就跳过；
• 保存部分原始响应用于排查；
• 记录失败样本；
• 单独写小测试验证解析器。

不要把解析逻辑写成一串硬索引，比如：

record["data"]["items"][0]["location"]["lat"]

这种写法在真实数据里很脆。更稳妥的方式是逐层判断。

10.5 编码/乱码如何处理

JSON 接口一般问题不大。如果导出 CSV 给 Excel 打开，中文字段可能出现乱码。本文使用：

encoding="utf-8-sig"

这样 Excel 兼容性更好。

如果网页采集遇到乱码，可以检查：

resp.encoding
resp.apparent_encoding

必要时手动设置：

resp.encoding = "utf-8"

但不要盲目设置。先看响应头和页面实际编码。

10.6 数据为空怎么办

数据为空不一定是程序错。可能原因很多：

• 物种名不存在或拼写不一致；
• 时间范围内没有记录；
• 海域 WKT 范围太小；
• 经纬度顺序写反；
• 参数名写错；
• 接口分页逻辑不对；
• 字段过滤过窄。

排查顺序建议：

WKT 最容易踩坑的是坐标顺序。WKT 通常写 lon lat，不是 lat lon。例如：

POLYGON((-75 35,-75 55,-45 55,-45 35,-75 35))

这里 -75 是经度，35 是纬度。

10.7 时间轴不动怎么办

Folium 生成时间轴地图时，需要 GeoJSON feature 里有可识别的时间属性。本文写入的是：

"properties": {
"time": "2020-03-01"
}

如果时间轴不动，检查：

• 是否有 time 字段；
• 时间格式是否类似 YYYY-MM-DD；
• GeoJSON 是否为空；
• feature geometry 是否有效；
• 浏览器控制台是否有 JS 报错。

如果数据量很大，HTML 会变得很重。此时应该考虑按时间切片加载，或者用专门的 WebGIS 服务发布图层。

1️⃣1️⃣ 进阶优化（可选但加分）

11.1 并发优化

本文没有使用并发，是故意的。对公开 API，先保证温和和稳定，再谈速度。

如果后续确实需要并发，可以考虑三种方案：

第一种，线程池：

from concurrent.futures import ThreadPoolExecutor, as_completed

def collect_one_window(args):
species_name, window = args
# call fetcher and parser here
return species_name, window.month, 0

with ThreadPoolExecutor(max_workers=2) as executor:
futures = [
executor.submit(collect_one_window, item)
for item in tasks
]

for future in as_completed(futures):
print(future.result())

注意，max_workers 不要太大。对于公共 API，2 到 4 已经不少了。

第二种，asyncio + aiohttp。适合大量小请求，但复杂度更高。

第三种，Scrapy。适合任务队列、去重、重试、日志、限速、扩展中间件都比较复杂的项目。

11.2 断点续跑

当前代码每次运行会继续插入原始表，因为 occurrence_id 是主键，重复记录会被忽略。但请求本身还是会重发。

更好的断点续跑方式是：

• request_log 记录每个 species + month + sea_area 是否完成；
• 启动时先查 request_log；
• 已完成的窗口跳过；
• 未完成的窗口继续抓。

示例函数：

def is_window_done(conn, species_name: str, sea_area: str, time_window: str) –> bool:
cur = conn.execute(
"""
SELECT COUNT(*) AS cnt
FROM request_log
WHERE species_name = ?
AND sea_area = ?
AND time_window = ?
""",
(species_name, sea_area, time_window),
)
row = cur.fetchone()
return row["cnt"] > 0

但要注意：如果上次采集失败却写了日志，就会误判完成。更严谨的做法是给 request_log 加 status 字段：

running / success / failed

开始采集前写 running，成功后改 success，失败后改 failed。

11.3 日志与监控

最低限度的日志应该包括：

• 请求 URL；
• 物种；
• 时间窗口；
• offset；
• 返回数量；
• 保存数量；
• 失败原因；
• 总耗时。

可以把日志写入文件：

from pathlib import Path
import logging

log_path = Path("logs/spider.log")
log_path.parent.mkdir(exist_ok=True)

logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
handlers=[
logging.FileHandler(log_path, encoding="utf-8"),
logging.StreamHandler(),
],
)

对于长期任务，可以统计：

• 成功窗口数；
• 失败窗口数；
• 空结果窗口数；
• 平均每页耗时；
• 平均每月记录数；
• 429 次数；
• 5xx 次数。

这些指标比“程序能跑”更有价值。采集任务一旦跑久了，日志就是你的眼睛。

11.4 定时任务

如果要每月自动更新，可以使用 cron。

Linux crontab 示例：

0 3 1 * * cd /opt/fisheries_dynamic_obis && /opt/fisheries_dynamic_obis/.venv/bin/python run_spider.py >> logs/cron.log 2>&1

含义是每月 1 日凌晨 3 点执行一次。

如果任务复杂，可以使用 Airflow 或 Prefect。它们可以管理依赖、重试、告警和任务状态。但小项目没必要一开始就上 Airflow。先用脚本把流程跑通，再决定是否平台化。

11.5 空间网格升级

本文用 1 度经纬度网格是为了简单可解释。真实项目可以升级为：

• Geohash；
• H3；
• S2；
• EEZ 区划；
• FAO fishing area；
• 自定义渔场边界；
• PostGIS 空间 join。

比如使用 H3，可以把经纬度编码成六边形网格，更适合做空间聚合。但 H3 需要额外依赖，本文为了保持代码轻量，没有加入。

11.6 强度指标升级

当前强度指标是记录数量。后续可以扩展为：

• 记录数量；
• 去重后的采样事件数量；
• 单位面积记录数量；
• 按数据集加权后的记录数量；
• 按采样努力校正后的相对指数；
• 不同来源一致性指标。

这里要特别谨慎。越接近生态解释，就越需要领域知识。爬虫只能负责把数据可靠地搬到分析桌面上，不能把简单 count 包装成科学结论。

11.7 多物种批量采集

把 species_list 扩展即可：

species_list = [
"Gadus morhua",
"Clupea harengus",
"Thunnus thynnus",
]

批量采集时建议加上：

• 每个物种单独日志；
• 每个物种单独输出文件；
• 失败物种清单；
• 可配置的采集范围；
• 限制最大记录数。

不要一次塞几十个物种、十几年时间、全球范围，然后直接开跑。先做一个物种一个月的小样本，确认字段和数据质量，再扩大。

11.8 与 Scrapy 集成

如果你更喜欢 Scrapy，可以把请求层改成 Scrapy Spider：

import scrapy

class ObisOccurrenceSpider(scrapy.Spider):
name = "obis_occurrence"

custom_settings = {
"DOWNLOAD_DELAY": 1.0,
"CONCURRENT_REQUESTS": 2,
"RETRY_TIMES": 2,
"FEED_EXPORT_ENCODING": "utf-8",
}

def start_requests(self):
url = "https://api.obis.org/v3/occurrence"
params = {
"scientificname": "Gadus morhua",
"startdate": "2020-01-01",
"enddate": "2020-01-31",
"geometry": "POLYGON((-75 35,-75 55,-45 55,-45 35,-75 35))",
"size": 500,
"offset": 0,
}
yield scrapy.FormRequest(
url=url,
method="GET",
formdata=params,
callback=self.parse,
)

def parse(self, response):
payload = response.json()
for row in payload.get("results", []):
yield {
"id": row.get("id"),
"scientificName": row.get("scientificName"),
"eventDate": row.get("eventDate"),
"decimalLongitude": row.get("decimalLongitude"),
"decimalLatitude": row.get("decimalLatitude"),
}

不过这段只是示意。对于 GET 参数，Scrapy 中更常见的写法是自己拼 query string 或使用 urllib.parse.urlencode。如果项目已经用 requests 跑得很好，没必要为了“更像爬虫”而换 Scrapy。

1️⃣2️⃣ 总结与延伸阅读

本文完成了一件比较完整的事情：把动态渔业资源分布图采集拆成了可落地的工程流程。

我们从需求出发，明确了目标字段：

• 物种/资源名；
• 海域；
• 时间；
• 强度指标；
• 来源。

然后选择公开 API 作为采集入口，没有去抓截图，也没有去模拟复杂页面操作。接着，我们把时间轴拆成月度窗口，把空间范围写成 WKT polygon，通过 requests Session 稳定请求 occurrence 数据。解析层负责字段映射和容错，存储层用 SQLite 保存原始记录和聚合记录，导出层生成 CSV、GeoJSON，最后用 Folium 做了一个可拖动时间轴的 HTML 地图。

这套流程的重点不是某一个接口，而是方法：

动态地图页面
→ 找图层接口
→ 拆时间轴
→ 分页采集
→ 解析 JSON / GeoJSON / MVT
→ 清洗坐标和时间
→ 空间聚合
→ 指标定义
→ 数据库存储
→ 可视化导出

在我看来，爬虫真正有意思的地方不在“把数据抓下来”这一步，而在于抓下来之后能不能解释清楚：字段是什么，指标怎么算，数据有什么局限，结果能不能复现，出了问题能不能排查。一个只会跑的脚本，很快会变成黑盒；一个结构清楚的采集项目，才有继续维护和扩展的价值。

下一步可以继续做这些扩展：

最后再提醒一句：动态地图采集不是越快越好，尤其是公共科学数据平台。慢一点、稳一点、可复现一点，往往比短时间抓很多数据更专业。爬虫工程的成熟，不是体现在并发数字上，而是体现在边界感、数据质量和长期维护能力上。

🌟 文末

好啦～以上就是本期的全部内容啦！如果你在实践过程中遇到任何疑问，欢迎在评论区留言交流，我看到都会尽量回复～咱们下期见！

小伙伴们在批阅的过程中，如果觉得文章不错，欢迎点赞、收藏、关注哦～ 三连就是对我写作道路上最好的鼓励与支持！ ❤️🔥

✅ 专栏持续更新中｜建议收藏 + 订阅

墙裂推荐订阅专栏 👉 《Python爬虫实战》，本专栏秉承着以“入门 → 进阶 → 工程化 → 项目落地”的路线持续更新，争取让每一期内容都做到：

✅ 讲得清楚（原理）｜✅ 跑得起来（代码）｜✅ 用得上（场景）｜✅ 扛得住（工程化）

📣 想系统提升的小伙伴：强烈建议先订阅专栏 《Python爬虫实战》，再按目录大纲顺序学习，效率十倍上升～

✅ 互动征集

想让我把【某站点/某反爬/某验证码/某分布式方案】等写成某期实战？

评论区留言告诉我你的需求，我会优先安排实现(更新)哒~

⭐️ 若喜欢我，就请关注我叭～（更新不迷路） ⭐️ 若对你有用，就请点赞支持一下叭～（给我一点点动力） ⭐️ 若有疑问，就请评论留言告诉我叭～（我会补坑 & 更新迭代）

✅ 免责声明

本文爬虫思路、相关技术和代码仅用于学习参考，对阅读本文后的进行爬虫行为的用户本作者不承担任何法律责任。

使用或者参考本项目即表示您已阅读并同意以下条款：

• 合法使用： 不得将本项目用于任何违法、违规或侵犯他人权益的行为，包括但不限于网络攻击、诈骗、绕过身份验证、未经授权的数据抓取等。
• 风险自负： 任何因使用本项目而产生的法律责任、技术风险或经济损失，由使用者自行承担，项目作者不承担任何形式的责任。
• 禁止滥用： 不得将本项目用于违法牟利、黑产活动或其他不当商业用途。
• 使用或者参考本项目即视为同意上述条款,即 “谁使用，谁负责” 。如不同意，请立即停止使用并删除本项目。！！！