写这个采集脚本的起因其实很朴素我日常要做的 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 天”。数据源返回内容请求量速度适合场景JSON API单个包的完整元数据每包 1 次请求慢但精确采集详情、增量更新XML-RPC API包名列表、下载量排名、版本发布信息1 次请求拉全量列表快快速获取包名清单Simple Index全量包名和文件链接下载约 300MB 文本看带宽与本地索引同步、离线分析搜索页面/网页列表搜索结果页 HTML每页 1 次请求中等按关键词找小众库数据不稳定我最开始想的是直接抓 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( total5, backoff_factor0.5, status_forcelist[429, 500, 502, 503, 504], allowed_methods[GET], ) adapter HTTPAdapter(max_retriesretry, pool_connections10, pool_maxsize20) session.mount(https://, adapter)Retry 的backoff_factor0.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, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesfieldnames) 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(fTotal 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, start1): 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.ThreadPoolExecutorfrom concurrent.futures import ThreadPoolExecutor, as_completed def fetch_package_detail(session, package_name): url fhttps://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_workers8) 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的下载量数据那要把下载量和包名对应起来做一次 matchdownload_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, filenamepypi_packages.csv): with open(filename, w, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesFIELD_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(fExported {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 本地编码打开 CSVPython 默认写进去的是 UTF-8 无 BOM两边对不上。解决方案就是写在open()里的encodingutf-8-sig。如果已经导出了乱码文件不用重跑采集直接重新读一遍再改写就行with open(bad.csv, r, encodingutf-8) as f: content f.read() with open(good.csv, w, encodingutf-8-sig) as f: f.write(content)另外提醒一句如果你用 pandas 的to_csv同样要带上encodingutf-8-sig否则一样会踩坑。4.4 断点续采避免从头再来全量采集几十万包耗时按小时算如果中途脚本崩了或者网络断了从头再来是最恼火的事情。所以这个项目一开始我就设计了断点续采机制伤亡成本很低实现也不复杂每成功采集一个包就立即写入 CSV不一次性留在内存里。做法是迭代包名列表边采边写或者每隔 500 条 flush 一次。我常用的是用csv.writer逐行追加with open(pypi_packages.csv, a, newline, encodingutf-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/目录存一份原始响应很多问题在分析阶段返工的时候都能直接从原始数据里找到答案不需要重新请求网络。