fix(translation,#19023): dédup iit.csv sans réordonnancement — chantier 1 (3739 → 2575 lignes, diff en pure suppression)

scripts/translation/dedup_cells_csv.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""Deduplique les lignes d'un CSV de cellules translations (notebook, cell_id).
+
+Contexte (#19023) : `translations/iit/iit.csv` portait une duplication
+systemique preexistante (~1164 cles (notebook, cell_id) en double, toutes
+byte-identiques). La regen integrale (T1 --full) reecrit 85k lignes a cause
+du reordonnancement lexicographique ; ce outil deduplique SANS reordonner :
+premiere occurrence conservee, lignes suivantes supprimees.
+
+Invariants verifies par l'outil (echec = rc 1, fichier inchange) :
+- les lignes dupliquees sont byte-identiques (toute divergence = refus,
+  arbitrage manuel requis : quelle ligne est conforme au hash notebook) ;
+- l'ordre des carnets (sequence des premieres occurrences) est inchange ;
+- le round-trip csv est byte-identique sur les lignes conservees.
+
+Usage :
+    python scripts/translation/dedup_cells_csv.py translations/iit/iit.csv
+    python scripts/translation/dedup_cells_csv.py --check <csv>   # mesure seule, ecriture prohibee
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import io
+import sys
+from pathlib import Path
+
+KEY_COLUMNS = 2  # notebook, cell_id
+
+
+def serialize(rows: list[list[str]]) -> str:
+    buffer = io.StringIO(newline="")
+    csv.writer(buffer, lineterminator="\n").writerows(rows)
+    return buffer.getvalue()
+
+
+def dedup(rows: list[list[str]]) -> tuple[list[list[str]], int, list[str]]:
+    """Renvoie (lignes dedupliquees, nombre supprime, erreurs).
+
+    Les lignes conservees gardent leur ordre relatif ; chaque cle
+    (notebook, cell_id) garde sa premiere occurrence. Une cle dont les
+    occurrences divergent n'est PAS tranchee : erreur nommee.
+    """
+    header, body = rows[0], rows[1:]
+    kept: list[list[str]] = []
+    seen: dict[tuple[str, str], list[str]] = {}
+    removed = 0
+    errors: list[str] = []
+    for row in body:
+        key = (row[0], row[1])
+        if key not in seen:
+            seen[key] = row
+            kept.append(row)
+            continue
+        if row != seen[key]:
+            errors.append(
+                f"divergence pour {key[0]} cell_id={key[1]} : "
+                "occurrences non identiques, arbitrage manuel requis"
+            )
+        removed += 1
+    return [header, *kept], removed, errors
+
+
+def notebook_order(rows: list[list[str]]) -> list[str]:
+    order: list[str] = []
+    seen: set[str] = set()
+    for row in rows[1:]:
+        if row[0] not in seen:
+            seen.add(row[0])
+            order.append(row[0])
+    return order
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("csv_path", type=Path, help="CSV de cellules (ex. translations/iit/iit.csv)")
+    parser.add_argument(
+        "--check",
+        action="store_true",
+        help="mesurer seulement : aucune ecriture, rc 0 meme si des doublons existent",
+    )
+    args = parser.parse_args(argv)
+
+    with io.open(args.csv_path, encoding="utf-8", newline="") as handle:
+        raw = handle.read()
+    rows = list(csv.reader(io.StringIO(raw, newline="")))
+    deduped, removed, errors = dedup(rows)
+    order_before = notebook_order(rows)
+    order_after = notebook_order(deduped)
+
+    failures = list(errors)
+    if order_before != order_after:
+        failures.append("ordre des carnets modifie : refus")
+    if serialize(rows) != raw:
+        failures.append(
+            "round-trip csv non byte-identique sur ce fichier : refus "
+            "(le writer ne reproduit pas la serialisation en place)"
+        )
+
+    print(f"fichier: {args.csv_path}")
+    print(f"lignes: {len(rows) - 1} -> {len(deduped) - 1} ({removed} doublons supprimables)")
+    print(f"carnets: {len(order_after)} (ordre inchange: {order_before == order_after})")
+    for error in failures:
+        print(f"ERREUR: {error}", file=sys.stderr)
+
+    if failures:
+        return 1
+    if args.check or not removed:
+        return 0
+
+    rewritten = serialize(deduped)
+    if parse_rows_from_string(rewritten) != deduped:
+        print("ERREUR: relecture de la sortie diverge", file=sys.stderr)
+        return 1
+    with io.open(args.csv_path, "w", encoding="utf-8", newline="") as handle:
+        handle.write(rewritten)
+    print(f"OK: {removed} doublons supprimes, ordre preserve, ecrit {args.csv_path}")
+    return 0
+
+
+def parse_rows_from_string(text: str) -> list[list[str]]:
+    return list(csv.reader(io.StringIO(text, newline="")))
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/translation/tests/test_dedup_cells_csv.py

@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+"""Tests pour scripts/translation/dedup_cells_csv.py — outil de déduplication
+des lignes du CSV de cellules (#19023, chantier 1). L'invariant central : la
+dédup ne réordonne RIEN (ni carnets ni lignes conservées) et refuse tout
+fichier qu'elle ne peut pas relire byte-identique — la regen intégrale (T1
+--full) reste réservée au resync, pas au nettoyage.
+
+Couvre : dédup d'occurrences byte-identiques, refus sur divergence (arbitrage
+manuel), mode --check sans écriture, garde round-trip (fichier CRLF refusé),
+ordre des carnets inchangé. stdlib-only, hermétique (tmp_path).
+"""
+
+import csv
+import io
+import sys
+from pathlib import Path
+
+import pytest
+
+HERE = Path(__file__).resolve().parent
+TRANSLATION_DIR = HERE.parent
+sys.path.insert(0, str(TRANSLATION_DIR))
+
+import dedup_cells_csv  # noqa: E402
+
+HEADER = ["notebook", "cell_id", "cell_type", "src_lang", "src_hash"]
+
+
+def _row(nb: str, cid: str, extra: str = "x") -> list[str]:
+    return [nb, cid, "markdown", "fr", extra]
+
+
+def _write_csv(path: Path, rows: list[list[str]], lineterminator: str = "\n") -> None:
+    buffer = io.StringIO(newline="")
+    csv.writer(buffer, lineterminator=lineterminator).writerows(rows)
+    with io.open(path, "w", encoding="utf-8", newline="") as handle:
+        handle.write(buffer.getvalue())
+
+
+# --- dédup pure : occurrences identiques
+
+def test_dedup_identical_duplicates_keeps_first_and_order():
+    rows = [
+        HEADER,
+        _row("nb-a", "c1", "h1"),
+        _row("nb-a", "c2", "h2"),
+        _row("nb-a", "c1", "h1"),  # doublon exact, plus tard dans le fichier
+        _row("nb-b", "c1", "h3"),
+    ]
+    deduped, removed, errors = dedup_cells_csv.dedup(rows)
+    assert removed == 1
+    assert errors == []
+    assert [r[4] for r in deduped[1:]] == ["h1", "h2", "h3"]
+    assert dedup_cells_csv.notebook_order(rows) == dedup_cells_csv.notebook_order(deduped)
+
+
+def test_dedup_no_duplicates_is_noop():
+    rows = [HEADER, _row("nb-a", "c1"), _row("nb-a", "c2")]
+    deduped, removed, errors = dedup_cells_csv.dedup(rows)
+    assert removed == 0
+    assert errors == []
+    assert deduped == rows
+
+
+# --- dédup pure : divergence = refus nommé
+
+def test_dedup_divergent_duplicates_refuses():
+    rows = [
+        HEADER,
+        _row("nb-a", "c1", "ancien-hash"),
+        _row("nb-a", "c1", "nouveau-hash"),  # divergence : arbitrage manuel
+    ]
+    deduped, removed, errors = dedup_cells_csv.dedup(rows)
+    assert removed == 1
+    assert len(errors) == 1
+    assert "nb-a" in errors[0] and "arbitrage" in errors[0]
+    # la divergence ne tranche pas : première occurrence conservée dans la
+    # sortie, MAIS main() refusera l'écriture (voir test dédié)
+
+
+# --- notebook_order
+
+def test_notebook_order_first_occurrences_only():
+    rows = [HEADER, _row("nb-b", "c1"), _row("nb-a", "c1"), _row("nb-b", "c2")]
+    assert dedup_cells_csv.notebook_order(rows) == ["nb-b", "nb-a"]
+
+
+# --- main() : écriture réelle, --check, gardes
+
+def test_main_dedups_real_file(tmp_path):
+    path = tmp_path / "cells.csv"
+    _write_csv(path, [HEADER, _row("nb-a", "c1"), _row("nb-a", "c1"), _row("nb-b", "c1")])
+    rc = dedup_cells_csv.main([str(path)])
+    assert rc == 0
+    with io.open(path, encoding="utf-8", newline="") as handle:
+        rows = list(csv.reader(handle))
+    assert len(rows) == 3  # header + 2 lignes uniques
+    assert [r[0] for r in rows[1:]] == ["nb-a", "nb-b"]
+
+
+def test_main_check_mode_writes_nothing(tmp_path):
+    path = tmp_path / "cells.csv"
+    _write_csv(path, [HEADER, _row("nb-a", "c1"), _row("nb-a", "c1")])
+    before = path.read_bytes()
+    rc = dedup_cells_csv.main(["--check", str(path)])
+    assert rc == 0
+    assert path.read_bytes() == before
+
+
+def test_main_refuses_non_roundtrippable_file(tmp_path):
+    # fichier sérialisé en CRLF : le writer (LF) ne le reproduirait pas
+    # byte-identique → refus, fichier inchangé
+    path = tmp_path / "cells.csv"
+    _write_csv(path, [HEADER, _row("nb-a", "c1"), _row("nb-a", "c1")], lineterminator="\r\n")
+    before = path.read_bytes()
+    rc = dedup_cells_csv.main([str(path)])
+    assert rc == 1
+    assert path.read_bytes() == before
+
+
+def test_main_refuses_divergent_duplicates_without_write(tmp_path):
+    path = tmp_path / "cells.csv"
+    _write_csv(
+        path,
+        [HEADER, _row("nb-a", "c1", "ancien"), _row("nb-a", "c1", "nouveau")],
+    )
+    before = path.read_bytes()
+    rc = dedup_cells_csv.main([str(path)])
+    assert rc == 1
+    assert path.read_bytes() == before