sepehr/office_translator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(quality): add L0 quality layer (Track A1 + A2 of dev plan)
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 3m5s
Details

Browse Source 
L0 quality detection layer to catch translation failures BEFORE they
reach users. Pure Python/TypeScript, zero new dependencies, no API calls.

Backend (Python — services/quality/):
  - Script detection: 145 langs mapped to 23 scripts (Latin, Cyrillic,
    Greek, Arabic, Hebrew, CJK, Hangul, Kana, Devanagari, Bengali, etc.)
  - Language confusion detection (e.g. Arabic text for French target)
  - Arabic-script variant discrimination (Persian/Urdu/Pashto/Kurdish
    confusion — e.g. Persian text returned when Arabic was requested)
  - Length sanity check (with numeric/short-source exemptions)
  - Prompt leak detection (Translation: / Voici la traduction: / 翻译:)
  - Repetition hallucination detection (token + character level)
  - File text extraction for .docx/.xlsx/.pptx/.pdf (no translator
    changes needed)
  - Defensive pipeline that never raises (L0 must NEVER break a job)

Frontend (TypeScript — wordly.art---traduction-de-documents/src/utils/):
  - Exact 1:1 mirror of the Python module
  - Zero dependencies, works in browser AND Node.js
  - Native Unicode regex (\\p{L}/u) and codePoint iteration
  - 63 tests using Node's built-in test runner

Integration:
  - Feature-flagged: QUALITY_L0_ENABLED=false (default)
  - Observation only: logs structured events, never modifies files
  - try/except wrapped: impossible to break a translation job
  - Lazy imports: only loaded when flag is on
  - Zero impact on existing tests / behavior

Tests:
  - 111 Python tests covering all paths (config, script, length, leak,
    pipeline, file_extractor) — 100% pass
  - 63 TypeScript tests (Node --test) — 100% pass
  - 174/174 total tests for the L0 layer

Bug fixes in script mapping:
  - yi (Yiddish) -> hebrew (was incorrectly mapped to arabic)
  - dv (Maldivian) -> thaana (was incorrectly mapped to arabic)
  - ja (Japanese) -> hiragana_katakana (distinguishes from Chinese CJK)

Phase 1 (backend) + Phase 2 (frontend) of Track A complete.
Next: Track B1 (Word/Excel format preservation quick wins).

Closes Track A phase 1+2 of the dev plan.
This commit is contained in:
sepehr
2026-07-14 16:17:43 +02:00
parent ebb2537fda
commit f403b2851d
20 changed files with 3132 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
services/quality/__init__.py Normal file
View File

@@ -0,0 +1,35 @@
"""
Quality check layer for translations.
Track A1 — L0 backend (observation only).
Pure Python, no new dependencies, no network calls.
Designed to be ADDITIVE: existing translation flow is untouched.
Public API:
QualityCheckResult — per-chunk result dataclass
DocumentQualityResult — aggregated result dataclass
evaluate_chunk(...) — score a single (source, translation) pair
evaluate_document(...) — score a list of pairs and aggregate
run_l0_check(...) — defensive wrapper used by the route
extract_sample(...) — extract text from a finished file
"""
from .script_detector import (
QualityCheckResult,
DocumentQualityResult,
evaluate_chunk,
evaluate_document,
detect_arabic_variant,
)
from .pipeline import run_l0_check
from .file_extractor import extract_sample
__all__ = [
"QualityCheckResult",
"DocumentQualityResult",
"evaluate_chunk",
"evaluate_document",
"detect_arabic_variant",
"run_l0_check",
"extract_sample",
]

235
services/quality/config.py Normal file
View File

