thzxx
28ff7c0a48
反爬措施: - 浏览器指纹头 build_browser_headers(): Sec-Ch-Ua/Sec-Fetch-*/Accept-Language/Accept-Encoding, 绕过 80%+ 轻量 WAF - 12 个 UA 池 (Chrome/Edge/Firefox x Win/macOS/Linux x v129-131) - 确定性 UA 轮换 get_ua_for_domain(): SHA-256 按域名固定 UA, 会话内稳定跨进程可复现 - WAF 指纹库 _detect_anti_bot(): 识别 Cloudflare/Imperva/PerimeterX/DataDome/Akamai/通用, 全文档扫描 - Retry-After 遵守: 429/503 读取 header (数字或 HTTP date) 作为最小重试延迟 - 退避封顶 60s (原无上限, N=10 时 1536s 卡死进程) 抓取稳定性: - requests.Session 复用: 连接池(10/host) + cookie 持久化 + TLS 会话恢复 - 超时分离 (connect, read) 元组, 避免大页面浪费已建连接 - Wayback Machine 兜底: 404/403/超时自动重试 web.archive.org, 默认启用 --no-fallback 关闭 - AdaptiveThrottle 自适应限流: 3 次失败翻倍延迟+减半并发, 5 次成功渐进恢复, 429 全局暂停 30s - readability-lite 提取: article/main 缺失时按文本密度选最可能正文 div 新增 CLI flags: - --fetch-report: 结构化抓取报告到 stderr (每 URL 状态/WAF 类型/兜底方式/字符数 + JSON 摘要) - --no-fallback: 禁用 Wayback 兜底 - --referer: 设置 Referer 头 (默认实例 URL) - --request-delay: 抓取请求间隔秒数 (默认 0.3, 自适应可能增大) fetch 结果新字段: anti_bot_detected (bool), waf_type (str|null), fallback_used (str|null) 测试: 新增 4 个测试文件 (test_browser_headers/test_anti_bot/test_wayback_fallback/test_adaptive_throttle), 451 个测试全部通过
325 lines
11 KiB
Python
325 lines
11 KiB
Python
"""Tests for v2.0.0 browser fingerprint headers and UA pool enhancements.
|
||
|
||
Covers: build_browser_headers (Chrome/Edge/Firefox variants, Referer,
|
||
Accept modes), get_ua_for_domain (deterministic per-domain, caching,
|
||
explicit override), parse_retry_after (numeric/HTTP date/edge cases),
|
||
compute_backoff_delay (cap enforcement), UA pool size and diversity.
|
||
"""
|
||
import sys
|
||
import os
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
|
||
|
||
from common import (
|
||
FALLBACK_UAS,
|
||
RETRY_BACKOFF_CAP,
|
||
build_browser_headers,
|
||
compute_backoff_delay,
|
||
get_ua_for_domain,
|
||
parse_retry_after,
|
||
reset_domain_ua_cache,
|
||
_ua_index_for_domain,
|
||
)
|
||
|
||
|
||
# ----- UA pool -----
|
||
|
||
def test_ua_pool_has_at_least_12_entries():
|
||
"""v2.0.0: expanded from 3 to 12 for diversity."""
|
||
assert len(FALLBACK_UAS) >= 12
|
||
|
||
|
||
def test_ua_pool_covers_multiple_browsers():
|
||
browsers = []
|
||
for ua in FALLBACK_UAS:
|
||
if "Edg/" in ua:
|
||
browsers.append("edge")
|
||
elif "Firefox/" in ua:
|
||
browsers.append("firefox")
|
||
elif "Chrome/" in ua:
|
||
browsers.append("chrome")
|
||
# 至少三种浏览器
|
||
assert "chrome" in browsers
|
||
assert "firefox" in browsers
|
||
assert "edge" in browsers
|
||
|
||
|
||
def test_ua_pool_covers_multiple_platforms():
|
||
platforms = []
|
||
for ua in FALLBACK_UAS:
|
||
if "Windows" in ua:
|
||
platforms.append("windows")
|
||
elif "Macintosh" in ua:
|
||
platforms.append("macos")
|
||
elif "Linux" in ua:
|
||
platforms.append("linux")
|
||
assert "windows" in platforms
|
||
assert "macos" in platforms
|
||
assert "linux" in platforms
|
||
|
||
|
||
# ----- get_ua_for_domain -----
|
||
|
||
def test_get_ua_for_domain_is_deterministic():
|
||
"""同一域名永远返回同一 UA（跨调用一致）。"""
|
||
reset_domain_ua_cache()
|
||
url = "https://example.com/page1"
|
||
ua1 = get_ua_for_domain(url)
|
||
ua2 = get_ua_for_domain(url)
|
||
assert ua1 == ua2
|
||
assert ua1 in FALLBACK_UAS
|
||
|
||
|
||
def test_get_ua_for_domain_same_domain_different_paths():
|
||
"""同域名不同路径返回同 UA。"""
|
||
reset_domain_ua_cache()
|
||
ua1 = get_ua_for_domain("https://example.com/a")
|
||
ua2 = get_ua_for_domain("https://example.com/b/c/d")
|
||
assert ua1 == ua2
|
||
|
||
|
||
def test_get_ua_for_domain_explicit_override():
|
||
"""显式 user_agent 优先于域名缓存。"""
|
||
reset_domain_ua_cache()
|
||
custom = "MyCustomBot/1.0"
|
||
ua = get_ua_for_domain("https://example.com", user_agent=custom)
|
||
assert ua == custom
|
||
|
||
|
||
def test_get_ua_for_domain_different_domains_may_differ():
|
||
"""不同域名可能映射到不同 UA（不一定，但缓存独立）。"""
|
||
reset_domain_ua_cache()
|
||
ua1 = get_ua_for_domain("https://aaa.example.com")
|
||
ua2 = get_ua_for_domain("https://bbb.example.com")
|
||
# 都是合法 UA
|
||
assert ua1 in FALLBACK_UAS
|
||
assert ua2 in FALLBACK_UAS
|
||
|
||
|
||
def test_get_ua_for_domain_invalid_url_returns_default():
|
||
"""无效 URL 返回第一个 UA（兜底）。"""
|
||
reset_domain_ua_cache()
|
||
ua = get_ua_for_domain("not-a-url")
|
||
assert ua == FALLBACK_UAS[0]
|
||
|
||
|
||
def test_get_ua_for_domain_caches_across_calls():
|
||
"""缓存生效：第二次调用不重新计算。"""
|
||
reset_domain_ua_cache()
|
||
url = "https://cached.example.com"
|
||
ua1 = get_ua_for_domain(url)
|
||
# 直接从缓存取
|
||
from common import _domain_ua_cache
|
||
domain_key = "cached.example.com"
|
||
assert domain_key in _domain_ua_cache
|
||
assert _domain_ua_cache[domain_key] == ua1
|
||
|
||
|
||
def test_ua_index_for_domain_is_stable_across_processes():
|
||
"""SHA-256 hash 保证跨进程一致（不像内置 hash 受 PYTHONHASHSEED 影响）。"""
|
||
idx1 = _ua_index_for_domain("example.com", len(FALLBACK_UAS))
|
||
idx2 = _ua_index_for_domain("example.com", len(FALLBACK_UAS))
|
||
assert idx1 == idx2
|
||
assert 0 <= idx1 < len(FALLBACK_UAS)
|
||
|
||
|
||
# ----- build_browser_headers -----
|
||
|
||
def test_build_headers_chrome_includes_sec_ch_ua():
|
||
"""Chrome UA 应生成 Sec-Ch-Ua 系列头。"""
|
||
chrome_ua = FALLBACK_UAS[0] # Chrome 131 Windows
|
||
headers = build_browser_headers(chrome_ua)
|
||
assert "Sec-Ch-Ua" in headers
|
||
assert "Sec-Ch-Ua-Mobile" in headers
|
||
assert "Sec-Ch-Ua-Platform" in headers
|
||
assert "Windows" in headers["Sec-Ch-Ua-Platform"]
|
||
|
||
|
||
def test_build_headers_firefox_excludes_sec_ch_ua():
|
||
"""Firefox UA 不应生成 Sec-Ch-Ua（Firefox 不发送此头）。"""
|
||
firefox_ua = next(ua for ua in FALLBACK_UAS if "Firefox/" in ua)
|
||
headers = build_browser_headers(firefox_ua)
|
||
assert "Sec-Ch-Ua" not in headers
|
||
assert "Sec-Ch-Ua-Mobile" not in headers
|
||
|
||
|
||
def test_build_headers_includes_accept_language():
|
||
"""所有浏览器都应有 Accept-Language。"""
|
||
headers = build_browser_headers(FALLBACK_UAS[0])
|
||
assert "Accept-Language" in headers
|
||
assert "en-US" in headers["Accept-Language"]
|
||
|
||
|
||
def test_build_headers_includes_accept_encoding():
|
||
"""Accept-Encoding 必须存在；Firefox 不发 br。"""
|
||
chrome_headers = build_browser_headers(FALLBACK_UAS[0])
|
||
assert "br" in chrome_headers["Accept-Encoding"]
|
||
|
||
firefox_ua = next(ua for ua in FALLBACK_UAS if "Firefox/" in ua)
|
||
firefox_headers = build_browser_headers(firefox_ua)
|
||
assert "br" not in firefox_headers["Accept-Encoding"]
|
||
assert "gzip" in firefox_headers["Accept-Encoding"]
|
||
|
||
|
||
def test_build_headers_html_mode_includes_upgrade_insecure():
|
||
"""HTML 模式下 Upgrade-Insecure-Requests=1。"""
|
||
headers = build_browser_headers(FALLBACK_UAS[0], accept_html=True)
|
||
assert headers["Upgrade-Insecure-Requests"] == "1"
|
||
|
||
|
||
def test_build_headers_json_mode_excludes_upgrade_insecure_navigation():
|
||
"""JSON 模式下不发送导航相关头。"""
|
||
headers = build_browser_headers(FALLBACK_UAS[0], accept_html=False)
|
||
assert headers["Upgrade-Insecure-Requests"] == "0"
|
||
assert headers["Sec-Fetch-Mode"] == "cors"
|
||
assert headers["Sec-Fetch-Dest"] == "empty"
|
||
|
||
|
||
def test_build_headers_sec_fetch_dest_document_for_html():
|
||
"""HTML 模式 Sec-Fetch-Dest=document。"""
|
||
headers = build_browser_headers(FALLBACK_UAS[0], accept_html=True)
|
||
assert headers["Sec-Fetch-Dest"] == "document"
|
||
assert headers["Sec-Fetch-Mode"] == "navigate"
|
||
|
||
|
||
def test_build_headers_referer_set_when_provided():
|
||
"""传入 referer 时设置 Referer 头。"""
|
||
headers = build_browser_headers(FALLBACK_UAS[0],
|
||
referer="https://google.com/")
|
||
assert headers["Referer"] == "https://google.com/"
|
||
|
||
|
||
def test_build_headers_no_referer_when_absent():
|
||
"""不传 referer 时不设置 Referer 头。"""
|
||
headers = build_browser_headers(FALLBACK_UAS[0])
|
||
assert "Referer" not in headers
|
||
|
||
|
||
def test_build_headers_sec_fetch_site_none_without_referer():
|
||
"""无 Referer 时 Sec-Fetch-Site=none（像地址栏直接访问）。"""
|
||
headers = build_browser_headers(FALLBACK_UAS[0], accept_html=True)
|
||
assert headers["Sec-Fetch-Site"] == "none"
|
||
|
||
|
||
def test_build_headers_sec_fetch_site_cross_site_with_referer():
|
||
"""有 Referer 时 Sec-Fetch-Site=cross-site。"""
|
||
headers = build_browser_headers(FALLBACK_UAS[0],
|
||
accept_html=True,
|
||
referer="https://google.com/")
|
||
assert headers["Sec-Fetch-Site"] == "cross-site"
|
||
|
||
|
||
def test_build_headers_edge_includes_edge_brand():
|
||
"""Edge UA 的 Sec-Ch-Ua 应包含 Microsoft Edge 品牌。"""
|
||
edge_ua = next(ua for ua in FALLBACK_UAS if "Edg/" in ua)
|
||
headers = build_browser_headers(edge_ua)
|
||
assert "Microsoft Edge" in headers["Sec-Ch-Ua"]
|
||
|
||
|
||
def test_build_headers_macos_platform():
|
||
"""macOS UA 的 Sec-Ch-Ua-Platform=macOS。"""
|
||
mac_ua = next(ua for ua in FALLBACK_UAS if "Macintosh" in ua and "Edg/" not in ua)
|
||
headers = build_browser_headers(mac_ua)
|
||
assert "macOS" in headers["Sec-Ch-Ua-Platform"]
|
||
|
||
|
||
def test_build_headers_linux_platform():
|
||
"""Linux UA 的 Sec-Ch-Ua-Platform=Linux。"""
|
||
linux_ua = next(ua for ua in FALLBACK_UAS if "Linux" in ua and "Firefox/" not in ua)
|
||
headers = build_browser_headers(linux_ua)
|
||
assert "Linux" in headers["Sec-Ch-Ua-Platform"]
|
||
|
||
|
||
def test_build_headers_user_agent_set():
|
||
"""UA 必须设置到 User-Agent 头。"""
|
||
headers = build_browser_headers(FALLBACK_UAS[0])
|
||
assert headers["User-Agent"] == FALLBACK_UAS[0]
|
||
|
||
|
||
def test_build_headers_connection_keep_alive():
|
||
"""Connection: keep-alive 支持 HTTP 持久连接。"""
|
||
headers = build_browser_headers(FALLBACK_UAS[0])
|
||
assert headers["Connection"] == "keep-alive"
|
||
|
||
|
||
# ----- parse_retry_after -----
|
||
|
||
def test_parse_retry_after_numeric_seconds():
|
||
"""纯数字格式：秒数。"""
|
||
assert parse_retry_after("30") == 30.0
|
||
assert parse_retry_after("0") == 0.0
|
||
assert parse_retry_after("120") == 120.0
|
||
|
||
|
||
def test_parse_retry_after_decimal():
|
||
"""小数秒数。"""
|
||
assert parse_retry_after("1.5") == 1.5
|
||
|
||
|
||
def test_parse_retry_after_empty():
|
||
"""空字符串返回 0。"""
|
||
assert parse_retry_after("") == 0.0
|
||
assert parse_retry_after(None) == 0.0
|
||
|
||
|
||
def test_parse_retry_after_http_date_future():
|
||
"""HTTP date 格式（未来时间）返回正秒数。"""
|
||
from datetime import datetime, timezone, timedelta
|
||
future = datetime.now(timezone.utc) + timedelta(seconds=60)
|
||
from email.utils import format_datetime
|
||
date_str = format_datetime(future)
|
||
seconds = parse_retry_after(date_str)
|
||
assert 50 < seconds < 70 # 允许一点时间漂移
|
||
|
||
|
||
def test_parse_retry_after_http_date_past():
|
||
"""HTTP date 格式（过去时间）返回 0（已过期）。"""
|
||
from datetime import datetime, timezone, timedelta
|
||
past = datetime.now(timezone.utc) - timedelta(seconds=60)
|
||
from email.utils import format_datetime
|
||
date_str = format_datetime(past)
|
||
assert parse_retry_after(date_str) == 0.0
|
||
|
||
|
||
def test_parse_retry_after_garbage():
|
||
"""无法解析的值返回 0。"""
|
||
assert parse_retry_after("not-a-date-or-number") == 0.0
|
||
|
||
|
||
def test_parse_retry_after_negative_numeric():
|
||
"""负数秒返回 0（不允许负等待）。"""
|
||
assert parse_retry_after("-5") == 0.0
|
||
|
||
|
||
# ----- compute_backoff_delay -----
|
||
|
||
def test_backoff_caps_at_60_seconds():
|
||
"""退避延迟不超过 60s 上限。"""
|
||
# attempt=20 会产生 1.5*2^20 ≈ 1.5M，远超上限
|
||
delay = compute_backoff_delay(20)
|
||
assert delay <= RETRY_BACKOFF_CAP
|
||
|
||
|
||
def test_backoff_increases_with_attempt():
|
||
"""退避延迟随 attempt 增加（允许抖动误差）。"""
|
||
# 多次取均值避免抖动干扰
|
||
import random
|
||
random.seed(42)
|
||
delays = [compute_backoff_delay(0) for _ in range(100)]
|
||
avg0 = sum(delays) / len(delays)
|
||
random.seed(42)
|
||
delays = [compute_backoff_delay(3) for _ in range(100)]
|
||
avg3 = sum(delays) / len(delays)
|
||
assert avg3 > avg0
|
||
|
||
|
||
def test_backoff_custom_cap():
|
||
"""自定义上限生效。"""
|
||
delay = compute_backoff_delay(20, cap=5.0)
|
||
assert delay <= 5.0
|
||
|
||
|
||
def test_backoff_attempt_zero_positive():
|
||
"""attempt=0 时延迟为正。"""
|
||
delay = compute_backoff_delay(0)
|
||
assert delay > 0
|