"""Platform Impact AI summary — cache table + weekly cron seed.

Revision ID: 282
Revises: 281
Create Date: 2026-06-12

1. **``insights_impact_summaries``** — cached bilingual LLM narrative
   for the leadership 平台成果 page. Mirrors ``ai_spending_summaries``
   (079): one row per ``(period, language)`` where ``period`` is the
   as-of DATE the aggregates were read; manual regenerate replaces the
   same-day row in place; readers take the newest row.
   ``generated_by_user_id`` mirrors ``identity_persons.id`` type AND
   collation via ``_fk_source_col_type`` + out-of-line FK (TiDB 3780
   discipline — see migrations 073/074 and PR #980).

2. **``scheduled_jobs``** seed — ``insights.impact_summary_generate``,
   weekly Monday 03:10 Asia/Shanghai (after the nightly crons quiet
   down, before CN work hours). Enabled; the handler soft-skips with a
   recorded reason when ``ai_spending.llm.api_key`` isn't configured.
   Hyphenated ``uuid4`` id (no-dash-UUID attach bug — 059/162/248).

Idempotent: table guarded by information_schema check; job insert
guarded by NOT EXISTS on handler_key.
"""
from __future__ import annotations

import uuid

import sqlalchemy as sa

from alembic import op

revision = "282"
down_revision = "281"
branch_labels = None
depends_on = None

_HANDLER = "insights.impact_summary_generate"
_DESC = (
    "每周一 03:10(Asia/Shanghai)用 Bailian LLM 生成平台成果页的双语高管"
    "摘要并落 insights_impact_summaries。复用 ai_spending.llm.* 平台级"
    "凭据;未配置 api_key 时记录跳过原因后退出。管理员也可在页面上手动"
    "重新生成(每人每日限额)。"
)


def _table_exists(bind, name: str) -> bool:
    return 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 _fk_source_col_type(
    bind,
    target_table: str,
    target_column: str,
    *,
    fallback_type: str = "char",
    fallback_collation: str | None = "utf8mb4_unicode_ci",
    length: int = 36,
):
    """Mirror an FK target's data type AND collation on the source
    column (TiDB errno-3780 discipline; canonical pattern from
    migrations 073/074/079)."""
    row = bind.execute(
        sa.text(
            "SELECT DATA_TYPE, COLLATION_NAME FROM "
            "information_schema.COLUMNS "
            "WHERE TABLE_SCHEMA = DATABASE() "
            "  AND TABLE_NAME = :t AND COLUMN_NAME = :c"
        ),
        {"t": target_table, "c": target_column},
    ).fetchone()
    if row:
        data_type = (row[0] or fallback_type).lower()
        collation = row[1] or fallback_collation
    else:
        data_type = fallback_type.lower()
        collation = fallback_collation
    type_cls = sa.CHAR if data_type == "char" else sa.String
    if collation:
        return type_cls(length, collation=collation)
    return type_cls(length)


def upgrade() -> None:
    bind = op.get_bind()

    if not _table_exists(bind, "insights_impact_summaries"):
        identity_persons_id_col = _fk_source_col_type(
            bind, "identity_persons", "id",
        )
        op.create_table(
            "insights_impact_summaries",
            sa.Column("id", sa.CHAR(36), primary_key=True),
            sa.Column("period", sa.String(10), nullable=False),
            sa.Column("language", sa.String(16), nullable=False),
            sa.Column("summary_md_en", sa.Text, nullable=True),
            sa.Column("summary_md_zh", sa.Text, nullable=True),
            sa.Column("highlights", sa.JSON, nullable=True),
            sa.Column("aggregates", sa.JSON, nullable=True),
            sa.Column("model_used", sa.String(128), nullable=True),
            sa.Column("input_token_count", sa.Integer, nullable=True),
            sa.Column("output_token_count", sa.Integer, nullable=True),
            sa.Column(
                "generated_by_user_id", identity_persons_id_col,
                nullable=True,
            ),
            sa.Column(
                "generated_at", sa.DateTime(timezone=True),
                server_default=sa.func.now(), nullable=False,
            ),
            sa.Column(
                "created_at", sa.DateTime(timezone=True),
                server_default=sa.func.now(), nullable=False,
            ),
            sa.Column(
                "updated_at", sa.DateTime(timezone=True),
                server_default=sa.func.now(), onupdate=sa.func.now(),
                nullable=False,
            ),
            sa.UniqueConstraint(
                "period", "language",
                name="uq_insights_impact_summaries_period_lang",
            ),
        )
        try:
            op.create_foreign_key(
                "fk_insights_impact_summaries_user",
                "insights_impact_summaries",
                "identity_persons",
                ["generated_by_user_id"],
                ["id"],
                ondelete="SET NULL",
            )
        except Exception:  # noqa: BLE001 — FK is best-effort hardening
            pass

    if _table_exists(bind, "scheduled_jobs"):
        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, 'Asia/Shanghai', 1, '{}', 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": "10 3 * * 1",
            },
        )


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