Compare commits

..
5 Commits
Author SHA1 Message Date
q792602257andClaude Opus 5 865bbe3724 feat(observability): 抓取会话逐次尝试与升级路径记成 event
重试链路的关键信息是「升级路径」,而属性表达不了过程:同名属性后写覆盖先写,
三次尝试跑完只剩最后一次的状态码,前两次为什么失败、升到哪一级全被盖掉。

- site_session / rakuma_session:每次尝试各记一条 scrape.attempt event,带
  attempt / outcome / status_code / 截断后的错误串;outcome 区分 http_error、
  bad_status、challenge、validate_failed
- site_session 的每次升级各记一条 scrape.escalate:rewarm_on_home、浏览器兜底
  recovered / failed。兜底失败也要记——不然链路里只剩「最终失败」,看不出浏览器
  这一级试过没有,而「没装 playwright」和「装了也被挡」是两个查法(reason 取
  BrowserFallback.unavailable_reason,为 None 时由 add_event 跳过)
- 三处 span.record_exception 换成 record_error,失败的错误码与可重试位跟着落下来
- 失败页面快照带上 url 与 profile,省得回头对是哪条通道的哪个地址

测试:test_scrape_telemetry.py 补一条会话层用例,用 MockTransport 让所有请求都回
挑战页,断言三次尝试各留一条 event(而不是只剩最后一次)、升级路径依次是
rewarm_on_home → browser,以及浏览器那一级的 outcome=failed 与 reason。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-28 16:18:25 +08:00
q792602257andClaude Opus 5 fd7d89ab0a test(observability): 补异常处理器的 server span 埋点测试
8381896 的最后一块配套测试。挂真实的 FastAPIInstrumentor 中间件(而不是手工造
span),断言「一次真实请求打进来、失败返回之后,server span 上有什么」:

- AppError 默认 status_code=400,链路里光看状态码只知道「客户端错了」,分不出
  是反爬阻断(3002,可重试)还是别的;断言 error.code / error.retryable /
  api.* 都落到了 span 上,异常栈作为 event 保留
- 兜底分支最需要埋点:对外只回 "Internal server error",断言真正的
  RuntimeError 类型与消息在链路上看得到
- 校验失败记错误码与字段消息,但不记 pydantic 那条没有诊断价值的长栈
- 一条成功路径对照,证明 ERROR 状态与 api.* 属性不是恒真

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-28 16:10:16 +08:00
q792602257andClaude Opus 5 8403b9b586 test(observability): 补抓取客户端与网关客户端的埋点测试
上一提交(8381896)的配套测试,钉住两个失败可见性问题:

- 抓取侧回归:失败分支原先写的是无参 `span.record_exception()`,而 exception
  是必填位置参数——每次抓取失败都会在记异常时抛 TypeError,把真正的原因(反爬
  阻断 / 解析失败)整个顶替掉,失败页面快照也永远落不下来(抛错在 snapshot
  之前)。测试断言业务异常原样抛出、stage 能区分 fetch/parse/delegate、
  parse 阶段确实落了快照且带得上 URL
- 网关侧:success=false code=6002 在 HTTP 层是 200,断言 span 记下了信封结论;
  以及空转长轮询在 suppressed() 里确实不产 span,抑制之外照常产(防止空断言
  是因为埋点根本没生效)
- 两个文件都各带一条成功路径对照,证明 ERROR 状态不是恒真

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-28 16:08:35 +08:00
q792602257andClaude Opus 5 b577d3ac8d fix(trading): 必填选项自动填值跳过「選択してください」占位项,并把选项开放给接口
trading 自动填 choice 时取 values[0],而必填 select 的 values[0] 恒为 id=0 的
「選択してください」——等于把「请选择」当答案提交。4 份真实样本一致(真值从
id=200 起)。同时 /api/item_detail 完全不返回 options,调用方即使想显式指定
choice 也无从知道合法取值。

- purchase_contract.py:新增 ItemOption / ItemOptionValue 与 parse_options /
  auto_choice_for / format_choice。占位判定以结构为主(value_id == 0),日文
  文案仅作兜底。放 shared 是因为「接口声明的合法取值」与「下单实际提交的值」
  必须同源,否则两边各判一次迟早再次分叉
- item.py / scrape.py:ItemDetailData 增 options、has_required_options、
  unfillable_required_options;只解析一次,两个派生结果都取自同一份结果
- site_interact.py:auto_choice_for 取第一个非占位候选;必填项填不出值时
  报错点名是哪些选项,让调用方知道该在 intent.choice 里补什么
- auto_choice_for 只自动填必填项:非必填项要不要选是业务决定,不是我们该替
  调用方做的选择
- README / docs:补 options[] → intent.choice、variants[] → intent.variant_id
  的对照,修掉 order-gateway 示例里已不存在的 "options": {} 字段

真账号验证(scripts/probe_option_choice.py,仅加购不结算不支付):两个商品
提交 確認した / 了解致しました。均被站点接受,购物车 count=2,跑完清空恢复
原状。探针刻意走生产的 add_to_cart_payload 并从其日志截获实际 payload——
probe_purchase_block_v2.py 自己抄了一遍字段构造,与生产代码同错,正是这个
bug 当初藏住的原因。