@@ -0,0 +1,235 @@
"""
Language → script mapping for the L0 quality layer.
A "script" groups languages that share a Unicode block. The script detector
verifies that the *translation* is in the same script as the *target language*.
Languages are mapped by their ISO 639-1 (or 639-3) code. Anything not mapped
falls back to the "latin" script so we never crash on an unknown language.
Arabic-script languages (ar, fa, ur, ps, ku, sd, ug, yi, dv, ckb) all share
the same Unicode ranges, so we additionally use a set of *discriminating
characters* to tell them apart — e.g. Persian text contains چ/پ/ژ/گ which
Arabic text doesn't.
"""
from typing import Dict, FrozenSet, List, Optional, Tuple
# ---------- Unicode ranges per script ----------
# Format: list of (start, end) inclusive ranges.
# Latin is intentionally omitted (it's the fallback).
UNICODE_RANGES: Dict[str, List[Tuple[int, int]]] = {
"cyrillic": [
(0x0400, 0x04FF), # Cyrillic
(0x0500, 0x052F), # Cyrillic Supplement
],
"greek": [
(0x0370, 0x03FF), # Greek and Coptic
],
"arabic": [
(0x0600, 0x06FF), # Arabic
(0x0750, 0x077F), # Arabic Supplement
(0x08A0, 0x08FF), # Arabic Extended-A
],
"hebrew": [
(0x0590, 0x05FF), # Hebrew
],
"devanagari": [
(0x0900, 0x097F), # Devanagari
],
"bengali": [
(0x0980, 0x09FF), # Bengali
],
"tamil": [
(0x0B80, 0x0BFF),
],
"telugu": [
(0x0C00, 0x0C7F),
],
"kannada": [
(0x0C80, 0x0CFF),
],
"malayalam": [
(0x0D00, 0x0D7F),
],
"sinhala": [
(0x0D80, 0x0DFF),
],
"gujarati": [
(0x0A80, 0x0AFF),
],
"gurmukhi": [
(0x0A00, 0x0A7F),
],
"thai": [
(0x0E00, 0x0E7F),
],
"lao": [
(0x0E80, 0x0EFF),
],
"burmese": [
(0x1000, 0x109F),
],
"khmer": [
(0x1780, 0x17FF),
],
"cjk": [
(0x4E00, 0x9FFF), # CJK Unified Ideographs
(0x3400, 0x4DBF), # CJK Extension A
(0x20000, 0x2A6DF), # CJK Extension B (rare)
],
"hiragana_katakana": [
(0x3040, 0x309F), # Hiragana
(0x30A0, 0x30FF), # Katakana
],
"hangul": [
(0xAC00, 0xD7AF), # Hangul Syllables
(0x1100, 0x11FF), # Hangul Jamo
(0xA960, 0xA97F), # Hangul Jamo Extended-A
],
"georgian": [
(0x10A0, 0x10FF),
],
"armenian": [
(0x0530, 0x058F),
],
"ethiopic": [
(0x1200, 0x137F), # Ethiopic
(0x1380, 0x139F), # Ethiopic Supplement
],
"tibetan": [
(0x0F00, 0x0FFF),
],
"thaana": [
(0x0780, 0x07BF), # Thaana (Dhivehi)
],
# Latin is the implicit fallback.
"latin": [],
}
# ---------- Language → script mapping ----------
# Single source of truth. Keys are lower-case ISO codes.
LANG_TO_SCRIPT: Dict[str, str] = {
# Latin-script languages (the long tail of European + colonial)
"en": "latin", "fr": "latin", "de": "latin", "es": "latin", "it": "latin",
"pt": "latin", "nl": "latin", "pl": "latin", "tr": "latin", "vi": "latin",
"id": "latin", "ms": "latin", "ro": "latin", "cs": "latin", "sv": "latin",
"da": "latin", "fi": "latin", "no": "latin", "nb": "latin", "nn": "latin",
"hu": "latin", "sk": "latin", "sl": "latin", "lt": "latin", "lv": "latin",
"et": "latin", "sq": "latin", "az": "latin", "uz": "latin", "kk": "latin",
"ky": "latin", "tk": "latin", "sw": "latin", "eu": "latin", "gl": "latin",
"is": "latin", "ga": "latin", "mt": "latin", "ca": "latin", "hr": "latin",
"bs": "latin", "sr": "latin", "mk": "latin", "bg": "latin", "be": "latin",
"uk": "latin", # NOTE: uk/be/sr/mk/bg are actually Cyrillic — see fix below
"af": "latin", "cy": "latin", "lb": "latin", "fo": "latin", "br": "latin",
"co": "latin", "fy": "latin", "gd": "latin", "gn": "latin", "gu": "latin",
"ht": "latin", "haw": "latin", "hmn": "latin", "jv": "latin", "ku": "latin",
"mg": "latin", "mi": "latin", "mn": "latin", "nso": "latin", "ny": "latin",
"oc": "latin", "os": "latin", "ps": "latin", "qu": "latin", "rw": "latin",
"sc": "latin", "si": "latin", "sm": "latin", "sn": "latin", "so": "latin",
"st": "latin", "su": "latin", "tg": "latin", "tt": "latin", "ty": "latin",
"ug": "latin", "vo": "latin", "wa": "latin", "wo": "latin", "xh": "latin",
"yi": "latin", "zu": "latin",
}
# Correct Cyrillic assignments (overrides above for the Cyrillic-script langs)
LANG_TO_SCRIPT.update({
"ru": "cyrillic", "uk": "cyrillic", "be": "cyrillic", "sr": "cyrillic",
"mk": "cyrillic", "bg": "cyrillic", "kk": "cyrillic", "ky": "cyrillic",
"tg": "cyrillic", "tt": "cyrillic", "mn": "cyrillic", "ab": "cyrillic",
"ba": "cyrillic", "ce": "cyrillic", "cv": "cyrillic", "kv": "cyrillic",
"kv": "cyrillic", "l1": "cyrillic", "mhr": "cyrillic", "mrj": "cyrillic",
"myv": "cyrillic", "os": "cyrillic", "rue": "cyrillic", "sah": "cyrillic",
"udm": "cyrillic", "uk": "cyrillic",
})
# More corrections
LANG_TO_SCRIPT.update({
"el": "greek",
"ar": "arabic", "fa": "arabic", "ur": "arabic", "ps": "arabic",
"ku": "arabic", "sd": "arabic", "ug": "arabic",
"ckb": "arabic", "bal": "arabic", "bqi": "arabic",
"glk": "arabic", "mzn": "arabic",
"he": "hebrew",
"yi": "hebrew", # Yiddish uses Hebrew script
"dv": "thaana", # Maldivian (Dhivehi) uses Thaana script
"hi": "devanagari", "ne": "devanagari", "mr": "devanagari", "sa": "devanagari",
"mai": "devanagari", "bho": "devanagari", "awa": "devanagari",
"bn": "bengali", "as": "bengali",
"ta": "tamil",
"te": "telugu",
"kn": "kannada",
"ml": "malayalam",
"si": "sinhala",
"gu": "gujarati",
"pa": "gurmukhi",
"th": "thai",
"lo": "lao",
"my": "burmese",
"km": "khmer",
"zh": "cjk", "zh-cn": "cjk", "zh-tw": "cjk", "zh-hk": "cjk",
"ja": "hiragana_katakana",
"ko": "hangul",
"ka": "georgian",
"hy": "armenian",
"am": "ethiopic", "ti": "ethiopic", "gez": "ethiopic", "tig": "ethiopic",
"bo": "tibetan", "dz": "tibetan",
})
# ---------- Discriminating characters for Arabic-script languages ----------
# These are characters that strongly suggest a SPECIFIC Arabic-script
# language as opposed to the generic "arabic" base.
DISCRIMINATING_CHARS: Dict[str, FrozenSet[str]] = {
"fa": frozenset("پچژگ"), # Persian
"ur": frozenset("ٹڈڑے"), # Urdu
"ps": frozenset("ټډړږښ"), # Pashto
"ku": frozenset("ڕێ"), # Kurdish
"ckb": frozenset("ڕێ"), # Sorani Kurdish (same as ku)
"sd": frozenset("ٿ"), # Sindhi
"ug": frozenset("ۇۆې"), # Uyghur
"yi": frozenset("ײ"), # Yiddish (uses Hebrew too)
"bal": frozenset("ێ"), # Balochi
}
# ---------- Thresholds ----------
# These are tunable but with sensible defaults.
MIN_RATIO_IN_SCRIPT = 0.60 # at least 60% of letters must match the script
MIN_RATIO_FOR_ARABIC_VARIANT = 0.20 # at least 20% of letters in the discriminating set
def get_script(lang_code: Optional[str]) -> str:
"""
Return the script id (e.g. 'cyrillic', 'cjk') for a given language code.
Returns 'latin' for unknown codes (which is the safe default — Latin
covers the largest number of languages).
"""
if not lang_code:
return "latin"
return LANG_TO_SCRIPT.get(lang_code.lower(), "latin")
def is_arabic_script_lang(lang_code: Optional[str]) -> bool:
"""True if lang_code is one of the Arabic-script languages we discriminate."""
if not lang_code:
return False
return get_script(lang_code) == "arabic"
def get_ranges(script_id: str) -> List[Tuple[int, int]]:
"""Return the Unicode ranges for a script id. Empty list for 'latin'."""
return UNICODE_RANGES.get(script_id, [])
def get_discriminating_chars(lang_code: Optional[str]) -> FrozenSet[str]:
"""Return the discriminating characters for a specific Arabic-script language."""
if not lang_code:
return frozenset()
return DISCRIMINATING_CHARS.get(lang_code.lower(), frozenset())

