文章目录

前言：为什么长时间的数据导出几乎总会中断，而这很正常

如果你曾启动过一次大规模的数据导出，你一定熟悉这种感觉。进程跑了好几个小时，进度到了百分之九十，然后断了。连接掉线，服务器返回错误，笔记本进入了睡眠状态。于是只能从头再来。这篇指南写出来就是让这种情况不再发生。

你最终能获得什么。你将学会构建一个能扛住中断的下载器。它能从文件中间续传，记得自己停在了哪一页，重跑时不会产生重复数据，甚至长时间中断后也能恢复。你会得到一个可以直接按需改造的 Python 骨架。

这篇指南写给谁。写给那些从 API 导出数据、下载大文件或通过代理抓取分页结果的工程师、分析师和开发者。难度中等。你需要理解 HTTP 基础，能看懂 Python 代码。不需要深入的网络编程知识。

需要提前掌握什么。基础 Python、HTTP 请求与响应的概念、什么是请求头和状态码。如果你用过 requests 库，这就够了。

需要多少时间。读完并理解这些概念大约四十分钟。按我们的骨架搭建一个可用的下载器，视数据源不同，需要一到三个小时。

关于主题的重要说明。我们不会讨论 429 之类的状态码和带延迟的重试策略（backoff）。那方面有单独的文章。这里的焦点只有一个：进程的状态及其恢复。如何保存进度，如何不丢失也不重复数据，如何从停下的地方继续。

建议：手边放个笔记本或单独的文件，把数据源的参数记下来：它是否支持断点续传，有没有分页导航，游标是什么格式。这些笔记每一步都会用上。

前期准备：工具、权限和环境

在写代码之前，先把工作环境搭好。这要花十分钟，但能省下几个小时的调试时间。

需要安装什么

  1. 安装 Python 3.10 或更新版本。在终端用 python --version 检查版本。
  2. 用 python -m venv venv 创建虚拟环境，让项目依赖不跟系统环境混在一起。
  3. 激活环境。Windows 上是 venv\Scripts\activate，macOS 和 Linux 上是 source venv/bin/activate。
  4. 用 pip install requests 安装 HTTP 请求库。
  5. 为了更高效地操作状态数据库，不需要额外安装：sqlite3 模块已包含在 Python 标准库中。

需要哪些权限

  • 数据源的访问权限：URL、令牌或 API 密钥（如果需要）。
  • 来自 Proxeon 的代理，包含地址、端口和认证信息。没有稳定的代理，稳定导出就失去意义，因为正是代理在分配负载、让连接变得可预测。
  • 磁盘空间用于存放状态文件和导出的数据本身。

检查 Proxeon 代理

  1. 取一条连接串，形如 http://登录名:密码@地址:端口。
  2. 用一个简单请求测试它。在终端执行 curl -x http://登录名:密码@地址:端口 https://api.ipify.org，确认返回的是代理的 IP 地址，而不是你自己的。

建议：把代理连接串保存在环境变量里，而不是写在代码里。这样你就不会不小心把密码提交到版本控制系统。在代码中用 os.environ 读取它。

⚠️ 注意：始终只处理你有合法访问权限的数据源。遵守服务条款及所有者规定的限制。Proxeon 代理用于合法的工程工作：负载分配、连接稳定和数据导出。

✅ 检查：如果命令 python -c "import requests, sqlite3" 没有报错，且通过代理的请求返回了代理服务器地址，环境就准备好了。

基础概念：用大白话解释稳定导出的词典

在写代码之前，先弄懂几个关键术语。没有它们，后面的步骤听起来会像咒语。

断点续传

断点续传就是从文件中断的那个字节继续下载。与其把文件重新下载一遍，你请求服务器只返回缺失的那一小块。它通过 HTTP 请求头 Range 实现。

检查点

检查点是保存下来的进度点。想象电子游戏里的存档。如果出了岔子，你回到最近的存档，而不是从头开始玩。在导出中，检查点存储的是你停在了哪一页或哪条记录。

游标

