"""阿里云折前口径 × 财务数仓(BigQuery)月度对账。

## 为什么需要它

Helix 的「业务折前成本」是从阿里云 BSS ``DescribeInstanceBill`` 逐行摄取
后按行白名单(``cost_bucket='usage'``)聚合出来的;财务侧口径在
BigQuery ``gcp-digital-bi.alibaba_cloud.monthly_en_*`` 上,规则是
「保留 Pay-As-You-Go / Purchase / Renewal / Upgrade-Downgrade,取
``pretax_gross_amount``,按 ``owner_account_id`` 汇总」。

两条链路**各自独立**:一个走 API 逐日拉,一个是数仓月表。口径设计上一致
(见 :mod:`app.domains.cmdb.cost_view` 的行白名单说明),但「设计一致」
和「今天仍然一致」是两回事 —— 驱动改一次分类、财务改一次过滤规则,就会
悄悄分叉,而**没有任何东西会报错**。这个对账把「一致」变成每月一次的
可证事实。

2026-08-16 首次实测(2026-08 当月至今,直接打两边):

===================  ===========  ===========  ========================
账号                  BQ(财务)     Helix        差异
===================  ===========  ===========  ========================
5901860086223871     172,640.19   166,781.73   −3.4%(Helix 少最后一天)
5108739859105002      67,833.73    65,513.50   −3.4%
5962436752113096       8,517.30     8,199.71   −3.7%
===================  ===========  ===========  ========================

逐服务对照证明差异**全部**来自数据新鲜度而非口径:Simple Log Service
两边都是 $88.94、Marketplace 都是 $0.55、ECS 都是 $0.04,只有当天仍在
增长的 Model Studio 差一天的量(Helix 末日 8/13,BQ 已到 8/14)。

## 为什么只做「已结月」

当月两边都在动,差异恒为噪声。默认对**上一个完整自然月**做,那时两边都
已定稿,任何差异都是真信号。

## 不做什么

**不改任何数字。** 这个模块只读、只比、只报告。发现漂移时它的产出是一条
告警和一份逐账号的差异清单,由人去判断该修哪一边 —— 自动「对齐」到某一
侧会把一个已知的分歧变成一个隐藏的分歧。
"""
from __future__ import annotations

import asyncio
import re
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Any

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

from app.core.safe_logging import get_safe_logger
from app.domains.cmdb.cost_view import fact_engineering_amount
from app.domains.cmdb.models import CmdbCloudScope, CmdbCostFact
from app.integrations.models import Integration

logger = get_safe_logger(__name__)

# 财务口径:保留的 ``item`` 值。与 ``cost_view`` 的行白名单一一对应 ——
# 排除 Refund / Reconciliation / Round-off Discount(= adjustment / other
# 桶)、SP 与资源包购买费(= commitment_fee)、集中采购 Support
# (= support);不减 invoice_discount / deducted_by_coupons
# (= contract_discount / credit 不参与相减);不使用实付字段。
FINANCE_KEPT_ITEMS: tuple[str, ...] = (
    "Pay-As-You-Go",
    "Purchase",
    "Renewal",
    "Upgrade/Downgrade",
)

# 数仓表名模板。财务按月一张表:``monthly_en_YYYYMM01``。
BQ_DATASET = "gcp-digital-bi.alibaba_cloud"
BQ_TABLE_PREFIX = "monthly_en_"

# 超过这个相对差异就报警(绝对值)。已结月两边都定稿,真实分歧通常远小于
# 它;设 1% 是为了容纳汇率/四舍五入的尾差而不是容纳口径分歧。
DEFAULT_DRIFT_THRESHOLD_PCT = 1.0
# 金额太小的账号不参与相对差异判定 —— $0.01 对 $0.02 是 100% 漂移但毫无
# 意义。低于这个绝对额只在报告里列出,不触发告警。
MIN_MATERIAL_USD = Decimal("50")

_TABLE_SUFFIX_RE = re.compile(r"^\d{8}$")


def bq_table_for_month(month_start: date) -> str:
    """``date(2026, 7, 1)`` → ``gcp-digital-bi.alibaba_cloud.monthly_en_20260701``。

    表名分量自己拼、并做形状校验 —— 它会进 SQL 字符串(BigQuery 的表名
    不能参数化),所以绝不能让调用方的输入直接落进去。
    """
    suffix = f"{month_start.year:04d}{month_start.month:02d}01"
    if not _TABLE_SUFFIX_RE.match(suffix):
        raise ValueError(f"bad month {month_start!r}")
    return f"{BQ_DATASET}.{BQ_TABLE_PREFIX}{suffix}"


