"""AI translation for awareness materials — closing the language gap that made an English UI speak Chinese. ## The bug this exists for Every read path in this domain already resolves language properly: the frontend passes its active i18next language, ``service.pick_block`` resolves it against ``LANG_FALLBACK`` and reports ``is_fallback_lang`` when the reader did not get what they asked for. None of that was broken. What was missing is **content**. On 2026-09-05 the published library was 46 materials: 30 comics carrying only a ``zh`` block and 16 articles with no ``ja``. So the fallback chain did exactly what it is designed to do — and an employee reading the English UI got a Chinese hook in the mascot bubble, on every page, with no way to tell whether it was a bug or a Chinese-only notice. That is the report this module answers. ## What it does Translates the baseline (``zh``) block of a material into the languages it is missing, using the org's shared Bailian/DashScope key — the same connector the copilot and the spend summaries use. Two calls per target language, deliberately split: * **title + summary** through ``chat_json``: short, structured, and the two strings the bubble and the library card actually render. * **body** (articles only) through ``chat_text``: markdown out, markdown in. Asking a model to JSON-escape a 1 KB markdown document is how you get ``BAILIAN_LLM_OUTPUT_TRUNCATED`` mid-object and a half-written translation. ## What it does NOT do * **It never touches the baseline block.** ``zh`` is authored by the security desk; nothing here rewrites it. * **It never translates artwork.** A comic's ``image_keys`` / ``cover_key`` are copied verbatim into the new block, because the pages are the same JPEGs. An English reader therefore gets an English title and hook over Chinese artwork — better than a Chinese title over Chinese artwork, and honest only because the reader is TOLD: the generated languages are recorded in ``SecawMaterial.mt_langs`` and every read surface reports ``machine_translated`` for the block it rendered, which the detail page turns into a notice naming the source language (and, for a comic, the untranslated artwork). * **It never publishes.** Status is not touched. A draft stays a draft. ## Why the marker is a column and not a key inside the block ``MaterialWrite.normalized_localized()`` whitelists the keys it persists, so an ``mt: true`` smuggled into the JSON block would be silently dropped by any admin save — including saves that never touched that language (a ``kind``-only edit round-trips every block through the same validator). The badge would then vanish for reasons unrelated to anybody reviewing anything. ``mt_langs`` is a column, cleared per-language by ``SecurityAwarenessService.update`` only for the languages whose block actually changed, and by the explicit "I reviewed this" endpoint. """ from __future__ import annotations import logging import uuid from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException, NotFoundError from app.domains.security_awareness.catalog import ( BASELINE_LANG, KIND_ARTICLE, KIND_COMIC, KIND_LINK, MATERIAL_LANGS, MAX_SUMMARY_LEN, MAX_TITLE_LEN, STATUS_ARCHIVED, STATUS_PUBLISHED, ) from app.domains.security_awareness.models import SecawMaterial from app.integrations.service import get_platform_setting logger = logging.getLogger(__name__) #: The shared AI / LLM connector. Same rows the copilot, the spend #: summaries and AI Labs read — one key to rotate, one place the spend #: shows up. This domain does not get its own credential. SETTING_LLM_KEY = "ai_spending.llm.api_key" SETTING_LLM_MODEL = "ai_spending.llm.model" SETTING_LLM_REGION = "ai_spending.llm.region" #: Longest article body we will hand the model in one shot. The ceiling #: is the model's OUTPUT budget, not its input window: Chinese → English #: roughly 1.3 tokens out per source character, and qwen-max tops out #: around 8k output tokens. Above this the reply would be truncated #: mid-sentence and ``chat_json``/``chat_text`` would hand back a #: half-written document that looks like a translation. The published #: library's longest body is 1,243 characters, so this is ~5x headroom; #: anything past it is a manual translation, and the error says so. MAX_BODY_CHARS = 6_000 #: Output budget for the body call. Sized against MAX_BODY_CHARS above. _BODY_MAX_TOKENS = 8_000 #: Output budget for the title+summary call. 200 + 500 characters of #: target-language text plus JSON scaffolding. _META_MAX_TOKENS = 1_200 #: Statuses worth spending tokens on by default. An archived material is #: retired — nobody will ever be shown it again — and translating it is #: paying for content that has no reader. BACKFILL_STATUSES: tuple[str, ...] = (STATUS_PUBLISHED,) _LANG_NAMES: dict[str, str] = { "zh": "Simplified Chinese (简体中文)", "en": "English", "ja": "Japanese (日本語)", } _SYSTEM = ( "You are a professional localization specialist for a technology " "company's internal security-awareness programme. You translate " "employee-facing security education content.\n" "Rules:\n" "1. Translate MEANING, not words. The output must read as if it was " "written natively for the target audience — office workers, not " "security engineers.\n" "2. Keep the register: direct, plain, slightly informal. Chinese " "awareness copy often uses rhetorical questions and short punchy " "clauses; keep that energy rather than flattening it into corporate " "prose.\n" "3. Do NOT translate product, company or platform names (Helix, " "Feishu / 飞书, PingCAP, TiDB, Google, GitHub, VPN, MFA, SSO, API), " "and keep technical terms in the form the target audience uses.\n" "4. Preserve markdown structure exactly: heading levels, list " "markers, bold/italic, code fences, links, and line breaks. Never " "add or remove sections, and never add a preface, a note about the " "translation, or a sign-off.\n" "5. Translate every sentence. Do not summarise, do not expand, do " "not editorialise, and never invent facts, URLs or policy that the " "source does not state.\n" "6. If a sentence refers to something the reader can only see in an " "image, translate it as written — the artwork is not yours to fix." ) @dataclass(slots=True) class MaterialOutcome: """What happened to one material.""" material_id: uuid.UUID #: Languages that were generated and persisted by this run. filled: list[str] = field(default_factory=list) #: Languages left alone — already present and ``overwrite`` was off. skipped: list[str] = field(default_factory=list) #: ``lang -> error message`` for the languages that failed. A failure #: on one language never blocks the others: two thirds of a #: translation is two thirds of the problem solved. failed: dict[str, str] = field(default_factory=dict) #: The generated blocks, ``lang -> block``. Returned so the editor #: can fill its form from what the model produced — including on the #: ``persist=False`` path, where nothing was written and the blocks #: are the only output. blocks: dict[str, dict[str, Any]] = field(default_factory=dict) #: False when this was a preview: the model ran, nothing was stored. persisted: bool = True def as_dict(self) -> dict[str, Any]: return { "material_id": str(self.material_id), "filled": list(self.filled), "skipped": list(self.skipped), "failed": dict(self.failed), "blocks": {k: dict(v) for k, v in self.blocks.items()}, "persisted": self.persisted, } def missing_langs(row: SecawMaterial) -> list[str]: """Languages this material cannot render, in catalog order. A block with no ``title`` does not count as present — ``pick_block`` skips it, so from a reader's point of view it does not exist. """ blocks = row.localized or {} out: list[str] = [] for lang in MATERIAL_LANGS: if lang == BASELINE_LANG: continue block = blocks.get(lang) if not (isinstance(block, dict) and block.get("title")): out.append(lang) return out class MaterialTranslator: """Fills a material's missing language blocks from its baseline.""" def __init__(self, db: AsyncSession): self.db = db self.audit = AuditService(db) self._client: Any | None = None # ── Public API ────────────────────────────────────────────────── async def translate_material( self, material_id: uuid.UUID, *, targets: list[str] | None = None, overwrite: bool = False, persist: bool = True, actor_id: uuid.UUID | None, trace_id: str | uuid.UUID, ) -> MaterialOutcome: """Translate one material into ``targets`` (default: everything it is missing). Idempotent by construction: with ``overwrite=False`` a language that already has a title is reported under ``skipped`` and costs nothing, so re-running the same call is free and changes nothing. ``persist=False`` runs the model and returns the blocks without writing. That is the material editor's path: the author is looking at the tab, and a "translate" button that saved behind their back would take the review away from the person best placed to do it — and would leave a translation behind even if they then cancelled the drawer. """ row = await self.db.get(SecawMaterial, material_id) if row is None: raise NotFoundError("security awareness material", str(material_id)) return await self._translate_row( row, targets=targets, overwrite=overwrite, persist=persist, actor_id=actor_id, trace_id=trace_id, ) async def fill_missing( self, *, limit: int, statuses: tuple[str, ...] = BACKFILL_STATUSES, actor_id: uuid.UUID | None = None, trace_id: str | uuid.UUID, ) -> dict[str, Any]: """Translate up to ``limit`` materials that are missing a language. Returns a summary the caller (endpoint or cron) reports verbatim. ``outstanding`` is the number of materials still missing something AFTER this run — a run that quietly leaves half the queue behind and a run that finished the job must not look the same. """ rows = await self._gap_rows(statuses) outcomes: list[MaterialOutcome] = [] for row in rows[:max(0, limit)]: outcome = await self._translate_row( row, targets=None, overwrite=False, actor_id=actor_id, trace_id=trace_id, ) outcomes.append(outcome) # Commit per material rather than once at the end. Each one # is independent, and the run is minutes long: nginx cuts a # request at 300s, and a single trailing commit would throw # away every translation already paid for the moment the # caller's connection went away. The cron commits again at # the end, which is then a no-op. await self.db.commit() remaining = await self._gap_rows(statuses) return { "considered": len(rows), "attempted": len(outcomes), "materials_filled": sum(1 for o in outcomes if o.filled), "langs_filled": sum(len(o.filled) for o in outcomes), "materials_failed": sum(1 for o in outcomes if o.failed), "outstanding": len(remaining), "details": [o.as_dict() for o in outcomes if o.filled or o.failed], } async def gap_summary( self, statuses: tuple[str, ...] = BACKFILL_STATUSES, ) -> dict[str, Any]: """How much of the library an ``en`` / ``ja`` reader cannot read. Cheap (no model calls) so the console can show the number next to the button that fixes it. """ rows = await self._gap_rows(statuses) per_lang: dict[str, int] = {} for row in rows: for lang in missing_langs(row): per_lang[lang] = per_lang.get(lang, 0) + 1 return { "materials_missing": len(rows), "per_lang": per_lang, "llm_configured": await self._has_key(), } # ── Internals ─────────────────────────────────────────────────── async def _gap_rows( self, statuses: tuple[str, ...], ) -> list[SecawMaterial]: stmt = select(SecawMaterial).order_by( SecawMaterial.is_pinned.desc(), SecawMaterial.display_order.asc(), ) if statuses: stmt = stmt.where(SecawMaterial.status.in_(statuses)) else: # pragma: no cover - defensive; callers always pass one stmt = stmt.where(SecawMaterial.status != STATUS_ARCHIVED) rows = (await self.db.execute(stmt)).scalars().all() return [r for r in rows if missing_langs(r) and self._baseline(r)] @staticmethod def _baseline(row: SecawMaterial) -> dict[str, Any] | None: """The block we translate FROM. Normally ``zh`` (the write schema requires it). A legacy row that somehow has no baseline is skipped rather than translated from an arbitrary language — translating a translation compounds error, and the honest fix for such a row is an author, not a model. """ block = (row.localized or {}).get(BASELINE_LANG) if isinstance(block, dict) and block.get("title"): return block return None async def _setting(self, key: str) -> str | None: value = await get_platform_setting(key, None, session=self.db) return value if isinstance(value, str) and value.strip() else None async def _has_key(self) -> bool: return bool(await self._setting(SETTING_LLM_KEY)) async def _get_client(self) -> Any: """Build (once per service instance) the shared Bailian client. Cached on the instance so a 20-material backfill does not re-read three platform settings and re-construct an httpx client forty times. """ if self._client is not None: return self._client from app.domains.bailian.llm_client import BailianLLMClient api_key = await self._setting(SETTING_LLM_KEY) if not api_key: raise AppException( code="BAILIAN_LLM_NO_KEY", message=( "未配置百炼(DashScope) API Key,无法自动翻译宣贯素材。" "请在 管理 → 连接器 → AI / LLM 配置后重试;在此之前可在" "素材编辑器里手动填写英文 / 日文内容。" ), status_code=503, ) self._client = BailianLLMClient( api_key=api_key, model=await self._setting(SETTING_LLM_MODEL), region=await self._setting(SETTING_LLM_REGION), purpose="security_awareness.translation") return self._client async def _translate_row( self, row: SecawMaterial, *, targets: list[str] | None, overwrite: bool, persist: bool = True, actor_id: uuid.UUID | None, trace_id: str | uuid.UUID, ) -> MaterialOutcome: outcome = MaterialOutcome(material_id=row.id, persisted=persist) source = self._baseline(row) if source is None: raise AppException( code="SECAW_NO_BASELINE_BLOCK", message=( f"material has no {BASELINE_LANG!r} content to " "translate from" ), status_code=409, ) wanted = [ lang for lang in (targets or missing_langs(row)) if lang in MATERIAL_LANGS and lang != BASELINE_LANG ] if not wanted: return outcome blocks = dict(row.localized or {}) todo: list[str] = [] for lang in wanted: existing = blocks.get(lang) has_content = isinstance(existing, dict) and existing.get("title") if has_content and not overwrite: outcome.skipped.append(lang) continue todo.append(lang) if not todo: return outcome # Resolve the client BEFORE the loop so a missing key fails once, # loudly, instead of once per language with the same message. client = await self._get_client() for lang in todo: try: generated = await self._translate_block( client, row.kind, source, lang, ) blocks[lang] = generated outcome.blocks[lang] = generated except AppException as exc: logger.warning( "secaw.translate.failed material=%s lang=%s code=%s: %s", row.id, lang, exc.code, exc.message, ) outcome.failed[lang] = f"{exc.code}: {exc.message}" except Exception as exc: # noqa: BLE001 - one lang must not # take the whole backfill down; the reason is recorded. logger.exception( "secaw.translate.error material=%s lang=%s", row.id, lang, ) outcome.failed[lang] = str(exc) or exc.__class__.__name__ outcome.filled = [lang for lang in todo if lang not in outcome.failed] if not outcome.filled or not persist: # Nothing landed, or this was a preview. Either way the row # stays untouched — no version bump and no audit entry # claiming a change that did not happen. return outcome row.localized = blocks marked = set(row.mt_langs or []) | set(outcome.filled) row.mt_langs = sorted(marked) # A translation IS a content change: an English reader is now # being shown different text. The version is what a campaign # snapshots, and "the card said something else when I confirmed" # has to stay answerable. row.version += 1 row.updated_at = datetime.now(UTC) if actor_id is not None: row.updated_by_person_id = actor_id await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user" if actor_id else "system", actor_user_id=actor_id, action="security_awareness.material.translate", status="success" if not outcome.failed else "partial", resource_type="secaw_material", resource_id=str(row.id), detail={ "filled": outcome.filled, "skipped": outcome.skipped, "failed": outcome.failed, "overwrite": overwrite, "source_lang": BASELINE_LANG, "version": row.version, "machine_translated_langs": list(row.mt_langs or []), }, ) return outcome async def _translate_block( self, client: Any, kind: str, source: dict[str, Any], lang: str, ) -> dict[str, Any]: """One language block, built from the baseline. Text is translated; everything else is carried over. The carried keys are the reason a comic can be "translated" at all — the pages are the same objects, and only the words around them change. """ target_name = _LANG_NAMES.get(lang, lang) title = str(source.get("title") or "").strip() summary = str(source.get("summary") or "").strip() meta = await client.chat_json( system=_SYSTEM, user=( f"Translate the following security-awareness card into " f"{target_name}.\n\n" f"TITLE (max {MAX_TITLE_LEN} characters in the output):\n" f"{title}\n\n" + ( f"SUMMARY — one line, max {MAX_SUMMARY_LEN} characters " f"in the output:\n{summary}\n\n" if summary else "SUMMARY: (none — return null)\n\n" ) + 'Return JSON: {"title": "...", "summary": "..." or null}' ), max_tokens=_META_MAX_TOKENS, temperature=0.2, ) new_title = _clip(meta.get("title"), MAX_TITLE_LEN) if not new_title: raise AppException( code="SECAW_TRANSLATION_EMPTY", message=f"model returned no title for {lang!r}", status_code=502, ) block: dict[str, Any] = {"title": new_title} new_summary = _clip(meta.get("summary"), MAX_SUMMARY_LEN) if new_summary: block["summary"] = new_summary if kind == KIND_ARTICLE: body = str(source.get("body") or "") if len(body) > MAX_BODY_CHARS: raise AppException( code="SECAW_BODY_TOO_LONG_TO_TRANSLATE", message=( f"article body is {len(body)} characters, over the " f"{MAX_BODY_CHARS}-character automatic-translation " "limit (the model's reply would be cut off " "mid-document). Translate this one by hand in the " "material editor." ), status_code=422, ) translated = (await client.chat_text( system=_SYSTEM, user=( f"Translate the following markdown document into " f"{target_name}. Output ONLY the translated markdown, " "with the same structure. Do not wrap it in a code " "fence.\n\n-----\n" + body + "\n-----" ), max_tokens=_BODY_MAX_TOKENS, temperature=0.2, )).strip() if not translated: raise AppException( code="SECAW_TRANSLATION_EMPTY", message=f"model returned no body for {lang!r}", status_code=502, ) block["body"] = translated if kind == KIND_COMIC: # Same JPEGs. The artwork is Chinese and stays Chinese — the # reader is told so by the ``machine_translated`` flag this # block earns, which the detail page renders as a notice. block["image_keys"] = list(source.get("image_keys") or []) if kind == KIND_LINK: block["link_url"] = source.get("link_url") if source.get("cover_key"): block["cover_key"] = source["cover_key"] return block def _clip(value: Any, limit: int) -> str: """Trim a model string to the column's cap. The prompt asks for a length and the model usually respects it, but a schema cap is not a suggestion — a 210-character title would be rejected by the write path the admin uses next, so we cut here rather than store something the editor cannot re-save. """ if not isinstance(value, str): return "" text = value.strip() return text[:limit] if len(text) > limit else text __all__ = [ "BACKFILL_STATUSES", "MAX_BODY_CHARS", "MaterialOutcome", "MaterialTranslator", "missing_langs", ]