Python 实战：监听 JSON / GraphQL 接口，构建交互式卫星任务时间线

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

全文目录：

• 🌟 开篇语
• 0️⃣ 前言（Preface）
• 1️⃣ 摘要（Abstract）
• 2️⃣ 背景与需求（Why）
• • 2.1 为什么要采集卫星任务时间线
  • • 1. 信息聚合
    • 2. 数据分析
    • 3. 自动化更新
    • 4. 可视化展示
  • 2.2 目标字段
  • 2.3 为什么任务阶段需要二次计算
• 3️⃣ 合规与注意事项
• • 3.1 robots.txt 的基本作用
  • 3.2 控制请求频率
  • 3.3 不采集敏感信息
  • 3.4 不绕过付费或登录限制
• 4️⃣ 技术选型与整体流程（What / How）
• • 4.1 静态页面、动态页面和 API 的区别
  • • 静态 HTML
    • 动态渲染页面
    • JSON / GraphQL API
  • 4.2 整体流程
  • 4.3 为什么选择 Requests
  • 4.4 为什么不把 BeautifulSoup 作为主解析器
  • 4.5 为什么保留 Playwright
• 5️⃣ 环境准备与依赖安装
• • 5.1 Python 版本
  • 5.2 创建虚拟环境
  • • Windows PowerShell
    • Windows CMD
    • macOS / Linux
  • 5.3 requirements.txt
  • 5.4 推荐项目结构
  • 5.5 app/**init**.py
• 6️⃣ 核心实现：请求层（Fetcher）
• • 6.1 配置文件 app/config.py
  • 6.2 数据模型 app/models.py
  • 6.3 请求器 app/fetcher.py
  • 6.4 Headers 应该怎么写
  • • User-Agent
    • Referer
    • Cookie
  • 6.5 Timeout 为什么必须设置
  • 6.6 重试与退避策略
• 7️⃣ 核心实现：解析层（Parser）
• • 7.1 app/parser.py
  • 7.2 列表页如何拿详情链接
  • 7.3 缺失字段的容错策略
  • • 第一层：安全访问
    • 第二层：字段回退
    • 第三层：默认文本
    • 第四层：单条失败不终止整页
  • 7.4 HTML、XPath 和 CSS 解析应该放在哪里
• 8️⃣ 数据存储与导出（Storage）
• • 8.1 字段映射表
  • 8.2 app/storage.py
  • 8.3 去重策略
  • • 1. `source_id`
    • 2. `source_url`
    • 3. `content_hash`
• 9️⃣ 交互式时间线页面
• • 9.1 app/renderer.py
• 🔟 主程序、运行方式与结果展示
• • 10.1 app/main.py
  • 10.2 启动命令
  • 10.3 输出位置
  • 10.4 打开时间线
  • 10.5 示例结果
• 1️⃣0️⃣ 常见问题与排错
• • 10.1 返回 403 怎么办
  • 10.2 返回 429 怎么办
  • 10.3 HTML 抓到空壳怎么办
  • 10.4 JSON 解析报错
  • 10.5 解析报错怎么办
  • 10.6 字段结构变化
  • 10.7 编码或乱码
  • 10.8 日期排序错误
  • 10.9 为什么未来任务日期会变化
• 1️⃣1️⃣ 监听 JSON / GraphQL 接口
• • 11.1 app/inspect_network.py
  • 11.2 运行监听器
  • 11.3 GraphQL 请求结构
  • 11.4 合法复现 GraphQL 请求
• 1️⃣2️⃣ 单元测试
• • 12.1 tests/test_parser.py
• 1️⃣3️⃣ 进阶优化
• • 13.1 并发采集
  • 13.2 asyncio
  • 13.3 断点续跑
  • 13.4 保存任务变更历史
  • 13.5 增量更新
  • 13.6 日志与监控
  • 13.7 定时任务
  • • Linux cron
    • Windows 任务计划程序
    • Airflow
  • 13.8 Scrapy 化
  • 13.9 数据质量校验
  • 13.10 状态映射配置化
• 1️⃣4️⃣ 完整运行检查清单
• • 环境检查
  • 模块检查
  • 测试检查
  • 小规模采集
  • 文件检查
  • SQLite 检查
• 1️⃣5️⃣ 总结与延伸阅读
• • 🌟 文末
  • • ✅ 专栏持续更新中｜建议收藏 + 订阅
    • ✅ 互动征集
    • ✅ 免责声明

🌟 开篇语

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

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

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

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

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

0️⃣ 前言（Preface）

很多航天信息页面看起来像普通网页，真正的数据却不一定写在 HTML 中。浏览器打开页面后，JavaScript 往往还会继续调用 JSON 或 GraphQL 接口，再把返回结果渲染成任务卡片、日历或时间线。

本项目将使用 Python、Requests、Playwright、SQLite 和原生 JavaScript，采集公开航天任务接口中的结构化数据，最终生成一个支持搜索、筛选和排序的交互式卫星任务时间线。

读完本文后，你将能够：

我个人很喜欢“时间线”类爬虫项目。它不只是把页面上的文字复制下来，而是需要同时考虑时间字段、任务状态、数据更新和交互展示。写到最后，你会发现它更像一个小型数据产品，而不是一段孤立的爬虫脚本。

1️⃣ 摘要（Abstract）

本文以公开的航天任务 JSON 接口为数据源，使用 Requests 完成分页采集，使用 Playwright 演示 JSON 与 GraphQL 接口监听，再经过字段解析、状态归一化、SQLite 去重存储和 HTML 渲染，生成可离线打开的交互式卫星任务时间线。

本项目重点解决以下问题：

• 如何识别网页背后的真实数据接口，而不是反复解析不稳定的 HTML。
• 如何将不同接口状态转换为“计划阶段、发射窗口、执行中、发射完成、发射失败、已取消”等统一阶段。
• 如何保证程序在接口超时、返回 429、字段缺失和重复运行时仍然能够稳定工作。
• 如何把爬虫结果从“一个 CSV 文件”升级成可以直接浏览和筛选的时间线页面。

本文选择 Launch Library 2 的公开 JSON 接口作为主数据源。它返回结构化任务数据，并提供分页、排序和日期筛选能力，很适合作为接口型爬虫练习对象。

需要特别说明的是，本项目关注的是公开航天任务信息的技术采集与整理，不采集个人隐私数据，不涉及登录绕过、访问控制规避或高频请求。

2️⃣ 背景与需求（Why）

2.1 为什么要采集卫星任务时间线

公开航天任务信息通常分散在新闻页面、机构公告、发射日历、任务详情页和第三方数据库中。对于普通阅读来说，逐页查看没有问题；但当我们希望进行数据分析或持续跟踪时，手工整理就会暴露出明显缺点：

• 同一任务的日期可能多次推迟。
• 不同网站对状态的命名不一致。
• 有的页面只展示任务简称，没有机构、轨道或任务类型。
• 历史任务和未来任务混在一起，不方便统计。
• 每次更新都要重新浏览大量页面。
• 数据无法直接用于图表、数据库查询或自动提醒。

因此，我们希望建立一个自动化管道：

公开任务接口
↓
分页采集原始 JSON
↓
字段解析与状态标准化
↓
SQLite 去重更新
↓
CSV / JSON 导出
↓
交互式时间线展示

这个项目至少可以服务于以下场景：

1. 信息聚合

将多个时间段内的任务整理到统一数据表中，避免重复浏览不同页面。

2. 数据分析

可以按机构、年份、轨道类型、任务状态统计任务数量，也可以研究任务日期变化。

3. 自动化更新

通过定时任务每天或每周运行一次，自动更新未来任务状态。

4. 可视化展示

将标准化后的数据渲染为时间线，并支持按任务名、机构、状态和阶段筛选。

2.2 目标字段

用户要求的核心字段为：

为了让数据更容易核验和扩展，项目还会保存以下辅助字段：

核心字段保持精简，辅助字段负责追溯。实际项目里，我通常不建议只保存眼前要展示的五列，因为将来一旦需要核对异常数据，如果没有原始标识、来源地址和更新时间，排错会非常痛苦。

2.3 为什么任务阶段需要二次计算

接口通常会提供类似以下状态：

Go for Launch
To Be Confirmed
To Be Determined
Launch Successful
Launch Failure
On Hold
Launch in Flight
Cancelled

这些状态可以直接展示，但不完全等同于适合时间线使用的“任务阶段”。

例如：

• 一个状态为“To Be Confirmed”、日期在三个月后的任务，应归入“计划阶段”。
• 一个状态为“Go for Launch”、距离发射不足 24 小时的任务，可以归入“发射窗口”。
• 一个状态为“Launch Successful”的任务，应归入“发射完成”。
• 一个已经开始但尚未完成的任务，可以归入“任务执行中”。

因此，status 保存接口原始含义，task_stage 保存项目内部的统一分类。两者不能简单地当成同一个字段。

3️⃣ 合规与注意事项

爬虫能不能实现，与是否应该采集，是两个不同的问题。技术实现之前，应先判断数据是否公开、访问是否合理、频率是否会给服务端造成负担。

3.1 robots.txt 的基本作用

robots.txt 通常位于网站根目录，例如：

https://example.com/robots.txt

它用于声明不同 User-Agent 是否允许访问某些路径。常见内容如下：

User-agent: *
Disallow: /private/
Allow: /public/
Crawl-delay: 2

需要注意：

可以用 Python 做一个基础检查：

from urllib.parse import urljoin
from urllib.robotparser import RobotFileParser

def check_robots(target_url: str, user_agent: str) –> bool:
robots_url = urljoin(target_url, "/robots.txt")

parser = RobotFileParser()
parser.set_url(robots_url)

try:
parser.read()
except OSError as exc:
print(f"robots.txt 读取失败：{exc}")
return False

allowed = parser.can_fetch(user_agent, target_url)
print(f"robots.txt: {robots_url}")
print(f"允许访问: {allowed}")
return allowed

这段代码只能作为辅助判断。正式运行前，仍需人工阅读目标站点的接口说明和服务条款。

3.2 控制请求频率

本文项目默认采用串行分页采集，并在页面请求之间加入随机等待：

time.sleep(random.uniform(0.8, 1.6))

这比几十个线程同时请求更稳妥。

对于公开数据接口，建议遵守以下原则：

• 不进行攻击式并发。
• 不使用无限重试。
• 遇到 429 时尊重 Retry-After。
• 对 500、502、503、504 等临时错误进行有限次数退避。
• 使用本地缓存，避免调试时重复请求相同页面。
• 只采集业务需要的数据范围。
• 不为了“跑得快”而持续更换出口地址规避限流。

一个只有几页数据的接口，没有必要开几十个线程。请求速度快几秒，对最终项目价值几乎没有影响，却会显著增加被限流和数据不完整的概率。

3.3 不采集敏感信息

本项目只处理公开任务数据，例如任务名称、机构、日期、状态和发射地点。

不应采集或保存：

• 个人身份信息。
• 未公开联系方式。
• 登录后的私人数据。
• 与公开任务无关的账户标识。
• 页面中意外暴露的令牌、Cookie 或密钥。
• 明确标记为内部使用的数据。

在监听浏览器网络请求时，尤其要避免把以下内容写入日志：

Authorization
Cookie
Set-Cookie
X-API-Key
access_token
refresh_token
password

本文的网络监听程序只保存请求地址、方法、GraphQL 操作名、变量和公开 JSON 响应，并对常见敏感请求头进行排除。

3.4 不绕过付费或登录限制

如果数据只能在付费订阅、账号登录或访问控制后查看，应按照站点提供的正常方式使用。

中性的处理方式是：

• 使用站点公开 API。
• 申请正式 API Key。
• 使用自己有权访问的数据。
• 联系数据提供方获得授权。
• 在无法确认授权时停止采集。

本文不涉及验证码绕过、登录破解、签名伪造、付费墙规避或访问控制绕过。

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

4.1 静态页面、动态页面和 API 的区别

静态 HTML

请求页面后，目标数据直接出现在响应 HTML 中。

适合工具：

requests + BeautifulSoup
requests + lxml
Scrapy Selector

动态渲染页面

初始 HTML 只有页面骨架，数据需要 JavaScript 执行后才能出现。

适合工具：

Playwright
浏览器开发者工具
直接查找网页背后的接口

JSON / GraphQL API

页面通过 XHR 或 Fetch 请求获取结构化数据。

适合工具：

requests
httpx
aiohttp
Scrapy
Playwright 网络监听

本项目属于 API 型采集。

Playwright 主要负责发现和验证接口，正式批量采集则使用 Requests。原因很简单：浏览器适合观察动态行为，但如果已经找到公开、稳定的 JSON 接口，再让浏览器逐页加载会增加资源开销和故障点。

4.2 整体流程

┌─────────────────────────────┐
│ 浏览器开发者工具 / Playwright │
│ 发现 JSON 或 GraphQL 接口 │
└──────────────┬──────────────┘
↓
┌─────────────────────────────┐
│ Fetcher 请求层 │
│ Session / Header / Timeout │
│ Retry / Backoff / Cache │
└──────────────┬──────────────┘
↓
┌─────────────────────────────┐
│ Parser 解析层 │
│ 嵌套字段提取 / 日期解析 │
│ 状态归一 / 缺失字段容错 │
└──────────────┬──────────────┘
↓
┌─────────────────────────────┐
│ Storage 存储层 │
│ SQLite Upsert / 内容摘要 │
│ CSV / JSON 导出 │
└──────────────┬──────────────┘
↓
┌─────────────────────────────┐
│ Renderer 展示层 │
│ 搜索 / 筛选 / 时间线 │
│ 独立 HTML 页面 │
└─────────────────────────────┘

4.3 为什么选择 Requests

Requests 适合本项目的原因：

• API 是普通 HTTPS GET 请求。
• 数据以 JSON 返回，不需要执行 JavaScript。
• 可以复用连接。
• 可以通过 HTTPAdapter 配置重试。
• 代码容易阅读和部署。
• 与 SQLite、CSV、JSON 等标准库组合简单。

4.4 为什么不把 BeautifulSoup 作为主解析器

BeautifulSoup 用于 HTML 解析，而本项目目标数据来自 JSON。

如果已经拿到如下结构：

{
"results": [
{
"id": "xxx",
"name": "Example Launch",
"status": {
"name": "Launch Successful"
}
}
]
}

继续把 JSON 转成字符串再交给 BeautifulSoup 没有意义。此时直接使用字典访问和容错函数更自然。

4.5 为什么保留 Playwright

接口发现阶段可能遇到：

• 接口地址由 JavaScript 动态生成。
• GraphQL 请求使用 POST。
• 请求参数藏在变量中。
• 页面调用多个接口，不知道哪个返回任务数据。
• 接口需要浏览器正常建立的会话 Cookie。

Playwright 可以监听页面的请求和响应，帮助我们确认：

请求方法
接口地址
Content-Type
GraphQL operationName
GraphQL variables
JSON 响应结构

确认接口后，应优先判断能否使用公开 API 直接请求，而不是长期依赖浏览器自动化。

5️⃣ 环境准备与依赖安装

5.1 Python 版本

推荐使用：

Python 3.11 或 Python 3.12

代码使用了类型注解、dataclass、pathlib 和现代 SQLite 写法。

检查版本：

python –version

Windows 某些环境需要使用：

py –version

5.2 创建虚拟环境

Windows PowerShell

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

Windows CMD

python -m venv .venv
.venv\\Scripts\\activate.bat

macOS / Linux

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

5.3 requirements.txt

新建 requirements.txt：

requests>=2.32,<3
urllib3>=2.2,<3
python-dateutil>=2.9,<3
playwright>=1.48,<2
pytest>=8,<10

安装依赖：

python -m pip install –upgrade pip
pip install -r requirements.txt

安装 Playwright 浏览器：

playwright install chromium

如果服务器只运行正式采集，不运行网络监听脚本，可以不安装 Playwright 和 Chromium。

5.4 推荐项目结构

satellite_timeline/
├─ app/
│ ├─ __init__.py
│ ├─ config.py
│ ├─ models.py
│ ├─ fetcher.py
│ ├─ parser.py
│ ├─ storage.py
│ ├─ renderer.py
│ ├─ main.py
│ └─ inspect_network.py
├─ data/
│ ├─ raw/
│ ├─ export/
│ └─ missions.db
├─ output/
│ └─ timeline.html
├─ logs/
├─ tests/
│ └─ test_parser.py
├─ .env.example
├─ requirements.txt
└─ README.md

创建目录：

mkdir satellite_timeline
cd satellite_timeline

mkdir app data output logs tests
mkdir data/raw data/export

Windows PowerShell 也可以使用：

New-Item –ItemType Directory app, data, output, logs, tests
New-Item –ItemType Directory data/raw, data/export

5.5 app/init.py

"""Satellite mission timeline package."""

__version__ = "1.0.0"

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

请求层不负责理解“任务名”或“任务阶段”，只负责稳定地拿到合法 JSON。

它需要处理：

• Session 复用。
• User-Agent。
• Referer。
• 连接和读取超时。
• 自动重试。
• 指数退避。
• 429 与 Retry-After。
• JSON 类型检查。
• 本地缓存。
• 分页。
• 异常信息。

6.1 配置文件 app/config.py

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path

ROOT_DIR = Path(__file__).resolve().parents[1]
DATA_DIR = ROOT_DIR / "data"
RAW_DIR = DATA_DIR / "raw"
EXPORT_DIR = DATA_DIR / "export"
OUTPUT_DIR = ROOT_DIR / "output"
LOG_DIR = ROOT_DIR / "logs"

@dataclass(frozen=True, slots=True)
class Settings:
api_base_url: str = os.getenv(
"SATELLITE_API_BASE_URL",
"https://ll.thespacedevs.com/2.3.0/launches/",
)

user_agent: str = os.getenv(
"CRAWLER_USER_AGENT",
"SatelliteTimelineResearch/1.0 "
"(educational project; replace-with-your-contact)",
)

referer: str = os.getenv(
"CRAWLER_REFERER",
"https://thespacedevs.com/llapi",
)

connect_timeout: float = float(
os.getenv("CONNECT_TIMEOUT", "5")
)

read_timeout: float = float(
os.getenv("READ_TIMEOUT", "30")
)

max_retries: int = int(
os.getenv("MAX_RETRIES", "4")
)

backoff_factor: float = float(
os.getenv("BACKOFF_FACTOR", "0.8")
)

request_interval_min: float = float(
os.getenv("REQUEST_INTERVAL_MIN", "0.8")
)

request_interval_max: float = float(
os.getenv("REQUEST_INTERVAL_MAX", "1.6")
)

cache_ttl_seconds: int = int(
os.getenv("CACHE_TTL_SECONDS", "21600")
)

database_path: Path = DATA_DIR / "missions.db"
csv_path: Path = EXPORT_DIR / "missions.csv"
json_path: Path = EXPORT_DIR / "missions.json"
html_path: Path = OUTPUT_DIR / "timeline.html"

def ensure_directories(self) –> None:
for directory in (
DATA_DIR,
RAW_DIR,
EXPORT_DIR,
OUTPUT_DIR,
LOG_DIR,
):
directory.mkdir(parents=True, exist_ok=True)

.env.example 可以写成：

SATELLITE_API_BASE_URL=https://ll.thespacedevs.com/2.3.0/launches/
CRAWLER_USER_AGENT=SatelliteTimelineResearch/1.0 (contact=your-email@example.com)
CRAWLER_REFERER=https://thespacedevs.com/llapi

CONNECT_TIMEOUT=5
READ_TIMEOUT=30
MAX_RETRIES=4
BACKOFF_FACTOR=0.8
REQUEST_INTERVAL_MIN=0.8
REQUEST_INTERVAL_MAX=1.6
CACHE_TTL_SECONDS=21600

本文没有强制引入 python-dotenv，因此环境变量可直接通过操作系统设置。需要自动读取 .env 时，再加入 python-dotenv 即可。

6.2 数据模型 app/models.py

from __future__ import annotations

from dataclasses import asdict, dataclass

@dataclass(slots=True)
class MissionRecord:
source_id: str
task_name: str
organization: str
scheduled_at: str | None
status: str
task_stage: str

launch_name: str
mission_type: str
orbit: str
location: str
detail: str

source_url: str
last_updated: str | None
content_hash: str
collected_at: str

def to_dict(self) –> dict[str, str | None]:
return asdict(self)

这里不直接保存原始 JSON，是因为结构化表更方便查询。不过请求层仍会将原始响应缓存到 data/raw/，遇到解析问题时可以回看。

6.3 请求器 app/fetcher.py

from __future__ import annotations

import hashlib
import json
import logging
import random
import time
from pathlib import Path
from typing import Any, Iterator
from urllib.parse import urlparse

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

from app.config import RAW_DIR, Settings

logger = logging.getLogger(__name__)

class FetchError(RuntimeError):
"""接口请求或响应校验失败。"""

class JSONFetcher:
def __init__(self, settings: Settings) –> None:
self.settings = settings
self.session = self._build_session()

def _build_session(self) –> requests.Session:
session = requests.Session()

session.headers.update(
{
"User-Agent": self.settings.user_agent,
"Accept": "application/json, text/plain;q=0.9, */*;q=0.1",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.7",
"Accept-Encoding": "gzip, deflate",
"Referer": self.settings.referer,
"Connection": "keep-alive",
}
)

