332 lines
14 KiB
Python
332 lines
14 KiB
Python
"""Translate character info to English before it reaches an image-prompt builder.
|
|
|
|
Character sheets are extracted from German books, so their free-text fields are
|
|
mostly German. The image models work best with English, and German leaks into
|
|
keyword logic ("hat dünne Arme" = "has thin arms" became a hat). This module is
|
|
the single translation layer:
|
|
|
|
* ``looks_german`` / ``needs_translation``: cheap, dependency-free detection.
|
|
* ``translate_fields``: one batched LLM call (JSON in, JSON out) for every field
|
|
of a character, with per-field validation, per-field fallback to the original
|
|
text, and a persistent cache keyed by (sha256 of the source, target language,
|
|
translator model, prompt version). The source records are never modified.
|
|
* The LLM is injected as a ``chat`` callable, so this module has no network or
|
|
database code of its own and tests never call a real model.
|
|
|
|
Never raises: every failure degrades to the untranslated original, reported in
|
|
``TranslationResult`` so the UI can say the prompt used untranslated text.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Callable, Protocol
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PROMPT_VERSION = "v1"
|
|
TARGET_LANG = "en"
|
|
ENV_FLAG = "PROMPT_TRANSLATION" # "off"/"0"/"false"/"no" disables; default on
|
|
MAX_BATCH_CHARS = 12000
|
|
|
|
ChatFn = Callable[[list, int], str]
|
|
|
|
|
|
class TranslationCache(Protocol):
|
|
def get(self, key: str) -> Any | None: ...
|
|
def put(self, key: str, value: Any) -> None: ...
|
|
|
|
|
|
class MemoryCache:
|
|
"""In-process cache (tests, and a fallback when the database is unavailable)."""
|
|
|
|
def __init__(self) -> None:
|
|
self.data: dict[str, Any] = {}
|
|
|
|
def get(self, key: str) -> Any | None:
|
|
return self.data.get(key)
|
|
|
|
def put(self, key: str, value: Any) -> None:
|
|
self.data[key] = value
|
|
|
|
|
|
def translation_enabled(settings: dict | None = None) -> bool:
|
|
"""Default ON; the env var (or a settings key ``prompt_translation``) turns it off."""
|
|
raw = os.environ.get(ENV_FLAG)
|
|
if raw is None and isinstance(settings, dict):
|
|
raw = settings.get("prompt_translation")
|
|
if raw is None:
|
|
return True
|
|
return str(raw).strip().lower() not in {"off", "0", "false", "no", "disabled"}
|
|
|
|
|
|
# ── language detection ─────────────────────────────────────────────────────────
|
|
_GERMAN_STOPWORDS = frozenset(
|
|
"der die das den dem des ein eine einen einem einer eines und oder aber nicht kein keine keinen "
|
|
"ist sind war waren wird werden wurde wurden sein ihr ihre ihren seine seinen sein er sie es wir "
|
|
"mit von zu zum zur bei nach aus für auf über unter vor hinter zwischen durch gegen ohne um "
|
|
"als wie auch sehr noch nur dass weil wenn dann hier dort mehr ihm ihn sich sein trägt tragen "
|
|
"hat hatte haben hatten".split()
|
|
) - {"hat", "haben", "um", "war", "sein"} # collide with English words or are too weak
|
|
# Words that are German-only in practice (kept apart from stop words that collide with English).
|
|
_GERMAN_VOCAB_WORDS = frozenset(
|
|
"mensch zwerg ork elf elfe mann frau junge mädchen kind bart haar haare augen arme arm beine "
|
|
"kleidung umhang mantel hemd hose stiefel rüstung kettenhemd schwert dolch axt bogen schild "
|
|
"alt jung groß klein dünn dick stark schwach schlank breit grau rot schwarz weiß braun blond "
|
|
"männlich weiblich krieger ritter könig königin hauptmann jahre jahr wolle leder".split()
|
|
)
|
|
_GERMAN_STEMS = ("zwerg", "mensch", "kleid", "umhang", "rüstung", "kettenhemd", "krieger")
|
|
_ENGLISH_STOPWORDS = frozenset(
|
|
"the a an and or but not no is are was were be been being has have had with without of to in on at "
|
|
"for from by his her their its he she they it we you who whom which that this these those very "
|
|
"wears wearing worn over under between through about as than then there here".split()
|
|
)
|
|
_UMLAUT_RE = re.compile(r"[äöüÄÖÜß]")
|
|
_WORD_RE = re.compile(r"[A-Za-zÄÖÜäöüß]+")
|
|
|
|
|
|
def _german_score(text: str) -> int:
|
|
words = [w.lower() for w in _WORD_RE.findall(text or "")]
|
|
score = sum(1 for w in words if w in _GERMAN_STOPWORDS or w in _GERMAN_VOCAB_WORDS)
|
|
score += sum(1 for w in words if len(w) >= 6 and any(stem in w for stem in _GERMAN_STEMS))
|
|
score += 2 * len(_UMLAUT_RE.findall(text or ""))
|
|
return score
|
|
|
|
|
|
def _english_score(text: str) -> int:
|
|
return sum(1 for w in _WORD_RE.findall(text or "") if w.lower() in _ENGLISH_STOPWORDS)
|
|
|
|
|
|
def looks_german(text: str) -> bool:
|
|
"""True when the text carries German markers and is not clearly English."""
|
|
text = str(text or "")
|
|
german = _german_score(text)
|
|
if german == 0:
|
|
return False
|
|
# An English sentence with one German proper name/umlaut stays English.
|
|
return german > _english_score(text) or german >= 3 and _english_score(text) < german * 2
|
|
|
|
|
|
def needs_translation(text: Any) -> bool:
|
|
"""Skip empty text, numbers, and text that is already English."""
|
|
if not isinstance(text, str):
|
|
return False
|
|
stripped = text.strip()
|
|
if not stripped or not re.search(r"[A-Za-zÄÖÜäöüß]", stripped):
|
|
return False
|
|
return looks_german(stripped)
|
|
|
|
|
|
# ── prompt ─────────────────────────────────────────────────────────────────────
|
|
_SYSTEM_PROMPT = (
|
|
"You are a precise translator inside an image-generation pipeline. The input is a JSON object whose "
|
|
"values are short texts about a fictional book character, usually German. Translate every value into "
|
|
"natural English.\n"
|
|
"Rules:\n"
|
|
"- Translate faithfully and keep the visual meaning. Do NOT add, drop, soften, explain or reorder traits.\n"
|
|
"- Keep proper names (people, places, peoples, ships, titles that are part of a name) exactly as written.\n"
|
|
"- Read grammar, not look-alike words: German 'hat' is usually the verb 'has' (\"hat dünne Arme\" = "
|
|
"\"has thin arms\"), 'Hut' is a hat.\n"
|
|
"- Species words: Mensch = human, Zwerg = dwarf, Ork = orc, Elf = elf, Kind = child. Keep brackets, "
|
|
"quotes and punctuation structure.\n"
|
|
"- A value that is already English stays unchanged. A value that is a JSON array stays an array with "
|
|
"the same number of items, each translated in place.\n"
|
|
"- Output ONLY one JSON object with exactly the same keys. No markdown, no comments, no explanations."
|
|
)
|
|
|
|
_ERROR_MARKERS = re.compile(
|
|
r"(as an ai|i cannot|i can't|i'm sorry|i am sorry|unable to (?:translate|comply)|"
|
|
r"error:|traceback|internal server error|bad gateway|model (?:not found|is loading))",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def _hash_source(value: Any) -> str:
|
|
raw = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, sort_keys=True)
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def cache_key(source: Any, model: str, lang: str = TARGET_LANG, version: str = PROMPT_VERSION) -> str:
|
|
return f"{_hash_source(source)}|{lang}|{model or '-'}|{version}"
|
|
|
|
|
|
# ── validation ─────────────────────────────────────────────────────────────────
|
|
def _name_tokens(names: list[str]) -> list[str]:
|
|
out: list[str] = []
|
|
for name in names or []:
|
|
name = str(name or "").strip()
|
|
if len(name) >= 3:
|
|
out.append(name)
|
|
out.extend(tok for tok in re.split(r"[\s,;]+", name) if len(tok) >= 4 and tok[:1].isupper())
|
|
return list(dict.fromkeys(out))
|
|
|
|
|
|
def validate_translation(source: str, result: Any, names: list[str] | None = None) -> str | None:
|
|
"""Return None when the translation is acceptable, else a short reason."""
|
|
if not isinstance(result, str) or not result.strip():
|
|
return "empty"
|
|
out = result.strip()
|
|
if _ERROR_MARKERS.search(out) and not _ERROR_MARKERS.search(source):
|
|
return "looks like an error or refusal"
|
|
low_src, low_out = len(source.strip()), len(out)
|
|
if low_out > low_src * 3 + 40 or low_out * 4 + 20 < low_src:
|
|
return "length out of range"
|
|
if out == source.strip() and _german_score(source) >= 2:
|
|
return "unchanged German"
|
|
if _german_score(out) >= 3 and _german_score(out) > _english_score(out):
|
|
return "still German"
|
|
for name in _name_tokens(names or []):
|
|
if re.search(re.escape(name), source, re.IGNORECASE) and name.casefold() not in out.casefold():
|
|
return f"name not preserved: {name}"
|
|
return None
|
|
|
|
|
|
def _validate_value(source: Any, result: Any, names: list[str]) -> str | None:
|
|
if isinstance(source, list):
|
|
if not isinstance(result, list) or len(result) != len(source):
|
|
return "list shape mismatch"
|
|
for src_item, out_item in zip(source, result):
|
|
if needs_translation(src_item):
|
|
reason = validate_translation(str(src_item), out_item, names)
|
|
if reason:
|
|
return reason
|
|
elif not isinstance(out_item, str) or not out_item.strip():
|
|
return "empty item"
|
|
return None
|
|
return validate_translation(str(source), result, names)
|
|
|
|
|
|
# ── parsing ────────────────────────────────────────────────────────────────────
|
|
def parse_json_object(raw: str) -> dict | None:
|
|
text = re.sub(r"<think>.*?</think>", "", raw or "", flags=re.DOTALL).strip()
|
|
text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text).strip()
|
|
candidates = [text]
|
|
start, end = text.find("{"), text.rfind("}")
|
|
if 0 <= start < end:
|
|
candidates.append(text[start:end + 1])
|
|
for candidate in candidates:
|
|
try:
|
|
parsed = json.loads(candidate)
|
|
except Exception:
|
|
continue
|
|
if isinstance(parsed, dict):
|
|
return parsed
|
|
return None
|
|
|
|
|
|
# ── main entry point ───────────────────────────────────────────────────────────
|
|
@dataclass
|
|
class TranslationResult:
|
|
values: dict[str, Any]
|
|
translated: list[str] = field(default_factory=list) # fields now in English (fresh or cached)
|
|
untranslated: list[str] = field(default_factory=list) # German fields that kept the original text
|
|
cache_hits: int = 0
|
|
skipped: list[str] = field(default_factory=list) # already English / empty
|
|
enabled: bool = True
|
|
error: str = ""
|
|
|
|
@property
|
|
def used_untranslated(self) -> bool:
|
|
return bool(self.untranslated)
|
|
|
|
def info(self) -> dict:
|
|
return {
|
|
"enabled": self.enabled,
|
|
"used_untranslated_text": self.used_untranslated,
|
|
"translated_fields": list(self.translated),
|
|
"untranslated_fields": list(self.untranslated),
|
|
"cache_hits": self.cache_hits,
|
|
"error": self.error,
|
|
}
|
|
|
|
|
|
def translate_fields(fields: dict[str, Any], *, chat: ChatFn | None, model: str = "",
|
|
cache: TranslationCache | None = None, names: list[str] | None = None,
|
|
enabled: bool = True) -> TranslationResult:
|
|
"""Return English versions of ``fields`` (str or list[str] values); never raises."""
|
|
values = dict(fields)
|
|
result = TranslationResult(values=values, enabled=enabled)
|
|
todo: dict[str, Any] = {}
|
|
for key, value in fields.items():
|
|
wants = (any(needs_translation(v) for v in value) if isinstance(value, list) else needs_translation(value))
|
|
if not wants:
|
|
result.skipped.append(key)
|
|
else:
|
|
todo[key] = value
|
|
if not todo:
|
|
return result
|
|
if not enabled:
|
|
result.untranslated = list(todo)
|
|
return result
|
|
names = list(names or [])
|
|
cache = cache if cache is not None else MemoryCache()
|
|
|
|
pending: dict[str, Any] = {}
|
|
for key, value in todo.items():
|
|
try:
|
|
hit = cache.get(cache_key(value, model))
|
|
except Exception as exc: # a broken cache must never block prompts
|
|
logger.warning("prompt translation cache read failed: %s", exc)
|
|
hit = None
|
|
if hit is not None and _validate_value(value, hit, names) is None:
|
|
values[key] = hit
|
|
result.translated.append(key)
|
|
result.cache_hits += 1
|
|
else:
|
|
pending[key] = value
|
|
if not pending:
|
|
return result
|
|
if chat is None:
|
|
result.untranslated = list(pending)
|
|
result.error = "no translation LLM configured"
|
|
return result
|
|
|
|
# Bounded batches keep one call inside the model's context.
|
|
batches: list[dict[str, Any]] = [{}]
|
|
size = 0
|
|
for key, value in pending.items():
|
|
length = len(json.dumps(value, ensure_ascii=False))
|
|
if batches[-1] and size + length > MAX_BATCH_CHARS:
|
|
batches.append({})
|
|
size = 0
|
|
batches[-1][key] = value
|
|
size += length
|
|
|
|
for batch in batches:
|
|
total = sum(len(json.dumps(v, ensure_ascii=False)) for v in batch.values())
|
|
messages = [
|
|
{"role": "system", "content": _SYSTEM_PROMPT},
|
|
{"role": "user", "content": json.dumps(batch, ensure_ascii=False)},
|
|
]
|
|
try:
|
|
raw = chat(messages, min(3072, max(400, total)))
|
|
parsed = parse_json_object(raw)
|
|
if parsed is None:
|
|
raise ValueError("translator did not return a JSON object")
|
|
except Exception as exc:
|
|
logger.warning("prompt translation failed (%s); using original text", exc)
|
|
result.error = str(exc)[:200]
|
|
result.untranslated.extend(batch)
|
|
continue
|
|
for key, source in batch.items():
|
|
candidate = parsed.get(key)
|
|
reason = _validate_value(source, candidate, names)
|
|
if reason:
|
|
logger.warning("prompt translation of %r rejected (%s); using original text", key, reason)
|
|
result.untranslated.append(key)
|
|
result.error = result.error or f"{key}: {reason}"
|
|
continue
|
|
if isinstance(candidate, str):
|
|
candidate = candidate.strip()
|
|
values[key] = candidate
|
|
result.translated.append(key)
|
|
try:
|
|
cache.put(cache_key(source, model), candidate)
|
|
except Exception as exc:
|
|
logger.warning("prompt translation cache write failed: %s", exc)
|
|
return result
|