q792602257andClaude Opus 5
865bbe3724
重试链路的关键信息是「升级路径」，而属性表达不了过程：同名属性后写覆盖先写， 三次尝试跑完只剩最后一次的状态码，前两次为什么失败、升到哪一级全被盖掉。 - site_session / rakuma_session：每次尝试各记一条 scrape.attempt event，带 attempt / outcome / status_code / 截断后的错误串；outcome 区分 http_error、 bad_status、challenge、validate_failed - site_session 的每次升级各记一条 scrape.escalate：rewarm_on_home、浏览器兜底 recovered / failed。兜底失败也要记——不然链路里只剩「最终失败」，看不出浏览器 这一级试过没有，而「没装 playwright」和「装了也被挡」是两个查法（reason 取 BrowserFallback.unavailable_reason，为 None 时由 add_event 跳过） - 三处 span.record_exception 换成 record_error，失败的错误码与可重试位跟着落下来 - 失败页面快照带上 url 与 profile，省得回头对是哪条通道的哪个地址 测试：test_scrape_telemetry.py 补一条会话层用例，用 MockTransport 让所有请求都回 挑战页，断言三次尝试各留一条 event（而不是只剩最后一次）、升级路径依次是 rewarm_on_home → browser，以及浏览器那一级的 outcome=failed 与 reason。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
419 lines
18 KiB
Python
419 lines
18 KiB
Python
"""站点会话：带 Akamai cookie 复用的 HTTP 抓取通道，附带浏览器兜底
|
|
|
|
乐天前置 Akamai Bot Manager，行为特征（实测）：
|
|
- 请求头不完整、无 cookie 时不封禁，而是把每个响应拖到 ~11s（与响应体大小无关）
|
|
- 补齐浏览器导航请求头并复用 Akamai 下发的 cookie 后，稳定在 ~0.6-0.9s
|
|
|
|
因此这里按「指纹画像」维护两条独立通道：搜索页用 PC 画像，商品详情页用手机
|
|
画像（详情页只有手机 UA 才返回带 __INITIAL_STATE__ 的统一模板）。两条通道
|
|
各自持有独立 cookie 罐，避免把 PC 指纹拿到的 cookie 混用到手机请求上。
|
|
|
|
**不预热首页**：Akamai 的 cookie 是随任意一个页面响应下发的，目标页自己就会带回
|
|
来，专门先打一次 `www.rakuten.co.jp/` 除了多一个出站请求（以及多一次被风控计数
|
|
的机会）之外没有额外收益——首个请求无论打哪个 URL 都是冷的 ~11s，之后都复用
|
|
cookie。首页只在**失败修复**路径上使用：目标页已经被挑战时，拿首页换一套干净
|
|
cookie 比继续拿目标页去撞更安全（见 `_rewarm_on_home`）。
|
|
|
|
抓取失败时的升级路径：换 cookie（首页重新预热）→ 浏览器兜底取 cookie → 放弃。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from opentelemetry import trace
|
|
|
|
from app.scraping.core import site
|
|
from app.shared.config import Settings
|
|
from app.shared.proxy import httpx_client_options
|
|
from app.shared.errors import (
|
|
ItemNotFoundError,
|
|
ResourceBusyError,
|
|
UpstreamBlockedError,
|
|
UpstreamRequestError,
|
|
)
|
|
from app.scraping.parsers.state import PageValidator, require_state_marker
|
|
from app.scraping.services.browser_fallback import BrowserFallback
|
|
from app.shared.telemetry import add_event, record_error, snapshot
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 逐次尝试 event 里错误串的截断长度：httpx 异常消息可能很长（带完整 URL 与底层
|
|
# socket 错误），这里只用来区分「这次是怎么失败的」，前半句够了
|
|
_ERROR_MAX_CHARS = 200
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class FetchedPage:
|
|
"""一次成功抓取的产物"""
|
|
|
|
html: str
|
|
url: str # 最终落地 URL；发生跳转时与请求地址不同
|
|
|
|
# Akamai 拦截页/挑战页的特征串
|
|
_BLOCK_MARKERS = (
|
|
"access denied",
|
|
"pardon our interruption",
|
|
"reference #",
|
|
"errors.edgesuite.net",
|
|
"/_sec/cp_challenge/",
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class _Profile:
|
|
"""一条指纹通道：独立的 httpx 客户端、cookie 罐与预热状态"""
|
|
|
|
name: str
|
|
mobile: bool
|
|
client: httpx.AsyncClient
|
|
lock: asyncio.Lock = field(default_factory=asyncio.Lock)
|
|
# Akamai cookie 罐的建立时刻（monotonic）。0 表示当前没有可复用的 cookie。
|
|
# 超过 session_ttl_seconds 就主动清空：拿着过期 cookie 去撞反而更容易被挑战。
|
|
cookies_at: float = 0.0
|
|
|
|
@property
|
|
def cookie_names(self) -> set[str]:
|
|
return {cookie.name for cookie in self.client.cookies.jar}
|
|
|
|
@property
|
|
def akamai_cookies(self) -> set[str]:
|
|
return self.cookie_names & set(site.AKAMAI_COOKIE_NAMES)
|
|
|
|
|
|
class SiteSession:
|
|
"""乐天站点抓取会话，管理 cookie 预热、并发限流与失败升级"""
|
|
|
|
def __init__(self, settings: Settings, browser_fallback: BrowserFallback):
|
|
self._settings = settings
|
|
self._browser = browser_fallback
|
|
self._semaphore = asyncio.Semaphore(settings.max_site_concurrency)
|
|
self._profiles: dict[str, _Profile] = {}
|
|
|
|
# ---- 生命周期 ----
|
|
|
|
async def start(self) -> None:
|
|
"""创建两条指纹通道的 HTTP 客户端"""
|
|
for name, mobile in (("pc", False), ("sp", True)):
|
|
if name in self._profiles:
|
|
continue
|
|
self._profiles[name] = _Profile(
|
|
name=name,
|
|
mobile=mobile,
|
|
client=httpx.AsyncClient(
|
|
headers=site.default_headers(mobile=mobile),
|
|
timeout=self._settings.request_timeout_seconds,
|
|
follow_redirects=True,
|
|
**httpx_client_options(self._settings),
|
|
http2=True,
|
|
),
|
|
)
|
|
logger.info(
|
|
"站点会话已就绪：profiles=%s concurrency=%s proxy=%s",
|
|
list(self._profiles),
|
|
self._settings.max_site_concurrency,
|
|
bool(self._settings.proxy_server),
|
|
)
|
|
|
|
async def close(self) -> None:
|
|
"""关闭所有 HTTP 客户端"""
|
|
for profile in self._profiles.values():
|
|
try:
|
|
await profile.client.aclose()
|
|
except Exception:
|
|
logger.debug("关闭 HTTP 客户端失败：profile=%s", profile.name, exc_info=True)
|
|
self._profiles.clear()
|
|
|
|
# ---- 状态 ----
|
|
|
|
def profile_status(self) -> dict[str, dict[str, Any]]:
|
|
"""各通道的 cookie 状态，供健康检查展示
|
|
|
|
`warmed` 保留原字段名（上游健康检查看板在用），语义是「当前有可复用的
|
|
Akamai cookie」——不再代表「已专门预热过首页」，因为正常路径不打首页了。
|
|
"""
|
|
now = time.monotonic()
|
|
return {
|
|
name: {
|
|
"warmed": profile.cookies_at > 0,
|
|
"age_seconds": round(now - profile.cookies_at, 1) if profile.cookies_at else None,
|
|
"cookies": sorted(profile.akamai_cookies),
|
|
}
|
|
for name, profile in self._profiles.items()
|
|
}
|
|
|
|
# ---- 抓取 ----
|
|
|
|
async def fetch_html(self, url: str, *, mobile: bool) -> str:
|
|
"""抓取要求含 __INITIAL_STATE__ 的页面，只取 HTML"""
|
|
page = await self.fetch(url, mobile=mobile)
|
|
return page.html
|
|
|
|
async def fetch(
|
|
self,
|
|
url: str,
|
|
*,
|
|
mobile: bool,
|
|
validator: PageValidator | None = None,
|
|
) -> FetchedPage:
|
|
"""抓取页面，返回 HTML 与最终落地地址
|
|
|
|
直接打目标页（不先访问首页），失败时按 换 cookie → 浏览器兜底 的顺序
|
|
逐级升级重试。
|
|
|
|
Args:
|
|
validator: 页面校验器，默认要求页面含 __INITIAL_STATE__。跨站抓取时
|
|
由调用方注入按落地域名分派的校验逻辑。
|
|
|
|
Raises:
|
|
ItemNotFoundError: 目标页面 404
|
|
UpstreamBlockedError: 反复被反爬阻断
|
|
UpstreamRequestError: 网络异常或上游 5xx
|
|
ResourceBusyError: 等待并发槽位超时
|
|
AppError: 校验器判定为不可重试的失败（如落到不支持的站点）
|
|
"""
|
|
tracer = trace.get_tracer(__name__)
|
|
profile = self._profiles["sp" if mobile else "pc"]
|
|
max_attempts = max(1, self._settings.http_max_attempts)
|
|
validate = validator or require_state_marker
|
|
last_error: str = "unknown error"
|
|
# 失败时供 trace snapshot 上报：保留最后一次拿到的响应体（含挑战页 / 5xx 响应）。
|
|
last_text: str | None = None
|
|
|
|
with tracer.start_as_current_span("scrape.fetch") as span:
|
|
span.set_attribute("scrape.url", url)
|
|
span.set_attribute("scrape.profile", profile.name)
|
|
|
|
try:
|
|
await asyncio.wait_for(
|
|
self._semaphore.acquire(),
|
|
timeout=self._settings.request_timeout_seconds,
|
|
)
|
|
except TimeoutError as exc:
|
|
err = ResourceBusyError()
|
|
record_error(span, err)
|
|
span.set_attribute("scrape.fail_reason", "ResourceBusyError")
|
|
raise err from exc
|
|
|
|
try:
|
|
for attempt in range(1, max_attempts + 1):
|
|
span.set_attribute("scrape.attempts", attempt)
|
|
# 过期 cookie 主动丢掉：带着它去撞比裸请求更容易吃挑战页
|
|
await self._drop_expired_cookies(profile)
|
|
try:
|
|
response = await profile.client.get(url)
|
|
except httpx.HTTPError as exc:
|
|
last_error = f"{type(exc).__name__}: {exc}"
|
|
# 每次尝试各记一条 event。属性同名后写覆盖先写，三次尝试跑完
|
|
# 只剩最后一次的状态码，前两次为什么失败、升级到哪一级全被
|
|
# 盖掉——而这条链路的关键信息恰恰是「升级路径」。
|
|
add_event(span, "scrape.attempt", {
|
|
"attempt": attempt,
|
|
"outcome": "http_error",
|
|
"error": last_error[:_ERROR_MAX_CHARS],
|
|
})
|
|
logger.warning(
|
|
"抓取请求异常：url=%s profile=%s attempt=%s/%s err=%s",
|
|
url, profile.name, attempt, max_attempts, last_error,
|
|
)
|
|
continue
|
|
|
|
# 任何响应都可能带 set-cookie（Akamai 不保证每次下发），拿到就
|
|
# 记下时刻，后续请求复用到 TTL 为止。
|
|
self._note_cookies(profile)
|
|
|
|
if response.status_code == 404:
|
|
err = ItemNotFoundError(f"Page not found: {url}")
|
|
record_error(span, err)
|
|
span.set_attribute("scrape.fail_reason", "ItemNotFoundError")
|
|
raise err
|
|
|
|
text = response.text
|
|
final_url = str(response.url)
|
|
span.set_attribute("scrape.last_status_code", response.status_code)
|
|
span.set_attribute("scrape.final_url", final_url)
|
|
span.set_attribute("scrape.html_bytes", len(text))
|
|
|
|
if response.status_code < 400:
|
|
# 校验器可能直接抛 AppError 表示「重试也没用」，此处不拦截
|
|
reason = validate(text, final_url)
|
|
if reason is None:
|
|
return FetchedPage(html=text, url=final_url)
|
|
last_error = self._refine_failure(reason, text)
|
|
challenged = last_error.startswith("challenge page detected")
|
|
if challenged:
|
|
span.set_attribute("scrape.challenge_detected", True)
|
|
outcome = "challenge" if challenged else "validate_failed"
|
|
else:
|
|
last_error = self._describe_error_status(response.status_code)
|
|
outcome = "bad_status"
|
|
last_text = text
|
|
add_event(span, "scrape.attempt", {
|
|
"attempt": attempt,
|
|
"outcome": outcome,
|
|
"status_code": response.status_code,
|
|
"html_bytes": len(text),
|
|
"error": last_error[:_ERROR_MAX_CHARS],
|
|
})
|
|
logger.warning(
|
|
"抓取结果异常：url=%s profile=%s attempt=%s/%s %s",
|
|
url, profile.name, attempt, max_attempts, last_error,
|
|
)
|
|
|
|
if attempt >= max_attempts:
|
|
break
|
|
|
|
# 第一次失败先便宜地换一套 cookie（丢掉旧的，用首页换新的）；
|
|
# 仍失败才动用浏览器
|
|
if attempt == 1:
|
|
await self._rewarm_on_home(profile)
|
|
span.set_attribute("scrape.rewarmed_on_home", True)
|
|
add_event(span, "scrape.escalate", {
|
|
"attempt": attempt, "to": "rewarm_on_home",
|
|
})
|
|
else:
|
|
page = await self._escalate_to_browser(profile, url, validate)
|
|
if page is not None:
|
|
span.set_attribute("scrape.fell_back_to_browser", True)
|
|
span.set_attribute("scrape.final_url", page.url)
|
|
add_event(span, "scrape.escalate", {
|
|
"attempt": attempt, "to": "browser", "outcome": "recovered",
|
|
})
|
|
return page
|
|
# 兜底没救回来（浏览器不可用，或取回的页面仍不合格）。不记
|
|
# 这条的话链路里只看得到「最终失败」，看不出浏览器这一级到底
|
|
# 试过没有——而「没装 playwright」和「装了也被挡」要分开查。
|
|
add_event(span, "scrape.escalate", {
|
|
"attempt": attempt,
|
|
"to": "browser",
|
|
"outcome": "failed",
|
|
"reason": self._browser.unavailable_reason,
|
|
})
|
|
|
|
if self._is_server_error(last_error):
|
|
err = UpstreamRequestError(f"Upstream request failed: {last_error}")
|
|
else:
|
|
err = UpstreamBlockedError(f"Blocked while fetching {url}: {last_error}")
|
|
record_error(span, err)
|
|
span.set_attribute("scrape.fail_reason", type(err).__name__)
|
|
snapshot(
|
|
span, "scrape.failed_html", last_text,
|
|
self._settings.otel_snapshot_max_bytes,
|
|
extra={"scrape.url": url, "scrape.profile": profile.name},
|
|
)
|
|
raise err
|
|
finally:
|
|
self._semaphore.release()
|
|
|
|
# ---- 内部 ----
|
|
|
|
@staticmethod
|
|
def _describe_error_status(status_code: int) -> str:
|
|
return (
|
|
f"upstream status {status_code}"
|
|
if status_code >= 500
|
|
else f"status {status_code}"
|
|
)
|
|
|
|
@staticmethod
|
|
def _refine_failure(reason: str, text: str) -> str:
|
|
"""页面内容校验失败时，优先报出更具体的反爬挑战页特征"""
|
|
lowered = text[:4000].lower()
|
|
matched = next((marker for marker in _BLOCK_MARKERS if marker in lowered), None)
|
|
return f"challenge page detected ({matched})" if matched else reason
|
|
|
|
@staticmethod
|
|
def _is_server_error(reason: str) -> bool:
|
|
return reason.startswith("upstream status") or reason.startswith("httpx") or "Error:" in reason
|
|
|
|
def _note_cookies(self, profile: _Profile) -> None:
|
|
"""目标页响应带回 Akamai cookie 时记下时刻，作为 TTL 起点
|
|
|
|
已经在计时的不重置：TTL 要从「这套 cookie 第一次出现」算起，每次响应都
|
|
刷新会让一套 cookie 被无限续命，反而绕过了 session_ttl_seconds 的本意。
|
|
"""
|
|
if profile.cookies_at:
|
|
return
|
|
if profile.akamai_cookies:
|
|
profile.cookies_at = time.monotonic()
|
|
|
|
async def _drop_expired_cookies(self, profile: _Profile) -> None:
|
|
"""cookie 罐超过 TTL 时清空，让下一次请求裸奔换一套新的"""
|
|
if not profile.cookies_at:
|
|
return
|
|
if time.monotonic() - profile.cookies_at <= self._settings.session_ttl_seconds:
|
|
return
|
|
async with profile.lock:
|
|
profile.client.cookies.clear()
|
|
profile.cookies_at = 0.0
|
|
logger.info("会话 cookie 已过期，已清空：profile=%s", profile.name)
|
|
|
|
async def _rewarm_on_home(self, profile: _Profile) -> None:
|
|
"""首次失败后的修复：丢掉旧 cookie，用首页换一套新的
|
|
|
|
这是首页 URL 唯一的用途。目标页已经吃了挑战页，继续拿同一个 URL 去撞
|
|
只会把挑战坐实；首页是站点最"无害"的入口，换 cookie 的成功率更高。
|
|
|
|
失败不阻断本次抓取（下一次 attempt 会裸请求目标页，只是慢），所以这里
|
|
只记日志。
|
|
"""
|
|
async with profile.lock:
|
|
profile.client.cookies.clear()
|
|
profile.cookies_at = 0.0
|
|
try:
|
|
response = await profile.client.get(self._settings.home_url)
|
|
except httpx.HTTPError as exc:
|
|
logger.warning("首页换 cookie 失败：profile=%s err=%s", profile.name, exc)
|
|
return
|
|
if profile.akamai_cookies:
|
|
profile.cookies_at = time.monotonic()
|
|
logger.info(
|
|
"已用首页换一套新 cookie：profile=%s status=%s cookies=%s",
|
|
profile.name,
|
|
response.status_code,
|
|
sorted(profile.akamai_cookies),
|
|
)
|
|
|
|
async def _escalate_to_browser(
|
|
self, profile: _Profile, url: str, validate: PageValidator
|
|
) -> FetchedPage | None:
|
|
"""用浏览器访问目标页，把 cookie 回灌给 HTTP 客户端
|
|
|
|
浏览器已经拿到合格页面时直接返回，省掉一次重复请求。
|
|
"""
|
|
visit = await self._browser.visit(url, mobile=profile.mobile)
|
|
if visit is None:
|
|
logger.warning(
|
|
"浏览器兜底不可用，放弃升级：profile=%s reason=%s",
|
|
profile.name,
|
|
self._browser.unavailable_reason,
|
|
)
|
|
return None
|
|
|
|
async with profile.lock:
|
|
for cookie in visit.cookies:
|
|
name = cookie.get("name")
|
|
value = cookie.get("value")
|
|
if not name or value is None:
|
|
continue
|
|
profile.client.cookies.set(
|
|
name,
|
|
value,
|
|
domain=cookie.get("domain") or "",
|
|
path=cookie.get("path") or "/",
|
|
)
|
|
profile.cookies_at = time.monotonic()
|
|
|
|
# 浏览器不回报最终 URL，这里以请求地址为准；跨站跳转场景下 HTTP 通道已先行
|
|
# 报错，走不到这一步。
|
|
if validate(visit.html, url) is None:
|
|
logger.info("浏览器兜底直接取回页面：url=%s profile=%s", url, profile.name)
|
|
return FetchedPage(html=visit.html, url=url)
|
|
|
|
logger.warning("浏览器兜底页面仍未通过校验：url=%s profile=%s", url, profile.name)
|
|
return None
|