简介：这是一份面向具备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"}, timeout=10, ) 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(strip=True) if title else "未找到节点")解析树的好处是你不必跟正则表达式纠缠。正则适合文本模式固定的场景，但网页标签经常多一个属性、换一层嵌套，正则就会漏匹配。用 CSS 选择器或 find 系列方法，锚定的是 DOM 结构而不是纯文本，容错率明显更高。
2.2 选型：requests 还是 httpx，BeautifulSoup 还是 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(text=resp.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?page=1" HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept": "text/html,application/xhtml+xml", } def fetch_page(url: str) -> str: """请求页面并返回 HTML 文本""" resp = requests.get(url, headers=HEADERS, 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(strip=True) if title_node else "", "price": price_node.get_text(strip=True) 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="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=list(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(strip=True) 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, f"https://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_retries=retries, pool_connections=10, pool_maxsize=20, ) 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_page=2) 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 后重新取 text：
resp.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_workers=4) 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% 就抬头查日志，而不是埋头改解析代码。做过几轮爬虫抓取后你会明白，大多数“抓取结果不对”不是解析器写错了，而是反爬策略换了姿势。早发现比事后补救省太多时间。我现在每次跑新目标站，都先抓一页、跑校验、看通过率，然后再放量。希望帮到你。
本文还有配套的精品资源，点击获取
