diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index e415e36..f888430 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -1,12 +1,23 @@ """Settings API: import filters, system stats.""" +import asyncio import secrets from quart import Blueprint, jsonify, request -from sqlalchemy import func, select +from sqlalchemy import func, or_, select from ..extensions import get_session -from ..models import AppSetting, Artist, ImageRecord, ImportBatch, ImportSettings, ImportTask, Tag +from ..models import ( + AppSetting, + Artist, + ImageRecord, + ImportBatch, + ImportSettings, + ImportTask, + Post, + Tag, +) +from ..services import interpreter_client as ic settings_bp = Blueprint("settings", __name__, url_prefix="/api") @@ -285,3 +296,45 @@ async def rotate_extension_api_key(): row.value = new_value await session.commit() return jsonify({"key": new_value}) + + +# --- Translation (#143): live status + manual "Translate now" -------------- + + +@settings_bp.route("/settings/translation/status", methods=["GET"]) +async def translation_status(): + """For the Settings card: is it on, is a URL set, is the service reachable, + and how many posts still await translation. Health runs the sync client in a + thread so the event loop isn't blocked.""" + async with get_session() as session: + cfg = await ImportSettings.load(session) + untranslated = (await session.execute( + select(func.count(Post.id)) + .where(Post.translated_source_lang.is_(None)) + .where(or_( + Post.post_title.is_not(None), Post.description.is_not(None), + )) + )).scalar_one() + base_url = (cfg.interpreter_base_url or "").strip() + healthy = await asyncio.to_thread(ic.health, base_url) if base_url else False + return jsonify({ + "enabled": cfg.translation_enabled, + "base_url_set": bool(base_url), + "healthy": healthy, + "untranslated_count": int(untranslated), + }) + + +@settings_bp.route("/settings/translation/run", methods=["POST"]) +async def translation_run(): + """Enqueue the translate sweep now (the Settings 'Translate now' button).""" + async with get_session() as session: + cfg = await ImportSettings.load(session) + if not cfg.translation_enabled or not (cfg.interpreter_base_url or "").strip(): + return jsonify( + {"error": "translation is disabled or no base URL is set"} + ), 400 + from ..tasks.translation import translate_posts + + r = translate_posts.delay() + return jsonify({"celery_task_id": r.id}), 202 diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index c8cd723..0e47473 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -35,6 +35,7 @@ def make_celery() -> Celery: "backend.app.tasks.backup", "backend.app.tasks.admin", "backend.app.tasks.library_audit", + "backend.app.tasks.translation", ], ) app.conf.update( @@ -63,6 +64,9 @@ def make_celery() -> Celery: "backend.app.tasks.backup.*": {"queue": "maintenance_long"}, "backend.app.tasks.admin.*": {"queue": "maintenance_long"}, "backend.app.tasks.library_audit.*": {"queue": "maintenance_long"}, + # Translation backfill hits the LLM (~1–6s/item) → the long lane so it + # never starves the quick self-healing sweeps (#143). + "backend.app.tasks.translation.*": {"queue": "maintenance_long"}, }, # Heavy ML tasks need fair dispatch — see ImageRepo's precedent. task_acks_late=True, @@ -161,6 +165,10 @@ def make_celery() -> Celery: "task": "backend.app.tasks.ml.prune_presentation_reviews", "schedule": 86400.0, # retention: drop resolved review flags >30d }, + "translate-posts-daily": { + "task": "backend.app.tasks.translation.translate_posts", + "schedule": 86400.0, # no-op unless translation configured + healthy + }, "snapshot-head-metrics-daily": { "task": "backend.app.tasks.maintenance.snapshot_head_metrics", "schedule": 86400.0, diff --git a/backend/app/tasks/translation.py b/backend/app/tasks/translation.py new file mode 100644 index 0000000..89c7b77 --- /dev/null +++ b/backend/app/tasks/translation.py @@ -0,0 +1,101 @@ +"""Post-text translation Celery task (milestone 143). + +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. +""" +import logging +from datetime import UTC, datetime + +from celery.exceptions import SoftTimeLimitExceeded +from sqlalchemy import or_, select + +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 + + +@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: + 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" + + posts = session.execute( + select(Post) + .where(Post.translated_source_lang.is_(None)) + .where(or_( + Post.post_title.is_not(None), Post.description.is_not(None), + )) + .limit(_MAX_POSTS_PER_RUN) + ).scalars().all() + + translated = 0 + try: + for post in posts: + translated += _translate_one(session, post, base_url, target) + session.commit() # per-post → an interrupted run keeps progress + except ic.InterpreterUnavailable: + session.rollback() + return f"interrupted (service down) — translated={translated}" + except ic.InterpreterBadRequest as e: + session.rollback() + log.warning("translate_posts bad request: %s", e) + return f"stopped (bad request) — translated={translated}" + except SoftTimeLimitExceeded: + return f"time limit — translated={translated}" + return f"translated={translated} scanned={len(posts)}" + + +def _translate_one(session, post, base_url: str, target: str) -> int: + """Translate one post's title/description in place. Returns 1 if it stored a + translation, 0 for passthrough/empty (which still marks the post handled).""" + title = (post.post_title or "").strip() + desc = (html_to_plain(post.description) if post.description else "") or "" + desc = desc.strip() + fields: list[tuple[str, str]] = [] + if title: + fields.append(("title", title)) + if desc: + fields.append(("desc", desc)) + if not fields: + post.translated_source_lang = target # nothing to translate → handled + return 0 + res = ic.translate([t for _, t in fields], base_url=base_url, target=target) + detected = res["detected_lang"] or target + if detected == target or res["engine"] == "none": + post.translated_source_lang = target # already target language → handled + return 0 + by_field = {f: res["translations"][i] for i, (f, _) in enumerate(fields)} + post.post_title_translated = by_field.get("title") + post.description_translated = by_field.get("desc") + post.translated_source_lang = detected + post.translation_engine_version = res["engine_version"] + post.translated_at = datetime.now(UTC) + return 1 diff --git a/tests/test_translate_posts.py b/tests/test_translate_posts.py new file mode 100644 index 0000000..c59580f --- /dev/null +++ b/tests/test_translate_posts.py @@ -0,0 +1,127 @@ +"""translate_posts sweep (#143) — integration, with a stubbed Interpreter client +and the task's session factory pointed at the test's db_sync.""" +import pytest +from sqlalchemy import select + +from backend.app.models import Artist, ImportSettings, Post +from backend.app.tasks.translation import translate_posts + +pytestmark = pytest.mark.integration + + +class _Ctx: + def __init__(self, s): + self.s = s + + def __enter__(self): + return self.s + + def __exit__(self, *a): + return False + + +def _sf(db_sync): + class _SM: + def __call__(self): + return _Ctx(db_sync) + + return _SM() + + +def _artist(db): + a = Artist(name="A", slug="a") + db.add(a) + db.flush() + return a + + +def _post(db, artist_id, *, title=None, desc=None, ext="p1"): + p = Post( + artist_id=artist_id, external_post_id=ext, + post_title=title, description=desc, + ) + db.add(p) + db.flush() + return p + + +def _enable(db, url="http://i.lan"): + cfg = db.execute( + select(ImportSettings).where(ImportSettings.id == 1) + ).scalar_one() + cfg.translation_enabled = True + cfg.interpreter_base_url = url + db.flush() + + +def _patch(monkeypatch, db_sync): + monkeypatch.setattr( + "backend.app.tasks.translation._sync_session_factory", + lambda: _sf(db_sync), + ) + + +def test_translate_posts_stores_translation(db_sync, monkeypatch): + _patch(monkeypatch, db_sync) + 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", + }, + ) + a = _artist(db_sync) + p = _post(db_sync, a.id, title="ねこ", desc="かわいい") + _enable(db_sync) + db_sync.commit() + + assert "translated=1" in translate_posts() + db_sync.refresh(p) + assert p.post_title_translated == "EN:ねこ" + assert p.description_translated == "EN:かわいい" + assert p.translated_source_lang == "ja" + assert p.translation_engine_version == "v1" + + +def test_translate_posts_passthrough_english_marks_handled(db_sync, monkeypatch): + _patch(monkeypatch, db_sync) + monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True) + monkeypatch.setattr( + "backend.app.tasks.translation.ic.translate", + lambda texts, **k: { + "translations": list(texts), "detected_lang": "en", + "engine": "none", "engine_version": None, + }, + ) + a = _artist(db_sync) + p = _post(db_sync, a.id, title="hello world", ext="p2") + _enable(db_sync) + db_sync.commit() + + translate_posts() + db_sync.refresh(p) + assert p.translated_source_lang == "en" # marked handled... + assert p.post_title_translated is None # ...but nothing to show + + +def test_translate_posts_disabled_is_noop(db_sync, monkeypatch): + _patch(monkeypatch, db_sync) + a = _artist(db_sync) + p = _post(db_sync, a.id, title="ねこ", ext="p3") + db_sync.commit() # translation_enabled defaults False + assert translate_posts() == "disabled" + db_sync.refresh(p) + assert p.translated_source_lang is None + + +def test_translate_posts_service_down_leaves_untranslated(db_sync, monkeypatch): + _patch(monkeypatch, db_sync) + monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: False) + a = _artist(db_sync) + p = _post(db_sync, a.id, title="ねこ", ext="p4") + _enable(db_sync) + db_sync.commit() + assert translate_posts() == "interpreter unavailable" + db_sync.refresh(p) + assert p.translated_source_lang is None # will retry next run