1
1. 获取 URL（种子页面）
2
2. 发送 HTTP 请求，下载页面内容（HTML/JSON）
3
3. 解析页面，提取数据 + 提取新 URL
4
4. 数据持久化（文件/数据库）
5
5. 对新 URL 重复上述过程（广度/深度优先）

1
# 访问 https://example.com/robots.txt 查看
2
User-agent: *              # 对所有爬虫
3
Disallow: /admin/          # 禁止爬 /admin/
4
Disallow: /private/
5
Allow: /public/            # 允许爬 /public/
6
Crawl-delay: 10            # 爬取间隔 10 秒

1
# Python 内置检查 robots 协议
2
from urllib.robotparser import RobotFileParser
3

4
rp = RobotFileParser()
5
rp.set_url("https://example.com/robots.txt")
6
rp.read()
7
print(rp.can_fetch("*", "https://example.com/data/"))   # True/False

1
# Session 自动管理 Cookie，登录后的请求都复用
2
session = requests.Session()
3
session.get("https://example.com/login")
4
session.post("https://example.com/login", data={"user": "tom", "pwd": "123"})
5
resp = session.get("https://example.com/profile")   # 已登录状态

1
# 下载图片/文件（用 content 二进制流式写入）
2
def download_image(url, filename):
3
    resp = requests.get(url, timeout=10)
4
    if resp.status_code == 200:
5
        with open(filename, "wb") as f:
6
            f.write(resp.content)
7

8
# 大文件流式下载（避免占满内存）
9
with requests.get(big_file_url, stream=True) as r:
10
    with open("file.zip", "wb") as f:
11
        for chunk in r.iter_content(chunk_size=8192):
12
            f.write(chunk)

1
from bs4 import BeautifulSoup
2

3
html = """
4
<div class="articles">
5
    <h2 id="title">文章标题</h2>
6
    <a href="/post/1" class="link">链接1</a>
7
    <a href="/post/2" class="link">链接2</a>
8
    <p>段落内容</p>
9
</div>
10
"""
11
soup = BeautifulSoup(html, "lxml")      # lxml 解析器（快）
12

13
# 查找节点
14
print(soup.title)                        # 标签
15
print(soup.find("h2", id="title").text)  # 按标签+属性找单个 → 文章标题
16
print(soup.find_all("a"))                # 找所有 a 标签
17

18
# CSS 选择器（select）
19
links = soup.select("div.articles a.link")    # 类选择器
20
for a in links:
21
    print(a.text, a["href"])             # 文本 + href 属性
22

23
# 获取属性
24
a = soup.find("a")
25
print(a.get("href"))                     # /post/1
26
print(a.attrs)                           # {'href': '/post/1', 'class': ['link']}
27

28
# 遍历
29
for child in soup.div.children:          # 直接子节点
30
    print(child)

1
tag              标签（div）
2
.class           类（.article）
3
#id              ID（#title）
4
div p            后代（div 内所有 p）
5
div > p          直接子级
6
div, span        多个选择器
7
a[href]          有 href 属性
8
a[href="/x"]     属性等于
9
a[href^="http"]  属性以...开头
10
li:nth-child(2)  第 2 个子元素

1
from lxml import etree
2

3
tree = etree.HTML(html)
4

5
# XPath 提取
6
titles = tree.xpath('//h2[@id="title"]/text()')      # 按 id 取文本
7
links = tree.xpath('//a[@class="link"]/@href')       # 取属性
8
items = tree.xpath('//div[@class="articles"]/a')     # 取节点
9

10
for item in items:
11
    print(item.xpath('./text()')[0])                 # 相对路径
12
    print(item.xpath('./@href')[0])

1
XPath 语法速查：
2
//node           任意位置的 node
3
/node            根下的 node
4
.                当前节点
5
..               父节点
6
@attr            属性
7
text()           文本
8
[条件]           过滤（//a[@href]、//li[1]）
9
*                通配符
10
|                或

1
# 打开浏览器 F12 → Network → XHR/Fetch，找到返回数据的接口
2
# 很多"动态页面"其实是请求 JSON 接口再渲染，直接爬接口最高效！
3
import requests
4

5
api = "https://example.com/api/articles?page=1"
6
resp = requests.get(api, headers=headers)
7
data = resp.json()                        # 直接拿到结构化 JSON
8
for item in data["results"]:
9
    print(item["title"])

