shop-java/tools/grab_golden_batch.py

# -*- coding: utf-8 -*-
"""批量抓取线上 PHP 黄金样本（商家后台 read 类接口）。

注意 URL 形式：?s=/store/<controller>.<sub>/<action>
    控制器在 app/store/controller 下，如
      app/store/controller/Store.php        -> /store/store/info
      app/store/controller/store/Menu.php   -> /store/store.menu/list
      app/store/controller/goods/Service.php-> /store/goods.service/list

用法：
    python grab_golden_batch.py                # 抓取全部
    python grab_golden_batch.py /store/store/info   # 只抓指定接口
"""
import re
import sys
import json
import requests
from pathlib import Path
from _root import ROOT

GOLDEN = Path(str(ROOT / "golden"))
GOLDEN.mkdir(exist_ok=True)

BASE = "https://shop.qinys.cn/index.php?s="

TARGETS = [
    ("/store/store/info", "store_info", None),
    ("/store/store.menu/list", "store_menu_list", None),
    ("/store/store.role/list", "store_role_list", {"page": 1, "listRows": 15}),
    ("/store/store.user/list", "store_user_list", {"page": 1, "listRows": 15}),
    ("/store/store.address/all", "store_address_all", None),
    ("/store/category/list", "category_list", None),
    ("/store/goods.service/list", "goods_service_list", {"page": 1, "listRows": 15}),
    ("/store/goods.service/all", "goods_service_all", None),
    ("/store/goods.comment/list", "goods_comment_list", {"page": 1, "listRows": 15}),
    ("/store/files/list", "files_list", {"page": 1, "listRows": 15}),
    ("/store/files.group/list", "files_group_list", None),
    ("/store/region/all", "region_all", None),
    ("/store/region/tree", "region_tree", None),
    ("/store/user/list", "user_list", {"page": 1, "listRows": 15}),
    ("/store/user.grade/all", "user_grade_all", None),
    ("/store/order/list", "order_list", {"page": 1, "listRows": 15, "dataType": "all"}),
    ("/store/order.delivery/list", "order_delivery_list", {"page": 1, "listRows": 15}),
    ("/store/order.refund/list", "order_refund_list", {"page": 1, "listRows": 15, "dataType": "all"}),
    ("/store/setting/detail", "setting_detail", {"key": "store"}),
    ("/store/setting.cache/items", "setting_cache_items", None),
    ("/store/setting.delivery/all", "setting_delivery_all", None),
    ("/store/setting.express/all", "setting_express_all", None),
    ("/store/setting.payment/options", "setting_payment_options", None),
    ("/store/setting.printer/all", "setting_printer_all", None),
    ("/store/page/list", "page_list", {"page": 1, "listRows": 15}),
    ("/store/market.coupon/list", "market_coupon_list", {"page": 1, "listRows": 15}),
    ("/store/market.points/log", "market_points_log", {"page": 1, "listRows": 15}),
    ("/store/market.recharge.plan/list", "recharge_plan_list", {"page": 1, "listRows": 15}),
    ("/store/content.article/list", "article_list", {"page": 1, "listRows": 15}),
    ("/store/content.article.category/list", "article_category_list", None),
    ("/store/content.help/list", "help_list", {"page": 1, "listRows": 15}),
    ("/store/user.balance/log", "user_balance_log", {"page": 1, "listRows": 15}),
    ("/store/statistics.data/survey", "statistics_survey", None),
    ("/store/statistics.data/initialize", "statistics_initialize", None),
]


def main():
    src = Path(r"F:\import goods\upload_goods.py").read_text(encoding="utf-8")
    token = re.search(r'TOKEN\s*=\s*"([^"]+)"', src).group(1)
    headers = {"Access-Token": token}

    only = set(sys.argv[1:])
    summary = []
    for path, name, params in TARGETS:
        if only and path not in only:
            continue
        try:
            r = requests.get(BASE + path, headers=headers, params=params or {}, timeout=60)
            try:
                d = r.json()
            except Exception:
                print(f"[非JSON] {path} HTTP={r.status_code} body={r.text[:200]}")
                summary.append((path, "NON-JSON", 0))
                continue
        except Exception as e:
            print(f"[异常] {path} {e}")
            summary.append((path, "EXC", 0))
            continue

        if d.get("status") == 200:
            (GOLDEN / f"{name}.json").write_text(
                json.dumps(d, ensure_ascii=False, indent=2), encoding="utf-8")
        data = d.get("data")
        size = len(json.dumps(data, ensure_ascii=False)) if data is not None else 0
        print(f"{path:42s} status={d.get('status')} msg={str(d.get('message'))[:40]} size={size}")
        summary.append((path, d.get("status"), size))

    print("\n=== 汇总 ===")
    for p, s, n in summary:
        print(f"{s}\t{n}\t{p}")


if __name__ == "__main__":
    main()