"""Seed the ``cmdb.cost_facts_chain_check`` daily cron.

Revision ID: 482
Revises: 481

每天把每个云账号每天的**日总额**(``cmdb_cloud_billing_daily``)与**逐行明细
之和**(``cmdb_cost_facts`` 的折前口径)对一下。两张表来自**两个不同的厂商
调用**,页面上的每个折前数字都来自后者。

抓的是「明细被静默截断」:明细是逐日翻页拉的,厂商提前返回一个空 ``NextToken``
就等于「这天到此为止」,而少拉的部分**没有任何痕迹** —— 任务写 ``status=ok``,
run summary 里的行数与实际入库行数完全相等(拉到多少写多少)。

2026 全年实测 3 起,全在阿里云国际站,而且**正是此前两次昂贵排查的答案**:

* dbaas-prod-dataplane-001 2026-05-06:日总 12,424.40 vs 明细 **200.83**
* dbaas-prod-dataplane-001 2026-05-05:日总 12,209.62 vs 明细 8,520.61
  —— 上面两条合起来就是 5 月 BQ 对账剩下的 $15,913
* dbaas-staging-dev-nextgen-dataplane 2026-07-31:473.88 vs 91.93
  —— 就是 7 月 BQ 对账报出的 $381.95

**这条检查不打任何厂商接口**(两张表本来就都有数据),却能在当天指出问题,
而 BQ 月度对账要到次月 14 号才发现,且只能定位到「账号×月」。两者互补:
互查负责快而窄(哪一天丢了),对账负责慢而权威(与财务口径是否一致)。

阈值 ``比值<0.85 且缺口≥$200`` 是在 2026 全年 11,716 个「账号×天」上量出来的
—— 恰好命中那 3 起、零误报。放宽到 5% 会在 AWS 上刷出 224 条噪声:AWS 两条
链路来自不同数据源(Cost Explorer vs BigQuery 里的 CUR),天然差几个百分点。
"""
import uuid

import sqlalchemy as sa

from alembic import op

revision = "482"
down_revision = "481"
branch_labels = None
depends_on = None

_KEY = "cmdb.cost_facts_chain_check"
_DISPLAY = "云成本明细 · 两条链路日级互查"
_DESCRIPTION = (
    "每天 08:30(Asia/Shanghai)把每个云账号每天的【日总额】"
    "(cmdb_cloud_billing_daily)与【逐行明细之和】(cmdb_cost_facts 的折前"
    "口径)对一下 —— 两张表来自两个不同的厂商调用,页面上的每个折前数字都"
    "来自后者。\n"
    "抓的是「明细被静默截断」:明细是逐日翻页拉的,厂商提前返回一个空 "
    "NextToken 就等于「这天到此为止」,而少拉的部分没有任何痕迹 —— 任务写 "
    "status=ok,run summary 里的行数与实际入库行数完全相等。\n"
    "2026 全年实测 3 起(阿里云国际站):5/06 日总 12,424 vs 明细 200.83、"
    "5/05 12,210 vs 8,521、7/31 473.88 vs 91.93 —— 前两起是 5 月对账剩下的 "
    "$15,913,第三起是 7 月对账报出的 $381.95。\n"
    "⚠️ 只读:发现了只报告 + 给出重拉窗口,绝不自动重拉 —— 重拉是 "
    "(scope, date) 先删后插的破坏性操作,而两条链路不一致不等于明细一定是"
    "错的那条。\n"
    "阈值 比值<0.85 且缺口≥$200 在 2026 全年 11,716 个账号×天上量出:恰好"
    "命中那 3 起、零误报。放宽到 5% 会在 AWS 上刷出 224 条噪声(AWS 两条链路"
    "来自不同数据源)。\n"
    "与 cmdb.cost_recon.aliyun_bq 互补:互查快而窄(当天、定位到哪一天丢了),"
    "对账慢而权威(次月、与财务口径是否一致)。\n"
    "config={\"lookback_days\":N} 覆写回看窗口(默认 45);"
    "{\"max_ratio\":R,\"min_gap_usd\":G} 覆写阈值。"
)
# 08:30 SH:排在每日 cost_facts(04:xx)与 billing_daily(06:xx)之后,
# 两条链路当天都已写完才比得有意义。
_CRON = "30 8 * * *"


def upgrade() -> None:
    op.get_bind().execute(
        sa.text(
            "INSERT INTO scheduled_jobs "
            "(id, name, handler_key, display_name, description, "
            " trigger_type, cron_expr, timezone, enabled, config, "
            " created_at, updated_at) "
            "SELECT :id, :name, :k, :dn, :desc, 'cron', :cron, "
            " 'Asia/Shanghai', 1, '{}', NOW(), NOW() "
            "FROM DUAL WHERE NOT EXISTS ("
            "  SELECT 1 FROM scheduled_jobs WHERE handler_key = :k)"
        ),
        {
            "id": str(uuid.uuid4()), "name": _KEY, "k": _KEY,
            "dn": _DISPLAY, "desc": _DESCRIPTION, "cron": _CRON,
        },
    )


def downgrade() -> None:
    op.get_bind().execute(
        sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :k"),
        {"k": _KEY},
    )