rakuten-api/app/scraping/services/rakuma_session.py

"""ラクマ（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()