"""Service-level coverage for ``BailianService.sync_usage_daily`` after
the 2026-06-07 MONTHLY-grain rewrite (QueryBillOverview source).

``query_usage_daily`` now emits ONE aggregate row per billing cycle,
dated first-of-month. ``sync_usage_daily`` upserts those rows and sweeps
stale ones — but **only within months that actually returned data this
run** (``seen_months``). That guard is the whole point of these tests:

  1. Multi-month window where one month transiently 200-OKs empty must
     NOT delete that month's existing good row (the data-loss fix).
  2. Pre-fix per-day rows left in a confirmed month by the old (wrong)
     per-day implementation MUST be swept, leaving only the monthly row.
  3. The new ``cost_*_after_discount`` columns persist through both the
     insert and the update branch.

We patch the client's ``query_usage_daily`` (via a stub returned from
``_get_client``) so no Aliyun / BSS call happens, and stub the FX-rate
read; everything else exercises the real service write path against a
real DB row + a seeded cloud integration.

Run isolated:
    PYTEST_XDIST_WORKER=bailianusage uv run --python 3.12 --group dev \\
        pytest tests/test_bailian/test_bailian_usage_sync.py
"""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from unittest.mock import AsyncMock

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.ai_spending.service import AiSpendingService
from app.domains.bailian.models import BailianApiKey, BailianUsageDaily
from app.domains.bailian.service import BailianService
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration

_RATE = 0.14


@pytest.mark.asyncio
async def test_bucket_bailian_propagates_overview_items_to_bucketrow_raw(
    db_session: AsyncSession,
):
    """Regression: ``_bucket_bailian`` must carry the loader's per-item
    ``raw_data["bss_overview_items"]`` onto ``BucketRow.raw`` — otherwise the
    drill-down (and the closed-month rollup that persists ``BucketRow.raw``)
    only ever see the hand-built subset and fall back to one opaque
    ``aggregate`` row. The first cut shipped the loader + ``get_detail``
    expansion but forgot this propagation, so prod still showed one row.

    2026-08-15:载体从 ``model_name="aggregate"`` 改成一条**明细行**。
    ``aggregate`` 是同一天的**汇总行**,与逐 model 明细行并存,任何求和
    都必须排除它,否则双计(常量 ``BAILIAN_AGGREGATE_MODEL``,
    ``apikey/spend.py`` 一直是这么过滤的,``ai_spending`` 漏了)。本用例
    要证的是「``bss_overview_items`` 会被带到 ``BucketRow.raw``」,与载体
    行是不是汇总行无关 —— 用汇总行做载体只是让这条用例顺带钉住了一个
    已经被判定为错误的行为。生产上 ``bailian_usage_daily`` 已无任何
    ``aggregate`` 行(2026-01~08 全部为明细行,同步逻辑会删除遗留汇总行),
    所以这次换载体不改变它的覆盖面。"""
    iid = await _seed_integration(db_session)
    items = [
        {"bill_account_name": "sa@pingcap.com",
         "commodity_code": "sfm_inferenceglobal_public_intl",
         "product_detail": "Model Studio Foundation Model Inference",
         "biz_type": "trusteeship", "net_usd": 374.21, "gross_usd": 1069.19},
        {"bill_account_name": "ml-team@pingcap.com",
         "commodity_code": "sfm_inferenceglobal_public_intl",
         "product_detail": "Model Studio Foundation Model Inference",
         "biz_type": "", "net_usd": 4353.50, "gross_usd": 12831.34},
    ]
    db_session.add(BailianUsageDaily(
        integration_id=iid,
        usage_date=date(2026, 5, 1),
        model_name="qwen3.5-plus",
        aliyun_account_id=None,
        cost_usd=Decimal("4727.710000"),
        cost_cny=Decimal("33769.357142"),
        cost_usd_gross=Decimal("13900.535030"),
        raw_data={"bss_overview_items": items,
                  "bss_overview_sample": items[0], "cycle": "2026-05"},
    ))
    await db_session.commit()

    rows = await AiSpendingService(db_session)._bucket_bailian(
        date(2026, 5, 1), date(2026, 6, 1),
    )
    assert len(rows) == 1
    carried = rows[0].raw.get("bss_overview_items")
    assert carried is not None and len(carried) == 2
    assert {x["bill_account_name"] for x in carried} == {
        "sa@pingcap.com", "ml-team@pingcap.com",
    }


