shaguabijia-app-server/app/repositories/coupon_state.py

"""领券今日状态读写:弹窗频控(engagement)+ 领券记录(claim)。

写操作按唯一键幂等 upsert,自带 commit + 并发 IntegrityError 兜底(对齐 price_observation)。
日期口径 = Asia/Shanghai 的自然日。
"""
from __future__ import annotations

import logging
from datetime import date, datetime, timezone
from zoneinfo import ZoneInfo

from sqlalchemy import delete, func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.models.coupon_state import (
    CouponClaimRecord,
    CouponDailyCompletion,
    CouponPromptEngagement,
    CouponSession,
)

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

_CN_TZ = ZoneInfo("Asia/Shanghai")


def today_cn() -> date:
    """Asia/Shanghai 的自然日(领券判断的"今天")。"""
    return datetime.now(_CN_TZ).date()


# ===== 弹窗频控(coupon_prompt_engagement)=====

def has_engaged_today(db: Session, device_id: str, package: str) -> bool:
    """这台设备今天**这个 App** 是否已对领券引导窗表达过意向(领/拒/弹出)。有 = 该 App 不再弹。
    频控按 (device, package, 日):美团弹过不影响淘宝/京东今天各自仍弹一次。"""
    row = db.execute(
        select(CouponPromptEngagement.id).where(
            CouponPromptEngagement.device_id == device_id,
            CouponPromptEngagement.package == package,
            CouponPromptEngagement.engage_date == today_cn(),
        )
    ).first()
    return row is not None


def mark_engagement(
    db: Session, device_id: str, package: str, user_id: int | None, engage_type: str
) -> None:
    """记今日意向(shown / claim_started / dismissed)。(device, package, 今天) 唯一,幂等 upsert。

    engage_type 升级口径(同一 (device,package,日) 多次调,只覆盖 type,不新增行):
    shown(自动弹出)→ claim_started(点一键领取)/ dismissed(点关闭)。判断只看"有没有这条"。
    """
    today = today_cn()
    row = db.execute(
        select(CouponPromptEngagement).where(
            CouponPromptEngagement.device_id == device_id,
            CouponPromptEngagement.package == package,
            CouponPromptEngagement.engage_date == today,
        )
    ).scalar_one_or_none()
    if row is not None:
        row.engage_type = engage_type
        if user_id is not None:
            row.user_id = user_id
    else:
        db.add(CouponPromptEngagement(
            device_id=device_id, package=package, user_id=user_id,
            engage_date=today, engage_type=engage_type,
        ))
    try:
        db.commit()
    except IntegrityError:
        # 并发下另一请求刚插了同 (device, package, 日) → 唯一约束撞,回滚忽略(本就幂等)。
        db.rollback()


def reset_today_engagement(db: Session, device_id: str) -> int:
    """删这台设备今天**所有 App** 的 engagement(开发设置「重置今日领券弹窗状态」调,测频控用)。
    删后各 App has_engaged_today → false,今天又都能弹。返回删除行数。
    (不按 package 过滤:重置是"把今天清干净从头测",清全部 App 最符合预期。)"""
    result = db.execute(
        delete(CouponPromptEngagement).where(
            CouponPromptEngagement.device_id == device_id,
            CouponPromptEngagement.engage_date == today_cn(),
        )
    )
    db.commit()
    return result.rowcount or 0


# ===== 今日跑完整轮(coupon_daily_completion)=====

def has_completed_today(db: Session, device_id: str) -> bool:
    """这台设备今天是否已跑完整轮领券(到 done 帧)。有 = 首页置灰、不能再领。"""
    row = db.execute(
        select(CouponDailyCompletion.id).where(
            CouponDailyCompletion.device_id == device_id,
            CouponDailyCompletion.complete_date == today_cn(),
        )
    ).first()
    return row is not None


