feat(notebook-tools,#13410): garde d'ancrage des cellules de densite

scripts/notebook_tools/check_density_anchor.py

@@ -0,0 +1,219 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Garde d'ancrage des cellules de densite.
+
+Une cellule markdown qui **lit un resultat** (« Lecture du resultat »,
+« Interpretation », « On observe que... ») doit etre ancree sur une cellule de
+code qui porte un **output reel**. Une lecture posee sur un stub d'exercice non
+rempli est doublement fautive : elle commente un resultat qui n'existe pas, et
+elle divulgue la reponse a l'etudiant qui n'a pas encore fait l'exercice.
+
+Aucun organe ne verifiait ca : `validate_pr_notebooks.py` mesure
+`execution_count` et les outputs d'erreur, pas le **rattachement** d'une prose a
+son ancre. La regle `cell-interpretation-ordering` decrit le geste, ce script le
+mesure.
+
+Usage :
+
+    # sur des fichiers locaux
+    python scripts/notebook_tools/check_density_anchor.py notebook.ipynb ...
+
+    # sur le diff d'une PR (compare head vs base, ne juge que les AJOUTS)
+    python scripts/notebook_tools/check_density_anchor.py --pr 16492
+
+    # sortie machine
+    python scripts/notebook_tools/check_density_anchor.py --pr 16492 --json
+
+Code de sortie : 0 si aucun defaut, 1 si au moins une lecture est mal ancree,
+2 en cas d'erreur d'invocation.
+"""
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import os
+import re
+import subprocess
+import sys
+
+# Une cellule markdown est une « lecture » si elle annonce qu'elle commente un
+# resultat. Volontairement large : un faux positif coute une relecture, un faux
+# negatif laisse passer une divulgation.
+LECTURE = re.compile(
+    r"(lecture du r|interpr[ée]tation|ce que (montre|dit|nous apprend)"
+    r"|r[ée]sultat\s*:|on (lit|observe|constate|voit)|attendu\s*:)",
+    re.I,
+)
+
+# Marqueurs d'une cellule de code laissee a l'etudiant.
+STUB = re.compile(
+    r"(#\s*TODO|Exercice a completer|Exercice à compléter|#\s*Indice"
+    r"|votre code ici|your code here|#\s*Etape \d)",
+    re.I,
+)
+
+# En dessous de ce volume, un output est considere comme un vestige (un prompt,
+# une ligne de bruit), pas comme un resultat qu'on peut commenter.
+MIN_OUTPUT_BYTES = 200
+
+
+def output_size(cell: dict) -> int:
+    """Volume total de sortie d'une cellule, texte et donnees riches confondus."""
+    total = 0
+    for out in cell.get("outputs", []):
+        total += len("".join(out.get("text", [])))
+        total += len(json.dumps(out.get("data", {}), ensure_ascii=False))
+    return total
+
+
+def _run(args: list[str]) -> str:
+    proc = subprocess.run(args, capture_output=True, text=True,
+                          encoding="utf-8", errors="replace")
+    if proc.returncode != 0:
+        raise RuntimeError((proc.stderr or "").strip()[:200])
+    return proc.stdout
+
+
+class BodyNotServed(Exception):
+    """L'API a repondu 200 sans porter le fichier ; l'argument est le sha du blob."""
+
+
+def decode_payload(payload: dict) -> list[dict]:
+    """Cellules d'un notebook depuis une reponse `contents` **ou** `blobs`.
+
+    L'API `contents` plafonne a 1 Mo : au-dela elle rend `200` avec
+    `content: ""` et `encoding: "none"`. C'est un succes HTTP qui ne porte pas
+    le fichier — le detecter ici evite un `JSONDecodeError` opaque a la ligne
+    suivante, et rend le sha qui permet de rattraper par l'API blobs (100 Mo).
+    """
+    content = (payload.get("content") or "").strip()
+    if not content:
+        raise BodyNotServed(payload.get("sha") or "")
+    raw = base64.b64decode(content).decode("utf-8", "replace")
+    return json.loads(raw)["cells"]
+
+
+def cells_at_ref(repo: str, path: str, ref: str) -> list[dict] | None:
+    """Cellules d'un notebook a une reference donnee, gros fichiers compris."""
+    try:
+        payload = json.loads(_run(["gh", "api",
+                                   "repos/%s/contents/%s?ref=%s" % (repo, path, ref)]))
+    except RuntimeError:
+        return None
+
+    try:
+        return decode_payload(payload)
+    except BodyNotServed as exc:
+        sha = str(exc)
+
+    # Blob > 1 Mo : l'API git/blobs le sert jusqu'a 100 Mo.
+    if not sha:
+        return None
+    try:
+        return decode_payload(json.loads(_run(["gh", "api",
+                                               "repos/%s/git/blobs/%s" % (repo, sha)])))
+    except (RuntimeError, BodyNotServed, KeyError, ValueError):
+        return None
+
+
+def audit(head_cells: list[dict], base_cells: list[dict] | None) -> list[dict]:
+    """Defauts d'ancrage parmi les cellules de lecture AJOUTEES.
+
+    `base_cells` a None fait auditer toutes les lectures du notebook (mode
+    fichier local) ; sinon seules les cellules absentes de la base sont jugees,
+    ce qui borne le verdict au diff.
+    """
+    known = set()
+    if base_cells is not None:
+        known = {"".join(c["source"]) for c in base_cells}
+
+    findings = []
+    for i, cell in enumerate(head_cells):
+        if cell["cell_type"] != "markdown":
+            continue
+        source = "".join(cell["source"])
+        if source in known or not LECTURE.search(source):
+            continue
+
+        # L'ancre est la cellule de code immediatement precedente.
+        j = i - 1
+        while j >= 0 and head_cells[j]["cell_type"] != "code":
+            j -= 1
+        if j < 0:
+            findings.append({"cell": i, "anchor": None,
+                             "reason": "lecture sans aucune cellule de code en amont"})
+            continue
+
+        anchor = head_cells[j]
+        size = output_size(anchor)
+        is_stub = bool(STUB.search("".join(anchor["source"])))
+        if size == 0:
+            findings.append({
+                "cell": i, "anchor": j, "output_bytes": 0, "stub": is_stub,
+                "reason": "ancre sans aucun output" + (" (stub d'exercice)" if is_stub else ""),
+            })
+        elif is_stub and size < MIN_OUTPUT_BYTES:
+            findings.append({
+                "cell": i, "anchor": j, "output_bytes": size, "stub": True,
+                "reason": "ancre = stub d'exercice, output residuel de %d octets" % size,
+            })
+    return findings
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__.split("\n")[0])
+    parser.add_argument("notebooks", nargs="*", help="fichiers .ipynb locaux")
+    parser.add_argument("--pr", type=int, help="numero de PR : n'audite que les cellules ajoutees")
+    parser.add_argument("--repo", default="jsboige/CoursIA")
+    parser.add_argument("--json", action="store_true", dest="as_json")
+    args = parser.parse_args(argv)
+
+    if not args.notebooks and args.pr is None:
+        parser.error("donner des notebooks ou --pr")
+
+    results = []
+
+    if args.pr is not None:
+        meta = json.loads(_run(["gh", "pr", "view", str(args.pr), "-R", args.repo,
+                                "--json", "headRefOid,baseRefOid,files"]))
+        paths = [f["path"] for f in meta["files"] if f["path"].endswith(".ipynb")]
+        for path in paths:
+            head = cells_at_ref(args.repo, path, meta["headRefOid"])
+            if head is None:
+                results.append({"path": path, "error": "notebook illisible au head (blob non servi par contents ni blobs)"})
+                continue
+            base = cells_at_ref(args.repo, path, meta["baseRefOid"]) or []
+            results.append({"path": path, "findings": audit(head, base)})
+
+    for path in args.notebooks:
+        with open(path, encoding="utf-8") as handle:
+            results.append({"path": path,
+                            "findings": audit(json.load(handle)["cells"], None)})
+
+    failed = any(r.get("findings") for r in results)
+
+    if args.as_json:
+        print(json.dumps({"ok": not failed, "results": results},
+                         ensure_ascii=False, indent=2))
+    else:
+        for result in results:
+            name = os.path.basename(result["path"])
+            if "error" in result:
+                print("??  %s : %s" % (name, result["error"]))
+            elif not result["findings"]:
+                print("OK  %s" % name)
+            else:
+                print("!!  %s" % name)
+                for f in result["findings"]:
+                    anchor = "c%s" % f["anchor"] if f["anchor"] is not None else "-"
+                    print("      cellule %-4s ancre %-5s %s" % (f["cell"], anchor, f["reason"]))
+        if failed:
+            print("\nUne cellule de lecture doit suivre une cellule de code qui a REELLEMENT produit\n"
+                  "le resultat commente. Sur un stub d'exercice, elle divulgue la reponse.")
+
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/notebook_tools/tests/test_check_density_anchor.py