游标是 API 给你的一个标记，让你可以请求下一批数据。它通常是一段字符串，形如 eyJvZmZzZXQiOjEwMH0。你把它发回去，服务器就知道从哪里继续。

幂等性

幂等性是操作的一种性质：重复执行不会改变结果。如果你用同一个键写入了两次相同的行，最终还是一行，而不是两行。这是在重试时防止重复数据的保护。

去重

去重是过滤掉重复记录。即使工作很规范，同一个对象也可能到达两次。去重保证它在你的最终数据集里只保留一条。

核心原则

稳定的下载器建立在这样一个理念上：进度需要持续保存，而不是只在结束时保存。任何一步都可能是中断前的最后一步。这意味着，每成功完成一小块工作后，状态就必须写入磁盘。这样恢复就只是读取状态并继续。

建议：记住任何导出都要问的三个问题。第一：我停在哪里？第二：如何不重复已经拿到的数据？第三：我不在的这段时间里，什么会失效？回答这些问题，就构成了稳定性。

第 1 步：通过 HTTP 的 Range 头实现断点续传

本阶段目标。学会下载大文件，使得中断后能从没下完的字节继续，而不是从零开始。

原理

HTTP 允许请求文件的某一部分，而不是整个文件。为此在请求中加上 Range 头。例如 Range: bytes=1048576- 的意思是：从第 1048576 字节开始，把后面的都给我。但首先得确认服务器是否支持。

  1. 用 HEAD 方法或普通 GET 请求文件，查看响应头。
  2. 找到 Accept-Ranges 头。如果值是 bytes，说明服务器支持断点续传。
  3. 如果没有这个头或值是 none，断点续传不可行。这种情况下只能一次性下载整个文件，或者找替代数据源。

检查是否支持断点续传

下面这段代码检查服务器能否返回文件的部分内容。

import requests
def supports_resume(url, proxies):
resp = requests.head(url, proxies=proxies, timeout=30, allow_redirects=True)
accept = resp.headers.get("Accept-Ranges", "none")
total = resp.headers.get("Content-Length")
return accept.lower() == "bytes", total

从中间续传文件

下面是核心代码。它查看本地已经下载了多少字节，然后只向服务器请求剩余部分。

import os
import requests
def download_resumable(url, dest, proxies):
already = 0
if os.path.exists(dest):
already = os.path.getsize(dest)
headers = {}
if already > 0:
headers["Range"] = f"bytes={already}-"
mode = "ab" if already > 0 else "wb"
with requests.get(url, headers=headers, proxies=proxies,
 stream=True, timeout=60) as r:
if already > 0 and r.status_code == 200:
mode = "wb"
already = 0
with open(dest, mode) as f:
for chunk in r.iter_content(chunk_size=65536):
if chunk:
f.write(chunk)
return os.path.getsize(dest)

来拆解几个关键点。如果服务器返回 206 状态，说明它确实返回了部分文件，追加写入会正确进行。如果服务器无视 Range 头返回了 200，说明它忽略了断点续传，返回整个文件。这种情况下我们切换到完全重写模式，避免把旧片段和新内容粘在一起、把文件弄坏。

⚠️ 注意：如果无法确定服务器返回的是 206，绝不要用 ab 模式追加数据。否则你会得到一个损坏的文件——开头是上次尝试的残余，后面却是新的完整文件。这样的文件打开会报错，而你要花时间去排查原因。

建议：不要直接下载到目标文件，而是先下到带 .part 后缀的临时文件。等下载完全结束后，再重命名为最终名称。这样你永远不会把完整文件和没下完的文件搞混。

完整性校验

完整下载后，最好检查文件有没有损坏。如果服务器返回了 Content-Length 头，把它跟磁盘上文件的实际大小比较。大小一致，文件就完整到达了。

def verify_size(dest, expected):
if expected is None:
return True
return os.path.getsize(dest) == int(expected)

✅ 检查：在下载到一半时关闭程序，中断它。再启动一次。日志里应该能看到请求带着 Range 头发出，文件是续传而不是重新开始的。最终大小与预期一致。

第 2 步：分页导出的检查点

本阶段目标。为按页返回数据的 API 配置进度保存，以便中断后从正确的页继续。

