"""Seed the AI-spending period-rollup refresh scheduled job.

Revision ID: 262
Revises: 261
Create Date: 2026-06-06

No schema change — registers the ``ai_spending.period_rollup_refresh`` cron
(daily 06:30 UTC, after cost_facts_sync ~04:00 + bailian.usage_sync ~05:00)
that materialises closed-month BucketRow sets into ai_spending_period_rollup
so the dashboard stops re-scanning ~2.7M cmdb_cost_facts rows per request.
Idempotent insert (skips when handler_key exists); hyphenated ``UUID()`` id to
avoid the no-dash-UUID attach bug (migrations 200/248).
"""
from __future__ import annotations

import json as _json
import uuid

import sqlalchemy as sa

from alembic import op

revision = "262"
down_revision = "261"
branch_labels = None
depends_on = None

_HANDLER = "ai_spending.period_rollup_refresh"
_DESC = (
    "每天 06:30 UTC 重算并物化已关闭月份的 AI 支出 BucketRow 集到 "
    "ai_spending_period_rollup;最近 2 个闭月每跳必算(吸收迟到 facts + "
    "新关闭月预热),更早闭月仅缺当前指纹时补算。闭月读直接命中物化表,"
    "免去每请求扫 270 万行 cmdb_cost_facts。"
)


def _table_exists(name: str) -> bool:
    return op.get_bind().execute(
        sa.text(
            "SELECT 1 FROM information_schema.TABLES "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n LIMIT 1"
        ),
        {"n": name},
    ).scalar() is not None


def upgrade() -> None:
    if not _table_exists("scheduled_jobs"):
        return
    op.get_bind().execute(
        sa.text(
            "INSERT INTO scheduled_jobs "
            "(id, name, display_name, description, handler_key, "
            " trigger_type, cron_expr, timezone, enabled, config, "
            " run_count, failure_count, created_at, updated_at) "
            "SELECT :id, :name, :display, :desc, :handler, 'cron', "
            ":cron, 'UTC', 1, :config, 0, 0, NOW(), NOW() "
            "FROM DUAL WHERE NOT EXISTS "
            "(SELECT 1 FROM scheduled_jobs WHERE handler_key = :handler)"
        ),
        {
            "id": str(uuid.uuid4()),
            "name": _HANDLER,
            "display": "AI 支出 · 周期物化表刷新",
            "desc": _DESC,
            "handler": _HANDLER,
            "cron": "30 6 * * *",
            "config": _json.dumps({"months_back": 13, "recent_months": 2}),
        },
    )


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