Appearance

Python 爬虫与自动化 ​

Python 常用于接口采集、网页解析、浏览器自动化、批量文件处理和定时办公任务。零基础学习时要先建立正确认知：爬虫不是“绕过限制拿数据”，而是用程序自动访问允许访问的数据来源；自动化也不是“暴力模拟人工点击”，而是把重复、可规则化、可审计的流程交给程序执行。

学完本页后，你应该能做到：

  1. 区分接口采集、HTML 解析、浏览器自动化、文件自动化的适用边界。
  2. 看懂一次 HTTP 请求从发起到响应的过程。
  3. 使用 requests 调用接口、处理超时、状态码、异常和 JSON。
  4. 使用 BeautifulSoup 解析 HTML，并知道页面结构变化为什么会导致脚本失效。
  5. 知道什么时候要用 Playwright，什么时候不应该用浏览器自动化。
  6. 能设计限速、重试、去重、断点续跑、日志和失败样本。
  7. 能写一个商业场景下的“医院资产公开接口采集 + CSV 落地”Demo。
  8. 能回答面试中关于爬虫稳定性、合规、反爬、数据质量和排查的问题。

合规边界先讲清楚 ​

爬虫和自动化必须遵守法律、用户协议、robots 规则、接口授权和数据安全要求。不要采集未授权数据，不要绕过登录、验证码、风控、权限控制，不要高频打爆对方服务，不要抓取个人隐私和敏感数据。

商业项目中的正确做法：

要做不要做
使用公开 API 或已授权接口绕过登录、验证码、权限
控制频率和并发高频压测式请求目标站点
记录来源、时间、版本数据来历不明，无法追溯
对敏感数据脱敏和加密明文保存账号、Token、隐私数据
失败可重试，过程可审计失败静默跳过，结果不可解释

这不是形式主义。线上采集一旦越权或频率失控，轻则账号封禁，重则法律和安全风险。

爬虫和自动化怎么区分 ​

爬虫更关注“获取数据”，自动化更关注“完成操作”。

类型目标工具例子
接口采集获取 JSON/XML 数据requests、httpx调公开 API 获取资产列表
HTML 解析从网页源码提取数据BeautifulSoup、lxml提取页面表格、链接
浏览器自动化模拟浏览器行为Playwright、Selenium登录后台、点击导出
文件自动化批量处理本地文件pathlib、csv、openpyxl批量重命名、清洗 Excel
定时自动化周期执行任务cron、APScheduler、XXL-JOB每天拉取报表

选择流程：

mermaid
flowchart TD
    A["需要获取或处理数据"] --> B{"是否有授权 API"}
    B -->|有| C["优先请求 API"]
    B -->|没有| D{"页面 HTML 是否包含数据"}
    D -->|是| E["requests + HTML 解析"]
    D -->|否| F{"是否需要登录/点击/下载"}
    F -->|是| G["Playwright 浏览器自动化"]
    F -->|否| H["重新确认数据来源和授权"]

能用接口就不要上浏览器。浏览器自动化更慢、更重、更容易受页面变化影响。

HTTP 请求的基本原理 ​

requests.get(url) 背后不是魔法。它大致经历这些步骤：

mermaid
flowchart TD
    A["准备 URL 和参数"] --> B["DNS 解析域名"]
    B --> C["建立 TCP 连接"]
    C --> D{"是否 HTTPS"}
    D -->|是| E["TLS 握手"]
    D -->|否| F["发送 HTTP 请求"]
    E --> F
    F --> G["服务端处理请求"]
    G --> H["返回状态码 响应头 响应体"]
    H --> I["客户端读取并解析响应"]

一次响应通常包含：

部分例子作用
状态码200、404、500表示请求结果
响应头Content-Type、Set-Cookie描述响应元信息
响应体HTML、JSON、文件字节真正的数据

常见状态码：

状态码含义处理建议
200成功正常解析
301/302重定向确认最终地址
400请求错误检查参数
401/403未认证或无权限检查授权，不要绕过
404不存在记录失败样本
429请求太频繁降低频率，暂停任务
500/502/503服务端异常有限重试

requests 入门 ​

安装：

shell
pip install requests

最小示例：

python
import requests

url = "https://example.com"
response = requests.get(url, timeout=10)

print(response.status_code)
print(response.text[:200])

生产代码不能只写这一点，至少要处理超时、状态码和异常：

python
import requests