def mark_completed_today(
    db: Session, device_id: str, user_id: int | None, trace_id: str | None = None
) -> None:
    """记今日已跑完整轮。(device, 今天) 唯一,幂等 upsert。到 done 即记,不管单券成败。"""
    today = today_cn()
    row = db.execute(
        select(CouponDailyCompletion).where(
            CouponDailyCompletion.device_id == device_id,
            CouponDailyCompletion.complete_date == today,
        )
    ).scalar_one_or_none()
    if row is not None:
        if user_id is not None:
            row.user_id = user_id
        if trace_id is not None:
            row.trace_id = trace_id
    else:
        db.add(CouponDailyCompletion(
            device_id=device_id, user_id=user_id,
            complete_date=today, trace_id=trace_id,
        ))
    try:
        db.commit()
    except IntegrityError:
        # 并发下另一请求刚插了同 (device, 日) → 唯一约束撞,回滚忽略(本就幂等)。
        db.rollback()


def reset_today_completion(db: Session, device_id: str) -> int:
    """删这台设备今天的"已完成"记录(开发设置「重置今日领券弹窗状态」全重置时调)。
    删后 has_completed_today → false,首页「去领取」卡恢复可点。返回删除行数。"""
    result = db.execute(
        delete(CouponDailyCompletion).where(
            CouponDailyCompletion.device_id == device_id,
            CouponDailyCompletion.complete_date == today_cn(),
        )
    )
    db.commit()
    return result.rowcount or 0


# ===== 领券记录(coupon_claim_record)=====

def record_claims(
    db: Session,
    device_id: str,
    user_id: int | None,
    trace_id: str | None,
    results: list[dict],
) -> int:
    """一批券领取结果幂等写入,返回写入(新增 + 更新)条数。

    results 单项取自 pricebot 的 last_coupon_result / done.coupon_results,识别字段:
    coupon_id(必需)/ status(必需)/ name / vendor / reason /(display_count)。
    (device, coupon_id, 今天) 唯一:重复上报同张券走更新(status 以最后一次为准)。
    """
    today = today_cn()
    written = 0
    seen: set[str] = set()  # 同批去重防御:autoflush=False 下同 coupon_id 重复会两次 add → 撞唯一约束回滚整批
    for r in results:
        coupon_id = r.get("coupon_id")
        status = r.get("status")
        if not coupon_id or not status or coupon_id in seen:
            continue  # 脏数据 / 同批重复跳过
        seen.add(coupon_id)
        count = r.get("display_count")
        if count is None:
            count = r.get("claimed_count")
        row = db.execute(
            select(CouponClaimRecord).where(
                CouponClaimRecord.device_id == device_id,
                CouponClaimRecord.coupon_id == coupon_id,
                CouponClaimRecord.claim_date == today,
            )
        ).scalar_one_or_none()
        if row is not None:
            row.status = status
            row.reason = r.get("reason")
            if user_id is not None:
                row.user_id = user_id
            if count is not None:
                row.claimed_count = count
            row.extra = r
        else:
            db.add(CouponClaimRecord(
                device_id=device_id, user_id=user_id,
                coupon_id=coupon_id, claim_date=today,
                status=status, vendor=r.get("vendor"), coupon_name=r.get("name"),
                claimed_count=count, trace_id=trace_id, reason=r.get("reason"),
                extra=r,
            ))
        written += 1
    if written == 0:
        return 0
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        logger.warning(
            "coupon_claim 并发幂等冲突 device=%s trace=%s,回滚", device_id, trace_id
        )
        return 0
    return written


# ===== 累计领券数(「我的」页战绩卡「领取优惠券 X 张」)=====

