shaguabijia-app-server/app/api/v1/coupon.py

"""领券业务透传端点。

把客户端的 POST /api/v1/coupon/step 请求原样转发给 pricebot-backend
的 /api/coupon/step。MVP 阶段**不鉴权**(见 docs/待办与技术债.md P1:
device_id 透传给 pricebot 区分设备,待补 JWT + device_id↔user_id 绑定后
才能做用户级画像)。

pricebot 协议文档:
  pricebot-backend/docs/projects/领券-客户端对接协议.md
"""
from __future__ import annotations

import json
import logging
from typing import Any

import httpx
from fastapi import APIRouter, HTTPException, Request, status
from fastapi.concurrency import run_in_threadpool

from app.api.deps import DbSession
from app.core.config import settings
from app.core.pricebot_router import pick_pricebot
from app.db.session import SessionLocal
from app.repositories import coupon_state as coupon_repo
from app.schemas.coupon_state import (
    CouponCompletedTodayOut,
    CouponPromptDismissIn,
    CouponPromptShouldShowOut,
)

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

router = APIRouter(prefix="/api/v1/coupon", tags=["coupon"])


def _to_int(v: object) -> int | None:
    """user_id 协议是字符串(登录态才带),转 int 存库;缺失/非法 → None。"""
    if v is None:
        return None
    try:
        return int(v)  # type: ignore[arg-type]
    except (TypeError, ValueError):
        return None


def _extract_coupon_results(resp_json: dict) -> list[dict]:
    """从 pricebot 响应抽本帧券结果,并**按 coupon_id 去重**。

    ⚠️ done/单券帧同时带 last_coupon_result(最后一张)+ action.params.coupon_results
    (全量含最后一张)→ 最后一张出现两次。必须去重:同批里同 coupon_id 两次,
    record_claims 在 autoflush=False 下两次 select 都查不到刚 add 的行 → 重复 add →
    commit 撞唯一约束 IntegrityError 回滚整批(done/单券记录全丢)。
    coupon_results(全量权威)覆盖 last_coupon_result。"""
    by_id: dict[str, dict] = {}
    last = resp_json.get("last_coupon_result")
    if isinstance(last, dict) and last.get("coupon_id"):
        by_id[last["coupon_id"]] = last
    params = (resp_json.get("action") or {}).get("params") or {}
    cr = params.get("coupon_results")
    if isinstance(cr, list):
        for x in cr:
            if isinstance(x, dict) and x.get("coupon_id"):
                by_id[x["coupon_id"]] = x
    return list(by_id.values())


def _mark_engagement_blocking(
    device_id: str, user_id: int | None, engage_type: str
) -> None:
    """独立 session 写 engagement(async 端点经 run_in_threadpool 调,不阻塞事件循环)。"""
    with SessionLocal() as db:
        coupon_repo.mark_engagement(db, device_id, user_id, engage_type)


def _record_claims_blocking(
    device_id: str, user_id: int | None, trace_id: str | None, results: list[dict]
) -> None:
    with SessionLocal() as db:
        coupon_repo.record_claims(db, device_id, user_id, trace_id, results)


def _mark_completed_blocking(
    device_id: str, user_id: int | None, trace_id: str | None
) -> None:
    """独立 session 写"今日已跑完整轮"(到 done 帧那刻调,best-effort 不阻塞领券)。"""
    with SessionLocal() as db:
        coupon_repo.mark_completed_today(db, device_id, user_id, trace_id)