def fetch_text(url: str) -> str:
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return response.text
    except requests.Timeout as exc:
        raise RuntimeError(f"请求超时: {url}") from exc
    except requests.HTTPError as exc:
        raise RuntimeError(f"HTTP 状态异常: {response.status_code} {url}") from exc
    except requests.RequestException as exc:
        raise RuntimeError(f"请求失败: {url}") from exc

为什么必须设置 timeout？如果不设置，网络卡住时程序可能一直等待，占住线程、连接、任务槽位，定时任务会越积越多。

Session、Header、Cookie ​

requests.Session 可以复用连接和保存 Cookie。

python
import requests

session = requests.Session()
session.headers.update({
    "User-Agent": "asset-collector/1.0",
})

response = session.get("https://example.com/api/assets", timeout=10)
response.raise_for_status()

Header 常用于：

Header作用
User-Agent标识客户端
AuthorizationToken 鉴权
Accept告诉服务端希望返回什么格式
Content-Type告诉服务端请求体格式

注意：不要把 Token、Cookie、账号密码写死在代码里。应从环境变量或配置中心读取，并避免打印到日志。

解析 JSON ​

接口返回 JSON 时：

python
import requests


def fetch_assets(api_url: str) -> list[dict]:
    response = requests.get(api_url, timeout=10)
    response.raise_for_status()
    data = response.json()
    return data["items"]

这段代码仍然不够稳。原因：

  1. 响应可能不是合法 JSON。
  2. JSON 结构可能没有 items。
  3. items 可能不是列表。
  4. 单条数据字段可能缺失。

更稳的写法：

python
import requests


def fetch_assets(api_url: str) -> list[dict]:
    response = requests.get(api_url, timeout=10)
    response.raise_for_status()

    try:
        data = response.json()
    except ValueError as exc:
        raise RuntimeError("响应不是合法 JSON") from exc

    items = data.get("items")
    if not isinstance(items, list):
        raise RuntimeError("JSON 缺少 items 列表")
    return items

解析成功只说明 JSON 语法合法，不说明业务字段正确。业务字段还要单独校验。

URL 参数和分页 ​

很多接口通过 query 参数分页：

python
import requests

response = requests.get(
    "https://example.com/api/assets",
    params={"page": 1, "size": 100},
    timeout=10,
)

不要手动拼接：

python
# 不推荐
url = f"https://example.com/api/assets?page={page}&keyword={keyword}"

原因：参数里可能有空格、中文、特殊符号，需要正确编码。params 会帮你处理。

分页采集流程：

mermaid
flowchart TD
    A["page = 1"] --> B["请求当前页"]
    B --> C["解析 items"]
    C --> D["保存或处理当前页"]
    D --> E{"是否还有下一页"}
    E -->|有| F["page 加 1"]
    F --> B
    E -->|没有| G["结束采集"]

分页采集要防止两个问题：

  1. 接口一直返回“有下一页”，导致死循环，所以要设置最大页数或空页退出。
  2. 采集过程中数据变化，导致重复或漏数据，所以最好使用稳定排序、游标、更新时间边界。

解析 HTML ​

安装：

shell
pip install beautifulsoup4

示例：

python
import requests
from bs4 import BeautifulSoup

html = requests.get("https://example.com", timeout=10).text
soup = BeautifulSoup(html, "html.parser")

title = soup.find("h1")
if title:
    print(title.get_text(strip=True))

links = soup.find_all("a")
for link in links:
    print(link.get("href"), link.get_text(strip=True))

HTML 解析常见问题：

问题原因做法
找不到元素页面结构变了或 JS 渲染保存失败样本，检查源码
文本多空格换行HTML 排版符号多get_text(strip=True)
链接是相对路径href="/a"用 urljoin 拼绝对地址
字段偶尔缺失页面数据不完整字段允许为空并记录原因

相对链接处理：

python
from urllib.parse import urljoin

base_url = "https://example.com/news/"
href = "/article/1"
print(urljoin(base_url, href))

HTML 页面为什么 requests 抓不到数据 ​

很多现代页面是前端框架渲染的。requests 拿到的是初始 HTML，真实数据由浏览器执行 JavaScript 后再请求接口加载。

排查流程：

mermaid
flowchart TD
    A["requests 拿到空页面"] --> B["打开浏览器开发者工具"]
    B --> C["查看 Network 请求"]
    C --> D{"是否有 JSON 接口"}
    D -->|有| E["优先直接请求 JSON 接口"]
    D -->|没有| F{"数据是否必须 JS 渲染"}
    F -->|是| G["考虑 Playwright"]
    F -->|否| H["检查请求头 Cookie 权限"]

