"""两条云账单链路的日级互查 —— 抓「明细被静默截断」。

每个云账号每天都被写进两张表,走的是**两个不同的厂商调用**:

* ``cmdb_cloud_billing_daily`` —— 账号×天的**日总额**(``cmdb.billing_daily.<slug>``)
* ``cmdb_cost_facts`` —— 账号×天×资源×服务的**逐行明细**(``cmdb.cost_facts.<slug>``)

页面上的每一个折前数字都来自后者。而后者是逐日翻页拉的,**厂商提前返回一个
空 NextToken 就等于「这天到此为止」** —— 少拉的部分没有任何痕迹:任务写
``status=ok``,run summary 里的 facts 数与实际入库行数完全相等(拉到多少写多少),
只有把它跟另一条链路对一下才看得出来。

2026-08-16 实测,全年 2026 一共 3 起,全在阿里云国际站:

| 账号 | 日期 | 日总额 | 明细 | 比 |
|---|---|---:|---:|---:|
| dbaas-prod-dataplane-001 | 2026-05-06 | 12,424.40 | **200.83** | 0.016 |
| dbaas-prod-dataplane-001 | 2026-05-05 | 12,209.62 | 8,520.61 | 0.698 |
| dbaas-staging-dev-nextgen-dataplane | 2026-07-31 | 473.88 | 91.93 | 0.194 |

前两起就是 5 月对账剩下的 $15,913,第三起就是 7 月对账报出的 $381.95 ——
**两个花了很久才查清的问题,这一条 SQL 当天就能指出来,且不打任何厂商接口**
(两张表本来就都有数据)。

## 阈值是量出来的,不是拍的

`aws` 的两条链路来自**不同数据源**(billing_daily 走 Cost Explorer,cost_facts
走 BigQuery 里的 CUR),天然差几个百分点;`gcp` 两边完全相等(2026 全年 3,221 天
比值恒为 1.0000,标准差 0.0001);`alicloud_intl` 平均比 1.03。所以「差 5% 就报」
会在 aws 上刷出 224 条噪声 —— 那正是[迁移 481](../../alembic/versions/481_monthclose_second_pass.py)
的注释里说过的那种假阳性:**它训练人忽略告警,比没有告警更坏。**

2026 全年 11,716 个「账号×天」上实测各档:

| 阈值 | aws | alicloud_intl | gcp |
|---|---:|---:|---:|
| 比值 <0.95 且缺口 ≥$50 | 224 | 3 | 0 |
| 比值 <0.90 且缺口 ≥$100 | 16 | 3 | 0 |
| **比值 <0.85 且缺口 ≥$200** | **0** | **3** | **0** |

取最后一档:**恰好命中那 3 起真事故,零误报**。两个条件缺一不可 —— 只看比值会
被小账号的分币噪声刷屏($0.01 对 $0.05 是 0.2 的比值),只看绝对额会漏掉小账号上
的整天丢失。

## 只读

发现了只报告 + 给出修复窗口,绝不自动重拉:重拉是 (scope, date) 先删后插的破坏性
操作,而「两条链路不一致」并不等于「明细一定是错的那条」—— 也可能是日总额那条坏了。
判断该修哪边是人的事。
"""
from __future__ import annotations

from dataclasses import dataclass
from datetime import date, timedelta
from decimal import Decimal
from typing import Any

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.cost_view import fact_engineering_amount
from app.domains.cmdb.models import (
    CmdbCloudBillingDaily,
    CmdbCloudScope,
    CmdbCostFact,
)
from app.integrations.models import Integration

# 比值下限:明细/日总额低于它才算「疑似整段丢失」而不是口径差。
MAX_HEALTHY_RATIO = 0.85
# 绝对额下限:小账号上 $0.01 对 $0.05 也是 0.2 的比值,但毫无意义。
MIN_MATERIAL_USD = Decimal("200")
# 回看窗口:覆盖月结第二遍(12 号)能改到的最远一天,即上上个月月初。
DEFAULT_LOOKBACK_DAYS = 45


@dataclass(slots=True)
class ChainDivergence:
    provider: str
    integration_slug: str
    scope_external_id: str
    scope_name: str | None
    day: date
    daily_total_usd: Decimal
    facts_total_usd: Decimal

    @property
    def gap_usd(self) -> Decimal:
        return self.daily_total_usd - self.facts_total_usd

    @property
    def ratio(self) -> float:
        if self.daily_total_usd == 0:
            return 1.0
        return float(self.facts_total_usd / self.daily_total_usd)

    def as_dict(self) -> dict[str, Any]:
        return {
            "provider": self.provider,
            "integration": self.integration_slug,
            "scope": self.scope_external_id,
            "scope_name": self.scope_name,
            "date": self.day.isoformat(),
            "daily_total_usd": float(self.daily_total_usd),
            "facts_total_usd": float(self.facts_total_usd),
            "gap_usd": float(self.gap_usd),
            "ratio": round(self.ratio, 4),
        }