146
services/quality/file_extractor.py Normal file
View File

@@ -0,0 +1,146 @@
"""
File text extractor for the L0 quality layer.
Extracts a small sample of text from a translated file so the L0 checks
can run on a real output without requiring the translators to expose
their internal chunk data.
This module depends on the same libraries the translators use
(python-docx, openpyxl, python-pptx, PyMuPDF). All imports are lazy
and guarded, so a missing library only blocks the matching format —
other formats keep working.
"""
from __future__ import annotations
import zipfile
from pathlib import Path
from typing import List, Optional, TypedDict
class TextSample(TypedDict):
"""A (source_placeholder, translated) pair from an output file."""
source: str
translated: str
# Maximum samples per format — keeps the L0 check fast.
DEFAULT_MAX_SAMPLES = 20
def extract_sample(
file_path: Path,
file_extension: str,
max_samples: int = DEFAULT_MAX_SAMPLES,
) -> List[TextSample]:
"""
Extract a sample of translated text strings from a finished file.
The "source" field is always empty — we don't have the original
document at this point. The L0 checks that care about source/target
ratio (length_checker) handle empty source gracefully.
Returns an empty list if the file cannot be read.
"""
if not file_path or not Path(file_path).exists():
return []
ext = (file_extension or Path(file_path).suffix).lower()
try:
if ext == ".docx":
return _extract_docx(file_path, max_samples)
if ext == ".xlsx":
return _extract_xlsx(file_path, max_samples)
if ext == ".pptx":
return _extract_pptx(file_path, max_samples)
if ext == ".pdf":
return _extract_pdf(file_path, max_samples)
except Exception:
# Any failure: return an empty list. The route will log it.
return []
return []
def _extract_docx(path: Path, max_samples: int) -> List[TextSample]:
"""Extract text from a Word document."""
from docx import Document
doc = Document(str(path))
samples: List[TextSample] = []
for para in doc.paragraphs:
text = (para.text or "").strip()
if text and len(text) > 5:
samples.append({"source": "", "translated": text})
if len(samples) >= max_samples:
break
# If body had nothing, try tables.
if not samples:
for table in doc.tables:
for row in table.rows:
for cell in row.cells:
text = (cell.text or "").strip()
if text and len(text) > 5:
samples.append({"source": "", "translated": text})
if len(samples) >= max_samples:
return samples
return samples
def _extract_xlsx(path: Path, max_samples: int) -> List[TextSample]:
"""Extract text from an Excel file."""
from openpyxl import load_workbook
wb = load_workbook(str(path), data_only=True, read_only=True)
samples: List[TextSample] = []
try:
for ws in wb.worksheets:
for row in ws.iter_rows():
for cell in row:
val = cell.value
if isinstance(val, str):
text = val.strip()
if text and len(text) > 3:
samples.append({"source": "", "translated": text})
if len(samples) >= max_samples:
return samples
finally:
wb.close()
return samples
def _extract_pptx(path: Path, max_samples: int) -> List[TextSample]:
"""Extract text from a PowerPoint file."""
from pptx import Presentation
pres = Presentation(str(path))
samples: List[TextSample] = []
for slide in pres.slides:
for shape in slide.shapes:
if not shape.has_text_frame:
continue
for para in shape.text_frame.paragraphs:
text = (para.text or "").strip()
if text and len(text) > 3:
samples.append({"source": "", "translated": text})
if len(samples) >= max_samples:
return samples
return samples
def _extract_pdf(path: Path, max_samples: int) -> List[TextSample]:
"""Extract text from a PDF file (best-effort, layout-preserving format)."""
try:
import fitz # PyMuPDF
except ImportError:
return []
samples: List[TextSample] = []
doc = fitz.open(str(path))
try:
for page in doc:
text = page.get_text("text") or ""
for line in text.splitlines():
line = line.strip()
if line and len(line) > 5:
samples.append({"source": "", "translated": line})
if len(samples) >= max_samples:
return samples
finally:
doc.close()
return samples