具体要保存什么

文件按字节续传，分页导出则完全是另一套逻辑。这里没有字节，只有页和记录。所以检查点里要存的是别的东西。

  • 游标——如果 API 基于游标工作。这是最可靠的方案，因为游标自己知道从哪里继续。
  • 页码或偏移量——如果 API 基于 offset 和 limit 工作。保存最后一个成功处理的页码。
  • 最后一条记录的标识符——如果可以按递增的 ID 或日期排序。这样下一次请求就要求返回 ID 大于已保存值的记录。
  • 已处理记录计数器——用于监控和汇报。

状态存在哪里

从简单到可靠，主要有三种选择。

  1. JSON 文件。最简单。把包含游标和计数器的字典在每页之后写入文件。适合单线程、非并行的导出。
  2. SQLite 数据库。更可靠。提供事务，所以状态不会在写入时因中断而损坏。适合数据量大、需要去重的场景。
  3. 外部数据库。适合大型分布式导出，多个进程分担同一份工作。

把检查点保存为 JSON

import json
import os
def save_checkpoint(path, cursor, page, last_id, count):
tmp = path + ".tmp"
data = {
"cursor": cursor,
"page": page,
"last_id": last_id,
"count": count,
}
with open(tmp, "w") as f:
json.dump(data, f)
os.replace(tmp, path)
def load_checkpoint(path):
if not os.path.exists(path):
return {"cursor": None, "page": 0, "last_id": None, "count": 0}
with open(path) as f:
return json.load(f)

注意临时文件的技巧。我们先写入带 .tmp 后缀的文件，然后用 os.replace 原子性地重命名它。这能防止程序在写检查点的一瞬间崩溃：旧检查点保持完整，而不会变成读不出来的半截 JSON。

⚠️ 注意：绝不要在没有临时文件的情况下直接把检查点覆盖写到同一个文件。写入中途中断会留给你一个损坏的检查点，恢复将变得不可能。原子替换彻底解决了这个问题。

建议：只有数据页真正写入存储之后，才保存检查点。顺序是这样的：拿到页，写入数据，然后更新检查点。如果把顺序调换，中断时你会跳过一页、丢失数据。

带检查点的主循环

def paginate(fetch_page, save_data, cp_path, proxies):
cp = load_checkpoint(cp_path)
cursor = cp["cursor"]
count = cp["count"]
while True:
items, next_cursor = fetch_page(cursor, proxies)
if not items:
break
save_data(items)
count += len(items)
last_id = items[-1].get("id")
save_checkpoint(cp_path, next_cursor, cp["page"] + 1,
last_id, count)
cursor = next_cursor
if next_cursor is None:
break
return count

✅ 检查：启动导出，让它处理几页，然后中断。打开检查点文件，确认里面写入了当前游标和计数器。再启动一次：导出应该从保存的游标继续，而不是从第一页。

第 3 步：幂等性——让重复不会产生副本

本阶段目标。让重新运行或重复某次请求不会在你的存储中产生相同的记录。

为什么会产重复

想象一下：你拿到了数据页，写入了文件，但程序在检查点更新之前挂了。下一次启动时你又请求同一页。数据再次返回并被二次写入。副本就是这样诞生的。这是中断不可避免的后果，必须在架构层面解决。

去重键

幂等性的核心工具是去重键。它是指能唯一确定一条记录的一个字段或多个字段的组合。选对键能解决一半的问题。

  • 自然 ID。如果记录有来自数据源的唯一标识符，就用它。这是理想的键。
  • 字段组合。如果没有单一 ID，就用几个稳定字段拼成一个键。比如邮箱加注册日期。
  • 内容哈希。如果根本没有稳定字段，就对整条记录算哈希。这是最后的手段，因为任何一个字段变化都会产生新键。

通过 UPSERT 实现无重复写入

如果你把结果存在 SQLite 或其他数据库里，就用忽略冲突的插入。这样重复写入同一键时什么也不会发生。

