"""宣贯素材机器翻译:``secaw_materials.mt_langs`` + 自动补齐 cron。

起因是一条员工反馈:**英文界面右下角的安全学习气泡是中文的。**

排查结论不是语言链路坏了 —— 前端把 i18next 的当前语言传给
``/security-awareness/bubble``,``service.pick_block`` 按
``catalog.LANG_FALLBACK`` 解析并回报 ``is_fallback_lang``,这一路都对。
缺的是**内容**:2026-09-05 线上 46 条已发布素材里,30 条漫画只有 ``zh``
一种语言,16 条文章没有 ``ja``。回退链于是老老实实地回退到了中文。

## 本迁移做两件事

1. ``secaw_materials`` 加 ``mt_langs JSON NULL`` —— 记录哪几种语言的
   block 是**模型写的、还没有人校对过**。

   为什么是一列而不是塞进 ``localized`` 的 block 里:写入路径
   ``schemas.MaterialWrite.normalized_localized()`` 是白名单,任何多余
   的键都会在管理员下一次保存时被丢掉 —— 包括只改 ``kind`` 的那种保存
   (它会把三种语言整个过一遍同一个校验器)。那样标记会因为跟「有没有人
   校对」完全无关的原因消失。放在列上,它只被两件事清掉:某种语言的正文
   真的被人改了(``service.update`` 逐语言比对),或者管理员显式点了
   「已人工校对」。

2. 种一条 cron ``security_awareness.translation_backfill`` —— 每小时把
   缺语言的已发布素材补上。**只补,不发布,不改中文原文**;漫画的图片
   原样沿用(同一批 JPEG,画面还是中文),所以译文 block 天然带
   ``machine_translated``,详情页会据此提示读者「正文机器翻译自中文,
   图片仍为中文」。

回填:不需要。``mt_langs`` 为 NULL 的历史行含义正确 —— 现有的每一个
翻译 block 都是人写的。
"""
from __future__ import annotations

import uuid

import sqlalchemy as sa

from alembic import op

revision = "607"
down_revision = "606"
branch_labels = None
depends_on = None

MATERIALS = "secaw_materials"
JOBS = "scheduled_jobs"
JOB_KEY = "security_awareness.translation_backfill"

_JOB_DESC = (
    "每小时补齐宣贯素材缺失的英文 / 日文。用平台共享的 AI/LLM 连接器"
    "(``ai_spending.llm.api_key``,和 copilot / 花费摘要同一把钥匙)把"
    "**已发布**素材的中文 block 译成缺的语言:标题 + 摘要走 chat_json,"
    "正文(article)走 chat_text —— 让模型把 1KB markdown 塞进 JSON 字符串"
    "是 ``BAILIAN_LLM_OUTPUT_TRUNCATED`` 的标准配方。\n"
    "**不改中文原文,不改发布状态,不译图片**:漫画的 image_keys / "
    "cover_key 原样复制(同一批 JPEG),所以英文读者拿到的是英文标题 + "
    "中文画面 —— 这比中文标题 + 中文画面好,但只有在**告诉读者**的前提下"
    "才诚实,所以生成的语言写进 ``secaw_materials.mt_langs``,每个读取面"
    "都会带 ``machine_translated``,详情页渲染成提示。\n"
    "一种语言失败不拖累其它语言,失败原因进 summary 与审计;每 tick 有"
    "条数上限(config.limit,默认 10),**剩余未处理条数写进 summary 的 "
    "``outstanding``** —— 一次悄悄少做一半的运行和一次做完的运行不该长得"
    "一样。没配 LLM key 且确实有缺口时直接 raise:这个任务此刻做不了它"
    "唯一的工作,记成 ok 就是把一句谎话写进 runs 表。"
)


def _has_table(bind, name: str) -> bool:
    return bool(bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.tables "
            "WHERE table_schema = DATABASE() AND table_name = :t LIMIT 1",
        ),
        {"t": name},
    ).scalar())


def _has_column(bind, table: str, column: str) -> bool:
    return bool(bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.columns "
            "WHERE table_schema = DATABASE() AND table_name = :t "
            "AND column_name = :c LIMIT 1",
        ),
        {"t": table, "c": column},
    ).scalar())


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

    if _has_table(bind, MATERIALS) and not _has_column(
        bind, MATERIALS, "mt_langs",
    ):
        op.add_column(
            MATERIALS,
            sa.Column("mt_langs", sa.JSON(), nullable=True),
        )

    if _has_table(bind, JOBS):
        bind.execute(
            sa.text(
                f"INSERT INTO {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, :config, 0, 0, NOW(), NOW() "
                "FROM DUAL WHERE NOT EXISTS "
                f"(SELECT 1 FROM {JOBS} WHERE handler_key = :handler)",
            ),
            {
                "id": str(uuid.uuid4()),
                "name": JOB_KEY,
                "display": "安全宣贯 · 自动补齐缺失语言",
                "desc": _JOB_DESC,
                "handler": JOB_KEY,
                # 每小时第 17 分。新素材当天就有译文,而按天跑会让一条今天
                # 发布的中文素材在英文界面上挂一整天。
                "cron": "17 * * * *",
                "config": '{"limit": 10}',
            },
        )


def downgrade() -> None:
    bind = op.get_bind()
    if _has_table(bind, JOBS):
        bind.execute(
            sa.text(f"DELETE FROM {JOBS} WHERE handler_key = :k"),
            {"k": JOB_KEY},
        )
    if _has_table(bind, MATERIALS) and _has_column(bind, MATERIALS, "mt_langs"):
        op.drop_column(MATERIALS, "mt_langs")