目录
6. 持久化：CSV / SQLite / MongoDB（示例：CSV）
7. 单机并发入门：ThreadPoolExecutor + 限速
第 一 章：单机爬虫起点与局限
1. 目标与读者
目标：写出稳定、可维护的单机爬虫；建立“工程化”的基础（日志、重试、限速、持久化）。
适合谁：已会 Python 基础语法，想把“脚本”升级为“靠谱工具”的同学。
2. 环境准备
Python ≥ 3.9
推荐库：
requests,beautifulsoup4,lxml,tenacity（重试）,loguru（日志，可选）
pip install requests beautifulsoup4 lxml tenacity loguru
3. 最小可用爬虫（MVP）
import requests
from bs4 import BeautifulSoup
resp = requests.get("https://example.com", timeout=10)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, "lxml")
print(soup.title.get_text(strip=True))
要点：务必设置 timeout；用 raise_for_status() 让异常显式暴露。
4. 抗脆弱：重试、超时、随机 UA、礼貌抓取
import random, time
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
UAS = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/118 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 Version/15.5 Safari/605.1.15",
]
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=8))
def fetch(url: str) -> str:
headers = {"User-Agent": random.choice(UAS)}
r = requests.get(url, headers=headers, timeout=10)
r.raise_for_status()
# 礼貌：简单限速，避免打爆网站
time.sleep(random.uniform(0.5, 1.5))
return r.text
要点：指数退避（wait_exponential）对临时性错误（429/5xx）更友好；加入随机延迟与随机 UA。
5. 三种解析方式：CSS / XPath / 正则
from bs4 import BeautifulSoup
from lxml import etree
import re
html = fetch("https://example.com")
# 1) CSS（BS4）
soup = BeautifulSoup(html, "lxml")
title_css = soup.select_one("title").get_text(strip=True)
# 2) XPath（lxml）
dom = etree.HTML(html)
title_xpath = dom.xpath("string(//title)")
# 3) 正则（兜底方案，不推荐首选）
match = re.search(r"<title>(.*?)</title>", html, flags=re.I|re.S)
title_re = match.group(1).strip() if match else None
print(title_css, title_xpath, title_re)
建议：优先 CSS/XPath；正则仅作兜底或局部抽取。
6. 持久化：CSV / SQLite / MongoDB（示例：CSV）
import csv, pathlib
from datetime import datetime
OUTPUT = pathlib.Path("data.csv")
def save_csv(rows):
exists = OUTPUT.exists()
with OUTPUT.open("a", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=["url", "title", "ts"])
if not exists:
w.writeheader()
for r in rows:
w.writerow(r)
rows = [{
"url": "https://example.com",
"title": title_css,
"ts": datetime.utcnow().isoformat(),
}]
save_csv(rows)
要点：统一字段；保存 UTC 时间；文件追加写入并自动建表头。
7. 单机并发入门：ThreadPoolExecutor + 限速
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urljoin
BASE = "https://example.com/"
PATHS = ["/", "#", "/?page=2", "/about"] # 示例路径
URLS = [urljoin(BASE, p) for p in PATHS]
def parse_title(url: str) -> dict:
html = fetch(url)
soup = BeautifulSoup(html, "lxml")
return {"url": url, "title": soup.title.get_text(strip=True)}
results = []
with ThreadPoolExecutor(max_workers=8) as pool:
futs = [pool.submit(parse_title, u) for u in URLS]
for fut in as_completed(futs):
try:
data = fut.result()
results.append({**data, "ts": datetime.utcnow().isoformat()})
except Exception as e:
print("error:", e)
save_csv(results)
建议：
对 网络 IO 密集任务，
ThreadPoolExecutor在单机就能显著加速。结合 重试 + 随机延迟 + 最大并发控制，平衡速度与合规。
8. 合规与风控清单（单机阶段必须养成的习惯）
尊重 robots.txt 与站点 ToS；必要时联系站点管理员取得授权。
限速与并发：设定全局 QPS/并发阈值；避免在敏感时段突发大量请求。
指纹收敛：
合理的
User-Agent与Accept-Language；避免固定
ORDER的请求序列与固定延时；
隐私与合规：避免采集个人敏感信息；遵循本地与目标站点司法辖区法律法规。
缓存策略：能缓存就缓存，减少对目标站点的重复压力。
9. 小结
单机阶段的关键是 稳定性与可维护性：日志、重试、限速、持久化。
先把“一个站点、一个功能”打磨到稳，再考虑堆并发、上分布式。
下一篇将进入 Scrapy：用框架把这些“好习惯”收敛为可复用的工程化能力。
10. 练习与思考题
将示例扩展为“抓取多页文章列表 + 详情页解析 + 写入 CSV”。
为
fetch()增加 HTTP 代理 与 会话级 Cookie 支持。设计一个速率限制器：同一域名每秒最多 2 个请求，并记录被 429/503 拒绝时的退避策略。
把 CSV 改为 SQLite，比较两者在增量更新时的优劣。