rakuten-api/app/scraping/parsers/rakuma/category.py

"""ラクマ 分类一览页 → RakumaCategoryData

站点其余页面都是 Rails 服务端渲染的老模板，只有 `/category` 换成了 Next.js
App Router：整棵分类树写在 RSC flight payload 里，一段段挂在
`self.__next_f.push([1, "<字符串>"])` 上。把这些字符串按顺序拼回去，就能拿到
`"categoryList":[{"id":...,"parentId":...,"name":...,"hasChild":...}, ...]`。

这份 categoryList 是**全量扁平树**（实测 1686 条：14 个顶层 + 169 个二级 +
1503 个三级），且与 URL 上的 `?category_id=` 无关——传任意分类或完全不传，
内容都一样。所以一次请求即可满足任意层级的查询，不需要像乐天 genre 那样
逐层下钻。

站点只给 id / parentId / name / hasChild 四个字段，没有商品数：商品数只在
`/category/{id}` 列表页的埋点属性上，要逐个分类多打一次请求，这里不做。
"""
from __future__ import annotations

import json
import re
from typing import Any

from app.shared.errors import ItemNotFoundError, ScrapeParseError
from app.scraping.models.scrape import RakumaCategoryData, RakumaCategoryNode
from app.scraping.utils.rakuma_urls import build_category_url

# flight payload 的分片：self.__next_f.push([1,"...JSON 字符串字面量..."])
_FLIGHT_CHUNK_RE = re.compile(r'self\.__next_f\.push\(\[1,("(?:[^"\\]|\\.)*")\]\)')
# 站点用 parentId=0 表示顶层分类，0 本身不是一个真实分类
ROOT_PARENT_ID = 0


def _flight_payload(html: str) -> str:
    """把 RSC flight payload 的所有分片按顺序拼成一整段文本

    每个分片是一个 JS 字符串字面量，转义规则与 JSON 一致，因此直接用
    json.loads 解码；单个分片解不开时跳过它而不是放弃整页——分类数据可能
    落在其他分片上。
    """
    parts: list[str] = []
    for match in _FLIGHT_CHUNK_RE.finditer(html):
        try:
            parts.append(json.loads(match.group(1)))
        except ValueError:
            continue
    return "".join(parts)


def _raw_categories(html: str) -> list[dict[str, Any]]:
    """从页面里取出扁平分类列表

    Raises:
        ScrapeParseError: 页面里没有 categoryList，或它不是非空数组
    """
    payload = _flight_payload(html)
    marker = payload.find('"categoryList":')
    if marker < 0:
        raise ScrapeParseError(
            "分类页中缺少 categoryList 数据；站点可能改版或返回了非预期页面"
        )

    start = payload.find("[", marker)
    if start < 0:
        raise ScrapeParseError("分类页的 categoryList 不是数组")

    try:
        raw, _ = json.JSONDecoder().raw_decode(payload[start:])
    except ValueError as exc:
        raise ScrapeParseError(f"分类页的 categoryList 解析失败：{exc}") from exc

    items = [item for item in raw if isinstance(item, dict) and item.get("id") is not None]
    if not items:
        raise ScrapeParseError("分类页的 categoryList 为空")
    return items


def _to_node(raw: dict[str, Any]) -> RakumaCategoryNode:
    category_id = str(raw.get("id"))
    return RakumaCategoryNode(
        category_id=category_id,
        name=str(raw.get("name") or ""),
        parent_id=str(raw.get("parentId") if raw.get("parentId") is not None else ROOT_PARENT_ID),
        is_leaf=not bool(raw.get("hasChild")),
        url=build_category_url(category_id),
    )


def _build_subtree(
    node: RakumaCategoryNode,
    children_of: dict[str, list[RakumaCategoryNode]],
) -> RakumaCategoryNode:
    """递归把下级分类挂到 children 上

    分类树只有三层，递归深度可控；节点在 categoryList 里 id 唯一，
    不会出现自环。
    """
    return node.model_copy(
        update={
            "children": [
                _build_subtree(child, children_of)
                for child in children_of.get(node.category_id, [])
            ]
        }
    )


def parse_categories(
    html: str, *, category_id: str | None, include_descendants: bool
) -> RakumaCategoryData:
    """解析分类树

    Args:
        category_id: 目标分类；None 表示取顶层分类列表
        include_descendants: children 里带上完整子树而非只有直接子级

    Raises:
        ScrapeParseError: 页面里没有分类树
        ItemNotFoundError: 目标分类不存在于站点分类树中
    """
    raw_items = _raw_categories(html)
    nodes = {str(raw["id"]): _to_node(raw) for raw in raw_items}

    children_of: dict[str, list[RakumaCategoryNode]] = {}
    for node in nodes.values():
        children_of.setdefault(node.parent_id, []).append(node)

    def resolve(node: RakumaCategoryNode) -> RakumaCategoryNode:
        return _build_subtree(node, children_of) if include_descendants else node

    root_children = children_of.get(str(ROOT_PARENT_ID), [])
    if category_id is None:
        return RakumaCategoryData(
            total_count=len(nodes),
            children=[resolve(node) for node in root_children],
        )

    target = nodes.get(category_id.strip())
    if target is None:
        raise ItemNotFoundError(f"分类树中未找到分类 {category_id}")

    ancestors: list[RakumaCategoryNode] = []
    parent = nodes.get(target.parent_id)
    while parent is not None:
        ancestors.insert(0, parent)
        parent = nodes.get(parent.parent_id)

    return RakumaCategoryData(
        category_id=target.category_id,
        name=target.name,
        full_name=" / ".join([node.name for node in ancestors] + [target.name]),
        is_leaf=target.is_leaf,
        url=target.url,
        total_count=len(nodes),
        ancestors=ancestors,
        children=[resolve(node) for node in children_of.get(target.category_id, [])],
    )