企业官网建设流程全解析

写这个采集脚本的起因其实很朴素：我日常要做的 Python 技术选型越来越多，每次找库都靠搜索引擎和社区文章翻半天，太零碎了。我就琢磨着，干脆把 PyPI 那几十万个包的结构化信息抓下来，导成 CSV，之后不管查作者、查许可证、看版本更新频率，直接当本地数据库用。这篇文章就把完整的实现过程拆开讲一遍，包括数据源怎么选、requests 怎么配才能不被限流、JSON 要解析哪些字段、最后怎么稳妥地落成 CSV，适合正在学爬虫想找个练手项目的朋友，也适合有包管理需求但不想手动一个一个查资料的人。

1. 项目整体设计：采集 PyPI 到底能做什么

1.1 采集目标：一份可用的 Python 包资产清单

先明确一下，这次采集不是要把 PyPI 上的安装文件下载下来，而是把“包的信息”采集下来。这两件事的差距非常大。安装文件动辄几百 MB，几十万个包归档下来至少得几个 TB，个人电脑根本扛不住；而包的元数据，除了 description 这类长文本之外，绝大多数都是很短的字符串、时间和版本号，全部汇总下来通常也就几百 MB 到 1 GB 左右，完全可控。

具体到字段上，我建议至少抓这几类：

• 基础信息：包名、当前版本、发布日期
• 作者信息：作者名、作者邮箱
• 项目信息：项目主页、项目链接
• 分发信息：Python 版本要求、许可证类型
• 描述信息：一句话简介、完整描述
• 下载信息：最近一个版本的文件上传时间、文件数量

拿到这些数据之后，你能做的事情一下就多起来了。比如找宝藏库：官方 PyPI 搜索只能按名字模糊匹配，但你自己导出的 CSV 可以直接按简介关键词、作者组织、许可证类型去筛选，效率完全不一样。还可以做合规审查，统计某个组织名下所有包的许可证，或者查某个包最近一年有没有更新，判断它是不是已经“死”了。甚至可以做生态趋势分析，比如对比新旧包的发布时间，看某个领域的热度变化。

1.2 数据源选型：官方接口与页面抓取的取舍

写 PyPI 采集器，第一个要决定的就是数据源。PyPI 本身对外提供了好几类数据接口，它们的数据粒度、请求成本和稳定性完全不一样。我实测对比过，选错方案直接决定你这个项目是“跑 5 分钟”还是“跑 5 天”。

我最开始想的是直接抓 PyPI 搜索页，因为页面上能看到包名、简介、最近更新时间，一个页面 25 条数据，用 requests 加 XPath 很容易解析。但实际操作下来发现，搜索页面的 HTML 结构会随着 PyPI 前端升级变化，而且分页请求多了之后容易触发限流。后端接口才是更好的选择。

JSON API 是每个包一个固定 URL，格式是https://pypi.org/pypi/{包名}/json，返回的是标准化 JSON，字段是 PyPI 官方定义的，长期稳定。XML-RPC 则是 PyPI 老牌接口，通过https://pypi.org/pypi这个地址提供 RPC 服务，可以直接调用list_packages()一次性拿到全部包名。Simple Index 是给 pip 用的安装索引，文件很大，但对爬虫来说它是一个天然的“包名大全”，适合批量同步。

所以我的选型结论是：用 XML-RPC 获取全量包名清单，再用 JSON API 按包名逐个采集详情。这样既不用去解析不稳定的 HTML，也不用下载几百 MB 的 Simple Index 再本地切割。需要说明的是，PyPI 官方对 XML-RPC 接口的态度比较保守，文档里明确说了不要高频调用list_packages()，所以我的方案只拿它取一次列表，后续详情全部走 JSON API，这样总请求量是可控的。

1.3 整体流程：从包名到 CSV 的数据管线

整个采集流程分成四个阶段，这也是我做过的爬虫项目里很典型的一条数据管线：

• 阶段一，包名获取：连接 XML-RPC，拿到 PyPI 全量包名的 list，或者用 Simple Index 做本地同步。这个阶段只发 1 到 2 次请求，耗时基本可以忽略。
• 阶段二，详情并发采集：把包名列表分批，通过多线程并发请求 JSON API，解析每个包的关键字段。这个阶段是大头，几十万个包再怎么并发也得跑上一段时间，需要设计限速和重试。
• 阶段三，数据清洗与规整：把解析结果是None的字段补默认值，把时间字符串统一格式，把版本号、许可证这类高频字段手工归类。不清洗的话，后面导出 CSV 你会发现一堆None和空白行，Excel 打开体验非常差。
• 阶段四，CSV 落地与归档：按固定字段顺序写入 CSV，注意用utf-8-sig编码而不是默认的utf-8，否则 Excel 直接打开会乱码。