def _today_window():
    today = datetime.now(UTC).date()
    return today


def _key_item(
    usage_date: date,
    *,
    account_id: str | None,
    api_key_id: str | None,
    model_name: str,
    workspace_id: str | None = None,
    cost_usd: float,
    cost_cny: float,
    gross_usd: float | None = None,
    gross_cny: float | None = None,
    ad_usd: float | None = None,
    ad_cny: float | None = None,
) -> dict:
    """Build one row in the shape ``query_instance_bill_daily`` emits:
    per-(day × account × API key × model)."""
    return {
        "usage_date": usage_date.isoformat(),
        "aliyun_account_id": account_id,
        "aliyun_account_name": None,
        "bailian_api_key_id": api_key_id,
        "workspace_id": workspace_id,
        "model_name": model_name,
        "channels": [],
        "input_tokens": 0,
        "output_tokens": 0,
        "request_count": 0,
        "cost_cny": cost_cny,
        "cost_usd": cost_usd,
        "cost_cny_gross": gross_cny,
        "cost_usd_gross": gross_usd,
        "cost_cny_after_discount": ad_cny,
        "cost_usd_after_discount": ad_usd,
        "raw": {"currency": "USD", "channels": []},
    }


class _StubClient:
    """Stands in for :class:`BailianClient`. Its
    ``query_instance_bill_daily`` is the seam the tests drive — returns
    whatever per-key rows we set, ignoring the (start, end) window the
    service computes."""

    def __init__(self, rows: list[dict]) -> None:
        self.rows = rows
        self.calls: list[tuple] = []

    async def query_instance_bill_daily(
        self, *, start_date: str, end_date: str, cny_to_usd_rate: float,
    ) -> list[dict]:
        self.calls.append((start_date, end_date, cny_to_usd_rate))
        return list(self.rows)


async def _seed_integration(db: AsyncSession) -> str:
    """A minimal Aliyun-family cloud integration so the usage rows'
    ``integration_id`` FK (RESTRICT → integrations.id) is satisfiable."""
    iid = uuid.uuid4()
    db.add(Integration(
        id=iid,
        kind="cloud",
        provider="alicloud_intl",
        name=f"bailian-{str(iid)[:6]}",
        slug=f"bl-{str(iid)[:6]}",
        enabled=True,
        is_primary=False,
    ))
    await db.commit()
    return str(iid)


def _service_with_stub(
    db: AsyncSession, stub: _StubClient, monkeypatch,
) -> BailianService:
    svc = BailianService(db)
    monkeypatch.setattr(svc, "_get_client", AsyncMock(return_value=stub))
    monkeypatch.setattr(
        svc, "_get_cny_to_usd_rate", AsyncMock(return_value=_RATE),
    )
    return svc


async def _rows_for(db: AsyncSession, iid: str) -> list[BailianUsageDaily]:
    db.expire_all()  # force a fresh load — prove values reached the DB
    res = await db.execute(
        select(BailianUsageDaily)
        .where(BailianUsageDaily.integration_id == iid)
        .order_by(BailianUsageDaily.usage_date)
    )
    return list(res.scalars().all())


@pytest.mark.asyncio
async def test_transiently_empty_day_survives_sweep(
    db_session: AsyncSession, monkeypatch,
):
    """Day A returns a per-key row; day B returns NOTHING this run (a
    transient empty response for B). A pre-existing good row in day B must
    SURVIVE — the sweep only deletes within ``seen_days``, and B isn't in
    it. Guards the data-loss fix."""
    iid = await _seed_integration(db_session)
    today = datetime.now(UTC).date()
    day_a = today
    day_b = today - timedelta(days=3)

    # Pre-seed an EXISTING good per-key row on day B.
    db_session.add(BailianUsageDaily(
        integration_id=iid,
        usage_date=day_b,
        model_name="qwen3.6-plus",
        aliyun_account_id="5464160084398809",
        bailian_api_key_id="132245",
        cost_usd=Decimal("111.110000"),
        cost_cny=Decimal("777.770000"),
    ))
    await db_session.commit()

    # Client returns ONLY day A — day B emits no row this run.
    stub = _StubClient([
        _key_item(day_a, account_id="5464160084398809", api_key_id="132245",
                  model_name="qwen3.6-plus", cost_usd=50.0, cost_cny=350.0),
    ])
    svc = _service_with_stub(db_session, stub, monkeypatch)
    result = await svc.sync_usage_daily(integration_id=iid, days=7)
    await db_session.commit()

    assert result["upserted"] == 1
    assert result["deleted_stale"] == 0  # day B NOT swept

    rows = await _rows_for(db_session, iid)
    by_day = {r.usage_date: r for r in rows}
    assert set(by_day) == {day_a, day_b}
    assert by_day[day_b].cost_usd == Decimal("111.11")  # original, intact
    assert by_day[day_a].cost_usd == Decimal("50")


