rakuten-api/app/shared/purchase_contract.py

"""乐天市场（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