"""Identity data-quality findings + runs + the daily scan cron.

Backs the Identity-Hub data-quality panel (PR3). Two tables:
``identity_data_quality_findings`` (deduped per (finding_type, dedup_key))
and ``identity_data_quality_runs`` (per-scan trend), plus a seeded
``identity.data_quality_scan`` cron (hyphenated UUID id to avoid the
no-dash-UUID attach bug — migrations 200/248/267).

Revises: 268 (widen_idempotency_scope — landed on main 2026-06-09 while this
stack was in review; renumbered 268→269 to resolve the revision-id race).
"""
from __future__ import annotations

import json
import uuid

import sqlalchemy as sa

from alembic import op
from app.core.migration_helpers import id_type_for, table_kw_for

revision = "269"
down_revision = "268"
branch_labels = None
depends_on = None

FINDINGS_TABLE = "identity_data_quality_findings"
RUNS_TABLE = "identity_data_quality_runs"
HANDLER_KEY = "identity.data_quality_scan"


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"
        ),
        {"n": name},
    ).scalar() is not None


def upgrade() -> None:
    bind = op.get_bind()
    persons_id = id_type_for("identity_persons")
    kw = table_kw_for("identity_persons")

    if not _table_exists(bind, FINDINGS_TABLE):
        op.create_table(
            FINDINGS_TABLE,
            sa.Column("id", persons_id, primary_key=True),
            sa.Column("finding_type", sa.String(64), nullable=False),
            sa.Column("dedup_key", sa.String(255), nullable=False),
            sa.Column(
                "severity", sa.String(16), nullable=False,
                server_default="medium",
            ),
            sa.Column("person_id", persons_id, nullable=True),
            sa.Column("subject_display", sa.String(255), nullable=True),
            sa.Column("title", sa.String(512), nullable=False),
            sa.Column("detail", sa.JSON(), nullable=True),
            sa.Column(
                "status", sa.String(16), nullable=False,
                server_default="open",
            ),
            sa.Column(
                "auto_resolved", sa.Boolean(), nullable=False,
                server_default="0",
            ),
            sa.Column(
                "first_seen_at", sa.DateTime(timezone=True), nullable=False,
                server_default=sa.func.now(),
            ),
            sa.Column(
                "last_seen_at", sa.DateTime(timezone=True), nullable=False,
                server_default=sa.func.now(),
            ),
            sa.Column("resolved_at", sa.DateTime(timezone=True), nullable=True),
            sa.Column("decided_by_person_id", persons_id, nullable=True),
            sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True),
            sa.Column("decision_note", sa.Text(), nullable=True),
            sa.Column(
                "created_at", sa.DateTime(timezone=True), nullable=False,
                server_default=sa.func.now(),
            ),
            sa.Column(
                "updated_at", sa.DateTime(timezone=True), nullable=False,
                server_default=sa.func.now(),
            ),
            sa.ForeignKeyConstraint(
                ["person_id"], ["identity_persons.id"],
                name="fk_identity_dq_finding_person", ondelete="SET NULL",
            ),
            sa.ForeignKeyConstraint(
                ["decided_by_person_id"], ["identity_persons.id"],
                name="fk_identity_dq_finding_decider", ondelete="SET NULL",
            ),
            sa.UniqueConstraint(
                "finding_type", "dedup_key",
                name="uq_identity_dq_finding_dedup",
            ),
            **kw,
        )
        op.create_index(
            "ix_identity_dq_finding_status_sev", FINDINGS_TABLE,
            ["status", "severity"],
        )
        op.create_index(
            "ix_identity_dq_finding_type_status", FINDINGS_TABLE,
            ["finding_type", "status"],
        )
        op.create_index(
            "ix_identity_dq_finding_person", FINDINGS_TABLE, ["person_id"],
        )

    if not _table_exists(bind, RUNS_TABLE):
        op.create_table(
            RUNS_TABLE,
            sa.Column("id", persons_id, primary_key=True),
            sa.Column("started_at", sa.DateTime(timezone=True), nullable=False),
            sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
            sa.Column("duration_ms", sa.Integer(), nullable=True),
            sa.Column("status", sa.String(16), nullable=False),
            sa.Column(
                "trigger_source", sa.String(64), nullable=False,
                server_default="cron",
            ),
            sa.Column("open_total", sa.Integer(), nullable=True),
            sa.Column("open_high", sa.Integer(), nullable=True),
            sa.Column("new_count", sa.Integer(), nullable=True),
            sa.Column("resolved_count", sa.Integer(), nullable=True),
            sa.Column("by_type", sa.JSON(), nullable=True),
            sa.Column("error_message", sa.Text(), nullable=True),
            **kw,
        )
        op.create_index(
            "ix_identity_dq_runs_started", RUNS_TABLE, ["started_at"],
        )

    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, '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_KEY,
                "display": "身份数据质量扫描",
                "desc": (
                    "每天 05:00 UTC 扫描身份图谱:离职/已删除仍可登录、HRIS 已清除"
                    "仍可登录、汇报对象已离职、缺汇报对象、下游同步持续失败。发现"
                    "即去重 upsert,消失即自动解决;新增高危去重推送 IT 管理员。"
                ),
                "handler": HANDLER_KEY,
                "cron": "0 5 * * *",
                "config": json.dumps({}),
            },
        )


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