"""钓鱼邮件鉴别 + 安全找茬 content banks.

* ``arcade_phishing_samples`` — rendered emails (phish/legit + flags).
* ``arcade_spot_scenes``      — office scenes with element lists (hazards).

created_by FK matches identity_persons charset/collation (error-3780
guard). Seeded with a curated starter set; idempotent per table.
"""
from __future__ import annotations

import uuid

import sqlalchemy as sa

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

revision = "414"
down_revision = "413"
branch_labels = None
depends_on = None

PHISH = "arcade_phishing_samples"
SPOT = "arcade_spot_scenes"


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()
    pid = id_type_for("identity_persons")
    kw = table_kw_for("identity_persons")

    seed_phish = False
    if not _table_exists(bind, PHISH):
        op.create_table(
            PHISH,
            sa.Column("id", pid, primary_key=True),
            sa.Column("sender_name", sa.String(128), nullable=False),
            sa.Column("sender_email", sa.String(256), nullable=False),
            sa.Column("subject", sa.String(256), nullable=False),
            sa.Column("body", sa.Text(), nullable=False),
            sa.Column("is_phishing", sa.Boolean(), nullable=False),
            sa.Column("flags", sa.JSON(), nullable=False),
            sa.Column("explanation", sa.Text(), nullable=True),
            sa.Column("status", sa.String(16), nullable=False, server_default="active"),
            sa.Column("source", sa.String(16), nullable=False, server_default="seed"),
            sa.Column("created_by", pid, 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(["created_by"], ["identity_persons.id"], name="fk_arcade_phish_created_by"),
            **kw,
        )
        op.create_index("ix_arcade_phish_status", PHISH, ["status"])
        seed_phish = True

    seed_spot = False
    if not _table_exists(bind, SPOT):
        op.create_table(
            SPOT,
            sa.Column("id", pid, primary_key=True),
            sa.Column("title", sa.String(128), nullable=False),
            sa.Column("elements", sa.JSON(), nullable=False),
            sa.Column("status", sa.String(16), nullable=False, server_default="active"),
            sa.Column("source", sa.String(16), nullable=False, server_default="seed"),
            sa.Column("created_by", pid, 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(["created_by"], ["identity_persons.id"], name="fk_arcade_spot_created_by"),
            **kw,
        )
        op.create_index("ix_arcade_spot_status", SPOT, ["status"])
        seed_spot = True

    if seed_phish:
        from app.domains.arcade.content_seed import PHISHING_SEED

        op.bulk_insert(
            sa.table(
                PHISH,
                sa.column("id", pid),
                sa.column("sender_name", sa.String(128)),
                sa.column("sender_email", sa.String(256)),
                sa.column("subject", sa.String(256)),
                sa.column("body", sa.Text()),
                sa.column("is_phishing", sa.Boolean()),
                sa.column("flags", sa.JSON()),
                sa.column("explanation", sa.Text()),
                sa.column("status", sa.String(16)),
                sa.column("source", sa.String(16)),
            ),
            [
                {
                    "id": str(uuid.uuid4()),
                    "sender_name": s["sender_name"],
                    "sender_email": s["sender_email"],
                    "subject": s["subject"],
                    "body": s["body"],
                    "is_phishing": s["is_phishing"],
                    "flags": s["flags"],
                    "explanation": s["explanation"],
                    "status": "active",
                    "source": "seed",
                }
                for s in PHISHING_SEED
            ],
        )

    if seed_spot:
        from app.domains.arcade.content_seed import SPOT_SEED

        op.bulk_insert(
            sa.table(
                SPOT,
                sa.column("id", pid),
                sa.column("title", sa.String(128)),
                sa.column("elements", sa.JSON()),
                sa.column("status", sa.String(16)),
                sa.column("source", sa.String(16)),
            ),
            [
                {
                    "id": str(uuid.uuid4()),
                    "title": s["title"],
                    "elements": s["elements"],
                    "status": "active",
                    "source": "seed",
                }
                for s in SPOT_SEED
            ],
        )


def downgrade() -> None:
    bind = op.get_bind()
    for name in (SPOT, PHISH):
        if _table_exists(bind, name):
            op.drop_table(name)