163 lines
6.3 KiB
Python
163 lines
6.3 KiB
Python
"""ラクマ（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 snapshot
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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()
|
|
span.record_exception(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}"
|
|
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}")
|
|
span.record_exception(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))
|
|
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}")
|
|
span.record_exception(err)
|
|
span.set_attribute("scrape.fail_reason", type(err).__name__)
|
|
snapshot(span, "scrape.failed_html", last_text, self._settings.otel_snapshot_max_bytes)
|
|
raise err
|
|
finally:
|
|
self._semaphore.release()
|