joy-cli/scripts/manhuang/validate-channels.py

#!/usr/bin/env python3
"""校验 configs/manhuang/channels.json（及本地覆盖）结构。

用法：
  python3 scripts/manhuang/validate-channels.py
  python3 scripts/manhuang/validate-channels.py --json
  python3 scripts/manhuang/validate-channels.py --path ~/.config/manhuang/channels.json
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[2]
REPO = ROOT / "configs" / "manhuang" / "channels.json"
LOCAL = Path.home() / ".config" / "manhuang" / "channels.json"
PRODUCT_TYPES = ROOT / "configs" / "manhuang" / "product-types.json"


def _load(path: Path) -> dict[str, Any]:
    data = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(data, dict):
        raise ValueError(f"{path}: root must be object")
    return data


def _pt_ids() -> set[str]:
    if not PRODUCT_TYPES.is_file():
        return set()
    try:
        doc = json.loads(PRODUCT_TYPES.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return set()
    types = doc.get("types") if isinstance(doc, dict) else None
    if not isinstance(types, list):
        return set()
    out: set[str] = set()
    for t in types:
        if isinstance(t, dict) and t.get("id"):
            out.add(str(t["id"]))
    return out


def validate(data: dict[str, Any], *, path: str) -> dict[str, Any]:
    errors: list[str] = []
    warnings: list[str] = []
    counts: dict[str, int] = {}
    known_pts = _pt_ids()

    if data.get("version") is None:
        warnings.append("missing version")

    def _list_section(key: str, *, require_base: bool = False) -> list[dict[str, Any]]:
        raw = data.get(key)
        if raw is None:
            warnings.append(f"missing section {key}")
            counts[key] = 0
            return []
        if not isinstance(raw, list):
            errors.append(f"{key} must be list")
            counts[key] = 0
            return []
        counts[key] = len(raw)
        rows: list[dict[str, Any]] = []
        seen: set[str] = set()
        for i, item in enumerate(raw):
            if not isinstance(item, dict):
                errors.append(f"{key}[{i}] not object")
                continue
            cid = str(item.get("id") or "").strip()
            if not cid:
                errors.append(f"{key}[{i}] missing id")
                continue
            if cid in seen:
                errors.append(f"{key}: duplicate id {cid}")
            seen.add(cid)
            base = str(item.get("base") or item.get("url") or "").strip()
            status = str(item.get("status") or "")
            # catalog_only / manual 聚合类允许无固定 base（发卡/TG 等）
            allow_no_base = status in ("catalog_only", "needs_key") or str(
                item.get("probe") or ""
            ) in ("manual", "rsshub_or_manual", "manual_watch")
            if require_base and not base and not allow_no_base:
                errors.append(f"{key}/{cid}: need base or url")
            elif require_base and not base and allow_no_base:
                warnings.append(f"{key}/{cid}: no base (catalog/manual — ok)")
            if key == "marketplaces" and status and status not in (
                "active",
                "watch_only",
                "catalog_only",
                "needs_key",
                "disabled",
            ):
                warnings.append(f"{key}/{cid}: unusual status={status}")
            for pt in item.get("product_types") or []:
                pts = str(pt)
                if known_pts and pts not in known_pts:
                    warnings.append(f"{key}/{cid}: unknown product_type {pts}")
            rows.append(item)
        return rows

    markets = _list_section("marketplaces", require_base=True)
    _list_section("sms_apis", require_base=False)
    _list_section("relay_price_feeds", require_base=False)
    _list_section("named_relay_stations", require_base=False)
    _list_section("free_llm_lists", require_base=False)

    demand = data.get("demand_sku_ids")
    if demand is None:
        warnings.append("missing demand_sku_ids")
        counts["demand_sku_ids"] = 0
    elif not isinstance(demand, list):
        errors.append("demand_sku_ids must be list")
        counts["demand_sku_ids"] = 0
    else:
        counts["demand_sku_ids"] = len(demand)
        for i, s in enumerate(demand):
            if not isinstance(s, str) or not s.strip():
                errors.append(f"demand_sku_ids[{i}] must be non-empty string")

    selfhost = data.get("selfhost")
    if selfhost is not None and not isinstance(selfhost, dict):
        errors.append("selfhost must be object")

    active = sum(1 for m in markets if m.get("status") == "active")
    counts["marketplaces_active"] = active

    ok = not errors
    return {
        "ok": ok,
        "path": path,
        "version": data.get("version"),
        "counts": counts,
        "errors": errors,
        "warnings": warnings,
        "error_n": len(errors),
        "warning_n": len(warnings),
    }


def main() -> int:
    ap = argparse.ArgumentParser(description="Validate manhuang channels.json")
    ap.add_argument("--path", type=Path, default=None, help="override path")
    ap.add_argument("--json", action="store_true", help="machine-readable output")
    args = ap.parse_args()

    path = args.path
    if path is None:
        path = LOCAL if LOCAL.is_file() else REPO
    if not path.is_file():
        msg = f"not found: {path}"
        if args.json:
            print(json.dumps({"ok": False, "errors": [msg]}, ensure_ascii=False, indent=2))
        else:
            print("FAIL", msg)
        return 2

    try:
        data = _load(path)
        result = validate(data, path=str(path))
    except (OSError, ValueError, json.JSONDecodeError) as e:
        if args.json:
            print(json.dumps({"ok": False, "path": str(path), "errors": [str(e)]}, ensure_ascii=False, indent=2))
        else:
            print("FAIL", path, e)
        return 2

    if args.json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print(f"path {result['path']}")
        print(f"version {result.get('version')} ok={result['ok']}")
        print("counts", result["counts"])
        if result["errors"]:
            print("errors:")
            for e in result["errors"]:
                print(" -", e)
        if result["warnings"]:
            print("warnings:")
            for w in result["warnings"][:30]:
                print(" -", w)
            if result["warning_n"] > 30:
                print(f" ... +{result['warning_n'] - 30} more")
        print("PASS" if result["ok"] else "FAIL")
    return 0 if result["ok"] else 1


if __name__ == "__main__":
    sys.exit(main())