Re-translate on Interpreter model change — artist-scoped + global (m146) #202

Merged
bvandeusen merged 1 commits from dev into main 2026-07-07 20:22:40 -04:00
6 changed files with 494 additions and 30 deletions
+36
View File
@@ -351,3 +351,39 @@ async def translation_run():
r = translate_posts.delay()
return jsonify({"celery_task_id": r.id}), 202
@settings_bp.route("/settings/translation/retranslate", methods=["POST"])
async def translation_retranslate():
"""Re-translate stored translations after a model change (m146). Body:
``{"artist_id": <int>}`` aims at one artist; ``{"all": true}`` re-runs every
artist. ``all`` must be explicit so an empty/typo body can't wipe everything.
Clears the scoped translations and enqueues the run-until-done retranslate
sweep (the Interpreter cache re-translates on a changed model, is cache-fast
otherwise). Same enabled + base-URL guard as 'Translate now'."""
body = await request.get_json(silent=True)
body = body if isinstance(body, dict) else {}
artist_id = body.get("artist_id")
do_all = bool(body.get("all"))
if artist_id is None and not do_all:
return jsonify(
{"error": "provide artist_id, or all=true to re-translate everything"}
), 400
if artist_id is not None:
try:
artist_id = int(artist_id)
except (TypeError, ValueError):
return jsonify({"error": "artist_id must be an integer"}), 400
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 retranslate_posts
# artist_id wins when both are sent; otherwise all=true → None (every artist).
artist_ids = [artist_id] if artist_id is not None else None
r = retranslate_posts.delay(artist_ids=artist_ids)
return jsonify({"celery_task_id": r.id}), 202
+156 -26
View File
@@ -1,16 +1,25 @@
"""Post-text translation Celery task (milestone 143).
"""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 or_, select
from sqlalchemy import func, or_, select, update
from ..celery_app import celery
from ..models import ImportSettings, Post
@@ -24,6 +33,9 @@ log = logging.getLogger(__name__)
# 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
@celery.task(
@@ -39,38 +51,156 @@ def translate_posts() -> str:
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"
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 = _translate_batch(session, posts, base_url, target)
return _summary(status, translated, len(posts))
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
@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 = _translate_batch(session, posts, base_url, target)
# run-until-done: only chase the tail when the chunk finished cleanly. A
# service interruption / bad request stops here and leaves the rest NULL
# for the daily beat or a fresh trigger to resume (rule-89 recovery).
# 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,
)
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) where status is one of
"ok" / "interrupted" (503/conn) / "stopped" (400) / "timeout"."""
translated = 0
for post in posts:
try:
for post in posts:
translated += _translate_one(session, post, base_url, target)
session.commit() # per-post → an interrupted run keeps progress
translated += _translate_one(session, post, base_url, target)
session.commit()
except ic.InterpreterUnavailable:
session.rollback()
return f"interrupted (service down) — translated={translated}"
return "interrupted", translated
except ic.InterpreterBadRequest as e:
session.rollback()
log.warning("translate_posts bad request: %s", e)
return f"stopped (bad request) — translated={translated}"
log.warning("translate bad request: %s", e)
return "stopped", translated
except SoftTimeLimitExceeded:
return f"time limit — translated={translated}"
return f"translated={translated} scanned={len(posts)}"
return "timeout", translated
return "ok", translated
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_one(session, post, base_url: str, target: str) -> int:
@@ -101,6 +101,47 @@
</v-card>
</v-dialog>
<section class="fc-artist-mgmt__sec">
<h2 class="fc-h2">Translation</h2>
<p class="fc-muted text-body-2 mb-3">
Re-translate every post by {{ overview.name }} through the current
Interpreter model — clears the stored translations and rebuilds them.
Use this after switching translation models to refresh just this artist.
</p>
<div class="d-flex align-center flex-wrap" style="gap: 10px;">
<v-btn
size="small" variant="tonal" rounded="pill"
prepend-icon="mdi-translate"
:loading="retranslating" :disabled="!translationEnabled"
@click="confirmRetranslate = true"
>Re-translate posts</v-btn>
<span v-if="!translationEnabled" class="fc-muted text-caption">
Translation is off — enable it in Settings → Maintenance.
</span>
</div>
</section>
<!-- m146: per-artist re-translate — clears + rebuilds this artist's stored
translations (used after a translation-model change). -->
<v-dialog v-model="confirmRetranslate" max-width="440">
<v-card>
<v-card-title class="fc-h2">Re-translate {{ overview.name }}?</v-card-title>
<v-card-text class="text-body-2">
Clears the stored translation for every post by
<strong>{{ overview.name }}</strong> and re-runs them through your
current Interpreter model. Unchanged text comes back from the cache
instantly. Runs in the background until done.
</v-card-text>
<v-card-actions>
<v-spacer />
<v-btn variant="text" @click="confirmRetranslate = false">Cancel</v-btn>
<v-btn color="accent" :loading="retranslating" @click="onRetranslate">
Re-translate
</v-btn>
</v-card-actions>
</v-card>
</v-dialog>
<section class="fc-artist-mgmt__sec">
<h2 class="fc-h2">Danger zone</h2>
<ArtistDangerZone
@@ -113,9 +154,11 @@
</template>
<script setup>
import { computed, ref } from 'vue'
import { computed, onMounted, ref } from 'vue'
import { useRouter, RouterLink } from 'vue-router'
import { useApi } from '../../composables/useApi.js'
import { useImportStore } from '../../stores/import.js'
import { useSourcesStore } from '../../stores/sources.js'
import { toast } from '../../utils/toast.js'
import ArtistDangerZone from './ArtistDangerZone.vue'
@@ -125,8 +168,38 @@ const props = defineProps({
})
const router = useRouter()
const api = useApi()
const importStore = useImportStore()
const sources = useSourcesStore()
// m146: per-artist re-translate. Gated on translation being enabled globally
// (the endpoint 400s otherwise) — read the shared import settings once.
const translationEnabled = ref(false)
const retranslating = ref(false)
const confirmRetranslate = ref(false)
onMounted(async () => {
try {
await importStore.loadSettings()
translationEnabled.value = !!importStore.settings?.translation_enabled
} catch { /* non-fatal — the button just stays disabled */ }
})
async function onRetranslate () {
retranslating.value = true
try {
await api.post('/api/settings/translation/retranslate', {
body: { artist_id: props.overview.id },
})
confirmRetranslate.value = false
toast({ text: `Re-translating ${props.overview.name}`, type: 'success' })
} catch (e) {
toast({ text: `Re-translate failed: ${e.message}`, type: 'error' })
} finally {
retranslating.value = false
}
}
// #130: move a source into another artist.
const moveOpen = ref(false)
const moveSource = ref(null)
@@ -47,16 +47,49 @@
prepend-icon="mdi-translate" :loading="running"
:disabled="!enabled || !baseUrl" @click="onRun"
>Translate now</v-btn>
<v-btn
size="small" variant="tonal" rounded="pill"
prepend-icon="mdi-refresh" :loading="retranslating"
:disabled="!enabled || !baseUrl" @click="confirmAll = true"
>Re-translate all</v-btn>
<span v-if="status" class="fc-muted text-caption">
{{ status.untranslated_count }}
post{{ status.untranslated_count === 1 ? '' : 's' }} awaiting translation
</span>
</div>
<p class="fc-muted text-caption mt-2 mb-0">
Use <strong>Re-translate all</strong> after switching the Interpreter model —
it clears every stored translation and re-runs it through the new model
(unchanged text is served from cache, so it's cheap). To refresh just one
artist, use the Re-translate button on that artist's page.
</p>
<v-alert
v-if="err" type="error" variant="tonal" density="compact"
class="mt-3" closable @click:close="err = null"
>{{ err }}</v-alert>
<!-- m146: re-translate-everything is a bigger hammer than 'Translate now'
(it clears + rebuilds all translations), so it asks first. -->
<v-dialog v-model="confirmAll" max-width="440">
<v-card>
<v-card-title>Re-translate everything?</v-card-title>
<v-card-text class="text-body-2">
This clears the stored translation for <strong>every</strong> post and
re-runs them through your current Interpreter model. Text that hasn't
changed comes back from the cache instantly; anything the new model
translates differently is refreshed. It runs in the background until
done.
</v-card-text>
<v-card-actions>
<v-spacer />
<v-btn variant="text" @click="confirmAll = false">Cancel</v-btn>
<v-btn
color="accent" :loading="retranslating" @click="onRetranslateAll"
>Re-translate all</v-btn>
</v-card-actions>
</v-card>
</v-dialog>
</MaintenanceTile>
</template>
@@ -75,6 +108,8 @@ const baseUrl = ref('')
const targetLang = ref('en')
const busy = ref(false)
const running = ref(false)
const retranslating = ref(false)
const confirmAll = ref(false)
const testing = ref(false)
const testResult = ref(null) // null = not tested this session; true/false = last result
const status = ref(null)
@@ -159,4 +194,18 @@ async function onRun() {
setTimeout(loadStatus, 1500) // let the sweep make a dent, then refresh
}
}
async function onRetranslateAll() {
retranslating.value = true
err.value = null
try {
await api.post('/api/settings/translation/retranslate', { body: { all: true } })
confirmAll.value = false
toast({ text: 'Re-translating all posts…', type: 'success' })
} catch (e) {
err.value = e.message
} finally {
retranslating.value = false
setTimeout(loadStatus, 1500) // reset spikes the untranslated count — refresh it
}
}
</script>
+62
View File
@@ -97,6 +97,68 @@ async def test_translation_run_enqueues_when_configured(client, monkeypatch):
assert (await resp.get_json())["celery_task_id"] == "t1"
def _fake_retranslate(monkeypatch, sink):
# Record the kwargs the endpoint enqueues with, return a fake AsyncResult.
monkeypatch.setattr(
"backend.app.tasks.translation.retranslate_posts.delay",
lambda *a, **k: sink.update(k) or type("R", (), {"id": "r1"})(),
)
@pytest.mark.asyncio
async def test_translation_retranslate_requires_config(client):
# Disabled → 400 even with an explicit scope (no reset is enqueued).
resp = await client.post(
"/api/settings/translation/retranslate", json={"all": True})
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_translation_retranslate_requires_target(client, monkeypatch):
# 'all' must be explicit: an empty body is a 400, so a stray call can't
# wipe every translation.
_fake_retranslate(monkeypatch, {})
await client.patch("/api/settings/import", json={
"translation_enabled": True, "interpreter_base_url": "http://i.lan",
})
resp = await client.post("/api/settings/translation/retranslate")
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_translation_retranslate_artist_scope(client, monkeypatch):
sink = {}
_fake_retranslate(monkeypatch, sink)
await client.patch("/api/settings/import", json={
"translation_enabled": True, "interpreter_base_url": "http://i.lan",
})
resp = await client.post(
"/api/settings/translation/retranslate", json={"artist_id": 7})
assert resp.status_code == 202
assert (await resp.get_json())["celery_task_id"] == "r1"
assert sink["artist_ids"] == [7]
@pytest.mark.asyncio
async def test_translation_retranslate_all_scope(client, monkeypatch):
sink = {}
_fake_retranslate(monkeypatch, sink)
await client.patch("/api/settings/import", json={
"translation_enabled": True, "interpreter_base_url": "http://i.lan",
})
resp = await client.post(
"/api/settings/translation/retranslate", json={"all": True})
assert resp.status_code == 202
assert sink["artist_ids"] is None
@pytest.mark.asyncio
async def test_translation_retranslate_bad_artist_id(client):
resp = await client.post(
"/api/settings/translation/retranslate", json={"artist_id": "abc"})
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_patch_rejects_non_object(client):
resp = await client.patch("/api/settings/import", json=[1, 2, 3])
+117 -3
View File
@@ -4,7 +4,7 @@ import pytest
from sqlalchemy import select
from backend.app.models import Artist, ImportSettings, Post
from backend.app.tasks.translation import translate_posts
from backend.app.tasks.translation import retranslate_posts, translate_posts
pytestmark = pytest.mark.integration
@@ -28,8 +28,8 @@ def _sf(db_sync):
return _SM()
def _artist(db):
a = Artist(name="A", slug="a")
def _artist(db, name="A", slug="a"):
a = Artist(name=name, slug=slug)
db.add(a)
db.flush()
return a
@@ -45,6 +45,15 @@ def _post(db, artist_id, *, title=None, desc=None, ext="p1"):
return p
def _mark_translated(p, *, lang="ja", ver="v1"):
"""Simulate a prior (old-model) translation so retranslate has something to
reset."""
p.post_title_translated = "OLD"
p.description_translated = "OLD"
p.translated_source_lang = lang
p.translation_engine_version = ver
def _enable(db, url="http://i.lan"):
cfg = db.execute(
select(ImportSettings).where(ImportSettings.id == 1)
@@ -125,3 +134,108 @@ def test_translate_posts_service_down_leaves_untranslated(db_sync, monkeypatch):
assert translate_posts() == "interpreter unavailable"
db_sync.refresh(p)
assert p.translated_source_lang is None # will retry next run
def _mock_new_model(monkeypatch, *, ver="v2"):
"""Interpreter is up and translates with a NEW engine version."""
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"NEW:{t}" for t in texts],
"detected_lang": "ja", "engine": "llm", "engine_version": ver,
},
)
def test_retranslate_resets_and_reruns(db_sync, monkeypatch):
# A post translated by the old model gets cleared + re-run through the new
# one (new engine_version proves the reset happened, not a cache short-circuit).
_patch(monkeypatch, db_sync)
_mock_new_model(monkeypatch)
a = _artist(db_sync)
p = _post(db_sync, a.id, title="ねこ", desc="かわいい")
_mark_translated(p)
_enable(db_sync)
db_sync.commit()
retranslate_posts(artist_ids=[a.id])
db_sync.refresh(p)
assert p.post_title_translated == "NEW:ねこ"
assert p.description_translated == "NEW:かわいい"
assert p.translation_engine_version == "v2"
def test_retranslate_artist_scope_leaves_others(db_sync, monkeypatch):
# Scoping to one artist must not touch another artist's translations.
_patch(monkeypatch, db_sync)
_mock_new_model(monkeypatch)
a1 = _artist(db_sync)
a2 = _artist(db_sync, name="B", slug="b")
p1 = _post(db_sync, a1.id, title="ねこ", ext="p1")
p2 = _post(db_sync, a2.id, title="いぬ", ext="p2")
_mark_translated(p1)
_mark_translated(p2)
_enable(db_sync)
db_sync.commit()
retranslate_posts(artist_ids=[a1.id])
db_sync.refresh(p1)
db_sync.refresh(p2)
assert p1.post_title_translated == "NEW:ねこ" # re-run
assert p1.translation_engine_version == "v2"
assert p2.post_title_translated == "OLD" # untouched
assert p2.translation_engine_version == "v1"
def test_retranslate_runs_until_done(db_sync, monkeypatch):
# With the chunk capped at 1 and 2 posts to redo, the first chunk re-enqueues
# itself (with _reset_done=True so it doesn't wipe the fresh work) to finish.
_patch(monkeypatch, db_sync)
_mock_new_model(monkeypatch)
monkeypatch.setattr("backend.app.tasks.translation._MAX_POSTS_PER_RUN", 1)
calls = []
monkeypatch.setattr(
"backend.app.tasks.translation.retranslate_posts.apply_async",
lambda *a, **k: calls.append((a, k)),
)
a = _artist(db_sync)
p1 = _post(db_sync, a.id, title="ねこ", ext="p1")
p2 = _post(db_sync, a.id, title="いぬ", ext="p2")
_mark_translated(p1)
_mark_translated(p2)
_enable(db_sync)
db_sync.commit()
retranslate_posts(artist_ids=[a.id])
assert len(calls) == 1 # one re-enqueue for the tail
args, _kw = calls[0]
assert args[1]["_reset_done"] is True
assert args[1]["artist_ids"] == [a.id]
def test_retranslate_disabled_does_not_reset(db_sync, monkeypatch):
# Never wipe translations we can't rebuild — a disabled service leaves them.
_patch(monkeypatch, db_sync)
a = _artist(db_sync)
p = _post(db_sync, a.id, title="ねこ", ext="p1")
_mark_translated(p)
db_sync.commit() # translation_enabled defaults False
assert retranslate_posts(artist_ids=[a.id]) == "disabled"
db_sync.refresh(p)
assert p.translated_source_lang == "ja"
assert p.post_title_translated == "OLD"
def test_retranslate_unhealthy_does_not_reset(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="p1")
_mark_translated(p)
_enable(db_sync)
db_sync.commit()
assert retranslate_posts(artist_ids=[a.id]) == "interpreter unavailable"
db_sync.refresh(p)
assert p.translated_source_lang == "ja"
assert p.post_title_translated == "OLD"