Cloudflare 防护下的合规采集实践：架构与关键代码

核心库速览#

本文示例主要基于以下 Python 库：

requests 用于 HTTP 会话管理、重试策略挂载、流式下载。
urllib3（Retry）用于网络抖动与 429/5xx 场景的指数退避重试。
beautifulsoup4 用于解析 HTML 并提取下载链接。
urllib.robotparser 用于读取并判断 robots.txt 访问许可。
playwright 用于动态页面访问；在挑战页场景仅做识别与人工兜底，不做自动绕过。
camoufox（浏览器方案可选核心）用于基于 Firefox 内核的自动化会话管理；适合需要完整浏览器上下文的下载流程。
browserforge（与 camoufox 常配套）用于提供屏幕与浏览器相关配置能力，提升自动化环境的一致性。

安装示例：

1
pip install requests urllib3 beautifulsoup4 playwright
2
pip install -U "camoufox[geoip]" browserforge
3
playwright install
4
python -m camoufox fetch

背景#

很多站点会使用 Cloudflare 保护下载链路与页面内容。
在工程实践里，真正可长期运行的方案不是“对抗防护”，而是合规访问 + 稳定采集管道。

本文给出一套可落地的技术框架，并附上关键代码：

识别挑战页并停止自动化流程。
检查 robots.txt 与访问权限。
限速、重试、缓存与日志审计。
必要时切换人工处理流程。

常见自动化脚本的技术结构#

典型下载脚本一般包含这些模块：

页面状态识别：判断是否命中挑战页、登录页、错误页。
会话管理：复用 cookie/session，减少重复握手成本。
链接提取：从资源页解析真实下载链接。
下载落盘：监听下载结果、重命名、冲突避让、失败重试。
超时控制：统一截止时间，避免无限等待。

这套结构本身是中立的，关键在于用途是否合规。

关键代码一：挑战页识别（仅识别，不绕过）#

1
from typing import Iterable
2

3
CF_MARKERS: Iterable[str] = (
4
    "just a moment",
5
    "cf-challenge",
6
    "enable javascript and cookies",
7
    "__cf_chl_",
8
    "cf-mitigated",
9
    "checking if the site connection is secure",
10
)
11

12

13
def is_cloudflare_challenge(url: str, title: str, body_html: str) -> bool:
14
    u = (url or "").lower()
15
    t = (title or "").lower()
16
    b = (body_html or "").lower()
17

18
    if "cdn-cgi/challenge-platform" in u or "__cf_chl_" in u:
19
        return True
20
    if "just a moment" in t:
21
        return True
22
    return any(marker in b for marker in CF_MARKERS)

用途说明：

用于检测并中止自动采集。
用于记录日志和告警。
不用于自动点击挑战控件或规避验证流程。

关键代码二：合规下载脚本（可运行示例）#

下面示例展示了“合规路径”：

先检查 robots.txt。
采用限速 + 指数退避重试。
命中挑战页后直接停止任务并提示人工处理。

1
#!/usr/bin/env python3
2
# -*- coding: utf-8 -*-
3

4
import argparse
5
import random
6
import time
7
from pathlib import Path
8
from urllib.parse import urljoin, urlparse
9
from urllib.robotparser import RobotFileParser
10

11
import requests
12
from bs4 import BeautifulSoup
13
from requests.adapters import HTTPAdapter
14
from urllib3.util.retry import Retry
15

16
CF_MARKERS = (
17
    "just a moment",
18
    "cf-challenge",
19
    "enable javascript and cookies",
20
    "__cf_chl_",
21
    "cf-mitigated",
22
    "checking if the site connection is secure",
23
)
24

25

26
def is_cloudflare_challenge(url: str, title: str, body_html: str) -> bool:
27
    u = (url or "").lower()
28
    t = (title or "").lower()
29
    b = (body_html or "").lower()
30
    if "cdn-cgi/challenge-platform" in u or "__cf_chl_" in u:
31
        return True
32
    if "just a moment" in t:
33
        return True
34
    return any(marker in b for marker in CF_MARKERS)
35

36

37
def build_session() -> requests.Session:
38
    session = requests.Session()
39
    retry = Retry(
40
        total=4,
41
        connect=4,
42
        read=4,
43
        backoff_factor=1.2,
44
        status_forcelist=(429, 500, 502, 503, 504),
45
        allowed_methods=("GET", "HEAD"),
46
        raise_on_status=False,
47
    )
48
    adapter = HTTPAdapter(max_retries=retry, pool_connections=20, pool_maxsize=20)
49
    session.mount("https://", adapter)
50
    session.mount("http://", adapter)
51
    session.headers.update(
52
        {
53
            "User-Agent": "MyCrawler/1.0 (+contact: admin@example.com)",
54
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
55
        }
56
    )
57
    return session
58

59

60
def check_robots_allowed(target_url: str, user_agent: str = "*") -> bool:
61
    parsed = urlparse(target_url)
