"""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. Two entry points: - ``translate_posts`` — the daily untranslated sweep (translated_source_lang IS NULL only); also the Settings "Translate now" button. - ``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 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() -> 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.""" 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, ) # 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): delay = _interrupt_backoff(retry_after) translate_posts.apply_async((), {}, 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 _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, or a passthrough (engine "none"). 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 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