retry = Retry(
total=self.settings.max_retries,
connect=self.settings.max_retries,
read=self.settings.max_retries,
status=self.settings.max_retries,
other=0,
allowed_methods=frozenset({"GET", "HEAD", "OPTIONS"}),
status_forcelist=(429, 500, 502, 503, 504),
backoff_factor=self.settings.backoff_factor,
respect_retry_after_header=True,
raise_on_status=False,
)

adapter = HTTPAdapter(
max_retries=retry,
pool_connections=4,
pool_maxsize=4,
)

session.mount("https://", adapter)
session.mount("http://", adapter)
return session

@staticmethod
def _cache_key(
url: str,
params: dict[str, Any] | None,
) –> str:
payload = {
"url": url,
"params": params or {},
}

serialized = json.dumps(
payload,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)

return hashlib.sha256(
serialized.encode("utf-8")
).hexdigest()

def _cache_path(
self,
url: str,
params: dict[str, Any] | None,
) –> Path:
return RAW_DIR / f"{self._cache_key(url, params)}.json"

def _read_cache(
self,
cache_path: Path,
) –> dict[str, Any] | None:
if not cache_path.exists():
return None

age = time.time() – cache_path.stat().st_mtime

if age > self.settings.cache_ttl_seconds:
return None

try:
with cache_path.open(
"r",
encoding="utf-8",
) as file:
payload = json.load(file)
except (OSError, json.JSONDecodeError) as exc:
logger.warning(
"缓存读取失败，将重新请求：%s",
exc,
)
return None