@@ -0,0 +1,143 @@
+# -*- coding: utf-8 -*-
+"""Tests du garde d'ancrage des cellules de densite.
+
+Les deux premiers tests sont les **controles positifs** : un garde qui ne peut
+pas echouer ne prouve rien quand il rend vert.
+"""
+import base64
+import json
+import sys
+
+import pytest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from check_density_anchor import (  # noqa: E402
+    MIN_OUTPUT_BYTES,
+    BodyNotServed,
+    audit,
+    decode_payload,
+    output_size,
+)
+
+
+def md(text):
+    return {"cell_type": "markdown", "source": [text]}
+
+
+def code(text, out_text=None):
+    cell = {"cell_type": "code", "source": [text], "outputs": [], "execution_count": 1}
+    if out_text is not None:
+        cell["outputs"] = [{"output_type": "stream", "text": [out_text]}]
+    return cell
+
+
+# --- controles positifs : le garde DOIT mordre ---------------------------------
+
+def test_mord_sur_lecture_ancree_sur_stub_sans_output():
+    cells = [code("# TODO : votre code ici\npass"),
+             md("### Lecture du resultat\nOn observe une convergence nette.")]
+    findings = audit(cells, base_cells=[])
+    assert len(findings) == 1
+    assert findings[0]["anchor"] == 0
+    assert findings[0]["stub"] is True
+    assert "stub" in findings[0]["reason"]
+
+
+def test_mord_sur_stub_a_output_residuel():
+    cells = [code("# TODO : completer\nprint('a faire')", out_text="a faire\n"),
+             md("### Interpretation\nLe score atteint 0.94.")]
+    findings = audit(cells, base_cells=[])
+    assert len(findings) == 1
+    assert 0 < findings[0]["output_bytes"] < MIN_OUTPUT_BYTES
+
+
+def test_mord_sur_lecture_sans_aucune_cellule_de_code_amont():
+    findings = audit([md("### Lecture du resultat\nvoir plus haut.")], base_cells=[])
+    assert len(findings) == 1
+    assert findings[0]["anchor"] is None
+
+
+# --- verts legitimes -----------------------------------------------------------
+
+def test_vert_quand_l_ancre_porte_un_vrai_output():
+    cells = [code("print(score)", out_text="x" * (MIN_OUTPUT_BYTES + 50)),
+             md("### Lecture du resultat\nLe score atteint 0.94.")]
+    assert audit(cells, base_cells=[]) == []
+
+
+def test_vert_sur_un_stub_sans_cellule_de_lecture():
+    """Un stub non commente est normal : c'est un exercice."""
+    cells = [code("# TODO : votre code ici\npass"),
+             md("## Exercice 2\nCompleter la fonction ci-dessus.")]
+    assert audit(cells, base_cells=[]) == []
+
+
+def test_ignore_les_cellules_deja_presentes_dans_la_base():
+    """Le verdict est borne au diff : une lecture preexistante n'est pas jugee."""
+    lecture = "### Lecture du resultat\nconvergence nette."
+    cells = [code("# TODO\npass"), md(lecture)]
+    assert audit(cells, base_cells=[md(lecture)]) == []
+
+
+def test_ancre_saute_les_markdown_intercales():
+    cells = [code("print(x)", out_text="y" * 400),
+             md("Transition sans lecture."),
+             md("### Interpretation\nLe resultat est stable.")]
+    assert audit(cells, base_cells=[]) == []
+
+
+def test_output_size_compte_les_donnees_riches():
+    cell = {"cell_type": "code", "source": [""],
+            "outputs": [{"output_type": "display_data",
+                         "data": {"image/png": "iVBOR" + "A" * 300}}]}
+    assert output_size(cell) > MIN_OUTPUT_BYTES
+
+
+def test_notebook_sans_cellule_de_lecture_est_vert():
+    cells = [code("import numpy", out_text="ok"), md("## Titre")]
+    assert audit(cells, base_cells=[]) == []
+
+
+def test_serialisation_json_du_verdict():
+    cells = [code("# TODO\npass"), md("### Lecture du resultat\nfoo.")]
+    json.dumps(audit(cells, base_cells=[]))
+
+
+# --- corps non servi : le defaut reproduit sur #16613 (notebook de 6,1 Mo) ------
+
+def _payload(cells, sha="abc123"):
+    body = json.dumps({"cells": cells}).encode("utf-8")
+    return {"sha": sha, "content": base64.b64encode(body).decode("ascii")}
+
+
+def test_decode_payload_lit_un_corps_normal():
+    assert decode_payload(_payload([md("titre")])) == [md("titre")]
+
+
+def test_decode_payload_signale_le_corps_vide_et_rend_le_sha():
+    """Controle positif : au-dela de 1 Mo, contents rend 200 avec content vide.
+
+    Avant ce garde, le `json.loads` de la ligne suivante levait un
+    `JSONDecodeError` opaque et tuait le mode --pr au premier gros notebook.
+    """
+    with pytest.raises(BodyNotServed) as exc:
+        decode_payload({"sha": "deadbeef", "content": "", "encoding": "none"})
+    assert str(exc.value) == "deadbeef"
+
+
+def test_decode_payload_tolere_le_base64_multiligne_de_l_api_blobs():
+    """L'API git/blobs rend son base64 decoupe en lignes ; contents non."""
+    body = json.dumps({"cells": [md("x")]}).encode("utf-8")
+    wrapped = base64.encodebytes(body).decode("ascii")  # decoupe en lignes
+    assert chr(10) in wrapped
+    assert decode_payload({"sha": "s", "content": wrapped}) == [md("x")]
+
+
+def test_decode_payload_sans_sha_rend_une_chaine_vide():
+    """Sans sha, le rattrapage par l'API blobs est impossible : le code appelant
+    doit pouvoir le distinguer d'un sha valide."""
+    with pytest.raises(BodyNotServed) as exc:
+        decode_payload({"content": ""})
+    assert str(exc.value) == ""