Internet Archive Collection 目录采集实战：用 Python 构建一个数字档案目录库
㊗️本期内容已收录至专栏《Python爬虫实战》，持续完善知识体系与项目实战，建议先订阅收藏，后续查阅更方便～
㊙️本期爬虫难度指数：⭐⭐☆☆☆（基础级）
🉐福利： 一次订阅后，专栏内的所有文章可永久免费看，持续更新中，保底1000+(篇)硬核实战内容。
全文目录：
🌟 开篇语
哈喽，各位小伙伴们你们好呀～我是【喵手】。
运营社区： C站 / 掘金 / 腾讯云 / 阿里云 / 华为云 / 51CTO
欢迎大家常来逛逛，一起学习，一起进步～🌟
我长期专注 Python 爬虫工程化实战，主理专栏👉 《Python爬虫实战》：从采集策略到反爬对抗，从数据清洗到分布式调度，持续输出可复用的方法论与可落地案例。内容主打一个“能跑、能用、能扩展”，让数据价值真正做到——抓得到、洗得净、用得上。
📌 专栏食用指南（建议收藏）
- ✅ 入门基础：环境搭建 / 请求与解析 / 数据落库
- ✅ 进阶提升：登录鉴权 / 动态渲染 / 反爬对抗
- ✅ 工程实战：异步并发 / 分布式调度 / 监控与容错
- ✅ 项目落地：数据治理 / 可视化分析 / 场景化应用
📣 专栏推广时间：如果你想系统学爬虫，而不是碎片化东拼西凑，欢迎订阅专栏👉《Python爬虫实战》👈，一次订阅后，专栏内的所有文章可永久免费阅读，持续更新中。
💕订阅后更新会优先推送，按目录学习更高效💯～
0️⃣ 前言（Preface）
一句话概括：本文会用 Python + requests + Internet Archive 官方搜索接口，采集 Internet Archive 的公开 Collection 目录，整理出 collection 名称、类型、条目数、说明、链接，并导出为结构化数据文件。
读完本文，你能获得：
- 一个可复现的 Internet Archive Collection 目录采集项目。
- 一套比较稳妥的“API 优先、页面解析兜底”的采集思路。
- 一份可以继续扩展成数字档案目录库、数据看板、检索索引的基础数据。
我个人更喜欢把这类项目叫作“目录型采集”，它和采集商品价格、新闻正文不太一样。它不追求页面炫技，也不需要浏览器自动化乱飞，而是要把公开元数据安静、稳定、克制地拿回来。Internet Archive 本身提供搜索 API、Metadata API、命令行工具和 Python 库等接口，官方文档也明确列出了相关工具和服务，所以这篇文章会优先使用公开 API，而不是硬扒 HTML 页面。
1️⃣ 摘要（Abstract）
本文要爬取的是 Internet Archive Collection 目录数据，使用 requests、urllib3 Retry、BeautifulSoup、CSV/JSONL 存储 完成采集、清洗、去重和导出。
读完本文，你可以获得：
- 如何通过 Internet Archive Search Scrape API 批量获取 collection 列表。
- 如何用 Metadata API 或搜索统计补齐条目数、说明等字段。
- 如何封装 Fetcher、Parser、Storage，让爬虫项目更像一个长期可维护的数据工程脚本。
Internet Archive 的搜索文档说明，它提供基于元数据的搜索 API；其中传统 advancedsearch 接口适合一般检索，而 cursor 形式的 scraping/search API 更适合深度分页。官方文档还特别说明，排序分页结果存在 10,000 条限制，而 scraping API 使用 cursor 机制继续向后取数据。 这也是本文选择 “Scrape API + cursor” 的主要原因。
2️⃣ 背景与需求（Why）
2.1 为什么要采集 Internet Archive Collection 目录？
Internet Archive 是一个大型数字资料库，里面既有文本、音频、视频、图片、软件，也有大量由主题、机构、项目、媒体类型组织起来的 Collection。对研究者、开发者、图书馆技术人员、数字人文爱好者来说，Collection 目录本身就很有价值。
举几个比较实际的场景：
第一，做数字档案目录库。
如果只靠网页搜索，很难知道有哪些 collection、每个 collection 大概是什么、里面有多少条目。把目录采集下来后，可以做成一个本地检索表。
第二，做信息聚合入口。
比如你想建立一个“公共领域资料导航站”，可以先按 collection 抽取目录，再手工筛选高质量 collection。
第三，做后续数据分析。
比如统计不同类型 collection 的数量、条目规模分布、说明文本关键词，或者筛选条目数较大的 collection 作为下一步采样对象。
第四，给爬虫任务做种子库。
很多时候我们不应该一上来就抓详情页或文件，而是先抓目录，确定范围，再分批处理。
Internet Archive 的 Metadata API 文档说明，Internet Archive 将其馆藏组织为 items，每个 item 都包含内容文件和元数据；Metadata API 可以在单次请求中获取一个 item 的元数据记录。对 collection 来说，我们可以把 collection 也当作一种特殊目录型 item 来理解。
2.2 本文目标站点
目标站点：
https://archive.org
主要使用接口：
https://archive.org/services/search/v1/scrape
https://archive.org/metadata/{identifier}
其中，search scrape 接口用于分页采集 collection 列表；metadata 接口用于在必要时读取单个 collection 的更完整元数据。官方文档里给出的 metadata read 方式就是对 /metadata/{identifier} 发起 HTTP GET 请求，并返回 JSON 格式的元数据。
2.3 目标字段清单
本文采集字段如下：
| 中文字段 | 英文字段 | 来源 | 说明 |
|---|---|---|---|
| collection 名称 | collection_name |
搜索结果中的 title，缺失时用 identifier |
用于展示 |
| collection 标识 | identifier |
搜索结果 | Internet Archive 唯一标识 |
| 类型 | collection_type |
type / mediatype |
通常为 collection 相关类型 |
| 条目数 | item_count |
优先 members，缺失时用 collection 查询统计 |
表示该 collection 下公开条目数量的近似统计 |
| 说明 | description |
搜索结果或 metadata | 清洗 HTML 后保存 |
| 链接 | url |
identifier 拼接 | https://archive.org/details/{identifier} |
| 抓取时间 | crawled_at |
本地生成 | 方便后续审计 |
| 内容 hash | content_hash |
本地生成 | 用于去重或变化检测 |
这里有一个小坑：不同接口、不同 collection 返回的字段并不总是完全一致。所以代码不会假设每个字段都一定存在，而是采用“优先取直接字段，缺失就兜底”的方式。Internet Archive 的 Metadata API 文档也提到，很多顶层字段是常见字段，但有些字段是可选的，取决于 item 内容本身。
3️⃣ 合规与注意事项（必写）
3.1 robots.txt 基本说明
写爬虫时，第一件事不是写代码，而是先看目标站点是否允许自动化访问。本文面向的是公开元数据接口，不涉及登录、不绕过限制、不采集个人敏感信息。
从当前公开 robots.txt 看，archive.org 的 robots 文件包含 sitemap，并对 /control/、/report/ 等路径做了 Disallow。本文不会访问这些路径，而是使用官方公开搜索和元数据接口。
不过，robots.txt 不是法律意见，也不是“拿到允许就可以无限请求”的通行证。它更像站点给自动化程序写的一份访问边界说明。工程上要同时考虑请求频率、接口负载、失败重试、缓存和 User-Agent 标识。
3.2 User-Agent 必须写清楚
Internet Archive 的自动化访问说明明确要求，所有自动化请求都应该带上描述性的 User-Agent，标识工具或 bot 名称、版本等信息。 所以本文代码不会使用默认的 python-requests/x.x.x，而是配置一个类似下面的 UA：
IACollectionCrawler/1.0 (+https://example.com/contact; for metadata research)
真实项目里建议把联系方式写清楚，比如项目页或邮箱。这样目标站点排查异常流量时，至少知道是谁在访问。
3.3 控制频率，不要攻击式并发
这类目录型采集没有必要高并发。本文默认每次请求后睡眠 1 秒，并且对 429、500、502、503、504 做指数退避。Internet Archive 官方自动化访问建议也提到，批量操作应该加延迟、尊重 429 Too Many Requests 和 Retry-After，并且限制并发。
我个人的建议是：
单机采集：1 个进程即可
默认间隔：1 秒
详情补全：只对缺失字段做
失败重试：最多 3~5 次
429：严格按 Retry-After 等待
3.4 不采集敏感信息，不绕过付费或登录限制
本文只采集公开 Collection 元数据，不采集用户账号信息，不抓取需要登录的数据，不绕过任何访问控制，也不下载大文件内容。即使某些公开 item 里可能带有上传者、评论等信息，本文也不会把它们作为采集目标。
爬虫技术本身是中性的，关键看你怎么用。目录库、公开元数据整理、个人学习和研究通常是合理方向；绕过权限、拖垮服务、采集敏感信息，则不应该做。
4️⃣ 技术选型与整体流程（What/How）
4.1 静态、动态还是 API？
本文属于 API 型采集。
原因很直接：目标数据是结构化元数据，而 Internet Archive 官方提供了搜索 API、Metadata API、命令行工具、Python 库等多种访问方式。官方 Tools and APIs 页面也列出了 Metadata API、Python Library、CLI、Views API、Wayback APIs 等服务。
因此，优先级是：
官方 API > 页面内 JSON > HTML 解析 > 浏览器自动化
本文不使用 Playwright。Playwright 很强，但它适合动态渲染、反爬较强、交互复杂的网页。Internet Archive Collection 目录这种结构化公开元数据任务，用 Playwright 反而显得太重。
4.2 整体流程
流程图用文字表示如下：
配置参数
↓
构造请求 Session
↓
调用 Search Scrape API
↓
读取 items / cursor
↓
解析 collection 字段
↓
必要时补全条目数或说明
↓
清洗 description HTML
↓
按 identifier / url 去重
↓
写入 CSV / JSONL
↓
输出日志与统计
用四个词概括就是：
采集 → 解析 → 清洗 → 存储
4.3 为什么选 requests？
本文选用：
requests + urllib3 Retry + BeautifulSoup + csv/json
原因如下：
-
requests 简洁稳定。
对 API 请求足够了。 -
urllib3 Retry 可以尊重 Retry-After。
对 429、503 这类情况更稳。 -
BeautifulSoup 用于清洗 description 里的 HTML。
虽然主体是 JSON，但 description 字段可能含 HTML 标签。 -
CSV/JSONL 更适合入门和审计。
CSV 方便 Excel 打开，JSONL 方便后续进入数据管道。
Scrapy 适合大型爬虫系统，内置调度、去重、并发、管道；但本文目标是一个可读、可复用的小项目，所以先不用 Scrapy。后文进阶部分会说明如何迁移。
5️⃣ 环境准备与依赖安装（可复现）
5.1 Python 版本
建议使用：
Python 3.10+
我建议最低不要低于 Python 3.10。不是因为旧版本完全不能写，而是类型注解、dataclass、异常处理和依赖兼容会舒服很多。
5.2 创建项目目录
推荐项目结构：
ia_collection_crawler_project/
├── README.md
├── requirements.txt
├── data/
│ ├── archive_collections.csv
│ └── archive_collections.jsonl
├── logs/
│ └── crawler.log
├── ia_collection_crawler/
│ ├── __init__.py
│ ├── config.py
│ ├── fetcher.py
│ ├── parser.py
│ ├── storage.py
│ └── main.py
└── tests/
└── test_parser.py
5.3 安装依赖
requirements.txt：
requests>=2.32.0
urllib3>=2.2.0
beautifulsoup4>=4.12.0
tqdm>=4.66.0
安装：
python -m venv .venv
# macOS / Linux
source .venv/bin/activate
# Windows PowerShell
# .venv\Scripts\Activate.ps1
pip install -r requirements.txt
6️⃣ 核心实现：请求层（Fetcher）
请求层的目标是把网络访问细节都封装起来。业务代码不应该到处写 requests.get()，否则后面加重试、换 UA、加日志会很难维护。
6.1 配置文件
ia_collection_crawler/config.py：
from __future__ import annotations
from dataclasses import dataclass, field
import os
from pathlib import Path
@dataclass(frozen=True)
class CrawlerConfig:
"""
采集配置。
设计原则：
1. URL、请求头、超时、重试、输出路径集中管理。
2. 允许通过环境变量覆盖 User-Agent，方便部署时写入真实项目联系方式。
3. 默认低频采集，不做攻击式并发。
"""
scrape_endpoint: str = "https://archive.org/services/search/v1/scrape"
metadata_endpoint_template: str = "https://archive.org/metadata/{identifier}"
# 只采集 collection 类型
query: str = "mediatype:collection"
# Internet Archive scrape API 文档中 fields 使用逗号分隔
fields: tuple[str, ...] = (
"identifier",
"title",
"mediatype",
"description",
"type",
"members",
)
# 官方文档示例中 count 最小通常按 100 使用；这里默认 100
page_size: int = 100
# 连接超时、读取超时
timeout: tuple[float, float] = (5.0, 30.0)
# 请求间隔，默认克制一点
sleep_seconds: float = 1.0
# 重试策略
max_retries: int = 4
backoff_factor: float = 1.5
# 输出
output_dir: Path = Path("data")
csv_filename: str = "archive_collections.csv"
jsonl_filename: str = "archive_collections.jsonl"
# 日志
log_dir: Path = Path("logs")
log_filename: str = "crawler.log"
# UA：真实项目建议替换为你的项目名、版本、联系页面或邮箱
user_agent: str = field(
default_factory=lambda: os.getenv(
"IA_CRAWLER_USER_AGENT",
"IACollectionCrawler/1.0 (+https://example.com/contact; metadata research)",
)
)
referer: str = "https://archive.org/advancedsearch.php"
def csv_path(self) -> Path:
return self.output_dir / self.csv_filename
def jsonl_path(self) -> Path:
return self.output_dir / self.jsonl_filename
def log_path(self) -> Path:
return self.log_dir / self.log_filename
6.2 Fetcher 实现
ia_collection_crawler/fetcher.py：
from __future__ import annotations
import logging
import random
import time
from typing import Any
import requests
from requests import Session
from requests.exceptions import RequestException, JSONDecodeError
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
from .config import CrawlerConfig
class FetchError(RuntimeError):
"""请求层统一异常。"""
class ArchiveFetcher:
"""
Internet Archive 请求封装。
负责：
1. 创建带重试的 requests.Session
2. 设置 headers
3. 设置 timeout
4. 处理 429 / 5xx
5. 提供 search_collections、fetch_metadata、fetch_collection_count 等方法
"""
RETRY_STATUS_CODES = (429, 500, 502, 503, 504)
def __init__(self, config: CrawlerConfig):
self.config = config
self.session = self._build_session()
self.logger = logging.getLogger(self.__class__.__name__)
def _build_session(self) -> Session:
session = requests.Session()
retry = Retry(
total=self.config.max_retries,
connect=self.config.max_retries,
read=self.config.max_retries,
status=self.config.max_retries,
status_forcelist=self.RETRY_STATUS_CODES,
allowed_methods=frozenset(["GET"]),
backoff_factor=self.config.backoff_factor,
respect_retry_after_header=True,
raise_on_status=False,
)
adapter = HTTPAdapter(max_retries=retry, pool_connections=8, pool_maxsize=8)
session.mount("https://", adapter)
session.mount("http://", adapter)
session.headers.update(
{
"User-Agent": self.config.user_agent,
"Accept": "application/json,text/plain,*/*",
"Accept-Encoding": "gzip, deflate",
"Referer": self.config.referer,
"Connection": "keep-alive",
}
)
return session
def polite_sleep(self) -> None:
"""
克制访问，加入轻微随机抖动，避免请求呈现机械固定节奏。
"""
base = self.config.sleep_seconds
jitter = random.uniform(0, base * 0.3)
time.sleep(base + jitter)
def get_json(self, url: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
"""
发起 GET 请求并解析 JSON。
注意：
- timeout 必须设置，避免永久卡住。
- status code 非 2xx 时给出可读日志。
- JSON 解析失败时输出响应前 300 字符，方便排查。
"""
try:
response = self.session.get(
url,
params=params,
timeout=self.config.timeout,
)
except RequestException as exc:
raise FetchError(f"Request failed: {exc}") from exc
if response.status_code >= 400:
text_preview = response.text[:300].replace("\n", " ")
raise FetchError(
f"HTTP {response.status_code} for {response.url}; "
f"body preview: {text_preview}"
)
try:
data = response.json()
except JSONDecodeError as exc:
text_preview = response.text[:300].replace("\n", " ")
raise FetchError(
f"JSON decode failed for {response.url}; body preview: {text_preview}"
) from exc
if not isinstance(data, dict):
raise FetchError(f"Unexpected JSON root type: {type(data)!r}")
return data
def search_collections(self, cursor: str | None = None) -> dict[str, Any]:
"""
调用 Internet Archive scrape API 获取 collection 列表。
返回结构通常包含：
{
"items": [...],
"count": 100,
"total": 12345,
"cursor": "..."
}
"""
params: dict[str, Any] = {
"q": self.config.query,
"fields": ",".join(self.config.fields),
"count": self.config.page_size,
}
if cursor:
params["cursor"] = cursor
self.logger.debug("Search collections params=%s", params)
return self.get_json(self.config.scrape_endpoint, params=params)
def fetch_metadata(self, identifier: str) -> dict[str, Any]:
"""
读取单个 item / collection 的 metadata。
"""
url = self.config.metadata_endpoint_template.format(identifier=identifier)
return self.get_json(url, params={"extended_err": 1})
def fetch_collection_count(self, identifier: str) -> int | None:
"""
当 search 结果里没有 members 字段时，用 total_only 兜底统计。
查询：
q=collection:{identifier}
total_only=true
这个统计会额外发请求，因此不建议对所有 collection 无脑调用。
"""
params: dict[str, Any] = {
"q": f"collection:{identifier}",
"total_only": "true",
"count": self.config.page_size,
}
try:
data = self.get_json(self.config.scrape_endpoint, params=params)
except FetchError as exc:
self.logger.warning(
"Fetch collection count failed: identifier=%s error=%s",
identifier,
exc,
)
return None
total = data.get("total")
if total is None:
return None
try:
return int(total)
except (TypeError, ValueError):
return None
6.3 请求层要点复盘
请求层必须做到这几件事：
| 项目 | 本文做法 |
|---|---|
| headers | 设置描述性 User-Agent、Accept、Accept-Encoding、Referer |
| timeout | 连接超时 5 秒，读取超时 30 秒 |
| session | 使用 requests.Session 复用连接 |
| cookie | 不需要登录，不使用 cookie |
| 失败处理 | 对 429/5xx 使用 Retry 和指数退避 |
| 频率控制 | 每页请求后 sleep，并加入轻微 jitter |
| JSON 异常 | 捕获并输出响应预览 |
这部分看起来不刺激，但它决定了项目能不能长期跑。很多爬虫坏掉，不是选择器写错，而是请求层太草率：没 timeout、没重试、没日志、没 UA，出问题时只能盯着一堆空文件发呆。
7️⃣ 核心实现：解析层（Parser）
本文主要解析 JSON，而不是 HTML。HTML 只在清洗 description 时用到。
7.1 Parser 设计思路
解析层负责把接口返回的原始字典变成统一数据结构。它不关心请求，也不关心存储。
目标：
raw item dict
↓
CollectionRecord
↓
dict
↓
CSV / JSONL
7.2 Parser 代码
ia_collection_crawler/parser.py：
from __future__ import annotations
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
import hashlib
import html
import re
from typing import Any, Callable
from bs4 import BeautifulSoup
@dataclass(slots=True)
class CollectionRecord:
"""
标准化后的 collection 记录。
"""
identifier: str
collection_name: str
collection_type: str
item_count: int | None
description: str
url: str
crawled_at: str
content_hash: str
def to_dict(self) -> dict[str, Any]:
return asdict(self)
def utc_now_iso() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def first_value(value: Any) -> Any:
"""
Internet Archive 某些字段可能是 list，也可能是 str/int。
这里统一取第一个有意义的值。
"""
if isinstance(value, list):
if not value:
return None
return value[0]
return value
def parse_int(value: Any) -> int | None:
value = first_value(value)
if value is None:
return None
if isinstance(value, int):
return value
if isinstance(value, float):
return int(value)
if isinstance(value, str):
value = value.strip().replace(",", "")
if not value:
return None
# 只提取纯数字，避免 "123 items" 之类字符串导致报错
match = re.search(r"\d+", value)
if match:
try:
return int(match.group(0))
except ValueError:
return None
return None
def clean_text(value: Any, max_length: int | None = None) -> str:
"""
清洗文本：
1. 处理 list / None
2. 去 HTML 标签
3. 反转义 HTML entity
4. 压缩空白字符
5. 可选截断
"""
value = first_value(value)
if value is None:
return ""
text = str(value)
if "<" in text and ">" in text:
soup = BeautifulSoup(text, "html.parser")
text = soup.get_text(" ", strip=True)
text = html.unescape(text)
text = re.sub(r"\s+", " ", text).strip()
if max_length is not None and len(text) > max_length:
text = text[: max_length - 1].rstrip() + "…"
return text
def make_url(identifier: str) -> str:
return f"https://archive.org/details/{identifier}"
def make_content_hash(*parts: str) -> str:
raw = "\n".join(parts).encode("utf-8", errors="ignore")
return hashlib.sha256(raw).hexdigest()
def extract_items(payload: dict[str, Any]) -> list[dict[str, Any]]:
"""
兼容不同接口形态。
Scrape API 常见：
{"items": [...]}
AdvancedSearch API 常见：
{"response": {"docs": [...]}}
这里做兼容，避免接口返回结构轻微变化时直接崩。
"""
if isinstance(payload.get("items"), list):
return payload["items"]
response = payload.get("response")
if isinstance(response, dict) and isinstance(response.get("docs"), list):
return response["docs"]
if isinstance(payload.get("results"), list):
return payload["results"]
return []
def parse_collection(
raw: dict[str, Any],
count_resolver: Callable[[str], int | None] | None = None,
max_description_length: int | None = 500,
) -> CollectionRecord | None:
"""
把原始 collection dict 解析为 CollectionRecord。
容错策略：
- identifier 缺失：丢弃，因为无法构造唯一 URL
- title 缺失：使用 identifier 作为名称
- type 缺失：使用 mediatype
- item_count 缺失：调用 count_resolver 兜底
- description 缺失：留空字符串
"""
identifier = clean_text(raw.get("identifier"))
if not identifier:
return None
collection_name = clean_text(raw.get("title")) or identifier
collection_type = (
clean_text(raw.get("type"))
or clean_text(raw.get("mediatype"))
or "collection"
)
item_count = parse_int(raw.get("members"))
if item_count is None and count_resolver is not None:
item_count = count_resolver(identifier)
description = clean_text(raw.get("description"), max_length=max_description_length)
url = make_url(identifier)
crawled_at = utc_now_iso()
content_hash = make_content_hash(
identifier,
collection_name,
collection_type,
str(item_count or ""),
description,
url,
)
return CollectionRecord(
identifier=identifier,
collection_name=collection_name,
collection_type=collection_type,
item_count=item_count,
description=description,
url=url,
crawled_at=crawled_at,
content_hash=content_hash,
)
7.3 列表页如何拿详情链接？
在 HTML 爬虫里，我们通常从列表页抽 <a href="">。但本文不是抓 HTML 页面，而是从 API 返回的 identifier 构造详情链接：
url = f"https://archive.org/details/{identifier}"
这比页面选择器稳定。页面 CSS 类名可能会改，但 identifier 是元数据体系里的核心字段。Internet Archive 的高级搜索页面也列出了可返回字段，包括 identifier、title、description、mediatype 等。
7.4 详情页如何抽字段？
本文默认不抓详情页 HTML。必要时请求：
GET /metadata/{identifier}
然后从 JSON 的 metadata 节点里补字段。官方 Metadata Read 文档说明，该接口会返回 item 的元数据 JSON，其中 metadata 里包含 mediatype、collection、title 等字段。
如果后续要补充更多字段，可以写成这样：
def enrich_from_metadata(raw_record: dict, metadata_payload: dict) -> dict:
metadata = metadata_payload.get("metadata") or {}
if not raw_record.get("description"):
raw_record["description"] = metadata.get("description", "")
if not raw_record.get("collection_name"):
raw_record["collection_name"] = metadata.get("title", raw_record["identifier"])
return raw_record
不过要注意：详情补全会造成 N+1 请求。如果你采集 10,000 个 collection，又给每个 collection 请求一次 metadata，就会变成 10,001 次请求。更稳妥的做法是：只有列表字段缺失时才补详情。
8️⃣ 数据存储与导出（Storage）
本文选择 CSV 和 JSONL 双输出。
CSV 适合表格软件打开；JSONL 适合后续进入 Elasticsearch、DuckDB、Spark、Python 数据分析脚本。
8.1 字段映射表
| 字段名 | 类型 | 示例值 |
|---|---|---|
identifier |
string | opensource |
collection_name |
string | Community Texts |
collection_type |
string | collection |
item_count |
integer/null | 123456 |
description |
string | A collection of public texts... |
url |
string | https://archive.org/details/opensource |
crawled_at |
string | 2026-06-09T12:00:00+00:00 |
content_hash |
string | SHA256 |
8.2 去重策略
本文采用两层去重：
第一层：identifier 唯一
第二层：url 唯一
理论上，Internet Archive 的详情 URL 是由 identifier 拼出来的，因此 identifier 唯一就够了。但在工程里，我还是会顺手保留 url 去重逻辑，方便将来兼容其他来源。
如果后续做“变化检测”，可以比较 content_hash：
同 identifier + content_hash 不变：跳过
同 identifier + content_hash 变化：更新
新 identifier：插入
8.3 Storage 代码
ia_collection_crawler/storage.py：
from __future__ import annotations
import csv
import json
from pathlib import Path
from typing import Iterable
from .parser import CollectionRecord
CSV_FIELDS = [
"identifier",
"collection_name",
"collection_type",
"item_count",
"description",
"url",
"crawled_at",
"content_hash",
]
def ensure_parent(path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
def dedupe_records(records: Iterable[CollectionRecord]) -> list[CollectionRecord]:
"""
按 identifier / url 去重。
如果重复出现，保留第一次出现的记录。
"""
result: list[CollectionRecord] = []
seen_identifiers: set[str] = set()
seen_urls: set[str] = set()
for record in records:
if record.identifier in seen_identifiers:
continue
if record.url in seen_urls:
continue
result.append(record)
seen_identifiers.add(record.identifier)
seen_urls.add(record.url)
return result
def write_csv(records: Iterable[CollectionRecord], path: Path) -> int:
ensure_parent(path)
rows = [record.to_dict() for record in records]
with path.open("w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)
writer.writeheader()
writer.writerows(rows)
return len(rows)
def write_jsonl(records: Iterable[CollectionRecord], path: Path) -> int:
ensure_parent(path)
count = 0
with path.open("w", encoding="utf-8") as f:
for record in records:
f.write(json.dumps(record.to_dict(), ensure_ascii=False) + "\n")
count += 1
return count
9️⃣ 运行方式与结果展示（必写）
9.1 主程序入口
ia_collection_crawler/main.py：
from __future__ import annotations
import argparse
import logging
import sys
from pathlib import Path
from tqdm import tqdm
from .config import CrawlerConfig
from .fetcher import ArchiveFetcher, FetchError
from .parser import CollectionRecord, extract_items, parse_collection
from .storage import dedupe_records, write_csv, write_jsonl
def setup_logging(config: CrawlerConfig, verbose: bool = False) -> None:
config.log_dir.mkdir(parents=True, exist_ok=True)
level = logging.DEBUG if verbose else logging.INFO
handlers = [
logging.StreamHandler(sys.stdout),
logging.FileHandler(config.log_path(), encoding="utf-8"),
]
logging.basicConfig(
level=level,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
handlers=handlers,
)
def crawl_collections(
config: CrawlerConfig,
limit: int | None = None,
resolve_count: bool = False,
max_description_length: int | None = 500,
) -> list[CollectionRecord]:
logger = logging.getLogger("crawl_collections")
fetcher = ArchiveFetcher(config)
records: list[CollectionRecord] = []
cursor: str | None = None
progress = tqdm(total=limit, desc="Crawling collections", unit="record")
try:
while True:
try:
payload = fetcher.search_collections(cursor=cursor)
except FetchError as exc:
logger.error("Search request failed: %s", exc)
break
items = extract_items(payload)
if not items:
logger.info("No more items found.")
break
logger.info(
"Fetched page: items=%s cursor_exists=%s total=%s",
len(items),
bool(payload.get("cursor")),
payload.get("total"),
)
for raw in items:
if limit is not None and len(records) >= limit:
break
resolver = fetcher.fetch_collection_count if resolve_count else None
record = parse_collection(
raw,
count_resolver=resolver,
max_description_length=max_description_length,
)
if record is None:
logger.warning("Skip item without identifier: %s", raw)
continue
records.append(record)
progress.update(1)
# 如果启用了 count_resolver，单条记录可能额外请求一次，频率要更克制
if resolve_count:
fetcher.polite_sleep()
if limit is not None and len(records) >= limit:
logger.info("Reach limit=%s, stop.", limit)
break
cursor = payload.get("cursor")
if not cursor:
logger.info("No cursor returned, stop.")
break
fetcher.polite_sleep()
finally:
progress.close()
deduped = dedupe_records(records)
logger.info("Records collected=%s deduped=%s", len(records), len(deduped))
return deduped
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Crawl Internet Archive collection directory metadata."
)
parser.add_argument(
"--limit",
type=int,
default=300,
help="Maximum number of collection records to collect. Default: 300",
)
parser.add_argument(
"--resolve-count",
action="store_true",
help="Resolve item_count by extra total_only query when members is missing.",
)
parser.add_argument(
"--sleep",
type=float,
default=1.0,
help="Sleep seconds between page requests. Default: 1.0",
)
parser.add_argument(
"--out-dir",
type=str,
default="data",
help="Output directory. Default: data",
)
parser.add_argument(
"--max-description-length",
type=int,
default=500,
help="Max description length. Use 0 for no truncation. Default: 500",
)
parser.add_argument(
"--verbose",
action="store_true",
help="Enable debug logs.",
)
return parser.parse_args()
def main() -> None:
args = parse_args()
max_description_length = (
None if args.max_description_length == 0 else args.max_description_length
)
config = CrawlerConfig(
sleep_seconds=args.sleep,
output_dir=Path(args.out_dir),
)
setup_logging(config, verbose=args.verbose)
logging.info("Start crawling Internet Archive collections.")
logging.info("User-Agent: %s", config.user_agent)
records = crawl_collections(
config=config,
limit=args.limit,
resolve_count=args.resolve_count,
max_description_length=max_description_length,
)
csv_count = write_csv(records, config.csv_path())
jsonl_count = write_jsonl(records, config.jsonl_path())
logging.info("CSV saved: %s rows=%s", config.csv_path(), csv_count)
logging.info("JSONL saved: %s rows=%s", config.jsonl_path(), jsonl_count)
print()
print("Done.")
print(f"CSV: {config.csv_path()}")
print(f"JSONL: {config.jsonl_path()}")
print(f"Rows: {len(records)}")
if __name__ == "__main__":
main()
9.2 包初始化文件
ia_collection_crawler/__init__.py：
__version__ = "1.0.0"
9.3 运行命令
在项目根目录执行：
python -m ia_collection_crawler.main --limit 300
如果你希望在 members 字段缺失时额外查询条目数：
python -m ia_collection_crawler.main --limit 300 --resolve-count --sleep 1.5
调试模式：
python -m ia_collection_crawler.main --limit 50 --verbose
不截断说明字段：
python -m ia_collection_crawler.main --limit 100 --max-description-length 0
自定义 User-Agent：
export IA_CRAWLER_USER_AGENT="MyArchiveCatalogBot/1.0 (+mailto:you@example.com)"
python -m ia_collection_crawler.main --limit 300
Windows PowerShell：
$env:IA_CRAWLER_USER_AGENT="MyArchiveCatalogBot/1.0 (+mailto:you@example.com)"
python -m ia_collection_crawler.main --limit 300
9.4 输出位置
默认输出：
data/archive_collections.csv
data/archive_collections.jsonl
logs/crawler.log
9.5 示例结果展示
下面是 CSV 的格式示例。注意，item_count 会随 Internet Archive 数据变化而变化，实际结果以你运行时接口返回为准。
| identifier | collection_name | collection_type | item_count | description | url |
|---|---|---|---|---|---|
opensource |
Community Texts |
collection |
运行时填充 |
Community-created texts and documents... |
https://archive.org/details/opensource |
opensource_audio |
Community Audio |
collection |
运行时填充 |
A collection of audio uploaded by users... |
https://archive.org/details/opensource_audio |
opensource_movies |
Community Video |
collection |
运行时填充 |
Community video collection... |
https://archive.org/details/opensource_movies |
opensource_image |
Community Images |
collection |
运行时填充 |
Images uploaded by the community... |
https://archive.org/details/opensource_image |
open_source_software |
Software Library |
collection |
运行时填充 |
Software and related materials... |
https://archive.org/details/open_source_software |
这些示例 collection 名称来自 Internet Archive 自动化访问最佳实践中提到的社区 collection 例子，例如 Community Texts、Community Video、Community Audio、Community Images、Community Data、Community Software 等。
🔟 常见问题与排错（强烈建议写）
10.1 403 怎么办？
403 通常表示访问被拒绝。常见原因：
- User-Agent 过于模糊或缺失。
- 请求频率太高。
- 访问了不该访问的路径。
- IP 或网络环境异常。
处理建议：
先降低频率
检查 User-Agent
确认请求的是公开 API
不要访问 robots.txt 禁止路径
不要加奇怪 cookie
本文不建议一上来就换代理。很多新手遇到 403 第一反应是代理池，但真正的问题往往是请求姿势不对。先把基础合规做好，比盲目换 IP 更重要。
10.2 429 怎么办？
429 表示请求过多。正确做法是：
减少请求频率
尊重 Retry-After
降低并发
增加缓存
减少详情补全请求
本文的 Retry 配置已经设置：
respect_retry_after_header=True
这意味着服务端返回 Retry-After 时，客户端会尽量遵守。Internet Archive 官方自动化访问建议也明确提到要尊重 429 和 Retry-After。
10.3 HTML 抓到空壳怎么办？
本文主体不抓 HTML 页面，所以不会太依赖页面渲染。如果你自己扩展时抓页面，发现 HTML 里没有目标内容，通常有几种可能：
- 数据由前端 JS 动态加载。
- 内容藏在接口返回的 JSON 里。
- 页面结构和你想象的不一样。
- 请求被服务端识别为异常。
处理方法：
打开浏览器开发者工具
查看 Network 面板
优先寻找 JSON 接口
能用公开 API 就不用 Playwright
对 Internet Archive 这类站点，先查官方 API 文档通常更省力。
10.4 解析报错怎么办？
解析报错大多来自字段缺失或类型不稳定。比如：
title = raw["title"]
如果 title 缺失就会 KeyError。
更稳妥：
title = raw.get("title") or raw.get("identifier") or ""
本文 parser 里做了这类容错：
collection_name = clean_text(raw.get("title")) or identifier
10.5 description 乱码怎么办？
本文 CSV 使用：
encoding="utf-8-sig"
这样 Windows Excel 打开时通常更友好。如果你在 Linux/macOS 或 pandas 中读取，utf-8 和 utf-8-sig 都可以处理。
读取 CSV：
import pandas as pd
df = pd.read_csv("data/archive_collections.csv", encoding="utf-8-sig")
print(df.head())
10.6 为什么条目数有时为空？
可能原因：
- 搜索结果没有返回
members字段。 - 某些 collection 字段结构不同。
- 统计接口暂时失败。
- collection 数据正在变化。
解决：
python -m ia_collection_crawler.main --limit 300 --resolve-count --sleep 1.5
--resolve-count 会在缺失时额外请求统计，但会增加请求量，所以不要开太大 limit。
10.7 为什么 total 和最终行数不一致？
Scrape API 的 cursor 分页适合深度遍历，但官方文档也提醒，Archive 数据会新增或删除；在分页过程中，结果集本身可能发生变化，因此不能保证每个 item 都绝对稳定地返回。
这不是你代码一定写错了，而是开放大型资料库的正常现象。对这种数据源，目录库应该允许“近似实时”，并保留 crawled_at。
1️⃣1️⃣ 进阶优化（可选但加分）
11.1 并发优化
本文默认单线程，因为目录采集没必要太急。如果要做并发，建议只对 metadata 补全做小规模线程池，并限制到 2~4 个 worker。
示例：
from concurrent.futures import ThreadPoolExecutor, as_completed
def resolve_counts_safely(fetcher, identifiers, max_workers=3):
result = {}
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_map = {
executor.submit(fetcher.fetch_collection_count, identifier): identifier
for identifier in identifiers
}
for future in as_completed(future_map):
identifier = future_map[future]
try:
result[identifier] = future.result()
except Exception:
result[identifier] = None
return result
不过我的建议仍然是：先不要并发。把单线程跑通、日志写清楚、失败可恢复，比一开始就追求速度更重要。
11.2 断点续跑
当前版本是一次性导出。如果数据量很大，可以改成断点续跑。
思路：
保存 cursor
保存已抓 identifier 集合
每处理一页就落盘
下次启动读取 state.json
从上次 cursor 继续
示例 state.json：
{
"cursor": "W3siaWRlbnRpZmllciI6IjE5NjEtTC0wNTkxNCJ9XQ==",
"seen_identifiers": [
"opensource",
"opensource_audio"
],
"updated_at": "2026-06-09T12:00:00+00:00"
}
可以新增 checkpoint.py：
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
def load_state(path: Path) -> dict[str, Any]:
if not path.exists():
return {
"cursor": None,
"seen_identifiers": [],
}
with path.open("r", encoding="utf-8") as f:
return json.load(f)
def save_state(path: Path, state: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
tmp_path = path.with_suffix(".tmp")
with tmp_path.open("w", encoding="utf-8") as f:
json.dump(state, f, ensure_ascii=False, indent=2)
tmp_path.replace(path)
这里用临时文件再 replace，是为了避免程序中途退出导致 state 文件写坏。
11.3 SQLite 存储
如果目录数据要长期维护，CSV 不够。SQLite 更适合增量更新。
建表：
CREATE TABLE IF NOT EXISTS archive_collections (
identifier TEXT PRIMARY KEY,
collection_name TEXT NOT NULL,
collection_type TEXT,
item_count INTEGER,
description TEXT,
url TEXT UNIQUE,
crawled_at TEXT,
content_hash TEXT
);
Python 写入：
import sqlite3
from pathlib import Path
from ia_collection_crawler.parser import CollectionRecord
def upsert_sqlite(records: list[CollectionRecord], db_path: Path) -> int:
db_path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(db_path)
try:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS archive_collections (
identifier TEXT PRIMARY KEY,
collection_name TEXT NOT NULL,
collection_type TEXT,
item_count INTEGER,
description TEXT,
url TEXT UNIQUE,
crawled_at TEXT,
content_hash TEXT
)
"""
)
sql = """
INSERT INTO archive_collections (
identifier,
collection_name,
collection_type,
item_count,
description,
url,
crawled_at,
content_hash
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(identifier) DO UPDATE SET
collection_name=excluded.collection_name,
collection_type=excluded.collection_type,
item_count=excluded.item_count,
description=excluded.description,
url=excluded.url,
crawled_at=excluded.crawled_at,
content_hash=excluded.content_hash
"""
rows = [
(
r.identifier,
r.collection_name,
r.collection_type,
r.item_count,
r.description,
r.url,
r.crawled_at,
r.content_hash,
)
for r in records
]
conn.executemany(sql, rows)
conn.commit()
return len(rows)
finally:
conn.close()
11.4 日志与监控
建议关注这些指标：
| 指标 | 说明 |
|---|---|
| 请求成功数 | 判断接口是否稳定 |
| 请求失败数 | 判断网络或访问限制 |
| 解析成功数 | 判断字段结构是否变化 |
| 解析失败数 | 判断数据脏值 |
| 去重前数量 | 判断原始采集规模 |
| 去重后数量 | 判断重复情况 |
| 每页耗时 | 判断接口响应速度 |
| 429 次数 | 判断频率是否过高 |
日志不是装饰品。真正跑数据时，日志就是你的黑匣子。
11.5 定时任务
Linux cron：
0 3 * * * cd /path/to/ia_collection_crawler_project && /path/to/.venv/bin/python -m ia_collection_crawler.main --limit 1000 >> logs/cron.log 2>&1
含义：
每天凌晨 3 点跑一次
采集 1000 条 collection 目录
日志追加到 logs/cron.log
如果要做企业级调度，可以考虑 Airflow、Prefect、Dagster。但个人项目 cron 已经够用。
11.6 迁移到 Scrapy
当你需要：
多任务调度
请求队列
自动去重
失败重试
中间件
管道存储
更完整日志
可以迁移到 Scrapy。
Spider 伪代码：
import scrapy
class IACollectionSpider(scrapy.Spider):
name = "ia_collections"
custom_settings = {
"DOWNLOAD_DELAY": 1.0,
"CONCURRENT_REQUESTS": 4,
"USER_AGENT": "IACollectionCrawler/1.0 (+https://example.com/contact)",
}
def start_requests(self):
url = "https://archive.org/services/search/v1/scrape"
params = {
"q": "mediatype:collection",
"fields": "identifier,title,mediatype,description,type,members",
"count": "100",
}
yield scrapy.FormRequest(
url=url,
method="GET",
formdata=params,
callback=self.parse,
)
def parse(self, response):
payload = response.json()
for item in payload.get("items", []):
yield {
"identifier": item.get("identifier"),
"title": item.get("title"),
"mediatype": item.get("mediatype"),
"description": item.get("description"),
"members": item.get("members"),
}
cursor = payload.get("cursor")
if cursor:
# 继续下一页
pass
但本文的 requests 版本已经足够完成目录库起步。
1️⃣2️⃣ 总结与延伸阅读
这篇文章完成了一个完整的 Internet Archive Collection 目录采集项目。它不是为了展示复杂反爬技巧，而是把一个真实可维护的数据采集流程拆开：请求层负责稳定访问，解析层负责字段容错，存储层负责导出和去重，主程序负责调度和日志。
本文完成的内容包括：
- 确定采集目标：Internet Archive Collection 目录。
- 明确字段：名称、类型、条目数、说明、链接。
- 选择技术路线：公开 API 优先，不使用浏览器自动化。
- 实现请求层：headers、timeout、session、Retry、退避。
- 实现解析层：JSON 解析、字段兜底、HTML 清洗。
- 实现存储层：CSV、JSONL、去重、hash。
- 给出运行命令、输出路径、示例结果和排错方法。
- 延伸到 SQLite、断点续跑、日志监控、定时任务和 Scrapy。
下一步你可以继续做三件事：
第一，把 CSV 导入 SQLite 或 DuckDB。
这样就可以做 SQL 查询和轻量分析。
第二，给目录库加检索界面。
比如用 FastAPI + SQLite 做一个本地搜索页。
第三，按 collection 做二级采集。
先筛出你关心的 collection，再采集 collection 下的 item 元数据，而不是一开始就全站铺开。
最后补一句个人经验：爬虫项目最难的部分往往不是“把数据拿下来”，而是“拿得稳定、拿得克制、拿得可追踪”。尤其面对 Internet Archive 这种公共文化基础设施，技术上能访问，不代表工程上应该粗暴访问。把频率降下来，把 UA 写清楚，把重试做好，把数据结构设计干净，这样的项目才更耐用。
🌟 文末
好啦～以上就是本期的全部内容啦！如果你在实践过程中遇到任何疑问，欢迎在评论区留言交流，我看到都会尽量回复～咱们下期见！
小伙伴们在批阅的过程中，如果觉得文章不错，欢迎点赞、收藏、关注哦～
三连就是对我写作道路上最好的鼓励与支持！ ❤️🔥
✅ 专栏持续更新中｜建议收藏 + 订阅
墙裂推荐订阅专栏 👉 《Python爬虫实战》，本专栏秉承着以“入门 → 进阶 → 工程化 → 项目落地”的路线持续更新，争取让每一期内容都做到：
✅ 讲得清楚（原理）｜✅ 跑得起来（代码）｜✅ 用得上（场景）｜✅ 扛得住（工程化）
📣 想系统提升的小伙伴：强烈建议先订阅专栏 《Python爬虫实战》，再按目录大纲顺序学习，效率十倍上升～
✅ 互动征集
想让我把【某站点/某反爬/某验证码/某分布式方案】等写成某期实战？
评论区留言告诉我你的需求，我会优先安排实现(更新)哒~
⭐️ 若喜欢我，就请关注我叭～（更新不迷路）
⭐️ 若对你有用，就请点赞支持一下叭～（给我一点点动力）
⭐️ 若有疑问，就请评论留言告诉我叭～（我会补坑 & 更新迭代）
✅ 免责声明
本文爬虫思路、相关技术和代码仅用于学习参考，对阅读本文后的进行爬虫行为的用户本作者不承担任何法律责任。
使用或者参考本项目即表示您已阅读并同意以下条款：
- 合法使用： 不得将本项目用于任何违法、违规或侵犯他人权益的行为，包括但不限于网络攻击、诈骗、绕过身份验证、未经授权的数据抓取等。
- 风险自负： 任何因使用本项目而产生的法律责任、技术风险或经济损失，由使用者自行承担，项目作者不承担任何形式的责任。
- 禁止滥用： 不得将本项目用于违法牟利、黑产活动或其他不当商业用途。
- 使用或者参考本项目即视为同意上述条款,即 “谁使用，谁负责” 。如不同意，请立即停止使用并删除本项目。！！！
更多推荐
8
0- 0
已为社区贡献20条内容
所有评论(0)