Python 爬虫实战：用 requests 抓取并保存文章列表

import requests

session = requests.Session()
session.headers.update({
    "User-Agent": "MyLearningBot/1.0 (+https://example.com/bot)",
    "Accept-Language": "zh-CN,zh;q=0.9",
})

resp = session.get(url, timeout=(3.05, 10))

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

retry = Retry(
    total=3,                          # 每个请求最多重试 3 次
    connect=2,                        # 其中连接失败最多重试 2 次
    backoff_factor=0.5,               # 退避基数：实际等待约 0s、1s、2s
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET"],          # 只自动重试幂等的 GET
    respect_retry_after_header=True,  # 尊重服务器指定的等待时间
)
adapter = HTTPAdapter(max_retries=retry)
session.mount("https://", adapter)
session.mount("http://", adapter)

python3 -m venv .venv
source .venv/bin/activate
python -m pip install requests beautifulsoup4

"""article_list_crawler.py —— 分页抓取文章列表并保存为 CSV。

用法：
    python article_list_crawler.py --start 1 --end 10
"""
import argparse
import csv
import logging
import os
import time
from urllib.parse import urljoin

import requests
from bs4 import BeautifulSoup
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 示例目标：example.com 是 RFC 2606 保留给文档的域名，只返回静态页。
# 跑真实数据前，把这里替换成你自己的目标，并核对下方选择器。
BASE_URL = "https://example.com/blog"

# 列表页选择器：以「article > h2 > a + time」这种博客结构为例。
# 每个目标的 HTML 都可能不同，先用浏览器 DevTools 确认再填。
SELECTORS = {
    "card": "article",
    "link": "h2 a[href]",
    "date": "time",
}

CSV_PATH = "articles.csv"
REQUEST_TIMEOUT = (3.05, 10)   # (连接超时, 读取超时)，单位：秒
DELAY_SECONDS = 1.5            # 每成功写入一页后等待的秒数