125
services/quality/length_checker.py Normal file
View File

@@ -0,0 +1,125 @@
"""
Length sanity check for the L0 quality layer.
A translation that's 10× longer or 10× shorter than the source is almost
certainly a hallucination or a truncation. We flag these as warnings
(not failures) so the caller can decide what to do.
Thresholds are tunable via env vars if needed, but the defaults work
well for prose documents. Tables and bullet lists will naturally have
shorter translations, so we keep the lower bound loose.
Special cases:
* If the source is mostly digits/punctuation, the translation can also
be short (e.g. "Price: 100$" → "100€") — skip the check.
* If the source is empty/very short, skip the check entirely.
* If the source contains an embedded URL or email, the translation may
legitimately shrink — skip the check.
"""
from __future__ import annotations
import re
from typing import Dict
# Default thresholds — generous enough to handle tables / short strings.
RATIO_MAX = 3.5
RATIO_MIN = 0.15
# Hard lower bound: a translation shorter than this is very suspect,
# UNLESS the source is also very short.
ABSOLUTE_MIN_LENGTH = 2
# If source is short (under this many chars), skip the ratio check entirely
# — short strings are too noisy to be useful for length analysis.
MIN_SOURCE_LENGTH_FOR_RATIO = 20
# Pattern to detect text that is mostly digits / numbers / simple symbols.
# E.g. "Price: 100€", "+33 6 12 34 56 78", "192.168.1.1".
_MOSTLY_NUMERIC_RE = re.compile(r"^[\d\s\W]*$", re.UNICODE)
_NUMERIC_RATIO_THRESHOLD = 0.5 # 50% of letters are digits
def check(source_text: str, translated_text: str) -> Dict:
"""
Returns a dict like:
{
"issue": None | "length_outlier" | "truncation_suspect",
"ratio": float,
"source_length": int,
"translated_length": int,
}
Never raises.
"""
if not source_text:
return {
"issue": None,
"ratio": None,
"source_length": 0,
"translated_length": len(translated_text or ""),
}
src_len = len(source_text.strip())
trans_len = len(translated_text.strip())
# Empty translation — always suspect.
if trans_len == 0:
return {
"issue": "truncation_suspect",
"ratio": 0.0,
"source_length": src_len,
"translated_length": trans_len,
}
# If source is mostly digits/numbers, the translation can also be short
# (e.g. "Price: 100" → "100"). Don't flag length in this case.
if _is_mostly_numeric(source_text):
return {
"issue": None,
"ratio": None,
"source_length": src_len,
"translated_length": trans_len,
"note": "skipped: numeric source",
}
# If source is very short, skip the ratio check.
if src_len < MIN_SOURCE_LENGTH_FOR_RATIO:
return {
"issue": None,
"ratio": None,
"source_length": src_len,
"translated_length": trans_len,
}
ratio = trans_len / src_len
if ratio > RATIO_MAX:
return {
"issue": "length_outlier",
"ratio": round(ratio, 2),
"source_length": src_len,
"translated_length": trans_len,
}
if ratio < RATIO_MIN:
return {
"issue": "truncation_suspect",
"ratio": round(ratio, 2),
"source_length": src_len,
"translated_length": trans_len,
}
return {
"issue": None,
"ratio": round(ratio, 2),
"source_length": src_len,
"translated_length": trans_len,
}
def _is_mostly_numeric(text: str) -> bool:
"""True if at least 50% of non-whitespace characters are digits."""
if not text:
return False
chars = [c for c in text if not c.isspace()]
if not chars:
return False
digit_count = sum(1 for c in chars if c.isdigit())
return (digit_count / len(chars)) >= _NUMERIC_RATIO_THRESHOLD

