"""Put ENGLISH comic artwork onto the awareness materials that had only
Chinese pages, and create the four whose topic had no Chinese comic at all.

## The bug this closes

Migration 607's translator did the honest half of the job: it machine-
translated each comic's title + summary into en / ja and **copied the
Chinese ``image_keys`` verbatim**, because a model cannot redraw a comic.
Its own docstring says so — "an English reader therefore gets an English
title and hook over Chinese artwork". So today every published comic
renders English chrome over pages the reader cannot read.

``secaw_materials.localized`` was always per-language down to the artwork
(``image_keys`` / ``cover_key`` live inside each language's block, and
``service.pick_block`` resolves ONE block whose own keys get presigned).
So the fix is data, not code: point the ``en`` block at English pages.

## What it does

* **21 merges** — upload the English JPEG, then set ``image_keys`` AND
  ``cover_key`` on the ``en`` and ``ja`` blocks of the matching material.
  ``cover_key`` matters as much as the pages: it is the library card's
  thumbnail, and an English page behind a Chinese thumbnail is still a
  card the reader cannot read. ``zh`` is never touched.
* **4 creations** — topics with no Chinese comic (AI tools, public Wi-Fi,
  removable media, working from home). All three blocks point at the
  English artwork, per the user's decision 2026-09-07 ("只有英语的就都显示
  英语的"); zh / ja get authored titles + summaries so the card reads in
  the reader's language even though the pages are English.

**Japanese points at the English artwork too** (user's decision, same
day). It matches the platform's own reasoning in
``core.locale.FALLBACK_CHAIN``: for a JP-based reader English is the
closer register, so English pages beat Chinese pages.

## What it deliberately does NOT do

* **It does not touch ``mt_langs``.** Every published comic currently has
  it empty, so no reader is shown a machine-translation notice — a
  pre-existing state this script did not create and does not change.
  Setting it now would be actively wrong: the comic variant of that
  notice tells the reader *the artwork is still Chinese*, which is the
  one thing this script makes untrue. Re-introducing the badge needs the
  notice copy split first (title-was-translated vs artwork-is-foreign),
  which is a code change, not a data one.
* **It does not touch ``zh``.** The Chinese pages are the authored
  original.

## Idempotency

The uploaded key ends with the source filename, so "has this material
already got this artwork" is answerable from the stored key: a material
whose ``en.image_keys`` already ends with the expected filename is
skipped, and nothing is re-uploaded. Re-running is a no-op.

``--dry-run`` resolves and reports without uploading or writing.

Usage::

    AWS_PROFILE=ljq-iam AWS_REGION=ap-southeast-1 \\
    DATABASE_URL_SYNC='mysql+pymysql://...' \\
        .venv/bin/python scripts/load_awareness_comic_artwork.py \\
            ~/Desktop/Security_Awareness_EN --dry-run
"""
from __future__ import annotations

import argparse
import json
import mimetypes
import sys
import uuid
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

import boto3
from sqlalchemy import create_engine, text

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from app.core.storage_uploads import (  # noqa: E402
    PURPOSE_MAX_BYTES,
    build_relative_upload_key,
)
from app.domains.security_awareness.presign import (  # noqa: E402
    AWARENESS_PURPOSE,
    is_awareness_key,
)

TABLE = "secaw_materials"

#: Attribution for the uploaded objects. The key layout embeds the
#: uploader's person id (``storage_router.parse_object_key`` recovers it
#: for the read ACL), so this is not decoration — it has to be a real
#: person, and it is the same one who uploaded the Chinese originals.
UPLOADER_EMAIL = "shuai.feng@pingcap.com"