logger.info(
"命中缓存：%s",
cache_path.name,
)
return payload

@staticmethod
def _write_cache(
cache_path: Path,
payload: dict[str, Any],
) –> None:
temporary_path = cache_path.with_suffix(".tmp")

with temporary_path.open(
"w",
encoding="utf-8",
) as file:
json.dump(
payload,
file,
ensure_ascii=False,
indent=2,
)

temporary_path.replace(cache_path)

def _polite_sleep(self) –> None:
delay = random.uniform(
self.settings.request_interval_min,
self.settings.request_interval_max,
)
time.sleep(delay)

def get_json(
self,
url: str,
params: dict[str, Any] | None = None,
use_cache: bool = True,
) –> dict[str, Any]:
cache_path = self._cache_path(url, params)

if use_cache:
cached = self._read_cache(cache_path)
if cached is not None:
return cached

logger.info(
"请求接口：%s params=%s",
url,
params,
)

try:
response = self.session.get(
url,
params=params,
timeout=(
self.settings.connect_timeout,
self.settings.read_timeout,
),
allow_redirects=True,
)
except requests.Timeout as exc:
raise FetchError(
f"请求超时：{url}"
) from exc
except requests.ConnectionError as exc:
raise FetchError(
f"连接失败：{url}"
) from exc
except requests.RequestException as exc:
raise FetchError(
f"请求异常：{url}，原因：{exc}"
) from exc

if response.status_code == 429:
retry_after = response.headers.get(
"Retry-After",
"未提供",
)
raise FetchError(
"接口返回 429，请降低请求频率。"
f"Retry-After={retry_after}"
)

try:
response.raise_for_status()
except requests.HTTPError as exc:
body_preview = response.text[:300].replace(
"\\n",
" ",
)
raise FetchError(
f"HTTP {response.status_code}: "
f"{response.url}; body={body_preview}"
) from exc

content_type = response.headers.get(
"Content-Type",
"",
).lower()

if "json" not in content_type:
body_preview = response.text[:300].replace(
"\\n",
" ",
)
raise FetchError(
"响应并非 JSON："
f"Content-Type={content_type}; "
f"body={body_preview}"
)

try:
payload = response.json()
except requests.JSONDecodeError as exc:
raise FetchError(
f"JSON 解码失败：{response.url}"
) from exc

if not isinstance(payload, dict):
raise FetchError(
"接口顶层结构不是 JSON 对象。"
)

self._write_cache(cache_path, payload)
self._polite_sleep()
return payload

def iter_pages(
self,
initial_url: str,
params: dict[str, Any],
max_pages: int | None = None,
use_cache: bool = True,
) –> Iterator[dict[str, Any]]:
original_host = urlparse(initial_url).hostname

next_url: str | None = initial_url
next_params: dict[str, Any] | None = params
page_number = 0

while next_url:
if max_pages is not None and page_number >= max_pages:
logger.info(
"达到最大页数限制：%s",
max_pages,
)
break

current_host = urlparse(next_url).hostname

if current_host != original_host:
raise FetchError(
"分页链接跳转到了非预期域名："
f"{next_url}"
)

payload = self.get_json(
next_url,
params=next_params,
use_cache=use_cache,
)

page_number += 1

result_count = len(
payload.get("results") or []
)

logger.info(
"完成第 %s 页，本页 %s 条",
page_number,
result_count,
)

yield payload

candidate = payload.get("next")

if candidate is not None and not isinstance(
candidate,
str,
):
raise FetchError(
"分页 next 字段不是字符串或 null。"
)

next_url = candidate

# next 通常已包含完整查询参数。
next_params = None

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

def __enter__(self) –> "JSONFetcher":
return self

def __exit__(
self,
exc_type: object,
exc_value: object,
traceback: object,
) –> None:
self.close()

6.4 Headers 应该怎么写

请求头不是越多越好。

本项目使用：

{
"User-Agent": "…",
"Accept": "application/json, …",
"Accept-Language": "…",
"Accept-Encoding": "gzip, deflate",
"Referer": "…",
}

User-Agent

推荐使用能够说明项目用途的 User-Agent，而不是随意伪装成某个浏览器：

SatelliteTimelineResearch/1.0

正式长期运行时，可以加入有效的项目主页或联系地址。

Referer

有些服务会检查来源页面，但公开 API 往往不强制要求 Referer。本文仍然展示该字段，是为了说明请求头配置方法。

不要伪造与实际业务完全无关的 Referer。

Cookie

当前公开接口不需要登录 Cookie，因此代码不手工写 Cookie。

requests.Session() 会自动保存服务端正常设置的 Cookie。如果站点需要合法登录，应使用自己的授权会话，并确认自动化访问符合规则。

6.5 Timeout 为什么必须设置

Requests 在没有指定 timeout 时，可能长时间等待。

本文设置：

timeout=(5, 30)

表示：

连接超时：5 秒
读取超时：30 秒

它不是整个下载过程绝对只能持续 30 秒，而是对连接和网络读取等待进行限制。

6.6 重试与退避策略

代码只对以下状态进行自动重试：

(429, 500, 502, 503, 504)

原因是这些错误通常具有临时性。

不建议对所有 4xx 状态重试。例如：

• 400 通常表示参数错误。
• 401 表示未授权。
• 403 表示服务器拒绝访问。
• 404 表示资源不存在。

参数写错后连续重试十次，不会让参数突然变正确。

指数退避的基本思想是：

第一次失败：较短等待
第二次失败：增加等待
第三次失败：进一步增加等待

这样可以避免服务端短暂异常时，客户端立即发起密集重试。

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

解析层负责把接口的嵌套结构转换成统一记录。

典型 JSON 结构可能是：