import sqlite3
def init_db(path):
conn = sqlite3.connect(path)
conn.execute(
"CREATE TABLE IF NOT EXISTS records ("
"dedup_key TEXT PRIMARY KEY, payload TEXT)"
)
conn.commit()
return conn
def save_records(conn, items):
rows = [(item["id"], json.dumps(item)) for item in items]
conn.executemany(
"INSERT OR IGNORE INTO records (dedup_key, payload) "
"VALUES (?, ?)", rows
)
conn.commit()

这里的关键细节是 dedup_key 字段上的 PRIMARY KEY。数据库自己会拒绝同一键的重复插入，因为 INSERT OR IGNORE 会默默吞掉冲突。你不需要手动检查记录是否已存在。数据库替你做了，而且做得很快。

建议：在项目开始时就把去重键选定，并把它固化到文档里。导出中途更换键意味着新旧记录不再能对应上，副本还是会出现。键的稳定性比它的优雅程度更重要。

✅ 检查：在同一批数据上连续跑两遍导出。用命令 SELECT COUNT(*) FROM records 统计数据库里的行数。第一遍和第二遍之后数字应该相同。

第 4 步：不让结果去重撑爆内存

本阶段目标。在数百万行数据上过滤重复记录，而不让已见键的集合占满整台电脑的内存。

朴素做法及其问题

最简单的去重：在内存里维护一个包含所有已见键的集合 set。对每条新记录检查键是否已在集合中。几十万行时它工作得很好。但在数百万、数千万行时，集合会膨胀并吃掉几十 GB 内存。程序变慢甚至崩溃。

方案一：依赖数据库

在大数据量下最简单可靠的做法是根本不把已见内容存在内存里，而是像上一步那样，把检查交给数据库的 PRIMARY KEY。数据库把索引存在磁盘上，而不是你的进程内存里。它能轻松处理数千万个键，而不会给内存带来压力。

方案二：记录哈希

当没有自然键时，对记录计算一个紧凑的哈希。无论记录本身多大，哈希都占固定的、很小的空间。