@pytest.mark.asyncio
async def test_legacy_aggregate_row_swept_on_cutover(
    db_session: AsyncSession, monkeypatch,
):
    """The previous ingest left ONE ``model_name='aggregate'`` row per
    month (dated first-of-month). The new per-key sync must delete that
    legacy row for the covered month so it doesn't double-count, while
    landing the new per-key rows."""
    iid = await _seed_integration(db_session)
    today = datetime.now(UTC).date()
    month_first = today.replace(day=1)

    db_session.add(BailianUsageDaily(
        integration_id=iid,
        usage_date=month_first,
        model_name="aggregate",
        aliyun_account_id=None,
        cost_usd=Decimal("11.640000"),
        cost_cny=Decimal("78.940000"),
    ))
    await db_session.commit()

    stub = _StubClient([
        _key_item(today, account_id="5464160084398809", api_key_id="132245",
                  model_name="qwen3.6-plus", cost_usd=9.0, cost_cny=63.0),
        _key_item(today, account_id="5464160084398809", api_key_id="132245",
                  model_name="deepseek-v4-pro", cost_usd=2.5, cost_cny=17.5),
    ])
    svc = _service_with_stub(db_session, stub, monkeypatch)
    result = await svc.sync_usage_daily(integration_id=iid, days=7)
    await db_session.commit()

    assert result["deleted_legacy"] == 1
    assert result["upserted"] == 2

    rows = await _rows_for(db_session, iid)
    assert all(r.model_name != "aggregate" for r in rows)  # legacy gone
    assert {r.model_name for r in rows} == {"qwen3.6-plus", "deepseek-v4-pro"}


@pytest.mark.asyncio
async def test_after_discount_columns_persist_insert_and_update(
    db_session: AsyncSession, monkeypatch,
):
    """``cost_*_after_discount`` (the middle 优惠后 tier) must persist on
    the INSERT branch and then be overwritten on the UPDATE branch when
    the same per-key key is re-synced."""
    iid = await _seed_integration(db_session)
    today = datetime.now(UTC).date()

    stub = _StubClient([
        _key_item(
            today, account_id="5464160084398809", api_key_id="132245",
            model_name="qwen3.6-plus",
            cost_usd=100.0, cost_cny=700.0,
            gross_usd=250.0, gross_cny=1750.0,
            ad_usd=180.0, ad_cny=1260.0,
        ),
    ])
    svc = _service_with_stub(db_session, stub, monkeypatch)

    # INSERT branch.
    result = await svc.sync_usage_daily(integration_id=iid, days=7)
    await db_session.commit()
    assert result["upserted"] == 1

    rows = await _rows_for(db_session, iid)
    assert len(rows) == 1
    row = rows[0]
    assert row.bailian_api_key_id == "132245"
    assert row.cost_usd_after_discount == Decimal("180")
    assert row.cost_cny_after_discount == Decimal("1260")
    assert row.cost_usd_gross == Decimal("250")

    # UPDATE branch — same (day, account, key, model), new after-discount.
    stub.rows = [
        _key_item(
            today, account_id="5464160084398809", api_key_id="132245",
            model_name="qwen3.6-plus",
            cost_usd=120.0, cost_cny=840.0,
            gross_usd=250.0, gross_cny=1750.0,
            ad_usd=200.0, ad_cny=1400.0,
        ),
    ]
    result = await svc.sync_usage_daily(integration_id=iid, days=7)
    await db_session.commit()
    assert result["upserted"] == 1
    assert result["deleted_stale"] == 0  # same key — upsert, not churn

    rows = await _rows_for(db_session, iid)
    assert len(rows) == 1  # updated in place, not duplicated
    row = rows[0]
    assert row.cost_usd_after_discount == Decimal("200")
    assert row.cost_cny_after_discount == Decimal("1400")
    assert row.cost_usd == Decimal("120")