@router.post("/step", summary="领券任务步进 (透传到 pricebot)")
async def coupon_step(
    request: Request,
) -> dict[str, Any]:
    """把请求体原样转发给 pricebot-backend 的 /api/coupon/step。

    - 鉴权: MVP 阶段不鉴权(待补 JWT,见 docs/待办与技术债.md P1)
    - 透传: 不做 schema 校验,pricebot 自己校验
    - 失败: 网络不可达 / pricebot 5xx → 502 + 友好 message
    """
    # 读原始字节,避免"反序列化→再序列化"的双重 JSON(省透传 CPU)。只 loads 一次拿
    # trace_id 做亲和 + 打日志,转发时直接发原始 bytes,不重新 dumps。
    raw = await request.body()
    try:
        meta = json.loads(raw)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"invalid json body: {e}") from e
    if not isinstance(meta, dict):
        meta = {}

    device_id = meta.get("device_id")
    user_id = _to_int(meta.get("user_id"))  # 登录态才带;判断不靠它,资产留痕用
    trace_id = meta.get("trace_id")

    # 领券任务首帧(step=0)= 用户已发起领券 → 记一条今日 engagement(claim_started),
    # 今天这台设备不再弹引导窗(对齐前台"点一键领取即 markEngaged")。写库失败绝不能
    # 连累领券主流程,整段吞掉。
    if device_id and meta.get("step") == 0:
        try:
            await run_in_threadpool(
                _mark_engagement_blocking, device_id, user_id, "claim_started"
            )
        except Exception as e:  # noqa: BLE001
            logger.warning("coupon engagement write failed: %s", e)

    # 按 trace_id 一致性 hash 选 pricebot 实例(同一领券任务的所有帧落同一进程)
    base = pick_pricebot(meta.get("trace_id"))
    url = f"{base.rstrip('/')}/api/coupon/step"
    timeout = settings.PRICEBOT_REQUEST_TIMEOUT_SEC

    logger.info(
        "coupon_step device_id=%s trace_id=%s step=%s",
        meta.get("device_id"),
        meta.get("trace_id"),
        meta.get("step"),
    )

    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            resp = await client.post(
                url, content=raw, headers={"Content-Type": "application/json"}
            )
    except httpx.RequestError as e:
        logger.error("[pricebot] request failed: %s", e)
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=f"pricebot upstream unreachable: {e}",
        ) from e

    if resp.status_code >= 500:
        logger.error(
            "[pricebot] 5xx status=%d body=%s",
            resp.status_code,
            resp.text[:500],
        )
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=f"pricebot upstream returned {resp.status_code}",
        )

    resp_json = resp.json()

    # 领券结果沉淀(资产):中间帧 last_coupon_result + done 帧 coupon_results 幂等写库。
    # 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
    if device_id:
        results = _extract_coupon_results(resp_json)
        if results:
            try:
                await run_in_threadpool(
                    _record_claims_blocking, device_id, user_id, trace_id, results
                )
            except Exception as e:  # noqa: BLE001
                logger.warning("coupon claim write failed: %s", e)

    # 整轮跑完(done 帧)→ 记一条"今日已完成",首页「去领取」卡据此置灰。
    # pricebot 把中途单券 done 改写成 wait+continue=true,只有整套全跑完那帧才保留
    # command=="done"(见 pricebot main.py),故 done 已等价"整轮完成"(用户决策 A 方案:
    # 到 done 即算,不管单券成败),无需再判 continue。写库失败绝不连累领券返回。
    action = resp_json.get("action") or {}
    if device_id and action.get("command") == "done":
        try:
            await run_in_threadpool(
                _mark_completed_blocking, device_id, user_id, trace_id
            )
        except Exception as e:  # noqa: BLE001
            logger.warning("coupon completion write failed: %s", e)

    return resp_json


@router.post("/prompt/dismiss", summary="用户拒绝/关闭领券引导窗(记今日已 engage)")
def coupon_prompt_dismiss(payload: CouponPromptDismissIn, db: DbSession) -> dict[str, bool]:
    """客户端点关闭引导窗时调用 → 记一条今日 engagement(dismissed),今天不再弹。

    server 在透传链路里看不到"用户拒绝"(拒绝不发起领券),故必须客户端通知。
    MVP 不鉴权,按 device_id 记。
    """
    coupon_repo.mark_engagement(db, payload.device_id, payload.user_id, "dismissed")
    return {"ok": True}


@router.get(
    "/prompt/should-show",
    response_model=CouponPromptShouldShowOut,
    summary="切到外卖 App 时是否还应弹领券引导窗",
)
def coupon_prompt_should_show(
    device_id: str, db: DbSession
) -> CouponPromptShouldShowOut:
    """今天这台设备已 engage(领或拒)过 → should_show=false。客户端据此决定弹不弹
    (纯后台判据,客户端不再做前台 SP 缓存判断)。"""
    return CouponPromptShouldShowOut(
        should_show=not coupon_repo.has_engaged_today(db, device_id)
    )


@router.post("/prompt/reset", summary="重置今日领券引导窗 engagement(开发测频控用)")
def coupon_prompt_reset(payload: CouponPromptDismissIn, db: DbSession) -> dict[str, bool]:
    """删这台设备今天的 engagement → has_engaged_today 变 false,今天又能弹。
    开发设置「重置今日领券弹窗状态」按钮调。MVP 不鉴权,按 device_id。"""
    coupon_repo.reset_today_engagement(db, payload.device_id)
    return {"ok": True}


@router.get(
    "/completed-today",
    response_model=CouponCompletedTodayOut,
    summary="这台设备今天是否已跑完整轮领券(到 done 帧)",
)
def coupon_completed_today(
    device_id: str, db: DbSession
) -> CouponCompletedTodayOut:
    """今天这台设备已跑完整轮(到 done)→ completed=true。客户端据此把首页「去领取」
    卡置灰、不可点(用户决策 A 方案:到 done 即算,不管单券成败)。判断维度 device_id
    必须与领券循环上报的一致(客户端两端都用 ANDROID_ID)。"""
    return CouponCompletedTodayOut(
        completed=coupon_repo.has_completed_today(db, device_id)
    )