{
"id": "任务ID",
"name": "火箭名称 | 任务名称",
"net": "2026-06-01T12:00:00Z",
"status": {
"name": "Go for Launch",
"abbrev": "Go"
},
"launch_service_provider": {
"name": "Example Agency"
},
"mission": {
"name": "Example Satellite",
"type": "Earth Science",
"description": "Mission description",
"orbit": {
"name": "Low Earth Orbit"
},
"agencies": []
}
}

字段并不保证全部存在，因此不能连续写出：

item["mission"]["orbit"]["name"]

任何一级缺失都会触发：

KeyError
TypeError

正确做法是封装安全访问函数。

7.1 app/parser.py

from __future__ import annotations

import hashlib
import json
import re
from datetime import datetime, timezone
from typing import Any

from dateutil.parser import isoparse

from app.models import MissionRecord

WHITESPACE_PATTERN = re.compile(r"\\s+")

def normalize_text(
value: Any,
default: str = "",
) –> str:
if value is None:
return default

text = str(value).strip()

if not text:
return default

return WHITESPACE_PATTERN.sub(" ", text)

def safe_get(
data: Any,
*keys: str,
default: Any = None,
) –> Any:
current = data

for key in keys:
if not isinstance(current, dict):
return default

if key not in current:
return default

current = current[key]

return default if current is None else current

def parse_datetime(
value: Any,
) –> datetime | None:
if not isinstance(value, str):
return None

value = value.strip()

if not value:
return None

try:
parsed = isoparse(value)
except (TypeError, ValueError, OverflowError):
return None

if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)

return parsed.astimezone(timezone.utc)

def normalize_datetime_string(
value: Any,
) –> str | None:
parsed = parse_datetime(value)

if parsed is None:
return None

return (
parsed.isoformat(timespec="seconds")
.replace("+00:00", "Z")
)

def join_agency_names(
mission: dict[str, Any],
) –> str:
agencies = mission.get("agencies")

if not isinstance(agencies, list):
return ""

names: list[str] = []

for agency in agencies:
if not isinstance(agency, dict):
continue

name = normalize_text(
agency.get("name"),
)

if name and name not in names:
names.append(name)

return " / ".join(names)

def derive_task_stage(
status_name: str,
status_abbrev: str,
scheduled_at: datetime | None,
now: datetime,
) –> str:
status_text = (
f"{status_name} {status_abbrev}"
).casefold()

if any(
keyword in status_text
for keyword in (
"cancelled",
"canceled",
"cancel",
)
):
return "已取消"

if any(
keyword in status_text
for keyword in (
"partial failure",
"partial success",
)
):
return "部分完成"

if any(
keyword in status_text
for keyword in (
"launch failure",
"failed",
"failure",
)
):
return "发射失败"

if any(
keyword in status_text
for keyword in (
"launch successful",
"success",
)
):
return "发射完成"

if any(
keyword in status_text
for keyword in (
"in flight",
"flight in progress",
"mission active",
)
):
return "任务执行中"

if any(
keyword in status_text
for keyword in (
"hold",
"paused",
"postponed",
)
):
return "暂停或延期"

if scheduled_at is None:
return "日期待确认"

seconds_to_launch = (
scheduled_at – now
).total_seconds()

if seconds_to_launch > 24 * 3600:
return "计划阶段"

if –6 * 3600 <= seconds_to_launch <= 24 * 3600:
return "发射窗口"

if seconds_to_launch < –6 * 3600:
return "状态待更新"

return "状态待确认"

def make_content_hash(
values: dict[str, Any],
) –> str:
serialized = json.dumps(
values,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)

return hashlib.sha256(
serialized.encode("utf-8")
).hexdigest()

def parse_mission(
item: dict[str, Any],
now: datetime | None = None,
) –> MissionRecord:
now = now or datetime.now(timezone.utc)

source_id = normalize_text(
item.get("id"),
)

if not source_id:
raise ValueError(
"任务缺少 id，无法建立唯一标识。"
)

mission = item.get("mission")

if not isinstance(mission, dict):
mission = {}

status_object = item.get("status")

if not isinstance(status_object, dict):
status_object = {}

provider = item.get(
"launch_service_provider"
)

if not isinstance(provider, dict):
provider = {}

pad = item.get("pad")

if not isinstance(pad, dict):
pad = {}

launch_name = normalize_text(
item.get("name"),
default="未命名发射任务",
)

task_name = normalize_text(
mission.get("name"),
default=launch_name,
)

mission_agencies = join_agency_names(
mission
)

organization = (
mission_agencies
or normalize_text(provider.get("name"))
or "未知机构"
)

scheduled_datetime = parse_datetime(
item.get("net")
)

scheduled_at = (
scheduled_datetime
.isoformat(timespec="seconds")
.replace("+00:00", "Z")
if scheduled_datetime
else None
)

status_name = normalize_text(
status_object.get("name"),
)

status_abbrev = normalize_text(
status_object.get("abbrev"),
)

status = (
status_name
or status_abbrev
or "状态未知"
)

task_stage = derive_task_stage(
status_name=status_name,
status_abbrev=status_abbrev,
scheduled_at=scheduled_datetime,
now=now,
)

orbit = normalize_text(
safe_get(
mission,
"orbit",
"name",
default="",
),
default="轨道未知",
)

location = normalize_text(
safe_get(
pad,
"location",
"name",
default="",
),
)

if not location:
location = normalize_text(
pad.get("name"),
default="地点未知",
)

mission_type = normalize_text(
mission.get("type"),
default="类型未知",
)

detail = normalize_text(
mission.get("description"),
default="暂无任务简介",
)

source_url = normalize_text(
item.get("url"),
)

last_updated = normalize_datetime_string(
item.get("last_updated")
)

collected_at = (
now.astimezone(timezone.utc)
.isoformat(timespec="seconds")
.replace("+00:00", "Z")
)

hash_fields = {
"source_id": source_id,
"task_name": task_name,
"organization": organization,
"scheduled_at": scheduled_at,
"status": status,
"task_stage": task_stage,
"mission_type": mission_type,
"orbit": orbit,
"location": location,
"detail": detail,
}

content_hash = make_content_hash(
hash_fields
)

return MissionRecord(
source_id=source_id,
task_name=task_name,
organization=organization,
scheduled_at=scheduled_at,
status=status,
task_stage=task_stage,
launch_name=launch_name,
mission_type=mission_type,
orbit=orbit,
location=location,
detail=detail,
source_url=source_url,
last_updated=last_updated,
content_hash=content_hash,
collected_at=collected_at,
)

def parse_payload(
payload: dict[str, Any],
now: datetime | None = None,
) –> tuple[list[MissionRecord], list[str]]:
results = payload.get("results")

if not isinstance(results, list):
raise ValueError(
"响应缺少 results 列表。"
)

records: list[MissionRecord] = []
errors: list[str] = []

for index, item in enumerate(results):
if not isinstance(item, dict):
errors.append(
f"第 {index} 条不是 JSON 对象。"
)
continue

try:
record = parse_mission(
item,
now=now,
)
except (TypeError, ValueError) as exc:
item_id = item.get(
"id",
"unknown",
)
errors.append(
f"任务 {item_id} 解析失败：{exc}"
)
continue

records.append(record)

return records, errors

7.2 列表页如何拿详情链接

传统 HTML 爬虫通常需要：

列表页 → 提取详情链接 → 请求详情页 → 解析字段

JSON API 中同样存在这个概念。

接口列表记录可能包含：

{
"id": "abc",
"url": "https://…/launches/abc/",
"name": "Example Mission"
}

url 就相当于详情链接。

不过，本项目使用 mode=normal 时，列表响应已经包含任务、状态、机构和轨道等字段，所以不需要逐条请求详情接口。

这能把请求数量从：

1 个列表请求 + N 个详情请求

降低为：

若干分页列表请求

当某些字段只存在于详情响应时，再按需请求详情，不要无条件把每一条都展开。

7.3 缺失字段的容错策略

本文采用四层容错：

第一层：安全访问

safe_get(mission, "orbit", "name", default="")

第二层：字段回退

任务名优先使用：

mission.name

缺失时回退到：

launch.name

第三层：默认文本

例如：

未知机构
轨道未知
地点未知
暂无任务简介

第四层：单条失败不终止整页

parse_payload() 会记录错误，并继续处理其他任务。

这点很重要。真实数据里总会出现少量异常记录。如果一条记录缺少 ID 就让整个任务退出，会导致其余正常数据也无法写入。

7.4 HTML、XPath 和 CSS 解析应该放在哪里

本项目主流程使用 JSON，不需要 XPath 或 CSS。

如果将来增加一个传统详情页面，可单独增加 HTML 解析器：

from bs4 import BeautifulSoup

def parse_html_detail(html: str) –> dict[str, str]:
soup = BeautifulSoup(html, "lxml")

title_node = soup.select_one("h1.mission-title")
agency_node = soup.select_one(".agency-name")

return {
"task_name": (
title_node.get_text(
" ",
strip=True,
)
if title_node
else ""
),
"organization": (
agency_node.get_text(
" ",
strip=True,
)
if agency_node
else ""
),
}

XPath 版本可以写成：

from lxml import html

def parse_html_with_xpath(
document: str,
) –> dict[str, str]:
tree = html.fromstring(document)

title = tree.xpath(
"normalize-space(//h1[contains(@class, 'mission-title')])"
)

agency = tree.xpath(
"normalize-space(//*[contains(@class, 'agency-name')])"
)

return {
"task_name": title,
"organization": agency,
}

