diff --git a/tools/meituan_playground.py b/tools/meituan_playground.py new file mode 100644 index 0000000..0e9f165 --- /dev/null +++ b/tools/meituan_playground.py @@ -0,0 +1,494 @@ +"""美团 CPS API Playground —— 本地调参/试 API 的可视化工具。 + +浏览器填参数 → 本地后端用 meituan.py 的同款签名打美团 → 返回结果, +页面渲染「卡片列表 + 可折叠 JSON 树」。AppSecret 只留在后端,浏览器不接触。 + +跑法: + /Users/pure/miniconda3/envs/price/bin/python tools/meituan_playground.py +然后开 http://127.0.0.1:8799 + +只读工具(只调 query 类接口),复用本仓库 .env 里的 MT_CPS 凭证。 +""" +from __future__ import annotations + +import asyncio +import json +import os +import sys +import time + +# 让 `from app...` 可导入 + pydantic-settings 从仓库根读 .env +HERE = os.path.dirname(os.path.abspath(__file__)) +ROOT = os.path.dirname(HERE) +os.chdir(ROOT) +sys.path.insert(0, ROOT) + +import httpx # noqa: E402 +from fastapi import FastAPI, Request # noqa: E402 +from fastapi.responses import HTMLResponse, JSONResponse # noqa: E402 + +from app.core.config import settings # noqa: E402 +from app.integrations.meituan import _content_md5, _sign # noqa: E402 + +app = FastAPI(title="Meituan CPS Playground") + +DEFAULT_PATH = "/cps_open/common/api/v1/query_coupon" + + +def call_raw(path: str, body_obj: dict) -> dict: + """用 meituan.py 同款 S-Ca 签名打美团,原样返回(不管 code 是否为 0,方便看错误体)。""" + body = json.dumps(body_obj, ensure_ascii=False).encode("utf-8") + md5 = _content_md5(body) + ts = str(int(time.time() * 1000)) + signed = {"S-Ca-App": settings.MT_CPS_APP_KEY, "S-Ca-Timestamp": ts} + sig = _sign(settings.MT_CPS_APP_SECRET, "POST", md5, path, signed) + headers = { + "Content-Type": "application/json", + "Content-MD5": md5, + "S-Ca-App": settings.MT_CPS_APP_KEY, + "S-Ca-Timestamp": ts, + "S-Ca-Signature-Headers": "S-Ca-Timestamp,S-Ca-App", + "S-Ca-Signature": sig, + } + url = f"{settings.MT_CPS_HOST}{path}" + t0 = time.time() + resp = httpx.post(url, content=body, headers=headers, timeout=settings.MT_CPS_TIMEOUT_SEC) + ms = int((time.time() - t0) * 1000) + try: + j = resp.json() + except Exception: + j = {"_raw_text": resp.text[:2000]} + return {"http_status": resp.status_code, "elapsed_ms": ms, "json": j} + + +def _is_rate_limited(j: dict) -> bool: + code = str(j.get("code")) + msg = str(j.get("message") or j.get("msg") or "") + return code == "402" or "频繁" in msg + + +@app.post("/api/query") +async def api_query(req: Request): + try: + payload = await req.json() + except Exception as e: + return JSONResponse({"ok": False, "error": f"请求体不是合法 JSON: {e}"}) + path = (payload.get("path") or DEFAULT_PATH).strip() + body = payload.get("body") + if not isinstance(body, dict): + return JSONResponse({"ok": False, "error": "body 必须是 JSON 对象"}) + if not settings.MT_CPS_APP_KEY or not settings.MT_CPS_APP_SECRET: + return JSONResponse({"ok": False, "error": "MT_CPS_APP_KEY / SECRET 未配置(.env)"}) + + # 美团这接口很容易 402「调用频繁」,交互式工具自动退避重试 3 次,UX 顺一点 + last = None + for a in range(3): + try: + out = call_raw(path, body) + except Exception as e: + return JSONResponse({"ok": False, "error": f"{type(e).__name__}: {e}"}) + if _is_rate_limited(out.get("json") or {}) and a < 2: + last = out + await asyncio.sleep(1.5 * (a + 1)) + continue + return JSONResponse({"ok": True, **out, "retries": a}) + return JSONResponse({"ok": True, **(last or {}), "retries": 2}) + + +@app.get("/", response_class=HTMLResponse) +def index(): + return HTML + + +HTML = r""" + + +美团 CPS Playground + + +
+ 🍔 美团 CPS Playground + + 就绪 +
+
+
+

快捷模板

+
渠道:榜榜单  搜搜索词  供供给
+
到店 · 团购
+
+
到家 · 外卖
+
+

测试坐标(点一下填经纬度)

+
+

参数(改完点「用表单生成 body」)

+
+ + + + + + + + + + + + + + + + + + + + + + + + + + +
+
+

请求 body(可直接手改,发送以这里为准)

+ +
+
经纬度填十进制(116.404)自动 ×100万;翻页按钮在中间卡片栏右上角(供给/搜索有下一页,榜单没有)。
+
+ +
+

卡片列表

+
点「发送请求」后,这里渲染商品卡片
+
+ +
+

原始 JSON(点三角折叠/展开)

+
+ + +
+
原始响应在这里
+
+
+ + +""" + + +if __name__ == "__main__": + import uvicorn + + port = int(os.environ.get("MT_PLAYGROUND_PORT", "8799")) + print(f"\n 美团 CPS Playground → http://127.0.0.1:{port}\n (Ctrl-C 退出)\n") + uvicorn.run(app, host="127.0.0.1", port=port, log_level="warning")