"""领券「平台成功率」埋点:coupon_id→平台映射 / 成功平台推导 / session platform_success 并集。

设计:docs/guides/领券成功率指标-设计与埋点.md
"""
from __future__ import annotations

from datetime import UTC, date, datetime

from sqlalchemy import delete, func, select

from app.admin.repositories.coupon_data import coupon_data_report
from app.db.session import SessionLocal
from app.models.coupon_state import CouponSession
from app.repositories.coupon_state import (
    coupon_id_to_platform,
    merge_session_platform_success,
    succeeded_platforms,
)


def _agg_session(
    trace: str, platforms, platform_success, *, status: str = "completed"
) -> CouponSession:
    """构造一条聚合测试用 session(started_date 固定 2020-01-02、app_env=prod,不 commit)。"""
    return CouponSession(
        trace_id=trace,
        device_id="d-agg",
        status=status,
        app_env="prod",
        platforms=platforms,
        platform_success=platform_success,
        started_at=datetime(2020, 1, 2, tzinfo=UTC),
        started_date=date(2020, 1, 2),
    )


def _make_session(db, trace_id: str, **kw) -> CouponSession:
    row = CouponSession(
        trace_id=trace_id,
        device_id="dev-merge",
        status=kw.pop("status", "started"),
        started_at=datetime(2020, 1, 1, tzinfo=UTC),
        started_date=date(2020, 1, 1),
        **kw,
    )
    db.add(row)
    db.commit()
    return row


def test_coupon_id_to_platform_prefix_mapping() -> None:
    """coupon_id 前缀 → 三档平台 id(与客户端 couponIdToPlatform 同词表)。"""
    assert coupon_id_to_platform("mt_banjia_zhoumo") == "meituan-waimai"
    assert coupon_id_to_platform("mt_cps_waimai_redpacket") == "meituan-waimai"  # 大众点评CPS 也挂 mt_ → 归美团
    assert coupon_id_to_platform("tb_vip_shangou_voucher") == "taobao-shanguang"
    assert coupon_id_to_platform("ele_hongbao") == "taobao-shanguang"
    assert coupon_id_to_platform("elm_hongbao") == "taobao-shanguang"
    assert coupon_id_to_platform("jd_redpacket") == "jd-waimai"


def test_coupon_id_to_platform_unknown_returns_none() -> None:
    """无法识别的前缀 / 空 → None(调用方跳过,不计入平台)。"""
    assert coupon_id_to_platform("weird_xxx") is None
    assert coupon_id_to_platform("") is None
    assert coupon_id_to_platform(None) is None  # type: ignore[arg-type]


def test_succeeded_platforms_filters_status_and_dedups() -> None:
    """只取 status∈{success,already_claimed} 的券,映射平台后去重;失败/跳过/无法识别的不计。"""
    results = [
        {"coupon_id": "mt_a", "status": "success"},
        {"coupon_id": "mt_b", "status": "already_claimed"},  # 也算成功 → 仍是美团
        {"coupon_id": "mt_c", "status": "failed"},           # 不算
        {"coupon_id": "tb_a", "status": "success"},
        {"coupon_id": "jd_a", "status": "skipped"},          # 不算
        {"coupon_id": "weird", "status": "success"},         # 无法识别平台 → 跳过
    ]
    assert set(succeeded_platforms(results)) == {"meituan-waimai", "taobao-shanguang"}


def test_succeeded_platforms_empty() -> None:
    assert succeeded_platforms([]) == []


def test_merge_platform_success_union_idempotent() -> None:
    """按 trace_id 把成功平台并入 platform_success:并集去重 + 按 DEFAULT_PLATFORMS 保序 + 重复并入幂等。"""
    db = SessionLocal()
    trace = "merge-union-1"
    try:
        _make_session(db, trace)
        merge_session_platform_success(db, trace, ["meituan-waimai"])
        merge_session_platform_success(db, trace, ["jd-waimai", "meituan-waimai"])  # 并集 + 已有幂等
        db.expire_all()
        row = db.execute(
            select(CouponSession).where(CouponSession.trace_id == trace)
        ).scalar_one()
        # 美团→淘宝→京东固定序;只含实际成功的美团/京东
        assert row.platform_success == ["meituan-waimai", "jd-waimai"]
    finally:
        db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
        db.commit()
        db.close()