不要一上来就用浏览器自动化。先找接口，接口稳定性通常比页面结构更好。

保存数据 ​

保存为 CSV：

python
import csv
from pathlib import Path

rows = [
    {"title": "文章1", "url": "https://example.com/1"},
    {"title": "文章2", "url": "https://example.com/2"},
]

with Path("articles.csv").open("w", newline="", encoding="utf-8-sig") as file:
    writer = csv.DictWriter(file, fieldnames=["title", "url"])
    writer.writeheader()
    writer.writerows(rows)

utf-8-sig 可以减少 Windows Excel 打开中文 CSV 乱码问题。如果是程序之间交换数据，优先使用普通 utf-8。

保存前要考虑：

  1. 唯一键是什么。
  2. 是否允许重复采集。
  3. 字段缺失怎么记录。
  4. 失败数据是否单独保存。
  5. 文件写到一半失败怎么办。

去重和断点续跑 ​

采集任务很少一次就完美跑完。要考虑网络失败、脚本中断、接口限流。

去重思路：

python
seen_ids: set[str] = set()

for item in items:
    item_id = item["id"]
    if item_id in seen_ids:
        continue
    seen_ids.add(item_id)
    save(item)

断点续跑可以记录最后处理的位置：

python
from pathlib import Path


def load_checkpoint(path: Path) -> int:
    if not path.exists():
        return 1
    return int(path.read_text(encoding="utf-8"))


def save_checkpoint(path: Path, page: int) -> None:
    path.write_text(str(page), encoding="utf-8")

流程：

mermaid
flowchart TD
    A["任务启动"] --> B["读取 checkpoint"]
    B --> C["从记录页码继续"]
    C --> D["采集并保存当前页"]
    D --> E["保存新的 checkpoint"]
    E --> F{"是否完成"}
    F -->|否| C
    F -->|是| G["删除或归档 checkpoint"]

断点必须在“数据成功保存后”再更新，否则中途失败会跳过未保存的数据。

限速和重试 ​

不要高频请求目标站点或接口。最简单的方式是加延迟：

python
import time

for url in urls:
    fetch(url)
    time.sleep(1)

更稳的重试要区分“可重试”和“不可重试”：

类型是否重试例子
超时可以有限重试网络抖动
502/503可以有限重试服务暂时不可用
429不应立刻重试请求太频繁，应降速
400通常不重试参数错误
401/403通常不重试权限问题

重试示例：

python
import time
import requests


def fetch_json(url: str, params: dict, retries: int = 3) -> dict:
    for attempt in range(1, retries + 1):
        try:
            response = requests.get(url, params=params, timeout=10)
            if response.status_code == 429:
                raise RuntimeError("请求过于频繁，停止任务")
            response.raise_for_status()
            return response.json()
        except (requests.Timeout, requests.ConnectionError) as exc:
            if attempt == retries:
                raise
            sleep_seconds = 2 ** attempt
            print(f"第 {attempt} 次失败: {exc}，{sleep_seconds}s 后重试")
            time.sleep(sleep_seconds)

不要无限重试，否则可能让程序卡住，也可能给对方服务造成更大压力。

浏览器自动化 Playwright ​

有些页面必须登录、点击、等待前端渲染或下载文件，这时可以使用 Playwright。

安装：

shell
pip install playwright
playwright install

基本示例：

python
from playwright.sync_api import sync_playwright


def capture_title(url: str) -> str:
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        page.goto(url, wait_until="networkidle")
        title = page.title()
        browser.close()
        return title

浏览器自动化流程：

mermaid
flowchart TD
    A["启动浏览器"] --> B["打开页面"]
    B --> C["等待页面或元素加载"]
    C --> D["点击 输入 下载 截图"]
    D --> E["读取页面或文件"]
    E --> F["保存结果"]
    F --> G["关闭浏览器"]

关键原则：

  1. 等元素，不要只 sleep。
  2. 操作后检查结果是否出现。
  3. 登录态和账号权限要合规管理。
  4. 浏览器要关闭，避免进程泄漏。
  5. 能用接口就不要用浏览器。

等待元素示例：

python
page.goto("https://example.com/report")
page.get_by_role("button", name="导出").click()
page.wait_for_selector("text=导出成功", timeout=10000)

商业 Demo：授权资产接口采集到 CSV ​

需求：从已授权的资产平台 API 分页采集资产数据，清洗后保存为 CSV。要求：

  1. Token 从环境变量读取。
  2. 每页最多 100 条。
  3. 设置超时。
  4. 遇到 429 停止。
  5. 支持断点续跑。
  6. 按 asset_code 去重。
  7. 输出成功数和失败样本。