#: ``(source filename, EXACT zh title of the material it belongs to)``.
#:
#: Reviewed by hand, and matched EXACTLY — only 7 of the 25 titles agree
#: character for character with production (the rest differ by spacing,
#: punctuation or wording: 三步 vs 3 步, 完美的密码 vs 好密码, 朋友圈七不 vs
#: 社交网络：七不发). Fuzzy matching would have been the wrong tool here:
#: production has four separate social-network comics, so "close enough"
#: could silently hang English pages on the wrong piece. A title that
#: does not resolve is an error, never a skip.
MERGE: list[tuple[str, str]] = [
    ("01_Spot_a_Phishing_Email_in_3_Steps.jpg", "3 步识破钓鱼邮件"),
    ("02_Setting_a_Perfect_Password.jpg", "设置一个好密码就是这么简单"),
    ("03_Ransomware_Held_Hostage.jpg", "你被勒索软件勒索了"),
    # Production also has an ARCHIVED 「什么是 CEO 诈骗」; the English piece
    # covers social engineering and CEO fraud together, so it goes on the
    # published social-engineering comic and the archived one stays retired.
    ("04_Social_Engineering_and_CEO_Fraud.jpg", "社会工程攻击:披着羊皮的狼"),
    ("05_10_Ways_to_Protect_Your_Mobile_Device.jpg", "10 招保护你的移动设备"),
    ("06_You_Are_a_Target_Too.jpg", "你也是黑客攻击的目标"),
    ("07_Clean_Desk_and_Office_Secrets.jpg", "如何杜绝办公机密泄漏"),
    ("08_Credential_Stuffing_Attacks.jpg", "一图读懂什么是撞库攻击"),
    ("09_Seven_Security_Blind_Spots.jpg", "7 大安全意识隐患,你占了几条"),
    ("10_Protecting_Customer_Data.jpg", "客户信息保护"),
    ("11_Six_Common_Attacks_and_Defenses.jpg", "你必须知道的 6 大网络攻击与对策"),
    ("12_Safe_App_Usage.jpg", "安全使用 App"),
    ("13_Business_Travel_Security.jpg", "差旅安全"),
    ("14_Malware_vs_Vulnerabilities.jpg", "恶意软件 VS 漏洞"),
    ("15_File_Sharing_Security.jpg", "文件共享安全"),
    # 七不发 ↔ 七不. Chosen over 社交网络上的安全 / 网络社交隐患多(上)(下),
    # which are three different social-network pieces.
    ("16_Social_Media_What_Not_to_Post.jpg", "朋友圈七不"),
    ("18_Five_Myths_of_Security_Awareness.jpg", "网络安全意识的五大误区"),
    ("19_Online_Shopping_Safety.jpg", "网络购物也是危机四伏"),
    ("20_Bluetooth_and_Wireless_Attacks.jpg", "蓝牙漏洞攻击"),
    ("21_Vishing_and_Smishing.jpg", "伪基站与电信诈骗"),
    ("22_Ive_Been_Hacked_What_Now.jpg", "遭到黑客攻击,我该怎么办"),
]

#: Topics with no Chinese comic at all → new materials. ``category`` is one
#: of ``catalog.MATERIAL_CATEGORIES`` (a closed list); ``display_order``
#: continues past production's current maximum so these land at the end of
#: the library rather than jumping the authored ordering.
NEW: list[dict[str, Any]] = [
    {
        "file": "17_AI_Tools_and_Company_Data.jpg",
        "category": "data",
        "tags": ["AI", "大模型", "数据外流", "深度伪造"],
        "display_order": 130,
        "localized": {
            "zh": {
                "title": "使用 AI 工具时如何避免泄密",
                "summary": "贴进公共 AI 工具的内容去了哪里、什么绝对不能贴、怎么用才安全,以及反过来——声音和视频也能被伪造。",
            },
            "en": {
                "title": "Using AI Tools Without Leaking Company Data",
                "summary": "What happens to text pasted into public AI tools, what must never be pasted, how to use AI safely, and the deepfake risk on the other side.",
            },
            "ja": {
                "title": "AIツールで情報を漏らさないために",
                "summary": "公開AIツールに貼り付けた内容はどこへ行くのか、絶対に貼ってはいけないもの、安全な使い方、そして逆方向のリスク——音声も映像も偽装できます。",
            },
        },
    },
    {
        "file": "23_Public_WiFi_Security.jpg",
        "category": "endpoint",
        "tags": ["公共WiFi", "邪恶双子", "VPN", "家用路由器"],
        "display_order": 131,
        "localized": {
            "zh": {
                "title": "免费 WiFi 从来都不免费",
                "summary": "伪造热点、假登录页与流量拦截是怎么工作的,六条安全使用规则,外加一份家用 WiFi 加固清单。",
            },
            "en": {
                "title": "Free Wi-Fi Is Never Really Free",
                "summary": "Evil-twin hotspots, fake captive portals and traffic interception — six safe-use rules, plus a short home Wi-Fi hardening checklist.",
            },
            "ja": {
                "title": "無料Wi-Fiは決して無料ではない",
                "summary": "偽アクセスポイント、偽の認証ページ、通信の傍受はどう行われるのか。安全に使う6つのルールと、自宅Wi-Fiの強化チェックリスト。",
            },
        },
    },
    {
        "file": "24_USB_and_Removable_Media.jpg",
        "category": "endpoint",
        "tags": ["U盘", "移动介质", "充电劫持", "加密"],
        "display_order": 132,
        "localized": {
            "zh": {
                "title": "移动介质安全:U 盘不是礼物",
                "summary": "移动介质的风险是双向的——捡到或收到的设备带恶意程序进来,没加密的 U 盘带公司数据出去。",
            },
            "en": {
                "title": "That Free USB Stick Is Not a Gift",
                "summary": "Removable media cuts both ways: malware arriving on a found or gifted device, and company data walking out on an unencrypted stick.",
            },
            "ja": {
                "title": "そのUSBメモリは贈り物ではない",
                "summary": "外部媒体のリスクは双方向です。拾った・もらった機器からマルウェアが入り、暗号化されていないUSBから会社のデータが出ていきます。",
            },
        },
    },
    {
        "file": "25_Working_From_Home_Security.jpg",
        "category": "endpoint",
        "tags": ["居家办公", "家用路由器", "VPN", "锁屏"],
        "display_order": 133,
        "localized": {
            "zh": {
                "title": "居家办公安全",
                "summary": "把家庭网络加固好、让工作留在工作设备上,以及那些跟着你走的日常习惯:锁屏、通话隐私、视频会议卫生。",
            },
            "en": {
                "title": "Staying Secure Working From Home",
                "summary": "Hardening the home network, keeping work on work devices, and the everyday habits that travel with you — screen lock, call privacy, video-call hygiene.",
            },
            "ja": {
                "title": "在宅勤務のセキュリティ",
                "summary": "自宅ネットワークの強化、業務は業務用端末で行うこと、そしてどこでも通用する日々の習慣——画面ロック、通話のプライバシー、ビデオ会議の作法。",
            },
        },
    },
]