但不要为了使用 BeautifulSoup 或 XPath 而使用它们。解析器应服从数据格式，而不是反过来。

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

SQLite 很适合该项目：

• Python 标准库自带。
• 不需要单独安装数据库服务。
• 支持唯一约束。
• 支持事务。
• 支持 Upsert。
• 可直接用 SQL 查询。
• 数据量达到几十万条仍然可以正常使用。

8.1 字段映射表

日期统一保存为 UTC ISO 8601 字符串，避免服务器时区不同导致排序混乱。

8.2 app/storage.py

from __future__ import annotations

import csv
import json
import sqlite3
from pathlib import Path
from typing import Iterable

from app.models import MissionRecord

CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS missions (
source_id TEXT PRIMARY KEY,
task_name TEXT NOT NULL,
organization TEXT NOT NULL,
scheduled_at TEXT,
status TEXT NOT NULL,
task_stage TEXT NOT NULL,

launch_name TEXT NOT NULL,
mission_type TEXT NOT NULL,
orbit TEXT NOT NULL,
location TEXT NOT NULL,
detail TEXT NOT NULL,

source_url TEXT UNIQUE,
last_updated TEXT,
content_hash TEXT NOT NULL,
collected_at TEXT NOT NULL
);
"""

CREATE_INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_missions_scheduled_at
ON missions(scheduled_at);
"""

CREATE_ORGANIZATION_INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_missions_organization
ON missions(organization);
"""

UPSERT_SQL = """
INSERT INTO missions (
source_id,
task_name,
organization,
scheduled_at,
status,
task_stage,
launch_name,
mission_type,
orbit,
location,
detail,
source_url,
last_updated,
content_hash,
collected_at
)
VALUES (
:source_id,
:task_name,
:organization,
:scheduled_at,
:status,
:task_stage,
:launch_name,
:mission_type,
:orbit,
:location,
:detail,
:source_url,
:last_updated,
:content_hash,
:collected_at
)
ON CONFLICT(source_id) DO UPDATE SET
task_name = excluded.task_name,
organization = excluded.organization,
scheduled_at = excluded.scheduled_at,
status = excluded.status,
task_stage = excluded.task_stage,
launch_name = excluded.launch_name,
mission_type = excluded.mission_type,
orbit = excluded.orbit,
location = excluded.location,
detail = excluded.detail,
source_url = excluded.source_url,
last_updated = excluded.last_updated,
content_hash = excluded.content_hash,
collected_at = excluded.collected_at;
"""

class MissionStorage:
def __init__(
self,
database_path: Path,
) –> None:
self.database_path = database_path
self.database_path.parent.mkdir(
parents=True,
exist_ok=True,
)

def connect(self) –> sqlite3.Connection:
connection = sqlite3.connect(
self.database_path
)
connection.row_factory = sqlite3.Row
return connection

def initialize(self) –> None:
with self.connect() as connection:
connection.execute(CREATE_TABLE_SQL)
connection.execute(CREATE_INDEX_SQL)
connection.execute(
CREATE_ORGANIZATION_INDEX_SQL
)
connection.commit()

def upsert(
self,
records: Iterable[MissionRecord],
) –> int:
rows = [
record.to_dict()
for record in records
]

if not rows:
return 0

with self.connect() as connection:
before = connection.total_changes

connection.executemany(
UPSERT_SQL,
rows,
)
connection.commit()

return (
connection.total_changes
– before
)

def fetch_all(
self,
) –> list[dict[str, str | None]]:
query = """
SELECT
source_id,
task_name,
organization,
scheduled_at,
status,
task_stage,
launch_name,
mission_type,
orbit,
location,
detail,
source_url,
last_updated,
content_hash,
collected_at
FROM missions
ORDER BY
CASE
WHEN scheduled_at IS NULL THEN 1
ELSE 0
END,
scheduled_at ASC,
task_name ASC;
"""

with self.connect() as connection:
rows = connection.execute(
query
).fetchall()

return [
dict(row)
for row in rows
]

def export_json(
self,
output_path: Path,
) –> int:
rows = self.fetch_all()

output_path.parent.mkdir(
parents=True,
exist_ok=True,
)

temporary_path = output_path.with_suffix(
".tmp"
)

with temporary_path.open(
"w",
encoding="utf-8",
) as file:
json.dump(
rows,
file,
ensure_ascii=False,
indent=2,
)

temporary_path.replace(output_path)
return len(rows)

def export_csv(
self,
output_path: Path,
) –> int:
rows = self.fetch_all()

output_path.parent.mkdir(
parents=True,
exist_ok=True,
)

fieldnames = [
"source_id",
"task_name",
"organization",
"scheduled_at",
"status",
"task_stage",
"launch_name",
"mission_type",
"orbit",
"location",
"detail",
"source_url",
"last_updated",
"content_hash",
"collected_at",
]

temporary_path = output_path.with_suffix(
".tmp"
)

with temporary_path.open(
"w",
encoding="utf-8-sig",
newline="",
) as file:
writer = csv.DictWriter(
file,
fieldnames=fieldnames,
extrasaction="ignore",
)
writer.writeheader()
writer.writerows(rows)

temporary_path.replace(output_path)
return len(rows)

8.3 去重策略

项目使用三种标识：

1. source_id

作为主键：

source_id TEXT PRIMARY KEY

同一个任务重复运行时执行更新，而不是新增一行。

2. source_url

建立唯一约束：

source_url TEXT UNIQUE

防止相同详情地址被不同流程重复写入。

3. content_hash

使用核心字段计算 SHA-256：

任务名
机构
时间
状态
阶段
轨道
地点
简介

content_hash 不直接作为唯一键，因为两个不同任务可能恰好有相同名称和状态。它主要用于判断内容是否变化，以及后续建立更新历史。

一个常见错误是把任务名设成唯一键。现实中同名任务并不少见，同一个任务系列也可能使用相似名称，因此任务名不适合作为唯一标识。

9️⃣ 交互式时间线页面

项目不依赖 Vue、React 或外部 CDN，而是生成一个独立 HTML 文件。

优点：

• 可离线打开。
• 不受 CDN 失效影响。
• 部署简单。
• 可直接放到静态网站。
• 适合学习数据到页面的完整过程。

9.1 app/renderer.py

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

HTML_TEMPLATE = r"""<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta
name="viewport"
content="width=device-width, initial-scale=1.0"
>
<title>Interactive Satellite Mission Timeline</title>

<style>
:root {
–background: #f4f6f9;
–panel: #ffffff;
–text: #172033;
–muted: #687086;
–line: #d9deea;
–accent: #315efb;
–border: #e4e8f0;
–shadow: 0 10px 32px rgba(31, 42, 68, 0.08);
}

* {
box-sizing: border-box;
}

body {
margin: 0;
background: var(–background);
color: var(–text);
font-family:
Inter,
"PingFang SC",
"Microsoft YaHei",
Arial,
sans-serif;
}

.page {
width: min(1180px, calc(100% – 32px));
margin: 0 auto;
padding: 42px 0 70px;
}

.hero {
padding: 32px;
border: 1px solid var(–border);
border-radius: 22px;
background: var(–panel);
box-shadow: var(–shadow);
}

.hero h1 {
margin: 0;
font-size: clamp(26px, 4vw, 44px);
line-height: 1.2;
}

.hero p {
max-width: 760px;
margin: 14px 0 0;
color: var(–muted);
line-height: 1.8;
}

.stats {
display: grid;
grid-template-columns:
repeat(auto-fit, minmax(150px, 1fr));
gap: 12px;
margin-top: 24px;
}

.stat-card {
padding: 18px;
border: 1px solid var(–border);
border-radius: 16px;
background: #fafbfe;
}

.stat-label {
color: var(–muted);
font-size: 13px;
}

.stat-value {
margin-top: 7px;
font-size: 26px;
font-weight: 700;
}

.controls {
display: grid;
grid-template-columns:
minmax(220px, 2fr)
repeat(3, minmax(150px, 1fr));
gap: 12px;
margin: 22px 0;
padding: 18px;
border: 1px solid var(–border);
border-radius: 18px;
background: var(–panel);
}

.control label {
display: block;
margin-bottom: 7px;
color: var(–muted);
font-size: 13px;
}

.control input,
.control select {
width: 100%;
height: 42px;
padding: 0 12px;
border: 1px solid var(–border);
border-radius: 11px;
background: #fff;
color: var(–text);
outline: none;
}

.control input:focus,
.control select:focus {
border-color: var(–accent);
box-shadow:
0 0 0 3px rgba(49, 94, 251, 0.12);
}

.result-bar {
display: flex;
justify-content: space-between;
gap: 12px;
align-items: center;
margin: 18px 0;
color: var(–muted);
}

.result-bar button {
border: 1px solid var(–border);
border-radius: 10px;
padding: 9px 13px;
background: var(–panel);
cursor: pointer;
}

.timeline {
position: relative;
padding: 10px 0;
}

.timeline::before {
content: "";
position: absolute;
top: 0;
bottom: 0;
left: 174px;
width: 2px;
background: var(–line);
}

.timeline-item {
display: grid;
grid-template-columns: 150px 1fr;
gap: 48px;
position: relative;
margin-bottom: 22px;
}

.timeline-date {
padding-top: 18px;
text-align: right;
color: var(–muted);
font-size: 13px;
line-height: 1.6;
}

.timeline-dot {
position: absolute;
left: 166px;
top: 25px;
width: 18px;
height: 18px;
border: 4px solid var(–background);
border-radius: 50%;
background: var(–accent);
box-shadow: 0 0 0 2px var(–accent);
}

