# dataflow-getcategory

**Repository Path**: lonlier/dataflow-getcategory

## Basic Information

- **Project Name**: dataflow-getcategory
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-07-13
- **Last Updated**: 2026-07-13

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# dataflow-getcategory

逐级抓取商城多级商品分类,整合成一棵统一的树形 `categories.json`,供下游程序按分类爬取商品列表。

## 特性

- 支持商城首页 HTML / API JSON 两种入口
- 用户以 Python 函数方式给出请求规则与提取规则,放于 `malls/<id>/rules/`,可反复编辑复用
- 每次 extract 返回**多级嵌套子树**(同一次响应里露出的层级),下钻时把子树嫁接成大一棵树
- 下钻交互循环:扫描全树所有"可下钻叶子" → 用户选样本 → 走通首分支 → 自动套用到同深度所有兄弟节点(跨顶层分支也复用)
- 不同分支层级数不同也兼容:短分支提前结束候选、长分支走到底后再询问/续生成更深一级规则
- 反爬优先级:随机 UA + Referer + 延时 + 重试 → `acw_sc__v2`/混淆 JS 挑战自动用 `py_mini_racer` 求解 cookie → 仍失败可选启用 `playwright` 浏览器兜底(`--allow-browser`)
- raw 按层级路径分目录保存:`raw/0_首页/1_手机/2_智能手机.html`
- 国内可用:依赖均走清华源;`py-mini-racer` 直接安装

## 环境

```powershell
uv sync
# (可选)启用浏览器兜底时:
$env:PLAYWRIGHT_DOWNLOAD_HOST="https://npmmirror.com/mirrors/playwright"
uv pip install playwright
uv run playwright install chromium
```

`pyproject.toml` 已配置:

```toml
[[tool.uv.index]]
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
default = true
```

## 使用

### 首次抓取

```powershell
uv run python -m getcategory https://www.example.com --id mymall
```

程序会:

1. 在 `malls/mymall/` 下生成 `config.yaml` 与 `level0_fetch.py` / `level0_extract.py` 模板
2. **段A**:提示编辑 `level0_fetch.py` → 回车 → 抓首页 → 保存 `raw/0_首页.html`(自动解 acw 挑战)
3. **段B**:提示打开 `raw/0_首页.html` 查看 → 据此编辑 `level0_extract.py` → 回车 → extract → 得顶层分类树
4. 进入下钻主循环(每轮两段式)

### 复用已有配置重跑(重新抓首页)

```powershell
uv run python -m getcategory --id mymall --resume
```

### 启用浏览器兜底

```powershell
uv run python -m getcategory --id mymall --allow-browser
```

### API 入口

```powershell
uv run python -m getcategory "https://api.example.com/categories" --id mymall --type json
```

然后在 `rules/level0_fetch.py` 中按需设置 `method=POST`、`headers`、`data` 等。

### 中断后从指定轮下钻继续

```powershell
uv run python -m getcategory --id mymall --resume-from 2
```

加载 `result/tree_state.json` 恢复整棵树,从第 2 轮下钻开始;0~1 轮的 raw/树保持不变,第 2 轮及之后的 raw 同名覆盖、规则文件可重新编辑。

## 目录结构

```
malls/<mall_id>/
├── config.yaml          商城元配置(起始URL、类型、UA池、延时、重试、代理、allow_browser)
├── rules/
│   ├── level0_fetch.py      第 0 轮(首页/API)请求规则
│   ├── level0_extract.py   第 0 轮提取规则
│   ├── level1_fetch.py      第 1 次下钻请求规则(按下钻轮次,非分类深度)
│   ├── level1_extract.py
│   ├── level2_*.py          ...每轮一个(程序交互时自动生成模板供编辑)
│   └── ...
├── raw/                     原始抓取文件,按层级路径分目录
│   ├── 0_首页.html
│   └── 0_首页/1_手机/2_智能手机.html
├── result/
│   ├── categories.json      最终整合后的树形结果
│   └── tree_state.json     每轮增量保存的树状态(供 --resume-from 使用)
└── meta.json                抓取日志、raw索引、自动套用失败记录、cookies
```

## 用户规则函数签名(固定)

> `N` 是【下钻轮次】(0=首页/API,1=第一次下钻,2=第二次...),与分类树深度**不对等**——一次 fetch 可能一次返回多级分类。

### `levelN_fetch.py`