#: Languages whose artwork this script owns. ``zh`` is never in here for a
#: merge (the Chinese pages are the authored original); for a creation the
#: caller passes all three.
MERGE_LANGS = ("en", "ja")


def _json(value: Any) -> Any:
    return json.loads(value) if isinstance(value, (str, bytes)) else value


def _storage_config(conn) -> tuple[str, str, str]:
    rows = dict(conn.execute(text(
        "SELECT `key`, value FROM platform_settings WHERE `key` IN "
        "('storage.s3.bucket','storage.s3.region','storage.s3.key_prefix')",
    )).fetchall())

    def one(key: str, default: str = "") -> str:
        raw = rows.get(key)
        value = _json(raw) if isinstance(raw, (str, bytes)) else raw
        return str(value or default)

    bucket, region = one("storage.s3.bucket"), one("storage.s3.region")
    if not bucket or not region:
        raise SystemExit("storage.s3.bucket / region not configured")
    return bucket, region, one("storage.s3.key_prefix")


def _upload(s3, bucket: str, prefix: str, key: str, path: Path, dry: bool) -> None:
    if dry:
        return
    s3.upload_file(
        str(path), bucket, f"{prefix}{key}",
        ExtraArgs={"ContentType": mimetypes.guess_type(path.name)[0] or "image/jpeg"},
    )