.mission-card {
min-width: 0;
padding: 22px;
border: 1px solid var(–border);
border-radius: 18px;
background: var(–panel);
box-shadow: var(–shadow);
}

.mission-card h2 {
margin: 0;
font-size: 20px;
overflow-wrap: anywhere;
}

.organization {
margin-top: 7px;
color: var(–muted);
}

.badges {
display: flex;
flex-wrap: wrap;
gap: 8px;
margin: 16px 0;
}

.badge {
display: inline-flex;
align-items: center;
min-height: 28px;
padding: 4px 10px;
border-radius: 999px;
background: #eef2ff;
color: #2746b5;
font-size: 12px;
}

.badge.stage {
background: #eef8f2;
color: #26704a;
}

.badge.status {
background: #fff5e8;
color: #8b5a19;
}

.mission-meta {
display: grid;
grid-template-columns:
repeat(auto-fit, minmax(170px, 1fr));
gap: 10px;
margin-top: 14px;
}

.meta-item {
padding: 11px 12px;
border-radius: 11px;
background: #f7f8fb;
}

.meta-key {
color: var(–muted);
font-size: 12px;
}

.meta-value {
margin-top: 5px;
font-size: 14px;
overflow-wrap: anywhere;
}

.description {
margin: 16px 0 0;
color: #4f586d;
line-height: 1.75;
}

.source-link {
display: inline-block;
margin-top: 15px;
color: var(–accent);
text-decoration: none;
font-size: 14px;
}

.empty-state {
padding: 55px 20px;
border: 1px dashed var(–border);
border-radius: 18px;
background: var(–panel);
color: var(–muted);
text-align: center;
}

@media (max-width: 850px) {
.controls {
grid-template-columns: 1fr 1fr;
}
}

@media (max-width: 650px) {
.page {
width: min(100% – 20px, 1180px);
padding-top: 16px;
}

.hero {
padding: 22px;
}

.controls {
grid-template-columns: 1fr;
}

.timeline::before {
left: 10px;
}

.timeline-item {
grid-template-columns: 1fr;
gap: 8px;
padding-left: 30px;
}

.timeline-date {
padding-top: 0;
text-align: left;
}

.timeline-dot {
left: 2px;
top: 7px;
}
}
</style>
</head>

<body>
<main class="page">
<section class="hero">
<h1>Interactive Satellite Mission Timeline</h1>

<p>
基于公开任务数据生成的交互式时间线。
可按任务名、机构、原始状态和标准任务阶段进行筛选。
</p>

<div class="stats">
<div class="stat-card">
<div class="stat-label">Total Records</div>
<div class="stat-value" id="totalCount">0</div>
</div>

<div class="stat-card">
<div class="stat-label">Visible Records</div>
<div class="stat-value" id="visibleCount">0</div>
</div>

<div class="stat-card">
<div class="stat-label">Organizations</div>
<div class="stat-value" id="organizationCount">0</div>
</div>

<div class="stat-card">
<div class="stat-label">Stages</div>
<div class="stat-value" id="stageCount">0</div>
</div>
</div>
</section>

<section class="controls">
<div class="control">
<label for="searchInput">Search</label>
<input
id="searchInput"
type="search"
placeholder="任务名、机构、地点、轨道……"
>
</div>

<div class="control">
<label for="organizationSelect">
Organization
</label>
<select id="organizationSelect">
<option value="">全部机构</option>
</select>
</div>

<div class="control">
<label for="statusSelect">Status</label>
<select id="statusSelect">
<option value="">全部状态</option>
</select>
</div>

<div class="control">
<label for="stageSelect">Stage</label>
<select id="stageSelect">
<option value="">全部阶段</option>
</select>
</div>
</section>

<div class="result-bar">
<span id="resultText">正在加载……</span>
<button id="resetButton" type="button">
重置筛选
</button>
</div>

<section id="timeline" class="timeline"></section>
</main>

<script>
const DATA = __MISSION_DATA__;

const searchInput =
document.getElementById("searchInput");

const organizationSelect =
document.getElementById("organizationSelect");

const statusSelect =
document.getElementById("statusSelect");

const stageSelect =
document.getElementById("stageSelect");

const resetButton =
document.getElementById("resetButton");

const timeline =
document.getElementById("timeline");

const resultText =
document.getElementById("resultText");

const totalCount =
document.getElementById("totalCount");

const visibleCount =
document.getElementById("visibleCount");

const organizationCount =
document.getElementById("organizationCount");

const stageCount =
document.getElementById("stageCount");

function uniqueSorted(values) {
return […new Set(
values.filter(Boolean)
)].sort((a, b) =>
a.localeCompare(b, "zh-CN")
);
}

function addOptions(select, values) {
values.forEach(value => {
const option =
document.createElement("option");

option.value = value;
option.textContent = value;
select.appendChild(option);
});
}

function createElement(
tag,
className,
text
) {
const node =
document.createElement(tag);

if (className) {
node.className = className;
}

if (text !== undefined && text !== null) {
node.textContent = String(text);
}

return node;
}

function formatDate(value) {
if (!value) {
return {
date: "日期待确认",
time: ""
};
}

const parsed = new Date(value);

if (Number.isNaN(parsed.getTime())) {
return {
date: value,
time: ""
};
}

return {
date: new Intl.DateTimeFormat(
"zh-CN",
{
year: "numeric",
month: "2-digit",
day: "2-digit",
timeZone: "UTC"
}
).format(parsed),

time: new Intl.DateTimeFormat(
"zh-CN",
{
hour: "2-digit",
minute: "2-digit",
hour12: false,
timeZone: "UTC"
}
).format(parsed) + " UTC"
};
}

function addMeta(
container,
key,
value
) {
const item =
createElement("div", "meta-item");

item.appendChild(
createElement(
"div",
"meta-key",
key
)
);

item.appendChild(
createElement(
"div",
"meta-value",
value || "未知"
)
);

container.appendChild(item);
}

function buildMissionCard(record) {
const item =
createElement("article", "timeline-item");

const formatted =
formatDate(record.scheduled_at);

const dateBox =
createElement("div", "timeline-date");

dateBox.appendChild(
createElement(
"div",
"",
formatted.date
)
);

dateBox.appendChild(
createElement(
"div",
"",
formatted.time
)
);

item.appendChild(dateBox);
item.appendChild(
createElement("div", "timeline-dot")
);

const card =
createElement("div", "mission-card");

card.appendChild(
createElement(
"h2",
"",
record.task_name
)
);

card.appendChild(
createElement(
"div",
"organization",
record.organization
)
);

const badges =
createElement("div", "badges");

badges.appendChild(
createElement(
"span",
"badge status",
record.status
)
);

badges.appendChild(
createElement(
"span",
"badge stage",
record.task_stage
)
);

badges.appendChild(
createElement(
"span",
"badge",
record.mission_type
)
);

card.appendChild(badges);

const meta =
createElement("div", "mission-meta");

addMeta(
meta,
"Orbit",
record.orbit
);

addMeta(
meta,
"Location",
record.location
);

addMeta(
meta,
"Launch Name",
record.launch_name
);

addMeta(
meta,
"Last Updated",
record.last_updated || "未知"
);

card.appendChild(meta);

card.appendChild(
createElement(
"p",
"description",
record.detail
)
);

if (record.source_url) {
const link =
createElement(
"a",
"source-link",
"查看原始数据"
);

link.href = record.source_url;
link.target = "_blank";
link.rel = "noopener noreferrer";

card.appendChild(link);
}

item.appendChild(card);
return item;
}

function timestamp(record) {
const value =
Date.parse(record.scheduled_at || "");

return Number.isNaN(value)
? Number.MAX_SAFE_INTEGER
: value;
}

function render() {
const keyword =
searchInput.value
.trim()
.toLocaleLowerCase("zh-CN");

const selectedOrganization =
organizationSelect.value;

const selectedStatus =
statusSelect.value;

const selectedStage =
stageSelect.value;

const filtered = DATA
.filter(record => {
const searchable = [
record.task_name,
record.organization,
record.status,
record.task_stage,
record.mission_type,
record.orbit,
record.location,
record.launch_name,
record.detail
]
.filter(Boolean)
.join(" ")
.toLocaleLowerCase("zh-CN");

const matchesKeyword =
!keyword ||
searchable.includes(keyword);

const matchesOrganization =
!selectedOrganization ||
record.organization ===
selectedOrganization;

const matchesStatus =
!selectedStatus ||
record.status === selectedStatus;

const matchesStage =
!selectedStage ||
record.task_stage === selectedStage;

return (
matchesKeyword &&
matchesOrganization &&
matchesStatus &&
matchesStage
);
})
.sort((a, b) =>
timestamp(a) – timestamp(b)
);

timeline.replaceChildren();

if (filtered.length === 0) {
timeline.appendChild(
createElement(
"div",
"empty-state",
"没有符合当前条件的任务。"
)
);
} else {
const fragment =
document.createDocumentFragment();

filtered.forEach(record => {
fragment.appendChild(
buildMissionCard(record)
);
});

timeline.appendChild(fragment);
}

visibleCount.textContent =
String(filtered.length);

resultText.textContent =
`当前显示 ${filtered.length} / ${DATA.length} 条任务`;
}

function initialize() {
const organizations =
uniqueSorted(
DATA.map(
record => record.organization
)
);

const statuses =
uniqueSorted(
DATA.map(
record => record.status
)
);

const stages =
uniqueSorted(
DATA.map(
record => record.task_stage
)
);

addOptions(
organizationSelect,
organizations
);

addOptions(
statusSelect,
statuses
);

addOptions(
stageSelect,
stages
);

totalCount.textContent =
String(DATA.length);

organizationCount.textContent =
String(organizations.length);

stageCount.textContent =
String(stages.length);

render();
}

