"""ラクマ(fril.jp)站点会话:单通道 HTTP 抓取 与乐天市场那条链路(app/services/site_session.py)分开维护,因为两站的 抓取前提完全不同: - 乐天前置 Akamai Bot Manager,无 cookie 时每个响应被拖到 ~11s,必须先访问 首页预热再复用 cookie;且搜索页与详情页要用不同 UA,需要两条指纹通道。 - ラクマ 实测没有这类限速:冷请求(无 cookie、无预热)即 0.5-1.1s,与预热后 持平;PC UA 在搜索页、详情页、店铺页上都能拿到完整模板。 因此这里只有一条通道、不做预热,也不接浏览器兜底——没有需要兜底的拦截行为。 若将来站点加了防护,再按乐天那套补预热与升级链路。 """ from __future__ import annotations import asyncio import logging from typing import Any import httpx from opentelemetry import trace from app.scraping.core import rakuma_site as 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.shared.telemetry import add_event, record_error, snapshot logger = logging.getLogger(__name__) # 逐次尝试 event 里错误串的截断长度:httpx 的异常消息可能很长(带完整 URL 与 # 底层 socket 错误),而这里只是用来区分「这次是怎么失败的」,前半句就够 _ERROR_MAX_CHARS = 200 class RakumaSession: """ラクマ 站点抓取会话,管理并发限流与失败重试""" def __init__(self, settings: Settings): self._settings = settings self._semaphore = asyncio.Semaphore(settings.max_site_concurrency) self._client: httpx.AsyncClient | None = None # ---- 生命周期 ---- async def start(self) -> None: """创建 HTTP 客户端""" if self._client is not None: return self._client = httpx.AsyncClient( headers=site.default_headers(), timeout=self._settings.request_timeout_seconds, follow_redirects=True, **httpx_client_options(self._settings), http2=True, ) logger.info( "ラクマ 会话已就绪:concurrency=%s proxy=%s", self._settings.max_site_concurrency, bool(self._settings.proxy_server), ) async def close(self) -> None: """关闭 HTTP 客户端""" if self._client is None: return try: await self._client.aclose() except Exception: logger.debug("关闭 ラクマ HTTP 客户端失败", exc_info=True) self._client = None # ---- 状态 ---- def status(self) -> dict[str, Any]: """会话状态,供健康检查展示""" return {"ready": self._client is not None} # ---- 抓取 ---- async def fetch_html(self, url: str) -> str: """抓取页面 HTML Raises: ItemNotFoundError: 目标页面 404(商品已下架或 ID 不存在) UpstreamRequestError: 网络异常或上游 5xx UpstreamBlockedError: 反复取不到正常页面 ResourceBusyError: 等待并发槽位超时 """ client = self._client if client is None: raise UpstreamRequestError("ラクマ 会话尚未初始化") tracer = trace.get_tracer(__name__) max_attempts = max(1, self._settings.http_max_attempts) last_error = "unknown error" 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", "rakuma") 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) try: response = await 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 attempt=%s/%s err=%s", url, attempt, max_attempts, last_error, ) continue 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 if response.status_code < 400: span.set_attribute("scrape.last_status_code", response.status_code) span.set_attribute("scrape.final_url", str(response.url)) span.set_attribute("scrape.html_bytes", len(response.text)) return response.text last_error = ( f"upstream status {response.status_code}" if response.status_code >= 500 else f"status {response.status_code}" ) last_text = response.text span.set_attribute("scrape.last_status_code", response.status_code) span.set_attribute("scrape.html_bytes", len(response.text)) add_event(span, "scrape.attempt", { "attempt": attempt, "outcome": "bad_status", "status_code": response.status_code, "error": last_error[:_ERROR_MAX_CHARS], }) logger.warning( "ラクマ 抓取结果异常:url=%s attempt=%s/%s %s", url, attempt, max_attempts, last_error, ) if last_error.startswith("upstream status") or ":" in last_error: err = UpstreamRequestError(f"Upstream request failed: {last_error}") else: err = UpstreamBlockedError(f"Failed to fetch {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": "rakuma"}, ) raise err finally: self._semaphore.release()