这套流程的核心思路是“接口优先，数据分层，并发受控”。接口优先解决了解析稳定性问题，数据分层把全量列表和单包详情分开，避免一个请求里承载太多数据导致失败重跑成本高；并发受控则是为了保证不对 PyPI 造成压力。整个过程里我最看重的是“可断点续跑”，因为几十万次网络请求无论如何都会遇到超时和中断，一旦中断就要能从上一次的位置继续，而不是从头再来。

2. 核心细节解析与实操要点

2.1 requests 请求策略：请求头、超时与重试

很多人写爬虫，requests.get 一发完事儿，采集小网站没什么问题，但面对 PyPI 这种高负载的公共服务，不加策略的话很容易被限流甚至封 IP。我的经验是三个关键配置缺一不可：请求头、超时、重试。

先说请求头。PyPI 会检查 User-Agent，虽然不强制要求必须是浏览器 UA，但一个语义化的 UA 能表明你是正常开发者而非恶意脚本，被误伤的概率低很多。我用的是：

headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 " "PyPI-Collector/1.0 (study project)" }

这里后面特意带上了PyPI-Collector/1.0，相当于告诉对方这是有目的、有标识的采集行为，比纯浏览器 UA 更“诚实”一些。部分公共服务会优先处理有明确标识的请求，纯浏览器 UA 反而显得可疑。

再说超时。网络请求必须设置timeout，否则遇到网络抖动时 requests 会一直挂着，线程池很快就全部阻塞，后面的请求全部排队，整个项目看起来像是死了。我用的经验值是连接超时 5 秒、读取超时 15 秒：

resp = session.get(url, timeout=(5, 15))

连接超时是 TCP 建连的超时阈值，读取超时是响应体的读取阈值。对 PyPI 这种响应体动辄几百 KB 的接口来说，15 秒读不完基本就是网络出问题了，再等也没意义。

最后是重试。我强烈建议不要每失败一次就去session.get重新发一次，而是用 urllib3 内置的 Retry 组件，它可以直接挂在 requests 的 Session 上，对指定状态码和网络异常自动退避重试：

