feat: 接入美团联盟 CPS 优惠券推荐模块

- 新增 app/core/meituan.py: 美团联盟 API 客户端 (S-Ca 签名, query_coupon, get_referral_link) - 新增 app/api/v1/meituan.py: /api/v1/meituan/coupons 和 /feed 路由, 支持按经纬度推荐 - 新增 app/schemas/meituan.py: 请求/响应 Pydantic Schema - app/main.py: 注册 meituan_router - app/core/config.py: 新增 MT_CPS_* 配置项, .env 路径改为绝对路径避免 CWD 问题 - .env.example: 补充美团联盟 CPS 配置模板 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-26 20:29:31 +08:00
parent 0d2a3e9d7a
commit cc9d9f03f7
6 changed files with 442 additions and 1 deletions
@@ -6,15 +6,18 @@
 from __future__ import annotations

 from functools import lru_cache
+from pathlib import Path
 from typing import Literal

 from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict

+_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+

 class Settings(BaseSettings):
    model_config = SettingsConfigDict(
-        env_file=".env",
+        env_file=str(_PROJECT_ROOT / ".env"),
        env_file_encoding="utf-8",
        case_sensitive=False,
        extra="ignore",
@@ -46,6 +49,13 @@ class Settings(BaseSettings):
    SMS_CODE_TTL_SEC: int = 300
    SMS_SEND_INTERVAL_SEC: int = 60

+    # ===== 美团联盟 CPS =====
+    MT_CPS_APP_KEY: str = ""
+    MT_CPS_APP_SECRET: str = ""
+    MT_CPS_HOST: str = "https://media.meituan.com"
+    MT_CPS_TIMEOUT_SEC: int = 15
+    MT_CPS_DEFAULT_SID: str = "sgbjia"
+
    # ===== CORS =====
    CORS_ALLOW_ORIGINS: str = ""

@@ -0,0 +1,140 @@
+"""美团联盟 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)