def build_session() -> requests.Session:
    """创建带默认头和重试策略的会话：身份、退避、限流尊重一次配齐。"""
    session = requests.Session()
    session.headers.update({
        "User-Agent": "MyLearningBot/1.0 (+https://example.com/bot)",
        "Accept-Language": "zh-CN,zh;q=0.9",
    })
    retry = Retry(
        total=3,
        connect=2,
        backoff_factor=0.5,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET"],
        respect_retry_after_header=True,
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session


def parse_listing(html: str) -> list[dict]:
    """从列表页 HTML 中提取 title / url / date。"""
    soup = BeautifulSoup(html, "html.parser")
    rows: list[dict] = []
    for card in soup.select(SELECTORS["card"]):
        link = card.select_one(SELECTORS["link"])
        if not link:
            continue
        date_tag = card.select_one(SELECTORS["date"])
        rows.append({
            "title": link.get_text(" ", strip=True),
            "url": urljoin(BASE_URL, link["href"]),
            "date": date_tag.get("datetime") if date_tag else "",
        })
    return rows


def append_to_csv(rows: list[dict]) -> None:
    """追加写入 CSV。utf-8-sig 带 BOM，Excel 打开中文不乱码。"""
    is_new = not os.path.exists(CSV_PATH)
    with open(CSV_PATH, "a", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=["title", "url", "date"])
        if is_new:
            writer.writeheader()
        writer.writerows(rows)


def load_done_urls() -> set[str]:
    """从已有 CSV 读回已处理的 URL，实现断点续抓。"""
    done: set[str] = set()
    if not os.path.exists(CSV_PATH):
        return done
    with open(CSV_PATH, newline="", encoding="utf-8-sig") as f:
        for row in csv.DictReader(f):
            if row.get("url"):
                done.add(row["url"])
    return done


def crawl(start: int, end: int) -> None:
    session = build_session()
    done = load_done_urls()
    for page in range(start, end + 1):
        url = f"{BASE_URL}?page={page}"
        try:
            resp = session.get(url, timeout=REQUEST_TIMEOUT)
            resp.raise_for_status()
        except requests.Timeout:
            logging.warning("[timeout] page=%s url=%s", page, url)
            continue
        except requests.HTTPError as exc:
            status = exc.response.status_code if exc.response is not None else 0
            logging.warning("[http %s] page=%s url=%s", status, page, url)
            if status == 404:
                logging.info("遇到 404：视为翻页结束，停止。")
                break
            continue
        except requests.RequestException as exc:
            logging.warning("[request failed] page=%s url=%s err=%s", page, url, exc)
            continue

        rows = [r for r in parse_listing(resp.text) if r["url"] not in done]
        if not rows:
            continue
        append_to_csv(rows)
        done.update(r["url"] for r in rows)
        logging.info("page=%s 新增 %d 条，累计 %d 条", page, len(rows), len(done))
        time.sleep(DELAY_SECONDS)


def main() -> None:
    parser = argparse.ArgumentParser(description="分页抓取文章列表并保存 CSV")
    parser.add_argument("--start", type=int, default=1, help="起始页码")
    parser.add_argument("--end", type=int, default=10, help="结束页码")
    args = parser.parse_args()
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    crawl(args.start, args.end)


if __name__ == "__main__":
    main()

python - <<'PY'
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

s = requests.Session()
retry = Retry(total=3, backoff_factor=0.5,
              status_forcelist=[429, 500, 502, 503, 504],
              allowed_methods=["GET"])
s.mount("https://", HTTPAdapter(max_retries=retry))

r = s.get("https://httpbin.org/headers", timeout=10)
print(r.json()["headers"]["User-Agent"])
PY

python - <<'PY'
import time, requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

s = requests.Session()
s.mount("https://", HTTPAdapter(max_retries=Retry(
    total=3, backoff_factor=0.5,
    status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET"])))

t0 = time.monotonic()
try:
    s.get("https://httpbin.org/status/503", timeout=10)
except requests.RequestException as exc:
    print(type(exc).__name__, "| 耗时 %.1f 秒" % (time.monotonic() - t0))
PY

python - <<'PY'
import time, requests
t0 = time.monotonic()
try:
    requests.get("https://httpbin.org/delay/5", timeout=(3.05, 2))
except requests.Timeout as exc:
    print(type(exc).__name__, "| 耗时 %.2f 秒" % (time.monotonic() - t0))
PY

python - <<'PY'
import time, requests
t0 = time.monotonic()
try:
    requests.get("http://192.0.2.1/", timeout=(3.05, 10))
except requests.Timeout as exc:
    print(type(exc).__name__, "| 耗时 %.2f 秒" % (time.monotonic() - t0))
PY

<!doctype html>
<html>
<head><meta charset="utf-8"><title>示例列表页</title></head>
<body>
  <article>
    <h2><a href="/posts/1">文章一</a></h2>
    <time datetime="2026-08-01">2026-08-01</time>
  </article>
  <article>
    <h2><a href="/posts/2">文章二</a></h2>
    <time datetime="2026-08-02">2026-08-02</time>
  </article>
</body>
</html>

python - <<'PY'
from article_list_crawler import parse_listing

rows = parse_listing(open("sample.html", encoding="utf-8").read())
assert len(rows) == 2, rows
assert rows[0]["url"] == "https://example.com/posts/1", rows[0]
assert rows[0]["date"] == "2026-08-01", rows[0]
print(rows)
PY

python - <<'PY'
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin

BASE = "https://books.toscrape.com/catalogue"
s = requests.Session()

for page in (1, 2):
    url = f"{BASE}/page-{page}.html"
    resp = s.get(url, timeout=10)
    resp.raise_for_status()
    soup = BeautifulSoup(resp.text, "html.parser")
    for card in soup.select("article.product_pod"):
        link = card.select_one("h3 a[title]")
        price = card.select_one("p.price_color")
        print(urljoin(BASE, link["href"]), "|", link["title"], "|", price.get_text(" ", strip=True))
PY