未覆盖:这两家店铺本身不校验该选项(旧的占位值当年也被收下),所以只证明新值
走得通、语义上才是真答案,证明不了旧值会被拒;必填自由文本项(
unfillable_required_options)无真实样本,仅离线测试覆盖。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-28 16:07:33 +08:00
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
20 changed files with 1863 additions and 109 deletions
+24 -2
View File
@@ -373,6 +373,20 @@ docker compose --profile gateway up -d
搜索结果中的 `shop.shop_code` + `item_code` 可直接用作本接口入参。
**下单前要决定的参数全在响应里**(规格与选项是两套东西,下单时走不同字段):
| 字段 | 下单时对应 | 说明 |
| --- | --- | --- |
| `sku.variants[].variant_id` | `intent.variant_id` | 规格组合(颜色 × 尺码等) |
| `options[]` | `intent.choice` | 店铺自定义选项,格式「选项名:取值名」 |
| `has_required_options` | — | `true` 时不给 `choice` 会被站点拒绝加购 |
| `unfillable_required_options` | — | 必填但无法自动选值的选项名,**必须**由调用方给值 |
`options[].values[].is_placeholder` 标出「選択してください」这类占位项——它们不是
合法取值,拼 `choice` 时要跳过(`selectable_value_count` 已是剔除占位项后的数量)。
`type="text"` 的选项是自由文本(如「【お名前】4文字まで」),站点不给候选值,
必填时只能由调用方给值。
部分官方店的商品页会跳转到独立子站,响应里的 `source` / `source_url` 会标明数据来源,
字段覆盖差异见下一节。
@@ -432,6 +446,7 @@ docker compose --profile gateway up -d
| `review` | ✅ | ✅ | ❌ 异步加载 | ❌ 异步加载 |
| `sku.variants`(规格组合) | ✅ | — 图书无规格 | ✅ 颜色 × 尺码 | — 单一规格 |
| `sku.attributes`(规格表) | ✅ | ✅ 出版社/ISBN 等 | 挂在各 variant 上 | ❌ |
| `options`(店铺自定义选项) | ✅ | ❌ 未解析 | ❌ 未解析 | ❌ 未解析 |
| `shipping` 运费明细 | ✅ | ❌ 仅库存措辞 | ❌ | 仅「是否含运费」 |
| `shop.shop_id` | ✅ | ❌ | ✅ | ✅ |
| `breadcrumbs[].url` | ✅ | ✅ | ❌ 站内分类编码,拼不出链接 | ✅ |
@@ -602,6 +617,8 @@ docker compose --profile gateway up -d
| `purchase_unit` | 起订单位 |
| `sku.inventory_type` | `multiple` 表示多规格,要看 `sku.variants[]` |
| `sku.variants[].variant_id` | 多规格商品的规格 ID(trading 加购多规格时按此选择) |
| `options[]` | 店铺自定义选项(必填项不给值站点拒绝加购),加购时走 `choice` 字段 |
| `unfillable_required_options` | 必填但无法自动选值的选项名,必须由调用方显式给值 |
交易服务对外接口(:31108,全部需要 Bearer token):
@@ -611,8 +628,13 @@ docker compose --profile gateway up -d
- `POST /api/cart/clear` — 清空购物车(UI 点击 `button[aria-label="削除"]`
- `POST /api/cart/remove` — 删除指定 `item_id`
底层共用 `app/shared/purchase_contract.py` 的常量字段构造(与 scraping 模型解耦)。
trading 加购时的字段选择策略:多规格挑第一个非售罄的 variant;必填选项拼「名:值」。
底层共用 `app/shared/purchase_contract.py` 的常量字段构造与**选项解析**(与 scraping
模型解耦)——`/api/item_detail` 对外暴露的 `options[]` 和 trading 自动填 `choice` 用的
是同一份解析与占位项判定,避免「接口说能选的值」与「下单实际填的值」不一致。
trading 加购时的字段选择策略:多规格挑第一个非售罄的 variant;必填选项拼「名:值」,
取第一个**非占位**候选值(`values[0]` 往往是「選択してください」,填它等于没选)。
必填项自动填不出来(自由文本项、候选值只剩占位项)且调用方没给 `choice` 时当场报错
并点名是哪几项,不拿占位值凑数去撞站点。
站点端点 `basketDomain` 逐商品不同(实测有 `sp.basket…``ts.sp.basket…`),不能写死。
几点子站差异(仅信息,trading 不覆盖子站加购):
+54 -1
View File
@@ -338,6 +338,43 @@ class SkuInfo(BaseModel):
variant_count: int = Field(default=0, description="SKU 组合总数;不受 include_sku_variants 影响,始终为真实组合数")
class ItemOptionValue(BaseModel):
"""店铺自定义选项的一个候选取值"""
value_id: int | None = Field(default=None, description="站点 values[].id 原值;给不出数字时为 null")
name: str = Field(default="", description="取值展示名,下单时 choice 里要用这个原文")
is_placeholder: bool = Field(
default=False,
description="是否为「選択してください」这类占位项。占位项不是合法取值,"
"提交它等于没选,构造 choice 时必须跳过",
)
class ItemOption(BaseModel):
"""一个店铺自定义选项(下单必填项的来源)
与 SKU 规格不是一回事:规格走 `sku.variants[].variant_id`,选项走下单接口的
`choice` 字段。店铺用它承载「名入れ文字」「配送方式确认」「レビュー依頼」等,
`is_required=true` 的选项不给值时站点会直接拒绝加购。
"""
option_id: int | None = Field(default=None, description="站点 options[].id 原值")
name: str = Field(default="", description="选项名,下单时 choice 的「名」部分要用这个原文")
type: str = Field(
default="",
description="站点原值:select(下拉,看 values)/ text(自由文本,values 为空)",
)
is_required: bool = Field(default=False, description="是否必填;必填项不给值时站点拒绝加购")
values: list[ItemOptionValue] = Field(
default_factory=list, description="候选取值;type=text 时为空"
)
selectable_value_count: int = Field(
default=0,
description="剔除占位项后真正可提交的候选数。为 0 且 is_required=true 时"
"无法自动选值,必须由调用方在 choice 里显式给出",
)
class ShippingInfo(BaseModel):
"""配送与运费信息"""
@@ -365,7 +402,8 @@ class ItemDetailData(BaseModel):
加购(构造 cart 请求)不在本服务范围内——加购需要已登录的乐天账号会话,
归 trading 服务(app.trading)。本响应只描述「商品状态」:能不能买
(purchase_condition / is_sold_out)、规格(sku.variants)、起订单位等。
(purchase_condition / is_sold_out)、规格(sku.variants)、起订单位
店铺自定义选项(options)等,即「下单前需要先决定哪些参数」。
"""
source: str = Field(default="ichiba", description="数据来源站点:ichiba / books / brandavenue / biccamera")
@@ -390,6 +428,21 @@ class ItemDetailData(BaseModel):
breadcrumbs: list[Breadcrumb] = Field(default_factory=list, description="分类面包屑")
shipping: ShippingInfo = Field(default_factory=ShippingInfo, description="该商品的配送与运费信息")
sku: SkuInfo = Field(default_factory=SkuInfo, description="SKU 信息")
options: list[ItemOption] = Field(
default_factory=list,
description="店铺自定义选项(站点 purchase.information.options)。与 SKU 规格不同:"
"规格选 sku.variants[].variant_id,选项走下单接口的 choice 字段。"
"空列表表示该商品页没有选项;子站(books / brandavenue / biccamera)暂不解析",
)
has_required_options: bool = Field(
default=False,
description="是否存在必填选项。为 true 时下单必须给 choice,否则站点拒绝加购",
)
unfillable_required_options: list[str] = Field(
default_factory=list,
description="必填但无法自动选值的选项名(自由文本项,或候选值只有占位项)。"
"非空时**必须**由调用方在下单 intent.choice 里显式给出这些项的取值",
)
class HealthData(BaseModel):
+42 -1
View File
@@ -11,9 +11,13 @@ from __future__ import annotations
from typing import Any
from app.shared.errors import ScrapeParseError
from app.shared.purchase_contract import ItemOption as SharedItemOption
from app.shared.purchase_contract import auto_choice_for, parse_options
from app.scraping.models.scrape import (
Breadcrumb,
ItemDetailData,
ItemOption,
ItemOptionValue,
ReviewSummary,
ShippingInfo,
ShopSummary,
@@ -31,6 +35,33 @@ _PURCHASABLE_CONDITION = "enabled"
# 库存类型 → 加购表单里的 inventory_flag(常量与基础字段构造在 app.shared.purchase_contract)
def _to_item_options(options: list[SharedItemOption]) -> list[ItemOption]:
"""把共用契约的 ItemOption dataclass 搬成对外的 pydantic 模型
解析与占位项判定都在 `app.shared.purchase_contract.parse_options`——trading
下单时用同一份逻辑挑 choice 取值,两侧对「哪个取值是合法的」必须完全一致,
否则本接口告诉上游能选的值、下单时却填了别的。本函数只做搬运,不加判断。
"""
return [
ItemOption(
option_id=option.option_id,
name=option.name,
type=option.type,
is_required=option.is_required,
values=[
ItemOptionValue(
value_id=value.value_id,
name=value.name,
is_placeholder=value.is_placeholder,
)
for value in option.values
],
selectable_value_count=len(option.selectable_values),
)
for option in options
]
def _parse_attributes(raw: Any) -> list[SkuAttribute]:
return [
SkuAttribute(title=as_str(attr.get("title")), value=as_str(attr.get("value")))
@@ -115,6 +146,13 @@ def parse_item_detail(
sell_type = _pick_sell_type(as_dict(purchase.get("sellType")))
purchase_condition = as_str(sell_type.get("purchaseCondition"))
purchase_information = as_dict(purchase.get("information"))
# 只解析一次,两个派生结果都从这份结果来
shared_options = parse_options(purchase_information)
# 无法自动选值的必填项:与 trading 自动填 choice 时的判定同源,上游据此知道
# 「哪些项必须自己给值」,而不是等下单时才被站点拒绝
_, unfillable_required = auto_choice_for(shared_options)
raw_sku = as_dict(purchase.get("sku"))
variants = _parse_variants(raw_sku.get("variants"))
sku = SkuInfo(
@@ -188,7 +226,7 @@ def parse_item_detail(
purchase_condition=purchase_condition,
# purchaseCondition 是站点判定能否下单的直接依据;缺失时不臆断为售罄
is_sold_out=bool(purchase_condition) and purchase_condition != _PURCHASABLE_CONDITION,
purchase_unit=as_int(as_dict(purchase.get("information")).get("unit")),
purchase_unit=as_int(purchase_information.get("unit")),
images=images,
shop=shop,
review=review,
@@ -196,4 +234,7 @@ def parse_item_detail(
breadcrumbs=breadcrumbs,
shipping=shipping,
sku=sku,
options=_to_item_options(shared_options),
has_required_options=any(option.is_required for option in shared_options),
unfillable_required_options=unfillable_required,
)
+26 -21
View File
@@ -37,7 +37,7 @@ from app.scraping.utils.rakuma_urls import (
split_item_url,
split_shop_url,
)
from app.shared.telemetry import snapshot
from app.shared.telemetry import record_parse_failure
logger = logging.getLogger(__name__)
tracer = trace.get_tracer(__name__)
@@ -77,10 +77,11 @@ class RakumaClient:
url, len(result.items), result.total_count,
)
return result
except Exception:
span.record_exception()
span.set_attribute("parse.fail_reason", "parse_error")
snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
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:
@@ -110,10 +111,11 @@ class RakumaClient:
data.category_id, data.name, len(data.children), data.total_count,
)
return data
except Exception:
span.record_exception()
span.set_attribute("parse.fail_reason", "parse_error")
snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
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:
@@ -139,10 +141,11 @@ class RakumaClient:
url, detail.item_name[:40], detail.price, detail.is_sold_out,
)
return detail
except Exception:
span.record_exception()
span.set_attribute("parse.fail_reason", "parse_error")
snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
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:
@@ -177,10 +180,11 @@ class RakumaClient:
shop_id, detail.shop_name, detail.item_count, detail.review_count,
)
return detail
except Exception:
span.record_exception()
span.set_attribute("parse.fail_reason", "parse_error")
snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
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:
@@ -205,10 +209,11 @@ class RakumaClient:
shop_id, len(result.items), result.total_count,
)
return result
except Exception:
span.record_exception()
span.set_attribute("parse.fail_reason", "parse_error")
snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
except Exception as exc:
record_parse_failure(
span, exc, html=html,
max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
)
raise
@staticmethod
+26 -5
View File
@@ -29,10 +29,14 @@ from app.shared.errors import (
UpstreamBlockedError,
UpstreamRequestError,
)
from app.shared.telemetry import snapshot
from app.shared.telemetry import add_event, record_error, snapshot
logger = logging.getLogger(__name__)
# 逐次尝试 event 里错误串的截断长度:httpx 的异常消息可能很长(带完整 URL 与
# 底层 socket 错误),而这里只是用来区分「这次是怎么失败的」,前半句就够
_ERROR_MAX_CHARS = 200
class RakumaSession:
"""ラクマ 站点抓取会话,管理并发限流与失败重试"""
@@ -108,7 +112,7 @@ class RakumaSession:
)
except TimeoutError as exc:
err = ResourceBusyError()
span.record_exception(err)
record_error(span, err)
span.set_attribute("scrape.fail_reason", "ResourceBusyError")
raise err from exc
@@ -119,6 +123,13 @@ class RakumaSession:
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,
@@ -127,7 +138,7 @@ class RakumaSession:
if response.status_code == 404:
err = ItemNotFoundError(f"Page not found: {url}")
span.record_exception(err)
record_error(span, err)
span.set_attribute("scrape.fail_reason", "ItemNotFoundError")
raise err
@@ -145,6 +156,12 @@ class RakumaSession:
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,
@@ -154,9 +171,13 @@ class RakumaSession:
err = UpstreamRequestError(f"Upstream request failed: {last_error}")
else:
err = UpstreamBlockedError(f"Failed to fetch {url}: {last_error}")
span.record_exception(err)
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)
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()
+26 -20
View File
@@ -45,7 +45,7 @@ from app.scraping.utils.urls import (
split_item_url,
split_shop_url,
)
from app.shared.telemetry import snapshot
from app.shared.telemetry import record_parse_failure
logger = logging.getLogger(__name__)
tracer = trace.get_tracer(__name__)
@@ -87,10 +87,11 @@ class RakutenClient:
url, len(result.items), result.ad_count, result.total_count,
)
return result
except Exception:
span.record_exception()
span.set_attribute("parse.fail_reason", "parse_error")
snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
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:
@@ -113,10 +114,11 @@ class RakutenClient:
result.genre_id or "root", result.name, len(result.children),
)
return result
except Exception:
span.record_exception()
span.set_attribute("parse.fail_reason", "parse_error")
snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
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:
@@ -142,10 +144,11 @@ class RakutenClient:
result.shop_code, result.shop_id, result.shop_name, result.review_count,
)
return result
except Exception:
span.record_exception()
span.set_attribute("parse.fail_reason", "parse_error")
snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
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:
@@ -170,9 +173,11 @@ class RakutenClient:
span.set_attribute("parse.items", len(result.items))
span.set_attribute("parse.total", result.total_count)
return result
except Exception:
span.record_exception()
span.set_attribute("parse.fail_reason", "parse_error")
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:
@@ -226,8 +231,9 @@ class RakutenClient:
url, detail.source, detail.item_name[:40], detail.price, detail.sku.variant_count,
)
return detail
except Exception:
span.record_exception()
span.set_attribute("parse.fail_reason", "parse_error")
snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
except Exception as exc:
record_parse_failure(
span, exc, html=html,
max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
)
raise
+47 -6
View File
@@ -38,10 +38,14 @@ from app.shared.errors import (
)
from app.scraping.parsers.state import PageValidator, require_state_marker
from app.scraping.services.browser_fallback import BrowserFallback
from app.shared.telemetry import snapshot
from app.shared.telemetry import add_event, record_error, snapshot
logger = logging.getLogger(__name__)
# 逐次尝试 event 里错误串的截断长度:httpx 异常消息可能很长(带完整 URL 与底层
# socket 错误),这里只用来区分「这次是怎么失败的」,前半句够了
_ERROR_MAX_CHARS = 200
@dataclass(slots=True)
class FetchedPage:
@@ -191,7 +195,7 @@ class SiteSession:
)
except TimeoutError as exc:
err = ResourceBusyError()
span.record_exception(err)
record_error(span, err)
span.set_attribute("scrape.fail_reason", "ResourceBusyError")
raise err from exc
@@ -204,6 +208,14 @@ class SiteSession:
response = await profile.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 profile=%s attempt=%s/%s err=%s",
url, profile.name, attempt, max_attempts, last_error,
@@ -216,7 +228,7 @@ class SiteSession:
if response.status_code == 404:
err = ItemNotFoundError(f"Page not found: {url}")
span.record_exception(err)
record_error(span, err)
span.set_attribute("scrape.fail_reason", "ItemNotFoundError")
raise err
@@ -232,11 +244,21 @@ class SiteSession:
if reason is None:
return FetchedPage(html=text, url=final_url)
last_error = self._refine_failure(reason, text)
if last_error.startswith("challenge page detected"):
challenged = last_error.startswith("challenge page detected")
if challenged:
span.set_attribute("scrape.challenge_detected", True)
outcome = "challenge" if challenged else "validate_failed"
else:
last_error = self._describe_error_status(response.status_code)
outcome = "bad_status"
last_text = text
add_event(span, "scrape.attempt", {
"attempt": attempt,
"outcome": outcome,
"status_code": response.status_code,
"html_bytes": len(text),
"error": last_error[:_ERROR_MAX_CHARS],
})
logger.warning(
"抓取结果异常:url=%s profile=%s attempt=%s/%s %s",
url, profile.name, attempt, max_attempts, last_error,
@@ -250,20 +272,39 @@ class SiteSession:
if attempt == 1:
await self._rewarm_on_home(profile)
span.set_attribute("scrape.rewarmed_on_home", True)
add_event(span, "scrape.escalate", {
"attempt": attempt, "to": "rewarm_on_home",
})
else:
page = await self._escalate_to_browser(profile, url, validate)
if page is not None:
span.set_attribute("scrape.fell_back_to_browser", True)
span.set_attribute("scrape.final_url", page.url)
add_event(span, "scrape.escalate", {
"attempt": attempt, "to": "browser", "outcome": "recovered",
})
return page
# 兜底没救回来(浏览器不可用,或取回的页面仍不合格)。不记
# 这条的话链路里只看得到「最终失败」,看不出浏览器这一级到底
# 试过没有——而「没装 playwright」和「装了也被挡」要分开查。
add_event(span, "scrape.escalate", {
"attempt": attempt,
"to": "browser",
"outcome": "failed",
"reason": self._browser.unavailable_reason,
})
if self._is_server_error(last_error):
err = UpstreamRequestError(f"Upstream request failed: {last_error}")
else:
err = UpstreamBlockedError(f"Blocked while fetching {url}: {last_error}")
span.record_exception(err)
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)
snapshot(
span, "scrape.failed_html", last_text,
self._settings.otel_snapshot_max_bytes,
extra={"scrape.url": url, "scrape.profile": profile.name},
)
raise err
finally:
self._semaphore.release()
+44 -3
View File
@@ -16,10 +16,12 @@ from typing import Any, Generic, TypeVar
from fastapi import Depends, FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from opentelemetry import trace
from pydantic import BaseModel, Field, ValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
from app.shared.errors import AppError, AuthenticationError
from app.shared.telemetry import record_envelope, record_error
logger = logging.getLogger(__name__)
@@ -98,16 +100,43 @@ def jsonable_errors(errors: list[dict]) -> list[dict]:
return [{key: value for key, value in error.items() if key != "ctx"} for error in errors]
def _trace_failure(
exc: BaseException | None, *, err_code: int, msg: str, status_code: int
) -> None:
"""把一次失败响应记到当前 server span 上(FastAPI 自动 instrumentation 建的那个)
这些处理器是所有对外失败的**唯一出口**,也是链路上唯一还知道「异常长什么样」
的地方:它们把异常吃掉换成 200/4xx 的信封响应,异常不再向上冒,自动
instrumentation 只看得到一个 HTTP 状态码。尤其 AppError 默认 status_code=400、
信封里 `success=false`,在 trace 里跟正常返回几乎分不出来——不在这里记一次,
上游报「调用失败了」时链路里根本找不到对应的错误。
span 没在录(otel 关闭、或 /health 这类被 excluded_urls 排除的路径)时
`set_attributes` / `record_error` 都作用在 NonRecordingSpan 上,是 noop,
不必额外判断。
"""
span = trace.get_current_span()
if exc is not None:
record_error(span, exc)
record_envelope(span, success=False, err_code=err_code, msg=msg, status_code=status_code)
def register_exception_handlers(app: FastAPI) -> None:
"""给应用挂上全套异常处理器
两个入口都调用它,保证抓取失败与下单失败返回的错误结构完全一致,
上游只需要按 code 分支,不必区分是哪个服务回的。
每个处理器除了构造响应,还把这次失败记到当前 server span 上(见
`_trace_failure`)——处理器是失败的唯一出口,不记就等于链路里没有这次失败。
"""
@app.exception_handler(AppError)
async def app_error_handler(_: Request, exc: AppError) -> JSONResponse:
"""业务异常处理器:返回结构化的错误响应"""
_trace_failure(
exc, err_code=exc.err_code, msg=exc.message, status_code=exc.status_code
)
return JSONResponse(
status_code=exc.status_code,
content=ApiResponse[None](
@@ -123,11 +152,15 @@ def register_exception_handlers(app: FastAPI) -> None:
async def validation_error_handler(_: Request, exc: RequestValidationError) -> JSONResponse:
"""请求参数校验异常处理器"""
errors = exc.errors()
msg = _format_validation_msg(errors)
# 校验失败不记异常本身(pydantic 的 ValidationError 栈很长且没有诊断价值),
# 只留错误码与整理后的字段消息——排查要看的是「哪个字段不合法」
_trace_failure(None, err_code=1002, msg=msg, status_code=422)
return JSONResponse(
status_code=422,
content=ApiResponse[object](
success=False,
msg=_format_validation_msg(errors),
msg=msg,
data=jsonable_errors(errors),
code=1002,
).model_dump(),
@@ -137,11 +170,13 @@ def register_exception_handlers(app: FastAPI) -> None:
async def pydantic_validation_error_handler(_: Request, exc: ValidationError) -> JSONResponse:
"""Pydantic 模型校验异常处理器"""
errors = exc.errors()
msg = _format_validation_msg(errors)
_trace_failure(None, err_code=1002, msg=msg, status_code=422)
return JSONResponse(
status_code=422,
content=ApiResponse[object](
success=False,
msg=_format_validation_msg(errors),
msg=msg,
data=jsonable_errors(errors),
code=1002,
).model_dump(),
@@ -152,11 +187,13 @@ def register_exception_handlers(app: FastAPI) -> None:
"""HTTP 异常处理器(404、500 等)"""
status_code = int(getattr(exc, "status_code", 500) or 500)
err_code = 1404 if status_code == 404 else 1500
detail = str(getattr(exc, "detail", "HTTP error"))
_trace_failure(None, err_code=err_code, msg=detail, status_code=status_code)
return JSONResponse(
status_code=status_code,
content=ApiResponse[None](
success=False,
msg=str(getattr(exc, "detail", "HTTP error")),
msg=detail,
data=None,
code=err_code,
).model_dump(),
@@ -167,6 +204,10 @@ def register_exception_handlers(app: FastAPI) -> None:
async def unhandled_exception_handler(_: Request, exc: Exception) -> JSONResponse:
"""兜底异常处理器:捕获所有未处理的异常"""
logger.exception("未处理异常:%s", exc)
# 这里最需要 record_error:对外只回一句无信息量的 "Internal server error",
# 真正的异常类型与栈只在本进程日志里。记到 span 上,链路里就能直接看到
# 是什么炸了,不必再去捞日志按时间对。
_trace_failure(exc, err_code=1500, msg="Internal server error", status_code=500)
return JSONResponse(
status_code=500,
content=ApiResponse[None](
+174 -2
View File
@@ -9,11 +9,13 @@
- `inventory_flag_for(inventory_type)` — 多规格判定
- `basket_domain_of(sell_type)` — 抽 basketDomain + 反转义 `\\u002F`
- `base_form_fields(shop_id, item_id, inventory_flag)` — 加购表单四件套
- `parse_options(information)` — 店铺自定义选项(`purchase.information.options[]`)
的结构化解析,含占位值识别;scraping 用它对外暴露选项,trading 用它自动填 choice
- `auto_choice_for(options)` / `format_choice(...)` — choice 表单值的构造
**不**放这里:
- `PurchaseInfo` pydantic 模型 — 是 scraping 的对外契约,4 个子站共用,留在 scraping/models
- variant_id / choice 自动选择策略 — trading 独有(scraping 把决策权留给上游)
- options 结构化解析 — scraping 独有(trading 用原始 dict)
- variant_id 自动选择策略 — trading 独有(scraping 把决策权留给上游)
- 子站(books/biccamera/brandavenue)的加购契约 — 各自独立,不走这里
依赖约束:仅 `typing.Any`(标准库),不 import scraping / trading / gateway,
@@ -21,6 +23,7 @@
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
# 站点前端构造加购表单时固定带的事件标识(普通购买 normalPurchase)
@@ -76,3 +79,172 @@ def base_form_fields(
"inventory_flag": inventory_flag,
"__event": NORMAL_PURCHASE_EVENT,
}
# ---- 店铺自定义选项(purchase.information.options[])----
#
# 「規格(SKU)」与「選項(option)」是两套完全不同的东西,加购表单里也走不同字段:
# - 規格:sku.variants[].variantId → 表单 variant_id
# - 選項:店铺自己配的下拉/文本框(名入れ文字、配送方式确认、レビュー依頼 等)
# → 表单 choice,格式「名:值」,多项用「,」连接
#
# 2026-08-28 用 .probe/ 下 3 份真实商品页样本 + tests/fixtures/item_with_options_state.json
# 核对过结构,得到两条**结构性**判据(比按日文文案猜稳):
#
# 1. `values[].id == 0` 是「請選擇」占位项,不是可提交的取值。三份真实样本里所有
# 必填 select 的 values[0] 都是 id=0 的「選択してください」,真实可选值从 id=200
# 起编号。此前 trading 侧自动填 choice 时取的就是 values[0],等于把「選択して
# ください」当答案提交上去(.probe/checkout/probe-v2.txt 里两条 "成功" 记录的
# choice 值就是它)——站点当时收下了,但那是店铺没做校验,不代表填对了。
# 2. `type == "text"` 的选项没有 values[](自由文本,如「【お名前】4文字まで」),
# **无法**自动填。必填且为 text 时只能交由调用方给值,不猜。
#
# 占位项文案本身(「選択してください」)作为辅助判据一起保留:id 编号规则是从 4 份
# 样本归纳的,万一某店铺不按 200 起编号,文案还能兜一层。两条判据命中任一即占位。
_PLACEHOLDER_VALUE_ID: int = 0
_PLACEHOLDER_VALUE_NAMES: frozenset[str] = frozenset({"選択してください", "選択して下さい"})
# 选项类型:下拉(有候选值)与自由文本(无候选值)
OPTION_TYPE_SELECT: str = "select"
OPTION_TYPE_TEXT: str = "text"
# choice 表单值的分隔符:项间用「,」,名与值之间用「:」
_CHOICE_PAIR_SEPARATOR: str = ","
_CHOICE_NAME_VALUE_SEPARATOR: str = ":"
@dataclass(slots=True)
class ItemOptionValue:
"""选项的一个候选取值
`is_placeholder=True` 表示这是「選択してください」这类占位项,提交它等于没选。
"""
value_id: int | None
name: str
is_placeholder: bool = False
@dataclass(slots=True)
class ItemOption:
"""一个店铺自定义选项
`is_free_text` 为 True 时 values 必然为空(站点 type="text"),取值只能由
调用方给;`selectable_values` 是剔掉占位项后真正可提交的候选。
"""
option_id: int | None
name: str
type: str
is_required: bool
values: list[ItemOptionValue] = field(default_factory=list)
@property
def is_free_text(self) -> bool:
return self.type == OPTION_TYPE_TEXT
@property
def selectable_values(self) -> list[ItemOptionValue]:
return [value for value in self.values if not value.is_placeholder]
@property
def can_auto_fill(self) -> bool:
"""能否在不问调用方的情况下自动给出一个合法取值"""
return bool(self.selectable_values)
def _coerce_option_id(raw: Any) -> int | None:
"""选项/取值的 id 收敛为 int;给不出数字时返回 None(不编造 0——0 有含义)"""
if isinstance(raw, bool) or not isinstance(raw, (int, float, str)):
return None
try:
return int(raw)
except (TypeError, ValueError):
return None
def _parse_option_value(raw: Any) -> ItemOptionValue | None:
"""解析单个候选取值;不是 dict 或没有名字的直接丢掉"""
if not isinstance(raw, dict):
return None
name = raw.get("name")
if not isinstance(name, str) or not name:
return None
value_id = _coerce_option_id(raw.get("id"))
is_placeholder = value_id == _PLACEHOLDER_VALUE_ID or name.strip() in _PLACEHOLDER_VALUE_NAMES
return ItemOptionValue(value_id=value_id, name=name, is_placeholder=is_placeholder)
def parse_options(information: Any) -> list[ItemOption]:
"""从 `purchase.information` 解析店铺自定义选项列表
`information` 传站点原始 dict(不是 options 数组本身);结构不符合预期时返回
空列表——**没有选项与解析不出选项在站点数据上长得一样**(缺 options 键),
这一层区分不出来,也不该假装能区分。
"""
if not isinstance(information, dict):
return []
raw_options = information.get("options")
if not isinstance(raw_options, list):
return []
options: list[ItemOption] = []
for raw in raw_options:
if not isinstance(raw, dict):
continue
name = raw.get("name")
if not isinstance(name, str) or not name:
continue
raw_type = raw.get("type")
option_type = raw_type if isinstance(raw_type, str) else ""
values = [
value
for value in (_parse_option_value(item) for item in raw.get("values") or [])
if value is not None
]
options.append(
ItemOption(
option_id=_coerce_option_id(raw.get("id")),
name=name,
type=option_type,
is_required=bool(raw.get("isRequired")),
values=values,
)
)
return options
def format_choice(pairs: Any) -> str:
"""把「名:值」序列拼成站点 choice 表单值
接受 str(原样透传,调用方自己拼好的)或可迭代的字符串序列。
"""
if isinstance(pairs, str):
return pairs
return _CHOICE_PAIR_SEPARATOR.join(str(pair) for pair in pairs)
def auto_choice_for(options: list[ItemOption]) -> tuple[str, list[str]]:
"""为必填选项自动挑取值,返回 (choice 表单值, 无法自动填的必填项名)
策略:每个必填选项取**第一个非占位**候选值,拼成「名:值」。自由文本必填项
(type="text",没有候选值)与候选值全是占位项的选项都自动填不了,其名字进
第二个返回值——调用方据此决定报错还是要求上游显式给 choice,**不拿占位值
凑数**(那正是本次修复的问题)。
只处理必填项:非必填项站点不强制,替上游擅自选(比如「置き配を希望する」)
等于替人做了业务决定。
"""
pairs: list[str] = []
unfillable: list[str] = []
for option in options:
if not option.is_required:
continue
selectable = option.selectable_values
if not selectable:
unfillable.append(option.name)
continue
pairs.append(
f"{option.name}{_CHOICE_NAME_VALUE_SEPARATOR}{selectable[0].name}"
)
return format_choice(pairs), unfillable
+144 -8
View File
@@ -34,7 +34,10 @@ from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.instrumentation.utils import suppress_instrumentation
from opentelemetry.instrumentation.utils import (
is_instrumentation_enabled,
suppress_instrumentation,
)
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
@@ -52,6 +55,10 @@ R = TypeVar("R")
logger = logging.getLogger(__name__)
# 错误信息类属性的截断长度。站点错误页抽出来的 msg 可能很长(_extract_error_message
# 拼两条提示),而属性值过长会把 OTLP 请求撑大;排查看的是前半句,够了。
_MSG_MAX_CHARS = 512
# 全局 provider 引用,用于 instrument_app / shutdown 时判断当前是否已初始化。
# 显式持有比依赖 trace.get_tracer_provider() 的类型判断更稳——后者在测试场景
# 下可能被其它用例改动全局状态。
@@ -147,11 +154,21 @@ def shutdown_telemetry() -> None:
_provider = None
def snapshot(span: Span, name: str, html: str | None, max_bytes: int) -> None:
def snapshot(
span: Span,
name: str,
html: str | None,
max_bytes: int,
*,
extra: Mapping[str, AttributeValue | None] | None = None,
) -> None:
"""把 HTML 作为 span event 上报,超 max_bytes 截断并标注。
用于解析失败时复现页面:span 自身只放结构化指标(items 数、source 等),
完整 HTML 体量大、含商品/价格内容,仅在失败分支通过 event 携带。
`extra` 用来带上「这份 HTML 是哪来的」——落地 URL、页面标题、证据文件路径
之类。光有一坨 HTML 还得自己回头对是哪一步的产物,附在同一条 event 上省事。
"""
if html is None or not html:
return
@@ -164,6 +181,9 @@ def snapshot(span: Span, name: str, html: str | None, max_bytes: int) -> None:
}
if truncated:
attributes["snapshot.truncated"] = True
for key, value in (extra or {}).items():
if value is not None:
attributes[key] = value
span.add_event(name, attributes=attributes)
@@ -174,25 +194,141 @@ def suppressed() -> Iterator[None]:
给「空转的长轮询」用:worker 每 30 秒问一次网关有没有活干,绝大多数时候
返回空。这些请求各自成为一条孤立 trace,量大且没有信息量——把观测后台刷满
的正是它们。领到任务后的每一次网关调用都在任务根 span 底下,不受影响。
**只挡自动 instrumentation**:OTel 那个上下文标记是给 instrumentation 库看的,
手工 `start_as_current_span` 不看它,照样会建 span。所以在这个上下文里手工埋点
要走 `span_unless_suppressed`,否则空转长轮询会从另一个口子把孤立 trace 放回来。
"""
with suppress_instrumentation():
yield
@contextmanager
def span_unless_suppressed(
tracer: trace.Tracer, name: str, *, kind: SpanKind = SpanKind.INTERNAL
) -> Iterator[Span]:
"""同 `start_as_current_span`,但在 `suppressed()` 里退化成 noop span。
手工埋点与 `suppressed()` 的配套件。`suppress_instrumentation` 只被
instrumentation 库尊重,手工建的 span 不受它影响——worker 的 `lease` /
`lease_query` 正是在 `suppressed()` 里调 `GatewayClient._request` 的,那里若
无条件建 span,空转的长轮询就会每 30 秒产出一条孤立 trace,等于绕开了
`suppressed()` 本来要解决的问题。
退化时给的是 `INVALID_SPAN`(NonRecordingSpan):`set_attributes` /
`record_error` / `record_envelope` 作用在它上面全是 noop,调用方不必分支。
"""
if not is_instrumentation_enabled():
yield trace.INVALID_SPAN
return
with tracer.start_as_current_span(name, kind=kind) as span:
yield span
def record_error(span: Span, exc: BaseException) -> None:
"""把异常记到 span 上并置 ERROR 状态。
单独抽出来是因为 AppError 带 `err_code`(对外错误码),排查时按码筛比按
异常类名筛更贴近上游看到的东西,值得单独落一个属性。
单独抽出来是因为 AppError 带的那几个字段(对外错误码 `err_code`、是否可重试
`retryable`、HTTP 状态码)正是排查时真正要看的东西:按码筛比按异常类名筛更
贴近上游看到的结果,而 `retryable` 直接决定这次失败该不该重来。异常消息也单独
落一个属性——`record_exception` 记的 event 在多数观测后台里要展开才看得到,
列表页按 `error.message` 筛不出来。
"""
span.record_exception(exc)
span.set_attribute("error.type", type(exc).__name__)
err_code = getattr(exc, "err_code", None)
if isinstance(err_code, int):
span.set_attribute("error.code", err_code)
retryable = getattr(exc, "retryable", None)
set_attributes(
span,
{
"error.type": type(exc).__name__,
"error.message": str(exc)[:_MSG_MAX_CHARS],
"error.code": err_code if isinstance(err_code := getattr(exc, "err_code", None), int) else None,
"error.retryable": retryable if isinstance(retryable, bool) else None,
"error.status_code": sc if isinstance(sc := getattr(exc, "status_code", None), int) else None,
},
)
span.set_status(Status(StatusCode.ERROR, f"{type(exc).__name__}: {exc}"))
def record_envelope(
span: Span,
*,
success: bool,
err_code: int | None = None,
msg: str | None = None,
status_code: int | None = None,
) -> None:
"""把 `ApiResponse` 信封的结果记到 span 上。
自动 instrumentation 只看 HTTP 层,而本项目的失败**在信封里**:一次
`success=false, code=6002` 的回报,HTTP 层跟成功的调用长得一模一样(很多还
是 200)。于是 trace 里只剩「调用发生过」,「这次到底成没成、错在哪个码」
全在 body 里,不显式记就永远看不到——这正是「只知道调用了、不知道异常怎么
来的」的直接原因。
出入两侧都用它:服务端异常处理器往 server span 上记(见 `shared.api`),
worker 解信封时往 client span 上记(见 `trading.worker.client`),同一套
`api.*` 属性名,一条 trace 里两侧的结论可以直接对上。
"""
set_attributes(
span,
{
"api.success": success,
"api.code": err_code,
"api.status_code": status_code,
"api.msg": msg[:_MSG_MAX_CHARS] if msg else None,
},
)
if not success:
span.set_status(Status(StatusCode.ERROR, msg or f"api.code={err_code}"))
def add_event(
span: Span, name: str, attributes: Mapping[str, AttributeValue | None]
) -> None:
"""记一条 span event,跳过 None 值属性。
「过程」不能用属性表达:同名属性后写覆盖先写,三次抓取尝试写完只剩最后一次
的状态码,中间那两次为什么失败、升级到哪一级全被盖掉了。每次尝试各记一条
event,链路里才看得出升级路径。
"""
span.add_event(
name,
attributes={key: value for key, value in attributes.items() if value is not None},
)
def record_parse_failure(
span: Span,
exc: BaseException,
*,
html: str | None = None,
max_bytes: int = 0,
url: str | None = None,
stage: str | None = None,
) -> None:
"""页面类操作失败的统一记法:错误详情 + 失败阶段 + 失败页面快照。
抓取侧与 ラクマ 侧一共十个接口的失败分支原本各写一遍同样三步,且都漏了最关键
的一件事——**失败在哪一步**。`html` 是否已拿到恰好就是判据:还是空说明页面根本
没取回来(通道 / 反爬 / 上游 5xx),非空说明取回了但解析不出(多半站点改版)。
两者的排查方向完全相反,所以作为属性直接落下来,不让人对着一坨 HTML 猜。
`stage` 可显式覆盖:像 `shop_items` 那种「转调另外两个接口」的编排方法,失败
既不在自己的 fetch 也不在自己的 parse,据 html 推断只会给出错的结论。
"""
record_error(span, exc)
set_attributes(
span,
{
"parse.stage": stage or ("fetch" if not html else "parse"),
# 保留原有属性名(观测后台的既有筛选条件),但值改成真实异常类名——
# 原先无论什么失败都写死 "parse_error",把反爬阻断也说成解析失败。
"parse.fail_reason": type(exc).__name__,
},
)
snapshot(span, "parse.failed_html", html, max_bytes, extra={"parse.url": url})
def traced(
name: str,
*,
+60 -19
View File
@@ -15,15 +15,25 @@ import logging
from typing import Any
import httpx
from opentelemetry import trace
from opentelemetry.trace import SpanKind
from app.shared.config import Settings
from app.shared.errors import AppError
from app.shared.proxy import httpx_client_options
from app.shared.task_state import OrderState, TaskStatus
from app.shared.telemetry import (
record_envelope,
record_error,
set_attributes,
span_unless_suppressed,
)
from app.trading.worker.models import LeaseTask, QueryTask
logger = logging.getLogger(__name__)
tracer = trace.get_tracer(__name__)
class GatewayClient:
"""网关 HTTP 客户端
@@ -49,27 +59,58 @@ class GatewayClient:
# ---- 基础封装 ----
async def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
"""发起请求并解信封。失败(success=False)抛 AppError"""
response = await self._client.request(method, path, **kwargs)
try:
body = response.json()
except ValueError as exc:
raise AppError(
message=f"网关响应不是合法 JSON:HTTP {response.status_code}",
code="GATEWAY_BAD_BODY",
err_code=3001,
retryable=True,
) from exc
"""发起请求并解信封。失败(success=False)抛 AppError
if not body.get("success"):
raise AppError(
message=body.get("msg", "网关返回失败"),
code="GATEWAY_ERROR",
err_code=int(body.get("code", 1500)),
retryable=False,
status_code=response.status_code,
整段套一个自己的 span,而不是依赖 httpx 自动 instrumentation 那个 CLIENT
span:**网关的失败在信封里,不在 HTTP 状态码上**。httpx 那个 span 在
`request()` 返回时就结束了,此时信封还没解——一次 `success=false, code=6002`
(租约无效)的调用在它看来是完成的 200 请求,链路里跟成功毫无区别。
本 span 活到解信封之后,所以能把「这次调用的结论」记下来。
"""
with span_unless_suppressed(
tracer,
f"gateway.{path.strip('/').replace('/', '.')}",
kind=SpanKind.CLIENT,
) as span:
set_attributes(span, {"gateway.method": method, "gateway.path": path})
response = await self._client.request(method, path, **kwargs)
try:
body = response.json()
except ValueError as exc:
err = AppError(
message=f"网关响应不是合法 JSON:HTTP {response.status_code}",
code="GATEWAY_BAD_BODY",
err_code=3001,
retryable=True,
)
record_error(span, err)
span.set_attribute("gateway.status_code", response.status_code)
raise err from exc
if not body.get("success"):
err_code = int(body.get("code", 1500))
msg = body.get("msg", "网关返回失败")
# 记成信封结果而不是只抛异常:上报被网关拒时,链路里能直接按
# api.code 筛出是哪一类拒绝(6002 租约无效 / 6003 状态不允许…),
# 不必回头翻 worker 日志。
record_envelope(
span,
success=False,
err_code=err_code,
msg=msg,
status_code=response.status_code,
)
raise AppError(
message=msg,
code="GATEWAY_ERROR",
err_code=err_code,
retryable=False,
status_code=response.status_code,
)
record_envelope(
span, success=True, status_code=response.status_code
)
return body
return body
# ---- 接口 ----
+28 -18
View File
@@ -109,9 +109,12 @@ from app.shared.errors import (
from app.shared.purchase_contract import (
INVENTORY_FLAG_DEFAULT,
INVENTORY_FLAG_MULTIPLE,
auto_choice_for,
base_form_fields,
basket_domain_of,
format_choice,
inventory_flag_for,
parse_options,
)
from app.shared.proxy import playwright_launch_proxy
from app.shared.task_state import OrderState
@@ -1102,10 +1105,18 @@ class SiteInteractor:
raise CartOperationError(
"多规格商品未选 variant,且 sku.variants 全部售罄或为空"
)
# 必填选项要求填了 choice
if fields["has_required_options"] and not fields["form_fields"].get(fields["options_field"]):
# 必填选项要求填了 choice。调用方没给 choice 时,只要存在「自动填不了」
# 的必填项(自由文本项,或候选值只剩占位项)就当场失败并点名是哪几项——
# 这些项非人工给值不可能成功,继续 POST 只会拿站点的
# 「未選択の項目からどれか1つ選んでください。」错误页,排查成本更高
if fields["has_required_options"] and not fields["form_fields"].get(
fields["options_field"]
):
unfillable = fields["unfillable_required_options"]
detail = f"{unfillable}" if unfillable else ""
raise CartOperationError(
"商品有必填选项但未提供 choice,且选项无候选值"
"商品有必填选项但未提供 choice,且这些必填项无法自动选值"
f"(需在 intent.choice 里按「选项名:取值名」显式给出){detail}"
)
payload = dict(fields["form_fields"])
@@ -2421,22 +2432,18 @@ def _extract_purchase_fields(state: dict, *, intent_override: dict | None) -> di
elif inventory_flag == INVENTORY_FLAG_DEFAULT and item.get("variantId"):
form_fields["variant_id"] = str(item.get("variantId"))
# 必填选项:调用方覆盖 > 自动填第一个候选值
options = information.get("options") or []
required_options = [o for o in options if o.get("isRequired")]
has_required = bool(required_options)
# 必填选项:调用方覆盖 > 自动填第一个**非占位**候选值
# 解析与占位项判定走 app.shared.purchase_contract(与 scraping 的
# /api/item_detail 同源),旧实现直接取 values[0],而必填 select 的 values[0]
# 恰恰是「選択してください」占位项,等于把「请选择」当答案提交上去。
options = parse_options(information)
has_required = any(option.is_required for option in options)
auto_choice, unfillable_required = auto_choice_for(options)
if intent_override.get("choice"):
# 调用方给的可能是 list 或 str
c = intent_override["choice"]
form_fields["choice"] = ",".join(c) if isinstance(c, list) else str(c)
elif has_required:
pairs: list[str] = []
for opt in required_options:
values = opt.get("values") or []
if values:
pairs.append(f"{opt.get('name')}:{values[0].get('name')}")
if pairs:
form_fields["choice"] = ",".join(pairs)
# 调用方给的可能是 list 或 str,两种都交给共用的格式化
form_fields["choice"] = format_choice(intent_override["choice"])
elif auto_choice:
form_fields["choice"] = auto_choice
return {
"basket_domain": basket_domain,
@@ -2446,6 +2453,9 @@ def _extract_purchase_fields(state: dict, *, intent_override: dict | None) -> di
"options_field": "choice" if options else "",
"options": options,
"has_required_options": has_required,
# 必填但自动填不了的选项名(自由文本项,或候选值只有占位项)。调用方没给
# choice 时这就是「非人工介入不可能成功」的直接依据,见 _add_to_cart_with_fields
"unfillable_required_options": unfillable_required,
"inventory_flag": inventory_flag,
"purchase_condition": sell_type.get("purchaseCondition"),
"min_price": sell_type.get("minPrice"),
+2 -2
View File
@@ -114,8 +114,8 @@ CREATE TABLE workers (
"intent": { // gateway 原样透传,结构由 trading 侧定义
"item_url": "https://item.rakuten.co.jp/shop/code/",
"quantity": 1,
"variant_id": "...", // 多规格商品必填
"options": {},
"variant_id": "...", // 多规格商品必填,取自 /api/item_detail 的 sku.variants[]
"choice": ["名入れ:希望する"], // 店铺自定义必填选项,格式「选项名:取值名」,取自 /api/item_detail 的 options[]
"max_total_yen": 30000 // 可选,覆盖本次的金额上限
},
"callback_url": "https://upstream.example.com/hooks/rakuten-order" // 可选,终结类事件通知,见 §4.8
+245
View File
@@ -0,0 +1,245 @@
"""必填选项 choice 取值的真账号验证:占位项修复是否真的被站点接受
背景2026-08-28 发现 trading 自动填 choice 时取 `values[0]`而必填 select
values[0] 恒为 id=0 選択してください占位项等于把请选择当答案提交
修复后取第一个非占位候选 app/shared/purchase_contract.py::auto_choice_for
离线测试只能证明我们填的值变了证明不了站点接受这个值后者必须真账号
实测本探针就为这一件事
**刻意走生产代码路径**SiteInteractor.add_to_cart_payload不重写字段构造逻辑
上一版探针 scripts/probe_purchase_block_v2.py 自己抄了一遍 form 构造结果那份
抄写与生产代码一起用了 values[0]两边同错就测不出问题实际提交的 payload
生产代码自己的 INFO 日志里抓加购请求basket=... payload=...确保记录的
就是真正发出去的东西
安全边界
- 只做清空购物车 加购 校验 清空购物车**绝不进入结算/支付**
- 加购完立即清空账号购物车恢复原状加购本身可逆不产生订单不扣款
- 不触碰 enter_checkout / submit_order / pay
用法
.venv/Scripts/python.exe scripts/probe_option_choice.py
"""
from __future__ import annotations
import asyncio
import json
import logging
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from app.shared.config import get_settings # noqa: E402
from app.shared.errors import AppError # noqa: E402
from app.shared.purchase_contract import auto_choice_for, parse_options # noqa: E402
from app.trading.services.auth_session import AuthSession # noqa: E402
from app.trading.worker.site_interact import ( # noqa: E402
SiteInteractor,
_parse_initial_state,
)
PROBE_DIR = Path(__file__).resolve().parent.parent / ".probe" / "options_choice"
PROBE_DIR.mkdir(parents=True, exist_ok=True)
# 目标商品:从 .probe/checkout/ 的历史样本里挑「有必填选项且占位项在 values[0]」的两个。
# 这两个正是旧实现填「選択してください」也被站点收下的商品——它们不能证明旧值是对的
# (店铺没做校验),但可以验证新值同样被接受,且是站点自己给的唯一非占位候选。
TARGETS = (
"https://item.rakuten.co.jp/waabbit/088-4p",
"https://item.rakuten.co.jp/aoimorinomise/121",
)
class _PayloadCapture(logging.Handler):
"""截获生产代码 `加购请求:basket=%s payload=%s` 这条 INFO 日志
目的是记录**真正发出去的 payload**而不是探针自己再算一遍
"""
def __init__(self) -> None:
super().__init__(level=logging.INFO)
self.payloads: list[dict] = []
def emit(self, record: logging.LogRecord) -> None:
if record.msg == "加购请求:basket=%s payload=%s":
basket, payload = record.args
self.payloads.append({"basket_domain": basket, "payload": dict(payload)})
def save(name: str, content: str) -> Path:
path = PROBE_DIR / name
path.write_text(content, encoding="utf-8")
print(f" saved -> {path.name} ({len(content)} bytes)")
return path
async def inspect_options(site: SiteInteractor, item_url: str) -> list[dict]:
"""只读打开商品页,把解析出的选项结构落盘,供人工核对占位项判定是否符合真实页面"""
page = await site._context.new_page()
try:
await page.goto(item_url, wait_until="domcontentloaded", timeout=30_000)
await page.wait_for_function(
"() => window.__INITIAL_STATE__ && window.__INITIAL_STATE__.purchase",
timeout=10_000,
)
html = await page.content()
finally:
await page.close()
state = _parse_initial_state(html) or {}
information = (state.get("purchase") or {}).get("information") or {}
options = parse_options(information)
auto_choice, unfillable = auto_choice_for(options)
slug = item_url.rstrip("/").replace("https://item.rakuten.co.jp/", "").replace("/", "_")
save(
f"options-{slug}.json",
json.dumps(
{
"item_url": item_url,
"raw_options": information.get("options"),
"parsed": [
{
"name": option.name,
"type": option.type,
"is_required": option.is_required,
"values": [
{
"value_id": value.value_id,
"name": value.name,
"is_placeholder": value.is_placeholder,
}
for value in option.values
],
}
for option in options
],
"auto_choice": auto_choice,
"unfillable_required_options": unfillable,
},
ensure_ascii=False,
indent=1,
),
)
for option in options:
if not option.is_required:
continue
names = [
f"{value.name}{'(占位)' if value.is_placeholder else ''}" for value in option.values
]
print(f" 必填项「{option.name[:30]}」候选={names}")
print(f" auto_choice = {auto_choice!r}")
if unfillable:
print(f" !! 自动填不了的必填项:{unfillable}")
return [{"name": o.name, "is_required": o.is_required} for o in options]
async def run() -> int:
settings = get_settings()
capture = _PayloadCapture()
logging.basicConfig(level=logging.WARNING)
# 必须显式把这个 logger 抬到 INFO:basicConfig 把 root 设成 WARNING,而目标
# logger 自身是 NOTSET,有效级别继承 root——INFO 记录会在到达 handler 之前
# 就被丢掉,capture 全程收不到东西(第一次跑就是这样,choice 打印成 None)
target_logger = logging.getLogger("app.trading.worker.site_interact")
target_logger.setLevel(logging.INFO)
target_logger.addHandler(capture)
auth = AuthSession(settings)
await auth.start()
print("=== 0. 登录态探测(storage_state 落盘于 2026-08-11,可能已过期)===")
status = await auth.check("rakuten")
print(f" logged_in={status.logged_in} detail={status.detail}")
if not status.logged_in:
print(" 登录态已失效——需要先跑 scripts/login.py 重新登录(可能要人工过验证码)")
await auth.close()
return 2
site = SiteInteractor(auth_session=auth, settings=settings)
await site.start()
results: list[dict] = []
try:
print("\n=== 1. 清空购物车(排除历史残留污染校验)===")
cleared = await site.clear_cart()
print(f" removed={cleared['removed_count']} cart_count={cleared['cart_count']}")
for index, item_url in enumerate(TARGETS, start=1):
print(f"\n=== 2.{index} {item_url} ===")
print(" -- 只读核对选项结构 --")
try:
await inspect_options(site, item_url)
except Exception as exc: # noqa: BLE001
print(f" 选项结构读取失败(不影响加购验证):{type(exc).__name__}: {exc}")
print(" -- 走生产路径加购(add_to_cart_payload)--")
before = len(capture.payloads)
entry: dict = {"item_url": item_url}
try:
added = await site.add_to_cart_payload(item_url=item_url, quantity=1)
entry["ok"] = True
entry["item_id"] = added["item_id"]
entry["cart_count"] = added["cart_count"]
print(
f" 加购成功 item_id={added['item_id']} cart_count={added['cart_count']}"
)
save(f"add-response-{index}.html", added.get("response_html") or "")
except AppError as exc:
entry["ok"] = False
entry["error"] = exc.message
print(f" 加购失败:{exc.message}")
except Exception as exc: # noqa: BLE001
entry["ok"] = False
entry["error"] = f"{type(exc).__name__}: {exc}"
print(f" 加购异常:{type(exc).__name__}: {exc}")
# 真正发出去的 payload(从生产代码日志截获)
sent = capture.payloads[before:]
if not sent:
# 抓不到就等于这次验证什么也没证明——必须显眼报出来,不能让
# 「加购成功」把它盖过去(第一次跑就是 logger 级别没抬导致静默为 None)
entry["capture_failed"] = True
print(" !! 未截获到 payload 日志,本次无法确认实际提交的 choice")
else:
entry["sent_payload"] = sent[-1]["payload"]
choice = sent[-1]["payload"].get("choice")
print(f" 实际提交 choice = {choice!r}")
if choice is None:
print(" !! payload 里没有 choice 字段")
elif "選択してください" in choice:
print(" !! 提交值里仍含占位项——修复未生效")
results.append(entry)
print("\n=== 3. 购物车状态(确认商品真的进车了)===")
try:
status_after = await site.cart_status()
print(f" {status_after}")
except AppError as exc:
print(f" 查询失败:{exc.message}")
print("\n=== 4. 清空购物车(恢复账号原状,不进入结算)===")
final = await site.clear_cart()
print(f" removed={final['removed_count']} cart_count={final['cart_count']}")
finally:
await site.close()
await auth.close()
save("summary.json", json.dumps(results, ensure_ascii=False, indent=1))
print("\n========== 汇总 ==========")
for entry in results:
mark = "OK " if entry.get("ok") else "FAIL"
choice = (entry.get("sent_payload") or {}).get("choice")
print(f" {mark} {entry['item_url']}")
print(f" choice={choice!r}")
if not entry.get("ok"):
print(f" error={entry.get('error')}")
print(f"\n探针输出目录:{PROBE_DIR}")
return 0
if __name__ == "__main__":
raise SystemExit(asyncio.run(run()))
+154
View File
@@ -0,0 +1,154 @@
"""对外失败响应的埋点测试:异常处理器是失败的唯一出口
`register_exception_handlers` 里的处理器把异常吃掉换成 `ApiResponse` 信封返回
异常不再向上冒**自动 instrumentation 之后只看得到一个 HTTP 状态码**
`AppError` 默认 `status_code=400`信封里 `success=false`在链路上跟正常返回几乎
分不出来兜底处理器更是只回一句无信息量的 "Internal server error"真正的异常
类型与栈只在本进程日志里所以处理器必须把这次失败记到当前 server span
这里挂真实的 FastAPIInstrumentor 中间件而不是手工造 span断言的就是一次
真实请求打进来失败返回之后server span 上有什么
"""
from __future__ import annotations
import pytest
from fastapi import APIRouter, FastAPI
from fastapi.testclient import TestClient
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.trace import StatusCode
from pydantic import BaseModel
from app.shared.api import ApiResponse, register_exception_handlers
from app.shared.errors import ItemNotFoundError, UpstreamBlockedError
class _Body(BaseModel):
keyword: str
@pytest.fixture
def app_and_spans():
"""挂了异常处理器 + 真实 OTel ASGI 中间件的最小应用
provider 显式传给 instrument_app避免碰全局 provider只允许设置一次
测试间共享会互相污染
"""
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
router = APIRouter()
@router.get("/api/blocked")
async def blocked() -> ApiResponse[None]:
raise UpstreamBlockedError("Blocked while fetching: challenge page detected")
@router.get("/api/missing")
async def missing() -> ApiResponse[None]:
raise ItemNotFoundError("Page not found: https://item.rakuten.co.jp/x/y/")
@router.get("/api/boom")
async def boom() -> ApiResponse[None]:
raise RuntimeError("unexpected explosion")
@router.post("/api/validated")
async def validated(_body: _Body) -> ApiResponse[None]:
return ApiResponse[None](success=True, msg="success", data=None, code=0)
app = FastAPI()
app.include_router(router)
register_exception_handlers(app)
FastAPIInstrumentor.instrument_app(app, tracer_provider=provider)
try:
yield app, exporter
finally:
FastAPIInstrumentor.uninstrument_app(app)
def _server_span(exporter: InMemorySpanExporter):
spans = exporter.get_finished_spans()
assert spans, "没有产生 server span"
return spans[-1]
def test_app_error_is_recorded_on_server_span(app_and_spans):
"""业务异常:错误码、可重试、异常类型都要落在 span 上
AppError 默认 status_code=400链路里光看状态码只知道客户端错了
不知道是被反爬阻断3002可重试还是别的什么
"""
app, exporter = app_and_spans
with TestClient(app) as http:
response = http.get("/api/blocked")
assert response.status_code == 400
assert response.json()["code"] == 3002
span = _server_span(exporter)
assert span.status.status_code is StatusCode.ERROR
assert span.attributes["error.type"] == "UpstreamBlockedError"
assert span.attributes["error.code"] == 3002
# 上游据此判断该不该重试,是失败分类里最要紧的一位
assert span.attributes["error.retryable"] is True
assert span.attributes["api.success"] is False
assert span.attributes["api.code"] == 3002
assert "challenge page" in span.attributes["api.msg"]
# 异常栈作为 event 保留,需要细看时能展开
assert any(event.name == "exception" for event in span.events)
def test_not_found_records_its_own_code(app_and_spans):
"""404 类业务异常与阻断类要能按码区分(4004 不可重试)"""
app, exporter = app_and_spans
with TestClient(app) as http:
assert http.get("/api/missing").status_code == 404
span = _server_span(exporter)
assert span.attributes["error.code"] == 4004
assert span.attributes["error.retryable"] is False
def test_unhandled_exception_records_real_cause(app_and_spans):
"""兜底分支最需要埋点:对外只回 "Internal server error",真因只在日志里"""
app, exporter = app_and_spans
with TestClient(app, raise_server_exceptions=False) as http:
response = http.get("/api/boom")
assert response.status_code == 500
assert response.json()["msg"] == "Internal server error"
span = _server_span(exporter)
assert span.status.status_code is StatusCode.ERROR
# 响应体里查不到的真因,链路上能直接看到
assert span.attributes["error.type"] == "RuntimeError"
assert "unexpected explosion" in span.attributes["error.message"]
assert span.attributes["api.code"] == 1500
def test_validation_error_records_field_message(app_and_spans):
"""参数校验失败记错误码与字段消息,但不记 pydantic 那条没有诊断价值的长栈"""
app, exporter = app_and_spans
with TestClient(app) as http:
response = http.post("/api/validated", json={})
assert response.status_code == 422
assert response.json()["code"] == 1002
span = _server_span(exporter)
assert span.status.status_code is StatusCode.ERROR
assert span.attributes["api.code"] == 1002
# 排查要看的是「哪个字段不合法」
assert "keyword" in span.attributes["api.msg"]
assert "error.type" not in span.attributes
def test_success_leaves_server_span_ok(app_and_spans):
"""成功请求不被误标 ERROR——作为对照说明失败断言不是恒真"""
app, exporter = app_and_spans
with TestClient(app) as http:
assert http.post("/api/validated", json={"keyword": "switch"}).status_code == 200
span = _server_span(exporter)
assert span.status.status_code is not StatusCode.ERROR
assert "api.success" not in span.attributes
assert "error.type" not in span.attributes
+73
View File
@@ -142,6 +142,79 @@ def test_item_detail_falls_back_to_request_shop_code_when_state_lacks_it(item_st
assert detail.shop.shop_code == "fallback"
# ---- 店铺自定义选项(下单必填项)----
def test_item_detail_exposes_shop_options_for_ordering(item_with_options_state):
"""下单要用的选项必须能从接口拿到:名字、类型、是否必填、候选取值"""
detail = parse_item_detail(
item_with_options_state, item_url="https://x", shop_code="kizamu",
include_sku_variants=True,
)
assert detail.has_required_options is True
# 真实样本:1 个必填 select(名入れ)+ 1 个非必填 select + 2 个自由文本
assert [option.name for option in detail.options] == [
"名入れ",
" 【お名前】※漢字・かな文字:4文字まで",
"【定型】",
"【定型】※〇寿・〇婚式・〇長を選択の場合のみ(1文字まで)",
]
required = detail.options[0]
assert required.is_required is True
assert required.type == "select"
assert required.option_id == 100
def test_item_detail_marks_placeholder_option_values_as_unselectable(item_with_options_state):
"""「選択してください」是占位项,不能作为下单取值——这是 choice 填错的根因"""
detail = parse_item_detail(
item_with_options_state, item_url="https://x", shop_code="kizamu",
include_sku_variants=True,
)
required = detail.options[0]
placeholder, first_real = required.values[0], required.values[1]
assert placeholder.name == "選択してください"
assert placeholder.is_placeholder is True
assert first_real.name == "希望する【次の項目で入力】"
assert first_real.is_placeholder is False
# 3 个候选里只有 2 个真正可提交
assert len(required.values) == 3
assert required.selectable_value_count == 2
def test_item_detail_free_text_option_has_no_values(item_with_options_state):
"""type=text 是自由文本,站点不给候选值,无法自动选"""
detail = parse_item_detail(
item_with_options_state, item_url="https://x", shop_code="kizamu",
include_sku_variants=True,
)
free_text = detail.options[1]
assert free_text.type == "text"
assert free_text.values == []
assert free_text.selectable_value_count == 0
def test_item_detail_reports_required_options_that_cannot_be_auto_filled(item_with_options_state):
"""必填自由文本项无法自动选值,必须点名交给调用方,不能静默蒙一个值"""
options = item_with_options_state["purchase"]["information"]["options"]
# 把那个自由文本项改成必填:店铺要求填「お名前」时就是这种形态
options[1]["isRequired"] = True
detail = parse_item_detail(
item_with_options_state, item_url="https://x", shop_code="kizamu",
include_sku_variants=True,
)
assert detail.unfillable_required_options == [" 【お名前】※漢字・かな文字:4文字まで"]
def test_item_detail_without_options_reports_empty(item_state):
"""没有选项的商品(真实样本 purchase.information 只有 unit)不应凭空造出选项"""
detail = parse_item_detail(
item_state, item_url="https://x", shop_code="s", include_sku_variants=False,
)
assert detail.options == []
assert detail.has_required_options is False
assert detail.unfillable_required_options == []
def test_item_detail_rejects_state_without_item_node():
with pytest.raises(ScrapeParseError):
parse_item_detail({}, item_url="https://x", shop_code="s", include_sku_variants=True)
+296
View File
@@ -0,0 +1,296 @@
"""抓取客户端失败分支的埋点测试:失败原因必须能从 span 上读出来
这里钉住的核心是一个真实存在过的 bug失败分支里写的是无参
`span.record_exception()`而该方法的 `exception` 是必填位置参数**每一次抓取
失败都会在记录异常时抛 TypeError**把真正的失败原因反爬阻断 / 解析失败 /
404整个替换掉链路上只剩调用发生过异常怎么来的完全看不到失败页面
快照也永远落不下来抛错发生在 snapshot 之前
用独立的 InMemory provider 断言 span不碰全局 providerOTel 的全局 provider
只允许设置一次测试间共享会互相污染 test_telemetry.py 同一套思路
"""
from __future__ import annotations
import json
import httpx
import pytest
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.trace import StatusCode
from app.scraping.models.scrape import (
ItemDetailRequest,
RakumaSearchRequest,
SearchRequest,
ShopItemsRequest,
)
from app.scraping.services import rakuma_client as rakuma_module
from app.scraping.services import rakuten_client as rakuten_module
from app.scraping.services import site_session as session_module
from app.scraping.services.rakuma_client import RakumaClient
from app.scraping.services.rakuten_client import RakutenClient
from app.scraping.services.site_session import SiteSession
from app.shared.config import Settings
from app.shared.errors import AppError, ScrapeParseError, UpstreamBlockedError
@pytest.fixture
def spans(monkeypatch) -> InMemorySpanExporter:
"""把两个抓取客户端模块级 tracer 换成写内存的,用于断言 span"""
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
tracer = provider.get_tracer("test")
monkeypatch.setattr(rakuten_module, "tracer", tracer)
monkeypatch.setattr(rakuma_module, "tracer", tracer)
return exporter
def _settings() -> Settings:
return Settings(_env_file=None, otel_snapshot_max_bytes=2_000_000)
class _StubSession:
"""按需返回 HTML 或抛异常的会话桩;两站的 fetch_html 签名差异在这里吸收"""
def __init__(self, *, html: str = "", error: Exception | None = None):
self._html = html
self._error = error
async def fetch_html(self, url: str, *, mobile: bool = False) -> str:
if self._error is not None:
raise self._error
return self._html
async def fetch(self, url: str, *, mobile: bool = False, validator=None):
if self._error is not None:
raise self._error
raise AssertionError("本测试不该走到 fetch 的成功分支")
def _span(exporter: InMemorySpanExporter, name: str):
return next(s for s in exporter.get_finished_spans() if s.name == name)
async def test_parse_failure_preserves_original_exception(spans):
"""回归:解析失败要原样抛出业务异常,不能被埋点自身的 TypeError 顶替
这是原 bug 最直接的后果上游拿到的不再是 4001解析失败而是一个
TypeError 兜底成的 500错误码表整个失效
"""
client = RakutenClient(_settings(), _StubSession(html="<html>no state</html>"))
with pytest.raises(ScrapeParseError) as excinfo:
await client.search(SearchRequest(keyword="switch"))
# 是业务异常而不是埋点炸出来的 TypeError
assert isinstance(excinfo.value, AppError)
assert excinfo.value.err_code == 4001
async def test_parse_failure_records_reason_and_page_snapshot(spans):
"""页面取回来了但解析不出:stage=parse,且失败页面 HTML 落成 event
站点改版了只能靠当时那份 HTML 判断所以快照必须真的落下来 bug
记异常那步先抛了snapshot 这行永远执行不到
"""
html = "<html>changed layout</html>"
client = RakutenClient(_settings(), _StubSession(html=html))
with pytest.raises(ScrapeParseError):
await client.search(SearchRequest(keyword="switch"))
span = _span(spans, "parse.rakuten.search")
assert span.status.status_code is StatusCode.ERROR
assert span.attributes["parse.stage"] == "parse"
# 真实异常类名,不是一律写死的 "parse_error"
assert span.attributes["parse.fail_reason"] == "ScrapeParseError"
assert span.attributes["error.code"] == 4001
assert span.attributes["error.retryable"] is False
snapshot = next(e for e in span.events if e.name == "parse.failed_html")
assert snapshot.attributes["snapshot.html"] == html
# 快照要能对上是哪个地址的页面
assert "search.rakuten.co.jp" in snapshot.attributes["parse.url"]
async def test_fetch_failure_marks_fetch_stage_without_snapshot(spans):
"""页面根本没取回来:stage=fetch,没有快照可落
parse 阶段的排查方向相反通道/反爬/上游 5xx而不是站点改版
所以要能直接按 parse.stage 分流而不是对着有没有 HTML
"""
blocked = UpstreamBlockedError("Blocked while fetching: challenge page detected")
client = RakutenClient(_settings(), _StubSession(error=blocked))
with pytest.raises(UpstreamBlockedError):
await client.search(SearchRequest(keyword="switch"))
span = _span(spans, "parse.rakuten.search")
assert span.attributes["parse.stage"] == "fetch"
assert span.attributes["parse.fail_reason"] == "UpstreamBlockedError"
# 反爬阻断是可重试的,这一位直接决定上游要不要重来
assert span.attributes["error.retryable"] is True
assert not [e for e in span.events if e.name == "parse.failed_html"]
async def test_item_detail_failure_records_reason(spans):
"""详情接口走的是 fetch()(带校验器)而非 fetch_html,失败分支同样要记全"""
blocked = UpstreamBlockedError("challenge page detected")
client = RakutenClient(_settings(), _StubSession(error=blocked))
with pytest.raises(UpstreamBlockedError):
await client.item_detail(
ItemDetailRequest(shop_code="someshop", item_code="10000001")
)
span = _span(spans, "parse.rakuten.item_detail")
assert span.status.status_code is StatusCode.ERROR
assert span.attributes["parse.fail_reason"] == "UpstreamBlockedError"
async def test_shop_items_marks_delegate_stage(spans):
"""shop_items 自己不抓页面:失败在转调的 shop_detail / search 里
html 推断会得出fetch 失败的错误结论它手里从来没有 html
所以这里显式标 delegate且不落快照真正的现场在被转调那个 span
"""
blocked = UpstreamBlockedError("challenge page detected")
client = RakutenClient(_settings(), _StubSession(error=blocked))
with pytest.raises(UpstreamBlockedError):
await client.shop_items(ShopItemsRequest(shop_code="someshop"))
span = _span(spans, "parse.rakuten.shop_items")
assert span.attributes["parse.stage"] == "delegate"
assert not [e for e in span.events if e.name == "parse.failed_html"]
# 被转调的那一步才是现场所在,它自己落了快照
inner = _span(spans, "parse.rakuten.shop_detail")
assert inner.attributes["parse.fail_reason"] == "UpstreamBlockedError"
async def test_rakuma_parse_failure_records_reason_and_snapshot(spans):
"""ラクマ 侧五个接口是同一套失败分支,同样要能读出原因与现场"""
html = "<html>rakuma changed</html>"
client = RakumaClient(_settings(), _StubSession(html=html))
with pytest.raises(AppError):
await client.search(RakumaSearchRequest(keyword="switch"))
span = _span(spans, "parse.rakuma.search")
assert span.status.status_code is StatusCode.ERROR
assert span.attributes["parse.stage"] == "parse"
assert "error.code" in span.attributes
snapshot = next(e for e in span.events if e.name == "parse.failed_html")
assert snapshot.attributes["snapshot.html"] == html
# ---- 会话层:逐次尝试与升级路径 ----
#
# 这一层的关键信息是「升级路径」:换 cookie → 浏览器兜底 → 放弃。属性表达不了过程
# (同名后写覆盖先写,三次尝试跑完只剩最后一次),所以每次尝试与每次升级各记一条
# event。site_session / rakuma_session 都在 fetch 内部 `trace.get_tracer(__name__)`,
# 所以这里替 `trace.get_tracer` 本身(与 test_telemetry.py 的 spans 夹具同一手法)。
@pytest.fixture
def session_spans(monkeypatch) -> InMemorySpanExporter:
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
monkeypatch.setattr(session_module.trace, "get_tracer", provider.get_tracer)
return exporter
class _FakeBrowser:
"""浏览器兜底替身:visit 恒不可用,让升级链走完整条路"""
def __init__(self) -> None:
self.unavailable_reason = "playwright is not installed"
async def visit(self, url: str, *, mobile: bool):
return None
async def close(self) -> None:
pass
async def _blocked_session() -> SiteSession:
"""所有请求都回 Akamai 挑战页的会话,用 MockTransport 拦截"""
settings = Settings(
_env_file=None, http_max_attempts=3, request_timeout_seconds=5.0,
max_site_concurrency=4,
)
session = SiteSession(settings, _FakeBrowser())
await session.start()
for profile in session._profiles.values():
await profile.client.aclose()
profile.client = httpx.AsyncClient(
transport=httpx.MockTransport(
lambda _r: httpx.Response(
200, text="<html>Access Denied. Reference #18.abc</html>"
)
),
follow_redirects=True,
)
return session
async def test_retry_attempts_and_escalation_are_recorded_as_events(session_spans):
"""每次尝试与每次升级各留一条 event,链路上能读出完整升级路径
这些信息无法用属性表达`scrape.attempts` 只剩最后一次的值前两次为什么失败
换过 cookie 没有浏览器兜底试过没有全被覆盖掉
"""
session = await _blocked_session()
try:
with pytest.raises(UpstreamBlockedError):
await session.fetch_html(
"https://search.rakuten.co.jp/search/mall/x/", mobile=False
)
finally:
await session.close()
span = _span(session_spans, "scrape.fetch")
assert span.status.status_code is StatusCode.ERROR
assert span.attributes["scrape.fail_reason"] == "UpstreamBlockedError"
assert span.attributes["scrape.challenge_detected"] is True
# 三次尝试都留下了自己的记录,而不是只剩最后一次
attempts = [e for e in span.events if e.name == "scrape.attempt"]
assert [e.attributes["attempt"] for e in attempts] == [1, 2, 3]
assert {e.attributes["outcome"] for e in attempts} == {"challenge"}
# 升级路径:第 1 次失败后换 cookie,第 2 次失败后动用浏览器(本例不可用)
escalations = [e for e in span.events if e.name == "scrape.escalate"]
assert [e.attributes["to"] for e in escalations] == ["rewarm_on_home", "browser"]
browser_step = escalations[-1]
assert browser_step.attributes["outcome"] == "failed"
# 「没装 playwright」与「装了也被挡」要能分开查
assert browser_step.attributes["reason"] == "playwright is not installed"
# 最终失败页面的快照带上来源,便于确认是哪条通道的哪个地址
failed = next(e for e in span.events if e.name == "scrape.failed_html")
assert failed.attributes["scrape.profile"] == "pc"
assert "search.rakuten.co.jp" in failed.attributes["scrape.url"]
async def test_successful_scrape_leaves_span_ok(spans, search_state):
"""成功路径不被误标 ERROR,不落失败快照,且照常记结果指标
作为上面那些失败断言的对照证明 ERROR 状态与 parse.fail_reason 是真的由失败
触发的不是每条 span 都长这样
"""
html = f"<script>window.__INITIAL_STATE__ = {json.dumps(search_state)};</script>"
client = RakutenClient(_settings(), _StubSession(html=html))
result = await client.search(SearchRequest(keyword="switch"))
span = _span(spans, "parse.rakuten.search")
assert span.status.status_code is not StatusCode.ERROR
assert "parse.fail_reason" not in span.attributes
assert "error.type" not in span.attributes
assert not [e for e in span.events if e.name == "parse.failed_html"]
# 成功时记的是结果指标,与失败侧属性互不重叠
assert span.attributes["parse.items"] == len(result.items)
+89
View File
@@ -180,6 +180,95 @@ def test_extract_fields_required_options_auto_picks_first_value():
assert fields["form_fields"]["choice"] == "サイズ:S"
def test_extract_fields_required_options_skips_placeholder_value():
"""回归:必填 select 的 values[0] 是「選択してください」占位项,不能当答案提交
真实商品页.probe/checkout/03-add-fail-2.html 3 份样本里必填 select
第一个候选恒为 id=0 的占位项真实取值从 id=200 旧实现直接取 values[0]
等于把請選擇填进 choice 提交上去
"""
state = {
"item": {"itemId": 1, "variantId": "v"},
"purchase": {
"sku": {"inventoryType": "single"},
"sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}},
"information": {
"options": [
{
"id": 100,
"name": "この商品は「トライタン製/プラスチック」です",
"type": "select",
"isRequired": True,
"values": [
{"id": 0, "name": "選択してください"},
{"id": 200, "name": "確認した"},
],
}
]
},
},
"shop": {"information": {"shopId": 1}},
}
fields = _extract_purchase_fields(state, intent_override={})
assert fields["has_required_options"] is True
assert fields["form_fields"]["choice"] == "この商品は「トライタン製/プラスチック」です:確認した"
assert fields["unfillable_required_options"] == []
def test_extract_fields_required_free_text_option_cannot_be_auto_filled():
"""必填自由文本项(type=text,无候选值)自动填不了,要点名报出来"""
state = {
"item": {"itemId": 1, "variantId": "v"},
"purchase": {
"sku": {"inventoryType": "single"},
"sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}},
"information": {
"options": [
{"id": 101, "name": "【お名前】", "type": "text", "isRequired": True},
]
},
},
"shop": {"information": {"shopId": 1}},
}
fields = _extract_purchase_fields(state, intent_override={})
assert fields["has_required_options"] is True
# 填不出来就不填,也不拿占位/空值凑数
assert "choice" not in fields["form_fields"]
assert fields["unfillable_required_options"] == ["【お名前】"]
def test_extract_fields_skips_optional_options_when_auto_filling():
"""非必填项不替上游做业务决定(如「置き配を希望する」),只自动填必填项"""
state = {
"item": {"itemId": 1, "variantId": "v"},
"purchase": {
"sku": {"inventoryType": "single"},
"sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}},
"information": {
"options": [
{
"id": 100,
"name": "必須確認",
"type": "select",
"isRequired": True,
"values": [{"id": 0, "name": "選択してください"}, {"id": 200, "name": "了解"}],
},
{
"id": 101,
"name": "「置き配」希望について",
"type": "select",
"isRequired": False,
"values": [{"id": 200, "name": "置き配を希望しない"}],
},
]
},
},
"shop": {"information": {"shopId": 1}},
}
fields = _extract_purchase_fields(state, intent_override={})
assert fields["form_fields"]["choice"] == "必須確認:了解"
def test_extract_fields_intent_choice_override_accepts_list_and_str():
state = {
"item": {"itemId": 1, "variantId": "v"},
+151 -1
View File
@@ -5,6 +5,9 @@
2. enabled=true + endpoint setup 注册真实 TracerProvidershutdown 复位
3. instrument_app setup 之前调用也要真的装上中间件三个服务都是导入期打桩
4. traced / set_attributes / record_error 的行为
5. 采集返回结果这一侧record_envelope信封失败在 HTTP 层看不出来
record_parse_failure失败在 fetch 还是 parsesnapshot失败页面快照
以及 span_unless_suppressed suppressed() 的配套关系
不打真实网络OTLPSpanExporter 创建时不发请求BatchSpanProcessor 异步批量
上报在没有 span 产生时也不会触发span 断言用独立的 InMemory provider不碰
@@ -27,7 +30,7 @@ from opentelemetry.util.http import parse_excluded_urls
from app.shared import telemetry
from app.shared.config import Settings
from app.shared.errors import OrderGuardError
from app.shared.errors import OrderGuardError, ScrapeParseError, UpstreamBlockedError
from app.shared.telemetry import is_initialized, setup_telemetry, shutdown_telemetry
@@ -189,6 +192,153 @@ def test_set_attributes_skips_none(spans):
assert "b" not in attributes
def test_record_error_keeps_app_error_fields(spans):
"""AppError 的排查字段(错误码/可重试/状态码/消息)都要落到属性上
只落 error.type 不够上游看到的是错误码该不该重试 retryable
record_exception 记的 event 在多数观测后台里要展开才看得到列表页筛不出来
"""
tracer = telemetry.trace.get_tracer(__name__)
with tracer.start_as_current_span("unit.err") as span:
telemetry.record_error(span, UpstreamBlockedError("Akamai 挑战页"))
attributes = spans.get_finished_spans()[0].attributes
assert attributes["error.type"] == "UpstreamBlockedError"
assert attributes["error.code"] == UpstreamBlockedError().err_code
assert attributes["error.retryable"] is True
assert attributes["error.status_code"] == 400
assert "Akamai 挑战页" in attributes["error.message"]
def test_record_error_on_plain_exception_omits_app_error_fields(spans):
"""非 AppError 不应凭空长出 error.code / error.retryable 属性"""
tracer = telemetry.trace.get_tracer(__name__)
with tracer.start_as_current_span("unit.plain") as span:
telemetry.record_error(span, RuntimeError("boom"))
attributes = spans.get_finished_spans()[0].attributes
assert attributes["error.type"] == "RuntimeError"
assert "error.code" not in attributes
assert "error.retryable" not in attributes
def test_record_envelope_failure_marks_span_error(spans):
"""信封失败要置 ERROR 并落错误码——HTTP 层看不出这次调用失败了"""
tracer = telemetry.trace.get_tracer(__name__)
with tracer.start_as_current_span("unit.envelope") as span:
telemetry.record_envelope(
span, success=False, err_code=6002, msg="租约无效", status_code=409
)
finished = spans.get_finished_spans()[0]
assert finished.status.status_code is StatusCode.ERROR
assert finished.attributes["api.success"] is False
assert finished.attributes["api.code"] == 6002
assert finished.attributes["api.status_code"] == 409
assert finished.attributes["api.msg"] == "租约无效"
def test_record_envelope_success_leaves_status_ok(spans):
tracer = telemetry.trace.get_tracer(__name__)
with tracer.start_as_current_span("unit.envelope_ok") as span:
telemetry.record_envelope(span, success=True, status_code=200)
finished = spans.get_finished_spans()[0]
assert finished.status.status_code is not StatusCode.ERROR
assert finished.attributes["api.success"] is True
def test_snapshot_truncates_and_carries_extra(spans):
"""超限 HTML 截断并标注,附带的来源信息(URL 等)也要落在同一条 event 上"""
tracer = telemetry.trace.get_tracer(__name__)
with tracer.start_as_current_span("unit.snapshot") as span:
telemetry.snapshot(
span, "parse.failed_html", "x" * 100, 10,
extra={"parse.url": "https://example.com/a", "parse.ignored": None},
)
event = spans.get_finished_spans()[0].events[0]
assert event.name == "parse.failed_html"
assert event.attributes["snapshot.html"] == "x" * 10
assert event.attributes["snapshot.original_bytes"] == 100
assert event.attributes["snapshot.truncated"] is True
assert event.attributes["parse.url"] == "https://example.com/a"
assert "parse.ignored" not in event.attributes
def test_snapshot_skips_empty_html(spans):
"""页面根本没取回来时不记空 event"""
tracer = telemetry.trace.get_tracer(__name__)
with tracer.start_as_current_span("unit.snapshot_empty") as span:
telemetry.snapshot(span, "parse.failed_html", None, 100)
telemetry.snapshot(span, "parse.failed_html", "", 100)
assert spans.get_finished_spans()[0].events == ()
def test_record_parse_failure_distinguishes_fetch_from_parse(spans):
"""失败阶段按「HTML 有没有拿到」区分:两者排查方向相反
html 为空页面没取回来通道/反爬/上游 5xx非空取回了但解析不出
多半站点改版fail_reason 要给真实异常类名不能一律写死 parse_error
"""
tracer = telemetry.trace.get_tracer(__name__)
with tracer.start_as_current_span("unit.fetch_fail") as span:
telemetry.record_parse_failure(
span, UpstreamBlockedError("挑战页"), html=None, max_bytes=100,
url="https://example.com/a",
)
with tracer.start_as_current_span("unit.parse_fail") as span:
telemetry.record_parse_failure(
span, ScrapeParseError("没有 state"), html="<html/>", max_bytes=100,
url="https://example.com/b",
)
with tracer.start_as_current_span("unit.delegate_fail") as span:
telemetry.record_parse_failure(
span, ScrapeParseError("转调失败"), stage="delegate",
)
by_name = {s.name: s for s in spans.get_finished_spans()}
fetch = by_name["unit.fetch_fail"]
assert fetch.attributes["parse.stage"] == "fetch"
assert fetch.attributes["parse.fail_reason"] == "UpstreamBlockedError"
# 页面没取回来,没有快照可落
assert [e.name for e in fetch.events] == ["exception"]
parsed = by_name["unit.parse_fail"]
assert parsed.attributes["parse.stage"] == "parse"
assert parsed.attributes["parse.fail_reason"] == "ScrapeParseError"
snapshot_event = next(e for e in parsed.events if e.name == "parse.failed_html")
assert snapshot_event.attributes["snapshot.html"] == "<html/>"
assert snapshot_event.attributes["parse.url"] == "https://example.com/b"
# 显式 stage 覆盖推断:编排方法的失败既不在自己的 fetch 也不在自己的 parse
assert by_name["unit.delegate_fail"].attributes["parse.stage"] == "delegate"
def test_span_unless_suppressed_is_noop_inside_suppressed(spans):
"""`suppressed()` 里手工埋点必须退化成 noop,否则空转长轮询绕开抑制刷满后台
`suppress_instrumentation` 只被 instrumentation 库尊重手工
`start_as_current_span` 不看它worker lease 正是在 suppressed() 里调
GatewayClient._request
"""
tracer = telemetry.trace.get_tracer(__name__)
with telemetry.suppressed():
with telemetry.span_unless_suppressed(tracer, "unit.suppressed") as span:
# 属性/异常写在 noop span 上不能报错,调用方不必分支
telemetry.set_attributes(span, {"a": 1})
telemetry.record_error(span, RuntimeError("boom"))
assert not span.is_recording()
assert spans.get_finished_spans() == ()
with telemetry.span_unless_suppressed(tracer, "unit.not_suppressed") as span:
assert span.is_recording()
assert [s.name for s in spans.get_finished_spans()] == ["unit.not_suppressed"]
def test_enabled_initializes_and_shutdown_releases():
"""enabled=true 时 setup 注册 TracerProvider,shutdown 后 _provider 复位"""
settings = Settings(
+158
View File
@@ -0,0 +1,158 @@
"""GatewayClient 埋点测试:信封里的失败必须在链路上看得见
要钉住的问题**网关的失败在响应信封里不在 HTTP 状态码上**一次
`success=false, code=6002`租约无效的回报httpx 自动 instrumentation 只看到
一个完成了的请求而且那个 CLIENT span `request()` 返回时就结束了此时信封
还没解所以它永远不可能带上这次调用的结论链路里于是只剩调用发生过
另一半是抑制worker 的空转长轮询在 `suppressed()` 里调 `_request`
`suppress_instrumentation` 只被 instrumentation 库尊重手工建的 span 不看它
手工埋点若不配合 `span_unless_suppressed`孤立 trace 会从这个口子重新灌回来
全部用 httpx.MockTransport 拦截不触达真实网关
"""
from __future__ import annotations
import httpx
import pytest
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.trace import SpanKind, StatusCode
from app.shared.config import Settings
from app.shared.errors import AppError
from app.shared.task_state import OrderState
from app.shared.telemetry import suppressed
from app.trading.worker import client as worker_client
from app.trading.worker.client import GatewayClient
@pytest.fixture
def spans(monkeypatch) -> InMemorySpanExporter:
"""把 client 模块级 tracer 换成写内存的(不碰全局 provider)"""
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
monkeypatch.setattr(worker_client, "tracer", provider.get_tracer("test"))
return exporter
def _client(handler) -> GatewayClient:
"""构建 GatewayClient 并把它的 httpx 客户端换成 MockTransport 版本"""
client = GatewayClient(
"https://gateway.example", "token", settings=Settings(_env_file=None)
)
client._client = httpx.AsyncClient(
base_url="https://gateway.example",
transport=httpx.MockTransport(handler),
)
return client
def _envelope(*, success: bool, code: int, msg: str, data=None, status: int = 200):
return httpx.Response(
status,
json={"success": success, "msg": msg, "data": data, "code": code},
)
async def test_envelope_failure_is_recorded_on_span(spans):
"""success=false 要把 span 置 ERROR 并落错误码——HTTP 层看这次是 200"""
gateway = _client(
lambda _r: _envelope(
success=False, code=6002, msg="租约无效:不是持有者", status=200
)
)
try:
with pytest.raises(AppError) as excinfo:
await gateway.report(
"task-1", "worker-1", state=OrderState.IN_CART, terminal=False
)
assert excinfo.value.err_code == 6002
finally:
await gateway.aclose()
span = spans.get_finished_spans()[0]
assert span.kind is SpanKind.CLIENT
assert span.status.status_code is StatusCode.ERROR
# 这次调用的结论——HTTP 状态码是 200,光看它分不出成败
assert span.attributes["api.success"] is False
assert span.attributes["api.code"] == 6002
assert span.attributes["api.status_code"] == 200
assert "租约无效" in span.attributes["api.msg"]
assert span.attributes["gateway.path"] == "/api/orders/task-1/report"
assert span.attributes["gateway.method"] == "POST"
async def test_successful_envelope_leaves_span_ok(spans):
"""success=true 不置 ERROR;作为对照说明上面的断言不是恒真"""
gateway = _client(
lambda _r: _envelope(success=True, code=0, msg="success", data={"recorded": True})
)
try:
await gateway.report("task-1", "worker-1", state=OrderState.IN_CART)
finally:
await gateway.aclose()
span = spans.get_finished_spans()[0]
assert span.status.status_code is not StatusCode.ERROR
assert span.attributes["api.success"] is True
assert "api.code" not in span.attributes
async def test_non_json_body_is_recorded_as_error(spans):
"""网关回了非 JSON(网关挂了/被反代拦了):记成可重试的 3001"""
gateway = _client(lambda _r: httpx.Response(502, text="<html>502 Bad Gateway</html>"))
try:
with pytest.raises(AppError) as excinfo:
await gateway.renew("task-1", "worker-1")
assert excinfo.value.err_code == 3001
finally:
await gateway.aclose()
span = spans.get_finished_spans()[0]
assert span.status.status_code is StatusCode.ERROR
assert span.attributes["error.code"] == 3001
# 这类失败是可重试的,与信封里的业务拒绝(retryable=False)要能区分开
assert span.attributes["error.retryable"] is True
assert span.attributes["gateway.status_code"] == 502
async def test_idle_long_poll_produces_no_span(spans):
"""回归:空转长轮询在 suppressed() 里不能产生 span
worker 30 lease 一次绝大多数返回空手工埋点若不看抑制标记这些
调用会各自成为一条孤立 trace 把观测后台刷满正是 suppressed() 要解决的问题
"""
gateway = _client(lambda _r: _envelope(success=True, code=0, msg="success", data=None))
try:
with suppressed():
assert await gateway.lease("worker-1", wait=0) is None
assert spans.get_finished_spans() == ()
# 抑制之外照常埋点,证明上面的空断言不是因为埋点根本没生效
assert await gateway.lease("worker-1", wait=0) is None
finally:
await gateway.aclose()
assert [s.attributes["gateway.path"] for s in spans.get_finished_spans()] == [
"/api/orders/lease"
]
async def test_leased_task_call_is_still_traced(spans):
"""领到任务后的调用不受抑制影响:它们挂在任务根 span 底下,是要看的那部分"""
gateway = _client(
lambda _r: _envelope(
success=True, code=0, msg="success",
data={"task_id": "task-1", "site": "rakuten", "lease_count": 1},
)
)
try:
task = await gateway.lease("worker-1", wait=0)
assert task is not None and task.task_id == "task-1"
finally:
await gateway.aclose()
assert len(spans.get_finished_spans()) == 1