import hashlib
import json
def record_hash(item):
raw = json.dumps(item, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()

这里的 sort_keys=True 参数至关重要。它保证内容相同的记录会产生相同的哈希，即使字段出现的顺序不同。没有这个排序，两个完全相同的对象可能得到不同的哈希并作为不同记录蒙混过关。

方案三：用布鲁姆过滤器节省内存

如果你确实需要在海量数据上做快速的内存检查，可以用布鲁姆过滤器。这种结构占用空间小，能快速回答某个键是否见过，或者肯定没见过。它有个特点：偶尔会错误地说某个键已经存在，虽然实际上没有。所以布鲁姆过滤器用作快速预筛，最终校验留给数据库。

  1. 用布鲁姆过滤器检查键。
  2. 如果过滤器说肯定没见过，直接写入数据库。
  3. 如果过滤器说可能见过，就去数据库做精确检查。

⚠️ 注意：不要试图用普通的内存集合去重数千万行。在普通笔记本上，这会在导出中途耗尽内存、导致进程崩溃。把负载转移到磁盘上的数据库，或者使用布鲁姆过滤器。

建议：如果你按批次导出，而一批内可能有重复，就在写入数据库之前用普通集合先把这一批去重。批次不大，内存不会受影响，而进入数据库的多余插入会少很多。

✅ 检查：在一个包含故意重复的大测试集上执行去重。确认唯一记录总数正确，并且进程的内存占用保持稳定、不随行数线性增长。

第 5 步：无损并行

本阶段目标。通过并行请求加速导出，同时不丢失任何一个任务，并正确重试失败的任务。

任务队列

安全并行的基础是任务队列。你事先把工作拆成互相独立的碎片。比如页列表或范围列表。把它们放进队列。若干个工作线程从队列取任务、执行并保存结果。如果某个线程挂了，它的任务可以放回队列交给别人。

限制并发数

不能启动无限个并行请求。那会压垮数据源和你的代理。正确的做法是把同时运行的工作线程数限制在合理的值。从少量开始，观察稳定性后再往上加。

from concurrent.futures import ThreadPoolExecutor, as_completed
def run_parallel(tasks, worker, proxies, max_workers=5):
results = []
failed = []
with ThreadPoolExecutor(max_workers=max_workers) as pool:
future_map = {
pool.submit(worker, t, proxies): t for t in tasks
}
for future in as_completed(future_map):
task = future_map[future]
try:
results.append(future.result())
except Exception:
failed.append(task)
return results, failed

重试失败的任务

收集到的 failed 列表不是丢失的数据，而是需要重试的任务清单。第一轮跑完后，把失败的任务再跑一遍。通常这就够把剩下的补齐了。

def run_with_retry(tasks, worker, proxies, rounds=3):
remaining = tasks
for _ in range(rounds):
done, remaining = run_parallel(remaining, worker, proxies)
if not remaining:
break
return remaining

Proxeon 代理在并行中的角色。并行工作时，代理分配连接，让导出更稳定、更可预测。每个工作线程用自己的连接，负载不会集中在一个点上。

⚠️ 注意：并行写入同一个文件或同一个检查点会产生数据竞争。两个工作线程可能互相覆盖状态。只把结果写入带事务的数据库，或者为每个工作线程使用单独的文件，再用一个独立的线程汇总检查点。

建议：把任务做小、做独立。如果一个任务覆盖的范围太大，它一中断就会丢弃大量工作。小任务重试成本低，几乎无感。

✅ 检查：启动并行导出，故意让一部分工作线程挂掉。经过几轮重试后，remaining 列表应该清空，最终数据集应该完整。把拿到的记录数与预期比较。

第 6 步：长时间中断后的恢复

本阶段目标。在两次尝试之间隔了很久的情况下正确继续导出，并弄清楚这段时间里什么可能已经失效。

什么会随时间失效

中断一分钟和暂停一天是两种不同的情况。长时间的间隔里，你的一部分状态可能失效。

  • 会话。很多服务只会把会话保留有限的时间。长时间暂停后服务器会忘记它，请求开始返回授权错误。
  • 访问令牌。API 令牌通常只有几分钟或几小时的有效期。过期令牌需要在继续之前刷新。
  • 游标。有些游标寿命很短。如果游标失效，就只能从最近一个稳定点开始，比如最后一条记录的标识符。
  • 数据本身。暂停期间，数据源里可能出现新记录，或旧记录发生变化。这会影响按 offset 分页导航时的偏移量。

安全恢复策略

  1. 启动时检查检查点的年龄。如果它很旧，就要做好准备：部分状态已经过时。
  2. 在第一个请求之前刷新访问令牌、新建会话。不要依赖旧的。
  3. 优先按最后一条记录的标识符恢复，而不是按页码。ID 是稳定的，而数据变化时页码会移位。
  4. 用保存的游标做一次试探请求。如果它返回游标无效的错误，就切换到按 last_id 恢复。
def resume(cp, fetch_by_id, fetch_by_cursor, proxies):
if cp["cursor"]:
try:
return fetch_by_cursor(cp["cursor"], proxies)
except CursorExpired:
pass
return fetch_by_id(cp["last_id"], proxies)

为什么按 ID 恢复更可靠。假设你停在按日期排序的第 50 页。你不在的这段时间里，开头新增了记录。现在第 50 页包含的完全是别的数据，而你会漏掉一部分记录。按最后一条记录的标识符恢复就不受影响：你只是请求所有大于已保存 ID 的内容。

建议：检查点里永远同时保存游标和最后一条记录的标识符。游标更快，而 ID 是在长时间暂停后游标失效时的保险绳。

✅ 检查：停止导出，等足够久让令牌或游标过期，然后重新启动。下载器应该刷新令牌、发现失效的游标，并按标识符继续，既不丢失也不重复记录。

第 7 步：一个现成的 Python 稳定下载器骨架

本阶段目标。把学过的一切汇成一个可用的整体框架，你可以按自己的数据源改造它。

下面这个骨架把检查点、通过数据库去重、令牌刷新和恢复整合在一起。获取页面的函数由你按具体的 API 填进去。

import os
import json
import sqlite3
import requests
class ResilientLoader:
def __init__(self, cp_path, db_path, proxies):
self.cp_path = cp_path
self.proxies = proxies
self.conn = sqlite3.connect(db_path)
self.conn.execute(
"CREATE TABLE IF NOT EXISTS records ("
"dedup_key TEXT PRIMARY KEY, payload TEXT)"
)
self.conn.commit()
def load_cp(self):
if not os.path.exists(self.cp_path):
return {"cursor": None, "last_id": None, "count": 0}
with open(self.cp_path) as f:
return json.load(f)
def save_cp(self, cp):
tmp = self.cp_path + ".tmp"
with open(tmp, "w") as f:
json.dump(cp, f)
os.replace(tmp, self.cp_path)
def save_records(self, items):
rows = [(str(i["id"]), json.dumps(i)) for i in items]
self.conn.executemany(
"INSERT OR IGNORE INTO records "
"(dedup_key, payload) VALUES (?, ?)", rows
)
self.conn.commit()
def run(self, fetch_page):
cp = self.load_cp()
while True:
items, next_cursor = fetch_page(
cp["cursor"], cp["last_id"], self.proxies
)
if not items:
break
self.save_records(items)
cp["count"] += len(items)
cp["last_id"] = items[-1]["id"]
cp["cursor"] = next_cursor
self.save_cp(cp)
if next_cursor is None:
break
return cp["count"]

针对你的数据源的获取页面函数示例。这里你实现请求逻辑和响应解析。

def fetch_page(cursor, last_id, proxies):
params = {"limit": 100}
if cursor:
params["cursor"] = cursor
elif last_id:
params["after_id"] = last_id
r = requests.get(
"https://example-source/api/records",
params=params, proxies=proxies, timeout=60
)
r.raise_for_status()
data = r.json()
return data["items"], data.get("next_cursor")

启动整个机制很简单。

proxies = {
"http": os.environ["PROXEON_URL"],
"https": os.environ["PROXEON_URL"],
}
loader = ResilientLoader("state.json", "out.db", proxies)
total = loader.run(fetch_page)
print("Всего записей:", total)

建议：在循环里加上每几百条记录的日志：时间、计数器、当前游标。这样你能看到进度，也容易发现导出卡在某个地方。

✅ 检查：在真实数据源上运行骨架，在中间中断，再启动。续传后的记录总数应与数据源的记录总数一致，而重复运行不会增加唯一行的计数。

结果检查：稳定导出清单

对照这份清单过一遍。如果所有条目都满足，你的下载器就真正稳定了。

  • 文件续传从没下完的字节继续，而不是从零。
  • 下载器能正确处理服务器忽略 Range 头的情况。
  • 检查点在每一页处理后保存，而不是只在结束时。
  • 检查点通过临时文件和重命名原子性地写入。
  • 重复运行不会在最终数据集里产生副本。
  • 去重的内存占用不随行数线性增长。
  • 并行工作线程不丢失失败任务并会重试它们。
  • 长时间暂停后令牌会刷新，失效的游标会被按 ID 恢复所取代。

如何测试

  1. 对一个小子集跑完整导出，记下记录数。
  2. 在同一子集上再跑一遍，确认数字没变。
  3. 在不同地方中断导出：开头、中间、接近结尾。
  4. 每次中断后重新启动，确认结果一致。

成功指标。唯一记录数在多次运行之间保持稳定。内存占用不会失控增长。恢复总是从保存点继续。续传后没有损坏的文件。

常见错误及解决办法

问题：续传后的文件打不开。原因：在服务器返回 200 并给出整个文件时，仍用 ab 模式追加数据。解决办法：检查响应状态，遇到 200 就从零重写文件。

问题：中断后导出从第一页重新开始。原因：检查点只在结束时保存，或者根本没保存。解决办法：每处理完一页就保存检查点，紧跟数据写入之后。

问题：检查点读不出来，JSON 损坏。原因：程序在直接写目标文件时崩溃。解决办法：写入临时文件，并用 os.replace 原子替换。

问题：最终数据集里有重复。原因：没有去重键，或键不稳定。解决办法：在可靠的键上设 PRIMARY KEY，并使用 INSERT OR IGNORE。

问题：大数据量下进程因内存不足而崩溃。原因：所有已见键都保存在内存的集合里。解决办法：把唯一性检查移到数据库，或使用布鲁姆过滤器。

问题：长时间暂停后请求返回授权错误。原因：令牌或会话在中断期间失效了。解决办法：每次启动时刷新令牌、新建会话。

问题：暂停后一部分记录被跳过或翻倍。原因：恢复是按页码进行的，而数据源里的数据发生了变化。解决办法：按最后一条记录的标识符恢复，而不是按 offset。

问题：并行工作线程丢失一部分数据。原因：多个工作线程写入同一个检查点、互相覆盖。解决办法：把结果写入带事务的数据库，而不是公共的状态文件。

其他能力和优化

批量写入

不要一行一行地写数据库。把几百条记录拼成一批，用 executemany 一次插入。在大数据量下这能成倍加快写入。

定期提交

不要每次写入都 commit，而是每几百行提交一次。太频繁的 commit 会拖慢数据库，太稀疏则在中途中丢更多数据。根据你的负载找到平衡点。

进度报告

加上剩余时间估算。知道处理页面的速度和记录总数，就能估算还要等多久。这对长时间导出很方便。

原始数据与处理后数据分开存储

把原始响应与解析后的记录分开存。如果以后你改变解析逻辑，就不用重新下载数据。只需把原始响应过一遍新解析器即可。

建议：把 Proxeon 代理配置成在整个导出过程中连接都保持稳定。稳定的连接能减少中断次数，意味着你的下载器更少进入恢复流程，运行也更快。

FAQ：稳定导出常见问题

怎么判断服务器是否支持文件续传？发送 HEAD 请求，看 Accept-Ranges 头。值是 bytes 表示支持。没有这个头或值是 none 表示续传不可行。

如果 API 不给游标，只给页码怎么办？保存页码，如果可能的话再保存最后一条记录的标识符。恢复时优先按 ID，因为数据变化时页码会移位。

检查点应该多久保存一次？每成功处理和写入一页之后。这样中断时最多丢掉一页的工作量，而不是整次导出。

能不能不用数据库去重？小数据量可以，用内存里的普通集合。数百万行时这样做因内存问题很危险。最好用带 PRIMARY KEY 的数据库或布鲁姆过滤器。

去重键选什么好？数据源的自然唯一 ID（如果有）。如果没有，就用稳定字段的组合。万不得已时用带键排序的整条记录哈希。

为什么按 ID 恢复比按页码更可靠？因为数据源里的数据会变。新记录会让页码移位，按页码你会漏掉或翻倍数据。ID 不受影响。

并行工作线程开多少合适？从少量开始，观察稳定性和数据源限制后再加。过度的并行弊大于利。

怎么安全地保存代理连接串？放在环境变量里，而不是代码里。用 os.environ 读取。这样密码不会进入版本控制系统。

游标在长时间暂停后失效了怎么办？捕获游标无效的错误，切换到按保存的最后一条记录标识符恢复。

需要校验下载文件的完整性吗？需要。把文件的真实大小与 Content-Length 头比较。如果服务器提供校验和，也一并校验。

结语：你现在掌握了什么，以及往哪里走

你走过了一条路：从一遇中断就崩溃的脆弱导出，到稳定的下载器。现在你有了所有工具，让中断不再是灾难，而变成普通的工作场景。

你掌握了什么。通过 Range 头并检查 Accept-Ranges 实现文件续传。通过原子检查点保存进度。在去重键层面实现幂等。在数百万行上不撑爆内存的去重。带失败任务重试的并行。长时间暂停后刷新令牌、替换失效游标的恢复。最重要的是，一个把这一切整合起来的现成 Python 框架。

接下来做什么。拿你真实的数据源，改造获取页面的函数。从小数据量开始，在中断上把恢复流程调通，然后再扩展。配置稳定的 Proxeon 代理，让连接在整个导出过程中都可预测。

往哪里发展。更深入地研究批量写入和布鲁姆过滤器。加上进度监控和时间估算。把原始数据和处理后数据分开存储。逐步……