def sum_claimed_count(db: Session, user_id: int) -> int:
    """该用户累计领到的优惠券张数。口径(2026-06-15 用户定):SUM(claimed_count) ——
    各成功领券记录的 pricebot 展示张数(claimed_count 列,存的是 display_count)之和,
    与领券完成时给用户看的「本次领了 N 张」同源。

    - 只算 status ∈ {success, already_claimed}:already_claimed=今日已领过,协议里算「已领到」;
      failed / skipped 不计。
    - claimed_count 为 0 的保持 0:那是同 count_group 合并去重项(pricebot 只让一条出数),不重复计;
      为 NULL 的兜底成 1(成功领到至少 1 张;实际 pricebot to_dict 恒下发 display_count,NULL 基本不出现)。
    - 维度 user_id:登录态领的券才归入。登录前匿名领的(user_id 为空)不算(产品可接受)。
    """
    total = db.execute(
        select(
            func.coalesce(
                func.sum(func.coalesce(CouponClaimRecord.claimed_count, 1)), 0
            )
        ).where(
            CouponClaimRecord.user_id == user_id,
            CouponClaimRecord.status.in_(("success", "already_claimed")),
        )
    ).scalar_one()
    return int(total or 0)


# ===== 领券任务流水(coupon_session,admin「领券数据」看板数据源)=====

def upsert_coupon_session(
    db: Session,
    *,
    trace_id: str,
    device_id: str,
    status: str,
    started_at_ms: int,
    user_id: int | None = None,
    platforms: list[str] | None = None,
    origin_package: str | None = None,
    device_model: str | None = None,
    rom: str | None = None,
    app_env: str | None = None,
    elapsed_ms: int | None = None,
    platform_elapsed: dict | None = None,
    claimed_count: int | None = None,
    trace_url: str | None = None,
) -> None:
    """一条领券流水按 trace_id 幂等 upsert(发起 started 建行、收尾终态更新同一行)。

    - 乱序/重复兜底:终态(completed/failed/abandoned)先到也建行;started 重复到不覆盖已有终态
      (状态只前进,不降级)。
    - started_at 由客户端墙钟毫秒转;started_date 取其 Asia/Shanghai 自然日(admin 按天聚合/筛选)。
    - 终态帧补 finished_at=服务端 now;各字段非空才写(避免 started 帧的 None 抹掉收尾值,反之亦然)。
    并发 IntegrityError 回滚忽略(本就幂等)。
    """
    started_at = datetime.fromtimestamp(started_at_ms / 1000, tz=timezone.utc)
    started_date = started_at.astimezone(_CN_TZ).date()
    is_terminal = status in ("completed", "failed", "abandoned")

    row = db.execute(
        select(CouponSession).where(CouponSession.trace_id == trace_id)
    ).scalar_one_or_none()

    if row is None:
        db.add(CouponSession(
            trace_id=trace_id,
            device_id=device_id,
            user_id=user_id,
            status=status,
            app_env=app_env,
            platforms=platforms,
            origin_package=origin_package,
            device_model=device_model,
            rom=rom,
            started_at=started_at,
            started_date=started_date,
            finished_at=datetime.now(timezone.utc) if is_terminal else None,
            elapsed_ms=elapsed_ms,
            platform_elapsed=platform_elapsed,
            claimed_count=claimed_count,
            trace_url=trace_url,
        ))
    else:
        # 状态只前进:started 帧重复到(如 START_STICKY 重启)不把已有终态降级回 started。
        if not (status == "started" and row.status in ("completed", "failed", "abandoned")):
            row.status = status
        if is_terminal:
            row.finished_at = datetime.now(timezone.utc)
        if user_id is not None:
            row.user_id = user_id
        if platforms is not None:
            row.platforms = platforms
        if origin_package is not None:
            row.origin_package = origin_package
        if device_model is not None:
            row.device_model = device_model
        if rom is not None:
            row.rom = rom
        if app_env is not None:
            row.app_env = app_env
        if elapsed_ms is not None:
            row.elapsed_ms = elapsed_ms
        if platform_elapsed is not None:
            row.platform_elapsed = platform_elapsed
        if claimed_count is not None:
            row.claimed_count = claimed_count
        if trace_url is not None:
            row.trace_url = trace_url
    try:
        db.commit()
    except IntegrityError:
        # 并发下另一请求刚插了同 trace_id → 唯一约束撞,回滚忽略(本就幂等)。
        db.rollback()