"""安全宣贯:素材库 + 曝光台账。

## 这是在解决什么

安全宣贯今天的形状是:右下角吉祥物气泡轮播 `platform_security_tips` 里的
**一行纯文本**(500 字符、只有中英两种语言、全局同一份),制度中心
`platform_policy_notices` 挂一堆**飞书外链**。两者都**不记录任何曝光**。

所以「证明这个员工被告知过钓鱼风险」这个问题,现在唯一诚实的答案是
「答不出来」—— 生产里 `platform_security_tips` 的读接口连 person_id 都
不需要。

这个迁移建的两张表就是为了让它答得出来:

| 表 | 是什么 |
|---|---|
| `secaw_materials` | 托管的素材:漫画(多张图)/ 文章(markdown 正文)/ 外链,每种最多三种语言 |
| `secaw_impressions` | **只追加**的曝光台账:谁、看了哪个素材、在哪个面、什么时候、看到的是第几版、渲染的是哪种语言 |

## 为什么本地化内容存 JSON 而不是 18 个列

三种语言 ×(标题 / 摘要 / 正文 / 图片列表 / 外链 / 封面)= 18 个绝大多数
时候是 NULL 的列,而它们之间唯一的关系 ——「这 6 个属于同一种语言」——
schema 本来也表达不出来。JSON 把这个分组显式化了,而且「这个素材有哪几种
语言」变成 `localized.keys()`,不是 6 次 NULL 判断。

形状在 `app/domains/security_awareness/schemas.py::LocalizedContent` 里强制
校验,service 只写 `normalized_localized()` 的返回值 —— 没有任何代码路径
直接往这一列塞 dict。`zh` 必填(语言回退链终点在它)。

图片列存的是 **S3 object key,不是 URL**:URL 要么在列里过期,要么更糟
—— 是一条永久公开链接。每次读都现签 10 分钟的 presigned GET。

## 为什么 material_id 上刻意没有 CASCADE

素材只归档、不删除(`status='archived'`)。真删只在「一次都没展示过」时
允许(service 里有闸),因为一条已经展示过的素材是**证据**:删了它会让
台账变成孤儿,并且悄悄把某个可能已经上报出去的覆盖率数字改小。
`person_id` 上是 CASCADE —— 硬删 person 只发生在数据纠正,那时候他的曝光
记录本来就是要一起清掉的个人数据。

Revision ID: 596
Revises: 595

⚠️ 迁移号要对着 `origin/main` 抢,而且每次 push 之前重新抢一次 —— 本仓踩过的
号是 462 / 472 / 582 / 587,两个分支同时拿同一个号,合进 main 之后链上出现
两个 head,后端每个测试分片和镜像构建一起红。
"""
from __future__ import annotations

import sqlalchemy as sa

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

revision = "596"
down_revision = "595"
branch_labels = None
depends_on = None

MATERIALS = "secaw_materials"
IMPRESSIONS = "secaw_impressions"


def _has_table(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()
    # FK 列的类型一律向 id_type_for 要 —— TiDB 3780 只认「和父列完全一致」,
    # 而 CI 的 TiDB parity 闸对手写 String(36)+FK 一票否决(它就是为 462/472
    # 两次生产卡链加的)。
    person_fk = id_type_for("identity_persons")

    if not _has_table(bind, MATERIALS):
        op.create_table(
            MATERIALS,
            sa.Column("id", sa.String(36), primary_key=True),
            # comic | article | link
            sa.Column(
                "kind", sa.String(16), nullable=False, server_default="article",
            ),
            # catalog.MATERIAL_CATEGORIES —— 闭集。自由文本的分类一年之后
            # 会有四十种「钓鱼」的写法,覆盖率报表就再也答不出「我们做了
            # 多少钓鱼宣贯」。
            sa.Column(
                "category", sa.String(32), nullable=False, server_default="other",
            ),
            sa.Column("tags", sa.JSON()),
            # {"zh": {...}, "en": {...}, "ja": {...}} —— 见模块文档。
            sa.Column("localized", sa.JSON(), nullable=False),
            # draft | published | archived
            sa.Column(
                "status", sa.String(16), nullable=False, server_default="draft",
            ),
            sa.Column(
                "is_pinned", sa.Boolean(), nullable=False, server_default="0",
            ),
            sa.Column(
                "display_order", sa.Integer(), nullable=False, server_default="0",
            ),
            sa.Column("estimated_minutes", sa.Integer()),
            # 内容变更时 +1。宣贯活动会记下自己发出去时的版本号,所以
            # 「我确认的时候文章写的不是这个」有答案。
            sa.Column(
                "version", sa.Integer(), nullable=False, server_default="1",
            ),
            sa.Column("published_at", sa.DateTime(timezone=True)),
            sa.Column("archived_at", sa.DateTime(timezone=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.Column(
                "created_by_person_id", person_fk,
                sa.ForeignKey("identity_persons.id", ondelete="SET NULL"),
            ),
            sa.Column(
                "updated_by_person_id", person_fk,
                sa.ForeignKey("identity_persons.id", ondelete="SET NULL"),
            ),
        )
        op.create_index(
            "ix_secaw_materials_live", MATERIALS, ["status", "published_at"],
        )
        op.create_index(
            "ix_secaw_materials_category", MATERIALS, ["category", "status"],
        )
        op.create_index(
            "ix_secaw_materials_pinned", MATERIALS,
            ["is_pinned", "display_order"],
        )

    if not _has_table(bind, IMPRESSIONS):
        op.create_table(
            IMPRESSIONS,
            sa.Column("id", sa.String(36), primary_key=True),
            sa.Column(
                "person_id", person_fk,
                sa.ForeignKey("identity_persons.id", ondelete="CASCADE"),
                nullable=False,
            ),
            # 刻意没有 ondelete —— 见模块文档。
            sa.Column(
                "material_id", id_type_for(MATERIALS),
                sa.ForeignKey(f"{MATERIALS}.id"),
                nullable=False,
            ),
            # detail | bubble | campaign_card
            sa.Column("surface", sa.String(24), nullable=False),
            # 写入时从素材上拷过来的版本号:素材之后还会被编辑,而争议
            # 恰恰是「他当时看到的是哪一版」。
            sa.Column(
                "material_version", sa.Integer(), nullable=False,
                server_default="1",
            ),
            sa.Column(
                "lang", sa.String(8), nullable=False, server_default="zh",
            ),
            sa.Column(
                "occurred_at", sa.DateTime(timezone=True),
                nullable=False, server_default=sa.func.now(),
            ),
        )
        # 去重探针 + 「我看过没有」的逐人读。
        op.create_index(
            "ix_secaw_impr_person_material", IMPRESSIONS,
            ["person_id", "material_id", "occurred_at"],
        )
        # 管理端的逐素材触达汇总。
        op.create_index(
            "ix_secaw_impr_material", IMPRESSIONS,
            ["material_id", "occurred_at"],
        )
        # 环境位频控要问的「这个人今天已经被展示了多少」。先建好,免得
        # 那个闸上线时还要再动一次 schema。
        op.create_index(
            "ix_secaw_impr_person_time", IMPRESSIONS,
            ["person_id", "occurred_at"],
        )


def downgrade() -> None:
    bind = op.get_bind()
    # 先删子表:MySQL / TiDB 都不允许先删被 FK 指着的父表。
    if _has_table(bind, IMPRESSIONS):
        op.drop_table(IMPRESSIONS)
    if _has_table(bind, MATERIALS):
        op.drop_table(MATERIALS)