from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry retry = Retry( total=5, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET"], ) adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=20) session.mount("https://", adapter)

Retry 的backoff_factor=0.5表示重试间隔会按指数增长，第一次 0.5 秒、第二次 1 秒、第三次 2 秒，这样既不会给服务器造成瞬时压力，也能在限流后留出窗口让请求恢复。status_forcelist里放的是 HTTP 状态码，429 是限流，5xx 是服务器临时错误，这些都值得重试，而不是直接判死。

2.2 JSON 字段解析技巧：从 dict 中安全取值

PyPI JSON API 的返回结构核心是一个info字段，下面挂着name、version、summary、author、license、requires_python、home_page等一大堆属性。另外还有一个urls是当前版本的文件列表，里面有upload_time、filename、packagetype这些信息。还有releases是历史版本字典，key 是版本号，value 是版本文件列表。

听起来简单，实际踩坑的地方在于：老包的字段经常是缺失的，而且每个包缺的字段不一样。有的老包没有author_email，有的没有home_page，还有的license是十六进制的字符串或者一长串自定义文本。如果你用data["info"]["author_email"]这种硬索引，遇到缺失直接 KeyError，整个脚本就崩了。

正确的做法是每一层都用.get()往后退，并且默认值给得宽松一点：

info = data.get("info", {}) name = info.get("name") or info.get("project_name") or "" version = info.get("version", "") summary = info.get("summary", "") author = info.get("author", "") author_email = info.get("author_email", "") license = info.get("license", "") requires_python = info.get("requires_python", "") home_page = info.get("home_page") or info.get("project_url", "") upload_time = data.get("urls", [{}])[0].get("upload_time", "") if data.get("urls") else ""

这里有个容易被忽视的细节：name有的老包可能返回的是空值，但project_name里会有值，所以要用or把两者桥接起来。upload_time不在info里面，而是在urls列表的第一个元素里，代表当前最新发布版本中第一个文件的上传时间。如果你只想取最新发布时间，urls[0].get("upload_time")是最快的路径。

还有一个细节是license字段。PyPI 现在推荐用 License Expression，但大量老包还是随便填的，比如"MIT License"、"BSD"、"Python Software Foundation License"，甚至直接是"UNKNOWN"。如果你想对许可证做统一分析，最好在清洗阶段做一次归一化，比如包含mit就归为 MIT，包含apache就归为 Apache，包含bsd就归为 BSD。这个工作放爬虫里头做会拖慢速度，更适合 CSV 导出来之后用 pandas 批量处理。

2.3 CSV 导出的编码与格式陷阱

CSV 导出看起来是最简单的一步，实际上雷区最多。第一个坑是编码，UTF-8 是 Python 默认编码，但 Excel 打开 UTF-8 无 BOM 的 CSV 时，中文大概率乱码。解决办法是用utf-8-sig编码，它会自动在文件开头写入 BOM 标记，Excel 识别后就用 UTF-8 解析：

with open("pypi_packages.csv", "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(data_rows)

第二个坑是换行符。Windows 的 Excel 对 CSV 换行要求比较严格，所以open必须加newline=""，这是 csv 模块官方文档特别强调的，不加的话 Windows 下每行末尾会多一个空行。

第三个坑是字段顺序。用csv.DictWriter时，fieldnames列表决定了列顺序，最好把英文标题固定写成一个模块级常量，方便复用。不要为了省事把字典键顺序当作列顺序，因为 Python 3.7+ 字典虽然有序，但你在构造行数据时的插入顺序和最后想要的列顺序大概率不一致。

3. 实操过程与核心环节实现

3.1 环境准备与依赖安装

这个项目的依赖很少，核心就是 requests。如果你打算做后续数据分析，可以额外装 pandas，但采集这一环完全用不到。建议在虚拟环境里装：

python -m venv venv source venv/bin/activate # Windows 下用 venv\Scripts\activate pip install requests

这里说一下为什么只用 requests 而不用 Scrapy。Scrapy 是很强大的爬虫框架，但它的学习曲线和工程复杂度对这个项目来说属于杀鸡用牛刀。Scrapy 的项目结构需要你定义 Item、写 Spider、配 Pipeline，光是初始化就够折腾一阵了，而 PyPI 本身是稳定 JSON 接口，不需要渲染、不需要解析复杂 HTML、不需要处理登录，用 requests 加线程池反而更直观、也更方便在 Jupyter 或者脚本里快速调试。

3.2 第一步：获取包名清单（XML-RPC 与 Simple 增量同步）

包名清单是整个采集的地图。如果你做全量采集，不能跳过这一步。我实测最省事的是通过 XML-RPC 调用list_packages()：

import xmlrpc.client client = xmlrpc.client.ServerProxy("https://pypi.org/pypi") packages = client.list_packages() print(f"Total packages: {len(packages)}")

这段代码会返回一个 Python 列表，里面是 PyPI 上所有包名的字符串，数量通常在几十万量级，网络正常的情况下几秒钟就能拉完。相比下载 Simple Index 那个几百 MB 的大文件，这个方式快太多了。

需要注意的是，官方对 XML-RPC 的list_packages有使用建议，不要频繁调用。我这个流程里只在启动时调用一次，后续增量更新我不会再调它，而是换成另一种思路：把包名存成本地文件，下次采集时先读本地文件，再用 Simple Index 或者网站最新列表去找到新增的包名。

另外，如果你只关心下载量大或者最近在活跃更新的包，可以用top_packages()拿下载量排名列表：

top_packages = client.top_packages(100) for rank, (name, downloads) in enumerate(top_packages, start=1): print(rank, name, downloads)

top_packages返回的是元组列表，每个元组是(包名, 下载量)，这个数据放到 CSV 里非常有用，等于天然给每个包加了一个“热度”维度，后续筛选宝藏库时直接按 download 排序就行。

全量包名存下来之后，我习惯把它分成多个小文件，每个文件 5000 个包名。为什么？因为如果只存一个超大的文本文件，中途续采时读起来不方便，而且如果脚本崩了，分片文件更容易定位断点。

3.3 第二步：多线程并发采集详情

拿到包名之后，最直接的做法是 for 循环一个个请求 JSON API。如果 PyPI 上有 60 万个包，每个请求平均 0.3 秒，循环跑下来是 50 个小时，损耗太大了。我改用多线程并发，把速度提升了十几倍。

Python 的多线程虽然受 GIL 限制，对于 IO 密集型任务来说依然非常有效，因为线程在等待网络响应时会让出 GIL，允许其他线程继续发送请求。我常用的写法是用concurrent.futures.ThreadPoolExecutor：

from concurrent.futures import ThreadPoolExecutor, as_completed def fetch_package_detail(session, package_name): url = f"https://pypi.org/pypi/{package_name}/json" try: resp = session.get(url, timeout=(5, 15)) if resp.status_code == 200: data = resp.json() return parse_package_info(package_name, data) else: return None except requests.exceptions.RequestException: return None results = [] with ThreadPoolExecutor(max_workers=8) as executor: future_map = { executor.submit(fetch_package_detail, session, pkg): pkg for pkg in package_list } for future in as_completed(future_map): result = future.result() if result: results.append(result)

线程数我建议控制在 8 到 16 之间，不要贪多。PyPI 是大型公共服务，过高的并发只会让你的 IP 更快触发限流。我实测过 8 线程和 32 线程，8 线程虽然慢一点，但基本上能稳定跑完；32 线程跑到一半会遇到大量 429。所以在这类爬虫里，“慢就是快”，控制并发是为了减少重试，重试反而更拖时间。

每个线程都要共用同一个 Session。Session 底层维护的是连接池，同一个 Session 的多个线程可以直接复用已建立的 TCP 连接，避免每次请求都重新握手。如果你在 fetch 函数内部每次 new 一个 Session，等于放弃了连接池优化。

3.4 第三步：字段清洗与增量标记

并发采集完的数据，不能直接导出，先要清洗。清洗包括两个层面。

第一个是单条记录的字段统一。比如时间字段，JSON 接口返回的是 ISO 格式的字符串，像2024-01-07T12:34:56，但有些老包时间格式不标准，导进数据库再筛选就麻烦了，我会全部规整成YYYY-MM-DD HH:MM:SS。再比如requires_python字段，有的包直接是空值，有的又是">=3.6"这种条件表达式，我保留原文，但把空值统一替换成"Unknown"，避免 CSV 里全是空单元格。

第二个是整个数据集级别的清洗。如果你用了top_packages的下载量数据，那要把下载量和包名对应起来，做一次 match：

download_map = {name: count for name, count in top_packages} for row in data_rows: row["download_count"] = download_map.get(row["name"], "")

这样导出的 CSV 直接就有了“下载量”这一列，后面按下载量排序选库就不用查两次数据了。

增量标记是我这个项目里比较独特的一个设计。采集方向不是一次性的，PyPI 每天都有新包，所以我给每一条记录加了一个collected_at字段，记录当前采集时间，下次重新采集时只需要对比这个时间就能筛出新增数据。代码里很简单：

from datetime import datetime, timezone row["collected_at"] = datetime.now(timezone.utc).isoformat()

别小看这一列，有了它之后，你以后可以每隔一段时间跑一次增量采集，对比分析哪些包被删除了、哪些包新增了、哪些包版本更新了，这些变化趋势对做社区观察特别有用。

3.5 第四步：CSV 导出与结果预览

采集加清洗完毕，最后写 CSV。我用的是标准库csv，没用 pandas，因为这里的导出逻辑不强依赖 DataFrame。下面是完整导出函数：

import csv FIELD_NAMES = [ "name", "version", "summary", "author", "author_email", "license", "requires_python", "home_page", "project_url", "upload_time", "download_count", "collected_at", ] def export_to_csv(data_rows, filename="pypi_packages.csv"): with open(filename, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=FIELD_NAMES) writer.writeheader() for row in data_rows: writer.writerow({key: row.get(key, "") for key in FIELD_NAMES})

这里做了一层字典规整：writerow的时候遍历FIELD_NAMES，从 row 中取值，取不到就填空字符串，这样能保证每一行列数完全一致，不会出现某行多一列某行少一列的情况。不同包的 JSON 返回字段多寡不一致是常态，这个规整步骤能少掉一堆奇怪的错误。

导出完成后，我习惯顺手打印一下总行数，再抽查几条记录看看内容是否符合预期：

print(f"Exported {len(data_rows)} rows to {filename}") print(data_rows[:5])

抽查这个动作是必须的，不要省略。代码跑完不代表数据是对的，看一眼是不是有空值、有没有乱码、字段顺序是不是乱了，几秒钟就能发现问题，等到用 Excel 打开才发现就晚了。

4. 常见问题与排查技巧实录

4.1 请求被打回：429 限流与退避重试

跑全量采集最容易遇到的就是 429。HTTP 429 表示“请求过多”，服务器明确告诉你被限流了。遇到 429 的第一个反应不是换代理，而是检查自己的并发数和重试策略。

我踩过最狠的一次，是把线程池开到 32，结果 5 分钟之后所有请求返回 429，并且持续了小半个小时才恢复。从那之后我总结出一个经验：凡是公共接口，初期的并发数一定要保守，先跑 100 个请求试试水，看返回状态码的分布，再动态调整线程数。正常情况 8 线程是最稳妥的。

另外一个细节是 429 响应头里可能会有 Retry-After 字段，表示服务器希望你等多久再试。如果你用 urllib3 的 Retry，它是不会自动读取这个字段的，只会按 backoff_factor 退避。想在爬虫里更专业地处理 429，可以手动检查：

if resp.status_code == 429: retry_after = resp.headers.get("Retry-After") if retry_after: time.sleep(int(retry_after) + 1)

这个检查放在重试逻辑之前，能显著降低继续被限流的概率。

4.2 老包数据缺失：让解析器“钝感”一点

PyPI 上存在大量上传于十几年前的老包，这些包的信息完整度远不如新包。有的连最基本的作者邮箱都没有，有的summary为空，有的home_page是已经失效的域名。解析的时候如果按理想情况写代码，等着你的就是满天飞的 KeyError 和 IndexError。

我的建议是解析函数里所有取值都用.get()，并且把默认值统一设置为空字符串而不是None。空字符串虽然也不美观，但在后续处理中比None好处理得多——你把 CSV 丢进 pandas、SQLite 或者直接拿 Excel 筛选都更方便。还有一个容易忽略的坑是：不要一遇到异常就直接丢到忽略列表里，建议专门收集失败包名，最后统一重试一次：

failed_packages = [] def fetch_with_record(session, package_name): try: return fetch_package_detail(session, package_name), None except Exception as e: return None, (package_name, str(e))

跑完第一批之后，用failed_packages再跑第二轮。这个机制非常实用，第一次跑 5 万个包通常会失败几百个，第二轮重试之后基本都能补齐，不用为了这几百个包从头再跑一遍。

4.3 中文乱码与 Excel 打开问题

CSV 导出之后双击打开乱码，这几乎是每个用 Python 写 CSV 的人都会遇到一次的问题。原因前面说过，Excel 默认按 ANSI 本地编码打开 CSV，Python 默认写进去的是 UTF-8 无 BOM，两边对不上。解决方案就是写在open()里的encoding="utf-8-sig"。

如果已经导出了乱码文件，不用重跑采集，直接重新读一遍再改写就行：

with open("bad.csv", "r", encoding="utf-8") as f: content = f.read() with open("good.csv", "w", encoding="utf-8-sig") as f: f.write(content)

另外提醒一句，如果你用 pandas 的to_csv，同样要带上encoding="utf-8-sig"，否则一样会踩坑。

4.4 断点续采：避免从头再来

全量采集几十万包，耗时按小时算，如果中途脚本崩了或者网络断了，从头再来是最恼火的事情。所以这个项目一开始我就设计了断点续采机制，伤亡成本很低，实现也不复杂：每成功采集一个包就立即写入 CSV，不一次性留在内存里。

做法是迭代包名列表，边采边写，或者每隔 500 条 flush 一次。我常用的是用csv.writer逐行追加：

with open("pypi_packages.csv", "a", newline="", encoding="utf-8-sig") as f: writer = csv.writer(f) for pkg in package_list: row = fetch_package_detail(session, pkg) if row: writer.writerow([row.get(k, "") for k in FIELD_NAMES])

搭配一个本地done.txt，记录已成功采集的包名，下次启动时先读进来做一个 set，遇到重复包名直接跳过。这个方案比在内存里维护结果 list 再最后一次性导出要可靠得多，至少不用担心辛辛苦苦跑了两小时，一个记忆体炸了全盘皆输。

4.5 单线程太慢：并发数与线程池的平衡

单线程 for 循环跑几万个包，你可能会等得不耐烦。但一味加大线程数反而适得其反。线程池并发数最优解的判断标准很简单：看请求失败率。如果失败率低于 1%，说明并发数还有余量；如果超过 5%，那就是太激进了。

我实测的感受是，8 线程跑 PyPI JSON API 很稳，失败率基本在 0.2% 左右；16 线程偶尔出 429，但重试能兜住；再往上就不推荐了。如果你只是采集几千个热门的包做分析，4 线程都够用，没必要追求极端速度，毕竟数据采集的重点是数据质量，跑得快但一堆缺失和失败，后面清洗成本更高。

这套项目跑完，我最大的收获其实不是 CSV 文件本身，而是想明白了“采集公共服务数据”的边界感。PyPI 是开放的，但开放不等于可以无限索取，控制并发、做好本地缓存、只取自己需要的字段，这是尊重数据源的基本做法。最后再分享一个小技巧：CSV 导出来之后别急着删除原始 JSON 快照，留着raw/目录存一份原始响应，很多问题在分析阶段返工的时候，都能直接从原始数据里找到答案，不需要重新请求网络。