62
    robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"
63
    rp = RobotFileParser()
64
    rp.set_url(robots_url)
65
    try:
66
        rp.read()
67
    except Exception:
68
        # robots 无法读取时，按保守策略返回 False 也可以
69
        return False
70
    return rp.can_fetch(user_agent, target_url)
71

72

73
def polite_get(session: requests.Session, url: str, min_delay: float = 1.0, max_delay: float = 2.5) -> requests.Response:
74
    time.sleep(random.uniform(min_delay, max_delay))
75
    return session.get(url, timeout=(10, 45))
76

77

78
def extract_download_link(resource_html: str, resource_url: str) -> str:
79
    soup = BeautifulSoup(resource_html, "html.parser")
80
    a = soup.select_one("a[href*='/download?']")
81
    if not a:
82
        raise RuntimeError("未找到下载链接，页面结构可能已变更。")
83
    href = a.get("href", "").strip()
84
    if not href:
85
        raise RuntimeError("下载链接为空。")
86
    return urljoin(resource_url, href)
87

88

89
def stream_download(session: requests.Session, download_url: str, output_dir: Path) -> Path:
90
    with session.get(download_url, timeout=(10, 60), stream=True) as resp:
91
        resp.raise_for_status()
92
        filename = "download.bin"
93
        cd = resp.headers.get("Content-Disposition", "")
94
        if "filename=" in cd:
95
            filename = cd.split("filename=")[-1].strip().strip('"')
96

97
        output_dir.mkdir(parents=True, exist_ok=True)
98
        target = output_dir / filename
99
        idx = 1
100
        while target.exists():
101
            target = output_dir / f"{target.stem}_{idx}{target.suffix}"
102
            idx += 1
103

104
        with target.open("wb") as f:
105
            for chunk in resp.iter_content(chunk_size=8192):
106
                if chunk:
107
                    f.write(chunk)
108
        return target
109

110

111
def main() -> int:
112
    parser = argparse.ArgumentParser(description="合规下载示例：挑战页检测 + robots 检查 + 稳定下载")
113
    parser.add_argument("--url", required=True, help="资源页 URL")
114
    parser.add_argument("--out", default="downloads", help="下载目录")
115
    args = parser.parse_args()
116

117
    url = args.url.strip()
118
    out_dir = Path(args.out).resolve()
119

120
    if not check_robots_allowed(url):
121
        raise SystemExit("robots.txt 不允许采集，任务终止。")
122

123
    session = build_session()
124
    res = polite_get(session, url)
125
    res.raise_for_status()
126

127
    title = ""
128
    try:
129
        soup = BeautifulSoup(res.text, "html.parser")
130
        title = (soup.title.text or "").strip() if soup.title else ""
131
    except Exception:
132
        pass
133

134
    if is_cloudflare_challenge(res.url, title, res.text):
135
        raise SystemExit("命中 Cloudflare 挑战页，自动流程停止，请走人工授权或官方 API。")
136

137
    download_url = extract_download_link(res.text, res.url)
138
    file_path = stream_download(session, download_url, out_dir)
139
    print(f"下载完成: {file_path}")
140
    return 0
141

142

143
if __name__ == "__main__":
144
    raise SystemExit(main())

关键代码三：浏览器自动化的“人工确认模式”#

某些场景必须使用浏览器（动态渲染、登录态页面）。
推荐模式：命中挑战页后停止自动动作，等待人工处理或授权。

1
from playwright.sync_api import sync_playwright
2

3

4
def is_cf_page(url: str, title: str, html: str) -> bool:
5
    text = f"{url}\n{title}\n{html}".lower()
6
    keys = ["cdn-cgi/challenge-platform", "just a moment", "__cf_chl_", "cf-challenge"]
7
    return any(k in text for k in keys)
8

9

10
with sync_playwright() as p:
11
    browser = p.chromium.launch(headless=False)
12
    page = browser.new_page()
13
    page.goto("https://example.com/resource", wait_until="domcontentloaded", timeout=60000)
14

15
    title = page.title()
16
    html = page.content()
17
    if is_cf_page(page.url, title, html):
18
        print("检测到挑战页：停止自动流程，请人工完成验证后再继续。")
19
    else:
20
        print("页面可访问，继续执行后续业务逻辑。")
21

22
    browser.close()

生产环境检查清单#

是否有官方 API、数据导出、授权渠道。
是否配置真实身份 User-Agent 与联系方式。
是否遵守 robots.txt 和服务条款。
是否设置限速、重试、超时、熔断。
是否记录审计日志（URL、时间、状态码、重试次数）。
是否准备人工兜底流程（验证码/登录过期）。

总结#

Cloudflare 场景下，工程目标应从“绕过”转向“合规可持续”：
可识别访问身份 + 可审计流程 + 稳定重试和限速策略 + 人工兜底机制。
这种路径更稳定，也更适合长期业务运行。