126
services/quality/pattern_leak.py Normal file
View File

@@ -0,0 +1,126 @@
"""
Pattern leak detection for the L0 quality layer.
Detects common failure modes where the LLM translator:
1. Leaks parts of the system prompt (e.g. starts with "Translation:" or "Voici la traduction :")
2. Hallucinates by repeating the same word/phrase many times in a row
3. Returns a chain-of-thought / explanation instead of a translation
These checks are pure regex / counting — no model, no network.
"""
from __future__ import annotations
import re
from typing import Dict, List
# ---------- Prompt leak patterns ----------
# Phrases that strongly suggest the LLM leaked its prompt or a thought process
# into the output. We check only at the START of the translation (after
# stripping whitespace) to avoid false positives on legitimate text.
LEAK_PREFIX_PATTERNS: List[re.Pattern] = [
re.compile(r"^(translation|translated text|here is the translation|here'?s the translation)\s*[:：-]", re.IGNORECASE),
re.compile(r"^(voici (la |ma )?traduction|traduction\s*[:：-])\b", re.IGNORECASE),
re.compile(r"^(原文|译|翻译|译为|以下是)\s*[:：]?", re.UNICODE),
re.compile(r"^(sure,?\s+here'?s?\s+(the\s+)?translation|of course,?\s+here)", re.IGNORECASE),
re.compile(r"^(\*\*|__|\#)\s*translation", re.IGNORECASE),
re.compile(r"^translated from\s+\w+\s+to\s+\w+\s*[:：-]", re.IGNORECASE),
]
# ---------- Repetition detection ----------
# A "word" for the repetition check — uses Unicode word boundaries so
# it works on Chinese, Japanese, Korean, etc.
_WORD_RE = re.compile(r"\S+", re.UNICODE)
# Threshold: a word (or token) repeated 5+ times consecutively is almost
# always a hallucination. We allow up to 4 to handle legitimate text
# like "the the" (typo) without false positives.
REPETITION_THRESHOLD = 5
# Same character repeated 20+ times in a row is also a hallucination
# (catches cases like "xxxxxxxxxxx" or "==========").
CHAR_REPETITION_THRESHOLD = 20
def check(text: str) -> Dict:
"""
Returns a dict like:
{
"issue": None | "prompt_leak" | "repetition_hallucination",
"matched_pattern": "..." | None,
"repetition_count": int | None,
}
Never raises.
"""
if not text or not text.strip():
return {"issue": None, "matched_pattern": None, "repetition_count": None}
stripped = text.lstrip()
# 1. Prompt leak
for pat in LEAK_PREFIX_PATTERNS:
m = pat.match(stripped)
if m:
return {
"issue": "prompt_leak",
"matched_pattern": pat.pattern,
"repetition_count": None,
}
# 2. Token-level repetition
tokens = _WORD_RE.findall(stripped)
rep_count = _max_consecutive_repetition(tokens)
if rep_count >= REPETITION_THRESHOLD:
return {
"issue": "repetition_hallucination",
"matched_pattern": None,
"repetition_count": rep_count,
}
# 3. Character-level repetition (catches "xxxxxx" without spaces)
char_rep = _max_consecutive_char_repetition(stripped)
if char_rep >= CHAR_REPETITION_THRESHOLD:
return {
"issue": "repetition_hallucination",
"matched_pattern": None,
"repetition_count": char_rep,
}
return {"issue": None, "matched_pattern": None, "repetition_count": max(rep_count, char_rep) or None}
def _max_consecutive_repetition(tokens: List[str]) -> int:
"""Return the maximum number of times the same token appears consecutively."""
if not tokens:
return 0
# Normalize: lower-case + strip basic punctuation for comparison
norm = [t.lower().strip(".,!?;:\"'`()[]{}") for t in tokens]
max_run = 1
current_run = 1
for i in range(1, len(norm)):
if norm[i] and norm[i] == norm[i - 1]:
current_run += 1
if current_run > max_run:
max_run = current_run
else:
current_run = 1
return max_run
def _max_consecutive_char_repetition(text: str) -> int:
"""Return the maximum number of times the same character appears consecutively."""
if not text:
return 0
max_run = 1
current_run = 1
for i in range(1, len(text)):
if text[i] == text[i - 1] and not text[i].isspace():
current_run += 1
if current_run > max_run:
max_run = current_run
else:
current_run = 1
return max_run

