1. 本课定位是什么、为何重要从 33 到 38你分别学会了 HTTP、稳健客户端、并发选型、协程、HTML 解析、接口优先。但它们若一直是散落的 demo仍然不像「能交付的东西」。工作里老板要的是输入源列表输出库里可查的数据并且跑第二遍不能乱套。这一课做迷你采集流水线配置源→拉 JSON→抽字段→SQLite 去重更新→汇总打印。明确不做 Selenium 和分布式。学完你应能讲清模块怎么拆、如何验收、如何和前几课能力一一对应。把3338收成一条可演示流水线配置源 → 拉取 JSON → 抽取字段 → 写入 SQLite去重/更新→ 汇总打印已学能力用在本项目Day33/34 HTTP requestsfetchDay35/36 并发可选扩展多源可改 gather/线程池Day37 解析/清洗直觉从 dict 取字段、校验Day38 接口优先不爬整页 HTMLDay2432 数据库习惯参数绑定、upsert、commit为何重要真实工作里「能跑通的端到端」比零散知识点更有说服力也练习模块边界与验收标准。本课不做Selenium、分布式爬虫、复杂调度系统。2. 流水线总览图先给一张总图SOURCES 进来经 fetch、parse、store 进 SQLite第二遍 upsert 不增行最后 COUNT 验收。有了图后面每一节都是在填某个方框的细节不会迷路。SOURCES 列表 | v ------- ------- ------- | fetch | -- | parse | -- | store | -- SQLite items ------- ------- ------- ^ | | 第二遍同源 | --------- upsert 不增行 ----- | v COUNT / 打印 rows 验收阶段输入输出fetchurldictJSONparsedictname, starsstoresource, name, stars表中 1 行插入或更新report连接COUNT 明细3. 本质各段职责与失败流水线本质是分段每段输入输出是什么、失败时怎么办。错误设计每次 INSERT 新行跑两遍变四行正确设计 UNIQUEUPSERT仍两行只刷新。把「修改前/后」记牢验收时才知道 COUNT 该看什么。步骤职责失败时fetchHTTP GETtimeout、UA网络错 / 4xx5xx → 抛错或重试parse取full_name、stargazers_count缺字段 → 降级空串/0 或跳过store写入同源更新SQL 错 → 事务回滚本示例简单 commitrun编排多源第二遍验证去重打印 COUNT 对照预期第一遍抓取后N 个源 → N 行。第二遍同源再抓仍是 N 行ON CONFLICT DO UPDATE不是 2N 行。修改前错误设计每次 INSERT 新行跑两遍变成 4 行重复源。修改后UNIQUE UPSERT仍 2 行字段刷新。4. 约束、坑与合规公开 API 有限流密钥不能进仓唯一键要设计对不要把 HTML 爬虫硬塞进本课目标。合规清单与安全课精神一致timeout、重试、UA、不采隐私。技术正确包含合规正确。约束/坑说明公开 API 限流控制频率设合理 timeoutToken 不进仓库需要鉴权时用环境变量唯一键设计sourceUNIQUE 才能稳定去重把 HTML 爬虫硬塞进本项目偏离「接口优先」教学目标无raise_for_status错误页被当数据star 写死断言真实数据会变断言会误伤合规清单项目级优先官方/公开 APItimeout 有限重试不把密钥写进 Git不采未授权隐私遵守对方 rate limitUser-Agent 标明自己的教学客户端5. 表设计详解items 表字段为何这样设source 逻辑名唯一name/stars 业务字段fetched_at 记录新鲜度。为什么不用 URL 当唯一键URL 易变、带 query。逻辑源名更稳、更好读。CREATETABLEitems(idINTEGERPRIMARYKEYAUTOINCREMENT,sourceTEXTNOTNULLUNIQUE,nameTEXTNOTNULL,starsINTEGERNOTNULL,fetched_atREALNOTNULL);列含义备注id自增主键内部用source逻辑源 id如github-cpython业务唯一name仓库全名如python/cpythonstarsstar 数会变upsert 更新fetched_at抓取时间戳time.time()为什么 source 用逻辑名而不是 URLURL 可能带 query、换域名逻辑名更稳定也好读。6. 能力分组代码怎么切init_db、fetch、store、main 各干什么对应测试与阅读友好。UPSERT SQL 在说什么用表解释插入与更新两种情况。函数短、SQL 参数化承接 Day28/32 习惯避免一个 200 行脚本揉成一团。函数输入输出注意init_db连接建表IF NOT EXISTSfetchurldicttimeout UA raise_for_statusstoresource, name, stars写入/更新参数绑定 UPSERTmain源列表打印过程与 COUNT可第二遍保持函数短——一个函数只做一件事便于测fetch/store。6.1 UPSERT 在说什么INSERTINTOitems(source,name,stars,fetched_at)VALUES(?,?,?,?)ONCONFLICT(source)DOUPDATESETnameexcluded.name,starsexcluded.stars,fetched_atexcluded.fetched_at情况行为source 不存在插入新行source 已存在更新 name/stars/fetched_at7. 数据源设计SOURCES 用逻辑名, URL元组列表扩展第三源只加一行。验收 COUNT 与源数量绑定改源列表时预期也要改——这是可验收的关键。SOURCES[(github-cpython,https://api.github.com/repos/python/cpython),(github-requests,https://api.github.com/repos/psf/requests),]字段含义元组第一项逻辑 source元组第二项真实请求 URL扩展第三个源再加一行元组即可验收 COUNT 变 3。8. 综合实践一键脚本跑 GitHub 两仓库第二遍 upsert看 COUNT2。star 数会变正常。请真实运行。这是阶段收官的主证据你能端到端交付而不只是会单课语法。mkdir-p~/python-lab/src/day39cd~/python-lab/src/day39 pipinstallrequestsWindows 推荐 Cygwin 或 WSL。catpipeline_demo.pyEOF # Day39: mini pipeline API sqlite import sqlite3 import time from pathlib import Path import requests DIR Path(__file__).resolve().parent DB DIR / pipeline.db SOURCES [ (github-cpython, https://api.github.com/repos/python/cpython), (github-requests, https://api.github.com/repos/psf/requests), ] def init_db(conn): conn.execute( CREATE TABLE IF NOT EXISTS items ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT NOT NULL, name TEXT NOT NULL, stars INTEGER NOT NULL, fetched_at REAL NOT NULL, UNIQUE(source) ) ) conn.commit() def fetch(url: str) - dict: r requests.get( url, timeout20, headers{User-Agent: python-lab-day39/1.0, Accept: application/json}, ) r.raise_for_status() return r.json() def store(conn, source: str, name: str, stars: int): conn.execute( INSERT INTO items (source, name, stars, fetched_at) VALUES (?, ?, ?, ?) ON CONFLICT(source) DO UPDATE SET nameexcluded.name, starsexcluded.stars, fetched_atexcluded.fetched_at , (source, name, stars, time.time()), ) conn.commit() def main(): if DB.exists(): DB.unlink() with sqlite3.connect(DB) as conn: init_db(conn) for source, url in SOURCES: data fetch(url) name data.get(full_name) or data.get(name) or stars int(data.get(stargazers_count) or 0) print(ffetched {source}: {name} stars{stars}) store(conn, source, name, stars) print(--- second pass (upsert same sources) ---) for source, url in SOURCES: data fetch(url) name data.get(full_name) or data.get(name) or stars int(data.get(stargazers_count) or 0) store(conn, source, name, stars) n conn.execute(SELECT COUNT(*) FROM items).fetchone()[0] print(--- count ---) print(n) print(--- rows ---) for row in conn.execute( SELECT id, source, name, stars FROM items ORDER BY id ): print(row) if __name__ __main__: main() EOFpython3 pipeline_demo.py实测输出star 数会变fetched github-cpython: python/cpython stars73867 fetched github-requests: psf/requests stars54147 --- second pass (upsert same sources) --- --- count --- 2 --- rows --- (1, github-cpython, python/cpython, 73867) (2, github-requests, psf/requests, 54147)第二遍前已有 2 行。第二遍后仍是 2 行字段被刷新——不是插入重复源。9. 验收清单务必过一遍用清单当「作业评分表」跑通、COUNT、去重、参数化与 UA、合规、错误可见。全部勾上本课才算完成而不是脚本能 print 就行。一行命令跑通COUNT(*) 源数量本例 2第二遍不产生重复source全程参数化 SQL、有 timeout 与 UA合规公开 API 控制频率失败时可手动改错 URL能看到异常而不是静默脏数据10. 可选扩展作业方向异步、线程池、重试、Redis、配置文件、日志——指向 35/36/30 等课。强调先串行正确再并发。顺序反了会放大错误排障更痛苦。扩展做法对应课异步拉取httpxasyncio.gatherDay36线程池拉取ThreadPoolExecutorDay35失败重试for sleep timeoutDay34Redis 去重/缓存记 source 或缓存 JSONDay30配置文件sources.yaml工程化日志logging 打到文件运维习惯扩展时先保证串行正确再加并发——并发会放大错误。11. 错误处理怎么加示例思路给 fetch 加重试循环的示例思路让短暂网络抖动不至于整条挂掉。修改前一次失败就崩修改后可恢复仍失败再抛——工程上更常见。deffetch(url:str)-dict:last_errNoneforattemptinrange(1,4):try:rrequests.get(url,timeout20,headers{...})r.raise_for_status()returnr.json()exceptrequests.RequestExceptionase:last_erre time.sleep(0.5*attempt)raiselast_err修改前一次网络抖就整条流水线挂。修改后短暂故障可恢复仍失败再抛出。12. 常见问答为何第二遍还请求、COUNT 不是 2 怎么办、403、能否改爬 HTML——集中答疑。帮助你独立排障而不是一出错就怀疑整课概念。Q为什么第二遍还要请求网络A演示 upsert 与「刷新 stars」生产可按 TTL 决定是否重抓。QCOUNT 不是 2A检查是否删库失败、是否改过 SOURCES、是否旧 DB 未删。Q403/rate limitA降频、加 Token、换时段不要死循环猛打。Q能否改成爬 HTMLA能但本课教学目标是接口流水线HTML 请回到 Day37 思路。13. 和「脚本随便写写」的差别对照随便写与本课结构分离模块、upsert、验收、密钥。这是从「练手」到「像项目」的差别列表写简历项目时也能用这套说法。随便写本课结构请求和 SQL 揉在一起fetch/store 分离每次插入新行唯一键 upsert无验收COUNT 第二遍密钥写死环境变量扩展14. 阶段收官对照一张表回顾 3339 你带走的能力形成阶段闭环。网络与异步不是终点而是能取数、能叠等待、能入库的底座。课你带走的能力33HTTP 五件套34稳健 requests 客户端35并发选型36asyncio httpx37HTML 解析入库38接口优先39端到端流水线15. 模块边界什么样叫「好拆」好拆与坏拆对照强调测试友好fetch 可 mock不必事事连外网。为以后写更大项目留接口意识。好不好fetch只负责 HTTPfetch里又写 SQL 又 print 排版store只负责写入全局到处sqlite3.connectmain只编排一个 200 行函数从头写到尾测试友好可以对fetch做 mock不必真连外网进阶。16. 验收脚本片段人工也可用 assert 锁住行数与源集合但不要断言 star 具体数字。把「顺眼」升级成「可自动检查」的一小步。nconn.execute(SELECT COUNT(*) FROM items).fetchone()[0]assertnlen(SOURCES)sources{row[0]forrowinconn.execute(SELECT source FROM items)}assertsources{sfors,_inSOURCES}修改前只看 print 是否「顺眼」。修改后用断言锁住「行数与源集合」。注意不要断言 star 的具体数字。17. 并发改造草图可选接 Day35/36线程池与 async 改造直觉并警告 SQLite 多线程写连接不安全。再次强调先串行正确再并发——和 35/36 课的选型、限流呼应。线程池版直觉# 伪代码withThreadPoolExecutor(max_workers4)asex:futs{ex.submit(fetch,url):sourceforsource,urlinSOURCES}forfutinas_completed(futs):sourcefuts[fut]datafut.result()store(conn,source,...)注意SQLite 多线程写同一连接不安全应「主线程统一 store」或每任务短连接并小心锁。async 版直觉gather多个 fetch回到主协程再 store。先串行正确再并发——顺序不要反。18. 故障演练建议错 URL、断网、第二遍、加源——主动演练建立预期。故障演练是工程习惯不是可选闲聊。演练操作期望错 URL改成不存在的仓库非 200进程报错退出断网拔网/防火墙超时或连接错误第二遍连续跑两轮COUNT 不变加源SOURCES 加一项COUNT119. 自我检查清单打勾图、UPSERT、COUNT、timeout/UA/绑定、扩展与陷阱。全过则本阶段实践目标达成。能画出 fetch → parse → store会 UPSERT 去重会用 COUNT 验收timeout/UA/参数绑定齐全知道可选扩展与并发陷阱20. 字段字典本项目逻辑名、JSON 来源、SQLite 列对照外加解析两行示例。换数据源时先填这种字典再写代码少返工。逻辑名JSON 来源SQLite 列源标识自定 SOURCES[0]source仓库名full_name / namenameStarstargazers_countstars抓取时间time.time()fetched_at解析示例namedata.get(full_name)ordata.get(name)orstarsint(data.get(stargazers_count)or0)21. 运行结果怎么读对照实测逐段解读 fetched 行、second pass、count、rowsCOUNT 为 4 时如何排查。会读输出才会判断实验成功还是环境/代码问题。fetched github-cpython: python/cpython stars73867片段含义fetched …fetchparse 成功stars数字当前公开 star会变second pass再次 upsertcount 2去重成功rows 两行与 SOURCES 一一对应若 count 为 4说明唯一约束没生效或跑了两次建库逻辑异常——检查UNIQUE(source)与是否每次unlinkDB。22. 提交作业前检查给学员交作业应贴完整输出、说明 SOURCES 与第二遍 COUNT、扩展另说、禁止贴 Token。这是课堂规范也是职业沟通的缩影。贴运行完整终端输出说明 SOURCES 列表说明第二遍 COUNT若做了扩展重试/异步单独说明不要贴 Token总结流水线四段拉、解析、存、汇总接口优先阶段收官。可重复可验收比「写出过一次请求」更重要。你可以继续走向 Web 服务对外提供 API或数据分析消费已入库数据。流水线 拉 → 解析 → 存 → 汇总。接口优先降低解析成本SQLite 承接持久化能力。「网络与异步」阶段收官HTTP → 客户端 → 并发 → 解析/接口 → 项目。可重复、可验收比「写出过一次请求」更重要。小练笔自测含 upsert、验收、安全与可选加源实践。先做后看答案。可选实践请真加第三源并验证 COUNT。题 1为什么第二遍 count 仍是 2题 2源改成 3 个仓库且均成功时验收 count 应是题 3可选为fetch增加失败重试 2 次思路即可。题 4判断本流水线必须以 Selenium 打开 GitHub 页面才能取 star。题 5store里用?占位的主要安全收益是什么题 6source列为什么建议 UNIQUE题 7raise_for_status放在fetch里而不是忽略状态码好处是题 8判断应用assert stars 73867作为长期自动化测试。题 9可选实践增加第三个公开仓库源确认 COUNT 为 3且第二遍仍为 3。题 10列出流水线四个阶段名称中文或英文。小练笔参考答案先独立完成。意思对即可。与 UPSERT、合规、参数绑定冲突的理解需修正。题 1UNIQUE(source)ON CONFLICT DO UPDATE同源更新不新增行。题 23题 3for attempt in range(3): try/except包裹requests.get失败 sleep 再试。题 4错公开 REST API 即可。题 5降低 SQL 注入风险参数与语句分离。题 6保证同一逻辑源只有一行支撑 upsert 去重。题 7尽早发现 4xx/5xx避免把错误响应当业务 JSON。题 8错star 会变应断言类型/存在性或允许范围。题 9以你运行为准。题 10拉取fetch、解析parse、存储store、汇总/验收report。合理即可