diff --git a/cli/build/i18n/__main__.py b/cli/build/i18n/__main__.py index 10bb727d2a..be3f0336b4 100644 --- a/cli/build/i18n/__main__.py +++ b/cli/build/i18n/__main__.py @@ -14,7 +14,7 @@ import json import os import sys import tempfile -from concurrent.futures import ProcessPoolExecutor +from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor from pathlib import Path from utils.cache.files import PROJECT_ROOT @@ -108,39 +108,71 @@ def prune(domains: list[str], requested: list[str]) -> int: return 0 -def translate(domain: str, requested: list[str]) -> int: - supported = translatable(load_languages(PROJECT_ROOT), domain) - unsupported = sorted(set(requested) - set(supported)) - if unsupported: - print(f"LibreTranslate does not support {unsupported}", file=sys.stderr) - return 2 - codes = [ - code - for code in (sorted(requested) if requested else supported) - if pending(read_catalog(catalog_path(PROJECT_ROOT, code, domain))) - ] - if not codes: - print(f"{domain}: nothing to translate") +def usable_cpus() -> int: + """Return the CPUs this process may run on.""" + try: + return len(os.sched_getaffinity(0)) + except (AttributeError, OSError): + return os.cpu_count() or 1 + + +def one_catalog(client: LibreTranslate, domain: str, code: str) -> None: + """Translate every pending entry of one catalog and write it back. + + Args: + client: translation client. + domain: catalog domain. + code: ISO 639-1 code of the catalog. + """ + path = catalog_path(PROJECT_ROOT, code, domain) + catalog = read_catalog(path) + todo = pending(catalog) + discarded = refused = damaged = 0 + refusal = "" + for start in range(0, len(todo), CHUNK_SIZE): + chunk = todo[start : start + CHUNK_SIZE] + outcome = client.translate([m.id for m in chunk], code) + discarded += apply(chunk, outcome.values) + refused += outcome.refused + damaged += outcome.damaged + refusal = outcome.refusal or refusal + write_catalog(path, catalog) + print(f"{domain}/{code}: {start + len(chunk)}/{len(todo)}", flush=True) + tail = f", last refusal {refusal}" if refused else "" + print( + f"{domain}/{code}: {len(todo) - discarded} translated, {discarded} discarded " + f"({refused} requests refused, {damaged} damaged a protected span){tail}", + flush=True, + ) + + +def translate(domains: list[str], requested: list[str]) -> int: + loaded = load_languages(PROJECT_ROOT) + work: list[tuple[str, str]] = [] + for domain in domains: + supported = translatable(loaded, domain) + unsupported = sorted(set(requested) - set(supported)) + if unsupported: + print(f"LibreTranslate does not support {unsupported}", file=sys.stderr) + return 2 + work += [ + (domain, code) + for code in (sorted(requested) if requested else supported) + if pending(read_catalog(catalog_path(PROJECT_ROOT, code, domain))) + ] + if not work: + print(f"{', '.join(domains)}: nothing to translate") return 0 - threads = os.cpu_count() or 1 - with server(PROJECT_ROOT, codes, threads) as url: - client = LibreTranslate(url, threads) + + cpus = usable_cpus() + codes = sorted({code for _, code in work}) + lanes = min(len(work), cpus) + with server(PROJECT_ROOT, codes, cpus) as url: + client = LibreTranslate(url, max(cpus // lanes, 1)) client.wait(codes, READY_TIMEOUT_SECONDS) - for code in codes: - path = catalog_path(PROJECT_ROOT, code, domain) - catalog = read_catalog(path) - todo = pending(catalog) - discarded = 0 - for start in range(0, len(todo), CHUNK_SIZE): - chunk = todo[start : start + CHUNK_SIZE] - discarded += apply(chunk, client.translate([m.id for m in chunk], code)) - write_catalog(path, catalog) - print(f"{domain}/{code}: {start + len(chunk)}/{len(todo)}", flush=True) - print( - f"{domain}/{code}: {len(todo) - discarded} translated, " - f"{discarded} discarded", - flush=True, - ) + with ThreadPoolExecutor(lanes) as pool: + for _ in pool.map(lambda job: one_catalog(client, *job), work): + pass return 0 @@ -152,7 +184,7 @@ def main() -> int: translate_parser = commands.add_parser( "translate", help="Machine-translate empty and fuzzy entries." ) - translate_parser.add_argument("--domain", choices=DOMAINS, required=True) + translate_parser.add_argument("--domain", choices=DOMAINS) translate_parser.add_argument( "--languages", default="", @@ -181,7 +213,10 @@ def main() -> int: [args.domain] if args.domain else list(DOMAINS), [c for c in args.languages.split(",") if c], ) - return translate(args.domain, [c for c in args.languages.split(",") if c]) + return translate( + [args.domain] if args.domain else list(DOMAINS), + [c for c in args.languages.split(",") if c], + ) if __name__ == "__main__": diff --git a/tests/unit/python/cli/build/i18n/test_main.py b/tests/unit/python/cli/build/i18n/test_main.py index b7d4b008d4..4c3875e688 100644 --- a/tests/unit/python/cli/build/i18n/test_main.py +++ b/tests/unit/python/cli/build/i18n/test_main.py @@ -9,6 +9,7 @@ from tempfile import TemporaryDirectory from utils.i18n.catalog import catalog_path, new_catalog, write_catalog from utils.i18n.languages import LANGUAGES_FILE +from utils.i18n.libretranslate import Outcome cli = importlib.import_module("cli.build.i18n.__main__") @@ -39,15 +40,15 @@ class TestTranslate(unittest.TestCase): def _run(self) -> tuple[int, mock.Mock]: started = mock.Mock(return_value=contextlib.nullcontext("http://lt")) client = mock.Mock() - client.return_value.translate.side_effect = lambda texts, code: ( - ["Bonjour"] * len(texts) + client.return_value.translate.side_effect = lambda texts, code: Outcome( + ["Bonjour"] * len(texts), 0, 0, "" ) with ( mock.patch.object(cli, "PROJECT_ROOT", self.root), mock.patch.object(cli, "server", started), mock.patch.object(cli, "LibreTranslate", client), ): - return cli.translate("core", []), started + return cli.translate(["core"], []), started def test_only_languages_with_pending_entries_start_the_container(self) -> None: _catalog(self.root, "fr", "") diff --git a/tests/unit/python/utils/i18n/test_libretranslate.py b/tests/unit/python/utils/i18n/test_libretranslate.py index 6698330480..3d4e3a8ba5 100644 --- a/tests/unit/python/utils/i18n/test_libretranslate.py +++ b/tests/unit/python/utils/i18n/test_libretranslate.py @@ -63,12 +63,93 @@ class TestTranslateFailures(unittest.TestCase): rejected = urllib.error.HTTPError( "http://libretranslate", 500, "boom", {}, None ) - with mock.patch.object(self.client, "_call", side_effect=rejected): + with ( + mock.patch("utils.i18n.libretranslate.time.sleep"), + mock.patch.object(self.client, "_call", side_effect=rejected), + ): self.assertEqual( - self.client.translate(["Hello", "World"], "de"), [None, None] + self.client.translate(["Hello", "World"], "de").values, [None, None] ) +class TestRefusalsAreRetried(unittest.TestCase): + """A server that buckles under the lanes must not cost entries for good.""" + + def setUp(self) -> None: + self.client = LibreTranslate("http://libretranslate", 1) + self.refusal = urllib.error.HTTPError( + "http://libretranslate", 500, "boom", {}, None + ) + + def test_a_transient_refusal_is_retried_until_it_succeeds(self) -> None: + answers = [self.refusal, self.refusal, {"translatedText": ["Hallo"]}] + with ( + mock.patch("utils.i18n.libretranslate.time.sleep"), + mock.patch.object(self.client, "_call", side_effect=answers), + ): + outcome = self.client.translate(["Hello"], "de") + + self.assertEqual(outcome.values, ["Hallo"]) + self.assertEqual(outcome.refused, 2) + self.assertEqual(outcome.damaged, 0) + + def test_a_permanent_refusal_is_counted_and_named(self) -> None: + with ( + mock.patch("utils.i18n.libretranslate.time.sleep"), + mock.patch.object(self.client, "_call", side_effect=self.refusal), + ): + outcome = self.client.translate(["Hello"], "de") + + self.assertEqual(outcome.values, [None]) + self.assertTrue(outcome.refused) + self.assertIn("HTTPError", outcome.refusal) + + def test_a_catalog_never_inherits_another_lane_s_counts(self) -> None: + answers = [ + self.refusal, + {"translatedText": ["Hallo"]}, + {"translatedText": ["Hi"]}, + ] + with ( + mock.patch("utils.i18n.libretranslate.time.sleep"), + mock.patch.object(self.client, "_call", side_effect=answers), + ): + first = self.client.translate(["Hello"], "de") + second = self.client.translate(["Hello"], "de") + + self.assertEqual(first.refused, 1) + self.assertEqual(second.refused, 0) + + +class TestScriptedTargetCodes(unittest.TestCase): + """LibreTranslate offers Chinese as zh-Hans/zh-Hant, never as bare zh.""" + + def setUp(self) -> None: + self.client = LibreTranslate("http://libretranslate", 1) + self.served = [{"code": "en", "targets": ["de", "zh-Hans", "zh-Hant"]}] + + def test_the_catalog_code_is_sent_as_the_code_the_server_serves(self) -> None: + with mock.patch.object( + self.client, "_call", return_value={"translatedText": ["你好"]} + ) as call: + self.client.translate(["Hello"], "zh") + + self.assertEqual(call.call_args.args[1]["target"], "zh-Hans") + + def test_waiting_accepts_a_server_that_only_names_the_script(self) -> None: + with mock.patch.object(self.client, "_call", return_value=self.served): + self.client.wait(["zh", "de"], timeout=0) + + def test_waiting_still_reports_a_target_nobody_serves(self) -> None: + with ( + mock.patch.object(self.client, "_call", return_value=self.served), + self.assertRaises(TimeoutError) as raised, + ): + self.client.wait(["fr"], timeout=0) + + self.assertIn("fr", str(raised.exception)) + + class TestAccelerated(unittest.TestCase): def _runtimes(self, stdout: str) -> mock.Mock: return mock.Mock(spec=subprocess.CompletedProcess, stdout=stdout) diff --git a/utils/i18n/libretranslate.py b/utils/i18n/libretranslate.py index 4a02b85a23..287362b2f3 100644 --- a/utils/i18n/libretranslate.py +++ b/utils/i18n/libretranslate.py @@ -5,14 +5,13 @@ from __future__ import annotations import json import os import subprocess -import sys import time import urllib.error import urllib.request from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, NamedTuple from utils.cache.yaml import load_yaml from utils.i18n.languages import SOURCE_LANGUAGE @@ -20,9 +19,42 @@ from utils.i18n.placeholders import has_words, mask, unmask from utils.roles.mapping import ROLE_FILE_META_SERVICES if TYPE_CHECKING: - from collections.abc import Iterator + from collections.abc import Iterable, Iterator + +class Outcome(NamedTuple): + """What one translation call produced, and why anything is missing. + + Args: + values: one translation per source, None where it was discarded. + refused: requests the server turned down, retries included. + damaged: translations that altered a protected span. + refusal: the last refusal's message, empty when none happened. + """ + + values: list[str | None] + refused: int + damaged: int + refusal: str + + +def merge(outcomes: Iterable[Outcome]) -> Outcome: + """Fold per-text outcomes back into one for the whole batch.""" + values: list[str | None] = [] + refused = damaged = 0 + refusal = "" + for outcome in outcomes: + values += outcome.values + refused += outcome.refused + damaged += outcome.damaged + refusal = outcome.refusal or refusal + return Outcome(values, refused, damaged, refusal) + ROLE = "web-svc-libretranslate" +# LibreTranslate names a few targets by script rather than by the bare ISO +# 639-1 code the catalogs use. A code missing from this map simply never turns +# up in /languages, and the readiness wait then spins until its timeout. +SERVER_CODES = {"zh": "zh-Hans"} SERVICES_FILE = Path("roles") / ROLE / ROLE_FILE_META_SERVICES CONTAINER_PORT = 5000 MODELS_VOLUME = "infinito-i18n-libretranslate" @@ -33,8 +65,11 @@ POLL_SECONDS = 5 READY_TIMEOUT_SECONDS = 3600 REQUEST_TIMEOUT_SECONDS = 600 DEPLOYED_TIMEOUT_SECONDS = 5 -DEPLOY_TIMEOUT_SECONDS = 3600 +DEPLOY_TIMEOUT_SECONDS = 4 * 3600 RUNNER_TIMEOUT_SECONDS = 300 +SERVICE_TIMEOUT_SECONDS = 300 +RETRY_ATTEMPTS = 4 +RETRY_BACKOFF_SECONDS = 2 INVENTORY_DIR = Path.home() / "inventories" / "infinito-i18n" DEPLOY_PID_FILE = Path("build") / "deploy.pid" GPU_STAMP = Path("build") / "i18n-gpu.stamp" @@ -80,6 +115,27 @@ def deployed(root: Path) -> str: return "" +def await_service(root: Path) -> str: + """Poll the deployed LibreTranslate until it answers, or give up. + + A service the playbook just created needs longer than one probe's timeout + to serve its first request, and a single miss would send the caller off to + spawn a throwaway container beside the one it just deployed. + + Args: + root: repository root. + + Returns: + The base URL, empty when nothing answered before the deadline. + """ + deadline = time.monotonic() + SERVICE_TIMEOUT_SECONDS + while True: + running = deployed(root) + if running or time.monotonic() > deadline: + return running + time.sleep(POLL_SECONDS) + + def unaccelerated(root: Path) -> bool: """Return whether the running LibreTranslate predates this host's GPU. @@ -186,7 +242,6 @@ def deploy(root: Path) -> str: """ from cli.administration.deploy.development.env import compose_file_args - development = [sys.executable, "-m", "cli.administration.deploy.development"] inside = [ "docker", "compose", @@ -212,6 +267,8 @@ def deploy(root: Path) -> str: "--force-recreate", "i18n", ], + [*inside[:-2], "sh", "-lc", "/usr/local/bin/package-frontend-ca.sh"], + [*inside[:-2], "systemctl", "daemon-reload"], [*inside[:-2], "bash", f"{environment(root)['INFINITO_SRC_DIR']}/{bootstrap}"], [ *inside, @@ -252,7 +309,7 @@ def deploy(root: Path) -> str: ) if index == 0: await_runner(name, outer) - running = deployed(root) + running = await_service(root) if running and accelerated(): (root / GPU_STAMP).parent.mkdir(exist_ok=True) (root / GPU_STAMP).touch() @@ -323,7 +380,13 @@ def container(image: str, codes: list[str], threads: int) -> Iterator[str]: if accelerated(): command += ["--gpus", "all"] image = f"{image}{CUDA_SUFFIX}" - subprocess.run([*command, image], check=True, capture_output=True) + started = subprocess.run( + [*command, image], capture_output=True, text=True, check=False + ) + if started.returncode: + raise RuntimeError( + f"docker run exited with {started.returncode}: {started.stderr.strip()}" + ) try: mapping = subprocess.run( ["docker", "port", name, f"{CONTAINER_PORT}/tcp"], @@ -362,6 +425,11 @@ class LibreTranslate: ) as response: return json.load(response) + @staticmethod + def server_code(code: str) -> str: + """Return the target code the server knows this catalog code by.""" + return SERVER_CODES.get(code, code) + def targets(self) -> set[str]: """Return the languages the server translates English into.""" for language in self._call("/languages"): @@ -377,10 +445,11 @@ class LibreTranslate: timeout: seconds to wait before giving up. """ deadline = time.monotonic() + timeout - missing = set(codes) + wanted = {self.server_code(code) for code in codes} + missing = set(wanted) while True: try: - missing = set(codes) - self.targets() + missing = wanted - self.targets() if not missing: return except (OSError, ValueError): @@ -391,30 +460,49 @@ class LibreTranslate: ) time.sleep(POLL_SECONDS) - def _batch(self, texts: list[str], target: str) -> list[str | None]: + def _post(self, masked: list, target: str) -> list | None: + """Return the server's translations, or None once the retries run out. + + The server refuses a share of the requests while every lane hammers it + at once, and those refusals used to discard their entries for good. + """ + refusals = 0 + refusal = "" + for attempt in range(RETRY_ATTEMPTS): + try: + result = self._call( + "/translate", + { + "q": [item.text for item in masked], + "source": SOURCE_LANGUAGE, + "target": self.server_code(target), + "format": "html", + }, + )["translatedText"] + except (urllib.error.HTTPError, ValueError, KeyError, TypeError) as exc: + refusals += 1 + refusal = f"{type(exc).__name__}: {exc}" + result = None + if isinstance(result, list) and len(result) == len(masked): + return result, refusals, refusal + time.sleep(RETRY_BACKOFF_SECONDS * 2**attempt) + return None, refusals, refusal + + def _batch(self, texts: list[str], target: str) -> Outcome: masked = [mask(text) for text in texts] - try: - result = self._call( - "/translate", - { - "q": [item.text for item in masked], - "source": SOURCE_LANGUAGE, - "target": target, - "format": "html", - }, - )["translatedText"] - except (urllib.error.HTTPError, ValueError, KeyError, TypeError): - result = None - if not isinstance(result, list) or len(result) != len(texts): + result, refused, refusal = self._post(masked, target) + if result is None: if len(texts) == 1: - return [None] - return [self._batch([text], target)[0] for text in texts] - return [ + return Outcome([None], refused, 0, refusal) + return merge(self._batch([text], target) for text in texts) + values = [ unmask(translated, item, text) for translated, item, text in zip(result, masked, texts, strict=True) ] + damaged = sum(1 for text in values if text is None) + return Outcome(values, refused, damaged, refusal) - def translate(self, texts: list[str], target: str) -> list[str | None]: + def translate(self, texts: list[str], target: str) -> Outcome: """Translate ``texts`` from English into ``target``. Args: @@ -422,8 +510,10 @@ class LibreTranslate: target: ISO 639-1 code. Returns: - One translation per source; ``None`` where the server rejected the - request or the translation damaged a protected span. + One translation per source, ``None`` where it was discarded, plus + the counts that say which of the two reasons applied. The counts + belong to this call: lanes share the client, so a field on the + client would mix another catalog's numbers into this one's. Raises: OSError: the server is unreachable; every further request would @@ -432,12 +522,17 @@ class LibreTranslate: results: list[str | None] = list(texts) wordy = [index for index, text in enumerate(texts) if has_words(text)] batches = [wordy[i : i + BATCH_SIZE] for i in range(0, len(wordy), BATCH_SIZE)] + refused = damaged = 0 + refusal = "" with ThreadPoolExecutor(self.workers) as pool: translated = pool.map( lambda batch: self._batch([texts[index] for index in batch], target), batches, ) - for batch, texts_out in zip(batches, translated, strict=True): - for index, text in zip(batch, texts_out, strict=True): + for batch, outcome in zip(batches, translated, strict=True): + for index, text in zip(batch, outcome.values, strict=True): results[index] = text - return results + refused += outcome.refused + damaged += outcome.damaged + refusal = outcome.refusal or refusal + return Outcome(results, refused, damaged, refusal)