Files
q792602257andClaude Opus 5 8381896eeb feat(observability): 失败响应与网关信封进链路,手工埋点尊重 suppressed
失败此前在 trace 里近乎不可见:异常处理器把异常吃掉换成信封响应,自动
instrumentation 只看到一个 HTTP 状态码,而 AppError 默认 400、信封里
success=false,跟正常返回分不出来。

- api.py:四个异常处理器(对外失败的唯一出口)各记一次 span;兜底处理器额外
  record_error——对外只回一句无信息量的错误文案,异常类型与栈只在本地日志里
- telemetry.py:新增 record_envelope / record_parse_failure /
  span_unless_suppressed;record_error 补 error.message / retryable /
  status_code;snapshot 支持 extra 带上「这份 HTML 是哪来的」
- worker/client.py:_request 自建 span,活到解信封之后。httpx 那个 CLIENT span
  在 request() 返回时就结束,此时信封还没解——success=false code=6002(租约
  无效)在它看来是完成的 200 请求
- span_unless_suppressed:suppress_instrumentation 只被 instrumentation 库尊重,
  手工 span 不看它,lease 空转长轮询会从这个口子把孤立 trace 放回来
- 两个 scraping client 的解析失败分支收拢到 record_parse_failure;shop_items
  显式标 stage=delegate 且不落快照(它自己不抓页面,按 html 推断只会得出
  「fetch 失败」的错误结论)

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-28 16:07:00 +08:00

225 lines
9.4 KiB
Python

