office_translator/services/quality/qa_report.py

"""
Post-translation QA report (no external API — pure heuristics).

Answers three user-facing questions about a finished job:
  - Are numbers preserved? (digit-token multiset source vs translation)
  - Did everything actually get translated? (untranslated-ratio heuristic)
  - A 0-100 confidence score combining both.

Never blocks a job: every failure degrades to "skipped".
"""

import re
from pathlib import Path
from typing import Dict, List, Optional

from core.logging import get_logger

logger = get_logger(__name__)

# Tokens that are never counted as "content words" for the untranslated ratio
_PUNCT_RE = re.compile(r"[^\w\s]", re.UNICODE)
_WORD_RE = re.compile(r"[\w']+", re.UNICODE)
_NUM_RE = re.compile(r"\d+(?:[.,]\d+)*", re.UNICODE)

# Latin-script vs non-Latin word detection for language confusion heuristics
_LATIN_RE = re.compile(r"[a-zA-Z]")


def _extract_text_pairs(source_path: Path, output_path: Path, file_extension: str):
    """Extract (source_text, translated_text) full-document strings.

    Reuses the quality layer's file extractor so every format is read the
    same way the L0 check reads it. Applied to the INPUT file the same
    extractor yields the SOURCE text (the "translated" field simply holds
    whatever text lives in the file).
    """
    from services.quality.file_extractor import extract_sample

    src_chunks = extract_sample(Path(source_path), file_extension, max_samples=10_000)
    out_chunks = extract_sample(Path(output_path), file_extension, max_samples=10_000)
    src = "\n".join(c["translated"] for c in src_chunks)
    out = "\n".join(c["translated"] for c in out_chunks)
    return src, out


def _number_multiset(text: str) -> List[str]:
    """Digit tokens with the decimal separator normalized (12,50 == 12.50).

    French/English differ on ',' vs '.'; a real translation keeps the value.
    """
    return sorted(n.replace(",", ".") for n in _NUM_RE.findall(text))


def _number_fidelity(source: str, translated: str) -> Optional[dict]:
    """Compare digit tokens: how many source numbers survived (order-insensitive)."""
    src_nums = _number_multiset(source)
    if not src_nums:
        return None
    out_nums = _number_multiset(translated)
    # multiset intersection
    from collections import Counter

    src_count = Counter(src_nums)
    out_count = Counter(out_nums)
    kept = sum((src_count & out_count).values())
    return {
        "source_numbers": len(src_nums),
        "preserved": kept,
        "fidelity": round(kept / len(src_nums), 3),
    }


def _untranslated_ratio(source: str, translated: str) -> Optional[float]:
    """Heuristic: share of source content-words still present verbatim in
    the output. ~0 for a real translation, ~1 when nothing was translated.
    """
    src_words = [w.lower() for w in _WORD_RE.findall(source) if len(w) > 3]
    if len(src_words) < 10:
        return None
    out_lower = translated.lower()
    hits = sum(1 for w in set(src_words) if w in out_lower)
    return round(hits / len(set(src_words)), 3)


def run_qa_report(
    source_path: Path, output_path: Path, target_lang: str, file_extension: str
) -> Optional[Dict]:
    """Compute the QA report for a finished translation job.

    Returns a dict with numbers fidelity, untranslated ratio and a
    0-100 score, or None if the report could not be computed.
    """
    try:
        source, translated = _extract_text_pairs(
            Path(source_path), Path(output_path), file_extension
        )
    except Exception as e:
        logger.warning("qa_report_extract_failed", error=str(e))
        return None

    if not source.strip() or not translated.strip():
        return None

    numbers = _number_fidelity(source, translated)
    untranslated = _untranslated_ratio(source, translated)

    score = 100.0
    if numbers:
        score *= 0.5 + 0.5 * numbers["fidelity"]
    if untranslated is not None and untranslated > 0:
        score *= max(0.0, 1.0 - untranslated)

    report = {
        "score": int(round(score)),
        "numbers": numbers,
        "untranslated_ratio": untranslated,
    }
    logger.info("qa_report_computed", **{k: v for k, v in report.items() if v is not None})
    return report