1
from selenium import webdriver
2
from selenium.webdriver.common.by import By
3
from selenium.webdriver.support.ui import WebDriverWait
4
from selenium.webdriver.support import expected_conditions as EC
5

6
# 启动浏览器
7
options = webdriver.ChromeOptions()
8
options.add_argument("--headless")        # 无头模式（不显示界面）
9
driver = webdriver.Chrome(options=options)
10

11
try:
12
    driver.get("https://example.com")
13

14
    # 显式等待（等元素出现，比 sleep 可靠）
15
    element = WebDriverWait(driver, 10).until(
16
        EC.presence_of_element_located((By.CSS_SELECTOR, ".article-title"))
17
    )
18

19
    # 查找元素
20
    titles = driver.find_elements(By.CSS_SELECTOR, ".article-title")
21
    for t in titles:
22
        print(t.text)
23

24
    # 模拟交互
25
    driver.find_element(By.ID, "load-more").click()      # 点击
26
    driver.find_element(By.NAME, "q").send_keys("python") # 输入
27
    driver.execute_script("window.scrollTo(0, 10000)")   # 滚动到底部（触发懒加载）
28

29
    # 获取渲染后的 HTML
30
    print(driver.page_source)
31
finally:
32
    driver.quit()                         # 关闭浏览器

1
requests vs Selenium：
2
┌──────────┬────────────────┬────────────────┐
3
│          │ requests       │ Selenium       │
4
├──────────┼────────────────┼────────────────┤
5
│ 执行 JS  │ 否             │ 是（真实浏览器）│
6
│ 速度     │ 快             │ 慢             │
7
│ 资源占用 │ 低             │ 高（开浏览器） │
8
│ 动态页面 │ 需找接口       │ 直接渲染       │
9
│ 适用     │ 静态/JSON 接口 │ 复杂交互/验证码│
10
└──────────┴────────────────┴────────────────┘
11
原则：能爬接口就别用 Selenium，性能差几十倍。

1
Scrapy 架构（五大组件）：
2
       ┌──────────────────────────────────────┐
3
       │              ENGINE（引擎）            │  总调度
4
       └──────────────────────────────────────┘
5
          │          │          │          │
6
   SCHEDULER     SPIDERS    PIPELINES  DOWNLOADER
7
   （调度器）    （爬虫）    （管道）    （下载器）
8
   管理待爬 URL  解析提取    处理数据    发请求下页面
9
                    │
10
              DOWNLOADER MIDDLEWARES（下载器中间件：代理、UA、重试）

1
# items.py：定义数据结构（可选，也可直接 yield dict）
2
import scrapy
3

4
class QuoteItem(scrapy.Item):
5
    text = scrapy.Field()
6
    author = scrapy.Field()
7
    tags = scrapy.Field()
8

9
# pipelines.py：处理数据（清洗、去重、存库）
10
class MySpiderPipeline:
11
    def process_item(self, item, spider):
12
        item["text"] = item["text"].strip()      # 清洗
13
        return item
14

15
    def open_spider(self, spider):               # 爬虫启动时（连数据库）
16
        pass
17
    def close_spider(self, spider):              # 爬虫关闭时（关连接）
18
        pass

1
# 多线程并发下载
2
from concurrent.futures import ThreadPoolExecutor
3

4
def fetch(url):
5
    return requests.get(url, headers=headers, timeout=10).text
6

7
urls = [f"https://example.com/page/{i}" for i in range(1, 50)]
8
with ThreadPoolExecutor(max_workers=10) as pool:
9
    results = list(pool.map(fetch, urls))
10

11
# 异步（aiohttp，高并发首选，见语言进阶篇）
12
# import aiohttp, asyncio

1
# 随机 User-Agent
2
import random
3
UA_LIST = [
4
    "Mozilla/5.0 ... Chrome/120.0",
5
    "Mozilla/5.0 ... Firefox/121.0",
6
    "Mozilla/5.0 ... Safari/605.1",
7
]
8
headers = {"User-Agent": random.choice(UA_LIST)}
9

10
# 代理池（IP 被封时切换）
11
proxies = {"http": "http://proxy1:8080", "https": "https://proxy1:8080"}
12
resp = requests.get(url, headers=headers, proxies=proxies, timeout=10)