"""安全宣贯:环境位 —— 气泡的频控开关,以及安全提示补日文。

## 频控是这个 PR 的主要内容,而它落在 platform_settings 上

三个键,带描述播种进 `platform_settings`,这样管理端的平台设置页能看见它们,
而不是只有代码里的默认值:

| key | 默认 | 是什么 |
|---|---|---|
| `security_awareness.bubble.enabled` | true | 气泡是否展示宣贯素材 |
| `security_awareness.bubble.daily_new_materials` | 1 | 一个人一天最多**遇到几个新素材** |
| `security_awareness.bubble.max_repeats_per_material` | 3 | 同一个素材对同一个人最多展示几次 |

播种而不是只留代码默认值,是因为**一个只存在于代码里的阈值,运维时改不动**。
读取端(`ambient.load_settings`)对坏值一律退回默认并记 warning:气泡挂在每
一个已登录页面上,一个畸形设置不能把应用外壳搞崩。

## `platform_security_tips.text_ja`

这张表是 2 语的(`text_zh` / `text_en`),而平台说三种语言。气泡在没有可展示
素材时会回退到这些提示,于是日语读者会拿到英文 —— 这不算错,但这张表**没有
一列能存日文**,所以连想修都修不了。补上这一列,并**不做回填**:NULL 就是
「还没翻译」的真值,渲染侧按 `ja → en → zh` 回退,而把英文复制进日文列会把
「没翻译」伪装成「翻译过了」,之后再也分不出来哪些是真的翻译过。

Revision ID: 598
Revises: 597

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

import json

import sqlalchemy as sa

from alembic import op

revision = "598"
down_revision = "597"
branch_labels = None
depends_on = None

TIPS = "platform_security_tips"
SETTINGS = "platform_settings"

_SETTINGS: tuple[tuple[str, object, str], ...] = (
    (
        "security_awareness.bubble.enabled",
        True,
        "右下角吉祥物气泡是否展示安全宣贯素材(漫画 / 文章 / 外链)。"
        "关掉之后气泡回退到旧的纯文本安全提示,不影响学习中心和宣贯活动。",
    ),
    (
        "security_awareness.bubble.daily_new_materials",
        1,
        "一个人一天最多在气泡里**遇到几个新素材**。默认 1 —— 环境位靠重复"
        "一个想法起作用,不是靠多。已经在今天展示过的素材不受这个上限影响"
        "(否则达到上限那一刻气泡会突然空掉),这个数管的是「新想法」的节奏。",
    ),
    (
        "security_awareness.bubble.max_repeats_per_material",
        3,
        "同一个素材对同一个人最多展示几次。默认 3 —— 三次之后它要么已经起"
        "作用了,要么不会起作用,第四次只会教会这个人忽略气泡。员工点开过的"
        "素材直接不再进气泡,不受这个数控制。",
    ),
)


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 _has_column(bind, table: str, column: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.columns "
            "WHERE table_schema = DATABASE() "
            "AND table_name = :t AND column_name = :c"
        ),
        {"t": table, "c": column},
    ).scalar() is not None


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

    # 三语补齐。刻意可空、刻意不回填 —— 见模块文档。
    if _has_table(bind, TIPS) and not _has_column(bind, TIPS, "text_ja"):
        op.add_column(TIPS, sa.Column("text_ja", sa.String(500), nullable=True))

    if _has_table(bind, SETTINGS):
        for key, value, description in _SETTINGS:
            bind.execute(
                sa.text(
                    f"INSERT INTO {SETTINGS} (`key`, `value`, description, "
                    " created_at, updated_at) "
                    "SELECT :k, :v, :d, NOW(), NOW() FROM DUAL "
                    f"WHERE NOT EXISTS (SELECT 1 FROM {SETTINGS} "
                    " WHERE `key` = :k)"
                ),
                {"k": key, "v": json.dumps(value), "d": description},
            )


def downgrade() -> None:
    bind = op.get_bind()
    if _has_table(bind, SETTINGS):
        for key, _value, _description in _SETTINGS:
            bind.execute(
                sa.text(f"DELETE FROM {SETTINGS} WHERE `key` = :k"), {"k": key},
            )
    if _has_table(bind, TIPS) and _has_column(bind, TIPS, "text_ja"):
        op.drop_column(TIPS, "text_ja")