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

240 lines
10 KiB
Python

"""乐天抓取客户端:把请求参数翻译成站点 URL,抓取后解析为结构化数据
两个接口分别走不同的指纹通道:
- 搜索:PC 通道(search.rakuten.co.jp 的搜索页)
- 详情:手机通道(item.rakuten.co.jp 只有手机 UA 才返回统一模板)
"""
from __future__ import annotations
import logging
from opentelemetry import trace
from app.scraping.core import site
from app.shared.config import Settings
from app.shared.errors import ScrapeParseError
from app.scraping.models.scrape import (
GenreData,
GenreRequest,
ItemDetailData,
ItemDetailRequest,
SearchRequest,
SearchResultData,
ShopDetailData,
ShopDetailRequest,
ShopItemsRequest,
)
from app.scraping.parsers.genre import parse_genres
from app.scraping.parsers.item import parse_item_detail
from app.scraping.parsers.search import parse_search
from app.scraping.parsers.shop import parse_shop_detail
from app.scraping.parsers.state import extract_initial_state
from app.scraping.parsers.subsites import (
SubsitePage,
build_item_page_validator,
host_of,
parse_subsite_item,
)
from app.scraping.services.site_session import SiteSession
from app.scraping.utils.urls import (
build_genre_url,
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 RakutenClient:
"""乐天市场抓取客户端"""
def __init__(self, settings: Settings, session: SiteSession):
self._settings = settings
self._session = session
async def search(self, payload: SearchRequest) -> SearchResultData:
"""抓取搜索结果列表"""
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.rakuten.search") as span:
span.set_attribute("parse.site", "rakuten")
span.set_attribute("parse.op", "search")
try:
html = await self._session.fetch_html(url, mobile=False)
state = extract_initial_state(html)
result = parse_search(
state,
request_url=url,
page=payload.page,
exclude_ads=payload.exclude_ads,
)
span.set_attribute("parse.items", len(result.items))
span.set_attribute("parse.ads", result.ad_count)
span.set_attribute("parse.total", result.total_count)
logger.info(
"搜索完成:url=%s items=%s ads=%s total=%s",
url, len(result.items), result.ad_count, 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 genres(self, payload: GenreRequest) -> GenreData:
"""抓取分类树:顶层分类列表,或指定分类的信息与直接子分类"""
genre_id = (payload.genre_id or "").strip() or None
url = build_genre_url(genre_id)
logger.info("抓取分类树:url=%s genre_id=%s", url, genre_id)
html: str | None = None
with tracer.start_as_current_span("parse.rakuten.genres") as span:
span.set_attribute("parse.site", "rakuten")
span.set_attribute("parse.op", "genres")
try:
html = await self._session.fetch_html(url, mobile=False)
state = extract_initial_state(html)
result = parse_genres(state, genre_id=genre_id)
span.set_attribute("parse.children", len(result.children))
logger.info(
"分类树完成:genre_id=%s name=%s children=%s",
result.genre_id or "root", result.name, len(result.children),
)
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 shop_detail(self, payload: ShopDetailRequest) -> ShopDetailData:
"""抓取商家详情"""
if payload.shop_url is not None:
shop_code = split_shop_url(str(payload.shop_url))
else:
shop_code = (payload.shop_code or "").strip()
url = build_shop_url(shop_code)
logger.info("抓取店铺页:url=%s", url)
html: str | None = None
with tracer.start_as_current_span("parse.rakuten.shop_detail") as span:
span.set_attribute("parse.site", "rakuten")
span.set_attribute("parse.op", "shop_detail")
try:
html = await self._session.fetch_html(url, mobile=False)
state = extract_initial_state(html)
result = parse_shop_detail(state, shop_code=shop_code, html=html)
span.set_attribute("parse.reviews", result.review_count)
logger.info(
"店铺详情完成:shop_code=%s shop_id=%s name=%s reviews=%s",
result.shop_code, result.shop_id, result.shop_name, result.review_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 shop_items(self, payload: ShopItemsRequest) -> SearchResultData:
"""抓取商家名下的商品列表
站点没有可分页抓取的「店铺内商品」页,店铺商品实际就是搜索结果按
`sid=` 限定店铺后的产物,因此这里转成一次搜索。只给 shop_code 时
需要先取一次店铺详情换出 shop_id。
"""
with tracer.start_as_current_span("parse.rakuten.shop_items") as span:
span.set_attribute("parse.site", "rakuten")
span.set_attribute("parse.op", "shop_items")
try:
shop_id = payload.shop_id
if shop_id is None:
detail = await self.shop_detail(ShopDetailRequest(shop_code=payload.shop_code))
if detail.shop_id is None:
raise ScrapeParseError(f"未能取得店铺 {payload.shop_code} 的 shop_id")
shop_id = detail.shop_id
result = await self.search(payload.to_search_request(shop_id))
span.set_attribute("parse.items", len(result.items))
span.set_attribute("parse.total", result.total_count)
return result
except Exception as exc:
# 本方法自己不抓页面,失败一定发生在它转调的 shop_detail / search
# 里(那两个 span 已各自记了自己的失败页面),所以显式标 delegate、
# 不落快照:按 html 推断只会得出「fetch 失败」的错误结论。
record_parse_failure(span, exc, stage="delegate")
raise
async def item_detail(self, payload: ItemDetailRequest) -> ItemDetailData:
"""抓取商品详情"""
if payload.item_url is not None:
shop_code, item_code = split_item_url(str(payload.item_url))
else:
# 模型校验已保证此分支下两者都非空
shop_code, item_code = payload.shop_code or "", payload.item_code or ""
# 统一收敛到规范地址,去掉 variantId 等查询参数——所有 SKU 组合都会随详情返回
url = build_item_url(shop_code, item_code)
logger.info("抓取商品详情:url=%s", url)
html: str | None = None
with tracer.start_as_current_span("parse.rakuten.item_detail") as span:
span.set_attribute("parse.site", "rakuten")
span.set_attribute("parse.op", "item_detail")
try:
# 部分官方旗舰店的商品页会跳出市场域名,落到各自的独立站点;
# 校验与解析都按最终落地域名分派,落到未登记站点时由校验器抛错。
page = await self._session.fetch(
url, mobile=True, validator=build_item_page_validator(url)
)
html = page.html
if host_of(page.url) == site.ITEM_HOST:
detail = parse_item_detail(
extract_initial_state(page.html),
item_url=url,
shop_code=shop_code,
include_sku_variants=payload.include_sku_variants,
)
else:
detail = parse_subsite_item(
SubsitePage(
html=page.html,
requested_url=url,
final_url=page.url,
shop_code=shop_code,
item_code=item_code,
include_sku_variants=payload.include_sku_variants,
)
)
span.set_attribute("parse.source", detail.source)
span.set_attribute("parse.price", detail.price)
span.set_attribute("parse.variant_count", detail.sku.variant_count)
logger.info(
"详情完成:url=%s source=%s name=%s price=%s skus=%s",
url, detail.source, detail.item_name[:40], detail.price, detail.sku.variant_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