[
searchInput,
organizationSelect,
statusSelect,
stageSelect
].forEach(element => {
element.addEventListener(
"input",
render
);

element.addEventListener(
"change",
render
);
});

resetButton.addEventListener(
"click",
() => {
searchInput.value = "";
organizationSelect.value = "";
statusSelect.value = "";
stageSelect.value = "";
render();
}
);

initialize();
</script>
</body>
</html>
"""

def render_timeline(
records: list[dict[str, Any]],
output_path: Path,
) –> None:
output_path.parent.mkdir(
parents=True,
exist_ok=True,
)

serialized = json.dumps(
records,
ensure_ascii=False,
separators=(",", ":"),
)

# 避免外部文本意外形成 </script>。
serialized = serialized.replace(
"</",
"<\\\\/",
)

document = HTML_TEMPLATE.replace(
"__MISSION_DATA__",
serialized,
)

temporary_path = output_path.with_suffix(
".tmp"
)

temporary_path.write_text(
document,
encoding="utf-8",
)

temporary_path.replace(output_path)

HTML 中没有把外部数据直接拼入 HTML 字符串，而是通过 textContent 创建节点，从而降低任务描述中异常字符破坏页面结构的风险。

🔟 主程序、运行方式与结果展示

10.1 app/main.py

from __future__ import annotations

import argparse
import logging
import sys
from datetime import date, datetime, timedelta, timezone
from pathlib import Path

from app.config import Settings
from app.fetcher import FetchError, JSONFetcher
from app.models import MissionRecord
from app.parser import parse_payload
from app.renderer import render_timeline
from app.storage import MissionStorage

def parse_date_argument(value: str) –> date:
try:
return date.fromisoformat(value)
except ValueError as exc:
raise argparse.ArgumentTypeError(
f"日期必须使用 YYYY-MM-DD：{value}"
) from exc

def build_argument_parser() –> argparse.ArgumentParser:
today = datetime.now(
timezone.utc
).date()

default_start = today – timedelta(
days=365
)

default_end = today + timedelta(
days=730
)

parser = argparse.ArgumentParser(
description=(
"采集公开航天任务数据，并生成交互式时间线。"
)
)

parser.add_argument(
"–start-date",
type=parse_date_argument,
default=default_start,
help=(
"开始日期，格式 YYYY-MM-DD；"
f"默认 {default_start}"
),
)

parser.add_argument(
"–end-date",
type=parse_date_argument,
default=default_end,
help=(
"结束日期，格式 YYYY-MM-DD；"
f"默认 {default_end}"
),
)

parser.add_argument(
"–page-size",
type=int,
default=50,
help="每页数量，范围 1–100，默认 50。",
)

parser.add_argument(
"–max-pages",
type=int,
default=None,
help="最多采集多少页，默认不额外限制。",
)

parser.add_argument(
"–max-records",
type=int,
default=None,
help="最多处理多少条任务，默认不限制。",
)

parser.add_argument(
"–no-cache",
action="store_true",
help="忽略本地缓存，重新请求接口。",
)

parser.add_argument(
"–database",
type=Path,
default=None,
help="自定义 SQLite 数据库路径。",
)

parser.add_argument(
"–verbose",
action="store_true",
help="输出详细调试日志。",
)

return parser

def validate_arguments(
args: argparse.Namespace,
) –> None:
if args.start_date > args.end_date:
raise ValueError(
"start-date 不能晚于 end-date。"
)

if not 1 <= args.page_size <= 100:
raise ValueError(
"page-size 必须位于 1–100。"
)

if (
args.max_pages is not None
and args.max_pages <= 0
):
raise ValueError(
"max-pages 必须大于 0。"
)

if (
args.max_records is not None
and args.max_records <= 0
):
raise ValueError(
"max-records 必须大于 0。"
)

def configure_logging(
verbose: bool,
) –> None:
level = (
logging.DEBUG
if verbose
else logging.INFO
)

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

def collect_records(
settings: Settings,
args: argparse.Namespace,
) –> tuple[list[MissionRecord], list[str]]:
params = {
"format": "json",
"mode": "normal",
"limit": args.page_size,
"offset": 0,
"ordering": "net",
"net__gte": (
f"{args.start_date.isoformat()}"
"T00:00:00Z"
),
"net__lte": (
f"{args.end_date.isoformat()}"
"T23:59:59Z"
),
}

records_by_id: dict[
str,
MissionRecord,
] = {}

all_errors: list[str] = []

parse_now = datetime.now(
timezone.utc
)

with JSONFetcher(settings) as fetcher:
for payload in fetcher.iter_pages(
settings.api_base_url,
params=params,
max_pages=args.max_pages,
use_cache=not args.no_cache,
):
records, errors = parse_payload(
payload,
now=parse_now,
)

all_errors.extend(errors)

for record in records:
records_by_id[
record.source_id
] = record

if (
args.max_records is not None
and len(records_by_id)
>= args.max_records
):
break

if (
args.max_records is not None
and len(records_by_id)
>= args.max_records
):
break

records = sorted(
records_by_id.values(),
key=lambda record: (
record.scheduled_at is None,
record.scheduled_at or "",
record.task_name,
),
)

return records, all_errors

def main() –> int:
parser = build_argument_parser()
args = parser.parse_args()

configure_logging(args.verbose)
logger = logging.getLogger("main")

try:
validate_arguments(args)
except ValueError as exc:
parser.error(str(exc))

settings = Settings()
settings.ensure_directories()

database_path = (
args.database
if args.database is not None
else settings.database_path
)

logger.info(
"采集日期范围：%s 至 %s",
args.start_date,
args.end_date,
)

try:
records, parse_errors = collect_records(
settings,
args,
)
except FetchError as exc:
logger.error("采集失败：%s", exc)
return 1
except KeyboardInterrupt:
logger.warning("用户中断运行。")
return 130

storage = MissionStorage(
database_path
)
storage.initialize()

changed_rows = storage.upsert(records)

csv_count = storage.export_csv(
settings.csv_path
)

json_count = storage.export_json(
settings.json_path
)

all_rows = storage.fetch_all()

render_timeline(
all_rows,
settings.html_path,
)

logger.info(
"本次解析任务：%s 条",
len(records),
)

logger.info(
"数据库受影响行数：%s",
changed_rows,
)

logger.info(
"CSV 导出：%s 条，路径：%s",
csv_count,
settings.csv_path,
)

logger.info(
"JSON 导出：%s 条，路径：%s",
json_count,
settings.json_path,
)

logger.info(
"时间线页面：%s",
settings.html_path,
)

if parse_errors:
logger.warning(
"共有 %s 条解析警告。",
len(parse_errors),
)

for error in parse_errors[:20]:
logger.warning("%s", error)

if len(parse_errors) > 20:
logger.warning(
"其余 %s 条警告未逐条显示。",
len(parse_errors) – 20,
)

return 0

if __name__ == "__main__":
sys.exit(main())

10.2 启动命令

在项目根目录运行：

python -m app.main

指定时间范围：

python -m app.main \\
–start-date 2025-01-01 \\
–end-date 2027-12-31

Windows PowerShell 可以写成一行：

python –m app.main —start-date 2025-01-01 —end–date 2027-12-31

调试时只抓两页：

python -m app.main \\
–page-size 10 \\
–max-pages 2 \\
–verbose

忽略缓存重新请求：

python -m app.main –no-cache

最多处理 100 条：

python -m app.main –max-records 100

10.3 输出位置

程序运行结束后会生成：

data/missions.db
data/export/missions.csv
data/export/missions.json
output/timeline.html

其中：

• missions.db：正式去重数据库。
• missions.csv：适合 Excel、Pandas 和人工查看。
• missions.json：适合其他程序或前端使用。
• timeline.html：交互式时间线页面。
• data/raw/*.json：接口原始响应缓存。

10.4 打开时间线

可以直接双击：

output/timeline.html

也可以启动本地静态服务器：

python -m http.server 8000

浏览器打开：

http://localhost:8000/output/timeline.html

使用本地服务器更接近正式部署环境，也方便后续增加独立 JSON 请求。

10.5 示例结果

下面仅展示输出结构，任务状态应以实际运行时接口返回结果为准。

CSV 形式类似：

source_id,task_name,organization,scheduled_at,status,task_stage
e3df2ecd-c239-472f-95e4-2b89b4f75800,Sputnik 1,Soviet Space Program,1957-10-04T19:28:34Z,Launch Successful,发射完成
f8c9f344-a6df-4f30-873a-90fe3a7840b3,Sputnik 2,Soviet Space Program,1957-11-03T02:30:00Z,Launch Successful,发射完成
535c1a09-97c8-4f96-bb64-6336d4bcb1fb,Vanguard,US Navy,1957-12-06T16:44:35Z,Launch Failure,发射失败
1b9e28d0-c531-44b0-9b37-244e62a6d3f4,Explorer 1,Army Ballistic Missile Agency,1958-02-01T03:47:56Z,Launch Successful,发射完成

1️⃣0️⃣ 常见问题与排错

10.1 返回 403 怎么办

403 表示服务器理解请求，但拒绝处理。

先检查：

不建议看到 403 后立刻堆砌几十个浏览器请求头。很多 403 与授权、路径和访问政策有关，不是加一个 sec-ch-ua 就能合理解决。

可以打印响应摘要：

print(response.status_code)
print(response.url)
print(response.headers)
print(response.text[:500])

但不要把响应中的令牌或 Cookie 贴到公开位置。

10.2 返回 429 怎么办

429 表示请求过多。

合理处理顺序：

停止继续并发
↓
读取 Retry-After
↓
延长请求间隔
↓
减少采集范围
↓
启用缓存
↓
降低定时任务频率

不要使用代理池去规避对方的频率限制。代理适合合法的网络出口管理，不应该被用来突破服务端的访问策略。

本文的 Retry 已设置：

respect_retry_after_header=True

同时手动检查最终响应是否仍为 429。

10.3 HTML 抓到空壳怎么办

例如 Requests 返回：

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

任务卡片完全不在 HTML 中。

处理步骤：

如果是 GraphQL，请重点看请求体：

{
"operationName": "UpcomingMissions",
"variables": {
"limit": 20,
"cursor": null
},
"query": "query UpcomingMissions…"
}

10.4 JSON 解析报错

常见错误：

JSONDecodeError

先检查：

print(response.status_code)
print(response.headers.get("Content-Type"))
print(response.text[:500])

可能原因：

• 返回的是 HTML 错误页。
• 接口地址失效。
• 触发限流。
• 服务端临时故障。
• URL 被重定向到登录页。
• 响应体为空。
• 接口返回 JSON Lines，而不是单个 JSON 对象。

不要直接写：

data = response.json()

然后忽略状态码和 Content-Type。

10.5 解析报错怎么办

如果出现：

TypeError: 'NoneType' object is not subscriptable

说明某一级字段是 None。

例如：

mission = item.get("mission")
orbit = mission["orbit"]["name"]

当 mission 为 None 时就会报错。

改为：

mission = item.get("mission")

if not isinstance(mission, dict):
mission = {}

然后使用：

orbit = safe_get(
mission,
"orbit",
"name",
default="轨道未知",
)

10.6 字段结构变化

接口版本升级后，可能发生：

launch_service_provider → provider
mission.orbit.name → mission.orbit.display_name

建议保留以下信息：

• 原始响应缓存。
• API 版本号。
• 单元测试。
• 解析警告日志。
• 字段映射文档。

接口 URL 中应尽量使用明确版本：

/2.3.0/

不要在生产项目中完全依赖“latest”别名，否则版本升级后可能在没有代码变更的情况下突然报错。

10.7 编码或乱码

JSON 通常使用 UTF-8。

写 JSON：

json.dump(
data,
file,
ensure_ascii=False,
indent=2,
)

写 CSV 给 Excel：

encoding="utf-8-sig"

如果误用系统默认编码，中文环境和英文环境下可能得到不同结果。

10.8 日期排序错误

字符串日期只有在统一格式和时区后才适合直接排序。

以下数据不能可靠混排：

2026-06-01 10:00
2026/06/01 09:00
June 1, 2026
2026-06-01T08:00:00Z
2026-06-01T10:00:00+02:00

本项目统一转换为：

2026-06-01T08:00:00Z

数据库和前端均按 UTC 排序，页面展示时明确写出 UTC。

10.9 为什么未来任务日期会变化

航天任务时间可能因为技术检查、发射窗口、天气、载荷准备或其他运行条件发生调整。

因此：

scheduled_at

表示当前数据源中的计划或实际时间，不应该在所有情况下解释为永远不变的最终时间。

更完整的项目应保存每次更新前后的值，形成日期变更历史。

1️⃣1️⃣ 监听 JSON / GraphQL 接口

下面编写一个通用 Playwright 网络监听器。

它可以：

• 监听 XHR、Fetch 和文档响应。
• 识别 JSON Content-Type。
• 识别 URL 中包含 graphql 的请求。
• 保存请求方法和地址。
• 保存 GraphQL operationName、variables。
• 保存公开 JSON 响应。
• 限制单个响应体大小。
• 输出 JSON Lines 文件。

11.1 app/inspect_network.py

from __future__ import annotations

import argparse
import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from playwright.sync_api import (
Request,
Response,
sync_playwright,
)

SENSITIVE_HEADERS = {
"authorization",
"cookie",
"set-cookie",
"proxy-authorization",
"x-api-key",
}

def utc_now_string() –> str:
return (
datetime.now(timezone.utc)
.isoformat(timespec="seconds")
.replace("+00:00", "Z")
)

def safe_headers(
headers: dict[str, str],
) –> dict[str, str]:
return {
key: value
for key, value in headers.items()
if key.casefold() not in SENSITIVE_HEADERS
}

def get_post_data(
request: Request,
) –> Any:
if request.method.upper() != "POST":
return None

try:
return request.post_data_json
except Exception:
return request.post_data

def parse_arguments() –> argparse.Namespace:
parser = argparse.ArgumentParser(
description=(
"监听页面中的 JSON、XHR、Fetch 和 "
"GraphQL 请求。"
)
)

parser.add_argument(
"–url",
required=True,
help="需要观察的公开页面地址。",
)

parser.add_argument(
"–output",
type=Path,
default=Path(
"data/raw/network_capture.jsonl"
),
help="捕获结果输出路径。",
)

parser.add_argument(
"–wait-seconds",
type=float,
default=8.0,
help="页面加载后额外等待时间。",
)

parser.add_argument(
"–timeout-seconds",
type=float,
default=45.0,
help="页面导航超时。",
)

parser.add_argument(
"–max-body-bytes",
type=int,
default=2_000_000,
help="单个响应体最大保存字节数。",
)

parser.add_argument(
"–headed",
action="store_true",
help="显示浏览器窗口。",
)

return parser.parse_args()

def main() –> int:
args = parse_arguments()

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

logger = logging.getLogger(
"network-inspector"
)

args.output.parent.mkdir(
parents=True,
exist_ok=True,
)

output_file = args.output.open(
"a",
encoding="utf-8",
)

def write_record(
record: dict[str, Any],
) –> None:
output_file.write(
json.dumps(
record,
ensure_ascii=False,
)
+ "\\n"
)
output_file.flush()

def handle_response(
response: Response,
) –> None:
request = response.request

content_type = response.headers.get(
"content-type",
"",
).casefold()

resource_type = request.resource_type
url_lower = response.url.casefold()

is_json = (
"application/json" in content_type
or "+json" in content_type
)

is_graphql = (
"graphql" in url_lower
)

is_api_resource = resource_type in {
"xhr",
"fetch",
}

if not (
is_json
or is_graphql
or is_api_resource
):
return

content_length_text = (
response.headers.get(
"content-length",
"",
)
)

if content_length_text.isdigit():
content_length = int(
content_length_text
)

if content_length > args.max_body_bytes:
logger.warning(
"跳过过大响应：%s bytes，%s",
content_length,
response.url,
)
return

try:
body = response.body()
except Exception as exc:
logger.warning(
"读取响应体失败：%s，%s",
response.url,
exc,
)
return

if len(body) > args.max_body_bytes:
logger.warning(
"跳过过大响应体：%s bytes，%s",
len(body),
response.url,
)
return

response_data: Any = None

try:
decoded = body.decode(
"utf-8",
errors="replace",
)
except Exception:
decoded = ""

if is_json or is_graphql:
try:
response_data = json.loads(
decoded
)
except json.JSONDecodeError:
response_data = decoded[:2000]
else:
# XHR / Fetch 不一定都是 JSON。
response_data = decoded[:2000]

post_data = get_post_data(
request
)

operation_name = None
variables = None

if isinstance(post_data, dict):
operation_name = post_data.get(
"operationName"
)
variables = post_data.get(
"variables"
)

record = {
"captured_at": utc_now_string(),
"request": {
"method": request.method,
"url": request.url,
"resource_type": resource_type,
"headers": safe_headers(
request.headers
),
"post_data": post_data,
},
"graphql": {
"detected": is_graphql,
"operation_name": operation_name,
"variables": variables,
},
"response": {
"status": response.status,
"url": response.url,
"content_type": content_type,
"headers": safe_headers(
response.headers
),
"body": response_data,
},
}

write_record(record)

logger.info(
"%s %s | HTTP %s | %s",
request.method,
resource_type,
response.status,
response.url,
)

try:
with sync_playwright() as playwright:
browser = playwright.chromium.launch(
headless=not args.headed
)

context = browser.new_context(
user_agent=(
"SatelliteTimelineResearch/1.0 "
"Playwright Network Inspector"
),
locale="zh-CN",
)

page = context.new_page()

page.on(
"response",
handle_response,
)

logger.info(
"打开页面：%s",
args.url,
)

page.goto(
args.url,
wait_until="domcontentloaded",
timeout=int(
args.timeout_seconds * 1000
),
)

page.wait_for_timeout(
int(
args.wait_seconds * 1000
)
)

browser.close()

except KeyboardInterrupt:
logger.warning("用户中断监听。")
return 130

except Exception as exc:
logger.exception(
"监听失败：%s",
exc,
)
return 1

finally:
output_file.close()

logger.info(
"捕获结果已写入：%s",
args.output,
)

return 0

if __name__ == "__main__":
raise SystemExit(main())

11.2 运行监听器

python -m app.inspect_network \\
–url "https://example.com/missions" \\
–headed

输出文件：

data/raw/network_capture.jsonl

Windows PowerShell：

python –m app.inspect_network —url "https://example.com/missions" —headed

当页面需要点击“加载更多”时，通用脚本还可以扩展：

button = page.get_by_role(
"button",
name="加载更多",
)

if button.is_visible():
button.click()

page.wait_for_timeout(3000)

对于无限滚动：

for _ in range(5):
page.mouse.wheel(0, 2500)
page.wait_for_timeout(1200)