"""乐天市场(ichiba)加购字段抽取的纯函数与常量 `__INITIAL_STATE__.purchase` 的字段映射在两处使用: - `app/scraping/parsers/item.py::_purchase_info` — `/api/item_detail` 返回的 PurchaseInfo - `app/trading/worker/site_interact.py::_extract_purchase_fields` — trading 加购链路 两者都解析同一份站点 JSON,本模块抽出**双方都要用**的核心逻辑: - 常量:普通购买事件标识、库存类型 → inventory_flag 映射 - `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 自动选择策略 — trading 独有(scraping 把决策权留给上游) - 子站(books/biccamera/brandavenue)的加购契约 — 各自独立,不走这里 依赖约束:仅 `typing.Any`(标准库),不 import scraping / trading / gateway, 符合架构测试 tests/test_architecture.py 的依赖方向规则。 """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any # 站点前端构造加购表单时固定带的事件标识(普通购买 normalPurchase) NORMAL_PURCHASE_EVENT: str = "ES01_003_001" # 库存类型 → inventory_flag # 多规格商品(sku.variants[] 非空)inventoryType="multiple" → flag "2" # 单一库存商品 → flag "1" # 注意:scraping 早期版本写过 int 2/1,但表单最终要 str,统一在这里就转成 str INVENTORY_FLAG_MULTIPLE: str = "2" INVENTORY_FLAG_DEFAULT: str = "1" _INVENTORY_FLAG_MAP: dict[str, str] = {"multiple": INVENTORY_FLAG_MULTIPLE} def inventory_flag_for(inventory_type: Any) -> str: """从 __INITIAL_STATE__.purchase.sku.inventoryType 取 inventory_flag inventory_type="multiple" → "2"(多规格,调用方需选 variant_id) 其他(含 None / 未知值)→ "1"(单一库存) """ key = inventory_type if isinstance(inventory_type, str) else "" return _INVENTORY_FLAG_MAP.get(key, INVENTORY_FLAG_DEFAULT) def basket_domain_of(sell_type: dict[str, Any]) -> str: """从 purchase.sellType.normalPurchase 抽 basketDomain 并反转义 站点 JSON 里 `/` 被编码为 `\\u002F`,POST 前要还原。 sell_type 不是 dict 或 basketDomain 缺失时返回空串。 """ if not isinstance(sell_type, dict): return "" raw = sell_type.get("basketDomain") if not isinstance(raw, str): return "" return raw.replace("\\u002F", "/") def base_form_fields( *, shop_id: Any, item_id: Any, inventory_flag: str, ) -> dict[str, str]: """构造加购表单的基础四件套 shop_bid / item_id / inventory_flag / __event 是 rakuten 主站加购的固定字段。 variant_id / choice / units 由调用方按业务策略再加(trading 自动选,scraping 留给上游)。 """ return { "shop_bid": str(shop_id) if shop_id else "", "item_id": str(item_id) if item_id else "", "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