diff --git a/scripts/notebook_tools/check_density_anchor.py b/scripts/notebook_tools/check_density_anchor.py new file mode 100644 index 0000000000..f17e39fa30 --- /dev/null +++ b/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()) diff --git a/scripts/notebook_tools/tests/test_check_density_anchor.py b/scripts/notebook_tools/tests/test_check_density_anchor.py new file mode 100644 index 0000000000..29b3cc1f27 --- /dev/null +++ b/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) == ""