def _new_key(person_id: uuid.UUID, filename: str) -> str:
    key = build_relative_upload_key(
        purpose=AWARENESS_PURPOSE, person_id=person_id, filename=filename,
    )
    # The read side refuses to sign a key it would not accept today, so a
    # key this script invents has to pass that same gate — otherwise the
    # pages upload fine and then render as broken images.
    if not is_awareness_key(key):
        raise SystemExit(f"built a key the read ACL would refuse: {key}")
    return key


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("source_dir", type=Path)
    ap.add_argument("--dry-run", action="store_true")
    args = ap.parse_args()
    dry = args.dry_run

    import os

    url = os.environ.get("DATABASE_URL_SYNC")
    if not url or not url.startswith("mysql+pymysql://"):
        raise SystemExit("DATABASE_URL_SYNC (mysql+pymysql://...) is required")

    files = {p.name: p for p in args.source_dir.glob("*.jpg")}
    cap = PURPOSE_MAX_BYTES[AWARENESS_PURPOSE]
    for name in [f for f, _ in MERGE] + [n["file"] for n in NEW]:
        if name not in files:
            raise SystemExit(f"missing source file: {name}")
        if files[name].stat().st_size > cap:
            raise SystemExit(f"{name} exceeds the {cap} byte cap for {AWARENESS_PURPOSE}")

    engine = create_engine(url, connect_args={"connect_timeout": 10, "read_timeout": 120})
    now = datetime.now(UTC).replace(tzinfo=None)
    merged = created = skipped = 0

    with engine.begin() as conn:
        print(f"target: {conn.execute(text('SELECT VERSION()')).scalar()} | "
              f"mode: {'DRY RUN' if dry else 'WRITE'}")
        bucket, region, prefix = _storage_config(conn)
        print(f"bucket: {bucket} | region: {region} | prefix: {prefix!r}")
        s3 = boto3.client("s3", region_name=region)

        person_id = conn.execute(text(
            "SELECT id FROM identity_persons WHERE work_email = :e"), {"e": UPLOADER_EMAIL},
        ).scalar()
        if not person_id:
            raise SystemExit(f"uploader {UPLOADER_EMAIL} not found")
        person_id = uuid.UUID(str(person_id))

        # ── merges ──────────────────────────────────────────────────
        rows = conn.execute(text(
            f"SELECT id, localized, version FROM {TABLE} WHERE kind = 'comic'",  # noqa: S608
        )).fetchall()
        by_title: dict[str, tuple] = {}
        for r in rows:
            blocks = _json(r[1]) or {}
            zh_title = ((blocks.get("zh") or {}).get("title") or "").strip()
            if zh_title:
                by_title[zh_title] = (r[0], blocks, r[2])

        for filename, zh_title in MERGE:
            hit = by_title.get(zh_title)
            if hit is None:
                raise SystemExit(
                    f"no comic titled {zh_title!r} — the mapping is stale, "
                    "resolve it by hand rather than guessing",
                )
            mat_id, blocks, version = hit
            already = (blocks.get("en") or {}).get("image_keys") or []
            if already and str(already[0]).endswith(filename):
                skipped += 1
                continue
            key = _new_key(person_id, filename)
            _upload(s3, bucket, prefix, key, files[filename], dry)
            updated = dict(blocks)
            for lang in MERGE_LANGS:
                block = dict(updated.get(lang) or {})
                if not block.get("title"):
                    # No block for this language at all: skip rather than
                    # invent one — a block with artwork and no title does
                    # not render (``pick_block`` requires a title).
                    continue
                block["image_keys"] = [key]
                block["cover_key"] = key
                updated[lang] = block
            if not dry:
                conn.execute(
                    text(
                        f"UPDATE {TABLE} SET localized = :loc, version = :v, "  # noqa: S608
                        "updated_at = :now, updated_by_person_id = :pid WHERE id = :id",
                    ),
                    {
                        "loc": json.dumps(updated, ensure_ascii=False),
                        # Artwork changing IS a content change, and a campaign
                        # records the version it went out with.
                        "v": int(version or 1) + 1,
                        "now": now, "pid": str(person_id), "id": mat_id,
                    },
                )
            merged += 1
            print(f"  merged  {filename:<46} → {zh_title}")

        # ── creations ───────────────────────────────────────────────
        existing_titles = set(by_title)
        for spec in NEW:
            zh_title = spec["localized"]["zh"]["title"]
            if zh_title in existing_titles:
                skipped += 1
                continue
            filename = spec["file"]
            key = _new_key(person_id, filename)
            _upload(s3, bucket, prefix, key, files[filename], dry)
            blocks = {
                lang: {**meta, "image_keys": [key], "cover_key": key}
                for lang, meta in spec["localized"].items()
            }
            if not dry:
                conn.execute(
                    text(
                        f"INSERT INTO {TABLE} "  # noqa: S608
                        "(id, kind, category, tags, localized, status, is_pinned, "
                        " display_order, version, published_at, created_at, updated_at, "
                        " created_by_person_id, updated_by_person_id) "
                        "VALUES (:id, 'comic', :cat, :tags, :loc, 'published', 0, "
                        " :order_, 1, :now, :now, :now, :pid, :pid)",
                    ),
                    {
                        "id": str(uuid.uuid4()), "cat": spec["category"],
                        "tags": json.dumps(spec["tags"], ensure_ascii=False),
                        "loc": json.dumps(blocks, ensure_ascii=False),
                        "order_": spec["display_order"], "now": now,
                        "pid": str(person_id),
                    },
                )
            created += 1
            print(f"  created {filename:<46} → {zh_title}")

    print(f"\nmerged={merged} created={created} skipped(already done)={skipped}")


if __name__ == "__main__":
    main()