"""Post-text translation Celery tasks (milestone 143 + re-translate m146). Backfills non-English post title/description to the target language via the self-hosted Interpreter LAN service, storing the result + engine_version so viewing is instant and re-runs are cache-fast. Sync (Celery workers are sync), same pattern as the other backfill sweeps. No-op unless translation is enabled, a base URL is set, and the service is healthy. Curator does NO language detection of its own: each field's translation is accepted or rejected purely on Interpreter's reported engine + detected language + confidence (see ``_accept``) — a short English title Interpreter mis-labels as a European language scores below the latin-script floor and is kept as-is. Two entry points: - ``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 (run-until-done). The Interpreter cache keys on engine_version: a changed model genuinely re-translates, an unchanged one is ~1ms cache-fast. """ import logging from datetime import UTC, datetime from celery.exceptions import SoftTimeLimitExceeded from sqlalchemy import func, or_, select, update from ..celery_app import celery from ..models import ImportSettings, Post from ..services import interpreter_client as ic from ..utils.text import html_to_plain from ._sync_engine import sync_session_factory as _sync_session_factory log = logging.getLogger(__name__) # Bound one run's work so it commits progress rather than dying at the time # limit; the rest resumes next cycle (idempotent — only untranslated posts are # picked, so an interrupted run just re-runs = rule-89 recovery). _MAX_POSTS_PER_RUN = 300 # Short gap between run-until-done chunks so a big re-translate finishes on its # own without monopolising the maintenance_long worker in one uninterrupted run. _RETRANSLATE_COUNTDOWN = 5 # When Interpreter drains mid-chunk (graceful restart), resume the bulk # re-translate after the service's Retry-After hint if it sent one, else this # default; capped so a bogus/huge hint can't park the work longer than the daily # beat's own safety net would. _INTERRUPT_BACKOFF = 60 _INTERRUPT_BACKOFF_MAX = 900 # --- Translation acceptance gate (Scribe rule 133; contract note #1347) ------- # Curator does NO language detection of its own — it accepts or rejects a # translation purely on Interpreter's reported detection. CJK is script-detected # (kana/hangul/kanji) and reliably high-confidence, so ja/ko/zh are trusted # outright (pure-kanji Japanese can even come back labelled zh ~0.75, still # correctly translated). Latin-script detection is a real statistical # probability, so a short English title Interpreter mis-labels as a European # language scores low; require it to clear this floor, else keep the original. # Calibrated 2026-07-09 against fresh (uncached) probes once Interpreter returned # real langdetect confidence: genuine German detected at 1.0, while a correctly- # detected but ambiguous latin string dipped to 0.86 — so the floor sits at 0.80, # below that band, to keep legitimate ambiguous non-English while still rejecting # genuinely-unsure guesses. Real langdetect also fixed the original mis-flag at # the source (short English titles now detect as English → passthrough), so this # floor is a safety net, not the primary fix. Re-tune via the "Test translation" # box (send fresh text — cache hits report 1.0). _CJK_LANGS = frozenset({"ja", "ko", "zh"}) _MIN_LATIN_CONFIDENCE = 0.80 def _interrupt_backoff(retry_after) -> int: """Seconds to wait before resuming after an Interpreter interruption: the server's Retry-After hint (capped), else the default.""" if retry_after and retry_after > 0: return int(min(retry_after, _INTERRUPT_BACKOFF_MAX)) return _INTERRUPT_BACKOFF @celery.task( name="backend.app.tasks.translation.translate_posts", soft_time_limit=1800, time_limit=2100, ) 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. ``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) if isinstance(ready, str): return ready base_url, target = ready posts = _select_untranslated(session, None, _MAX_POSTS_PER_RUN) status, translated, retry_after = _translate_batch( session, posts, base_url, target, ) # 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((), {"drain": drain}, countdown=delay) log.info( "translate_posts: interrupted (service draining) → retry in %ss", delay, ) return _summary(status, translated, len(posts)) @celery.task( name="backend.app.tasks.translation.retranslate_posts", soft_time_limit=1800, time_limit=2100, ) def retranslate_posts(artist_ids=None, _reset_done=False) -> str: """Re-translate posts after a model change. On the first call resets the stored translation columns to NULL for the scoped posts — all artists when ``artist_ids`` is falsy, else ``WHERE artist_id IN artist_ids`` — so the normal untranslated selection re-runs them through Interpreter. Then handles one bounded chunk and re-enqueues itself until the scoped backlog is drained (run-until-done; ``_reset_done`` guards against wiping fresh work on the follow-up chunks). No reset happens unless the service is configured + healthy, so translations are never cleared when they can't be rebuilt.""" SessionLocal = _sync_session_factory() with SessionLocal() as session: ready = _translation_config(session) if isinstance(ready, str): return ready base_url, target = ready if not _reset_done: n = _reset_translations(session, artist_ids) log.info( "retranslate_posts: reset %d post(s) (artist_ids=%s)", n, artist_ids, ) posts = _select_untranslated(session, artist_ids, _MAX_POSTS_PER_RUN) status, translated, retry_after = _translate_batch( session, posts, base_url, target, ) # run-until-done: chase the tail when the chunk finished cleanly. # Termination is guaranteed: every handled post leaves the NULL set, so # the remaining count strictly decreases each chunk. if status == "ok": remaining = _count_untranslated(session, artist_ids) if remaining: retranslate_posts.apply_async( (), {"artist_ids": artist_ids, "_reset_done": True}, countdown=_RETRANSLATE_COUNTDOWN, ) log.info( "retranslate_posts: %d remaining → re-enqueued", remaining, ) # Interpreter drained mid-chunk (graceful restart / 429 / 5xx): don't # stall the bulk re-translate until the daily beat — resume it after the # service's Retry-After hint (or the default backoff), with # _reset_done=True so the fresh work is never re-wiped. Self-terminating: # if the service is still down next run, _translation_config's health gate # returns "interpreter unavailable" early and nothing re-enqueues. elif status == "interrupted" and _count_untranslated(session, artist_ids): delay = _interrupt_backoff(retry_after) retranslate_posts.apply_async( (), {"artist_ids": artist_ids, "_reset_done": True}, countdown=delay, ) log.info( "retranslate_posts: interrupted (service draining) → retry in %ss", delay, ) return _summary(status, translated, len(posts)) def _translation_config(session): """Resolve (base_url, target) if translation is enabled + healthy, else a short status string ("disabled" / "interpreter unavailable") for the caller to return. Centralises the guard so translate/retranslate agree exactly.""" cfg = ImportSettings.load_sync(session) if not cfg.translation_enabled or not cfg.interpreter_base_url.strip(): return "disabled" base_url = cfg.interpreter_base_url.strip() target = (cfg.translation_target_lang or "en").strip() or "en" if not ic.health(base_url): return "interpreter unavailable" return base_url, target def _untranslated_filter(stmt, artist_ids): """Add the untranslated-post predicate (+ optional artist scope) to a select/count over Post. Untranslated = translated_source_lang IS NULL with some text to translate.""" stmt = stmt.where( Post.translated_source_lang.is_(None) ).where(or_( Post.post_title.is_not(None), Post.description.is_not(None), )) if artist_ids: stmt = stmt.where(Post.artist_id.in_(artist_ids)) return stmt def _select_untranslated(session, artist_ids, limit): return session.execute( _untranslated_filter(select(Post), artist_ids).limit(limit) ).scalars().all() def _count_untranslated(session, artist_ids) -> int: return int(session.execute( _untranslated_filter(select(func.count(Post.id)), artist_ids) ).scalar_one()) def _reset_translations(session, artist_ids) -> int: """Clear the stored translation columns for scoped, already-handled posts so the untranslated sweep re-runs them. Only touches rows that were translated (translated_source_lang IS NOT NULL) — untranslated rows are already NULL. Returns the row count reset (commits it).""" stmt = ( update(Post) .where(Post.translated_source_lang.is_not(None)) .values( post_title_translated=None, description_translated=None, translated_source_lang=None, translation_engine_version=None, translated_at=None, ) ) if artist_ids: stmt = stmt.where(Post.artist_id.in_(artist_ids)) n = session.execute(stmt).rowcount session.commit() return n def _translate_batch(session, posts, base_url: str, target: str): """Translate each post in the chunk with a per-post commit (an interrupted run keeps progress). Returns (status, translated, retry_after) where status is one of "ok" / "interrupted" (unavailable/drain) / "stopped" (400) / "timeout"; retry_after is the server's Retry-After hint in seconds on an interrupt, else None.""" translated = 0 for post in posts: try: translated += _translate_one(session, post, base_url, target) session.commit() except ic.InterpreterUnavailable as e: session.rollback() return "interrupted", translated, getattr(e, "retry_after", None) except ic.InterpreterBadRequest as e: session.rollback() log.warning("translate bad request: %s", e) return "stopped", translated, None except SoftTimeLimitExceeded: return "timeout", translated, None return "ok", translated, None def _summary(status: str, translated: int, scanned: int) -> str: if status == "interrupted": return f"interrupted (service down) — translated={translated}" if status == "stopped": return f"stopped (bad request) — translated={translated}" if status == "timeout": return f"time limit — translated={translated}" return f"translated={translated} scanned={scanned}" def _accept(language: str, confidence) -> bool: """Should curator store Interpreter's translation, or keep the original? Consumes ONLY Interpreter's own detection (curator does none, Scribe rule 133): CJK languages are trusted regardless of the reported number; a latin-script detection must clear ``_MIN_LATIN_CONFIDENCE``. A missing confidence fails OPEN (accept), so a service that omits the field never causes translations to be silently dropped.""" if (language or "").split("-", 1)[0].lower() in _CJK_LANGS: return True if confidence is None: return True return confidence >= _MIN_LATIN_CONFIDENCE def _translate_field(text: str, base_url: str, target: str): """Translate ONE field independently. Returns (translated, source_lang, engine_version), or (None, None, None) when there's nothing to store — empty text, already the target language, a passthrough (engine "none"), or a latin-script detection the acceptance gate rejects as too low-confidence (kept as the original). Per-field (not aggregate first-item) detection so a non-English description still gets translated when the title is already English (mixed-language posts).""" if not text: return None, None, None res = ic.translate([text], base_url=base_url, target=target) detected = res["detected_lang"] or target if detected == target or res["engine"] == "none": return None, None, None # Trust only Interpreter's own detection (Scribe rule 133): keep the original # when a latin-script detection doesn't clear the confidence floor, so a short # English title mis-labelled as e.g. German isn't rewritten into the archive. if not _accept(detected, res.get("detected_confidence")): return None, None, None return res["translations"][0], detected, res["engine_version"] def _translate_one(session, post, base_url: str, target: str) -> int: """Translate a post's title/description in place, each field independently. Returns 1 if it stored at least one translation, 0 when the post is all passthrough/empty (still marks it handled so the sweep won't revisit it).""" title = (post.post_title or "").strip() desc = (html_to_plain(post.description) if post.description else "") or "" desc = desc.strip() title_tr, title_lang, title_ev = _translate_field(title, base_url, target) desc_tr, desc_lang, desc_ev = _translate_field(desc, base_url, target) if title_tr is None and desc_tr is None: # Nothing non-target (or nothing to translate) → handled, store nothing. post.translated_source_lang = target return 0 post.post_title_translated = title_tr post.description_translated = desc_tr # Source lang = whichever field was actually non-target (for a mixed post, # the translated field's language, not the already-English one). post.translated_source_lang = title_lang or desc_lang post.translation_engine_version = title_ev or desc_ev post.translated_at = datetime.now(UTC) return 1