"""领券今日状态读写:弹窗频控(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 (
    CouponClaimEvent,
    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 session_app_env(db: Session, trace_id: str | None) -> str | None:
    """按 trace_id 取 coupon_session.app_env(每券成功率表打环境标用);无 trace_id / 查不到 → None。"""
    if not trace_id:
        return None
    return db.execute(
        select(CouponSession.app_env).where(CouponSession.trace_id == trace_id)
    ).scalar_one_or_none()


def record_claims(
    db: Session,
    device_id: str,
    user_id: int | None,
    trace_id: str | None,
    results: list[dict],
    app_env: str | None = None,
) -> int:
    """一批券领取结果同时写入每日资产表和逐次事件表。

    results 单项取自 pricebot 的 last_coupon_result / done.coupon_results,识别字段:
    coupon_id(必需)/ status(必需)/ name / vendor / reason /(display_count)。
    - CouponClaimRecord 按 (device, coupon_id, 今天) 幂等，供每日资产口径使用。
    - CouponClaimEvent 按 (trace_id, coupon_id) 幂等，供 admin 逐场统计使用。
    """
    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
            if app_env is not None:
                row.app_env = app_env
            row.extra = r
        else:
            db.add(CouponClaimRecord(
                device_id=device_id, user_id=user_id,
                coupon_id=coupon_id, claim_date=today,
                status=status, app_env=app_env,
                vendor=r.get("vendor"), coupon_name=r.get("name"),
                claimed_count=count, trace_id=trace_id, reason=r.get("reason"),
                extra=r,
            ))
        if trace_id:
            event = db.execute(
                select(CouponClaimEvent).where(
                    CouponClaimEvent.trace_id == trace_id,
                    CouponClaimEvent.coupon_id == coupon_id,
                )
            ).scalar_one_or_none()
            if event is not None:
                event.device_id = device_id
                event.status = status
                event.reason = r.get("reason")
                event.vendor = r.get("vendor")
                event.coupon_name = r.get("name")
                event.extra = r
                if user_id is not None:
                    event.user_id = user_id
                if count is not None:
                    event.claimed_count = count
                if app_env is not None:
                    event.app_env = app_env
            else:
                db.add(CouponClaimEvent(
                    trace_id=trace_id,
                    device_id=device_id,
                    user_id=user_id,
                    coupon_id=coupon_id,
                    claim_date=today,
                    status=status,
                    app_env=app_env,
                    vendor=r.get("vendor"),
                    coupon_name=r.get("name"),
                    claimed_count=count,
                    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_id → 平台;成功平台集)=====

# 成功语义:success + already_claimed 算成功(pricebot 代码 emit already_claimed,协议 enum 漏了);
# failed / skipped 不算。与 sum_claimed_count 同口径。
_SUCCESS_STATUSES = frozenset({"success", "already_claimed"})

# 三档平台 id 及固定序(美团→淘宝→京东),与客户端 DEFAULT_PLATFORM_ORDER 对齐。
DEFAULT_PLATFORMS: tuple[str, ...] = ("meituan-waimai", "taobao-shanguang", "jd-waimai")


def coupon_id_to_platform(coupon_id: str | None) -> str | None:
    """coupon_id 前缀 → 平台 id;无法识别 / 空 → None。

    与客户端 `CouponForegroundService.couponIdToPlatform` 同词表:
    mt_→美团外卖 / tb_·ele_·elm_→淘宝闪购 / jd_→京东外卖。
    """
    if not coupon_id:
        return None
    if coupon_id.startswith("mt_"):
        return "meituan-waimai"
    if coupon_id.startswith(("tb_", "ele_", "elm_")):
        return "taobao-shanguang"
    if coupon_id.startswith("jd_"):
        return "jd-waimai"
    return None


def succeeded_platforms(results: list[dict]) -> list[str]:
    """一批券结果 → 至少领到一张的平台集(按 DEFAULT_PLATFORMS 去重保序)。

    只取 status∈{success, already_claimed} 的券;失败/跳过、无法识别平台的券跳过。
    """
    ok: set[str] = set()
    for r in results:
        if r.get("status") in _SUCCESS_STATUSES:
            platform = coupon_id_to_platform(r.get("coupon_id"))
            if platform is not None:
                ok.add(platform)
    return [p for p in DEFAULT_PLATFORMS if p in ok]


# ===== 领券任务流水(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()


def merge_session_platform_success(
    db: Session, trace_id: str, platforms: list[str]
) -> None:
    """把本帧「成功平台」并入 coupon_session.platform_success(按 trace_id,并集幂等,按 DEFAULT_PLATFORMS 保序)。

    - 领券 /step 每逢带券结果的帧调一次(平台成败布尔,跨帧取并集天然幂等,不重复计)。
    - 读不到该 trace_id 的行 → **静默跳过**(不建兜底行;设计 §5:started 帧几乎必先落库)。
    - 并集无变化(该平台已记过)→ 不写库,省一次 UPDATE。
    - fire-and-forget:调用方已吞异常;并发唯一冲突回滚忽略。
    """
    if not platforms:
        return
    row = db.execute(
        select(CouponSession).where(CouponSession.trace_id == trace_id)
    ).scalar_one_or_none()
    if row is None:
        return
    merged = set(row.platform_success or []) | set(platforms)
    new_list = [p for p in DEFAULT_PLATFORMS if p in merged]
    if new_list == (row.platform_success or []):
        return  # 幂等:无新平台,不写
    row.platform_success = new_list
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