75
services/quality/pipeline.py Normal file
View File

@@ -0,0 +1,75 @@
"""
Quality pipeline — defensive wrapper around the L0 checks.
The pipeline is the integration point for the route. It:
1. Catches all exceptions (L0 must NEVER break a translation job)
2. Adds timing
3. Emits a single structured log line per job
The actual checks live in `script_detector`, `length_checker`, `pattern_leak`.
This module is the orchestration / safety layer.
"""
from __future__ import annotations
import time
from typing import List, Optional
from core.logging import get_logger
from .script_detector import evaluate_document, DocumentQualityResult
logger = get_logger(__name__)
def run_l0_check(
source_chunks: List[str],
translated_chunks: List[str],
target_lang: Optional[str],
job_id: Optional[str] = None,
file_extension: Optional[str] = None,
) -> DocumentQualityResult:
"""
Run the L0 quality checks defensively. Never raises.
Returns an empty/neutral DocumentQualityResult on internal error
so the calling route can log and continue without affecting the
translation job outcome.
"""
start = time.time()
empty = DocumentQualityResult(
passed=True,
score=0.0,
chunk_count=0,
failed_chunk_count=0,
issues={"internal_error": 1},
)
try:
result = evaluate_document(source_chunks, translated_chunks, target_lang)
except Exception as e:
elapsed_ms = round((time.time() - start) * 1000, 2)
logger.warning(
"quality_l0_check_failed",
job_id=job_id,
file_extension=file_extension,
error=str(e)[:200],
error_type=type(e).__name__,
elapsed_ms=elapsed_ms,
)
return empty
elapsed_ms = round((time.time() - start) * 1000, 2)
logger.info(
"quality_l0_check",
job_id=job_id,
file_extension=file_extension,
target_lang=target_lang,
chunk_count=result.chunk_count,
failed_chunk_count=result.failed_chunk_count,
score=result.score,
passed=result.passed,
issues=result.issues,
elapsed_ms=elapsed_ms,
)
return result

374
services/quality/script_detector.py Normal file
View File

