diff --git a/services/quality/script_detector.py b/services/quality/script_detector.py index 21f5fd5..bb4557c 100644 --- a/services/quality/script_detector.py +++ b/services/quality/script_detector.py @@ -162,6 +162,31 @@ def detect_arabic_variant( ), } + # No discriminating character of ANY Arabic-script variant was found. + # For a long text this is itself a strong signal: a Persian/Urdu/Pashto + # text of this length almost always contains at least one of its + # specific letters (Persian پ چ ژ گ are common letters), while plain + # Arabic never does. Short texts stay tolerant — the signal is too + # weak to tell plain Arabic from e.g. Persian without those letters. + target_chars = _config.get_discriminating_chars(claimed_lang) + if ( + not detected + and claimed_lang + and claimed_lang.lower() != "ar" + and target_chars + and letters >= 25 + ): + return { + "verdict": "fail", + "claimed_lang": claimed_lang, + "detected_variants": [], + "arabic_ratio": round(arabic_ratio, 3), + "reason": ( + f"target={claimed_lang} but none of its specific letters appear " + f"in a {letters}-letter Arabic-script text — likely plain Arabic" + ), + } + return { "verdict": "pass", "claimed_lang": claimed_lang, @@ -173,31 +198,14 @@ def detect_arabic_variant( # ---------- Per-chunk evaluation ---------- -def evaluate_chunk( - source_text: str, - translated_text: str, - target_lang: Optional[str], -) -> QualityCheckResult: +def _script_checks(text: str, target_lang: Optional[str]) -> tuple: """ - Run the L0 checks on a single (source, translation) pair. + Script checks shared by evaluate_chunk (QA) and script_issue (retry). - Returns a QualityCheckResult. The function is purely defensive — it - never raises; any internal error results in a "skip" result. + Returns (issues, details): issues is a list of + "wrong_script"/"wrong_arabic_variant" strings, details carries the + diagnostics evaluate_chunk logs. """ - 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 = {} @@ -262,6 +270,68 @@ def evaluate_chunk( if variant_result["verdict"] == "fail": issues.append("wrong_arabic_variant") + return issues, details + + +def script_issue(translated_text: str, target_lang: Optional[str]) -> Optional[str]: + """ + Standalone wrong-script check, usable by the translators to decide + whether a delivered translation must be retried. + + Returns a short human-readable reason when the text is NOT written + in the script expected for target_lang (wrong script, or wrong + Arabic-script variant such as Arabic delivered for Persian), and + None when the script is acceptable. Purely heuristic, never raises. + """ + try: + text = (translated_text or "").strip() + if not text: + return None + issues, details = _script_checks(text, (target_lang or "").lower() or None) + if not issues: + return None + if "wrong_arabic_variant" in issues: + variant = details.get("arabic_variant") or {} + return variant.get("reason") or "wrong_arabic_variant" + return details.get("reason") or issues[0] + except Exception as e: # defensive: retry decisions must never crash a job + logger.warning("script_issue_error", error=str(e)[:200]) + return None + + +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 + Arabic variant (shared with script_issue) --- + script_issues, script_details = _script_checks(text, target_lang) + issues.extend(script_issues) + details.update(script_details) + # --- Length sanity --- length_result = length_checker.check(source_text, text) details["length"] = length_result @@ -291,8 +361,8 @@ def evaluate_chunk( passed=passed, score=round(score, 3), issues=issues, - detected_script=detected_script, - expected_script=expected_script, + detected_script=script_details.get("detected_script"), + expected_script=script_details.get("expected_script"), details=details, ) diff --git a/tests/test_translators/test_script_retry.py b/tests/test_translators/test_script_retry.py new file mode 100644 index 0000000..a961b0a --- /dev/null +++ b/tests/test_translators/test_script_retry.py @@ -0,0 +1,212 @@ +""" +Tests du garde-fou d'écriture (wrong-script retry). + +Scénario principal : un LLM livre de l'arabe alors que la cible est le +persan — même écriture, mauvaise langue, indétectable à l'œil d'un +non-lecteur. Le détecteur doit le signaler et le traducteur doit +redemander une fois avec une consigne renforcée, sans jamais risquer +de dégrader le document (la 2e tentative ne remplace que si elle passe +le même contrôle). +""" + +from docx import Document + +from services.providers.base import TranslationProvider +from services.providers.schemas import TranslationRequest, TranslationResponse +from services.quality.script_detector import script_issue +from translators.segments import retry_wrong_script +from translators.word_translator import WordTranslator + +# Arabe pur (aucune lettre persane spécifique), bien plus de 25 lettres. +ARABIC_FOR_PERSIAN = "مرحبا بكم في صفحة المشاريع الجديدة الخاصة بنا" +# Persan correct : contient پ et ژ (lettres spécifiques du persan). +GOOD_PERSIAN = "به صفحه پروژه های جدید ما خوش آمدید" +SOURCE_EN = "Welcome to our new projects page" + + +class SequencedMockProvider(TranslationProvider): + """Renvoie une mauvaise traduction, puis la bonne dès que la requête + porte la consigne de retentative (détectée via le custom prompt).""" + + def __init__(self, wrong: dict, corrected: dict): + self._wrong = wrong + self._corrected = corrected + self.calls = [] # (texte, porte_la_consigne_de_retry) + + def get_name(self) -> str: + return "sequenced-mock" + + def is_available(self) -> bool: + return True + + def translate_text(self, request: TranslationRequest) -> TranslationResponse: + custom = (request.metadata or {}).get("custom_prompt") or "" + is_retry = "CRITICAL RETRY" in custom + self.calls.append((request.text, is_retry)) + table = self._corrected if is_retry else self._wrong + return TranslationResponse( + translated_text=table.get(request.text, request.text), + provider_name="sequenced-mock", + source_language=request.source_language, + ) + + def translate_batch(self, requests: list) -> list: + return [self.translate_text(req) for req in requests] + + +# --------------------------------------------------------------------------- +# script_issue — détection seule +# --------------------------------------------------------------------------- + + +class TestScriptIssue: + def test_pure_arabic_for_persian_flagged(self): + reason = script_issue(ARABIC_FOR_PERSIAN, "fa") + assert reason is not None + assert "fa" in reason + + def test_correct_persian_passes(self): + assert script_issue(GOOD_PERSIAN, "fa") is None + + def test_latin_for_persian_flagged(self): + assert script_issue("Bienvenue sur notre page", "fa") is not None + + def test_cyrillic_for_french_flagged(self): + reason = script_issue("Добро пожаловать на новую страницу", "fr") + assert reason is not None + + def test_short_pure_arabic_stays_tolerated(self): + # Texte court : pas assez de signal pour distinguer arabe et + # persan sans lettre discriminante — comportement historique. + assert script_issue("السلام عليكم", "fa") is None + + def test_identity_and_empty_are_never_flagged(self): + assert script_issue("", "fa") is None + assert script_issue("2024", "fa") is None + + +# --------------------------------------------------------------------------- +# retry_wrong_script — mécanisme seul +# --------------------------------------------------------------------------- + + +class TestRetryWrongScript: + def test_retry_replaces_with_valid_translation(self): + def translate_fn(texts, hint): + assert "CRITICAL RETRY" in hint + assert texts == [SOURCE_EN] + return [GOOD_PERSIAN] + + result, fixed = retry_wrong_script( + [SOURCE_EN], [ARABIC_FOR_PERSIAN], "fa", translate_fn + ) + assert fixed == 1 + assert result[0] == GOOD_PERSIAN + + def test_keeps_first_attempt_when_retry_still_wrong(self): + def translate_fn(texts, hint): + return [ARABIC_FOR_PERSIAN] # la 2e tentative est tout aussi fausse + + result, fixed = retry_wrong_script( + [SOURCE_EN], [ARABIC_FOR_PERSIAN], "fa", translate_fn + ) + assert fixed == 0 + assert result[0] == ARABIC_FOR_PERSIAN + + def test_identity_output_never_triggers_retry(self): + def translate_fn(texts, hint): # pragma: no cover — ne doit pas servir + raise AssertionError("no retry expected for identity output") + + result, fixed = retry_wrong_script( + ["Microsoft 2024"], ["Microsoft 2024"], "fa", translate_fn + ) + assert fixed == 0 + assert result == ["Microsoft 2024"] + + def test_correct_translation_never_triggers_retry(self): + def translate_fn(texts, hint): # pragma: no cover + raise AssertionError("no retry expected for a correct translation") + + result, fixed = retry_wrong_script( + [SOURCE_EN], [GOOD_PERSIAN], "fa", translate_fn + ) + assert fixed == 0 + assert result == [GOOD_PERSIAN] + + def test_provider_error_keeps_originals(self): + def translate_fn(texts, hint): + raise RuntimeError("boom") + + result, fixed = retry_wrong_script( + [SOURCE_EN], [ARABIC_FOR_PERSIAN], "fa", translate_fn + ) + assert fixed == 0 + assert result == [ARABIC_FOR_PERSIAN] + + def test_only_offenders_are_resent(self): + sent = [] + + def translate_fn(texts, hint): + sent.extend(texts) + return [GOOD_PERSIAN] * len(texts) + + sources = [SOURCE_EN, "Second text", "Third text"] + translations = [ARABIC_FOR_PERSIAN, "متنی که کاملاً درست است", "سلام دوست من"] + # La 2e contient ک ? non — vérifions plutôt avec un persan valide + # pour montrer que seules les traductions fautives repartent. + translations[1] = "این متن کاملاً درست است" + translations[2] = "این هم درست است" + + result, fixed = retry_wrong_script(sources, translations, "fa", translate_fn) + assert sent == [SOURCE_EN] + assert fixed == 1 + + +# --------------------------------------------------------------------------- +# Intégration — le traducteur Word corrige un lot entier +# --------------------------------------------------------------------------- + + +class TestWordIntegration: + def test_word_retries_arabic_delivered_for_persian(self, tmp_path): + provider = SequencedMockProvider( + {SOURCE_EN: ARABIC_FOR_PERSIAN}, + {SOURCE_EN: GOOD_PERSIAN}, + ) + translator = WordTranslator(provider=provider) + + doc = Document() + doc.add_paragraph(SOURCE_EN) + input_file = tmp_path / "input.docx" + doc.save(input_file) + + output_file = tmp_path / "output.docx" + translator.translate_file(input_file, output_file, "fa") + + # Une demande initiale + une retentative avec consigne renforcée + non_retry = [c for c in provider.calls if not c[1]] + retry = [c for c in provider.calls if c[1]] + assert len(non_retry) == 1 + assert len(retry) == 1 + assert retry[0][0] == SOURCE_EN + + doc_out = Document(output_file) + assert GOOD_PERSIAN in doc_out.paragraphs[0].text + + def test_word_no_retry_for_latin_target(self, tmp_path): + provider = SequencedMockProvider( + {SOURCE_EN: "Bienvenue sur notre page"}, {} + ) + translator = WordTranslator(provider=provider) + + doc = Document() + doc.add_paragraph(SOURCE_EN) + input_file = tmp_path / "input.docx" + doc.save(input_file) + + output_file = tmp_path / "output.docx" + translator.translate_file(input_file, output_file, "fr") + + # Traduction latine correcte pour une cible latine : une seule demande + assert len(provider.calls) == 1 + assert not provider.calls[0][1] diff --git a/translators/excel_translator.py b/translators/excel_translator.py index 2804135..d1abb21 100644 --- a/translators/excel_translator.py +++ b/translators/excel_translator.py @@ -526,6 +526,24 @@ class ExcelTranslator: ) miss_pos += 1 + # Wrong-script guard: re-ask once (with a reinforced instruction) + # for translations delivered in the wrong alphabet — e.g. Arabic + # delivered for a Persian target. + from translators.segments import retry_wrong_script + + def _retry_translate(retry_texts, hint): + if self._provider is not None: + return self._translate_with_provider( + retry_texts, target_language, source_language, extra_prompt=hint + ) + return self._translate_with_legacy( + retry_texts, target_language, source_language + ) + + translated, _script_fixed = retry_wrong_script( + texts, translated, target_language, _retry_translate + ) + recorder = getattr(self, "_segment_recorder", None) if recorder is not None: recorder.record_pairs(zip(texts, translated)) @@ -539,7 +557,8 @@ class ExcelTranslator: return dict(self._translation_stats) def _translate_with_provider( - self, texts: List[str], target_language: str, source_language: str + self, texts: List[str], target_language: str, source_language: str, + extra_prompt: Optional[str] = None, ) -> List[str]: """Translate using the TranslationProvider.translate_batch() interface.""" from services.providers.base import TranslationProvider as NewTranslationProvider @@ -557,6 +576,10 @@ class ExcelTranslator: if is_new_style: from services.providers.schemas import TranslationRequest custom_prompt = getattr(self, "_custom_prompt", None) + if extra_prompt: + custom_prompt = ( + f"{custom_prompt}\n\n{extra_prompt}" if custom_prompt else extra_prompt + ) metadata = {"custom_prompt": custom_prompt} if custom_prompt else None requests = [ diff --git a/translators/pdf_translator.py b/translators/pdf_translator.py index 1dd2dd3..9ce19df 100644 --- a/translators/pdf_translator.py +++ b/translators/pdf_translator.py @@ -1702,7 +1702,8 @@ class PDFTranslator: # ------------------------------------------------------------------ # def _translate_with_provider( - self, texts: List[str], target_language: str, source_language: str + self, texts: List[str], target_language: str, source_language: str, + extra_prompt: Optional[str] = None, ) -> List[str]: """Translate using the TranslationProvider interface (handles old & new styles).""" from services.providers.base import TranslationProvider as NewTranslationProvider @@ -1720,6 +1721,10 @@ class PDFTranslator: if is_new_style: from services.providers.schemas import TranslationRequest custom_prompt = getattr(self, "_custom_prompt", None) + if extra_prompt: + custom_prompt = ( + f"{custom_prompt}\n\n{extra_prompt}" if custom_prompt else extra_prompt + ) metadata = {"custom_prompt": custom_prompt} if custom_prompt else None requests = [ @@ -1812,6 +1817,25 @@ class PDFTranslator: logger.warning("legacy_translate_failed", error=str(e)) translated = texts + # Wrong-script guard: re-ask once (with a reinforced instruction) + # for translations delivered in the wrong alphabet — e.g. Arabic + # delivered for a Persian target. + from translators.segments import retry_wrong_script + + def _retry_translate(retry_texts, hint): + if self._provider is not None: + return self._translate_with_provider( + retry_texts, target_language, source_language, extra_prompt=hint + ) + from services.translation_service import translation_service + return translation_service.translate_batch( + retry_texts, target_language, source_language + ) + + translated, _script_fixed = retry_wrong_script( + texts, translated, target_language, _retry_translate + ) + changed = sum(1 for orig, trans in zip(texts, translated) if orig != trans and trans.strip()) self._translation_stats["changed"] += changed diff --git a/translators/segments.py b/translators/segments.py index 7d687e4..3f2e77c 100644 --- a/translators/segments.py +++ b/translators/segments.py @@ -72,3 +72,117 @@ def apply_overrides( else: misses.append(i) return hits, misses + + +# --------------------------------------------------------------------------- +# Wrong-script retry — turn the script detector into an actionable guard +# --------------------------------------------------------------------------- + +def _script_retry_hint(target_language: str, reason: str) -> str: + """Extra instructions appended to the retry request's custom prompt. + + Tells the model, in its own working language (English), that the + previous attempt was rejected for the wrong script/variant and what + the output must look like this time. + """ + from core.languages import language_name + from services.quality import config as _qa_config + + name = language_name(target_language) or target_language + lines = [ + "CRITICAL RETRY: your previous attempt was rejected because the " + f"output was not written in {name}.", + f"Rejected output reason: {reason}", + f"Write the translation strictly in {name}.", + ] + target_chars = _qa_config.get_discriminating_chars( + (target_language or "").lower().split("-")[0] + ) + if target_chars: + sample = " ".join(sorted(target_chars)) + lines.append( + f"It must be {name}, not plain Arabic: use the letters specific " + f"to {name} ({sample}) wherever the words require them." + ) + return " ".join(lines) + + +def retry_wrong_script( + sources: List[str], + translations: List[str], + target_language: str, + translate_fn, +) -> Tuple[List[str], int]: + """Retry the translations delivered in the wrong script, once each. + + The most damaging silent failure of a translation pipeline is the + right language family with the wrong alphabet: Arabic delivered for + Persian looks almost correct to a non-reader and ships unnoticed. + This guard checks every delivered translation against the target + language's expected script (and, for Arabic-script languages, the + expected variant), and re-asks the model once for each offender with + an explicit instruction. A retry only replaces the first attempt when + it passes the same check; otherwise the first attempt is kept. + + translate_fn(texts, extra_prompt) must translate a list of texts in + one call, appending extra_prompt to the request's custom prompt. + + Returns (translations, number_of_replacements). Never raises: any + error in detection or retry keeps the original translations. + """ + if not translations or not target_language: + return translations, 0 + + try: + from services.quality.script_detector import script_issue + except Exception: + return translations, 0 + + # Only genuinely translated segments can be "wrong": identity output + # (numbers, proper names kept as-is) must not trigger a retry. + offenders = [] + reasons = [] + for i, (src, trans) in enumerate(zip(sources, translations)): + if not trans or trans.strip() == (src or "").strip(): + continue + try: + reason = script_issue(trans, target_language) + except Exception: + continue + if reason: + offenders.append(i) + reasons.append(reason) + + if not offenders: + return translations, 0 + + hint = _script_retry_hint(target_language, reasons[0]) + try: + retried = translate_fn([sources[i] for i in offenders], hint) + except Exception as e: + logger.warning("script_retry_failed", error=str(e)[:200], count=len(offenders)) + return translations, 0 + + fixed = 0 + result = list(translations) + for pos, i in enumerate(offenders): + if pos >= len(retried): + break + candidate = retried[pos] + if not candidate or not candidate.strip(): + continue + try: + still_wrong = script_issue(candidate, target_language) + except Exception: + still_wrong = True # do not swap in an unverifiable retry + if not still_wrong: + result[i] = candidate + fixed += 1 + + logger.info( + "script_retry", + target=target_language, + offenders=len(offenders), + fixed=fixed, + ) + return result, fixed diff --git a/translators/word_translator.py b/translators/word_translator.py index 196ffb3..6317dbc 100644 --- a/translators/word_translator.py +++ b/translators/word_translator.py @@ -763,6 +763,25 @@ class WordTranslator: ) miss_pos += 1 + # Wrong-script guard: re-ask once (with a reinforced instruction) + # for translations delivered in the wrong alphabet — e.g. Arabic + # delivered for a Persian target, which looks almost right to a + # non-reader and ships unnoticed. + from translators.segments import retry_wrong_script + + def _retry_translate(retry_texts, hint): + if self._provider is not None: + return self._translate_with_provider( + retry_texts, target_language, source_language, extra_prompt=hint + ) + return self._translate_with_legacy( + retry_texts, target_language, source_language, extra_prompt=hint + ) + + translated, _script_fixed = retry_wrong_script( + texts, translated, target_language, _retry_translate + ) + recorder = getattr(self, "_segment_recorder", None) if recorder is not None: recorder.record_pairs(zip(texts, translated)) @@ -776,7 +795,8 @@ class WordTranslator: return dict(self._translation_stats) def _translate_with_provider( - self, texts: List[str], target_language: str, source_language: str + self, texts: List[str], target_language: str, source_language: str, + extra_prompt: Optional[str] = None, ) -> List[str]: """Translate using the TranslationProvider.translate_batch() interface.""" from services.providers.base import TranslationProvider as NewTranslationProvider @@ -794,8 +814,12 @@ class WordTranslator: if is_new_style: from services.providers.schemas import TranslationRequest custom_prompt = getattr(self, "_custom_prompt", None) + if extra_prompt: + custom_prompt = ( + f"{custom_prompt}\n\n{extra_prompt}" if custom_prompt else extra_prompt + ) metadata = {"custom_prompt": custom_prompt} if custom_prompt else None - + requests = [ TranslationRequest( text=t, @@ -809,7 +833,7 @@ class WordTranslator: translated = [resp.translated_text for resp in responses] else: translated = self._provider.translate_batch(texts, target_language, source_language) - + # Fallback: keep original text for any empty/failed result return [ t if (t and t.strip()) else orig @@ -817,9 +841,15 @@ class WordTranslator: ] def _translate_with_legacy( - self, texts: List[str], target_language: str, source_language: str + self, texts: List[str], target_language: str, source_language: str, + extra_prompt: Optional[str] = None, ) -> List[str]: - """Fallback to legacy translation_service for backward compatibility.""" + """Fallback to legacy translation_service for backward compatibility. + + The legacy service has no custom-prompt parameter: extra_prompt is + accepted for signature parity but cannot be forwarded — the retry + still happens, just without the reinforced instruction. + """ from services.translation_service import translation_service _log_info(