def finance_caliber_sql(table: str) -> str:
    """财务口径 SQL —— 逐字对应他们给的那条。

    ``item`` 白名单用字面量拼进去:值来自本模块的常量而非外部输入,且
    ``FINANCE_KEPT_ITEMS`` 里没有引号/反斜杠(有断言守着)。
    """
    for item in FINANCE_KEPT_ITEMS:
        if "'" in item or "\\" in item:
            raise ValueError(f"unsafe item literal {item!r}")
    items = ", ".join(f"'{i}'" for i in FINANCE_KEPT_ITEMS)
    return (
        "SELECT owner_account_id, owner_account_name, "
        "SUM(pretax_gross_amount) AS business_list_cost\n"
        f"FROM `{table}`\n"
        f"WHERE item IN ({items})\n"
        "GROUP BY owner_account_id, owner_account_name"
    )


@dataclass(slots=True)
class AccountDrift:
    account_id: str
    account_name: str | None
    bq_usd: Decimal
    helix_usd: Decimal

    @property
    def delta_usd(self) -> Decimal:
        return self.helix_usd - self.bq_usd

    @property
    def delta_pct(self) -> float | None:
        """相对 BQ(财务侧)的偏差。BQ 为 0 时无意义 → None。"""
        if self.bq_usd == 0:
            return None
        return float(self.delta_usd / self.bq_usd * 100)

    @property
    def is_material(self) -> bool:
        """够大到值得报警吗。"""
        if max(abs(self.bq_usd), abs(self.helix_usd)) < MIN_MATERIAL_USD:
            return False
        pct = self.delta_pct
        return pct is not None and abs(pct) > DEFAULT_DRIFT_THRESHOLD_PCT

    def as_dict(self) -> dict[str, Any]:
        pct = self.delta_pct
        return {
            "account_id": self.account_id,
            "account_name": self.account_name,
            "bq_usd": float(round(self.bq_usd, 2)),
            "helix_usd": float(round(self.helix_usd, 2)),
            "delta_usd": float(round(self.delta_usd, 2)),
            "delta_pct": round(pct, 2) if pct is not None else None,
            "material": self.is_material,
        }


async def helix_list_cost_by_account(
    session: AsyncSession, *, month_start: date, month_end_excl: date,
) -> dict[str, Decimal]:
    """Helix 侧同口径:阿里云各账号的折前成本(按 scope 的 external_id)。

    金额表达式复用 :func:`fact_engineering_amount` —— **不重新实现口径**。
    对账的价值全在「两边各自独立算」,而 Helix 这一侧必须是页面上真正在
    用的那个表达式,否则对的是一个没人看的数。
    """
    rows = (await session.execute(
        select(
            CmdbCloudScope.external_id,
            func.coalesce(func.sum(fact_engineering_amount()), 0),
        )
        .select_from(CmdbCostFact)
        .join(CmdbCloudScope, CmdbCloudScope.id == CmdbCostFact.scope_id)
        .join(Integration, Integration.id == CmdbCostFact.integration_id)
        .where(
            Integration.provider.in_(("alicloud_intl", "aliyun")),
            CmdbCostFact.date >= month_start,
            CmdbCostFact.date < month_end_excl,
        )
        .group_by(CmdbCloudScope.external_id),
    )).all()
    return {
        str(acct): Decimal(str(total or 0))
        for acct, total in rows
        if acct
    }


def build_drift_report(
    *,
    month_start: date,
    bq_rows: list[dict[str, Any]],
    helix_by_account: dict[str, Decimal],
) -> dict[str, Any]:
    """把两侧结果合成一份逐账号的差异报告。

    **两侧的账号取并集**,不是交集:只在一边出现的账号才是最值得看的
    信号(Helix 缺一个账号 = 整个账号的花费没进任何看板;BQ 缺 =
    我们在统计一个财务不认的账号)。取交集会把这两种情况一起藏掉。
    """
    bq_by_account: dict[str, Decimal] = {}
    names: dict[str, str | None] = {}
    for row in bq_rows:
        acct = str(row.get("owner_account_id") or "").strip()
        if not acct:
            continue
        bq_by_account[acct] = Decimal(str(row.get("business_list_cost") or 0))
        names[acct] = row.get("owner_account_name")

    drifts = [
        AccountDrift(
            account_id=acct,
            account_name=names.get(acct),
            bq_usd=bq_by_account.get(acct, Decimal(0)),
            helix_usd=helix_by_account.get(acct, Decimal(0)),
        )
        for acct in sorted(set(bq_by_account) | set(helix_by_account))
    ]
    drifts.sort(key=lambda d: abs(d.delta_usd), reverse=True)
    material = [d for d in drifts if d.is_material]
    return {
        "month": month_start.strftime("%Y-%m"),
        "bq_table": bq_table_for_month(month_start),
        "accounts_compared": len(drifts),
        "bq_total_usd": float(round(sum(bq_by_account.values(), Decimal(0)), 2)),
        "helix_total_usd": float(
            round(sum(helix_by_account.values(), Decimal(0)), 2),
        ),
        "material_drift_count": len(material),
        "threshold_pct": DEFAULT_DRIFT_THRESHOLD_PCT,
        "min_material_usd": float(MIN_MATERIAL_USD),
        "accounts": [d.as_dict() for d in drifts],
    }