def test_merge_platform_success_missing_row_skips() -> None:
    """trace_id 无对应行 → 静默跳过:不建兜底行、不抛异常(设计 §5)。"""
    db = SessionLocal()
    try:
        merge_session_platform_success(db, "no-such-trace", ["meituan-waimai"])
        n = db.execute(
            select(func.count()).select_from(CouponSession).where(
                CouponSession.trace_id == "no-such-trace"
            )
        ).scalar_one()
        assert n == 0
    finally:
        db.close()


def test_coupon_data_success_rates() -> None:
    """admin 聚合 ②整单成功率 / ③点位成功率:平台粒度,含空 platforms→全领三档、abandoned 入基数。

    A 勾美团+淘宝、两个都成 → 整单成功;点位 2/2
    B 勾三档、只成美团       → 非整单;点位 1/3
    C 全领(platforms空→3)、三档全成 → 整单成功;点位 3/3
    D abandoned、勾美团、无成功平台 → 非整单;点位 0/1(仍进基数)
    发起数=4;整单成功=2 → 0.5;点位 6/9 → 0.6667
    """
    db = SessionLocal()
    try:
        db.add_all([
            _agg_session("agg-A", ["meituan-waimai", "taobao-shanguang"],
                         ["meituan-waimai", "taobao-shanguang"]),
            _agg_session("agg-B", ["meituan-waimai", "taobao-shanguang", "jd-waimai"],
                         ["meituan-waimai"]),
            _agg_session("agg-C", [], ["meituan-waimai", "taobao-shanguang", "jd-waimai"]),
            _agg_session("agg-D", ["meituan-waimai"], None, status="abandoned"),
        ])
        db.flush()  # 同会话可见,不 commit(finally 回滚保持隔离)
        s = coupon_data_report(
            db, date_from="2020-01-02", date_to="2020-01-02", app_env="prod"
        )["summary"]
        assert s["started_count"] == 4
        assert s["full_success_count"] == 2
        assert s["full_success_rate"] == 0.5
        assert s["point_success_count"] == 6
        assert s["point_total_count"] == 9
        assert s["point_success_rate"] == round(6 / 9, 4)
        # ③ 分平台点位成功率(§12):美团 3/4、淘宝 2/3、京东 1/2。
        # 分平台(成功,总)= 美团(3,4)+淘宝(2,3)+京东(1,2) = (6,9),
        # 其和正好等于上面已断言的 point_success_count=6 / point_total_count=9(和不变量)。
        assert s["per_platform"] == {
            "meituan-waimai": 0.75,
            "taobao-shanguang": round(2 / 3, 4),
            "jd-waimai": 0.5,
        }
    finally:
        db.rollback()
        db.close()


def test_coupon_data_success_rates_empty_range() -> None:
    """区间无 session → 发起数 0、两个率为 None(不除零)。"""
    db = SessionLocal()
    try:
        s = coupon_data_report(
            db, date_from="2019-01-01", date_to="2019-01-01", app_env="prod"
        )["summary"]
        assert s["started_count"] == 0
        assert s["full_success_rate"] is None
        assert s["point_success_rate"] is None
    finally:
        db.close()


def test_coupon_data_summary_schema_exposes_rates() -> None:
    """schema 契约:CouponDataSummary 暴露 ②③ 字段,键名与 repo 输出一致(router 直接 **summary 构造)。"""
    from app.admin.schemas.coupon_data import CouponDataSummary

    db = SessionLocal()
    try:
        db.add(_agg_session("agg-schema-1", ["meituan-waimai"], ["meituan-waimai"]))
        db.flush()
        summary = coupon_data_report(
            db, date_from="2020-01-02", date_to="2020-01-02", app_env="prod"
        )["summary"]
        dumped = CouponDataSummary(**summary).model_dump()
        assert dumped["full_success_rate"] == summary["full_success_rate"]
        assert dumped["point_success_rate"] == summary["point_success_rate"]
        assert dumped["full_success_count"] == summary["full_success_count"]
        assert dumped["point_success_count"] == summary["point_success_count"]
        assert dumped["point_total_count"] == summary["point_total_count"]
        assert dumped["per_platform"] == summary["per_platform"]
    finally:
        db.rollback()
        db.close()