```python
def build_request(ctx: dict) -> dict:
    """ctx = {
        'parent': Node | None,        # 上一级节点(首页轮次 N=0 时为 None)
        'parent_raw_path': str,        # 上级 raw 文件相对路径
        'cookies': dict,              # 会话 cookie
        'level': int,                 # 【分类深度】(与轮次 N 不一定相等;仅作参考)
    }
    返回: {
        'url': str,
        'method': 'GET' | 'POST',
        'headers': dict,
        'params': dict,
        'data': dict,
        'cookies': dict,
        'raw_type': 'html' | 'json',
    }
    """
```

### `levelN_extract.py`

```python
def extract(content, ctx: dict) -> list[dict]:
    """content: str(html) 或 dict|list(json)
    ctx = {
        'parent': Node | None,        # 首页轮次为 None;下钻轮次为被下钻的叶子
        'raw_type': 'html' | 'json',
        'raw_path': str,              # 本次响应已保存的相对路径
        'level': int,                 # 分类深度(仅作参考)
        'cookies': dict,
    }
    返回: 本次响应中可见的【完整多级子树】
    [
      {
        'name': str,          # 分类名,缺失填 ''
        'id': str,            # 分类ID,缺失填 ''
        'url': str,           # 该分类的商品列表页/可下钻页 URL
        'api': str,           # 该分类对应的商品列表 API(若有)
        'extra': dict,        # 任意附加键值
        'drillable': bool,    # True=本节点还有未露出的下级子分类(需再请求)
                              # False=本节点已是最底子分类(叶子),保留 url/api 给下游
                              # 缺省时:
                              #   children 非空(中间节点) → False
                              #   children 空(叶子) → 有 url/api 才 True,否则 False
        'children': [ ...递归同schema,叶子 children=[]... ],
      },
      ...
    ]

    提示:
    - extract 只负责"本次响应里能看到的层级",不必递归到需要再请求才能拿到的层
    - 一次响应若露出多级(如首页侧栏 0→1→2),请一并解析成嵌套树
    - 需要再请求才能看到更深层,把本次最底层节点的 drillable 设 True
    - 【重要】只返回"以 ctx.parent 为根的下一层(及本次可见的更下级)",
      不要把上级链路或父节点也当 child 返回!否则会触发环检测导致本轮失败
    - HTML 解析建议用 selectolax(已安装):
        from selectolax.parser import HTMLParser
        tree = HTMLParser(content)
    - JSON 直接 dict/list 访问即可
    """
```

## 下钻与整合机制

**核心概念:规则文件按【下钻请求轮次】编号,与分类树深度不对等。**

一次 fetch 可能一次返回多级分类(首页侧栏常露出 0→1→2 三级),所以同一次请求只对应一套 fetch/extract 规则,用"下钻轮次"编号:

- round 0 = 首页/API(规则文件 `level0_fetch.py` / `level0_extract.py`)
- 第 1 次选某叶子下钻 = round 1(规则文件 `level1_fetch.py` / `level1_extract.py`)
- 第 2 次下钻 = round 2,依此类推

**每轮分两段,中间暂停看 raw**:

```
round 0 (首页):
  段A  暂停:编辑 level0_fetch.py         → 回车 → fetch 首页 → 保存 raw/0_首页.html
  段B  暂停:看 raw 编 level0_extract.py  → 回车 → extract → 得顶层分类树

round N (下钻):
  询问是否继续下钻 → 选样本叶子
  段A  暂停:编辑 level{N}_fetch.py         → 回车 → fetch 样本 → 保存 raw
  段B  暂停:看 raw 编 level{N}_extract.py   → 回车 → extract → 环检测 → 嫁接到样本.children
       同分类深度兄弟用本轮规则自动 fetch+extract 套用 → 增量保存 tree_state.json
  回到"询问是否继续下钻"
```

为什么要分两段:fetch 之前你不知道响应内容,无法准确写 extract(更无法正确判断哪些节点 `drillable=True`)。分两段后你先看 raw 实际内容,再决定 extract 怎么写、哪些子节点还可继续下钻。

下钻主循环:

1. 扫描整棵树所有"可下钻叶子"(`drillable=True` 且 `children==[]`),按分类深度升序列出
2. 询问是否继续下钻;否 → 保存最终大树
3. 用户从候选选一个**样本叶子**
4. **段A**:生成 `level{round}_fetch.py` 模板 → 暂停编辑 → fetch 样本 → 保存 raw
5. **段B**:生成 `level{round}_extract.py` 模板 → 暂停看 raw 编辑 → extract → **环检测**(若返回子树里出现样本自身或其任一祖先,按 url/api/name 指纹判定,立刻拦截,指向 extract 文件让用户修) → 嫁接到样本 `children`
6. **同分类深度兄弟自动套用**:把本轮 fetch+extract 规则套到所有分类深度与样本相同、且仍可下钻的兄弟叶子(跨顶层分支也复用),各自动 fetch+extract+嫁接;失败节点写入 `meta.json`,不中断
7. 增量保存 `result/tree_state.json`;回到步骤 1
8. 当全树无任何可下钻叶子(或用户选"否")时,整合大树保存为 `result/categories.json`

不同分支层级数不同也兼容:

- 短分支的叶子 `drillable=False` → 自动退出候选
- 长分支走到更深时,在下一轮询问中选该深度样本,程序自动生成更深的 `level{round}_*.py` 模板(轮次号递增,与分类深度无关)

## 输出 `categories.json`

```json
{
  "mall_id": "mymall",
  "start_url": "...",
  "start_type": "html",
  "fetched_at": "2026-07-06T...",
  "tree": [
    {
      "name": "手机", "id": "1", "url": "...", "api": "", "extra": {},
      "drillable": false,
      "children": [
        { "name": "智能手机", "id": "1.1", "url": "...", "api": "", "extra": {},
          "drillable": false,
          "children": [ ... ] }
      ]
    }
  ],
  "tree_with_runtime": [ ... 同上但含 _depth/_raw_path/_parent_id 便于调试 ... ],
  "meta": { "raw_index": [...], "drill_log": [...], "auto_failures": [...], "cookies": {...} }
}
```

下游程序按 `drillable=false` 的节点取其 `url`(商品列表页)或 `api`(商品列表API)爬商品列表。

## 反爬策略

1. **请求层**:随机 UA(内置 5 条主流)+ 上级 URL 作 Referer + 1~3 秒随机延时 + 指数退避重试(2/4/8s,3 次)
2. **挑战检测**:响应含 `acw_sc__v2` + `arg1` 等标志 → 走 `acw_challenge.py`
   - 用 `py-mini-racer`(轻量 V8)执行该 JS,自动中立化反调试自检陷阱:
     - `String.prototype.match` 永不返回 null(避免 `k()`/`l()` 陷阱链 `null.toString()` 抛错)
     - `RegExp.prototype.test` 恒 true(让 `a0j` 自检 `dQvDam` 得 -1 避开 `Hbwxmd` 无限 push 致 OOM;主反调试三元走 `b()` 分支,其内部 try/catch 吞掉异常)
     - `Function.prototype.toString` 返回中性串(避免源码特征检测)
   - 从 `document.cookie` 提取 `acw_sc__v2`,注入 cookie 后重发(最多 2 轮)
   - **已用真实站点验证**:`www.tiangen.com`(阿里云盾 a0j/a0i 数组洗牌 + k/l/b 反调试),`test_acw.py` 与 `tests/sample_acw_tiangen.html` 为回归 fixture
3. **浏览器兜底**(可选):挑战仍失败且确认为真正 SPA 时,`--allow-browser` 启用 `playwright` headless 取渲染后 HTML
4. **meta 记录**:每节点抓取状态(`challenge_used`/`browser_used`)、自动套用失败原因

## 已知边界

- 同一深度但**结构差异大**的分支(例如 L2 在 A 分支是 HTML、在 B 分支是 JSON):单一 `level2_extract.py` 失败时仅记入 `meta.json`,可在该文件中按 `ctx['parent']` 信息分支判断;后续可扩展为 `level2_extract_alt.py` 备选规则。

## 自检

```powershell
uv run python smoke_test.py    # 端到端下钻流程(无网络)
uv run python test_acw.py      # acw_sc__v2 挑战求解(真实 tiangen fixture)
```

- `smoke_test.py` 在临时目录构造假商城(无网络),模拟多级下钻与同层自动套用全流程,验证 `categories.json` 与 raw 层级路径。
- `test_acw.py` 用 `tests/sample_acw_tiangen.html`(真实阿里云盾挑战页)验证 solver 能算出正确 `acw_sc__v2` cookie。