完整 Demo：

python
import csv
import os
import time
from dataclasses import dataclass
from pathlib import Path

import requests


@dataclass(frozen=True)
class Asset:
    asset_code: str
    asset_name: str
    department: str
    source_system: str


def load_checkpoint(path: Path) -> int:
    if not path.exists():
        return 1
    return int(path.read_text(encoding="utf-8"))


def save_checkpoint(path: Path, page: int) -> None:
    path.write_text(str(page), encoding="utf-8")


def fetch_page(session: requests.Session, api_url: str, page: int, size: int) -> dict:
    response = session.get(
        api_url,
        params={"page": page, "size": size},
        timeout=10,
    )
    if response.status_code == 429:
        raise RuntimeError("请求过于频繁，任务停止")
    response.raise_for_status()
    return response.json()


def parse_asset(raw: dict) -> Asset:
    asset_code = str(raw.get("asset_code") or "").strip()
    asset_name = str(raw.get("asset_name") or "").strip()
    department = str(raw.get("department") or "").strip()
    source_system = str(raw.get("source_system") or "").strip()

    if not asset_code:
        raise ValueError("asset_code 为空")
    if not asset_name:
        raise ValueError(f"asset_name 为空: {asset_code}")

    return Asset(asset_code, asset_name, department, source_system)


def write_assets(path: Path, assets: list[Asset]) -> None:
    with path.open("w", encoding="utf-8-sig", newline="") as file:
        writer = csv.DictWriter(
            file,
            fieldnames=["asset_code", "asset_name", "department", "source_system"],
        )
        writer.writeheader()
        for asset in assets:
            writer.writerow(
                {
                    "asset_code": asset.asset_code,
                    "asset_name": asset.asset_name,
                    "department": asset.department,
                    "source_system": asset.source_system,
                }
            )


def collect_assets(api_url: str, output: Path, checkpoint: Path) -> None:
    token = os.environ["ASSET_API_TOKEN"]
    session = requests.Session()
    session.headers.update({
        "Authorization": f"Bearer {token}",
        "User-Agent": "asset-collector/1.0",
    })

    page = load_checkpoint(checkpoint)
    size = 100
    assets: list[Asset] = []
    seen_codes: set[str] = set()
    errors: list[str] = []

    while True:
        data = fetch_page(session, api_url, page, size)
        items = data.get("items", [])
        if not items:
            break

        for raw in items:
            try:
                asset = parse_asset(raw)
            except ValueError as exc:
                errors.append(str(exc))
                continue

            if asset.asset_code in seen_codes:
                continue
            seen_codes.add(asset.asset_code)
            assets.append(asset)

        write_assets(output, assets)
        save_checkpoint(checkpoint, page + 1)

        if len(items) < size:
            break

        page += 1
        time.sleep(1)

    print(f"采集完成 success={len(assets)} failed={len(errors)}")
    if errors:
        print(f"失败样本: {errors[:3]}")


if __name__ == "__main__":
    collect_assets(
        api_url="https://example.com/api/assets",
        output=Path("assets.csv"),
        checkpoint=Path("asset_checkpoint.txt"),
    )

这个 Demo 的生产意识：

  1. Token 不写死，从环境变量读取。
  2. 通过 Session 复用连接和统一 Header。
  3. 用分页控制单次返回量。
  4. 用 checkpoint 支持断点续跑。
  5. 保存成功后再更新 checkpoint。
  6. 对单条脏数据记录错误，不让整批中断。
  7. 用唯一字段去重。
  8. 对 429 立刻停止，而不是继续重试。

采集链路：

mermaid
flowchart TD
    A["读取 Token 和 checkpoint"] --> B["请求当前页 API"]
    B --> C{"状态码是否正常"}
    C -->|429| D["停止任务并降速"]
    C -->|异常| E["记录错误或有限重试"]
    C -->|正常| F["解析 JSON items"]
    F --> G["逐条校验字段"]
    G --> H{"字段是否合法"}
    H -->|否| I["记录失败样本"]
    H -->|是| J["按 asset_code 去重"]
    J --> K["写入 CSV"]
    K --> L["更新 checkpoint"]
    L --> M{"是否还有下一页"}
    M -->|有| B
    M -->|没有| N["输出采集统计"]

数据质量和幂等 ​

采集数据不是拿到就完事。商业系统更关注数据是否可信、是否可追溯、是否能重复执行。

