From ffdbdbaf074c6e94c9e0608755f3483524aebc76 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 8 Jul 2026 20:13:49 -0400 Subject: [PATCH] feat(translation): 8h sweep + drain button + detection probe (#1376) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Throughput: translate_posts now runs every 8h (was daily) as the steady-state cadence for newly-imported posts, and the Settings "Translate now" button runs it in drain mode (run-until-done, no reset) so one press clears the whole untranslated backlog instead of a single 300-post chunk. The interrupt/backoff re-enqueue now preserves the drain flag so a bulk drain resumes cleanly after an Interpreter restart. Misdetection groundwork: surface the detector's confidence from the Interpreter client (it was in the detectedLanguage payload but discarded) and add a read-only "Test translation" box — POST /settings/translation/ probe + TranslationCard UI — that shows detected language + confidence + engine + result for pasted text, without saving. Lets the operator see why a short/abbreviation-heavy English title gets mis-detected so the detection guard (min-length + confidence floor) can be tuned from real numbers. The guard itself follows once the mis-detected cases are probed. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CgZP9v2otxVJymiYsnVuMy --- backend/app/api/settings.py | 44 +++++++++- backend/app/celery_app.py | 8 +- backend/app/services/interpreter_client.py | 10 ++- backend/app/tasks/translation.py | 44 +++++++--- .../components/settings/TranslationCard.vue | 81 +++++++++++++++++++ tests/test_api_settings.py | 43 +++++++++- tests/test_interpreter_client.py | 1 + tests/test_translate_posts.py | 63 ++++++++++++++- 8 files changed, 274 insertions(+), 20 deletions(-) diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index 53e4a8c..fb2c052 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -364,9 +364,49 @@ async def translation_test(): return jsonify({"healthy": healthy}) +@settings_bp.route("/settings/translation/probe", methods=["POST"]) +async def translation_probe(): + """Diagnostic for the Settings 'Test translation' box: translate a pasted + snippet WITHOUT saving anything, returning what Interpreter *detected* + (language + confidence) alongside the result. Lets the operator see why a + given string was (mis-)detected — e.g. a short English title flagged as + another language — so a detection guard can be tuned from real numbers. + Read-only: no post is touched. Uses the currently-saved base URL + target.""" + body = await request.get_json(silent=True) + body = body if isinstance(body, dict) else {} + text = (body.get("text") or "").strip() + if not text: + return jsonify({"error": "provide text to translate"}), 400 + async with get_session() as session: + cfg = await ImportSettings.load(session) + base_url = (cfg.interpreter_base_url or "").strip() + if not base_url: + return jsonify({"error": "no Interpreter base URL is set"}), 400 + target = (cfg.translation_target_lang or "en").strip() or "en" + try: + res = await asyncio.to_thread( + ic.translate, [text], base_url=base_url, target=target, + ) + except ic.InterpreterUnavailable as e: + return jsonify({"error": f"Interpreter unavailable: {e}"}), 503 + except ic.InterpreterBadRequest as e: + return jsonify({"error": f"Interpreter rejected the request: {e}"}), 400 + translations = res.get("translations") or [] + return jsonify({ + "target": target, + "detected_lang": res.get("detected_lang"), + "detected_confidence": res.get("detected_confidence"), + "engine": res.get("engine"), + "engine_version": res.get("engine_version"), + "translated": translations[0] if translations else None, + }) + + @settings_bp.route("/settings/translation/run", methods=["POST"]) async def translation_run(): - """Enqueue the translate sweep now (the Settings 'Translate now' button).""" + """Enqueue the translate sweep now (the Settings 'Translate now' button). + Runs in drain mode — run-until-done — so one press chases the whole + untranslated backlog to zero rather than a single 300-post chunk.""" async with get_session() as session: cfg = await ImportSettings.load(session) if not cfg.translation_enabled or not (cfg.interpreter_base_url or "").strip(): @@ -375,7 +415,7 @@ async def translation_run(): ), 400 from ..tasks.translation import translate_posts - r = translate_posts.delay() + r = translate_posts.delay(drain=True) return jsonify({"celery_task_id": r.id}), 202 diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 967ece2..f657af3 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -178,9 +178,13 @@ def make_celery() -> Celery: "task": "backend.app.tasks.ml.prune_presentation_reviews", "schedule": 86400.0, # retention: drop resolved review flags >30d }, - "translate-posts-daily": { + "translate-posts-8h": { "task": "backend.app.tasks.translation.translate_posts", - "schedule": 86400.0, # no-op unless translation configured + healthy + "schedule": 28800.0, # every 8h: steady-state cadence for the + # trickle of newly-imported posts (no-op unless translation + # configured + healthy). One bounded 300-chunk per fire — the + # one-time backlog drains via the "Translate now" button + # (drain=True, run-until-done), not this sweep. }, "snapshot-head-metrics-daily": { "task": "backend.app.tasks.maintenance.snapshot_head_metrics", diff --git a/backend/app/services/interpreter_client.py b/backend/app/services/interpreter_client.py index c4db1a3..696460f 100644 --- a/backend/app/services/interpreter_client.py +++ b/backend/app/services/interpreter_client.py @@ -96,8 +96,9 @@ def translate( ) -> dict: """Translate a batch. Returns:: - {"translations": [str, ...], # SAME order & length as `texts` - "detected_lang": str | None, # aggregate: describes the FIRST item + {"translations": [str, ...], # SAME order & length as `texts` + "detected_lang": str | None, # aggregate: describes the FIRST item + "detected_confidence": float | None, # detector's confidence, if given "engine": str | None, "engine_version": str | None} @@ -110,6 +111,7 @@ def translate( """ if not texts: return {"translations": [], "detected_lang": None, + "detected_confidence": None, "engine": None, "engine_version": None} try: r = session.post( @@ -140,9 +142,11 @@ def translate( if not isinstance(out, list): out = [out] interp = body.get("interpreter") or {} + detected = body.get("detectedLanguage") or {} return { "translations": out, - "detected_lang": (body.get("detectedLanguage") or {}).get("language"), + "detected_lang": detected.get("language"), + "detected_confidence": detected.get("confidence"), "engine": interp.get("engine"), "engine_version": interp.get("engine_version"), } diff --git a/backend/app/tasks/translation.py b/backend/app/tasks/translation.py index 1ed6b1a..e287d86 100644 --- a/backend/app/tasks/translation.py +++ b/backend/app/tasks/translation.py @@ -7,8 +7,10 @@ same pattern as the other backfill sweeps. No-op unless translation is enabled, a base URL is set, and the service is healthy. Two entry points: -- ``translate_posts`` — the daily untranslated sweep (translated_source_lang IS - NULL only); also the Settings "Translate now" button. +- ``translate_posts`` — the periodic untranslated sweep (every 8h; + translated_source_lang IS NULL only), one bounded chunk per fire. The Settings + "Translate now" button calls it with ``drain=True`` to chase the tail + run-until-done and clear the whole backlog in a single press. - ``retranslate_posts`` — after a model change, resets the stored translation columns for a scoped set of posts (all, or a given set of artists) so the untranslated selection re-runs them, then chases the tail until drained @@ -56,13 +58,19 @@ def _interrupt_backoff(retry_after) -> int: name="backend.app.tasks.translation.translate_posts", soft_time_limit=1800, time_limit=2100, ) -def translate_posts() -> str: +def translate_posts(drain: bool = False) -> str: """Translate untranslated non-English post title/description (default target en) via Interpreter. Per-post [title, description] batch → per-post detected language. Passthrough / already-target posts are marked handled (source == target) with no stored translation. 503 or a connection error interrupts the run (retry next cycle); 400 stops it (fix config); posts done so far stay - committed. No-op unless configured + healthy. Returns a summary string.""" + committed. No-op unless configured + healthy. Returns a summary string. + + ``drain`` (the Settings "Translate now" button) chases the tail + run-until-done: on a clean chunk with work still remaining it re-enqueues + itself until the untranslated backlog is zero, like ``retranslate_posts``. The + periodic beat leaves it False → one bounded chunk per fire (the 8h cadence + drives the rest, enough for the trickle of newly-imported posts).""" SessionLocal = _sync_session_factory() with SessionLocal() as session: ready = _translation_config(session) @@ -73,14 +81,28 @@ def translate_posts() -> str: status, translated, retry_after = _translate_batch( session, posts, base_url, target, ) - # Steady-state daily sweep: one chunk per run on success (the beat drives - # the rest — no tail-chase). But if Interpreter drained mid-chunk, don't - # make a manual "Translate now" wait a whole day — re-enqueue after its - # Retry-After hint (or the default backoff). Self-terminating: once the - # service is down the health gate returns "interpreter unavailable" early. - if status == "interrupted" and _count_untranslated(session, None): + # Drain mode (manual "Translate now"): chase the tail until the backlog is + # zero, so one press clears the whole pile instead of one 300-chunk. + # Termination is guaranteed — every handled post leaves the untranslated + # set, so the remaining count strictly decreases each chunk. + if status == "ok" and drain: + remaining = _count_untranslated(session, None) + if remaining: + translate_posts.apply_async( + (), {"drain": True}, countdown=_RETRANSLATE_COUNTDOWN, + ) + log.info( + "translate_posts: draining — %d remaining → re-enqueued", + remaining, + ) + # Interpreter drained mid-chunk (graceful restart): don't make a manual + # "Translate now" wait a whole beat cycle — re-enqueue after its + # Retry-After hint (or the default backoff), preserving `drain` so a bulk + # drain resumes cleanly. Self-terminating: once the service is down the + # health gate returns "interpreter unavailable" early. + elif status == "interrupted" and _count_untranslated(session, None): delay = _interrupt_backoff(retry_after) - translate_posts.apply_async((), {}, countdown=delay) + translate_posts.apply_async((), {"drain": drain}, countdown=delay) log.info( "translate_posts: interrupted (service draining) → retry in %ss", delay, diff --git a/frontend/src/components/settings/TranslationCard.vue b/frontend/src/components/settings/TranslationCard.vue index 1e5db35..501e47a 100644 --- a/frontend/src/components/settings/TranslationCard.vue +++ b/frontend/src/components/settings/TranslationCard.vue @@ -78,6 +78,61 @@ artist, use the Re-translate button on that artist's page.

+ + + +
+ Test translation +

+ Paste a title or snippet to see what Interpreter detects and returns — + handy for checking why a short string is mis-detected. Nothing is saved. +

+
+ +
+ Test translation +
+ +
+ + Detected: + {{ probeResult.detected_lang || '—' }} + + + Confidence: + {{ fmtConf(probeResult.detected_confidence) }} + + + Engine: + {{ probeResult.engine || '—' }} + + Target: {{ probeResult.target }} +
+ +
+ Result: +
+

{{ + probeResult.translated || '(no change)' + }}

+ + { if (!enabled.value || !baseUrl.value) return 'grey' return status.value?.healthy ? 'success' : 'error' @@ -170,6 +236,21 @@ async function onTest() { } } +async function onProbe () { + probing.value = true + err.value = null + probeResult.value = null + try { + probeResult.value = await api.post('/api/settings/translation/probe', { + body: { text: probeText.value.trim() }, + }) + } catch (e) { + err.value = e.message + } finally { + probing.value = false + } +} + onMounted(async () => { try { await store.loadSettings() diff --git a/tests/test_api_settings.py b/tests/test_api_settings.py index f2800cf..d2522de 100644 --- a/tests/test_api_settings.py +++ b/tests/test_api_settings.py @@ -113,9 +113,10 @@ async def test_translation_test_connection(client, monkeypatch): @pytest.mark.asyncio async def test_translation_run_enqueues_when_configured(client, monkeypatch): + sink = {} monkeypatch.setattr( "backend.app.tasks.translation.translate_posts.delay", - lambda *a, **k: type("R", (), {"id": "t1"})(), + lambda *a, **k: sink.update(k) or type("R", (), {"id": "t1"})(), ) await client.patch("/api/settings/import", json={ "translation_enabled": True, "interpreter_base_url": "http://i.lan", @@ -123,6 +124,46 @@ async def test_translation_run_enqueues_when_configured(client, monkeypatch): resp = await client.post("/api/settings/translation/run") assert resp.status_code == 202 assert (await resp.get_json())["celery_task_id"] == "t1" + assert sink["drain"] is True # "Translate now" chases the whole backlog + + +@pytest.mark.asyncio +async def test_translation_probe_requires_text(client): + resp = await client.post( + "/api/settings/translation/probe", json={"text": " "}) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_translation_probe_requires_base_url(client): + # Text given but no Interpreter URL saved → nothing to probe against. + resp = await client.post( + "/api/settings/translation/probe", json={"text": "hello"}) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_translation_probe_returns_detection(client, monkeypatch): + # The diagnostic surfaces detected language + confidence + result, unsaved. + monkeypatch.setattr( + "backend.app.api.settings.ic.translate", + lambda texts, **k: { + "translations": ["Work in Progress"], + "detected_lang": "de", "detected_confidence": 71.5, + "engine": "llm", "engine_version": "v1", + }, + ) + await client.patch("/api/settings/import", json={ + "translation_enabled": True, "interpreter_base_url": "http://i.lan", + }) + resp = await client.post( + "/api/settings/translation/probe", json={"text": "WIP"}) + assert resp.status_code == 200 + body = await resp.get_json() + assert body["detected_lang"] == "de" + assert body["detected_confidence"] == 71.5 + assert body["translated"] == "Work in Progress" + assert body["target"] == "en" def _fake_retranslate(monkeypatch, sink): diff --git a/tests/test_interpreter_client.py b/tests/test_interpreter_client.py index 0c4adbc..453338f 100644 --- a/tests/test_interpreter_client.py +++ b/tests/test_interpreter_client.py @@ -34,6 +34,7 @@ def test_translate_maps_batch_in_order(monkeypatch): out = ic.translate(["ねこが可愛い", "金髪ギャル！"], base_url="http://i.lan") assert out["translations"] == ["The cat is cute", "Blonde gal!"] assert out["detected_lang"] == "ja" + assert out["detected_confidence"] == 0.98 # surfaced for the detection guard assert out["engine_version"] == "ollama:x:12b" assert captured["json"]["q"] == ["ねこが可愛い", "金髪ギャル！"] assert captured["json"]["target"] == "en" diff --git a/tests/test_translate_posts.py b/tests/test_translate_posts.py index 14d9605..8358303 100644 --- a/tests/test_translate_posts.py +++ b/tests/test_translate_posts.py @@ -5,7 +5,11 @@ from sqlalchemy import select from backend.app.models import Artist, ImportSettings, Post from backend.app.services import interpreter_client as ic -from backend.app.tasks.translation import retranslate_posts, translate_posts +from backend.app.tasks.translation import ( + _RETRANSLATE_COUNTDOWN, + retranslate_posts, + translate_posts, +) pytestmark = pytest.mark.integration @@ -191,6 +195,63 @@ def test_translate_posts_interrupt_reenqueues_after_backoff(db_sync, monkeypatch assert calls[0][1]["countdown"] == 30 +def _mock_ok(monkeypatch): + monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True) + monkeypatch.setattr( + "backend.app.tasks.translation.ic.translate", + lambda texts, **k: { + "translations": [f"EN:{t}" for t in texts], + "detected_lang": "ja", "engine": "llm", "engine_version": "v1", + }, + ) + + +def test_translate_posts_drain_chases_tail(db_sync, monkeypatch): + # "Translate now" (drain=True): with the chunk capped at 1 and 2 untranslated + # posts, the first chunk re-enqueues itself — drain preserved — to finish the + # backlog rather than waiting for the next beat. + _patch(monkeypatch, db_sync) + _mock_ok(monkeypatch) + monkeypatch.setattr("backend.app.tasks.translation._MAX_POSTS_PER_RUN", 1) + calls = [] + monkeypatch.setattr( + "backend.app.tasks.translation.translate_posts.apply_async", + lambda *a, **k: calls.append((a, k)), + ) + a = _artist(db_sync) + _post(db_sync, a.id, title="ねこ", ext="p1") + _post(db_sync, a.id, title="いぬ", ext="p2") + _enable(db_sync) + db_sync.commit() + + translate_posts(drain=True) + assert len(calls) == 1 # one re-enqueue for the tail + args, kw = calls[0] + assert args[1] == {"drain": True} # drain flag carried forward + assert kw["countdown"] == _RETRANSLATE_COUNTDOWN + + +def test_translate_posts_beat_single_chunk_no_tail_chase(db_sync, monkeypatch): + # The periodic beat (drain defaults False) does exactly ONE chunk and never + # re-enqueues — the 8h cadence drives the rest. + _patch(monkeypatch, db_sync) + _mock_ok(monkeypatch) + monkeypatch.setattr("backend.app.tasks.translation._MAX_POSTS_PER_RUN", 1) + calls = [] + monkeypatch.setattr( + "backend.app.tasks.translation.translate_posts.apply_async", + lambda *a, **k: calls.append((a, k)), + ) + a = _artist(db_sync) + _post(db_sync, a.id, title="ねこ", ext="p1") + _post(db_sync, a.id, title="いぬ", ext="p2") + _enable(db_sync) + db_sync.commit() + + translate_posts() # drain=False + assert calls == [] # no self-re-enqueue + + def _mock_new_model(monkeypatch, *, ver="v2"): """Interpreter is up and translates with a NEW engine version.""" monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)