"""Machine-translate the documentation and stage each language site for the build. Internal docs tooling, not part of the SDK. Maintained entirely by LLM; expect breaking changes to its CLI, formats and internals at any time. Nothing here is a supported interface. English under `docs/` is the source of truth. Every language in `i18n/languages.yml` gets generated pages under `i18n//pages/`, driven by that language's hand-written `instructions.md` and `glossary.json` plus the shared `i18n/general-prompt.md`. Corrections go into those inputs, never into the generated pages. Each generated page records, in its own front matter, the English section hashes it reflects; nothing else tracks state. Usage (from the repository root): python scripts/docs/translations.py status [--lang CODE] python scripts/docs/translations.py translate --lang CODE [--pages PATH ...] python scripts/docs/translations.py stage [--lang CODE] Only `translate` calls the model (credentials come from the environment, e.g. `ANTHROPIC_API_KEY`) and needs the `translate` dependency group; `DOCS_TRANSLATE_MODEL`, if set, replaces the registry's `model` for that run. Exit codes: 0 done, 1 some page failed, 2 configuration or credential error. """ import argparse import hashlib import importlib import json import os import posixpath import re import shutil import sys from collections import Counter from collections.abc import Callable, Iterator, Sequence from dataclasses import dataclass, field from pathlib import Path from typing import Any, Literal, Protocol, cast, get_args import markdown import yaml import zensical.config from build_config import ROOT, Language, Registry, load_registry, nav_page_paths, staged_docs_dir, staged_titles_file from llms_txt import CODE_SPAN, FRONT_MATTER, page_url from zensical.config import ConfigurationError # Zensical annotates its config loader `-> dict`; this is the shape its own renderer relies on. parse_mkdocs_config = cast("Callable[[str], dict[str, Any]]", getattr(zensical.config, "parse_mkdocs_config")) # Bumped only when the generated-file contract changes; older files then read as missing. TOOL_VERSION = 1 # `max_tokens` per request: several times the longest page, leaving room for # any thinking the model does, while inside the ceiling streaming allows. OUTPUT_TOKEN_BUDGET = 64_000 # Repair turns fed back to the model after the first reply before a page fails. MAX_REPAIRS = 2 NOTICES_PAGE = "i18n/notices.md" # The nav page the notices link to for how the translations are made. TRANSLATIONS_DOC = "translations.md" API_DIR = "api" Status = Literal["missing", "outdated", "current"] NoticeKind = Literal["translated", "outdated", "english"] class ConfigError(Exception): """Unusable configuration, inputs or credentials (exit code 2).""" class PageError(Exception): """One page cannot be translated; the run continues with the next page.""" # ---- Markdown structure: fences, headings, code spans, links ---- # An ATX heading the way the renderer reads it: hashes at column 0, no space # required after them, an optional closing hash run, backslash escapes honoured. HEADING = re.compile(r"^(?P#{1,6})(?!#)(?P(?:\\.|[^\\\n])*?)#*[ \t]*$") # A heading's trailing attr_list block(s), matched with or without the whitespace # attr_list itself needs, so blocks the model glued to CJK text or doubled are # still seen; `body` is the last block's, the one attr_list reads. Each block # parses one way only (`{` to the next `}`), so no run of them can backtrack. HEADING_ATTRS = re.compile(r"(?:[ \t]*\{(?P[^}\n]*)\})+[ \t]*$") ESCAPE = re.compile(r"\\(?P[!\"#$%&'()*+,\-./:;<=>?@\[\\\]^_`{|}~])") # An underscore that is not word-internal turns into emphasis before attr_list # reads the block, so it must be written escaped inside `{#...}`. _BOUNDARY_UNDERSCORE = re.compile(r"(?`{3,}|~{3,})(?P[^\n]*)$") LINK = re.compile(r"!?\[[^\]]*\]\((?P[^)\s]*)[^)]*\)") _URL = re.compile(r"[a-zA-Z][a-zA-Z0-9+.-]*://(?:(?![<>)\]])[!-~])+") _TAG = re.compile(r"\n]*>|", re.DOTALL) _MARKER = re.compile(r"^[ \t]*(?P!!!|\?\?\?\+?|===)[ \t]+(?P[\w-]+)?") # A bullet or ordered item at any depth, and a table row: block structure whose # count a faithful translation keeps, whatever the language. _LIST_ITEM = re.compile(r"^[ \t]*(?:[-*+]|\d{1,9}[.)])(?:[ \t]|$)") _TABLE_ROW = re.compile(r"^[ \t]*\|") _WRAPPER = re.compile( r"\A[ \t]*\n*(?P`{3,}|~{3,})[^\n]*\n(?P.*)\n(?P`{3,}|~{3,})[ \t]*\n*\Z", re.DOTALL ) # Bracketed text followed by `(` or `[` is a link label, never a placeholder. _PLACEHOLDERS = ( re.compile(r"\[\s*(?:translation|rest of|remaining)[^\]]*\](?![(\[])", re.IGNORECASE), re.compile(r"\((?:content )?omitted[^)]*\)", re.IGNORECASE), re.compile(r"\[\s*\.\.\.\s*\](?![(\[])"), ) _COMMENT = re.compile(r"", re.DOTALL) _ABRIDGED = re.compile(r"\b(?:omitted|continues|truncated|abridged|remaining|rest of)\b", re.IGNORECASE) # A language site builds no API reference; links into `api/` go to the English one. _API_LINK = re.compile(r"\]\((?:\.\./)*(?Papi/[^)#\s]+\.md)") @dataclass(frozen=True) class Heading: """One ATX heading; `text` excludes any trailing `{...}` attr block.""" line: int level: int text: str anchor: str | None @dataclass(frozen=True) class FenceLines: """Line indices of one fenced block; an unclosed fence runs to the last line.""" opener: int closer: int closed: bool def split_front_matter(text: str) -> tuple[str | None, str]: """`(front matter YAML, body)`; the YAML is None when the page has no front matter block.""" match = FRONT_MATTER.match(text) return (match["block"], text[match.end() :]) if match else (None, text) def fence_ranges(lines: Sequence[str]) -> list[FenceLines]: """The fenced blocks of `lines`, in order (CommonMark opener/closer rules).""" found: list[FenceLines] = [] opener: re.Match[str] | None = None start = 0 for index, line in enumerate(lines): match = _FENCE.match(line) if opener is None: if match and not (match["run"][0] == "`" and "`" in match["info"]): opener, start = match, index continue run = opener["run"] if match and match["run"][0] == run[0] and len(match["run"]) >= len(run) and not match["info"].strip(): found.append(FenceLines(start, index, closed=True)) opener = None if opener is not None: found.append(FenceLines(start, len(lines) - 1, closed=False)) return found def _blank(text: str, spans: list[tuple[int, int]]) -> str: """Replace each span with same-length spaces, keeping newlines so positions and lines survive.""" chars = list(text) for start, end in spans: for index in range(start, end): if chars[index] != "\n": chars[index] = " " return "".join(chars) def _mask_fences(text: str) -> str: """Blank every fenced block, marker lines included (same length, same lines).""" lines = text.split("\n") offsets = [0] for line in lines: offsets.append(offsets[-1] + len(line) + 1) return _blank(text, [(offsets[fence.opener], offsets[fence.closer + 1] - 1) for fence in fence_ranges(lines)]) def parse_headings(text: str) -> list[Heading]: """The ATX headings of `text` in order, with the id any trailing `{#...}` block pins.""" found: list[Heading] = [] for index, line in enumerate(_mask_fences(text).split("\n")): if not (match := HEADING.match(line)): continue heading, anchor = match["text"].strip(), None if attrs := HEADING_ATTRS.search(heading): tokens = attrs["body"].removeprefix(":").split() # `{: ...}` is attr_list's other spelling ids = [token[1:] for token in tokens if token.startswith("#") and len(token) > 1] anchor = ESCAPE.sub(r"\g", ids[-1]) if ids else None heading = heading[: attrs.start()].rstrip() found.append(Heading(index, len(match["hashes"]), heading, anchor)) return found def anchor_source_form(anchor: str) -> str: """The rendered id as it must be written inside `{#...}` to survive the emphasis pass.""" return _BOUNDARY_UNDERSCORE.sub(r"\\_", anchor) def mask_code(text: str) -> str: """Blank fenced blocks and inline code spans (same length, same lines).""" masked = _mask_fences(text) return _blank(masked, [match.span() for match in CODE_SPAN.finditer(masked)]) def mask(text: str) -> str: """Blank everything that is not translatable prose: code, link targets, bare URLs, HTML tags.""" masked = mask_code(text) spans = [match.span("target") for match in LINK.finditer(masked)] spans += [match.span() for pattern in (_URL, _TAG) for match in pattern.finditer(masked)] return _blank(masked, spans) def code_spans(text: str) -> Counter[str]: """The inline code spans of the prose (fenced code excluded), counted by content.""" return Counter(match.group(2).strip() for match in CODE_SPAN.finditer(_mask_fences(text))) def link_targets(text: str) -> Counter[str]: """The targets of the markdown links and images written in the prose of `text`, counted.""" return Counter(match["target"] for match in LINK.finditer(mask_code(text))) def markers(text: str) -> list[str]: """The admonition (`!!!`/`???`) markers with their type keyword, and the tab (`===`) markers, in order.""" matches = [match for line in _mask_fences(text).split("\n") if (match := _MARKER.match(line))] return ["===" if match["marker"] == "===" else f"{match['marker']} {match['kind'] or '?'}" for match in matches] def block_counts(text: str) -> dict[str, int]: """How many list items and table rows the prose of `text` has (fenced code excluded).""" lines = mask_code(text).split("\n") patterns = {"list items": _LIST_ITEM, "table rows": _TABLE_ROW} return {kind: sum(1 for line in lines if pattern.match(line)) for kind, pattern in patterns.items()} def abridgements(text: str) -> Counter[str]: """Placeholder phrases and "omitted" comments of the kind a model leaves where it cut a page short.""" prose = mask(text) found = [match.group(0) for pattern in _PLACEHOLDERS for match in pattern.finditer(prose)] found += [c.group(0) for c in _COMMENT.finditer(mask_code(text)) if _ABRIDGED.search(c["body"])] return Counter(found) # ---- Sections, hashes and the provenance front matter of a generated page ---- def sections(body: str) -> list[str]: """The page as intro + one string per `##` section; `"".join(sections(body)) == body`. Blank lines just above a `##` heading belong to that heading's section, so a section's bytes never depend on the section after it. """ lines = body.split("\n") offsets = [0] for line in lines: offsets.append(offsets[-1] + len(line) + 1) starts = [0] for heading in parse_headings(body): if heading.level != 2: continue first = heading.line while first > 0 and not lines[first - 1].strip(): first -= 1 starts.append(offsets[first]) return [body[start:end] for start, end in zip(starts, [*starts[1:], len(body)])] def section_label(section: str, index: int) -> str: """How a section is named in prompts and status lines.""" if index == 0: return "the introduction (everything before the first `##` heading)" return section.strip("\n").split("\n", 1)[0].strip() def section_hashes(body: str) -> list[str]: """The first 16 hex digits of each section's sha256.""" return [hashlib.sha256(section.encode("utf-8")).hexdigest()[:16] for section in sections(body)] def with_provenance(body: str, hashes: Sequence[str]) -> str: """The generated file: the English section hashes in front matter, then the translated body.""" record = {"translation": {"sections": list(hashes), "tool": TOOL_VERSION}} header = yaml.safe_dump(record, sort_keys=False, default_flow_style=None, width=2**16) return f"---\n{header}---\n{body}" def read_provenance(front_matter: str | None) -> tuple[str, ...] | None: """The section hashes a generated page records, or None if absent, unreadable or from another tool version.""" try: record = yaml.safe_load(front_matter or "")["translation"] return tuple(str(value) for value in record["sections"]) if record["tool"] == TOOL_VERSION else None except (yaml.YAMLError, TypeError, KeyError): return None # ---- The repository: registry, nav pages, prompt inputs and the renderer's heading ids ---- @dataclass(frozen=True) class Term: source: str target: str note: str = "" avoid: Sequence[str] = () @dataclass(frozen=True) class Glossary: keep: Sequence[str] terms: Sequence[Term] def _strings(value: object) -> bool: """Whether `value` is a JSON list of strings (a bare string is not).""" return isinstance(value, list) and all(isinstance(item, str) for item in cast("list[object]", value)) def load_glossary(path: Path) -> Glossary: """Parse `glossary.json`: `keep` strings and `terms` objects with the `Term` fields. Raises: ConfigError: The file is missing, not JSON, or not that shape. """ try: raw = json.loads(path.read_text(encoding="utf-8")) keep, entries = raw["keep"], [dict(entry) for entry in raw["terms"]] except (OSError, json.JSONDecodeError, TypeError, KeyError, ValueError) as exc: raise ConfigError(f"{path}: {exc!r}") from exc if not _strings(keep): raise ConfigError(f"{path}: `keep` must be a list of strings") for entry in entries: texts = all(isinstance(entry.get(key, ""), str) for key in ("source", "target", "note")) if not (texts and _strings(entry.get("avoid", []))): raise ConfigError(f"{path}: `terms` entry {entry} needs string `source`/`target`/`note` and a list `avoid`") try: terms = tuple(Term(**entry) for entry in entries) # a missing or unknown key is a TypeError except TypeError as exc: raise ConfigError(f"{path}: {exc!r}") from exc return Glossary(tuple(keep), terms) @dataclass(frozen=True) class Inputs: """One language's prompt inputs.""" language: Language general_prompt: str instructions: str glossary: Glossary @dataclass(frozen=True) class Page: """A translatable page: its display key, English source file and generated translation file.""" key: str source: Path target: Path @dataclass(frozen=True) class Notice: title: str body: str def parse_notices(body: str) -> dict[str, Notice]: """The `##` sections of a notices page keyed by their pinned id.""" lines = body.split("\n") headings = [heading for heading in parse_headings(body) if heading.level == 2] found: dict[str, Notice] = {} for position, heading in enumerate(headings): end = headings[position + 1].line if position + 1 < len(headings) else len(lines) if heading.anchor: found[heading.anchor] = Notice(heading.text, "\n".join(lines[heading.line + 1 : end]).strip("\n")) return found @dataclass class Repo: """Everything the commands read from a checkout rooted at `root`.""" root: Path docs: Path i18n: Path registry: Registry prose_pages: list[str] translatable: list[str] renderer: markdown.Markdown def language(self, code: str) -> Language: for language in self.registry.languages: if language.code == code: return language known = ", ".join(language.code for language in self.registry.languages) raise ConfigError(f"unknown language {code!r} (i18n/languages.yml has: {known})") def inputs(self, language: Language) -> Inputs: try: general = (self.i18n / "general-prompt.md").read_text(encoding="utf-8") instructions = (self.i18n / language.code / "instructions.md").read_text(encoding="utf-8") except OSError as exc: raise ConfigError(str(exc)) from exc return Inputs(language, general, instructions, load_glossary(self.i18n / language.code / "glossary.json")) def pages(self, language: Language) -> list[Page]: """The language's translatable pages in nav order, then the notices page.""" generated = self.i18n / language.code / "pages" pages = [Page(key, self.docs / key, generated / key) for key in self.translatable] return [*pages, Page(NOTICES_PAGE, self.root / NOTICES_PAGE, self.i18n / language.code / "notices.md")] def heading_ids(self, body: str) -> list[str]: """The ids the site renderer gives the page's headings, paired one-to-one with `parse_headings(body)`. Raises: PageError: The renderer sees headings the source scan does not (setext, indented, HTML). """ self.renderer.reset() self.renderer.convert(body) tokens = cast("list[dict[str, Any]]", getattr(self.renderer, "toc_tokens", [])) ids, found = [str(token["id"]) for token in _flatten(tokens)], parse_headings(body) if len(ids) != len(found): raise PageError(f"the page renders {len(ids)} headings but {len(found)} are ATX headings at column 0") return ids def _flatten(tokens: list[dict[str, Any]]) -> Iterator[dict[str, Any]]: for token in tokens: yield token yield from _flatten(cast("list[dict[str, Any]]", token["children"])) def _renderer(root: Path) -> markdown.Markdown: """A python-markdown instance configured with the extensions the site build uses.""" try: config = parse_mkdocs_config(str(root / "mkdocs.yml")) except (OSError, yaml.YAMLError, ConfigurationError) as exc: raise ConfigError(f"cannot load {root / 'mkdocs.yml'}: {exc}") from exc configs = cast("dict[str, dict[str, Any]]", config["mdx_configs"]) # Snippet paths resolve against the build's working directory, the # repository root; only heading ids are read here, so a missing one is not # this renderer's failure. snippets = configs.setdefault("pymdownx.snippets", {}) snippets["base_path"] = [str(root / base) for base in cast("list[str]", snippets.get("base_path", ["."]))] snippets["check_paths"] = False return markdown.Markdown(extensions=config["markdown_extensions"], extension_configs=configs) def _excluded(page: str, patterns: Sequence[str]) -> bool: """`dir/**` excludes a subtree; any other pattern is an exact page path.""" return any(page.startswith(p.removesuffix("**")) if p.endswith("/**") else page == p for p in patterns) def read_english(path: Path) -> str: """An English page's body; any front matter is dropped here, once, so no later step ever sees it. Raises: ConfigError: The file cannot be read. """ try: return split_front_matter(path.read_text(encoding="utf-8"))[1] except OSError as exc: raise ConfigError(f"cannot read {path}: {exc}") from exc def load_repo(root: Path) -> Repo: """Read the registry and the nav of the checkout at `root`, and check its English notices. Raises: ConfigError: Any of them is missing or malformed. """ try: registry = load_registry(root) config: object = yaml.safe_load((root / "mkdocs.yml").read_text(encoding="utf-8")) except (OSError, yaml.YAMLError, ValueError) as exc: # build_config reports registry problems as ValueError raise ConfigError(str(exc)) from exc nav = cast("dict[str, Any]", config).get("nav") if isinstance(config, dict) else None if not isinstance(nav, list): raise ConfigError(f"{root / 'mkdocs.yml'}: no nav list") prose = [p for p in nav_page_paths(cast("list[Any]", nav)) if p.endswith(".md") and not p.startswith(f"{API_DIR}/")] notices = parse_notices(read_english(root / NOTICES_PAGE)) if missing := [kind for kind in get_args(NoticeKind) if kind not in notices]: raise ConfigError(f"{root / NOTICES_PAGE}: missing `## ... {{#id}}` sections for {missing}") translatable = [page for page in prose if not _excluded(page, registry.exclude)] return Repo(root, root / "docs", root / "i18n", registry, prose, translatable, _renderer(root)) # ---- Page status ---- @dataclass(frozen=True) class Translation: """A generated page: the English section hashes it records (one per section of its body) and its body.""" sections: tuple[str, ...] body: str def recorded_sections(translation: Translation) -> dict[str, str]: """A generated page's sections keyed by the English section hash each records.""" return dict(zip(translation.sections, sections(translation.body), strict=True)) @dataclass(frozen=True) class PageState: """One page classified against the current English text; `translation` is None unless usable.""" page: Page english: str hashes: list[str] translation: Translation | None status: Status changed: list[int] = field(default_factory=list[int]) note: str = "" def classify(page: Page) -> PageState: """Compare the page's recorded section hashes with the current English ones. Raises: ConfigError: The English source cannot be read. """ english = read_english(page.source) hashes = section_hashes(english) if not page.target.is_file(): return PageState(page, english, hashes, None, "missing") front_matter, body = split_front_matter(page.target.read_text(encoding="utf-8")) recorded = read_provenance(front_matter) if recorded is None or len(recorded) != len(sections(body)): return PageState(page, english, hashes, None, "missing", note="unreadable front matter, retranslated whole") translation = Translation(recorded, body) if tuple(hashes) == recorded: return PageState(page, english, hashes, translation, "current") changed = [index for index, value in enumerate(hashes) if value not in set(recorded)] labels = ", ".join(section_label(sections(english)[index], index) for index in changed) note = f"English changed in: {labels}" if changed else "English sections removed or reordered" return PageState(page, english, hashes, translation, "outdated", changed, note) def removable(repo: Repo, language: Language) -> list[str]: """Generated pages whose English page left the translatable set (the fix is `git rm`).""" generated = repo.i18n / language.code / "pages" on_disk = sorted(path.relative_to(generated).as_posix() for path in generated.rglob("*.md")) return [page for page in on_disk if page not in set(repo.translatable)] # ---- Prompts and the model client ---- @dataclass(frozen=True) class Message: role: Literal["user", "assistant"] content: str @dataclass class Usage: input_tokens: int = 0 output_tokens: int = 0 cache_write_tokens: int = 0 cache_read_tokens: int = 0 def add(self, other: "Usage") -> None: self.input_tokens += other.input_tokens self.output_tokens += other.output_tokens self.cache_write_tokens += other.cache_write_tokens self.cache_read_tokens += other.cache_read_tokens def __str__(self) -> str: return ( f"{self.input_tokens} input / {self.output_tokens} output / " f"{self.cache_write_tokens} cache-write / {self.cache_read_tokens} cache-read tokens" ) @dataclass(frozen=True) class Completion: text: str usage: Usage stop_reason: str | None = "end_turn" class Translator(Protocol): """Anything that answers a conversation (`ConfigError`: credentials rejected; `PageError`: request failed).""" def complete(self, *, model: str, system: str, messages: Sequence[Message], max_tokens: int) -> Completion: ... def anthropic_translator() -> Translator: """The Claude Messages API client, streaming, with the system prompt as one cached block. `anthropic` lives in the non-default `translate` dependency group, so it is imported here, by name: offline commands and type checking never need it. Raises: ConfigError: The `translate` dependency group is not installed, or no credentials are configured. """ try: sdk = importlib.import_module("anthropic") except ImportError as exc: raise ConfigError( "the anthropic package is not installed; run with `uv run --frozen --group translate`" ) from exc # The SDK resolves every credential source it knows at construction; fail # here, before any page work, rather than on the first request. try: client = sdk.Anthropic() except sdk.AnthropicError as exc: # e.g. a credential profile it was pointed at is unreadable raise ConfigError(f"cannot set up the API client: {exc}") from exc if not (client.api_key or client.auth_token or client.credentials): raise ConfigError("no API credentials: set ANTHROPIC_API_KEY") class AnthropicTranslator: def complete(self, *, model: str, system: str, messages: Sequence[Message], max_tokens: int) -> Completion: prefix = [{"type": "text", "text": system, "cache_control": {"type": "ephemeral", "ttl": "1h"}}] turns = [{"role": message.role, "content": message.content} for message in messages] try: with client.messages.stream(model=model, max_tokens=max_tokens, system=prefix, messages=turns) as s: reply = s.get_final_message() except (sdk.AuthenticationError, sdk.PermissionDeniedError) as exc: raise ConfigError(f"the API rejected the credentials: {exc.message}") from exc except sdk.APIError as exc: raise PageError(f"API request failed: {exc.message}") from exc usage = Usage( reply.usage.input_tokens, reply.usage.output_tokens, reply.usage.cache_creation_input_tokens or 0, reply.usage.cache_read_input_tokens or 0, ) text = "".join(block.text for block in reply.content if block.type == "text") return Completion(text, usage, reply.stop_reason) return AnthropicTranslator() def glossary_prompt(glossary: Glossary) -> str: lines = ["## Glossary", "", "These terms always stay in English, spelled exactly like this:", ""] lines += [f"- {term}" for term in glossary.keep] if glossary.terms: lines += ["", "Use these renderings; the notes are binding:", ""] for term in glossary.terms: entry = f"- {term.source} → {term.target}" if term.avoid: entry += f" (never: {', '.join(term.avoid)})" if term.note: entry += f". {term.note}" lines.append(entry) return "\n".join(lines) def system_prompt(inputs: Inputs) -> str: """The cacheable prefix shared by every page of a language: rules, instructions, glossary.""" header = f"# Target language: {inputs.language.name} (`{inputs.language.code}`)" parts = (inputs.general_prompt, header, inputs.instructions, glossary_prompt(inputs.glossary)) return "\n\n".join(part.strip() for part in parts) def translate_request(english: str) -> str: return ( "Translate the following Markdown page. Return only the translated page.\n\n" f"\n{english}\n" ) def update_request(english: str, changed: Sequence[str], previous: str) -> str: listed = "\n".join(f"- {label}" for label in changed) return ( "This page was translated before. Retranslate it: translate the sections listed below\n" "afresh from the current English, applying the current language instructions and glossary\n" "(their previous wording may be outdated); everywhere else, reproduce the previous\n" "translation line by line, changing nothing. Keep the retranslated sections consistent in\n" "terminology and tone with their surroundings. A section is the introduction before the\n" "first `##` heading, or one `##` heading with everything under it.\n\n" f"Sections to retranslate:\n\n{listed}\n\n" f"Current English page:\n\n\n{english}\n\n\n" f"Previous translation of the page:\n\n\n{previous}\n\n\n" "Return only the full translated page." ) def repair_request(findings: Sequence[str]) -> str: listed = "\n".join(f"- {finding}" for finding in findings) return ( "Your translation broke the following structural rules. Fix each problem and return the\n" f"full corrected page, changing nothing else:\n\n{listed}" ) # ---- Re-imposing the English structure on a reply, and validating what cannot be re-imposed ---- @dataclass(frozen=True) class Mismatch: """A reply whose structure differs from the English; each finding says what to fix.""" findings: list[str] def unwrap(english: str, reply: str) -> str: """Drop a code fence wrapping the whole reply, and end it with exactly the English's trailing newlines.""" match = None if english.startswith(("```", "~~~")) else _WRAPPER.match(reply) if match and match["close"][0] == match["open"][0] and len(match["close"]) >= len(match["open"]): reply = match["body"] return reply.rstrip("\n") + english[len(english.rstrip("\n")) :] def reimpose(english: str, ids: Sequence[str], reply: str) -> str | Mismatch: """Copy over the reply what the model must never change, and check the links it placed. `ids` are the renderer's ids for the English headings (`Repo.heading_ids`). Fences are copied opener-through-closer within each section and `{#id}` blocks are pinned onto the translated headings positionally; each needs matching counts. Link and image targets are never moved (a translation may reorder links): each section must carry the same targets as its English, however placed. Checks go section by section, so any assembly of passing sections passes too. """ findings: list[str] = [] text = _restore_fences(english, reply, findings) text = _pin_headings(english, ids, text, findings) _check_targets(english, text, findings) return Mismatch(findings) if findings else text def _paired_sections(english: str, text: str) -> list[tuple[str, str, str]]: """`(label, English section, its counterpart)` per section, or the pages whole if their counts differ.""" want, got = sections(english), sections(text) if len(want) != len(got): # the heading finding says so already return [("the page", english, text)] return [(section_label(source, index), source, output) for index, (source, output) in enumerate(zip(want, got))] def _restore_fences(english: str, reply: str, findings: list[str]) -> str: if unclosed := [fence for fence in fence_ranges(reply.split("\n")) if not fence.closed]: findings.append(f"the code fence opened on line {unclosed[0].opener + 1} is never closed") return reply restored: list[str] = [] for label, source, output in _paired_sections(english, reply): kept, lines = source.split("\n"), output.split("\n") want, got = fence_ranges(kept), fence_ranges(lines) if len(want) != len(got): findings.append( f"{label}: {len(got)} code fences vs {len(want)} in the English: keep each where it is, add none" ) restored.append(output) continue result: list[str] = [] cursor = 0 for expected, found in zip(want, got): result += lines[cursor : found.opener] result += kept[expected.opener : expected.closer + 1] cursor = found.closer + 1 restored.append("\n".join([*result, *lines[cursor:]])) return "".join(restored) def _pin_headings(english: str, ids: Sequence[str], text: str, findings: list[str]) -> str: want, got = parse_headings(english), parse_headings(text) if len(want) != len(got): findings.append(f"{len(got)} headings vs {len(want)} in the English: keep every heading, and no others") return text if wrong := [(w, g) for w, g in zip(want, got) if w.level != g.level]: findings.extend(f"`{g.text}` is a level-{g.level} heading but `{w.text}` is level {w.level}" for w, g in wrong) return text lines = text.split("\n") for heading, anchor in zip(got, ids, strict=True): # `Repo.heading_ids` pairs ids with these headings lines[heading.line] = f"{'#' * heading.level} {heading.text} {{#{anchor_source_form(anchor)}}}" return "\n".join(lines) def _check_targets(english: str, text: str, findings: list[str]) -> None: missing: Counter[str] = Counter() extra: Counter[str] = Counter() for _, source, output in _paired_sections(english, text): expected, found = link_targets(source), link_targets(output) missing += expected - found extra += found - expected if missing: findings.append(f"missing links to {sorted(missing.elements())}: keep every link of the English where it is") if extra: findings.append(f"unexpected links to {sorted(extra.elements())}: add no links of your own") def validate(english: str, output: str, glossary: Glossary, label: str = "the page") -> list[str]: """Findings for what re-imposition cannot fix (an empty list means `output`, called `label`, passes).""" findings: list[str] = [] want, got = code_spans(english), code_spans(output) if missing := sorted((want - got).elements()): findings.append(f"missing inline code {missing}: copy every `code span` of the English") if extra := sorted((got - want).elements()): findings.append(f"unexpected inline code {extra}: use only the English `code spans`") if markers(english) != markers(output): findings.append( f"block markers {markers(output)} vs {markers(english)} in the English:" " keep each `!!!`/`???`/`===` line and its type" ) counted = block_counts(output) findings.extend( f"{label}: {counted[kind]} {kind} vs {count} in the English: translate them one for one, dropping none" for kind, count in block_counts(english).items() if counted[kind] != count ) folded = mask(output).casefold() findings.extend( f"banned rendering {avoid!r} of {term.source!r} appears: use {term.target!r}" for term in glossary.terms for avoid in term.avoid if avoid.casefold() in folded ) # Whatever the English itself carries is content, not an abridgement. placeholders = sorted((abridgements(output) - abridgements(english)).elements()) findings.extend(f"placeholder {found!r}: translate the whole page, never abridge it" for found in placeholders) return findings # ---- translate ---- @dataclass(frozen=True) class Job: """One page selected for translation: which sections the model rewrites, and the prior text to keep.""" state: PageState open: list[int] previous: Translation | None def select_jobs(states: Sequence[PageState], pages: Sequence[str]) -> list[Job]: """The missing and outdated pages with their changed sections open, or exactly `pages`, whole and fresh. Raises: ConfigError: A `pages` entry is not a translatable page. """ if unknown := [page for page in pages if page not in {state.page.key for state in states}]: raise ConfigError(f"not translatable pages (nav paths such as servers/tools.md): {unknown}") if pages: # a previous translation would carry nothing forward and only anchor the model on it return [_fresh(state) for state in states if state.page.key in pages] return [ _fresh(state) if state.translation is None else Job(state, state.changed, state.translation) for state in states if state.status != "current" ] def _fresh(state: PageState) -> Job: """The whole page from the English alone.""" return Job(state, list(range(len(state.hashes))), None) def build_messages(job: Job) -> list[Message]: english = job.state.english if job.previous is None: return [Message("user", translate_request(english))] labels = [section_label(text, index) for index, text in enumerate(sections(english)) if index in job.open] return [Message("user", update_request(english, labels, job.previous.body))] def carry_forward(job: Job, output: str) -> str: """Overwrite every section the model was not asked to rewrite with its previous translation. A section left closed is one whose English hash the previous file records (that is how `classify` closes it), so the lookup cannot miss. """ if job.previous is None: return output prior = recorded_sections(job.previous) paired = zip(job.state.hashes, sections(output), strict=True) return "".join(text if index in job.open else prior[value] for index, (value, text) in enumerate(paired)) def reassemble(repo: Repo, job: Job) -> str: """The page rebuilt from its recorded translations alone, for a job with no open section (no model call). Raises: PageError: The rebuilt page no longer fits the English structure. """ english = job.state.english result = reimpose(english, repo.heading_ids(english), carry_forward(job, english)) if isinstance(result, Mismatch): raise PageError("; ".join(result.findings)) return result def _validate_open(english: str, body: str, job: Job, glossary: Glossary) -> list[str]: """`validate` over the sections this run rewrites; a carried section is published text, not this run's to fix.""" paired = enumerate(zip(sections(english), sections(body), strict=True)) rewritten = [ (section_label(source, index), source, output) for index, (source, output) in paired if index in job.open ] return [finding for label, source, output in rewritten for finding in validate(source, output, glossary, label)] def translate_page(repo: Repo, inputs: Inputs, job: Job, translator: Translator, model: str, usage: Usage) -> str: """Translate one page and return its body; token usage accumulates into `usage`. Each reply has its carried sections overwritten before any check runs, so only text this run keeps can cost a repair turn or fail the page. Raises: PageError: The page could not be produced (API failure, refusal, or unrepairable structure). ConfigError: The credentials were rejected. """ if not job.open: return reassemble(repo, job) english = job.state.english ids = repo.heading_ids(english) system, messages = system_prompt(inputs), build_messages(job) findings: list[str] = [] for _ in range(1 + MAX_REPAIRS): completion = translator.complete(model=model, system=system, messages=messages, max_tokens=OUTPUT_TOKEN_BUDGET) usage.add(completion.usage) if completion.stop_reason == "max_tokens": raise PageError(f"the reply was cut off at {OUTPUT_TOKEN_BUDGET} output tokens") if completion.stop_reason == "refusal": raise PageError("the model declined to translate this page") reply = unwrap(english, completion.text) # Only a reply whose sections line up with the English can be assembled; one that # does not is checked as it stands and fails the heading check. if len(sections(reply)) == len(job.state.hashes): reply = carry_forward(job, reply) result = reimpose(english, ids, reply) # today's ids pinned on carried sections too if isinstance(result, Mismatch): findings = result.findings elif not (findings := _validate_open(english, result, job, inputs.glossary)): # checked as it would be written return result messages += [Message("assistant", completion.text), Message("user", repair_request(findings))] raise PageError(f"unfixed after {MAX_REPAIRS} repairs: " + "; ".join(findings)) def command_translate(repo: Repo, args: argparse.Namespace, translator: Translator | None) -> int: language = repo.language(args.lang) inputs = repo.inputs(language) jobs = select_jobs([classify(page) for page in repo.pages(language)], args.pages) if not jobs: print(f"{language.code}: nothing to translate") return 0 model = os.environ.get("DOCS_TRANSLATE_MODEL") or repo.registry.model # never recorded in the generated files # Only a job with open sections calls the model; a run without one needs no client and no # credentials. Otherwise both are set up here, so bad credentials fail before any page work. if translator is None and any(job.open for job in jobs): translator = anthropic_translator() usage, failed = Usage(), False for job in jobs: page = job.state.page try: body = translate_page(repo, inputs, job, translator, model, usage) if translator else reassemble(repo, job) except PageError as exc: failed = True print(f"error: {page.key}: {exc}", file=sys.stderr) continue page.target.parent.mkdir(parents=True, exist_ok=True) page.target.write_text(with_provenance(body, job.state.hashes), encoding="utf-8", newline="\n") print(f"translated: {page.key} ({len(job.open)} of {len(job.state.hashes)} sections)", flush=True) print(f"usage: {usage}") return 1 if failed else 0 # ---- stage ---- def serve(state: PageState) -> tuple[str, NoticeKind]: """What a language site shows for a page: its stored translation exactly as generated, else English. A generated page had its heading ids and code pinned against the English it was made from, so served verbatim it can never pair prose with another section's code or ids; bringing it up to date is the translate run's job. """ if state.translation is None: return state.english, "english" return state.translation.body, "translated" if state.status == "current" else "outdated" def english_site(page: str) -> str: """The English site root as a link from `page` on a language site reads it. The renderer resolves a page's relative links against its source path, so this climbs out of the page's directory, then out of the language site. """ return "../" * page.count("/") + "../" def render_notice(notice: Notice, kind: NoticeKind, page: str) -> str: """The notice as an admonition (collapsed for translated pages) with its placeholder links filled. Links are relative to the staged page, so they hold under whatever path the sites are served: the English page is the same path one site up, and the translations page is this site's own `translations.md`. """ body = notice.body.replace("(ENGLISH_PAGE)", f"({english_site(page)}{page_url(page)})") body = body.replace("(TRANSLATIONS_PAGE)", f"({posixpath.relpath(TRANSLATIONS_DOC, posixpath.dirname(page))})") marker = "???" if kind == "translated" else "!!!" title = notice.title.replace('"', "'") lines = [f'{marker} note "{title}"', "", *(f" {line}" if line.strip() else "" for line in body.split("\n"))] return "\n".join(lines) def page_title(body: str) -> Heading | None: """The page's first `#` heading, if it has one.""" return next((heading for heading in parse_headings(body) if heading.level == 1), None) def inject_notice(body: str, notice: str) -> str: """Place the notice right after the page's first `#` heading, or first if there is none.""" lines = body.split("\n") title = page_title(body) if title is None: return f"{notice}\n\n{body}" rest = "\n".join(lines[title.line + 1 :]).lstrip("\n") return "\n".join([*lines[: title.line + 1], "", notice, "", rest]) def stage(repo: Repo, language: Language) -> Path: """Build `.build/i18n//docs`: the English tree minus `api/`, translations overlaid, notices injected. Beside it, `titles.json` records each staged page's `#` heading for `build_config.py --lang` to title nav sections with. Only English and the generated pages are read (never the prompt inputs), so nothing a language maintainer edits by hand can break a site build. """ states = {state.page.key: state for state in map(classify, repo.pages(language))} notices_page = states.pop(NOTICES_PAGE) # a kind its translation lacks (added since) stays English notices = {**parse_notices(notices_page.english), **parse_notices(serve(notices_page)[0])} target, titles_file = staged_docs_dir(language.code, repo.root), staged_titles_file(language.code, repo.root) # The titles file goes last, so it only ever sits beside a complete tree. titles_file.unlink(missing_ok=True) shutil.rmtree(target, ignore_errors=True) def left_out(directory: str, names: list[str]) -> list[str]: return [n for n in names if n.startswith(".") or (n == API_DIR and Path(directory) == repo.docs)] shutil.copytree(repo.docs, target, ignore=left_out) titles: dict[str, str] = {} for page in repo.prose_pages: # a page excluded from translation is staged as its English page body, kind = serve(states[page]) if page in states else (read_english(repo.docs / page), "english") # The one API reference is the English site's. body = _API_LINK.sub(lambda match: f"]({english_site(page)}{page_url(match['page'])}", body) if title := page_title(body): titles[page] = title.text body = inject_notice(body, render_notice(notices[kind], kind, page)) (target / page).write_text(body, encoding="utf-8", newline="\n") listing = json.dumps(titles, ensure_ascii=False, indent=0, sort_keys=True) titles_file.write_text(listing + "\n", encoding="utf-8", newline="\n") return target def command_stage(repo: Repo, args: argparse.Namespace) -> int: for language in [repo.language(args.lang)] if args.lang else repo.registry.languages: target = stage(repo, language) print(f"staged {language.code} at {target.relative_to(repo.root).as_posix()}", flush=True) return 0 # ---- status ---- def command_status(repo: Repo, args: argparse.Namespace) -> int: languages = [repo.language(args.lang)] if args.lang else repo.registry.languages for language in languages: states = [classify(page) for page in repo.pages(language)] strays = removable(repo, language) counts = Counter(state.status for state in states) print( f"{language.code} ({language.name}): {counts['missing']} missing, {counts['outdated']} outdated," f" {counts['current']} current, {len(strays)} removable" ) for state in states: if state.status != "current": print(f" {state.status:<9} {state.page.key}" + (f" ({state.note})" if state.note else "")) for page in strays: print(f" {'removable':<9} {page} (git rm i18n/{language.code}/pages/{page})") return 0 # ---- Command line ---- def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="translations.py", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter ) commands = parser.add_subparsers(dest="command", required=True) status = commands.add_parser("status", help="what each language is missing") status.add_argument("--lang", metavar="CODE") translate = commands.add_parser("translate", help="translate missing and outdated pages (calls the model)") translate.add_argument("--lang", metavar="CODE", required=True) translate.add_argument( "--pages", nargs="+", metavar="PATH", default=[], help="re-translate exactly these pages from scratch" ) staged = commands.add_parser("stage", help="assemble .build/i18n/CODE/docs for the site build") staged.add_argument("--lang", metavar="CODE", help="stage this language only (default: every language)") return parser def main(argv: Sequence[str] | None = None, *, root: Path = ROOT, translator: Translator | None = None) -> int: """Run one command against the checkout at `root`; returns the exit code.""" args = _parser().parse_args(argv) try: repo = load_repo(root) if args.command == "status": return command_status(repo, args) if args.command == "translate": return command_translate(repo, args, translator) return command_stage(repo, args) except ConfigError as exc: print(f"translations: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())