async def fetch_bq_finance_rows(
    session: AsyncSession, *, month_start: date,
) -> list[dict[str, Any]]:
    """跑财务口径 SQL,拿回 ``[{owner_account_id, owner_account_name,
    business_list_cost}]``。

    凭证复用 **GCP 集成的服务账号**(``gcp-primary``)—— 财务数仓在
    ``gcp-digital-bi``,而 Helix 已经有那个租户的 SA;不为对账单独引入
    第二套凭证。SA 需要 ``alibaba_cloud`` 数据集的 ``dataViewer``
    (2026-08-16 已授:``pch-resource-manager@gcp-digital-infra``)。

    权限缺失时**抛出可操作的错误**而不是静默返回空 —— 空结果会被下游
    读成「财务侧这个月没花钱」,进而报出一份全是负漂移的假报告。
    """
    from app.integrations.service import build_driver_by_id

    driver = await build_driver_by_id("gcp-primary", session=session)
    table = bq_table_for_month(month_start)
    sql = finance_caliber_sql(table)
    return await asyncio.to_thread(_run_bq_sync, driver, table, sql)


def _run_bq_sync(driver: Any, table: str, sql: str) -> list[dict[str, Any]]:
    try:
        from google.cloud import bigquery  # type: ignore
    except ImportError as exc:  # pragma: no cover — 部署缺依赖
        raise RuntimeError(
            "google-cloud-bigquery is not installed; cannot reconcile "
            "against the finance warehouse",
        ) from exc

    # Client 绑到**表所在项目**(``gcp-digital-bi``),不是 SA 的 home
    # 项目 —— 与 ``gcp_cloud._list_cost_facts_sync`` 同一个理由:SA 的
    # home 项目可能没开 BigQuery API,job 会硬失败。
    bq_project = table.split(".", 1)[0]
    client = bigquery.Client(
        credentials=driver._credentials(), project=bq_project,
    )
    try:
        rows = list(client.query(sql).result())
    except Exception as exc:  # noqa: BLE001 — 转成可操作的说明
        raise RuntimeError(
            f"BigQuery reconciliation query failed on {table}: {exc}. "
            f"If this is a permission error, grant "
            f"roles/bigquery.dataViewer on the {BQ_DATASET} dataset to "
            f"the gcp-primary integration's service account.",
        ) from exc
    return [
        {
            "owner_account_id": r.get("owner_account_id"),
            "owner_account_name": r.get("owner_account_name"),
            "business_list_cost": r.get("business_list_cost"),
        }
        for r in rows
    ]


async def reconcile_month(
    session: AsyncSession, *, month_start: date,
) -> dict[str, Any]:
    """对某个自然月做一次对账,返回逐账号差异报告(**只读**)。"""
    if month_start.day != 1:
        raise ValueError(f"month_start must be the 1st, got {month_start}")
    month_end_excl = (
        date(month_start.year + 1, 1, 1) if month_start.month == 12
        else date(month_start.year, month_start.month + 1, 1)
    )
    bq_rows = await fetch_bq_finance_rows(session, month_start=month_start)
    helix = await helix_list_cost_by_account(
        session, month_start=month_start, month_end_excl=month_end_excl,
    )
    report = build_drift_report(
        month_start=month_start, bq_rows=bq_rows, helix_by_account=helix,
    )
    logger.info(
        "cmdb.cost_recon.aliyun_bq month=%s accounts=%s material=%s "
        "bq_total=%s helix_total=%s",
        report["month"], report["accounts_compared"],
        report["material_drift_count"], report["bq_total_usd"],
        report["helix_total_usd"],
    )
    return report