def test_step_writes_platform_success_to_session(client) -> None:
    """/step 集成:pricebot 返 coupon_results → 本 trace 的 coupon_session.platform_success 落库(仅成功平台)。"""
    from unittest.mock import MagicMock, patch

    import httpx

    from app.models.coupon_state import CouponClaimRecord, CouponDailyCompletion

    trace = "int-trace-1"
    device = "dev-int-1"

    # 预置一条 started session(模拟 /session started 已先落库)
    db = SessionLocal()
    try:
        db.add(CouponSession(
            trace_id=trace, device_id=device, status="started", app_env="dev",
            platforms=["meituan-waimai", "taobao-shanguang"],
            started_at=datetime(2020, 1, 5, tzinfo=UTC),
            started_date=date(2020, 1, 5),
        ))
        db.commit()
    finally:
        db.close()

    fake_resp = {
        "success": True,
        "action": {"command": "done", "params": {
            "information": "已领 1 张",
            "coupon_results": [
                {"coupon_id": "mt_x", "name": "美团券", "vendor": "meituan_internal", "status": "success"},
                {"coupon_id": "tb_y", "name": "淘宝券", "vendor": "taobao", "status": "failed"},
            ],
        }},
        "continue": False,
    }

    async def fake_post(self, url, **kw):
        m = MagicMock()
        m.status_code = 200
        m.json = lambda: fake_resp
        return m

    body = {
        "device_id": device, "trace_id": trace, "step": 5,
        "screen_state": {"screen": {"width": 1080, "height": 2340, "density": 3.0},
                         "foreground": {"package": "x", "activity": ""}, "windows": []},
    }

    try:
        with patch.object(httpx.AsyncClient, "post", fake_post):
            r = client.post("/api/v1/coupon/step", json=body)
        assert r.status_code == 200, r.text

        db = SessionLocal()
        try:
            row = db.execute(
                select(CouponSession).where(CouponSession.trace_id == trace)
            ).scalar_one()
            assert row.platform_success == ["meituan-waimai"]  # tb 失败不计入
            claims = db.execute(
                select(CouponClaimRecord).where(CouponClaimRecord.device_id == device)
            ).scalars().all()
            assert claims and all(c.app_env == "dev" for c in claims)  # session app_env 打标
        finally:
            db.close()
    finally:
        db = SessionLocal()
        try:
            db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
            db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.device_id == device))
            db.execute(delete(CouponDailyCompletion).where(CouponDailyCompletion.device_id == device))
            db.commit()
        finally:
            db.close()


def test_coupon_data_status_filter() -> None:
    """状态多选过滤(方案 A):整个视图按选中状态算;None/空=全部。"""
    db = SessionLocal()
    try:
        db.add_all([
            _agg_session("st-A", ["meituan-waimai"], ["meituan-waimai"], status="completed"),
            _agg_session("st-B", ["meituan-waimai"], ["meituan-waimai"], status="started"),
            _agg_session("st-C", ["meituan-waimai"], None, status="failed"),
            _agg_session("st-D", ["meituan-waimai"], ["meituan-waimai"], status="abandoned"),
        ])
        db.flush()
        base = dict(date_from="2020-01-02", date_to="2020-01-02", app_env="prod")
        # None = 全部 4 发起
        assert coupon_data_report(db, **base)["summary"]["started_count"] == 4
        # 排除 started → 3 发起(整个视图,发起数也随之变)
        sub = coupon_data_report(db, **base, statuses=["completed", "failed", "abandoned"])["summary"]
        assert sub["started_count"] == 3
        assert sub["completed_count"] == 1
        # 只 completed → 发起数=1、整单成功率基数=1(该 completed 是整单成功)
        comp = coupon_data_report(db, **base, statuses=["completed"])["summary"]
        assert comp["started_count"] == 1
        assert comp["full_success_rate"] == 1.0
    finally:
        db.rollback()
        db.close()