"""AI translation of awareness materials.

The bug this feature answers: an employee on the English UI was shown a
Chinese hook in the mascot bubble (2026-09-05). The language plumbing was
never broken — 30 of 46 published materials simply had no ``en`` block,
so the fallback chain did its job and served Chinese.

What is pinned here is everything that makes the fix safe rather than
merely present:

* the baseline is never rewritten, and the status is never touched;
* a comic's artwork is carried over, not invented;
* the generated language is MARKED, and the mark survives edits to other
  languages while being cleared by an edit to its own;
* a preview does not persist;
* "no LLM key" fails loudly instead of quietly doing nothing.
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import select

from app.domains.security_awareness.models import SecawMaterial
from app.integrations.models import PlatformSetting
from tests.test_security_awareness.conftest import (
    ADMIN_BASE,
    SELF_BASE,
    article_payload,
    awareness_key,
)

pytestmark = pytest.mark.asyncio


# ── A stand-in for the vendor ───────────────────────────────────────


class WiredFakeLLM:
    """A DashScope client double that enforces the call contract.

    Deliberately strict about the kwargs it is handed: a double that
    accepts ``**kwargs`` and returns a canned dict keeps passing after
    the real client's signature moves, which is exactly how a green
    suite ends up certifying a call that would 400 in production.
    """

    def __init__(self, api_key: str, *, model=None, region=None, **_extra):
        assert api_key, "client must not be built without a key"
        self.model = model
        self.region = region
        self.json_calls: list[dict] = []
        self.text_calls: list[dict] = []

    async def chat_json(self, *, system: str, user: str, max_tokens: int,
                        temperature: float) -> dict:
        assert system and "localization" in system.lower()
        assert max_tokens > 0
        self.json_calls.append({"system": system, "user": user})
        lang = "EN" if "English" in user else "JA"
        return {"title": f"[{lang}] title", "summary": f"[{lang}] summary"}

    async def chat_text(self, *, system: str, user: str, max_tokens: int,
                        temperature: float) -> str:
        assert system and max_tokens > 0
        self.text_calls.append({"system": system, "user": user})
        return "## Translated heading\nTranslated body."


@pytest.fixture
def fake_llm(monkeypatch):
    """Point the translator's lazy import at the double.

    Patched on the module the translator imports FROM, so the lazy
    ``from app.domains.bailian.llm_client import BailianLLMClient``
    inside ``_get_client`` resolves to it.
    """
    made: list[WiredFakeLLM] = []

    def _factory(api_key: str, **kwargs):
        client = WiredFakeLLM(api_key, **kwargs)
        made.append(client)
        return client

    monkeypatch.setattr(
        "app.domains.bailian.llm_client.BailianLLMClient", _factory,
    )
    return made


async def _set_llm_key(db, value: str = "sk-test-key") -> None:
    row = await db.get(PlatformSetting, "ai_spending.llm.api_key")
    if row is None:
        db.add(PlatformSetting(key="ai_spending.llm.api_key", value=value))
    else:
        row.value = value
    await db.commit()


def _zh_only_comic(person_id: uuid.UUID, *, publish: bool = True) -> dict:
    """A comic exactly as the security desk actually authors them: one
    Chinese block, Chinese artwork, nothing else. This IS the shape that
    produced the bug report."""
    return {
        "kind": "comic",
        "category": "phishing",
        "tags": ["钓鱼"],
        "publish": publish,
        "localized": {
            "zh": {
                "title": "3 步识破钓鱼邮件",
                "summary": "看发件人、看链接、看诉求。",
                "image_keys": [
                    awareness_key(person_id, "p1.jpg"),
                    awareness_key(person_id, "p2.jpg"),
                ],
                "cover_key": awareness_key(person_id, "p1.jpg"),
            },
        },
    }


# ── Filling the gap ─────────────────────────────────────────────────


async def test_translate_fills_missing_langs_without_touching_the_original(
    client, db_session, security_officer, fake_llm,
):
    officer, oh = security_officer
    await _set_llm_key(db_session)
    mid = (await client.post(
        f"{ADMIN_BASE}/materials", json=_zh_only_comic(officer.id), headers=oh,
    )).json()["data"]["id"]

    r = await client.post(
        f"{ADMIN_BASE}/materials/{mid}/translate", json={}, headers=oh,
    )
    assert r.status_code == 200, r.text
    data = r.json()["data"]
    assert sorted(data["filled"]) == ["en", "ja"]
    assert data["failed"] == {}
    assert data["persisted"] is True

    row = await db_session.get(SecawMaterial, uuid.UUID(mid))
    await db_session.refresh(row)
    # The Chinese block is byte-identical — the security desk authored it.
    assert row.localized["zh"]["title"] == "3 步识破钓鱼邮件"
    assert row.localized["zh"]["summary"] == "看发件人、看链接、看诉求。"
    # The new blocks carry translated text and the SAME artwork: the
    # pages are the same JPEGs and nothing here can redraw them.
    assert row.localized["en"]["title"] == "[EN] title"
    assert row.localized["en"]["image_keys"] == row.localized["zh"]["image_keys"]
    assert row.localized["en"]["cover_key"] == row.localized["zh"]["cover_key"]
    assert row.localized["ja"]["title"] == "[JA] title"
    # Marked, so every read surface can say a machine wrote it.
    assert sorted(row.mt_langs) == ["en", "ja"]
    # Content changed, so the version moved; publication did not.
    assert row.version == 2
    assert row.status == "published"


async def test_translate_is_idempotent_and_skips_what_exists(
    client, db_session, security_officer, fake_llm,
):
    """Re-running must cost nothing and change nothing.

    The bulk button and the hourly cron both re-enter this path over the
    whole library; if a second pass re-translated what the first one
    wrote, the badge would never settle and the bill would never stop.
    """
    officer, oh = security_officer
    await _set_llm_key(db_session)
    mid = (await client.post(
        f"{ADMIN_BASE}/materials", json=_zh_only_comic(officer.id), headers=oh,
    )).json()["data"]["id"]

    await client.post(f"{ADMIN_BASE}/materials/{mid}/translate", json={},
                      headers=oh)
    first_calls = len(fake_llm[0].json_calls)

    again = await client.post(
        f"{ADMIN_BASE}/materials/{mid}/translate", json={}, headers=oh,
    )
    assert again.json()["data"]["filled"] == []

    row = await db_session.get(SecawMaterial, uuid.UUID(mid))
    await db_session.refresh(row)
    assert row.version == 2, "a no-op translate must not bump the version"
    # No second client was built and no further model call went out.
    assert len(fake_llm) == 1
    assert len(fake_llm[0].json_calls) == first_calls


async def test_article_body_goes_through_the_text_call_not_json(
    client, db_session, security_officer, fake_llm,
):
    """Markdown is translated as markdown.

    Asking the model to JSON-escape a body is how the reply gets cut off
    at ``max_tokens`` mid-object and comes back as a half-written
    document — see ``BAILIAN_LLM_OUTPUT_TRUNCATED``.
    """
    _officer, oh = security_officer
    await _set_llm_key(db_session)
    mid = (await client.post(
        f"{ADMIN_BASE}/materials",
        json=article_payload(publish=True, body="## 标题\n正文内容"),
        headers=oh,
    )).json()["data"]["id"]

    await client.post(
        f"{ADMIN_BASE}/materials/{mid}/translate",
        json={"targets": ["en"]}, headers=oh,
    )
    fake = fake_llm[0]
    assert len(fake.text_calls) == 1
    assert "## 标题" in fake.text_calls[0]["user"]

    row = await db_session.get(SecawMaterial, uuid.UUID(mid))
    await db_session.refresh(row)
    assert row.localized["en"]["body"] == "## Translated heading\nTranslated body."


async def test_preview_runs_the_model_and_writes_nothing(
    client, db_session, security_officer, fake_llm,
):
    """The editor's path.

    A translate button that saved behind the author's back would also
    leave a translation behind when they cancelled the drawer.
    """
    officer, oh = security_officer
    await _set_llm_key(db_session)
    mid = (await client.post(
        f"{ADMIN_BASE}/materials", json=_zh_only_comic(officer.id), headers=oh,
    )).json()["data"]["id"]

    r = await client.post(
        f"{ADMIN_BASE}/materials/{mid}/translate",
        json={"targets": ["en"], "overwrite": True, "persist": False},
        headers=oh,
    )
    assert r.status_code == 200, r.text
    data = r.json()["data"]
    assert data["persisted"] is False
    assert data["blocks"]["en"]["title"] == "[EN] title"

    row = await db_session.get(SecawMaterial, uuid.UUID(mid))
    await db_session.refresh(row)
    assert "en" not in row.localized
    assert not (row.mt_langs or [])
    assert row.version == 1


# ── The marker, and how it is cleared ───────────────────────────────


async def test_reader_is_told_the_block_was_machine_written(
    client, db_session, security_officer, reader, fake_llm,
):
    officer, oh = security_officer
    _person, rh = reader
    await _set_llm_key(db_session)
    mid = (await client.post(
        f"{ADMIN_BASE}/materials", json=_zh_only_comic(officer.id), headers=oh,
    )).json()["data"]["id"]
    await client.post(f"{ADMIN_BASE}/materials/{mid}/translate", json={},
                      headers=oh)

    en = (await client.get(
        f"{SELF_BASE}/materials/{mid}?lang=en", headers=rh,
    )).json()["data"]
    assert en["lang"] == "en"
    # NOT a fallback: they got the language they asked for. The two
    # claims are different and the UI renders different notices.
    assert en["is_fallback_lang"] is False
    assert en["machine_translated"] is True
    assert en["machine_translated_from"] == "zh"

    zh = (await client.get(
        f"{SELF_BASE}/materials/{mid}?lang=zh", headers=rh,
    )).json()["data"]
    assert zh["machine_translated"] is False


async def test_editing_one_language_clears_only_its_own_badge(
    client, db_session, security_officer, fake_llm,
):
    """A human edit is a review — of the language they edited.

    Blanket-clearing would launder the Japanese copy the moment somebody
    fixed an English typo.
    """
    officer, oh = security_officer
    await _set_llm_key(db_session)
    mid = (await client.post(
        f"{ADMIN_BASE}/materials", json=_zh_only_comic(officer.id), headers=oh,
    )).json()["data"]["id"]
    await client.post(f"{ADMIN_BASE}/materials/{mid}/translate", json={},
                      headers=oh)

    row = await db_session.get(SecawMaterial, uuid.UUID(mid))
    await db_session.refresh(row)
    localized = {
        lang: {
            "title": block["title"],
            "summary": block.get("summary"),
            "image_keys": list(block.get("image_keys") or []),
            "cover_key": block.get("cover_key"),
        }
        for lang, block in row.localized.items()
    }
    localized["en"]["title"] = "Three checks before you click"

    r = await client.put(
        f"{ADMIN_BASE}/materials/{mid}",
        json={"localized": localized}, headers=oh,
    )
    assert r.status_code == 200, r.text

    await db_session.refresh(row)
    assert row.mt_langs == ["ja"], "only the edited language is reviewed"
    assert row.localized["en"]["title"] == "Three checks before you click"


async def test_an_unrelated_content_edit_keeps_every_badge(
    client, db_session, security_officer, fake_llm,
):
    """Recategorising is a content edit, but it reviewed nothing.

    The clearing rule compares blocks per language rather than firing on
    ``content_changed`` — otherwise fixing a category would silently
    certify two machine translations nobody has read. This is the
    regression that a blanket clear would introduce, and the reason the
    marker is a column the update path can reason about instead of a key
    the write whitelist would drop.
    """
    _officer, oh = security_officer
    await _set_llm_key(db_session)
    mid = (await client.post(
        f"{ADMIN_BASE}/materials",
        json=article_payload(publish=True), headers=oh,
    )).json()["data"]["id"]
    await client.post(f"{ADMIN_BASE}/materials/{mid}/translate", json={},
                      headers=oh)

    row = await db_session.get(SecawMaterial, uuid.UUID(mid))
    await db_session.refresh(row)
    assert sorted(row.mt_langs) == ["en", "ja"]
    version_before = row.version

    r = await client.put(
        f"{ADMIN_BASE}/materials/{mid}", json={"category": "policy"},
        headers=oh,
    )
    assert r.status_code == 200, r.text
    await db_session.refresh(row)
    assert row.category == "policy"
    assert row.version == version_before + 1, "a category change IS content"
    assert sorted(row.mt_langs) == ["en", "ja"]


async def test_mark_reviewed_clears_one_language_and_is_idempotent(
    client, db_session, security_officer, fake_llm,
):
    """The common review outcome is "it is fine", which produces no edit.

    Without this the badge would be permanent on exactly the
    translations that came out well — and a badge that can never be
    cleared stops being read.
    """
    officer, oh = security_officer
    await _set_llm_key(db_session)
    mid = (await client.post(
        f"{ADMIN_BASE}/materials", json=_zh_only_comic(officer.id), headers=oh,
    )).json()["data"]["id"]
    await client.post(f"{ADMIN_BASE}/materials/{mid}/translate", json={},
                      headers=oh)

    r = await client.post(
        f"{ADMIN_BASE}/materials/{mid}/translation-reviewed",
        json={"lang": "en"}, headers=oh,
    )
    assert r.status_code == 200, r.text
    assert r.json()["data"]["mt_langs"] == ["ja"]

    row = await db_session.get(SecawMaterial, uuid.UUID(mid))
    await db_session.refresh(row)
    version_before = row.version

    again = await client.post(
        f"{ADMIN_BASE}/materials/{mid}/translation-reviewed",
        json={"lang": "en"}, headers=oh,
    )
    assert again.status_code == 200
    assert again.json()["data"]["mt_langs"] == ["ja"]
    await db_session.refresh(row)
    assert row.version == version_before, "a review is not a content change"


# ── Gaps, bulk fill, and degradation ────────────────────────────────


async def test_gap_report_counts_only_what_a_reader_would_miss(
    client, db_session, security_officer, fake_llm,
):
    officer, oh = security_officer
    await _set_llm_key(db_session)
    await client.post(f"{ADMIN_BASE}/materials",
                      json=_zh_only_comic(officer.id), headers=oh)
    # A draft is not shown to anybody, so it is not part of the gap the
    # employee-facing report is about.
    await client.post(f"{ADMIN_BASE}/materials",
                      json=article_payload(publish=False), headers=oh)

    gaps = (await client.get(
        f"{ADMIN_BASE}/translation-gaps", headers=oh,
    )).json()["data"]
    assert gaps["materials_missing"] == 1
    assert gaps["per_lang"] == {"en": 1, "ja": 1}
    assert gaps["llm_configured"] is True


async def test_bulk_fill_reports_what_it_left_behind(
    client, db_session, security_officer, fake_llm,
):
    """A run that quietly leaves half the queue behind must not look
    like one that finished."""
    officer, oh = security_officer
    await _set_llm_key(db_session)
    for _ in range(3):
        await client.post(f"{ADMIN_BASE}/materials",
                          json=_zh_only_comic(officer.id), headers=oh)

    r = await client.post(
        f"{ADMIN_BASE}/materials/translate-missing",
        json={"limit": 2}, headers=oh,
    )
    assert r.status_code == 200, r.text
    data = r.json()["data"]
    assert data["considered"] == 3
    assert data["attempted"] == 2
    assert data["materials_filled"] == 2
    assert data["langs_filled"] == 4
    assert data["outstanding"] == 1


async def test_no_llm_key_is_a_503_that_says_what_to_do(
    client, db_session, security_officer,
):
    """Degradation contract: every field the model fills is one a human
    can type, so the feature is optional — but it must SAY so."""
    officer, oh = security_officer
    row = await db_session.get(PlatformSetting, "ai_spending.llm.api_key")
    if row is not None:
        await db_session.delete(row)
        await db_session.commit()

    mid = (await client.post(
        f"{ADMIN_BASE}/materials", json=_zh_only_comic(officer.id), headers=oh,
    )).json()["data"]["id"]

    r = await client.post(
        f"{ADMIN_BASE}/materials/{mid}/translate", json={}, headers=oh,
    )
    assert r.status_code == 503
    assert r.json()["errors"][0]["code"] == "BAILIAN_LLM_NO_KEY"


async def test_backfill_cron_raises_when_there_is_work_and_no_key(
    db_session, security_officer,
):
    """A handler that returned a tidy "skipped" here would land
    ``runs.status='ok'`` and the job would look healthy for as long as
    the connector stayed empty — the failure mode
    ``feedback_handler_status_must_raise`` exists to prevent."""
    from app.domains.security_awareness.scheduler_handlers import (
        run_translation_backfill,
    )

    officer, _oh = security_officer
    setting = await db_session.get(PlatformSetting, "ai_spending.llm.api_key")
    if setting is not None:
        await db_session.delete(setting)
    db_session.add(SecawMaterial(
        id=uuid.uuid4(),
        kind="comic",
        category="phishing",
        tags=[],
        localized={
            "zh": {
                "title": "只有中文",
                "image_keys": [awareness_key(officer.id, "p1.jpg")],
            },
        },
        status="published",
        version=1,
    ))
    await db_session.commit()

    with pytest.raises(RuntimeError, match="no AI/LLM key"):
        await run_translation_backfill(db_session, {})


async def test_backfill_cron_is_quiet_when_the_library_is_complete(
    db_session,
):
    from app.domains.security_awareness.scheduler_handlers import (
        run_translation_backfill,
    )

    existing = (await db_session.execute(select(SecawMaterial))).scalars().all()
    assert not existing, "this test relies on an empty library"

    summary = await run_translation_backfill(db_session, {})
    assert summary["materials_missing"] == 0
    assert summary["outstanding"] == 0


async def test_employee_cannot_translate(client, db_session, reader,
                                         security_officer, fake_llm):
    officer, oh = security_officer
    _person, rh = reader
    await _set_llm_key(db_session)
    mid = (await client.post(
        f"{ADMIN_BASE}/materials", json=_zh_only_comic(officer.id), headers=oh,
    )).json()["data"]["id"]

    for path, body in (
        (f"{ADMIN_BASE}/materials/{mid}/translate", {}),
        (f"{ADMIN_BASE}/materials/translate-missing", {"limit": 1}),
        (f"{ADMIN_BASE}/materials/{mid}/translation-reviewed", {"lang": "en"}),
    ):
        r = await client.post(path, json=body, headers=rh)
        assert r.status_code == 403, f"{path} -> {r.status_code}"
    assert (await client.get(
        f"{ADMIN_BASE}/translation-gaps", headers=rh,
    )).status_code == 403