"""双口径回填验收断言（成本口径改造 PR2 配套运维脚本）。

用法（在 server/ 目录，指向目标 DB 的环境下）::

    uv run python scripts/verify_cost_dual_measure_backfill.py \
        --provider gcp --start 2026-05-01 --end 2026-05-31 \
        --expect-list 910010.34 --tolerance 1.0

    uv run python scripts/verify_cost_dual_measure_backfill.py \
        --provider aws --start 2026-06-10 --end 2026-06-16 \
        --expect-list 1414157 --tolerance 5.0

已核验的期望值（2026-07-27 用云厂商账单原始数据只读查询得出，
详见 docs/CHANGELOG.md 对应条目）：

* GCP 2026-05 整月 business_list_cost = **$910,010.34**
  （= 牌价 $976,161.95 − Support $66,151.61；Invoice 行 at_list=0）。
* AWS 2026-06-10..16 白名单周样本 ≈ **$1,414,157**
  （Usage $983,566.63 + SavingsPlanCoveredUsage $430,590.57）。

脚本同时输出结构性体检（无需期望值也应成立）：

* 发票口径列在回填前后必须逐日一致（回填只新增列，绝不改 amount_usd）
  —— 这里核对 SUM(amount_usd) 与 billing_daily SUM(total_cost_usd)
  的量级一致性（两表口径本就允许小差异，仅告警不判失败）。
* 白名单纯度：amount_usd_list 非 NULL 的行 cost_bucket 必须全是
  'usage'（白名单泄漏检测）。
* GCP 零线修复：给出窗口内 list 全 0 的 (integration, scope, day) 数
  （回填后 5 月上半月的 GCP 项目不应再整段为 0）。

对不上 → 非零退出码 + 明细，绝不静默。只读脚本，不写任何数据。
"""
from __future__ import annotations

import argparse
import asyncio
import sys
from datetime import date

from sqlalchemy import func, select

# 生产 SSM 脚本惯例:先 import app.main 完成 mapper 配置。
import app.main  # noqa: F401
from app.core.database import async_session_factory
from app.domains.cmdb.models import CmdbCloudBillingDaily, CmdbCostFact
from app.integrations.models import Integration


async def run(provider: str, start: date, end: date,
              expect_list: float | None, tolerance: float) -> int:
    failures: list[str] = []
    async with async_session_factory() as session:
        integ_ids = [
            str(r) for r in (await session.execute(
                select(Integration.id).where(
                    Integration.kind == "cloud",
                    Integration.provider == provider,
                ),
            )).scalars().all()
        ]
        if not integ_ids:
            print(f"FAIL: no cloud integration with provider={provider}")
            return 2

        base = (
            select(
                func.sum(CmdbCostFact.amount_usd),
                func.sum(CmdbCostFact.amount_usd_list),
                func.sum(CmdbCostFact.amount_usd_public),
                func.count(),
            )
            .where(CmdbCostFact.integration_id.in_(integ_ids))
            .where(CmdbCostFact.date >= start)
            .where(CmdbCostFact.date <= end)
        )
        invoice_sum, list_sum, public_sum, n = (
            await session.execute(base)
        ).one()
        print(
            f"[{provider}] {start}..{end}  rows={n}\n"
            f"  invoice SUM(amount_usd)        = {float(invoice_sum or 0):,.2f}\n"
            f"  list    SUM(amount_usd_list)   = {float(list_sum or 0):,.2f}\n"
            f"  public  SUM(amount_usd_public) = {float(public_sum or 0):,.2f}"
        )

        if expect_list is not None:
            got = float(list_sum or 0)
            if abs(got - expect_list) > tolerance:
                failures.append(
                    f"list sum {got:,.2f} != expected {expect_list:,.2f} "
                    f"(tolerance {tolerance})"
                )
            else:
                print(f"  ✅ list sum matches expected {expect_list:,.2f}")

        # 白名单纯度：list 非 NULL 的行必须 bucket='usage'。
        leak = (await session.execute(
            select(func.count())
            .where(CmdbCostFact.integration_id.in_(integ_ids))
            .where(CmdbCostFact.date >= start)
            .where(CmdbCostFact.date <= end)
            .where(CmdbCostFact.amount_usd_list.is_not(None))
            .where(CmdbCostFact.cost_bucket != "usage"),
        )).scalar()
        if leak:
            failures.append(
                f"whitelist purity violated: {leak} rows carry a list "
                "measure with cost_bucket != 'usage'"
            )
        else:
            print("  ✅ whitelist purity: every list-bearing row is bucket=usage")

        # billing_daily 侧量级对照（提示性,不判失败——两表粒度不同）。
        daily_invoice, daily_list = (await session.execute(
            select(
                func.sum(CmdbCloudBillingDaily.total_cost_usd),
                func.sum(CmdbCloudBillingDaily.total_cost_usd_list),
            )
            .where(CmdbCloudBillingDaily.integration_id.in_(integ_ids))
            .where(CmdbCloudBillingDaily.date >= start)
            .where(CmdbCloudBillingDaily.date <= end),
        )).one()
        print(
            f"  (billing_daily: invoice={float(daily_invoice or 0):,.2f} "
            f"list={float(daily_list or 0):,.2f})"
        )

        # 零线体检：窗口内 list 恒 0/NULL 的天数占比（GCP 5 月上半月
        # 修复的验收信号——回填后不应整段为 0）。
        zero_days = (await session.execute(
            select(func.count())
            .select_from(CmdbCloudBillingDaily)
            .where(CmdbCloudBillingDaily.integration_id.in_(integ_ids))
            .where(CmdbCloudBillingDaily.date >= start)
            .where(CmdbCloudBillingDaily.date <= end)
            .where(CmdbCloudBillingDaily.total_cost_usd != 0)
            .where(func.coalesce(
                CmdbCloudBillingDaily.total_cost_usd_list, 0) == 0),
        )).scalar()
        print(
            f"  (days with nonzero invoice but zero/NULL list: {zero_days} — "
            "expect ~0 after backfill on list-wired providers)"
        )

    if failures:
        print("\nFAIL:")
        for f in failures:
            print(f"  ✗ {f}")
        return 1
    print("\nOK — dual-measure backfill assertions passed.")
    return 0


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--provider", required=True,
                    help="cloud provider slug prefix, e.g. aws / gcp")
    ap.add_argument("--start", required=True, type=date.fromisoformat)
    ap.add_argument("--end", required=True, type=date.fromisoformat)
    ap.add_argument("--expect-list", type=float, default=None,
                    help="expected SUM(amount_usd_list) for the window")
    ap.add_argument("--tolerance", type=float, default=1.0)
    args = ap.parse_args()
    sys.exit(asyncio.run(run(
        args.provider, args.start, args.end,
        args.expect_list, args.tolerance,
    )))


if __name__ == "__main__":
    main()