joy-cli/scripts/manhuang/build-guishi-handbooks.py

#!/usr/bin/env python3
"""游侠/灰产手册 → 鬼市 C2C 结构化索引。

读候选路径（存在才入）：
  docs/intelligence/xianyu-*.md
  ~/.cache 或 scout 同步的 handbooks
写出：~/.cache/manhuang/guishi/handbooks.json
"""
from __future__ import annotations

import json
import re
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

def _out_path() -> Path:
    home = Path.home()
    if (home / "data" / "manhuang").is_dir():
        return home / "data" / "manhuang" / "guishi" / "handbooks.json"
    return home / ".cache" / "manhuang" / "guishi" / "handbooks.json"


OUT = _out_path()

CANDIDATES = [
    Path.home() / "docs/intelligence/xianyu-data-trading-handbook.md",
    Path.home() / "docs/intelligence/xianyu-hidden-market-handbook.md",
    Path.home() / "data/manhuang/handbooks/xianyu-data-trading-handbook.md",
    Path.home() / "data/manhuang/handbooks/xianyu-hidden-market-handbook.md",
    Path("/Users/feibisi-studio/Projects/linuxjoy/docs/intelligence/xianyu-data-trading-handbook.md"),
    Path("/Users/feibisi-studio/Projects/linuxjoy/docs/intelligence/xianyu-hidden-market-handbook.md"),
    Path("/Users/feibisi-studio/Projects/joy-cli/docs/intelligence/xianyu-data-trading-handbook.md"),
    Path("/Users/feibisi-studio/Projects/joy-cli/docs/intelligence/xianyu-hidden-market-handbook.md"),
]


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _parse_md(path: Path) -> dict[str, Any] | None:
    if not path.is_file():
        return None
    try:
        text = path.read_text(encoding="utf-8", errors="ignore")
    except OSError:
        return None
    lines = text.splitlines()
    title = (lines[0].lstrip("# ").strip() if lines else path.name)[:120]
    secs = [ln[3:].strip() for ln in lines if ln.startswith("## ")]
    # 粗抽「平台/渠道」线索
    platforms = sorted(set(re.findall(
        r"(闲鱼|淘宝|拼多多|Telegram|TG|微信|QQ|发卡|接码|号商|担保|中介)",
        text,
    )))
    kind = "c2c" if re.search(r"闲鱼|C2C|隐蔽|灰产", title + text[:400], re.I) else "handbook"
    return {
        "title": title,
        "path": str(path),
        "kind": kind,
        "section_n": len(secs),
        "sections": secs[:16],
        "platforms": platforms[:20],
        "chars": len(text),
        "mtime": path.stat().st_mtime,
        "age_s": round(max(0.0, time.time() - path.stat().st_mtime), 1),
        "market_layer": "guishi",
        "sku_for": ["account", "otp", "c2c_deal"],
        "next": "手册线索 → 鬼市 C2C 卡；成交回黑市担保场或本地跟进",
    }


def build() -> dict[str, Any]:
    items = []
    seen = set()
    for p in CANDIDATES:
        key = str(p.resolve()) if p.exists() else str(p)
        if key in seen:
            continue
        row = _parse_md(p)
        if not row:
            continue
        seen.add(key)
        items.append(row)
    # 也并入 tab-caches handbooks_index 灰产项
    tc = Path.home() / ".cache" / "manhuang" / "tab-caches.json"
    if tc.is_file():
        try:
            data = json.loads(tc.read_text(encoding="utf-8"))
            for h in ((data.get("handbooks_index") or {}).get("items") or []):
                if not isinstance(h, dict):
                    continue
                if h.get("kind") not in ("gray", "c2c") and not re.search(
                    r"闲鱼|灰|隐蔽|c2c", str(h.get("title") or ""), re.I
                ):
                    continue
                path = str(h.get("path") or h.get("title"))
                if path in seen:
                    continue
                seen.add(path)
                items.append({
                    "title": h.get("title"),
                    "path": h.get("path"),
                    "kind": "c2c",
                    "section_n": h.get("section_n") or len(h.get("sections") or []),
                    "sections": (h.get("sections") or [])[:12],
                    "platforms": [],
                    "market_layer": "guishi",
                    "source": "tab_caches",
                    "sku_for": ["account", "c2c_deal"],
                    "next": "游侠手册 → 鬼市",
                })
        except (OSError, json.JSONDecodeError):
            pass

    out = {
        "ts": _now(),
        "ok": bool(items),
        "n": len(items),
        "items": items,
        "note": "鬼市 C2C / 游侠手册索引",
    }
    OUT.parent.mkdir(parents=True, exist_ok=True)
    OUT.write_text(json.dumps(out, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return out


def main() -> int:
    r = build()
    print("wrote", OUT, "n=", r.get("n"))
    for it in r.get("items") or []:
        print("-", (it.get("title") or "")[:50], "secs", it.get("section_n"))
    return 0 if r.get("ok") else 1


if __name__ == "__main__":
    raise SystemExit(main())