"""ラクマ 抓取客户端:把请求参数翻译成站点 URL,抓取后解析为结构化数据
五个接口都走同一条 HTTP 通道(ラクマ 无 Akamai 限速,不需要分指纹通道):
搜索、分类、商品详情、卖家详情、卖家商品列表。
"""
from __future__ import annotations
import asyncio
import logging
from opentelemetry import trace
from app.scraping.core import rakuma_site as site
from app.shared.config import Settings
from app.scraping.models.scrape import (
RakumaCategoryData,
RakumaCategoryRequest,
RakumaItemDetailData,
RakumaItemDetailRequest,
RakumaSearchRequest,
RakumaSearchResultData,
RakumaShopDetailData,
RakumaShopDetailRequest,
RakumaShopItemsData,
RakumaShopItemsRequest,
)
from app.scraping.parsers.rakuma.category import parse_categories
from app.scraping.parsers.rakuma.item import parse_item_detail
from app.scraping.parsers.rakuma.search import parse_search
from app.scraping.parsers.rakuma.shop import parse_shop_detail, parse_shop_items
from app.scraping.services.rakuma_session import RakumaSession
from app.scraping.utils.rakuma_urls import (
build_item_url,
build_search_url,
build_shop_url,
normalize_search_url,
split_item_url,
split_shop_url,
)
from app.shared.telemetry import record_parse_failure
logger = logging.getLogger(__name__)
tracer = trace.get_tracer(__name__)
class RakumaClient:
"""ラクマ(fril.jp)抓取客户端"""
def __init__(self, settings: Settings, session: RakumaSession):
self._settings = settings
self._session = session
async def search(self, payload: RakumaSearchRequest) -> RakumaSearchResultData:
"""抓取搜索结果列表"""
if payload.search_url is not None:
url = normalize_search_url(str(payload.search_url), payload.page)
else:
url = build_search_url(payload)
logger.info("抓取 ラクマ 搜索页:url=%s", url)
html: str | None = None
with tracer.start_as_current_span("parse.rakuma.search") as span:
span.set_attribute("parse.site", "rakuma")
span.set_attribute("parse.op", "search")
try:
html = await self._session.fetch_html(url)
result = parse_search(
html,
request_url=url,
page=payload.page,
keyword=payload.keyword.strip(),
)
span.set_attribute("parse.items", len(result.items))
span.set_attribute("parse.total", result.total_count)
logger.info(
"ラクマ 搜索完成:url=%s items=%s total=%s",
url, len(result.items), result.total_count,
)
return result
except Exception as exc:
record_parse_failure(
span, exc, html=html,
max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
)
raise
async def categories(self, payload: RakumaCategoryRequest) -> RakumaCategoryData:
"""抓取分类树
分类一览页一次就返回整棵树(与 URL 上的 category_id 无关),因此不论
查哪一层都只打一次请求,页面地址也固定不带参数。
"""
url = site.CATEGORY_LIST_URL
logger.info("抓取 ラクマ 分类页:category_id=%s", payload.category_id)
html: str | None = None
with tracer.start_as_current_span("parse.rakuma.categories") as span:
span.set_attribute("parse.site", "rakuma")
span.set_attribute("parse.op", "categories")
try:
html = await self._session.fetch_html(url)
data = parse_categories(
html,
category_id=payload.category_id,
include_descendants=payload.include_descendants,
)
span.set_attribute("parse.children", len(data.children))
span.set_attribute("parse.tree", data.total_count)
logger.info(
"ラクマ 分类完成:category_id=%s name=%s children=%s tree=%s",
data.category_id, data.name, len(data.children), data.total_count,
)
return data
except Exception as exc:
record_parse_failure(
span, exc, html=html,
max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
)
raise
async def item_detail(self, payload: RakumaItemDetailRequest) -> RakumaItemDetailData:
"""抓取商品详情"""
if payload.item_url is not None:
item_id = split_item_url(str(payload.item_url))
else:
item_id = (payload.item_id or "").strip()
url = build_item_url(item_id)
logger.info("抓取 ラクマ 商品详情:url=%s", url)
html: str | None = None
with tracer.start_as_current_span("parse.rakuma.item_detail") as span:
span.set_attribute("parse.site", "rakuma")
span.set_attribute("parse.op", "item_detail")
try:
html = await self._session.fetch_html(url)
detail = parse_item_detail(html, item_id=item_id, item_url=url)
span.set_attribute("parse.price", detail.price)
span.set_attribute("parse.sold_out", detail.is_sold_out)
logger.info(
"ラクマ 详情完成:url=%s name=%s price=%s sold_out=%s",
url, detail.item_name[:40], detail.price, detail.is_sold_out,
)
return detail
except Exception as exc:
record_parse_failure(
span, exc, html=html,
max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
)
raise
async def shop_detail(self, payload: RakumaShopDetailRequest) -> RakumaShopDetailData:
"""抓取卖家详情
评价明细在单独的 /review 页上,仅在 include_reviews=true 时并发多取一次。
"""
shop_id = self._resolve_shop_id(payload.shop_url, payload.shop_id)
url = build_shop_url(shop_id)
logger.info("抓取 ラクマ 卖家详情:url=%s reviews=%s", url, payload.include_reviews)
html: str | None = None
with tracer.start_as_current_span("parse.rakuma.shop_detail") as span:
span.set_attribute("parse.site", "rakuma")
span.set_attribute("parse.op", "shop_detail")
try:
if payload.include_reviews:
html, review_html = await asyncio.gather(
self._session.fetch_html(url),
self._session.fetch_html(build_shop_url(shop_id, review=True)),
)
else:
html, review_html = await self._session.fetch_html(url), None
detail = parse_shop_detail(
html, shop_id=shop_id, shop_url=url, review_html=review_html
)
span.set_attribute("parse.items", detail.item_count)
span.set_attribute("parse.reviews", detail.review_count)
logger.info(
"ラクマ 卖家详情完成:shop_id=%s name=%s items=%s reviews=%s",
shop_id, detail.shop_name, detail.item_count, detail.review_count,
)
return detail
except Exception as exc:
record_parse_failure(
span, exc, html=html,
max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
)
raise
async def shop_items(self, payload: RakumaShopItemsRequest) -> RakumaShopItemsData:
"""抓取卖家的商品列表"""
shop_id = self._resolve_shop_id(payload.shop_url, payload.shop_id)
url = build_shop_url(shop_id, page=payload.page)
logger.info("抓取 ラクマ 卖家商品:url=%s", url)
html: str | None = None
with tracer.start_as_current_span("parse.rakuma.shop_items") as span:
span.set_attribute("parse.site", "rakuma")
span.set_attribute("parse.op", "shop_items")
try:
html = await self._session.fetch_html(url)
result = parse_shop_items(
html, shop_id=shop_id, request_url=url, page=payload.page
)
span.set_attribute("parse.items", len(result.items))
span.set_attribute("parse.total", result.total_count)
logger.info(
"ラクマ 卖家商品完成:shop_id=%s items=%s total=%s",
shop_id, len(result.items), result.total_count,
)
return result
except Exception as exc:
record_parse_failure(
span, exc, html=html,
max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
)
raise
@staticmethod
def _resolve_shop_id(shop_url: object, shop_id: str | None) -> str:
"""从 shop_url 或 shop_id 里取出店铺 hash(模型校验已保证两者不同时为空)"""
if shop_url is not None:
return split_shop_url(str(shop_url))
return (shop_id or "").strip()