"""美团 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() # trust_env=False: 美团是国内域名,强制直连绕开本机代理(代理会掐断 TLS 握手,报 SSL EOF) resp = httpx.post(url, content=body, headers=headers, timeout=settings.MT_CPS_TIMEOUT_SEC, trust_env=False) 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")