@pytest.mark.asyncio
async def test_list_usage_daily_carries_owner_leaver_indicators(
    db_session: AsyncSession,
):
    """list_usage_daily must surface owner_employee_no /
    owner_employment_status (in lock-step with whichever owner won —
    here the API-key owner) so the admin usage table renders the owner
    cell through <PersonTag> with the 已离职 treatment for a departed
    key owner."""
    iid = await _seed_integration(db_session)
    leaver = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"keyowner-{uuid.uuid4().hex[:6]}@pingcap.com",
        display_name="Key Owner",
        employee_no="P0654",
        employment_status="left",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(leaver)
    await db_session.flush()

    key = BailianApiKey(
        id=uuid.uuid4(),
        integration_id=iid,
        api_key_id=f"ak-{uuid.uuid4().hex[:10]}",
        display_name="prod key",
        owner_person_id=str(leaver.id),
    )
    db_session.add(key)
    await db_session.flush()

    db_session.add(BailianUsageDaily(
        integration_id=iid,
        usage_date=date(2026, 5, 1),
        model_name="qwen-max",
        api_key_pk=key.id,
        aliyun_account_id=None,
        input_tokens=10,
        output_tokens=20,
        request_count=1,
        cost_usd=Decimal("1.000000"),
        cost_cny=Decimal("7.000000"),
    ))
    await db_session.commit()

    out = await BailianService(db_session).list_usage_daily(integration_id=iid)
    assert len(out) == 1
    row = out[0]
    assert row.owner_person_id == str(leaver.id)
    assert row.owner_employee_no == "P0654"
    assert row.owner_employment_status == "left"


@pytest.mark.asyncio
async def test_workspace_collision_folds_not_crashes(
    db_session: AsyncSession, monkeypatch,
):
    """Two source rows that differ ONLY by workspace_id collapse onto the
    same DB chargeback grain (date, account, api_key, model — the unique key
    carries NO workspace). The old code issued a second INSERT for the
    second row → IntegrityError on flush → the whole integration's usage
    sync failed every tick. The service must FOLD them: one persisted row
    whose native costs are the SUM, with workspace nulled (ambiguous)."""
    iid = await _seed_integration(db_session)
    today = datetime.now(UTC).date()
    stub = _StubClient([
        _key_item(today, account_id="acct1", api_key_id="k1",
                  model_name="qwen-max", workspace_id="ws-A",
                  cost_usd=10.0, cost_cny=70.0,
                  gross_usd=12.0, gross_cny=84.0, ad_usd=11.0, ad_cny=77.0),
        _key_item(today, account_id="acct1", api_key_id="k1",
                  model_name="qwen-max", workspace_id="ws-B",
                  cost_usd=5.0, cost_cny=35.0,
                  gross_usd=6.0, gross_cny=42.0, ad_usd=5.5, ad_cny=38.5),
    ])
    svc = _service_with_stub(db_session, stub, monkeypatch)
    # Must NOT raise (the old IntegrityError-on-flush path).
    result = await svc.sync_usage_daily(integration_id=iid, days=7)
    await db_session.commit()

    assert result["upserted"] == 1  # two source rows folded into one grain
    rows = await _rows_for(db_session, iid)
    assert len(rows) == 1
    row = rows[0]
    assert row.cost_usd == Decimal("15")      # 10 + 5
    assert row.cost_cny == Decimal("105")     # 70 + 35
    assert row.cost_usd_gross == Decimal("18")  # 12 + 6
    assert row.cost_usd_after_discount == Decimal("16.5")  # 11 + 5.5
    # Workspace is ambiguous at this grain → nulled, not one-of-several.
    assert row.workspace_id_raw is None
    assert row.workspace_pk is None