shaguabijia-app-server/app/integrations/meituan.py

"""美团联盟 CPS 开放接口客户端。

签名方式：类阿里云网关 S-Ca 头，但 stringToSign **只有 4 段**——
  METHOD\n + Content-MD5\n + Headers + Url
没有 Accept / Content-Type / Date 行（加了就签名失败）。
"""
from __future__ import annotations

import base64
import hashlib
import hmac
import json
import logging
import time
from typing import Any

import httpx

from app.core.config import settings

logger = logging.getLogger("shagua.meituan")


class MeituanCpsError(Exception):
    """美团 CPS 接口调用失败。"""


def _content_md5(body: bytes) -> str:
    return base64.b64encode(hashlib.md5(body).digest()).decode()


def _sign(secret: str, method: str, content_md5: str, path: str,
          signed_headers: dict[str, str]) -> str:
    block = "".join(
        f"{k}:{signed_headers[k]}\n"
        for k in sorted(signed_headers)
    )
    sts = f"{method}\n{content_md5}\n{block}{path}"
    return base64.b64encode(
        hmac.new(secret.encode(), sts.encode(), hashlib.sha256).digest()
    ).decode()


def _call(path: str, body_obj: dict[str, Any]) -> dict[str, Any]:
    if not settings.MT_CPS_APP_KEY or not settings.MT_CPS_APP_SECRET:
        raise MeituanCpsError("MT_CPS_APP_KEY / MT_CPS_APP_SECRET not configured")

    body = json.dumps(body_obj, ensure_ascii=False).encode("utf-8")
    content_md5 = _content_md5(body)
    ts = str(int(time.time() * 1000))

    signed_headers = {"S-Ca-App": settings.MT_CPS_APP_KEY, "S-Ca-Timestamp": ts}
    signature = _sign(settings.MT_CPS_APP_SECRET, "POST", content_md5, path, signed_headers)

    headers = {
        "Content-Type": "application/json",
        "Content-MD5": content_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": signature,
    }

    url = f"{settings.MT_CPS_HOST}{path}"
    try:
        resp = httpx.post(url, content=body, headers=headers, timeout=settings.MT_CPS_TIMEOUT_SEC)
    except httpx.HTTPError as e:
        logger.exception("[MT] http error calling %s", url)
        raise MeituanCpsError(f"meituan http error: {e}") from e

    if resp.status_code != 200:
        logger.error("[MT] HTTP %s body=%s", resp.status_code, resp.text[:500])
        raise MeituanCpsError(f"meituan http {resp.status_code}")

    data = resp.json()
    if data.get("code") != 0:
        logger.error("[MT] api error code=%s message=%s", data.get("code"), data.get("message"))
        raise MeituanCpsError(f"code={data.get('code')} {data.get('message')}")

    return data


# ────────────────────── 业务方法 ──────────────────────

def query_coupon(
    *,
    longitude: float,
    latitude: float,
    platform: int = 1,
    biz_line: int | None = None,
    list_topic_id: int | None = 3,
    search_text: str | None = None,
    search_id: str | None = None,
    sort_field: int | None = None,
    page_no: int = 1,
    page_size: int = 20,
) -> dict[str, Any]:
    body: dict[str, Any] = {
        "platform": platform,
        "longitude": int(longitude * 1_000_000),
        "latitude": int(latitude * 1_000_000),
        "pageNo": page_no,
        "pageSize": page_size,
    }
    if biz_line is not None:
        body["bizLine"] = biz_line

    if search_text:
        body["searchText"] = search_text
        if sort_field is None:
            sort_field = 6
    elif list_topic_id is not None:
        body["listTopiId"] = list_topic_id

    if sort_field is not None:
        body["sortField"] = sort_field
    if search_id:
        body["searchId"] = search_id

    return _call("/cps_open/common/api/v1/query_coupon", body)


def get_referral_link(
    *,
    product_view_sign: str,
    platform: int = 1,
    biz_line: int | None = None,
    sid: str | None = None,
    link_type_list: list[int] | None = None,
) -> dict[str, Any]:
    body: dict[str, Any] = {"productViewSign": product_view_sign}
    if platform == 2:
        body["platform"] = 2
        if biz_line is not None:
            body["bizLine"] = biz_line

    body["sid"] = sid or settings.MT_CPS_DEFAULT_SID
    body["linkTypeList"] = link_type_list or [1, 3]

    return _call("/cps_open/common/api/v1/get_referral_link", body)