简介这是一份面向具备Python基础与网络常识的学习者的网络数据爬虫代码集合旨在解决手动收集网页信息效率低、重复劳动多的问题。内容覆盖从requests请求、BeautifulSoup解析到CSV文件存储的常见抓取流程既有numbers、strings、list等基础语法片段也包含豆瓣读书简介、新东方头像下载、Glassdoor城市职位抓取等贴近真实场景的实战样例可支撑网页内容、电商产品、行情数据等自动化采集适用于数据分析师、NLP工程师、爬虫爱好者入门与参考。压缩包共21个文件以16个py脚本为主辅以2个txt文本与2个csv数据样例并含1张jpg图片整体大小仅34KB轻量便于按需拆解示例代码以独立脚本组织便于对照运行与二次修改。目前已有4379人学习下载。通过学习可快速掌握请求发送、页面解析、文件读写与结构化存储的完整链路同时了解爬取频率控制、法律合规及数据使用边界等注意事项。1. Python网络数据爬虫抓取代码先想清楚四件事再动手你搜“Python网络数据爬虫抓取代码”大概率是手上有个明确的数据源想把列表页、详情页里的字段落成表格。这一步本身不难真正让人翻车的是另外四件事目标站点的页面结构变了、反爬策略换了姿势、字段格式不统一、下游要的数据结构和抓回来的对不上。本文按我平时搭一个中小型爬虫的顺序来写——从一次 HTTP 请求开始到把数据写进 CSV 或 SQLite最后处理并发和请求频率。适合刚入门但希望代码能复用的开发者也适合在公司内部做临时数据采集的工程师。整篇代码都用 requests、BeautifulSoup 和标准库实现改一下选择器就能在本地跑通不需要引入重量级框架。2. 爬虫的底层逻辑从一次 HTTP 请求到表格里的一行数据2.1 一条数据的旅程HTTP 请求、响应与解析树先抛开代码讲流程。网络数据抓取的本质是四段接力构造请求、发送请求、解析响应、保存结果。大多数“爬虫抓不了”的报错都发生在第一段和第三段——要么请求没被服务器认可要么解析器没找对节点。常见的做法是先用 requests 发送 GET 请求import requests resp requests.get( https://demo-data.example/items, headers{User-Agent: Mozilla/5.0}, timeout10, ) print(resp.status_code) print(resp.text[:200])这段代码做了三件事用 requests.get 发出 GET 请求通过 headers 传递一个浏览器标识要求 10 秒内收到响应头否则直接抛超时异常。参数说明timeout 是超时控制只传一个数字表示建立连接和读取响应共用同一个上限。我习惯传元组 (3, 10)连接阶段等 3 秒读取响应体等 10 秒。这样某个慢接口不会拖死整个脚本但也不会因为网络抖动秒断。headers 里至少要有 User-Agent很多站点会拒绝默认的 python-requests 标识。拿到响应后下一步是把 HTML 文本变成可查询的解析树。用 BeautifulSoup 是入门最快的方式它把网页标签转成树结构再用 find 或 select 定位节点from bs4 import BeautifulSoup soup BeautifulSoup(resp.text, html.parser) title soup.select_one(h3.item-title) print(title.get_text(stripTrue) if title else 未找到节点)解析树的好处是你不必跟正则表达式纠缠。正则适合文本模式固定的场景但网页标签经常多一个属性、换一层嵌套正则就会漏匹配。用 CSS 选择器或 find 系列方法锚定的是 DOM 结构而不是纯文本容错率明显更高。2.2 选型requests 还是 httpxBeautifulSoup 还是 parsel新手最容易纠结两个选型问题我直接给结论中小型同步抓取用 requests要大规模异步并发再上 httpx。网页解析用 BeautifulSoup 起步单日抓取量到几十万条时换 parsel。requests 和 httpx 的核心差异在于异步支持和 API 风格。httpx 内置 async 接口适合 IO 密集型的大规模采集requests 生态最成熟遇到问题搜解决方案时资料最多。对绝大多数中小型任务requests 完全够用不必为了“看起来新”而引入 httpx。解析库这边BeautifulSoup 默认用 html.parser速度一般但容错好遇到不规范标签也能兜住。parsel 是基于 lxml 的类 Scrapy 选择器写法更接近 CSS 选择器解析速度高一个量级。如果目标是几万条以上的页面我建议直接用 parsel核心代码量几乎不变from parsel import Selector sel Selector(textresp.text) title sel.css(h3.item-title::text).get()这里的 get() 取第一个匹配getall() 取全部匹配。::text 表示只要文本节点不要外层 HTML 标签。对应到 BeautifulSoup就是 select_one().get_text() 和 select() 的差别。还有一处容易被忽略requests 的 resp.text 默认按响应头里的编码解码但很多站点响应头不写编码或写错。稳妥做法是显式指定编码或者先用 apparent_encoding 推断一次。这部分展开放在第 5 章避坑里因为乱码问题往往要到批量跑的时候才暴露。3. 写出第一版能跑通的中小型抓取脚本请求、解析与翻页3.1 最小可运行版本一个列表页的完整抓取代码我一般会先写一个“只抓一页”的版本确认解析逻辑没问题再往上加翻页。这一步如果跑不通直接打印响应内容看结构别急着写完整代码。import csv import requests from bs4 import BeautifulSoup BASE_URL https://demo-data.example/items?page1 HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Accept: text/html,application/xhtmlxml, } def fetch_page(url: str) - str: 请求页面并返回 HTML 文本 resp requests.get(url, headersHEADERS, timeout(3, 10)) resp.raise_for_status() return resp.text def parse_items(html: str) - list[dict]: 从 HTML 里提取标题和价格字段 soup BeautifulSoup(html, html.parser) rows [] for item in soup.select(div.row-item): title_node item.select_one(h3.title) price_node item.select_one(span.price) rows.append({ title: title_node.get_text(stripTrue) if title_node else , price: price_node.get_text(stripTrue) if price_node else , }) return rows def save_to_csv(rows: list[dict], path: str items.csv) - None: 追加写入 CSV首次写入自动补表头 if not rows: return with open(path, a, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnameslist(rows[0].keys())) if f.tell() 0: writer.writeheader() writer.writerows(rows) if __name__ __main__: html fetch_page(BASE_URL) data parse_items(html) save_to_csv(data) print(f本页抓取 {len(data)} 条)逻辑说明fetch_page 只负责请求和状态码检查parse_items 用 CSS 选择器先定位容器节点 div.row-item再在容器内找标题和价格节点save_to_csv 用追加模式写入避免重复抓取时覆盖之前的数据。参数说明newline 是 CSV 写入的固定写法否则 Windows 上每行之间会多一个空行。encoding 用 utf-8-sig这样用 Excel 打开 CSV 时中文不乱码如果下游是 Linux 文本处理管道建议改回 utf-8。raise_for_status() 会在状态码为 4xx/5xx 时直接抛异常省去自己判断状态码的代码。这段代码跑通后你手里就有一个可扩展的骨架。之后改字段、加翻页、换解析库都不需要重写整个流程。3.2 翻页与详情页抓取用循环和会话把代码串起来列表页一般只能拿到摘要字段完整数据经常在详情页里。常见做法是两层循环外层遍历页码内层遍历列表项对每个详情链接单独发一次请求。from urllib.parse import urljoin import time def fetch_page_session(session: requests.Session, url: str) - str: 用会话对象请求页面 resp session.get(url, timeout(3, 10)) resp.raise_for_status() return resp.text def fetch_detail(session: requests.Session, detail_url: str) - dict: 请求详情页并抽取描述字段 html fetch_page_session(session, detail_url) soup BeautifulSoup(html, html.parser) desc_node soup.select_one(div.description) return { desc: desc_node.get_text(stripTrue) if desc_node else , url: detail_url, } def crawl_all(max_page: int 5): session requests.Session() session.headers.update(HEADERS) all_rows [] for page in range(1, max_page 1): list_html fetch_page_session(session, fhttps://demo-data.example/items?page{page}) soup BeautifulSoup(list_html, html.parser) links [a[href] for a in soup.select(div.row-item a.detail-link)] for href in links: detail_url urljoin(https://demo-data.example, href) detail fetch_detail(session, detail_url) detail[page] page all_rows.append(detail) time.sleep(0.5) time.sleep(1) return all_rows逻辑说明Session 复用 TCP 连接和 Cookie比每次重新建连快很多urljoin 把相对路径补成完整 URL避免手拼字符串时漏掉协议或域名。参数说明time.sleep(0.5) 是详情页之间的等待时间建议不要小于 0.2目标站点有访问监控时尤其如此。max_page 控制抓取深度先设为 1 到 2 页验证数据质量再放开。3.3 请求头、超时与重试参数让代码在第一天下班后还能活着脚本跑几个小时就中断是常见问题。要稳定得把重试逻辑放进请求函数里而不是让它直接抛异常让整个任务中断。import time import requests from requests.adapters import HTTPAdapter def build_session(retries: int 3) - requests.Session: 构建带重试策略的会话 session requests.Session() adapter HTTPAdapter( max_retriesretries, pool_connections10, pool_maxsize20, ) session.mount(http://, adapter) session.mount(https://, adapter) return session逻辑说明HTTPAdapter 负责连接池和重试策略pool_connections 是缓存的主机连接数pool_maxsize 是每个主机的最大连接池大小。对中小型任务默认值够用不需要调得很大。配合循环调用时我会在连接池重试之外再加一层指数退避for attempt in range(3): try: resp session.get(url, timeout(3, 10)) resp.raise_for_status() break except requests.RequestException: if attempt 2: raise time.sleep(2 ** attempt)指数退避的参数含义第一次失败等 1 秒第二次等 2 秒第三次直接抛出异常。相比固定 sleep这种方式能应对瞬时波动也不会在服务器恢复前白白等待很长时间。注意重试只解决临时性网络问题。如果返回的是 403 或验证码页面重试多少次都没用要回到第 5 章的请求头方案去排查。4. 数据清洗与落盘选择抓下来只是开始能用的数据才算数4.1 三个高频脏数据场景缺失、空格、格式混排抓回来的数据十个里通常有两三个不能直接用。最常见的三种形态字段缺失导致 None 混入、文本首尾有换行和全角空格、价格或日期格式五花八门。我习惯在解析函数里顺手做清洗而不是落盘后再统一处理。def clean_text(value: str | None) - str: 压缩连续空白去首尾空格 if not value: return return .join(value.split()) def clean_price(value: str | None) - float | None: 把 1,299.00 这类文本转成 float if not value: return None digits .join(ch for ch in value if ch.isdigit() or ch .) try: return float(digits) except ValueError: return None逻辑说明clean_text 把换行、全角空格、连续空格统一压缩成单个 ASCII 空格clean_price 只保留数字和小数点再转成 float能处理“1,299.00”这类混排文本。参数说明这两个函数是纯函数不依赖外部状态可以放进独立工具模块里复用。注意 clean_price 处理不了“元/斤”这类带单位的后缀如果单位本身是另一个字段要先拆分再清洗。4.2 四种落盘方式对比CSV、JSON Lines、Excel、SQLite方式适合场景优点注意点CSV临时分析、交付给业务Excel 可直接打开、通用字段含逗号或换行时依赖 csv 模块转义JSON Lines日志采集、程序间传递一行一条可追加不适合人工阅读Excel给不写代码的同事交付格式直观、可加样式大批量写入慢几万行就明显卡顿SQLite增量抓取、频繁条件查询支持索引和去重需要写一点 SQL抓一千条以内的临时任务我首选 CSV调试最方便。抓几万条且后面要按条件过滤就换 SQLite。JSON Lines 适合把数据喂给下游程序Excel 只用于最终交付不要把它当主力存储。CSV 写入最稳妥的方式就是用 csv.DictWriter。字段里包含换行或逗号时csv 模块会自动加引号包裹不要手动拼逗号分隔的字符串。4.3 去重与增量抓取用一列主键挡住重复数据重复是批量抓取最头疼的问题。列表页偶尔会重复返回同一批数据不加去重第二次跑任务就多出双份记录。最简单的办法是维护一个已见过的主键集合seen: set[str] set() rows crawl_all(max_page2) fresh_rows [] for row in rows: key f{row[title]}:{row[url]} if key in seen: continue seen.add(key) fresh_rows.append(row)逻辑说明主键由标题和 URL 拼接而成能区分标题相同但链接不同的页面。如果目标站给每条数据提供了唯一 ID优先用 ID 做主键更简洁。如果落到 SQLite用 UNIQUE 约束做去重比内存集合更省心CREATE TABLE IF NOT EXISTS items ( id INTEGER PRIMARY KEY, title TEXT NOT NULL, url TEXT UNIQUE NOT NULL, price REAL, fetched_at TEXT );插入数据时用 INSERT OR IGNORE重复 URL 会被数据库直接拒绝。增量抓取场景下第一次全量之后只插入新 URL内存里不需要维护任何状态。5. 爬虫抓取避坑指南本地跑得好、一上线就翻车的 5 个典型问题这部分内容基本是我自己踩坑换来的血泪经验每一条都对应真实发生过的故障现象。5.1 403 与验证码请求头太敷衍是头号原因现象本地用 requests 直接 get状态码 403浏览器打开同一个地址却完全正常。 原因默认的 python-requests User-Agent 和缺头少尾的请求在服务端日志里一眼就能认出来轻则拒绝访问重则弹验证码。 解决把浏览器开发者工具里的完整请求头搬过来至少带 User-Agent、Accept、Accept-Language。有些站点还会校验 Referer 和 Origin遇到 403 时逐项补齐。只加一个 User-Agent 就反复重试容易把 IP 送上风险名单。5.2 请求太密被限制频率阈值和退避参数怎么设现象前 50 条数据正常第 60 条开始连续 503 或连接超时第二天再跑又能恢复。 原因服务端做了单位时间请求次数限制当前 IP 的请求速率超过了阈值。 解决把单次抓取改成“请求→等待→请求”的节拍。我常用的起步参数是详情页之间 sleep(0.5)列表页之间 sleep(1)跑完一轮看日志里的错误率再逐步收紧。需要更大吞吐时不要无脑加线程用第 6 章的线程池加限速组合。5.3 解析结果为空网页里藏着 JS 动态渲染现象解析代码在别的页面测试正常换成目标站点后 select 返回 None打印响应文本也搜不到目标内容。 原因数据不是服务端直接渲染在 HTML 里的而是浏览器执行 JavaScript 后从接口注入。 解决优先找页面背后的数据接口。打开浏览器网络面板筛选 Fetch/XHR找到返回 JSON 的那个请求直接抓接口而不是解析 HTML。这个方案比上无头浏览器轻量得多返回的还是结构化 JSON省掉解析步骤。确实没有现成接口时再考虑无头浏览器但并发能力和稳定性都会明显下降。5.4 数据乱码编码声明不一致导致的“口口”现象打印 resp.text 时中文变成乱码写进 CSV 后打开也是乱码。 原因requests 按响应头里的 charset 解码但一部分站点不声明 charset 或声明错误requests 会回退到 ISO-8859-1。 解决手动指定编码。先用 apparent_encoding 推断再赋值给 resp.encoding 后重新取 textresp.encoding resp.apparent_encoding text resp.text常见的中文站点编码是 utf-8 或 gb18030。gb18030 兼容 GBK比 gbk 更不容易翻车。还有少部分站点声明 utf-8 但实际混着 GBK 内容这种只能靠清洗函数做兜底没有万能解法最终以肉眼能读为准。5.5 登录态与重定向Cookie 过期导致抓到的不是同一张页面现象脚本运行一段时间后返回内容从数据页变成了登录页解析节点全部失效程序还在继续写入空数据。 原因目标站设置会话超时Cookie 过期后服务端把请求重定向到统一登录地址而 requests 默认会自动跟随重定向所以不会报错。 解决先手动登录一次把 Cookie 放进 requests.Session。可以手工从浏览器复制 Cookie 字符串也可以把登录流程脚本化。Session 会带着 Cookie 请求后续页面但要注意 Cookie 本身有时效。更稳健的做法是在解析结果里加一道校验发现页面标题不是预期格式就停止抓取避免把登录页当数据写进库。6. 进阶技巧并发抓取与请求频率控制的实测搭配6.1 并发下载的正确姿势线程池加限速中小型爬虫的单线程瓶颈是网络 IO用线程池能省一大半时间。但并发必须和频率控制绑在一起否则就是在替目标站管理员向自己的 IP 下手。from concurrent.futures import ThreadPoolExecutor, as_completed detail_urls [...] # 已经从列表页收集到全部详情链接 def fetch_one(detail_url: str) - dict: 抓取单个详情页内部自带等待 time.sleep(0.2) # 本地限速防止单线程打出过高频率 return fetch_detail(session, detail_url) with ThreadPoolExecutor(max_workers4) as executor: future_map {executor.submit(fetch_one, url): url for url in detail_urls} for future in as_completed(future_map): try: detail future.result() print(OK, detail[url]) except Exception as exc: print(FAIL, future_map[future], exc)max_workers 建议从 4 起步跑完一轮看错误率再决定是否调到 8。并发上去后每个线程内的 sleep 要相应调小否则整体请求速率会翻倍。这里的 time.sleep(0.2) 是每个线程内的等待间隔结合 4 个线程整体等效大约每秒 5 个请求已经接近多数常规站点容忍度的边缘。注意线程只有线程安全要求时直接用同一个 Session 即可requests 的 Session 在并发场景下是线程安全的但如果你在每个线程里各建一个 Session连接池会重新预热收益反而下降。6.2 验证别抓错用一个校验函数在落盘前做最后检查并发之后最容易出现的不是请求失败而是悄悄抓到错误数据。我习惯在落盘前加一道只读校验关键字段不为空、URL 属于预期域名、价格字段合法。三个条件全部通过才写入最终文件。def is_valid(row: dict) - bool: 落盘前的数据质量校验 return ( bool(row.get(title)) and row.get(url, ).startswith(https://demo-data.example) and row.get(price, 0) is not None )按这个思路每批数据落盘前先统计通过率通过率低于 95% 就抬头查日志而不是埋头改解析代码。做过几轮爬虫抓取后你会明白大多数“抓取结果不对”不是解析器写错了而是反爬策略换了姿势。早发现比事后补救省太多时间。我现在每次跑新目标站都先抓一页、跑校验、看通过率然后再放量。希望帮到你。本文还有配套的精品资源点击获取
简介这是一份面向具备Python基础与网络常识的学习者的网络数据爬虫代码集合旨在解决手动收集网页信息效率低、重复劳动多的问题。内容覆盖从requests请求、BeautifulSoup解析到CSV文件存储的常见抓取流程既有numbers、strings、list等基础语法片段也包含豆瓣读书简介、新东方头像下载、Glassdoor城市职位抓取等贴近真实场景的实战样例可支撑网页内容、电商产品、行情数据等自动化采集适用于数据分析师、NLP工程师、爬虫爱好者入门与参考。压缩包共21个文件以16个py脚本为主辅以2个txt文本与2个csv数据样例并含1张jpg图片整体大小仅34KB轻量便于按需拆解示例代码以独立脚本组织便于对照运行与二次修改。目前已有4379人学习下载。通过学习可快速掌握请求发送、页面解析、文件读写与结构化存储的完整链路同时了解爬取频率控制、法律合规及数据使用边界等注意事项。1. Python网络数据爬虫抓取代码先想清楚四件事再动手你搜“Python网络数据爬虫抓取代码”大概率是手上有个明确的数据源想把列表页、详情页里的字段落成表格。这一步本身不难真正让人翻车的是另外四件事目标站点的页面结构变了、反爬策略换了姿势、字段格式不统一、下游要的数据结构和抓回来的对不上。本文按我平时搭一个中小型爬虫的顺序来写——从一次 HTTP 请求开始到把数据写进 CSV 或 SQLite最后处理并发和请求频率。适合刚入门但希望代码能复用的开发者也适合在公司内部做临时数据采集的工程师。整篇代码都用 requests、BeautifulSoup 和标准库实现改一下选择器就能在本地跑通不需要引入重量级框架。2. 爬虫的底层逻辑从一次 HTTP 请求到表格里的一行数据2.1 一条数据的旅程HTTP 请求、响应与解析树先抛开代码讲流程。网络数据抓取的本质是四段接力构造请求、发送请求、解析响应、保存结果。大多数“爬虫抓不了”的报错都发生在第一段和第三段——要么请求没被服务器认可要么解析器没找对节点。常见的做法是先用 requests 发送 GET 请求import requests resp requests.get( https://demo-data.example/items, headers{User-Agent: Mozilla/5.0}, timeout10, ) print(resp.status_code) print(resp.text[:200])这段代码做了三件事用 requests.get 发出 GET 请求通过 headers 传递一个浏览器标识要求 10 秒内收到响应头否则直接抛超时异常。参数说明timeout 是超时控制只传一个数字表示建立连接和读取响应共用同一个上限。我习惯传元组 (3, 10)连接阶段等 3 秒读取响应体等 10 秒。这样某个慢接口不会拖死整个脚本但也不会因为网络抖动秒断。headers 里至少要有 User-Agent很多站点会拒绝默认的 python-requests 标识。拿到响应后下一步是把 HTML 文本变成可查询的解析树。用 BeautifulSoup 是入门最快的方式它把网页标签转成树结构再用 find 或 select 定位节点from bs4 import BeautifulSoup soup BeautifulSoup(resp.text, html.parser) title soup.select_one(h3.item-title) print(title.get_text(stripTrue) if title else 未找到节点)解析树的好处是你不必跟正则表达式纠缠。正则适合文本模式固定的场景但网页标签经常多一个属性、换一层嵌套正则就会漏匹配。用 CSS 选择器或 find 系列方法锚定的是 DOM 结构而不是纯文本容错率明显更高。2.2 选型requests 还是 httpxBeautifulSoup 还是 parsel新手最容易纠结两个选型问题我直接给结论中小型同步抓取用 requests要大规模异步并发再上 httpx。网页解析用 BeautifulSoup 起步单日抓取量到几十万条时换 parsel。requests 和 httpx 的核心差异在于异步支持和 API 风格。httpx 内置 async 接口适合 IO 密集型的大规模采集requests 生态最成熟遇到问题搜解决方案时资料最多。对绝大多数中小型任务requests 完全够用不必为了“看起来新”而引入 httpx。解析库这边BeautifulSoup 默认用 html.parser速度一般但容错好遇到不规范标签也能兜住。parsel 是基于 lxml 的类 Scrapy 选择器写法更接近 CSS 选择器解析速度高一个量级。如果目标是几万条以上的页面我建议直接用 parsel核心代码量几乎不变from parsel import Selector sel Selector(textresp.text) title sel.css(h3.item-title::text).get()这里的 get() 取第一个匹配getall() 取全部匹配。::text 表示只要文本节点不要外层 HTML 标签。对应到 BeautifulSoup就是 select_one().get_text() 和 select() 的差别。还有一处容易被忽略requests 的 resp.text 默认按响应头里的编码解码但很多站点响应头不写编码或写错。稳妥做法是显式指定编码或者先用 apparent_encoding 推断一次。这部分展开放在第 5 章避坑里因为乱码问题往往要到批量跑的时候才暴露。3. 写出第一版能跑通的中小型抓取脚本请求、解析与翻页3.1 最小可运行版本一个列表页的完整抓取代码我一般会先写一个“只抓一页”的版本确认解析逻辑没问题再往上加翻页。这一步如果跑不通直接打印响应内容看结构别急着写完整代码。import csv import requests from bs4 import BeautifulSoup BASE_URL https://demo-data.example/items?page1 HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Accept: text/html,application/xhtmlxml, } def fetch_page(url: str) - str: 请求页面并返回 HTML 文本 resp requests.get(url, headersHEADERS, timeout(3, 10)) resp.raise_for_status() return resp.text def parse_items(html: str) - list[dict]: 从 HTML 里提取标题和价格字段 soup BeautifulSoup(html, html.parser) rows [] for item in soup.select(div.row-item): title_node item.select_one(h3.title) price_node item.select_one(span.price) rows.append({ title: title_node.get_text(stripTrue) if title_node else , price: price_node.get_text(stripTrue) if price_node else , }) return rows def save_to_csv(rows: list[dict], path: str items.csv) - None: 追加写入 CSV首次写入自动补表头 if not rows: return with open(path, a, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnameslist(rows[0].keys())) if f.tell() 0: writer.writeheader() writer.writerows(rows) if __name__ __main__: html fetch_page(BASE_URL) data parse_items(html) save_to_csv(data) print(f本页抓取 {len(data)} 条)逻辑说明fetch_page 只负责请求和状态码检查parse_items 用 CSS 选择器先定位容器节点 div.row-item再在容器内找标题和价格节点save_to_csv 用追加模式写入避免重复抓取时覆盖之前的数据。参数说明newline 是 CSV 写入的固定写法否则 Windows 上每行之间会多一个空行。encoding 用 utf-8-sig这样用 Excel 打开 CSV 时中文不乱码如果下游是 Linux 文本处理管道建议改回 utf-8。raise_for_status() 会在状态码为 4xx/5xx 时直接抛异常省去自己判断状态码的代码。这段代码跑通后你手里就有一个可扩展的骨架。之后改字段、加翻页、换解析库都不需要重写整个流程。3.2 翻页与详情页抓取用循环和会话把代码串起来列表页一般只能拿到摘要字段完整数据经常在详情页里。常见做法是两层循环外层遍历页码内层遍历列表项对每个详情链接单独发一次请求。from urllib.parse import urljoin import time def fetch_page_session(session: requests.Session, url: str) - str: 用会话对象请求页面 resp session.get(url, timeout(3, 10)) resp.raise_for_status() return resp.text def fetch_detail(session: requests.Session, detail_url: str) - dict: 请求详情页并抽取描述字段 html fetch_page_session(session, detail_url) soup BeautifulSoup(html, html.parser) desc_node soup.select_one(div.description) return { desc: desc_node.get_text(stripTrue) if desc_node else , url: detail_url, } def crawl_all(max_page: int 5): session requests.Session() session.headers.update(HEADERS) all_rows [] for page in range(1, max_page 1): list_html fetch_page_session(session, fhttps://demo-data.example/items?page{page}) soup BeautifulSoup(list_html, html.parser) links [a[href] for a in soup.select(div.row-item a.detail-link)] for href in links: detail_url urljoin(https://demo-data.example, href) detail fetch_detail(session, detail_url) detail[page] page all_rows.append(detail) time.sleep(0.5) time.sleep(1) return all_rows逻辑说明Session 复用 TCP 连接和 Cookie比每次重新建连快很多urljoin 把相对路径补成完整 URL避免手拼字符串时漏掉协议或域名。参数说明time.sleep(0.5) 是详情页之间的等待时间建议不要小于 0.2目标站点有访问监控时尤其如此。max_page 控制抓取深度先设为 1 到 2 页验证数据质量再放开。3.3 请求头、超时与重试参数让代码在第一天下班后还能活着脚本跑几个小时就中断是常见问题。要稳定得把重试逻辑放进请求函数里而不是让它直接抛异常让整个任务中断。import time import requests from requests.adapters import HTTPAdapter def build_session(retries: int 3) - requests.Session: 构建带重试策略的会话 session requests.Session() adapter HTTPAdapter( max_retriesretries, pool_connections10, pool_maxsize20, ) session.mount(http://, adapter) session.mount(https://, adapter) return session逻辑说明HTTPAdapter 负责连接池和重试策略pool_connections 是缓存的主机连接数pool_maxsize 是每个主机的最大连接池大小。对中小型任务默认值够用不需要调得很大。配合循环调用时我会在连接池重试之外再加一层指数退避for attempt in range(3): try: resp session.get(url, timeout(3, 10)) resp.raise_for_status() break except requests.RequestException: if attempt 2: raise time.sleep(2 ** attempt)指数退避的参数含义第一次失败等 1 秒第二次等 2 秒第三次直接抛出异常。相比固定 sleep这种方式能应对瞬时波动也不会在服务器恢复前白白等待很长时间。注意重试只解决临时性网络问题。如果返回的是 403 或验证码页面重试多少次都没用要回到第 5 章的请求头方案去排查。4. 数据清洗与落盘选择抓下来只是开始能用的数据才算数4.1 三个高频脏数据场景缺失、空格、格式混排抓回来的数据十个里通常有两三个不能直接用。最常见的三种形态字段缺失导致 None 混入、文本首尾有换行和全角空格、价格或日期格式五花八门。我习惯在解析函数里顺手做清洗而不是落盘后再统一处理。def clean_text(value: str | None) - str: 压缩连续空白去首尾空格 if not value: return return .join(value.split()) def clean_price(value: str | None) - float | None: 把 1,299.00 这类文本转成 float if not value: return None digits .join(ch for ch in value if ch.isdigit() or ch .) try: return float(digits) except ValueError: return None逻辑说明clean_text 把换行、全角空格、连续空格统一压缩成单个 ASCII 空格clean_price 只保留数字和小数点再转成 float能处理“1,299.00”这类混排文本。参数说明这两个函数是纯函数不依赖外部状态可以放进独立工具模块里复用。注意 clean_price 处理不了“元/斤”这类带单位的后缀如果单位本身是另一个字段要先拆分再清洗。4.2 四种落盘方式对比CSV、JSON Lines、Excel、SQLite方式适合场景优点注意点CSV临时分析、交付给业务Excel 可直接打开、通用字段含逗号或换行时依赖 csv 模块转义JSON Lines日志采集、程序间传递一行一条可追加不适合人工阅读Excel给不写代码的同事交付格式直观、可加样式大批量写入慢几万行就明显卡顿SQLite增量抓取、频繁条件查询支持索引和去重需要写一点 SQL抓一千条以内的临时任务我首选 CSV调试最方便。抓几万条且后面要按条件过滤就换 SQLite。JSON Lines 适合把数据喂给下游程序Excel 只用于最终交付不要把它当主力存储。CSV 写入最稳妥的方式就是用 csv.DictWriter。字段里包含换行或逗号时csv 模块会自动加引号包裹不要手动拼逗号分隔的字符串。4.3 去重与增量抓取用一列主键挡住重复数据重复是批量抓取最头疼的问题。列表页偶尔会重复返回同一批数据不加去重第二次跑任务就多出双份记录。最简单的办法是维护一个已见过的主键集合seen: set[str] set() rows crawl_all(max_page2) fresh_rows [] for row in rows: key f{row[title]}:{row[url]} if key in seen: continue seen.add(key) fresh_rows.append(row)逻辑说明主键由标题和 URL 拼接而成能区分标题相同但链接不同的页面。如果目标站给每条数据提供了唯一 ID优先用 ID 做主键更简洁。如果落到 SQLite用 UNIQUE 约束做去重比内存集合更省心CREATE TABLE IF NOT EXISTS items ( id INTEGER PRIMARY KEY, title TEXT NOT NULL, url TEXT UNIQUE NOT NULL, price REAL, fetched_at TEXT );插入数据时用 INSERT OR IGNORE重复 URL 会被数据库直接拒绝。增量抓取场景下第一次全量之后只插入新 URL内存里不需要维护任何状态。5. 爬虫抓取避坑指南本地跑得好、一上线就翻车的 5 个典型问题这部分内容基本是我自己踩坑换来的血泪经验每一条都对应真实发生过的故障现象。5.1 403 与验证码请求头太敷衍是头号原因现象本地用 requests 直接 get状态码 403浏览器打开同一个地址却完全正常。 原因默认的 python-requests User-Agent 和缺头少尾的请求在服务端日志里一眼就能认出来轻则拒绝访问重则弹验证码。 解决把浏览器开发者工具里的完整请求头搬过来至少带 User-Agent、Accept、Accept-Language。有些站点还会校验 Referer 和 Origin遇到 403 时逐项补齐。只加一个 User-Agent 就反复重试容易把 IP 送上风险名单。5.2 请求太密被限制频率阈值和退避参数怎么设现象前 50 条数据正常第 60 条开始连续 503 或连接超时第二天再跑又能恢复。 原因服务端做了单位时间请求次数限制当前 IP 的请求速率超过了阈值。 解决把单次抓取改成“请求→等待→请求”的节拍。我常用的起步参数是详情页之间 sleep(0.5)列表页之间 sleep(1)跑完一轮看日志里的错误率再逐步收紧。需要更大吞吐时不要无脑加线程用第 6 章的线程池加限速组合。5.3 解析结果为空网页里藏着 JS 动态渲染现象解析代码在别的页面测试正常换成目标站点后 select 返回 None打印响应文本也搜不到目标内容。 原因数据不是服务端直接渲染在 HTML 里的而是浏览器执行 JavaScript 后从接口注入。 解决优先找页面背后的数据接口。打开浏览器网络面板筛选 Fetch/XHR找到返回 JSON 的那个请求直接抓接口而不是解析 HTML。这个方案比上无头浏览器轻量得多返回的还是结构化 JSON省掉解析步骤。确实没有现成接口时再考虑无头浏览器但并发能力和稳定性都会明显下降。5.4 数据乱码编码声明不一致导致的“口口”现象打印 resp.text 时中文变成乱码写进 CSV 后打开也是乱码。 原因requests 按响应头里的 charset 解码但一部分站点不声明 charset 或声明错误requests 会回退到 ISO-8859-1。 解决手动指定编码。先用 apparent_encoding 推断再赋值给 resp.encoding 后重新取 textresp.encoding resp.apparent_encoding text resp.text常见的中文站点编码是 utf-8 或 gb18030。gb18030 兼容 GBK比 gbk 更不容易翻车。还有少部分站点声明 utf-8 但实际混着 GBK 内容这种只能靠清洗函数做兜底没有万能解法最终以肉眼能读为准。5.5 登录态与重定向Cookie 过期导致抓到的不是同一张页面现象脚本运行一段时间后返回内容从数据页变成了登录页解析节点全部失效程序还在继续写入空数据。 原因目标站设置会话超时Cookie 过期后服务端把请求重定向到统一登录地址而 requests 默认会自动跟随重定向所以不会报错。 解决先手动登录一次把 Cookie 放进 requests.Session。可以手工从浏览器复制 Cookie 字符串也可以把登录流程脚本化。Session 会带着 Cookie 请求后续页面但要注意 Cookie 本身有时效。更稳健的做法是在解析结果里加一道校验发现页面标题不是预期格式就停止抓取避免把登录页当数据写进库。6. 进阶技巧并发抓取与请求频率控制的实测搭配6.1 并发下载的正确姿势线程池加限速中小型爬虫的单线程瓶颈是网络 IO用线程池能省一大半时间。但并发必须和频率控制绑在一起否则就是在替目标站管理员向自己的 IP 下手。from concurrent.futures import ThreadPoolExecutor, as_completed detail_urls [...] # 已经从列表页收集到全部详情链接 def fetch_one(detail_url: str) - dict: 抓取单个详情页内部自带等待 time.sleep(0.2) # 本地限速防止单线程打出过高频率 return fetch_detail(session, detail_url) with ThreadPoolExecutor(max_workers4) as executor: future_map {executor.submit(fetch_one, url): url for url in detail_urls} for future in as_completed(future_map): try: detail future.result() print(OK, detail[url]) except Exception as exc: print(FAIL, future_map[future], exc)max_workers 建议从 4 起步跑完一轮看错误率再决定是否调到 8。并发上去后每个线程内的 sleep 要相应调小否则整体请求速率会翻倍。这里的 time.sleep(0.2) 是每个线程内的等待间隔结合 4 个线程整体等效大约每秒 5 个请求已经接近多数常规站点容忍度的边缘。注意线程只有线程安全要求时直接用同一个 Session 即可requests 的 Session 在并发场景下是线程安全的但如果你在每个线程里各建一个 Session连接池会重新预热收益反而下降。6.2 验证别抓错用一个校验函数在落盘前做最后检查并发之后最容易出现的不是请求失败而是悄悄抓到错误数据。我习惯在落盘前加一道只读校验关键字段不为空、URL 属于预期域名、价格字段合法。三个条件全部通过才写入最终文件。def is_valid(row: dict) - bool: 落盘前的数据质量校验 return ( bool(row.get(title)) and row.get(url, ).startswith(https://demo-data.example) and row.get(price, 0) is not None )按这个思路每批数据落盘前先统计通过率通过率低于 95% 就抬头查日志而不是埋头改解析代码。做过几轮爬虫抓取后你会明白大多数“抓取结果不对”不是解析器写错了而是反爬策略换了姿势。早发现比事后补救省太多时间。我现在每次跑新目标站都先抓一页、跑校验、看通过率然后再放量。希望帮到你。本文还有配套的精品资源点击获取