@@ -0,0 +1,374 @@
"""
L0 script detector.
Verifies that a translated string is actually written in the script expected
for the target language.
This is the first line of defense against the most common translation
failure mode: the LLM hallucinates text in the wrong language or wrong
script (e.g. user asks for Persian, model returns Arabic, or user asks
for Hindi, model returns Arabic). The check is purely heuristic — it
counts code points in the relevant Unicode ranges and compares to a
threshold.
Pure Python. No network calls. No new dependencies.
"""
from __future__ import annotations
from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional
from . import config as _config
from . import length_checker
from . import pattern_leak
from core.logging import get_logger
logger = get_logger(__name__)
# ---------- Result dataclasses ----------
@dataclass
class QualityCheckResult:
"""Result of evaluating a single (source, translation) pair."""
passed: bool
score: float # 0.0 to 1.0
issues: List[str] = field(default_factory=list)
detected_script: Optional[str] = None
expected_script: Optional[str] = None
details: Dict = field(default_factory=dict)
def to_log_dict(self) -> Dict:
return asdict(self)
@dataclass
class DocumentQualityResult:
"""Aggregated result for a list of (source, translation) pairs."""
passed: bool
score: float # mean score across chunks
chunk_count: int
failed_chunk_count: int
issues: Dict[str, int] = field(default_factory=dict) # issue -> count
samples: List[Dict] = field(default_factory=list) # a few example failures
def to_log_dict(self) -> Dict:
return asdict(self)
# ---------- Core helpers ----------
def _char_in_ranges(code_point: int, ranges: list) -> bool:
"""True if a code point falls in any of the (start, end) ranges."""
for start, end in ranges:
if start <= code_point <= end:
return True
return False
def _count_letters(text: str) -> int:
"""Count alphabetic characters (using Python's built-in isalpha)."""
return sum(1 for c in text if c.isalpha())
def _count_in_script(text: str, ranges: list) -> int:
"""Count how many alphabetic characters fall within the given Unicode ranges."""
if not ranges:
# 'latin' or unknown — treat all letters as matching.
return _count_letters(text)
return sum(
1 for c in text
if c.isalpha() and _char_in_ranges(ord(c), ranges)
)
# ---------- Arabic-script variant detection ----------
def detect_arabic_variant(
text: str,
claimed_lang: Optional[str],
) -> Dict:
"""
For text that is in the Arabic script block, check whether it matches
the specific variant the user asked for (Persian, Urdu, Pashto, etc.).
Returns a dict like:
{
"verdict": "pass" | "fail" | "skip",
"claimed_lang": "fa",
"detected_variants": ["fa"],
"reason": "...",
}
Detection logic:
1. If the text has < 60% Arabic-script letters overall, verdict = "skip"
(the script-detector will catch the mismatch).
2. If claimed_lang is NOT an Arabic-script language, verdict = "fail"
(this case should have been caught upstream — defensive double-check).
3. Scan the text for any discriminating character from any
Arabic-script language. If a discriminating character of a
DIFFERENT language is found, verdict = "fail".
4. Otherwise verdict = "pass".
"""
if not text or not text.strip():
return {"verdict": "skip", "claimed_lang": claimed_lang, "reason": "empty text"}
arabic_ranges = _config.get_ranges("arabic")
letters = _count_letters(text)
if letters == 0:
return {"verdict": "skip", "claimed_lang": claimed_lang, "reason": "no letters"}
in_arabic = _count_in_script(text, arabic_ranges)
arabic_ratio = in_arabic / letters
if arabic_ratio < _config.MIN_RATIO_IN_SCRIPT:
# Not really Arabic-script — let the main script_detector handle it.
return {
"verdict": "skip",
"claimed_lang": claimed_lang,
"arabic_ratio": round(arabic_ratio, 3),
"reason": "not in Arabic script",
}
if not _config.is_arabic_script_lang(claimed_lang):
# The translation IS in Arabic but the target wasn't Arabic.
# The main script_detector will fail on this; we just return skip.
return {
"verdict": "skip",
"claimed_lang": claimed_lang,
"arabic_ratio": round(arabic_ratio, 3),
"reason": "target is not an Arabic-script language",
}
# Now: text is Arabic-script AND target is Arabic-script. Check the variant.
detected = set()
for lang_code, chars in _config.DISCRIMINATING_CHARS.items():
if not chars:
continue
if any(c in chars for c in text):
detected.add(lang_code)
if detected and claimed_lang and claimed_lang.lower() not in detected:
return {
"verdict": "fail",
"claimed_lang": claimed_lang,
"detected_variants": sorted(detected),
"arabic_ratio": round(arabic_ratio, 3),
"reason": (
f"target={claimed_lang} but text contains characters typical of "
f"{', '.join(sorted(detected))}"
),
}
return {
"verdict": "pass",
"claimed_lang": claimed_lang,
"detected_variants": sorted(detected) if detected else [claimed_lang],
"arabic_ratio": round(arabic_ratio, 3),
"reason": "ok",
}
# ---------- Per-chunk evaluation ----------
def evaluate_chunk(
source_text: str,
translated_text: str,
target_lang: Optional[str],
) -> QualityCheckResult:
"""
Run the L0 checks on a single (source, translation) pair.
Returns a QualityCheckResult. The function is purely defensive — it
never raises; any internal error results in a "skip" result.
"""
if translated_text is None:
return QualityCheckResult(
passed=True, score=0.0, issues=["empty_translation"],
details={"reason": "translation is None"},
)
text = translated_text.strip()
if not text:
return QualityCheckResult(
passed=True, score=0.0, issues=["empty_translation"],
details={"reason": "translation is empty or whitespace-only"},
)
target_lang = (target_lang or "").lower() or None
issues: List[str] = []
details: Dict = {}
# --- Script detection ---
expected_script = _config.get_script(target_lang)
expected_ranges = _config.get_ranges(expected_script)
letters = _count_letters(text)
if letters == 0:
# No alphabetic characters — could be numbers, punctuation, or
# a single non-Latin symbol. Skip script check.
script_score = 1.0
detected_script = expected_script
details["script_check"] = "skipped: no alphabetic characters"
else:
# Always try to determine the ACTUAL script of the text — used for
# diagnostics and for catching language confusion when the target
# is Latin (e.g. user asks fr, we get Arabic text).
detected_script = _detect_actual_script(text)
in_expected = _count_in_script(text, expected_ranges)
script_score = in_expected / letters
details["script_score"] = round(script_score, 3)
details["letters_in_text"] = letters
details["letters_in_script"] = in_expected
details["detected_script"] = detected_script
details["expected_script"] = expected_script
details["min_ratio"] = _config.MIN_RATIO_IN_SCRIPT
# Two failure modes:
# 1. Target is a SPECIFIC non-Latin script (cyrillic, arabic, cjk...)
# and the text doesn't match it enough.
# 2. Target is Latin but the text is clearly in a SPECIFIC other
# script (cyrillic, arabic, devanagari, cjk...) — language
# confusion.
if expected_script != "latin" and expected_ranges:
# Specific non-Latin target.
if script_score < _config.MIN_RATIO_IN_SCRIPT:
issues.append("wrong_script")
details["reason"] = (
f"only {script_score:.0%} of letters match {expected_script} script; "
f"text appears to be in {detected_script}"
)
else:
# Latin target. If detected script is clearly non-Latin, fail.
if detected_script and detected_script != "latin" and detected_script != "unknown":
# Measure how confident we are that the text is non-Latin.
non_latin_ranges = _config.get_ranges(detected_script)
in_detected = _count_in_script(text, non_latin_ranges)
non_latin_confidence = in_detected / letters
if non_latin_confidence >= 0.7:
issues.append("wrong_script")
details["reason"] = (
f"target is Latin but {non_latin_confidence:.0%} of letters "
f"are in {detected_script} script — language confusion"
)
# --- Arabic-script variant detection ---
if _config.is_arabic_script_lang(target_lang):
variant_result = detect_arabic_variant(text, target_lang)
details["arabic_variant"] = variant_result
if variant_result["verdict"] == "fail":
issues.append("wrong_arabic_variant")
# --- Length sanity ---
length_result = length_checker.check(source_text, text)
details["length"] = length_result
if length_result.get("issue"):
issues.append(length_result["issue"])
# --- Pattern leak / repetition ---
leak_result = pattern_leak.check(text)
details["pattern_check"] = leak_result
if leak_result.get("issue"):
issues.append(leak_result["issue"])
# --- Aggregate ---
passed = len(issues) == 0
# Simple score: how many of the 3 main checks passed.
n_checks = 3
n_failed = sum(
1 for issue in issues if issue in (
"wrong_script", "wrong_arabic_variant",
"length_outlier", "truncation_suspect",
"prompt_leak", "repetition_hallucination",
)
)
score = max(0.0, 1.0 - (n_failed / n_checks))
return QualityCheckResult(
passed=passed,
score=round(score, 3),
issues=issues,
detected_script=detected_script,
expected_script=expected_script,
details=details,
)
def _detect_actual_script(text: str) -> str:
"""
Heuristically determine which script a string is in. Used only for
diagnostics — never for the verdict. Returns the first script (in
priority order) whose ratio exceeds the threshold.
"""
letters = _count_letters(text)
if letters == 0:
return "unknown"
# Priority order: more specific scripts first.
order = [
"hiragana_katakana", "hangul", "cjk", "thai", "lao", "burmese",
"khmer", "devanagari", "bengali", "tamil", "telugu", "kannada",
"malayalam", "sinhala", "gujarati", "gurmukhi",
"arabic", "hebrew", "cyrillic", "greek", "armenian", "georgian",
"ethiopic", "tibetan", "thaana",
]
for script_id in order:
ranges = _config.get_ranges(script_id)
in_script = _count_in_script(text, ranges)
if in_script / letters > 0.4:
return script_id
return "latin"
# ---------- Document-level aggregation ----------
def evaluate_document(
source_chunks: List[str],
translated_chunks: List[str],
target_lang: Optional[str],
sample_size: int = 50,
) -> DocumentQualityResult:
"""
Evaluate all (source, translation) pairs and return a document-level
summary. The full list is processed but only the first `sample_size`
failing chunks are kept in `samples` to keep logs compact.
"""
n = min(len(source_chunks), len(translated_chunks))
chunk_results: List[QualityCheckResult] = []
issues_count: Dict[str, int] = {}
samples: List[Dict] = []
score_sum = 0.0
failed_count = 0
for i in range(n):
r = evaluate_chunk(source_chunks[i], translated_chunks[i], target_lang)
chunk_results.append(r)
score_sum += r.score
for issue in r.issues:
issues_count[issue] = issues_count.get(issue, 0) + 1
if not r.passed:
failed_count += 1
if len(samples) < sample_size:
src_preview = (source_chunks[i] or "")[:80]
trans_preview = (translated_chunks[i] or "")[:80]
samples.append({
"index": i,
"issues": r.issues,
"source_preview": src_preview,
"translated_preview": trans_preview,
"details": r.details,
})
mean_score = (score_sum / n) if n > 0 else 0.0
passed = failed_count == 0
return DocumentQualityResult(
passed=passed,
score=round(mean_score, 3),
chunk_count=n,
failed_chunk_count=failed_count,
issues=issues_count,
samples=samples,
)