"""bailian_usage_daily 折扣前(gross / 原价)成本列 + 历史回填。

Revision ID: 259
Revises: 258
Create Date: 2026-06-06

AI 大盘的金额一直是**折扣后**(Aliyun BSS ``PretaxAmount`` = 应付,税前)。
本次新增**折扣前**(原价 ``PretaxGrossAmount``)成本,使大盘能展示原价 /
折扣额 / 折扣率。

* ``cost_cny_gross`` / ``cost_usd_gross`` —— 与现有 ``cost_cny`` /
  ``cost_usd``(折后)成对,口径完全一致:CNY 为 BSS 原始币种,USD 用
  **每行摄取时冻结的同一汇率**派生(避免读时汇率漂移让 gross/net 不可比)。
  两列可空 —— 未来若出现没有原价口径的 ingest 路径(DashScope 等),行仍可落库。

* 一次性回填:原价数据**早已在库**,存于 ``raw_data.bss_item.PretaxGrossAmount``
  (BSS ``QueryAccountBill`` 每行原始 item),无需重拉。先从 JSON 取 CNY 原价,
  再按该行冻结汇率 ``cost_usd / cost_cny`` 推 USD 原价。只回填 NULL 行,幂等;
  ``cost_cny<=0`` 的行(被大盘跳过)不参与 USD 推算。

幂等:列创建走 ``_column_exists`` 守卫;回填只更新 ``IS NULL`` 行。Downgrade 删列。
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "259"
down_revision = "258"
branch_labels = None
depends_on = None

_TABLE = "bailian_usage_daily"


def _column_exists(table: str, column: str) -> bool:
    conn = op.get_bind()
    result = conn.execute(
        sa.text(
            "SELECT COUNT(*) FROM information_schema.columns "
            "WHERE table_schema = DATABASE() "
            "AND table_name = :t AND column_name = :c"
        ),
        {"t": table, "c": column},
    )
    return (result.scalar() or 0) > 0


def upgrade() -> None:
    if not _column_exists(_TABLE, "cost_cny_gross"):
        op.add_column(
            _TABLE,
            sa.Column("cost_cny_gross", sa.Numeric(14, 6), nullable=True),
        )
    if not _column_exists(_TABLE, "cost_usd_gross"):
        op.add_column(
            _TABLE,
            sa.Column("cost_usd_gross", sa.Numeric(14, 6), nullable=True),
        )

    conn = op.get_bind()
    # 1) CNY 原价 ← raw_data.bss_item.PretaxGrossAmount(数据已在库)。
    #    同样要求 cost_cny > 0,与步骤 2 的 USD 推算守卫对齐 —— 否则
    #    cost_cny<=0 的行会被填上 cost_cny_gross 却留 cost_usd_gross=NULL,
    #    两列不一致。BSS 摄取已跳过 PretaxAmount<=0,实际无此类行,纯防御。
    conn.execute(sa.text(
        f"UPDATE {_TABLE} "
        "SET cost_cny_gross = CAST("
        "  JSON_UNQUOTE(JSON_EXTRACT(raw_data, '$.bss_item.PretaxGrossAmount'))"
        "  AS DECIMAL(18, 6)) "
        "WHERE cost_cny_gross IS NULL "
        "  AND cost_cny > 0 "
        "  AND raw_data IS NOT NULL "
        "  AND JSON_EXTRACT(raw_data, '$.bss_item.PretaxGrossAmount') IS NOT NULL"
    ))
    # 2) USD 原价 ← CNY 原价 × 该行冻结汇率(cost_usd / cost_cny)。
    #    cost_cny<=0 的行(大盘已跳过)不推 USD,留 NULL。
    conn.execute(sa.text(
        f"UPDATE {_TABLE} "
        "SET cost_usd_gross = ROUND(cost_cny_gross * (cost_usd / cost_cny), 6) "
        "WHERE cost_usd_gross IS NULL "
        "  AND cost_cny_gross IS NOT NULL "
        "  AND cost_cny > 0"
    ))


def downgrade() -> None:
    if _column_exists(_TABLE, "cost_usd_gross"):
        op.drop_column(_TABLE, "cost_usd_gross")
    if _column_exists(_TABLE, "cost_cny_gross"):
        op.drop_column(_TABLE, "cost_cny_gross")