"""Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
from __future__ import annotations

from datetime import UTC, datetime

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import event

from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.admin.repositories import queries
from app.db.session import SessionLocal, engine
from app.models.comparison import ComparisonRecord
from app.models.feedback import Feedback
from app.models.invite import InviteRelation
from app.models.savings import SavingsRecord
from app.models.user import User
from app.models.wallet import CashTransaction, WithdrawOrder
from app.repositories import user as user_repo
from app.repositories import wallet as wallet_repo


@pytest.fixture()
def admin_client() -> TestClient:
    return TestClient(admin_app)


@pytest.fixture()
def admin_token() -> str:
    db = SessionLocal()
    try:
        if admin_repo.get_by_username(db, "m2_admin") is None:
            admin_repo.create_admin(db, username="m2_admin", password="m2-pass", role="super_admin")
    finally:
        db.close()
    c = TestClient(admin_app)
    r = c.post("/admin/api/auth/login", json={"username": "m2_admin", "password": "m2-pass"})
    return r.json()["access_token"]


def _auth(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


def _seed_user_with_data(phone: str) -> int:
    """造一个用户 + 金币流水 + 现金流水 + 提现单 + 反馈,返回 user_id。"""
    db = SessionLocal()
    try:
        user = user_repo.upsert_user_for_login(db, phone=phone, register_channel="sms")
        uid = user.id
        wallet_repo.grant_coins(db, uid, 5000, biz_type="signin", remark="测试发金币")
        db.commit()
        db.add(CashTransaction(
            user_id=uid, amount_cents=-100, balance_after_cents=0, biz_type="withdraw", remark="t"
        ))
        db.add(WithdrawOrder(
            user_id=uid, out_bill_no=f"test{uid}billno0001", amount_cents=100, status="success"
        ))
        db.add(Feedback(user_id=uid, content="测试反馈内容", contact="wx_test", status="pending"))
        db.commit()
        return uid
    finally:
        db.close()


def test_dashboard_overview(admin_client: TestClient, admin_token: str) -> None:
    _seed_user_with_data("13800000001")
    r = admin_client.get("/admin/api/stats/overview", headers=_auth(admin_token))
    assert r.status_code == 200, r.text
    data = r.json()
    assert data["users"]["total"] >= 1
    assert data["coins"]["granted_total"] >= 5000
    assert "success_rate" in data["comparison"]
    assert data["cps"]["available"] is True
    assert "meituan_order_count" in data["cps"]
    assert "jd_order_count" in data["cps"]


def test_dashboard_period_comparison_is_aggregated_by_backend(
    admin_client: TestClient, admin_token: str
) -> None:
    created_at = datetime(2037, 1, 15, 12)
    rows = [
        ("dashboard-aggregate-success", "success", 101, 0.1),
        ("dashboard-aggregate-failed", "failed", 200, 0.2),
        ("dashboard-aggregate-cancelled", "cancelled", 300, 0.3),
        ("dashboard-aggregate-running", "running", 400, 0.4),
    ]
    db = SessionLocal()
    try:
        for trace_id, status, total_ms, llm_cost_yuan in rows:
            db.add(
                ComparisonRecord(
                    trace_id=trace_id,
                    status=status,
                    total_ms=total_ms,
                    llm_cost_yuan=llm_cost_yuan,
                    created_at=created_at,
                )
            )
        db.commit()
    finally:
        db.close()

    response = admin_client.get(
        "/admin/api/stats/overview",
        params={"date_from": "2037-01-15", "date_to": "2037-01-15"},
        headers=_auth(admin_token),
    )
    assert response.status_code == 200, response.text
    comparison = response.json()["period"]["comparison"]
    assert comparison["total"] == 4
    assert comparison["completed"] == 2
    assert comparison["cancelled"] == 1
    assert comparison["success"] == 1
    assert comparison["success_rate"] == 0.3333
    assert comparison["median_duration_ms"] == 151
    assert comparison["p95_duration_ms"] == 195
    assert comparison["token_cost_total_yuan"] == pytest.approx(1.0)


def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:
    uid = _seed_user_with_data("13800000002")
    r = admin_client.get("/admin/api/users", headers=_auth(admin_token))
    assert r.status_code == 200, r.text
    assert "items" in r.json()

    r = admin_client.get(f"/admin/api/users/{uid}", headers=_auth(admin_token))
    assert r.status_code == 200, r.text
    d = r.json()
    assert d["user"]["id"] == uid
    assert d["coin_balance"] == 5000
    assert d["withdraw_total"] >= 1
    assert d["feedback_total"] >= 1

    assert admin_client.get("/admin/api/users/999999", headers=_auth(admin_token)).status_code == 404


def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
    admin_client: TestClient, admin_token: str
) -> None:
    """旧库缺少无关新列时，提现详情的统计和金币记录仍应可读。"""
    uid = _seed_user_with_data("13800000022")

    def reject_full_ad_reward_projection(
        _conn, _cursor, statement: str, _parameters, _context, _executemany
    ) -> None:
        if "ad_reward_record.boost_round_id" in statement:
            raise AssertionError("提现详情不应查询未使用的 boost_round_id")
        if "FROM user" in statement and "user.phone" in statement:
            raise AssertionError("奖励统计的用户存在性检查不应展开完整 user 表")

    event.listen(engine, "before_cursor_execute", reject_full_ad_reward_projection)
    try:
        stats = admin_client.get(
            f"/admin/api/users/{uid}/reward-stats", headers=_auth(admin_token)
        )
        records = admin_client.get(
            f"/admin/api/users/{uid}/coin-records",
            params={"limit": 10, "cursor": 0},
            headers=_auth(admin_token),
        )
    finally:
        event.remove(engine, "before_cursor_execute", reject_full_ad_reward_projection)

    assert stats.status_code == 200, stats.text
    assert records.status_code == 200, records.text


def test_user_reward_stats_can_scope_withdrawals_by_account(
    admin_client: TestClient, admin_token: str
) -> None:
    uid = _seed_user_with_data("13800000023")
    db = SessionLocal()
    try:
        account = wallet_repo.get_or_create_account(db, uid)
        account.cash_balance_cents = 123
        account.invite_cash_balance_cents = 456
        db.add_all(
            [
                WithdrawOrder(
                    user_id=uid,
                    out_bill_no="rewardstatsinvite00000001",
                    amount_cents=250,
                    source="invite_cash",
                    status="success",
                ),
                WithdrawOrder(
                    user_id=uid,
                    out_bill_no="rewardstatsinvite00000002",
                    amount_cents=50,
                    source="invite_cash",
                    status="reviewing",
                ),
            ]
        )
        db.commit()
    finally:
        db.close()

    coin = admin_client.get(
        f"/admin/api/users/{uid}/reward-stats",
        params={"withdraw_source": "coin_cash"},
        headers=_auth(admin_token),
    )
    invite = admin_client.get(
        f"/admin/api/users/{uid}/reward-stats",
        params={"withdraw_source": "invite_cash"},
        headers=_auth(admin_token),
    )
    assert coin.status_code == 200, coin.text
    assert invite.status_code == 200, invite.text
    assert coin.json()["withdraw_success_cents"] == 100
    assert coin.json()["withdraw_total"] == 1
    assert coin.json()["cash_balance_cents"] == 123
    assert invite.json()["withdraw_success_cents"] == 250
    assert invite.json()["withdraw_total"] == 2
    assert invite.json()["cash_balance_cents"] == 456


def test_user_reward_stats_draw_ecpm_uses_all_filtered_impressions(
    admin_client: TestClient, admin_token: str
) -> None:
    """Draw 平均 eCPM 应与收益报表一致，不能只平均成功发奖记录。"""
    from app.models.ad_ecpm import AdEcpmRecord
    from app.models.ad_feed_reward import AdFeedRewardRecord

    uid = _seed_user_with_data("13800000024")
    created_at = datetime(2038, 1, 15, 4, tzinfo=UTC)
    db = SessionLocal()
    try:
        db.add_all(
            [
                AdFeedRewardRecord(
                    client_event_id="reward-stats-granted-high",
                    ad_session_id="reward-stats-granted-high",
                    user_id=uid,
                    reward_date="2038-01-15",
                    duration_seconds=10,
                    unit_count=1,
                    ecpm_raw="9000",
                    ad_type="draw",
                    feed_scene="coupon",
                    app_env="prod",
                    our_code_id="104098712",
                    coin=9,
                    status="granted",
                    created_at=created_at,
                ),
                AdEcpmRecord(
                    user_id=uid,
                    ad_type="draw",
                    feed_scene="coupon",
                    ad_session_id="reward-stats-impression-low",
                    app_env="prod",
                    our_code_id="104098712",
                    ecpm_raw="1000",
                    report_date="2038-01-15",
                    created_at=created_at,
                ),
                AdEcpmRecord(
                    user_id=uid,
                    ad_type="feed",
                    feed_scene="coupon",
                    ad_session_id="reward-stats-impression-mid",
                    app_env="prod",
                    our_code_id="104098712",
                    ecpm_raw="3000",
                    report_date="2038-01-15",
                    created_at=created_at,
                ),
                # 同用户但不同场景/环境/非业务代码位，均不应进入本次详情筛选。
                AdEcpmRecord(
                    user_id=uid,
                    ad_type="draw",
                    feed_scene="comparison",
                    ad_session_id="reward-stats-other-scene",
                    app_env="prod",
                    our_code_id="104098712",
                    ecpm_raw="7000",
                    report_date="2038-01-15",
                    created_at=created_at,
                ),
                AdEcpmRecord(
                    user_id=uid,
                    ad_type="draw",
                    feed_scene="coupon",
                    ad_session_id="reward-stats-test-env",
                    app_env="test",
                    our_code_id="104127529",
                    ecpm_raw="8000",
                    report_date="2038-01-15",
                    created_at=created_at,
                ),
                AdEcpmRecord(
                    user_id=uid,
                    ad_type="draw",
                    feed_scene="coupon",
                    ad_session_id="reward-stats-non-business",
                    app_env="prod",
                    our_code_id="demo-slot",
                    ecpm_raw="9000",
                    report_date="2038-01-15",
                    created_at=created_at,
                ),
            ]
        )
        db.commit()
    finally:
        db.close()

    response = admin_client.get(
        f"/admin/api/users/{uid}/reward-stats",
        params={
            "date_from": "2038-01-15T00:00:00Z",
            "date_to": "2038-01-15T23:59:59Z",
            "app_env": "prod",
            "revenue_scope": "business",
            "feed_scene": "coupon",
        },
        headers=_auth(admin_token),
    )
    assert response.status_code == 200, response.text
    data = response.json()
    assert data["feed_count"] == 1
    # 全部真实展示 (1000 + 3000) / 2；不能返回成功发奖记录的 9000。
    assert data["feed_avg_ecpm"] == 2000.0


def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
    """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。"""
    naive = datetime(2038, 1, 1, 8, 0)
    aware = datetime(2038, 1, 1, 7, 0, tzinfo=UTC)

    rows = [{"created_at": aware}, {"created_at": naive}]
    rows.sort(key=queries._coin_record_sort_key, reverse=True)

    assert rows == [{"created_at": naive}, {"created_at": aware}]


def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> None:
    _seed_user_with_data("13800000003")
    r = admin_client.get("/admin/api/users", params={"status": "active"}, headers=_auth(admin_token))
    assert r.status_code == 200
    assert all(u["status"] == "active" for u in r.json()["items"])


def test_user_list_sort_filter_range(admin_client: TestClient, admin_token: str) -> None:
    """用户列表:渠道/昵称筛选 + id 升降序 + 时间范围 + 非法 sort_by 422。"""
    db = SessionLocal()
    try:
        u1 = user_repo.upsert_user_for_login(db, phone="13811110001", register_channel="sms")
        u1.nickname = "排序测试甲"
        u2 = user_repo.upsert_user_for_login(db, phone="13811110002", register_channel="wechat")
        u2.nickname = "排序测试乙"
        db.commit()
        id1, id2 = u1.id, u2.id
    finally:
        db.close()

    # 渠道精确筛选
    r = admin_client.get(
        "/admin/api/users", params={"register_channel": "wechat"}, headers=_auth(admin_token)
    )
    assert r.status_code == 200, r.text
    assert all(u["register_channel"] == "wechat" for u in r.json()["items"])

    # 昵称模糊筛选
    r = admin_client.get(
        "/admin/api/users", params={"nickname": "排序测试", "limit": 100}, headers=_auth(admin_token)
    )
    ids = {u["id"] for u in r.json()["items"]}
    assert id1 in ids and id2 in ids

    # id 升序 / 降序
    asc_ids = [
        u["id"]
        for u in admin_client.get(
            "/admin/api/users",
            params={"sort_by": "id", "sort_order": "asc", "limit": 100},
            headers=_auth(admin_token),
        ).json()["items"]
    ]
    assert asc_ids == sorted(asc_ids)
    desc_ids = [
        u["id"]
        for u in admin_client.get(
            "/admin/api/users",
            params={"sort_by": "id", "sort_order": "desc", "limit": 100},
            headers=_auth(admin_token),
        ).json()["items"]
    ]
    assert desc_ids == sorted(desc_ids, reverse=True)

    # 时间范围:2000 年之前无人;之后有人(验证范围条件确实生效)
    assert (
        admin_client.get(
            "/admin/api/users", params={"created_to": "2000-01-01T00:00:00Z"},
            headers=_auth(admin_token),
        ).json()["items"]
        == []
    )
    assert (
        len(
            admin_client.get(
                "/admin/api/users", params={"created_from": "2000-01-01T00:00:00Z", "limit": 100},
                headers=_auth(admin_token),
            ).json()["items"]
        )
        >= 1
    )

    # 非法 sort_by → 422
    assert (
        admin_client.get(
            "/admin/api/users", params={"sort_by": "phone"}, headers=_auth(admin_token)
        ).status_code
        == 422
    )


def test_wallet_and_withdraw_lists(admin_client: TestClient, admin_token: str) -> None:
    uid = _seed_user_with_data("13800000004")
    r = admin_client.get(
        "/admin/api/wallet/coin-transactions", params={"user_id": uid}, headers=_auth(admin_token)
    )
    assert r.status_code == 200
    assert len(r.json()["items"]) >= 1

    r = admin_client.get("/admin/api/withdraws", params={"user_id": uid}, headers=_auth(admin_token))
    assert r.status_code == 200
    assert len(r.json()["items"]) >= 1
    assert all(o["status"] == "success" for o in r.json()["items"])


def test_feedback_list(admin_client: TestClient, admin_token: str) -> None:
    _seed_user_with_data("13800000005")
    r = admin_client.get(
        "/admin/api/feedbacks", params={"status": "pending"}, headers=_auth(admin_token)
    )
    assert r.status_code == 200
    assert all(f["status"] == "pending" for f in r.json()["items"])


def test_feedback_list_filter_by_source(admin_client: TestClient, admin_token: str) -> None:
    """反馈列表按反馈类型(source)筛选;返回项带 source/scene 字段。"""
    db = SessionLocal()
    try:
        uid = user_repo.upsert_user_for_login(db, phone="13800009001", register_channel="sms").id
        db.add(Feedback(user_id=uid, content="比价反馈", contact="", status="pending",
                        source="comparison", scene="找错商品"))
        db.add(Feedback(user_id=uid, content="普通反馈", contact="", status="pending",
                        source="profile"))
        db.commit()
    finally:
        db.close()

    r = admin_client.get(
        "/admin/api/feedbacks", params={"user_id": uid, "source": "comparison"},
        headers=_auth(admin_token),
    )
    assert r.status_code == 200, r.text
    items = r.json()["items"]
    assert items and all(it["source"] == "comparison" for it in items)
    assert any(it["scene"] == "找错商品" for it in items)

    r = admin_client.get(
        "/admin/api/feedbacks", params={"user_id": uid, "source": "profile"},
        headers=_auth(admin_token),
    )
    assert all(it["source"] == "profile" for it in r.json()["items"])

    # 非法 source 值 → 422
    assert admin_client.get(
        "/admin/api/feedbacks", params={"source": "bogus"}, headers=_auth(admin_token)
    ).status_code == 422


def test_withdraw_list_exposes_source_and_filters(
    admin_client: TestClient, admin_token: str
) -> None:
    """提现列表带 source 字段(coin_cash/invite_cash),并可按提现类型筛选。"""
    db = SessionLocal()
    try:
        uid = user_repo.upsert_user_for_login(db, phone="13800009002", register_channel="sms").id
        # 直接从 ORM 查询，验证列表富化返回人工风险结论。
        user = db.get(User, uid)
        assert user is not None
        user.is_high_risk = True
        user.high_risk_note = "邀请行为异常"
        db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coin0001",
                             amount_cents=100, status="success", source="coin_cash"))
        db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invite001",
                             amount_cents=200, status="success", source="invite_cash"))
        db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coinreview",
                             amount_cents=50, status="reviewing", source="coin_cash"))
        db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invitereview",
                             amount_cents=80, status="reviewing", source="invite_cash"))
        db.commit()
    finally:
        db.close()

    r = admin_client.get("/admin/api/withdraws", params={"user_id": uid}, headers=_auth(admin_token))
    assert r.status_code == 200, r.text
    assert {it["source"] for it in r.json()["items"]} == {"coin_cash", "invite_cash"}

    r = admin_client.get(
        "/admin/api/withdraws", params={"user_id": uid, "source": "invite_cash"},
        headers=_auth(admin_token),
    )
    items = r.json()["items"]
    assert items and all(it["source"] == "invite_cash" for it in items)
    assert all(it["is_high_risk"] is True for it in items)
    assert all(it["high_risk_note"] == "邀请行为异常" for it in items)
    # 累计提现按当前审核页账户隔离，不把普通现金的 100 分串进邀请金。
    assert all(it["cumulative_success_cents"] == 200 for it in items)

    invite_summary = admin_client.get(
        "/admin/api/withdraws/summary",
        params={"source": "invite_cash"},
        headers=_auth(admin_token),
    ).json()
    other_summary = admin_client.get(
        "/admin/api/withdraws/summary",
        params={"source": "coin_cash"},
        headers=_auth(admin_token),
    ).json()
    assert invite_summary["reviewing_count"] >= 1
    assert other_summary["reviewing_count"] >= 1
    assert invite_summary["reviewing_amount_cents"] >= 80
    assert other_summary["reviewing_amount_cents"] >= 50


def test_invite_withdraw_detail_contains_invitee_first_actions(
    admin_client: TestClient, admin_token: str
) -> None:
    db = SessionLocal()
    try:
        inviter = user_repo.upsert_user_for_login(
            db, phone="13800009031", register_channel="sms"
        )
        invitee = user_repo.upsert_user_for_login(
            db, phone="13800009032", register_channel="sms"
        )
        db.add(
            InviteRelation(
                inviter_user_id=inviter.id,
                invitee_user_id=invitee.id,
                status="effective",
                compare_reward_granted=True,
                compare_reward_cents=50,
            )
        )
        db.add(
            ComparisonRecord(
                user_id=invitee.id,
                trace_id="invite-detail-first-compare",
                status="success",
                store_name="首次比价商家",
                product_names="商品甲、商品乙",
            )
        )
        db.add(
            SavingsRecord(
                user_id=invitee.id,
                order_amount_cents=1888,
                saved_amount_cents=300,
                source="compare",
                shop_name="首次下单商家",
                title="首次下单商品",
                dishes=[],
            )
        )
        bill = "inviteoverviewbill0001"
        db.add(
            WithdrawOrder(
                user_id=inviter.id,
                out_bill_no=bill,
                amount_cents=100,
                status="reviewing",
                source="invite_cash",
            )
        )
        db.commit()
    finally:
        db.close()

    response = admin_client.get(
        f"/admin/api/withdraws/{bill}", headers=_auth(admin_token)
    )
    assert response.status_code == 200, response.text
    overview = response.json()["invite_overview"]
    assert overview["invite_total"] == 1
    assert overview["invite_success_total"] == 1
    item = overview["items"][0]
    assert item["phone"] == "13800009032"
    assert item["first_compare_store"] == "首次比价商家"
    assert item["first_compare_products"] == "商品甲、商品乙"
    assert item["first_order_store"] == "首次下单商家"
    assert item["first_order_products"] == "首次下单商品"
    assert item["first_order_amount_cents"] == 1888

    filtered = admin_client.get(
        f"/admin/api/withdraws/{bill}",
        params={
            "date_from": "2099-01-01T00:00:00Z",
            "date_to": "2099-01-02T00:00:00Z",
        },
        headers=_auth(admin_token),
    )
    assert filtered.status_code == 200, filtered.text
    assert filtered.json()["invite_overview"]["invite_total"] == 0


def test_withdraw_summary_counts_match_status_lists_by_source(
    admin_client: TestClient, admin_token: str
) -> None:
    """每个审核页签的汇总数必须等于相同 source、status 的列表总数。"""
    db = SessionLocal()
    try:
        uid = user_repo.upsert_user_for_login(
            db, phone="13800009003", register_channel="sms"
        ).id
        for index, status in enumerate(
            ("reviewing", "pending", "success", "rejected", "failed"), start=1
        ):
            db.add(
                WithdrawOrder(
                    user_id=uid,
                    out_bill_no=f"summary{uid}invite{index}",
                    amount_cents=index * 10,
                    status=status,
                    source="invite_cash",
                )
            )
        db.add(
            WithdrawOrder(
                user_id=uid,
                out_bill_no=f"summary{uid}coinfailed",
                amount_cents=99,
                status="failed",
                source="coin_cash",
            )
        )
        db.commit()
    finally:
        db.close()

    status_fields = {
        "reviewing": "reviewing_count",
        "pending": "pending_count",
        "success": "success_count",
        "rejected": "rejected_count",
        "failed": "failed_count",
    }
    for source in ("coin_cash", "invite_cash"):
        summary_response = admin_client.get(
            "/admin/api/withdraws/summary",
            params={"source": source},
            headers=_auth(admin_token),
        )
        assert summary_response.status_code == 200, summary_response.text
        summary = summary_response.json()

        for status, field in status_fields.items():
            list_response = admin_client.get(
                "/admin/api/withdraws",
                params={"source": source, "status": status, "limit": 1},
                headers=_auth(admin_token),
            )
            assert list_response.status_code == 200, list_response.text
            assert summary[field] == list_response.json()["total"]

        reviewing_response = admin_client.get(
            "/admin/api/withdraws",
            params={"source": source, "status": "reviewing", "limit": 100},
            headers=_auth(admin_token),
        )
        assert reviewing_response.status_code == 200, reviewing_response.text
        reviewing_body = reviewing_response.json()
        assert reviewing_body["total"] == len(reviewing_body["items"])
        assert summary["reviewing_amount_cents"] == sum(
            item["amount_cents"] for item in reviewing_body["items"]
        )

        all_response = admin_client.get(
            "/admin/api/withdraws",
            params={"source": source, "limit": 1},
            headers=_auth(admin_token),
        )
        assert all_response.status_code == 200, all_response.text
        assert summary["total_count"] == all_response.json()["total"]
        assert summary["total_count"] == sum(
            summary[field] for field in status_fields.values()
        )

    assert admin_client.get(
        "/admin/api/withdraws/summary",
        params={"source": "unknown"},
        headers=_auth(admin_token),
    ).status_code == 422


def test_comparison_records_show_readable_device_and_rom_version(
    admin_client: TestClient, admin_token: str
) -> None:
    db = SessionLocal()
    try:
        user = user_repo.upsert_user_for_login(
            db, phone="13800009041", register_channel="sms"
        )
        record = ComparisonRecord(
            user_id=user.id,
            trace_id="comparison-readable-device",
            status="success",
            device_model="V2166BA",
            device_manufacturer="vivo",
            rom_vendor="vivo",
            rom_name="OriginOS",
            rom_version=4,
            android_version="14",
        )
        db.add(record)
        db.commit()
        record_id = record.id
        user_id = user.id
    finally:
        db.close()

    response = admin_client.get(
        "/admin/api/comparison-records",
        params={"user_id": user_id},
        headers=_auth(admin_token),
    )
    assert response.status_code == 200, response.text
    item = response.json()["items"][0]
    assert item["device_model"] == "V2166BA"
    assert item["device_model_name"] == "vivo Y77e"
    assert item["rom_name"] == "OriginOS"
    assert item["rom_version"] == 4

    detail = admin_client.get(
        f"/admin/api/comparison-records/{record_id}",
        headers=_auth(admin_token),
    )
    assert detail.status_code == 200, detail.text
    assert detail.json()["device_model_name"] == "vivo Y77e"
    assert detail.json()["rom_version"] == 4


def test_comparison_records_show_real_order_status(
    admin_client: TestClient, admin_token: str
) -> None:
    db = SessionLocal()
    try:
        user = user_repo.upsert_user_for_login(
            db, phone="13800009042", register_channel="sms"
        )
        ordered_record = ComparisonRecord(
            user_id=user.id,
            trace_id="comparison-ordered-shop",
            status="success",
            store_name="真实下单店",
        )
        unordered_record = ComparisonRecord(
            user_id=user.id,
            trace_id="comparison-demo-order-shop",
            status="success",
            store_name="演示下单店",
        )
        db.add_all([ordered_record, unordered_record])
        db.flush()
        db.add_all(
            [
                SavingsRecord(
                    user_id=user.id,
                    order_amount_cents=1800,
                    saved_amount_cents=300,
                    shop_name="真实下单店",
                    source="compare",
                    client_event_id="admin-comparison-real-order",
                ),
                SavingsRecord(
                    user_id=user.id,
                    order_amount_cents=1500,
                    saved_amount_cents=200,
                    shop_name="演示下单店",
                    source="demo",
                ),
            ]
        )
        db.commit()
        ordered_record_id = ordered_record.id
        user_id = user.id
    finally:
        db.close()

    response = admin_client.get(
        "/admin/api/comparison-records",
        params={"user_id": user_id},
        headers=_auth(admin_token),
    )
    assert response.status_code == 200, response.text
    by_trace = {item["trace_id"]: item for item in response.json()["items"]}
    assert by_trace["comparison-ordered-shop"]["ordered"] is True
    assert by_trace["comparison-demo-order-shop"]["ordered"] is False

    detail = admin_client.get(
        f"/admin/api/comparison-records/{ordered_record_id}",
        headers=_auth(admin_token),
    )
    assert detail.status_code == 200, detail.text
    assert detail.json()["ordered"] is True


def test_ad_coin_audit_full_count_truncate_and_only_mismatch(
    admin_client: TestClient, admin_token: str
) -> None:
    """A+B:total/mismatch_count 按全量统计(不受 limit 影响),truncated 旗标 + only_mismatch 过滤。

    用 capped 行造确定性数据(应发恒 0,coin==0 即一致),不依赖发奖公式:
    3 条 coin=0(一致) + 2 条 coin=7(不一致) → 全量 total=5、mismatch=2。
    """
    from app.models.ad_reward import AdRewardRecord

    d = "2020-01-15"  # 固定历史日 + 独立 user,隔离其它用例数据
    db = SessionLocal()
    try:
        uid = user_repo.upsert_user_for_login(db, phone="13800009999", register_channel="sms").id
        for i in range(3):
            db.add(AdRewardRecord(
                trans_id=f"adaudit-ok-{i}", user_id=uid, coin=0, status="capped",
                reward_scene="reward_video", reward_date=d,
            ))
        for i in range(2):
            db.add(AdRewardRecord(
                trans_id=f"adaudit-bad-{i}", user_id=uid, coin=7, status="capped",
                reward_scene="reward_video", reward_date=d,
            ))
        db.commit()
    finally:
        db.close()

    base = {"date": d, "user_id": uid}

    # 全量:total=5、mismatch=2、不截断、返回 5 条
    body = admin_client.get(
        "/admin/api/ad-coin-audit", params={**base, "limit": 100}, headers=_auth(admin_token)
    ).json()
    assert body["total"] == 5
    assert body["mismatch_count"] == 2
    assert body["truncated"] is False
    assert len(body["items"]) == 5

    # 截断:limit=2 → 统计仍全量、truncated=True、只回 2 条
    body = admin_client.get(
        "/admin/api/ad-coin-audit", params={**base, "limit": 2}, headers=_auth(admin_token)
    ).json()
    assert body["total"] == 5 and body["mismatch_count"] == 2
    assert body["truncated"] is True
    assert len(body["items"]) == 2

    # only_mismatch:只回 ✗ 行(2 条),统计仍全量、不截断
    body = admin_client.get(
        "/admin/api/ad-coin-audit",
        params={**base, "limit": 100, "only_mismatch": True},
        headers=_auth(admin_token),
    ).json()
    assert body["total"] == 5 and body["mismatch_count"] == 2
    assert body["truncated"] is False
    assert len(body["items"]) == 2
    assert all(it["matched"] is False for it in body["items"])


def test_read_apis_require_auth(admin_client: TestClient) -> None:
    """所有 M2 读接口未带 token → 401(router 级 get_current_admin 守卫)。"""
    for path in [
        "/admin/api/stats/overview",
        "/admin/api/users",
        "/admin/api/wallet/coin-transactions",
        "/admin/api/withdraws",
        "/admin/api/feedbacks",
    ]:
        assert admin_client.get(path).status_code == 401, path


def test_period_comparison_reward_coin_from_feed_scene(
    admin_client: TestClient, admin_token: str
) -> None:
    """比价奖励金币口径:按 ad_feed_reward_record.feed_scene='comparison' 的实发金币
    (status=granted)汇总,而非查从不写入的 biz_type 桶(修复大盘该卡恒 0)。
    too_short(未发奖)不计。"""
    from app.models.ad_feed_reward import AdFeedRewardRecord

    d = "2021-06-15"  # 独立历史日,隔离其它用例数据
    db = SessionLocal()
    try:
        uid = user_repo.upsert_user_for_login(db, phone="13800008801", register_channel="sms").id
        db.add(AdFeedRewardRecord(
            client_event_id="cmp-fs-granted", user_id=uid, reward_date=d,
            ecpm_raw="0", coin=123, feed_scene="comparison", status="granted",
        ))
        db.add(AdFeedRewardRecord(
            client_event_id="cmp-fs-tooshort", user_id=uid, reward_date=d,
            ecpm_raw="0", coin=99, feed_scene="comparison", status="too_short",
        ))
        db.commit()
    finally:
        db.close()

    r = admin_client.get(
        "/admin/api/stats/overview", params={"date_from": d, "date_to": d},
        headers=_auth(admin_token),
    )
    assert r.status_code == 200, r.text
    assert r.json()["period"]["coins"]["comparison_reward_coin_total"] == 123


def test_period_coupon_reward_coin_from_feed_scene(
    admin_client: TestClient, admin_token: str
) -> None:
    """领券奖励金币口径:按 ad_feed_reward_record.feed_scene='coupon' 的实发金币汇总。"""
    from app.models.ad_feed_reward import AdFeedRewardRecord

    d = "2021-06-16"
    db = SessionLocal()
    try:
        uid = user_repo.upsert_user_for_login(db, phone="13800008802", register_channel="sms").id
        db.add(AdFeedRewardRecord(
            client_event_id="cpn-fs-granted", user_id=uid, reward_date=d,
            ecpm_raw="0", coin=456, feed_scene="coupon", status="granted",
        ))
        db.commit()
    finally:
        db.close()

    r = admin_client.get(
        "/admin/api/stats/overview", params={"date_from": d, "date_to": d},
        headers=_auth(admin_token),
    )
    assert r.status_code == 200, r.text
    assert r.json()["period"]["coins"]["coupon_reward_coin_total"] == 456


def test_period_coupon_reward_excludes_reward_video(
    admin_client: TestClient, admin_token: str
) -> None:
    """领券奖励金币不再把激励视频金币算进来(reward_video/ad_reward 从领券桶拆出);
    激励视频仍单独计入 reward_video_coin_total。"""
    from datetime import datetime

    from app.models.wallet import CoinTransaction

    d = "2021-06-17"
    db = SessionLocal()
    try:
        uid = user_repo.upsert_user_for_login(db, phone="13800008803", register_channel="sms").id
        db.add(CoinTransaction(
            user_id=uid, amount=50, balance_after=50, biz_type="reward_video",
            ref_id="rv-split-1", created_at=datetime(2021, 6, 17, 12, 0, 0),
        ))
        db.commit()
    finally:
        db.close()

    r = admin_client.get(
        "/admin/api/stats/overview", params={"date_from": d, "date_to": d},
        headers=_auth(admin_token),
    )
    assert r.status_code == 200, r.text
    coins = r.json()["period"]["coins"]
    assert coins["coupon_reward_coin_total"] == 0        # 激励视频不计入领券奖励
    assert coins["reward_video_coin_total"] == 50        # 仍计入激励视频卡


def test_period_regular_task_coin_uses_explicit_allowlist(
    admin_client: TestClient, admin_token: str
) -> None:
    """常规任务金币只加明确任务来源，领券/比价及未知新类型不得自动混入。"""
    from datetime import datetime

    from app.models.wallet import CoinTransaction

    d = "2021-06-18"
    included = {
        "signin": 100,
        "task_enable_notification": 300,
        "task_other": 400,
        "price_report_reward": 500,
        "feedback_reward": 600,
    }
    excluded = {
        "signin_boost": 200,
        "feed_ad_reward_coupon": 700,
        "feed_ad_reward_comparison": 800,
        "feed_ad_reward": 900,
        "reward_video": 1000,
        "admin_grant": 1100,
        "invite_inviter": 1200,
        "future_unknown_reward": 1300,
    }
    db = SessionLocal()
    try:
        uid = user_repo.upsert_user_for_login(
            db, phone="13800008804", register_channel="sms"
        ).id
        balance = 0
        rows = []
        for index, (biz_type, amount) in enumerate((included | excluded).items(), start=1):
            balance += amount
            rows.append(CoinTransaction(
                user_id=uid,
                amount=amount,
                balance_after=balance,
                biz_type=biz_type,
                ref_id=f"regular-task-{index}",
                created_at=datetime(2021, 6, 18, 12, 0, index),
            ))
        db.add_all(rows)
        db.commit()
    finally:
        db.close()

    response = admin_client.get(
        "/admin/api/stats/overview",
        params={"date_from": d, "date_to": d},
        headers=_auth(admin_token),
    )
    assert response.status_code == 200, response.text
    coins = response.json()["period"]["coins"]
    assert coins["regular_task_coin_total"] == sum(included.values())
    assert coins["task_coin_total"] == 700
    assert coins["reward_video_coin_total"] == 1200


def test_period_signin_boost_moves_to_reward_video_without_double_count(
    admin_client: TestClient, admin_token: str
) -> None:
    """历史签到膨胀归看视频桶，不再进常规任务；本期发放总额不变且不重复计算。"""
    from datetime import datetime

    from app.models.wallet import CoinTransaction

    d = "2021-06-19"
    rows = [
        ("signin_boost", 200),
        ("reward_video", 100),
        ("signin", 50),
    ]
    db = SessionLocal()
    try:
        uid = user_repo.upsert_user_for_login(
            db, phone="13800008805", register_channel="sms"
        ).id
        balance = 0
        for index, (biz_type, amount) in enumerate(rows, start=1):
            balance += amount
            db.add(CoinTransaction(
                user_id=uid,
                amount=amount,
                balance_after=balance,
                biz_type=biz_type,
                ref_id=f"signin-boost-route-{index}",
                created_at=datetime(2021, 6, 19, 12, 0, index),
            ))
        db.commit()
    finally:
        db.close()

    response = admin_client.get(
        "/admin/api/stats/overview",
        params={"date_from": d, "date_to": d},
        headers=_auth(admin_token),
    )
    assert response.status_code == 200, response.text
    coins = response.json()["period"]["coins"]
    assert coins["granted_total"] == 350
    assert coins["reward_video_coin_total"] == 300
    assert coins["regular_task_coin_total"] == 50
    assert coins["signin_boost_coin_total"] == 200