async def find_chain_divergences(
    session: AsyncSession,
    *,
    start: date,
    end: date,
    max_ratio: float = MAX_HEALTHY_RATIO,
    min_gap_usd: Decimal = MIN_MATERIAL_USD,
) -> list[ChainDivergence]:
    """明细链路显著低于日总额链路的 (账号, 天),按缺口从大到小。

    明细侧**复用 `fact_engineering_amount()`** —— 页面上真正在用的那个表达式。
    自己重写一遍聚合的话,对上了也不能证明页面是对的。
    """
    facts = (
        select(
            CmdbCostFact.scope_id.label("scope_id"),
            CmdbCostFact.date.label("day"),
            func.sum(fact_engineering_amount()).label("amt"),
        )
        .where(CmdbCostFact.date >= start, CmdbCostFact.date <= end)
        .group_by(CmdbCostFact.scope_id, CmdbCostFact.date)
        .subquery()
    )

    stmt = (
        select(
            Integration.provider,
            Integration.slug,
            CmdbCloudScope.external_id,
            CmdbCloudScope.display_name,
            CmdbCloudBillingDaily.date,
            CmdbCloudBillingDaily.total_cost_usd_list,
            facts.c.amt,
        )
        .join(
            CmdbCloudScope,
            CmdbCloudScope.id == CmdbCloudBillingDaily.scope_id,
        )
        .join(Integration, Integration.id == CmdbCloudScope.integration_id)
        # OUTER,不是 INNER。明细侧的 GROUP BY 只会为「至少有一行」的
        # (scope, date) 产出分组,所以 INNER JOIN 会把**一行都没拉到的那天**
        # 整个丢掉 —— 而那正是本检查要抓的最严重形态(整天丢失)。用 INNER
        # 的话,``divergence_count`` 在损失最大的时候恰好是 0,给出一个「明细
        # 链路健康」的假保证。下面 ``row[6] or 0`` 的空值兜底就是为这条准备的。
        .outerjoin(
            facts,
            (facts.c.scope_id == CmdbCloudBillingDaily.scope_id)
            & (facts.c.day == CmdbCloudBillingDaily.date),
        )
        .where(
            CmdbCloudBillingDaily.date >= start,
            CmdbCloudBillingDaily.date <= end,
            # 没写折前列的厂商(azure / tencent / ucloud)没有可比的口径,
            # 拿实付去比折前只会得到一堆假漂移。
            CmdbCloudBillingDaily.total_cost_usd_list.is_not(None),
            CmdbCloudBillingDaily.total_cost_usd_list > 0,
        )
    )

    out: list[ChainDivergence] = []
    for row in (await session.execute(stmt)).all():
        daily = Decimal(str(row[5]))
        amt = Decimal(str(row[6] or 0))
        if daily - amt < min_gap_usd:
            continue
        if float(amt / daily) >= max_ratio:
            continue
        out.append(ChainDivergence(
            provider=row[0], integration_slug=row[1],
            scope_external_id=row[2], scope_name=row[3],
            day=row[4], daily_total_usd=daily, facts_total_usd=amt,
        ))
    out.sort(key=lambda d: d.gap_usd, reverse=True)
    return out


def repair_windows(
    divergences: list[ChainDivergence],
) -> list[dict[str, str]]:
    """把命中的天按 (集成, 连续区间) 合并成可直接用的重拉窗口。

    命中往往连着几天(一次翻页故障可能跨日),逐天重拉是浪费;而跨集成合并
    则会把一个账号的问题变成所有账号的重拉。
    """
    by_integration: dict[str, list[date]] = {}
    for d in divergences:
        by_integration.setdefault(d.integration_slug, []).append(d.day)

    windows: list[dict[str, str]] = []
    for slug, days in sorted(by_integration.items()):
        ordered = sorted(set(days))
        run_start = prev = ordered[0]
        for day in ordered[1:]:
            if day - prev > timedelta(days=1):
                windows.append({
                    "integration": slug,
                    "start_date": run_start.isoformat(),
                    "end_date": prev.isoformat(),
                })
                run_start = day
            prev = day
        windows.append({
            "integration": slug,
            "start_date": run_start.isoformat(),
            "end_date": prev.isoformat(),
        })
    return windows


async def crosscheck_recent(
    session: AsyncSession,
    *,
    today: date,
    lookback_days: int = DEFAULT_LOOKBACK_DAYS,
    max_ratio: float = MAX_HEALTHY_RATIO,
    min_gap_usd: Decimal = MIN_MATERIAL_USD,
) -> dict[str, Any]:
    """回看窗口内的互查报告(**只读**)。"""
    end = today - timedelta(days=1)  # 今天还在累积,天然不齐
    start = end - timedelta(days=max(lookback_days, 1) - 1)
    found = await find_chain_divergences(
        session, start=start, end=end,
        max_ratio=max_ratio, min_gap_usd=min_gap_usd,
    )
    return {
        "window": {"start": start.isoformat(), "end": end.isoformat()},
        "max_ratio": max_ratio,
        "min_gap_usd": float(min_gap_usd),
        "divergence_count": len(found),
        "total_gap_usd": float(sum((d.gap_usd for d in found), Decimal("0"))),
        "divergences": [d.as_dict() for d in found],
        "repair_windows": repair_windows(found),
    }