能力说明
唯一键用 asset_code、URL、业务 ID 去重
来源记录记录来源系统、接口地址、采集时间
失败样本保存解析失败的原始数据
幂等同一批数据重复执行不会重复入库
校验必填字段、类型、枚举值、范围
审计谁在什么时候采集了什么

如果要入库，建议先写入“采集原始表”或“批次表”，再转换成业务表。这样出问题时能回看原始数据。

生产排查流程 ​

mermaid
flowchart TD
    A["采集任务异常"] --> B{"请求是否成功"}
    B -->|否| C["看状态码 超时 DNS TLS 代理"]
    B -->|是| D{"响应结构是否变化"}
    D -->|是| E["保存样本 更新解析规则"]
    D -->|否| F{"是否被限流"}
    F -->|是| G["降低频率 暂停任务 检查 429"]
    F -->|否| H{"数据是否重复或缺失"}
    H -->|重复| I["检查唯一键和 checkpoint"]
    H -->|缺失| J["检查分页 排序 时间边界"]
    H -->|正常| K["检查保存文件 数据库 日志"]

排查清单：

现象重点检查
Timeout网络、目标服务、timeout 是否过短、代理
403Token、权限、来源 IP、授权范围
429请求频率、并发数、重试策略
JSON 解析失败响应是否是 HTML 错误页、登录页
HTML 解析为空页面是否 JS 渲染、选择器是否失效
数据重复唯一键、断点更新时机、分页排序
数据漏采页码变化、数据实时新增、时间窗口
定时任务失败工作目录、环境变量、解释器、权限

常见坑 ​

问题后果正确做法
不设置 timeout程序卡住，任务堆积所有请求设置超时
无限重试打爆对方服务，自己也卡住有限重试，指数退避
不处理 429越请求越被限制降速、暂停、告警
页面选择器写死页面一改脚本就坏保存失败样本，选择稳定字段
Token 写死密钥泄露环境变量或密钥管理
没有去重重复入库设计唯一键和幂等逻辑
断点更新太早中断后漏数据成功保存后再更新 checkpoint
只看成功数脏数据被忽略保存失败样本和原因

面试标准回答 ​

requests 抓到的页面没有数据怎么办？ ​

标准回答：先判断数据是否由 JavaScript 后加载。打开浏览器开发者工具看 Network，如果有 JSON 接口，优先请求接口；如果没有接口且必须执行 JS，再考虑 Playwright。还要检查 Header、Cookie、权限和登录状态。

爬虫怎么保证稳定性？ ​

标准回答：要设置超时、有限重试、限速、断点续跑、去重、失败样本保存、日志和告警。解析逻辑要允许字段缺失，页面结构变化时能记录原始响应，避免静默丢数据。

为什么要限速？ ​

标准回答：限速是为了不影响对方服务，也避免触发 429、封禁、风控。商业采集要尊重授权范围和服务承载能力，请求频率、并发数和重试策略都应该可配置。

requests 和 Playwright 怎么选？ ​

标准回答：能用授权 API 或静态 HTML 就用 requests，因为轻量、稳定、快；页面需要登录、点击、等待 JS 渲染、下载文件或截图时才用 Playwright。Playwright 更重，维护成本更高。

如何处理重复数据和断点续跑？ ​

标准回答：重复数据用业务唯一键去重，例如 URL、商品 ID、资产编号；断点续跑要保存页码、游标或时间边界，并且在数据成功保存后再更新 checkpoint，避免中断后漏采。

采集到的数据如何保证质量？ ​

标准回答：采集后要做字段校验、类型转换、枚举校验、唯一键去重、来源记录、失败样本保存。重要数据最好保留原始记录和批次号，方便追溯和重新处理。

练习 ​

  1. 用 requests 请求一个网页并打印状态码。
  2. 给请求增加 timeout、raise_for_status() 和异常处理。
  3. 调用一个公开 JSON API，校验返回结构。
  4. 用 BeautifulSoup 提取页面中所有链接。
  5. 把标题和链接保存成 CSV。
  6. 给循环抓取增加 1 秒延迟。
  7. 为采集任务设计唯一键和 checkpoint。
  8. 模拟 3 条脏数据，保存失败原因。

关联知识点 ​

小结 ​

Python 爬虫与自动化的重点不是“能不能抓到”，而是“能否合规、稳定、可维护、可排查地采集”。真正可用的采集程序要理解 HTTP、状态码、超时、重试、限速、解析、去重、断点续跑、数据校验、日志和安全边界。能用接口就用接口，需要浏览器时再上 Playwright。