feat(translation): backfill sweep + beat + manual trigger (#143 step 3)
tasks/translation.py — translate_posts: picks untranslated posts (title OR description non-empty), per-post [title, description] batch via the Interpreter client, stores translations + detected lang + engine_version; passthrough / already-target posts are marked handled with no stored translation. 503 or a connection error interrupts (retry next cycle), 400 stops (fix config), per-post commit keeps progress; wall-clock bounded. Wired into celery (maintenance_long lane) + a daily beat. No-op unless enabled + base URL set + healthy. GET /settings/translation/status + POST .../run for the Settings card. Task tests (stubbed client, monkeypatched session). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM
This commit is contained in:
@@ -1,12 +1,23 @@
|
|||||||
"""Settings API: import filters, system stats."""
|
"""Settings API: import filters, system stats."""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
import secrets
|
import secrets
|
||||||
|
|
||||||
from quart import Blueprint, jsonify, request
|
from quart import Blueprint, jsonify, request
|
||||||
from sqlalchemy import func, select
|
from sqlalchemy import func, or_, select
|
||||||
|
|
||||||
from ..extensions import get_session
|
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")
|
settings_bp = Blueprint("settings", __name__, url_prefix="/api")
|
||||||
|
|
||||||
@@ -285,3 +296,45 @@ async def rotate_extension_api_key():
|
|||||||
row.value = new_value
|
row.value = new_value
|
||||||
await session.commit()
|
await session.commit()
|
||||||
return jsonify({"key": new_value})
|
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
|
||||||
|
|||||||
@@ -35,6 +35,7 @@ def make_celery() -> Celery:
|
|||||||
"backend.app.tasks.backup",
|
"backend.app.tasks.backup",
|
||||||
"backend.app.tasks.admin",
|
"backend.app.tasks.admin",
|
||||||
"backend.app.tasks.library_audit",
|
"backend.app.tasks.library_audit",
|
||||||
|
"backend.app.tasks.translation",
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
app.conf.update(
|
app.conf.update(
|
||||||
@@ -63,6 +64,9 @@ def make_celery() -> Celery:
|
|||||||
"backend.app.tasks.backup.*": {"queue": "maintenance_long"},
|
"backend.app.tasks.backup.*": {"queue": "maintenance_long"},
|
||||||
"backend.app.tasks.admin.*": {"queue": "maintenance_long"},
|
"backend.app.tasks.admin.*": {"queue": "maintenance_long"},
|
||||||
"backend.app.tasks.library_audit.*": {"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.
|
# Heavy ML tasks need fair dispatch — see ImageRepo's precedent.
|
||||||
task_acks_late=True,
|
task_acks_late=True,
|
||||||
@@ -161,6 +165,10 @@ def make_celery() -> Celery:
|
|||||||
"task": "backend.app.tasks.ml.prune_presentation_reviews",
|
"task": "backend.app.tasks.ml.prune_presentation_reviews",
|
||||||
"schedule": 86400.0, # retention: drop resolved review flags >30d
|
"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": {
|
"snapshot-head-metrics-daily": {
|
||||||
"task": "backend.app.tasks.maintenance.snapshot_head_metrics",
|
"task": "backend.app.tasks.maintenance.snapshot_head_metrics",
|
||||||
"schedule": 86400.0,
|
"schedule": 86400.0,
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -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
|
||||||
Reference in New Issue
Block a user