FabledCurator/backend/app/tasks/translation.py

"""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