Compare commits

...
7 Commits
Author SHA1 Message Date
sebastien b9e9d00e7d Restructuration de l'application 2026-08-22 13:39:20 +02:00
sebastien dc25b5fefa Splitting of Readme.org 2026-08-22 13:17:54 +02:00
sebastien 5a7ddd407f Reuse enonce groups if availabel in annotating_by_label 2026-08-22 12:22:46 +02:00
sebastien 92a9f9883e extension de export/import aux autres modes de génération 2026-08-22 12:14:20 +02:00
sebastien 9af05f2c86 Améliorations GUI 2026-08-22 11:55:17 +02:00
sebastien ce9f385a0a Untrack binaries 2026-08-22 11:50:20 +02:00
sebastien 81dc658640 Améliorations GUI 2026-08-22 11:49:13 +02:00
51 changed files with 1751 additions and 9144 deletions
+115
View File
@@ -0,0 +1,115 @@
#+title: Architecture et conventions de développement
#+author: Sébastien Miquel
#+date: 22-08-2026
#+OPTIONS:
Ce document décrit les invariants techniques du projet. Il s'adresse
aux développeurs et aux agents logiciels qui modifient le dépôt.
- [[file:Readme.org][Guide de démarrage]]
- [[file:Script.org][Référence des étapes et des scripts]]
* API commune pour les scripts
Le paquet =copienator= centralise les chemins d'une évaluation et les
écritures JSON sûres. Un script ne devrait donc plus reconstruire les
chemins partagés à la main :
#+BEGIN_SRC python
from copienator import EvaluationWorkspace, atomic_write_json
workspace = EvaluationWorkspace("Interro")
atomic_write_json(workspace.correction_file, corrections)
#+END_SRC
=EvaluationWorkspace.discover(path)= retrouve également la racine d'une
évaluation à partir d'un fichier ou d'un sous-dossier. Sa construction
ne crée aucun fichier. La création explicite de
=.copienator/logs/= et =.copienator/runs/= se fait avec
=workspace.ensure_control_directories()=. Le chemin
=.copienator/state.sqlite3= est réservé à une future couche d'état
transactionnelle.
=atomic_write_json= écrit d'abord dans un fichier temporaire situé dans
le même dossier, synchronise son contenu, puis remplace la destination.
Une interruption ne laisse donc pas un JSON partiellement écrit. Pour
une modification concurrente de type lire-modifier-écrire, utiliser
=atomic_update_json= : cet utilitaire protège l'opération complète avec
un verrou inter-processus Linux/Windows.
* Convention des scripts standardisés
Les scripts standardisés exposent =build_parser()=, =run(...)= et
=main(argv=None)=. Leur import ne lance aucun traitement. Ils acceptent
le dossier d'évaluation comme premier argument positionnel, utilisent
=EvaluationWorkspace= pour les chemins partagés et peuvent afficher la
trace complète d'une erreur avec =--verbose=.
Les codes de sortie communs sont :
| Code | Signification |
|------+---------------|
| 0 | réussite |
| 1 | erreur de traitement |
| 2 | arguments invalides |
| 3 | évaluation ou prérequis invalides |
| 4 | traitement partiel, avec avertissements |
| 130 | interruption par l'utilisateur |
Le GUI distingue notamment un traitement partiel d'un échec. Toutes les
commandes du paquet suivent désormais cette convention. Elles sont
regroupées dans =copienator.commands= et exposées par le répartiteur
=python -m copienator=.
* Organisation du code
- Le paquet =copienator= contient les abstractions partagées par les
scripts, notamment =EvaluationWorkspace= et les écritures atomiques.
- Le paquet =copienator_gui= contient la définition du workflow, l'état
persistant, le lanceur de processus et l'application Tk.
- Le sous-paquet =copienator.commands= contient les points d'entrée.
Leur import ne doit pas déclencher de traitement.
- =copienator.dispatcher= expose le CLI unifié et =copienator_gui=
contient l'application Tk.
* État et exécution du GUI
L'état d'une évaluation est conservé dans =.copienator-gui.json=. Les
valeurs des arguments, le dernier mode exécuté, les statuts et un
historique borné y sont enregistrés atomiquement. Les sorties des
processus sont conservées dans =.copienator/logs/=. Le GUI lance les
scripts dans des processus séparés afin de pouvoir afficher leur sortie,
leur transmettre une saisie et les interrompre.
Les retours en arrière ne suppriment pas les artefacts. Ils marquent les
étapes suivantes comme étant à revalider. Les automatismes de première
visite sont également persistés et ne doivent pas être rejoués lors d'un
retour en arrière.
* Invariants à préserver
- Utiliser =EvaluationWorkspace= pour les chemins partagés d'une
évaluation au lieu de reconstruire ces chemins dans chaque script.
- Publier les JSON et les ensembles de fichiers importants de façon
atomique. Une interruption ou un résultat partiel doit conserver la
dernière sortie complète lorsqu'elle existe.
- Conserver la convention =build_parser()=, =run(...)= et
=main(argv=None)= pour les scripts standardisés.
- Ne lancer aucun traitement lors de l'import d'un module.
- Retourner les codes de sortie communs afin que le GUI distingue une
réussite, un résultat partiel, un échec et une interruption.
- Préserver les modifications manuelles et les fichiers déjà présents,
sauf lorsqu'une option explicite comme =--overwrite= ou =--reset=
autorise leur remplacement.
- Les labels deviennent des noms de fichiers. Ils doivent donc respecter
les contraintes de la plateforme, notamment l'interdiction de =:= sous
Windows.
* Règle de classement de la documentation
- Une information nécessaire pour installer ou démarrer l'application
appartient à [[file:Readme.org][Readme.org]].
- Une commande, un argument ou une étape du workflow appartient à
[[file:Script.org][Script.org]].
- Une convention interne, un invariant ou une décision d'architecture
appartient au présent document.
+37 -447
View File
@@ -1,10 +1,10 @@
#+title: Script #+title: Copienator
#+author: Sébastien Miquel #+author: Sébastien Miquel
#+date: 14-03-2026 #+date: 14-03-2026
# Time-stamp: <20-08-26 11:37> # Time-stamp: <22-08-26 12:22>
#+OPTIONS: #+OPTIONS:
* Méta * Présentation
** Quézaco ** Quézaco
Ce dépôt contient un certain nombre de script Python que j'utilise Ce dépôt contient un certain nombre de script Python que j'utilise
@@ -21,23 +21,6 @@ pour faire corriger des copies par Gemini.
4. Ces annotations manuscrites sont lues et recompilées en une 4. Ces annotations manuscrites sont lues et recompilées en une
version de la copie pour l'élève. version de la copie pour l'élève.
** Disclaimer
J'utilise régulièrement cet outil et j'en suis satisfait, mais j'ai
fait peu d'efforts pour le rendre universel et simple à l'emploi.
Plusieurs parties de mon workflow sont spécifiques à mes
représentations internes des sujets d'examens et/ou à mon environment.
L'utilisation de ce système nécessite une familiarité avec python et
la ligne de commande unix ; cette familiarité n'est probablement pas
suffisante en l'état, mais en théorie il devrait être possible de le
faire tourner sur le dossier =Example= qui contient une copie
initiale, les fichiers correspondant au sujet de l'interro, et des
examples du rendu final (dans le sous dossier =BGnot=).
Cette situation s'améliorera peut-être, mais faciliter l'utilisation
de ce système n'est pas une priorité.
** Limitations ** Limitations
Pour l'instant, la correction est faite question par question : le LLM Pour l'instant, la correction est faite question par question : le LLM
@@ -47,13 +30,8 @@ question précédente ou autre.
*** Noms de labels sous Windows *** Noms de labels sous Windows
Certains labels sont utilisés directement comme noms de fichiers et de Les labels de questions sont utilisés comme noms de fichiers. Les
dossiers. Les labels contenant notamment =:= ne sont pas acceptés par labels contenant notamment =:= ne sont pas acceptés par Windows.
Windows. Sous Windows, les scripts refusent ces labels avant de
poursuivre et affichent la liste des valeurs à corriger. Aucune
conversion silencieuse n'est appliquée ; utiliser par exemple
=Ex 1 - a)= au lieu de =Ex 1 : a)=. Le diagnostic du GUI permet de les
repérer avant de lancer une étape.
** Requirements ** Requirements
@@ -63,8 +41,13 @@ Libraries :
#+BEGIN_SRC bash #+BEGIN_SRC bash
pip install numpy pandas matplotlib pillow pydantic pypdf pdf2image reportlab img2pdf pymupdf ftfy ezodf google pip install numpy pandas matplotlib pillow pydantic pypdf pdf2image reportlab img2pdf pymupdf ftfy ezodf google
pip install -e .
#+END_SRC #+END_SRC
La seconde commande installe les exécutables =copienator= et
=copienator-gui=. Sans installation, les mêmes commandes restent
accessibles avec =python -m copienator=.
*** Poppler (for pdf2image) *** Poppler (for pdf2image)
+ Linux : install poppler-utils + Linux : install poppler-utils
@@ -75,9 +58,6 @@ pip install numpy pandas matplotlib pillow pydantic pypdf pdf2image reportlab im
Il faut créer une clef API pour Gemini (pas facile). Il faut créer une clef API pour Gemini (pas facile).
NB : Lors de la création, google offre (offrait ?) 300€ d'utilisation,
mais seulement pendant les trois mois à venir.
Puis ajouter =GEMINI_API_KEY= à l'environnement avec : Puis ajouter =GEMINI_API_KEY= à l'environnement avec :
#+BEGIN_SRC bash #+BEGIN_SRC bash
@@ -91,23 +71,37 @@ ou éventuellement, la renseigner directement dans le fichier
Copier `default_config.py` en `config.py`. Éventuellement le modifier. Copier `default_config.py` en `config.py`. Éventuellement le modifier.
*** Interface graphique ** Correction d'un paquet de copies
1. Créer un fichier =names= dans le dossier courant, avec les
noms/prénoms des élèves, un par ligne
2. Créer un dossier correspondant à l'évaluation (=Interro= dans la
suite)
3. Suivre les instructions suivantes
** Interface graphique
Lancer l'assistant avec : Lancer l'assistant avec :
#+BEGIN_SRC bash #+BEGIN_SRC bash
python gui.py python -m copienator gui
#+END_SRC #+END_SRC
On peut aussi ouvrir directement une évaluation avec On peut aussi ouvrir directement une évaluation avec =python -m copienator gui
=python gui.py Interro=. L'interface conserve l'état et l'historique Interro=. L'interface conserve l'état et l'historique des étapes dans
des étapes dans =Interro/.copienator-gui.json=, et les sorties complètes =Interro/.copienator-gui.json=, et les sorties complètes dans
dans =Interro/.copienator/logs/=. Une relance d'une étape antérieure ne =Interro/.copienator/logs/=. Une relance d'une étape antérieure ne
supprime aucun résultat ; les étapes suivantes sont seulement marquées supprime aucun résultat ; les étapes suivantes sont seulement marquées
comme étant à revalider. comme étant à revalider.
La variable =SHOW_PERSONAL_STEPS= de =config.py= permet d'afficher ou de Après la réussite d'un script, l'interface sélectionne automatiquement
masquer les étapes propres au workflow personnel. l'étape suivante. =Découper les réponses par question= et =Regrouper
les réponses= démarrent automatiquement lors de leur première visite.
Avec le mode =Correction immédiate=, les trois étapes batch sont alors
marquées =Ignorée= à leur première visite. L'étape de résolution
manuelle est traitée de la même façon si =manual_resolutions.txt= est
absent, vide ou ne contient que des commentaires. Ces automatismes ne
se répètent pas lors d'un retour en arrière.
Le bouton =Diagnostic…= vérifie les modules Python, Poppler, LaTeX, Le bouton =Diagnostic…= vérifie les modules Python, Poppler, LaTeX,
PDF Arranger et la configuration Gemini. Sous Windows, les exécutables PDF Arranger et la configuration Gemini. Sous Windows, les exécutables
@@ -119,413 +113,9 @@ Les chemins des étapes personnelles peuvent être adaptés avec
=CURRENT_SCORE_ODS_PATH=, =FINAL_SCORE_ODS_PATH=, =CURRENT_SCORE_ODS_PATH=, =FINAL_SCORE_ODS_PATH=,
=FINAL_SCORE_OUTPUT_DIR= et =FINAL_SCORE_FONT_PATH= dans =config.py=. =FINAL_SCORE_OUTPUT_DIR= et =FINAL_SCORE_FONT_PATH= dans =config.py=.
*** API commune pour les scripts * Documentation complémentaire
Le paquet =copienator= centralise les chemins d'une évaluation et les - [[file:Script.org][Référence des étapes et des scripts]] : commandes, arguments,
écritures JSON sûres. Un script ne devrait donc plus reconstruire les prérequis, fichiers produits et parcours alternatifs.
chemins partagés à la main : - [[file:Architecture.org][Architecture et conventions de développement]] : API commune,
état partagé, écritures sûres et règles de maintenance.
#+BEGIN_SRC python
from copienator import EvaluationWorkspace, atomic_write_json
workspace = EvaluationWorkspace("Interro")
atomic_write_json(workspace.correction_file, corrections)
#+END_SRC
=EvaluationWorkspace.discover(path)= retrouve également la racine d'une
évaluation à partir d'un fichier ou d'un sous-dossier. Sa construction
ne crée aucun fichier. La création explicite de
=.copienator/logs/= et =.copienator/runs/= se fait avec
=workspace.ensure_control_directories()=. Le chemin
=.copienator/state.sqlite3= est réservé à une future couche d'état
transactionnelle.
=atomic_write_json= écrit d'abord dans un fichier temporaire situé dans
le même dossier, synchronise son contenu, puis remplace la destination.
Une interruption ne laisse donc pas un JSON partiellement écrit. Pour
une modification concurrente de type lire-modifier-écrire, utiliser
=atomic_update_json= : cet utilitaire protège l'opération complète avec
un verrou inter-processus Linux/Windows.
*** Convention des scripts standardisés
Les scripts standardisés exposent =build_parser()=, =run(...)= et
=main(argv=None)=. Leur import ne lance aucun traitement. Ils acceptent
le dossier d'évaluation comme premier argument positionnel, utilisent
=EvaluationWorkspace= pour les chemins partagés et peuvent afficher la
trace complète d'une erreur avec =--verbose=.
Les codes de sortie communs sont :
| Code | Signification |
|------+---------------|
| 0 | réussite |
| 1 | erreur de traitement |
| 2 | arguments invalides |
| 3 | évaluation ou prérequis invalides |
| 4 | traitement partiel, avec avertissements |
| 130 | interruption par l'utilisateur |
Le GUI distingue notamment un traitement partiel d'un échec. Les
scripts migrés vers cette convention sont actuellement :
- =copies_tools.py=, =grouping.py= et =verify_groups.py= ;
- =post-correction.py= et =resolve_manual.py= ;
- =page_splitter.py=, =cutleft.py=, =plotting.py= et
=splitting_int.py= ;
- =gemini_for_labels.py= ;
- =gemini_for_enonce.py= et =enonce_info.py= ;
- =correction.py=, =submit_batches.py=, =batch_status.py= et
=fetch_batched_results.py= ;
- =annotating.py=, =annotating_with_checks.py= et
=annotating_by_label.py= ;
- =reading_annotations.py= et =reading_grouped_annotations.py= ;
- =export.py=, =import.py= et =giving_names.py=.
** Correction d'un paquet de copies
1. Créer un fichier =names= dans le dossier courant, avec les
noms/prénoms des élèves, un par ligne
2. Créer un dossier correspondant à l'évaluation (=Interro= dans la
suite)
3. Suivre les instructions suivantes
* Étapes et Script
Utiliser `python gui.py` ou `python gui.py Interro` pour lancer un GUI
qui suit automatiquement les étapes décrites ci-dessous.
** Prétraitement de l'énoncé
Dans le dossier de l'évaluation, mettre les fichiers suivants de l'évaluation :
`enonce.pdf`, `enonce.tex`, `correction.tex`.
- `python gemini_for_enonce.py Interro` or
`python gemini_for_enonce.py Interro --restart`
À partir des trois fichiers précédents, se charge de détecter les
labels des questions et leur contenu.
Les questions vont également être regroupées. Par la suite, quand
des requêtes de corrections seront effectuées sur une question,
seulement les énoncés des questions du groupe seront envoyés (et le
corrigé de la question). Il faut donc que chaque groupe contienne
si possible le contexte nécessaire pour comprendre la question.
Une fenêtre s'ouvre pour permettre d'éditer le résultat. Ne pas
hésiter à faire des groupes plus gros que les groupes par défaut.
Après relecture le script génère :
+ un fichier `labels` avec les labels des questions
+ Un dossier `Text` avec le contenu textuel des questions,
regroupées.
+ Un dossier `Sol` avec le contenu textuel du corrigé, question par
question.
+ Un dossier `Text2`, qui compile un fichier `.tex` pour chaque
question (utilisé pour compiler un rendu pdf du corrigé pour
chaque question)
+ Un dossier `Sol2`, qui compile un fichier `.tex` pour chaque
correction de chaque question.
+ Un dossier `Persp` avec des instruction de barème pour chaque
question.
Éventuellement : vérifier et modifier les barèmes dans `Persp`.
- Alternative personnelle : `python enonce_info.py Interro`
Ces deux commandes suivent la convention des scripts standardisés.
Leur import ne lance aucun traitement et les erreurs partielles sont
distinguées des échecs. Les réponses d'extraction mises en cache par
=gemini_for_enonce.py= et le fichier =labels= sont publiés
atomiquement.
** Prétraitement des copies
Mettre les copies scannées au format pdf dans =Interro=.
1. =python copies_tools.py rotate Interro= (facultatif)
Retourne tous les pdf de 180°, si la photocopie a été faite à
l'envers.
2. =python copies_tools.py rename Interro=
change le nom des copies en =Copie{id}.pdf=
3. =python page_splitter.py Interro=
Découpe des copies A3 en pages A4, en retirant les pages vides.
Pour chaque page double il est possible
+ de garder les deux pages
+ de ne garder qu'une des deux pages.
+ de jeter les deux pages
+ de déplacer la délimitation à droite/gauche
Fix issues with =python page_splitter.py Interro14/Copies/Copie01.pdf=
Le PDF transformé est construit dans un dossier temporaire. La
copie produite et la sauvegarde dans =Copies Originales= sont
ensuite installées avec rollback : une erreur conserve les deux
versions précédentes. Une relance ciblée lit directement la
sauvegarde originale sans la déplacer au préalable.
4. =python cutleft.py Interro=
Découpe la partie gauche des copies, là où il devrait y avoir les
labels des exercices/questions.
=python cutleft.py Interro --fullpage= to use the full page always.
Rerun on a single file with =python cutleft.py Interro/Copies/Copie01.pdf=
Les images et le fichier =_schema.json= d'une copie sont remplacés
ensemble. Fermer l'outil juste après la dernière validation ne peut
donc plus interrompre un thread de sauvegarde en arrière-plan.
** Labelisation et regroupement
Optional : Set proxy with ~export HTTPS_PROXY="http://10.0.0.1:3128"~
1. =python gemini_for_labels.py Interro=, avec éventuellement =--overwrite=
Fait des requêtes à Gemini pour identifier les labels des
questions dans images générées à partir des parties gauches des copies.
Une copie PDF ou une image précise de =Cutleft= peut également être
ciblée. Plusieurs cibles de la même évaluation sont acceptées. Les
parties d'une copie restent traitées séquentiellement afin de
conserver les labels précédents comme contexte, tandis que les
copies différentes sont traitées en parallèle. Chaque réponse JSON
validée est écrite atomiquement. Une cible sans image correspondante
produit le code de sortie 4.
2. =python plotting.py Interro=
Permet de vérifier visuellement les labels trouvés.
+ Sous linux, on peut faire =e= pour ouvrir le fichier .json et
l'éditer a la main.
+ Quand un label est manquant, il est possible de cliquer sur
l'image, ce qui copie les coordonnées dans le presse papier
(sous linux…), puis on peut l'ajouter à la main.
+ Utilisation de `_`, `|…` et `…|` :
+ `|…` n'est pas arrêté verticalement par son type opposé.
+ `…|` est stoppé horizontalement par le `|…` le plus proche.
Pour modifier une seule copie :
=python plotting.py Interro/Copies/Copie01.pdf=
Les coordonnées agrégées sont écrites atomiquement dans le JSON de
la copie. Fermer la fenêtre avant la fin d'une copie ne remplace pas
son JSON par un résultat incomplet.
It also generates les =Copie01.json=, à partir des =Copie01_01.json=
En cas de soucis, (par exemple les pages ne sont pas dans le bon ordre)
- Réordonner les pages du fichier pdf
- Rerun =python cutleft.py Interro/Copie{id}=
- Rerun =python gemini_for_labels.py Interro/Copie{id}=
3. =python splitting_int.py Interro=
Découpe les copies suivant les exercices
Peut-être appelé avec une seule copie.
Les réponses d'une copie sont préparées dans un dossier temporaire,
puis remplacent ensemble le dossier précédent. En cas d'erreur,
l'ancienne version est conservée. Les réponses devenues obsolètes
restent archivées dans le sous-dossier =Missing=.
4. =python grouping.py Interro=
Regroupe les mêmes questions de différentes copies en groupes de
tailles raisonnables.
5. Facultatif : =python verify_groups.py Interro=
Vérifie que chaque réponse PDF apparaît bien dans les métadonnées
des groupes. La commande renvoie un code non nul si une réponse est
absente ou si la vérification est incomplète.
** Correction et annotation
Optional : Set proxy with ~export HTTPS_PROXY="http://10.0.0.1:3128"~
1. Il faut créer des persp, pour indication de comment corriger, et
relancer =enonce_info.py=
2. =python correction.py Interro --limit 240= OU
=python correction.py Interro/Par\ label/Ex\ 2/Group_1.jpg= OU
=python correction.py Interro --overwrite=
Fais les requêtes de correction à Gemini.
=correction.py= peut être relancé sans supprimer son état. Les
fichiers =correction.json= et =correction_progress.json= sont mis à
jour atomiquement. Avec =--overwrite=, leur version précédente reste
en place jusqu'à la première écriture réussie de la nouvelle
exécution. =--reset= est la seule option qui supprime explicitement
cet état et restaure les fichiers =*_old.pdf=.
L'argument =limit= limite le nombre de requêtes à Gemini Pro
(chères), pour une version low cost, passer =--limit 0=, toutes
les requêtes seront sur Gemini Flash.
Will it resume ? It seems so. Best to wait a bit.
Pour diminuer le coût, il est possible de batch les requêtes, qui
seront alors traitées sous au plus 24h.
+ =python correction.py Interro --batch=
+ OU =python correction.py Interro --batch-from 'Ex 4'=
+ =python submit_batches.py Interro=
+ =python batch_status.py=
+ =python fetch_batched_results.py Interro=
+ =python correction.py Interro --deal-with-batched=
Les quatre commandes de ce flux suivent la convention des scripts
standardisés. Les fichiers de requêtes et le résultat JSONL combiné
sont publiés atomiquement : une interruption ne laisse pas de fichier
final partiellement écrit. =submit_batches.py= conserve aussi les
identifiants distants dans =batch_jobs.json= ; la récupération les
utilise en priorité et garde la recherche par nom pour les anciens
batchs. =batch_status.py --download JOB --output
resultat.jsonl= permet aussi de télécharger atomiquement le résultat
d'un job particulier.
3. =python post-correction.py Interro=
- Essaye de corriger des erreurs d'encodage/d'accents dans
=correction.json=.
- aussi échappe les `_` en dehors du mode math, pour LaTeX.
4. Résolution manuel de conflits, s'il y en a.
Edit `manual_resolutions.txt`. Use :
+ `->` or `x>` : Here set a pipe `|` before or after the new_label name
+ `-x` : replace the goal
+ `ss` : do nothing
+ `sx` : stay, and remove goal.
+ `xx` : move to goal.
+ `xs` : remove old, keep goal.
Then call `python resolve_manual.py Interro`
5. Call `python correction.py Interro --refaire`.
** Génération des copies annotées
1. =python annotating.py Interro= (facultatif)
Ajoute les annotations Gemini aux copies, enregistrées dans le dossier =Anot=.
On peut passer l'argument =--overwrite=.
OU
2. =python annotating_with_checks.py Interro=
Ajoute les annotations Gemini, et des checkboxes à cocher.
Enregistrées dans le dossier =Bnot=,
=--overwrite=
Une seule copie peut être ciblée avec, par exemple,
=python annotating_with_checks.py Interro/Copies/Copie01.pdf=.
Le mode =--refaire= exige un fichier =refaire.json= et écrit dans
=BRnot=.
OU
2. =python annotating_by_label.py Interro= dans =BGnot=
Ajoute les annotations Gemini, et des checkboxes, et regroupe les
réponses par question.
Enregistrées dans =BGnot=
_Needs_ : label_groups file (made automatically by this function),
qui dit quelles questions regrouper.
Dans ces trois modes, les métadonnées JSON sont écrites atomiquement.
Lors d'une régénération, les nouvelles sorties sont préparées dans un
dossier temporaire voisin. Une erreur de rendu conserve donc la sortie
précédente ; pour =BGnot --overwrite=, le dossier complet n'est remplacé
que si tous les groupes ont été produits.
3. =python export.py Interro= (gestion perso)
Cela déplace les groupes dans le dossier configuré par =EXPORT_DIR=
(par défaut =Export=).
Il faut ensuite annoter les fichiers dans `EXPORT_DIR` avec une
tablette graphique.
** Lecture de la correction manuscrite
_Before_ : vider le dossier configuré par =IMPORT_DIR= (par défaut
=Import=), puis y copier ou synchroniser les fichiers depuis la tablette.
1. =python import.py Interro=
Une fois les corrections manuelles appliquées aux fichiers
=Concat.pdf=, il faut enregistrer le fichier annoté au même endroit,
sous le nom =Concat_annotated.pdf=.
2. =python reading_annotations.py Interro=
Lit les =Concat_annotated= dans =Bnot=, regénère les copies avec
les modifications. Les fichiers générés (=score.json=,
=Concat.jpg=, etc.) sont préparés séparément puis installés
ensemble. Les entrées du dossier =Bnot= restent en place et une
erreur de génération conserve les anciennes sorties.
OU
2. =python reading_grouped_annotations.py Interro=
Idem, mais pour =BGnot=. Les tâches parallèles remontent leurs
erreurs au processus principal au lieu de les ignorer.
3. =python giving_names.py Interro BGnot=
Crée un dossier =A Rendre= avec des liens symboliques vers
+ La copie à rendre
+ un fichier =score.json= qui contient les notes par question
Si un nom est =Unknown= : renommer à la main le dossier et le fichier dedans.
4. Éventuellement, faire des modifications manuelles aux =score.json=.
Puis
- `python reading_annotations.py --update-score Interro`
- `python reading_grouped_annotations.py --update-score Interro`
pour mettre à jour les scores dans les images.
4. (gestion perso)
+ =gestion_classe ne= pour créer l'interro puis
+ =gestion_classe we= (set barème here)
+ =python update_ods.py Interro=
ou =python update_ods.py Interro --sum= (en l'absence de barème)
+ =gestion_classe re=
+ =gestion_classe wsent=
+ =python add_final_score.py Interro21=
(this makes files in =Server/copies=)
5. (gestion perso)
+ Deploy =miqmacs-copies-assets=, and
+ update the copies from =miqmacs.fr/admin=.
6. (gestion perso) Impression d'une copie. Via Evince » print to pdf.
* Autres
** Recorrection d'une seule copie (peu testé)
!! Attention, refaire ne marchera pas si tu fais une annotation non
groupée into refaire !!
1. Redécoupage
+ =python plotting.py InterroTest/Copie01.pdf=
+ =python splitting_int.py InterroTest/Copie20.pdf=
2. Créer =refaire.json=, avec un contenu comme
: [["Copie02", []],
: ["Copie01", ["Ex 1 : 1)"]]]
3. Appeler =correction= avec --refaire. Il doit créer des groupes
individuels, faire des requêtes, et remplacer les corrections
précédentes (à sauver ailleurs).
Ou non, si tu veux le faire à la main.
4. ?? Si je fais refaire, avant d'avoir créer les annotating with
checks, que se passe-t-il ???
5. Appeler =annotating_with_checks.py --refaire --overwrite=
6. =python export.py --refaire Interro24=
6. =python import.py --refaire Interro24=
7. =python reading_grouped_annotations.py --refaire Interro24=
Avec =--refaire=, =refaire.json= et le dossier =BRnot= sont des
prérequis obligatoires ; leur absence produit le code de sortie 3.
** Exemple de replotting, refaire d'une copie
1. replot it.
2. `python splitting_int.py DS09VA/Copies/Copie25.pdf`
this will get rid of old/new.
!! Attention, et si ça dégage un new : bad bad bad.
3. Make `refaire.json`, avec la copie, et les labels à refaire.
4. `python correction.py DS09VA --refaire`
5. `python annotating_with_checks.py DS09VA --refaire`
6. `python import.py Interro24 --refaire`
+353
View File
@@ -0,0 +1,353 @@
#+title: Référence des étapes et des scripts
#+author: Sébastien Miquel
#+date: 22-08-2026
#+OPTIONS:
Ce document décrit le workflow détaillé, les commandes disponibles et
les parcours alternatifs.
- [[file:Readme.org][Guide de démarrage]]
- [[file:Architecture.org][Architecture et conventions de développement]]
* Étapes et Script
Utiliser `python -m copienator gui` ou `python -m copienator gui Interro` pour lancer un GUI
qui suit automatiquement les étapes décrites ci-dessous.
** Prétraitement de l'énoncé
Dans le dossier de l'évaluation, mettre les fichiers suivants de l'évaluation :
`enonce.pdf`, `enonce.tex`, `correction.tex`.
- `python -m copienator statement Interro` or
`python -m copienator statement Interro --restart`
À partir des trois fichiers précédents, se charge de détecter les
labels des questions et leur contenu.
Les questions vont également être regroupées. Par la suite, quand
des requêtes de corrections seront effectuées sur une question,
seulement les énoncés des questions du groupe seront envoyés (et le
corrigé de la question). Il faut donc que chaque groupe contienne
si possible le contexte nécessaire pour comprendre la question.
Une fenêtre s'ouvre pour permettre d'éditer le résultat. Ne pas
hésiter à faire des groupes plus gros que les groupes par défaut.
Après relecture le script génère :
+ un fichier `labels` avec les labels des questions
+ Un dossier `Text` avec le contenu textuel des questions,
regroupées.
+ Un dossier `Sol` avec le contenu textuel du corrigé, question par
question.
+ Un dossier `Text2`, qui compile un fichier `.tex` pour chaque
question (utilisé pour compiler un rendu pdf du corrigé pour
chaque question)
+ Un dossier `Sol2`, qui compile un fichier `.tex` pour chaque
correction de chaque question.
+ Un dossier `Persp` avec des instruction de barème pour chaque
question.
Éventuellement : vérifier et modifier les barèmes dans `Persp`.
- Alternative personnelle : `python -m copienator statement-personal Interro`
Ces deux commandes suivent la convention des scripts standardisés.
Leur import ne lance aucun traitement et les erreurs partielles sont
distinguées des échecs. Les réponses d'extraction mises en cache par
=copienator statement= et le fichier =labels= sont publiés
atomiquement.
** Prétraitement des copies
Mettre les copies scannées au format pdf dans =Interro=.
1. =python -m copienator copies rotate Interro= (facultatif)
Retourne tous les pdf de 180°, si la photocopie a été faite à
l'envers.
2. =python -m copienator copies rename Interro=
change le nom des copies en =Copie{id}.pdf=
3. =python -m copienator page-split Interro=
Découpe des copies A3 en pages A4, en retirant les pages vides.
Pour chaque page double il est possible
+ de garder les deux pages
+ de ne garder qu'une des deux pages.
+ de jeter les deux pages
+ de déplacer la délimitation à droite/gauche
Fix issues with =python -m copienator page-split Interro14/Copies/Copie01.pdf=
Le PDF transformé est construit dans un dossier temporaire. La
copie produite et la sauvegarde dans =Copies Originales= sont
ensuite installées avec rollback : une erreur conserve les deux
versions précédentes. Une relance ciblée lit directement la
sauvegarde originale sans la déplacer au préalable.
4. =python -m copienator crop-labels Interro=
Découpe la partie gauche des copies, là où il devrait y avoir les
labels des exercices/questions.
=python -m copienator crop-labels Interro --fullpage= to use the full page always.
Rerun on a single file with =python -m copienator crop-labels Interro/Copies/Copie01.pdf=
Les images et le fichier =_schema.json= d'une copie sont remplacés
ensemble. Fermer l'outil juste après la dernière validation ne peut
donc plus interrompre un thread de sauvegarde en arrière-plan.
** Labelisation et regroupement
Optional : Set proxy with ~export HTTPS_PROXY="http://10.0.0.1:3128"~
1. =python -m copienator labels Interro=, avec éventuellement =--overwrite=
Fait des requêtes à Gemini pour identifier les labels des
questions dans images générées à partir des parties gauches des copies.
Une copie PDF ou une image précise de =Cutleft= peut également être
ciblée. Plusieurs cibles de la même évaluation sont acceptées. Les
parties d'une copie restent traitées séquentiellement afin de
conserver les labels précédents comme contexte, tandis que les
copies différentes sont traitées en parallèle. Chaque réponse JSON
validée est écrite atomiquement. Une cible sans image correspondante
produit le code de sortie 4.
2. =python -m copienator review-labels Interro=
Permet de vérifier visuellement les labels trouvés.
+ Sous linux, on peut faire =e= pour ouvrir le fichier .json et
l'éditer a la main.
+ Quand un label est manquant, il est possible de cliquer sur
l'image, ce qui copie les coordonnées dans le presse papier
(sous linux…), puis on peut l'ajouter à la main.
+ Utilisation de `_`, `|…` et `…|` :
+ `|…` n'est pas arrêté verticalement par son type opposé.
+ `…|` est stoppé horizontalement par le `|…` le plus proche.
Pour modifier une seule copie :
=python -m copienator review-labels Interro/Copies/Copie01.pdf=
Les coordonnées agrégées sont écrites atomiquement dans le JSON de
la copie. Fermer la fenêtre avant la fin d'une copie ne remplace pas
son JSON par un résultat incomplet.
It also generates les =Copie01.json=, à partir des =Copie01_01.json=
En cas de soucis, (par exemple les pages ne sont pas dans le bon ordre)
- Réordonner les pages du fichier pdf
- Rerun =python -m copienator crop-labels Interro/Copie{id}=
- Rerun =python -m copienator labels Interro/Copie{id}=
3. =python -m copienator split-answers Interro=
Découpe les copies suivant les exercices
Peut-être appelé avec une seule copie.
Les réponses d'une copie sont préparées dans un dossier temporaire,
puis remplacent ensemble le dossier précédent. En cas d'erreur,
l'ancienne version est conservée. Les réponses devenues obsolètes
restent archivées dans le sous-dossier =Missing=.
4. =python -m copienator group-answers Interro=
Regroupe les mêmes questions de différentes copies en groupes de
tailles raisonnables.
5. Facultatif : =python -m copienator verify-groups Interro=
Vérifie que chaque réponse PDF apparaît bien dans les métadonnées
des groupes. La commande renvoie un code non nul si une réponse est
absente ou si la vérification est incomplète.
** Correction et annotation
Optional : Set proxy with ~export HTTPS_PROXY="http://10.0.0.1:3128"~
1. Il faut créer des persp, pour indication de comment corriger, et
relancer =copienator statement-personal=
2. =python -m copienator correct Interro --limit 240= OU
=python -m copienator correct Interro/Par\ label/Ex\ 2/Group_1.jpg= OU
=python -m copienator correct Interro --overwrite=
Fais les requêtes de correction à Gemini.
=copienator correct= peut être relancé sans supprimer son état. Les
fichiers =correction.json= et =correction_progress.json= sont mis à
jour atomiquement. Avec =--overwrite=, leur version précédente reste
en place jusqu'à la première écriture réussie de la nouvelle
exécution. =--reset= est la seule option qui supprime explicitement
cet état et restaure les fichiers =*_old.pdf=.
L'argument =limit= limite le nombre de requêtes à Gemini Pro
(chères), pour une version low cost, passer =--limit 0=, toutes
les requêtes seront sur Gemini Flash.
Pour diminuer le coût, il est possible de batch les requêtes, qui
seront alors traitées sous au plus 24h.
+ =python -m copienator correct Interro --batch=
+ OU =python -m copienator correct Interro --batch-from 'Ex 4'=
+ =python -m copienator batch-submit Interro=
+ =python -m copienator batch-status=
+ =python -m copienator batch-fetch Interro=
+ =python -m copienator correct Interro --deal-with-batched=
Les quatre commandes de ce flux suivent la convention des scripts
standardisés. Les fichiers de requêtes et le résultat JSONL combiné
sont publiés atomiquement : une interruption ne laisse pas de fichier
final partiellement écrit. =copienator batch-submit= conserve aussi les
identifiants distants dans =batch_jobs.json= ; la récupération les
utilise en priorité et garde la recherche par nom pour les anciens
batchs. =python -m copienator batch-status --download JOB --output
resultat.jsonl= permet aussi de télécharger atomiquement le résultat
d'un job particulier.
3. =python -m copienator post-correction Interro=
- Essaye de corriger des erreurs d'encodage/d'accents dans
=correction.json=.
- aussi échappe les `_` en dehors du mode math, pour LaTeX.
4. Résolution manuel de conflits, s'il y en a.
Edit `manual_resolutions.txt`. Use :
+ `->` or `x>` : Here set a pipe `|` before or after the new_label name
+ `-x` : replace the goal
+ `ss` : do nothing
+ `sx` : stay, and remove goal.
+ `xx` : move to goal.
+ `xs` : remove old, keep goal.
Then call `python -m copienator resolve-manual Interro`
5. Call `python -m copienator correct Interro --refaire`.
** Génération des copies annotées
1. =python -m copienator annotate-simple Interro= (facultatif)
Ajoute les annotations Gemini aux copies, enregistrées dans le dossier =Anot=.
On peut passer l'argument =--overwrite=.
OU
2. =python -m copienator annotate-checks Interro=
Ajoute les annotations Gemini, et des checkboxes à cocher.
Enregistrées dans le dossier =Bnot=,
=--overwrite=
Une seule copie peut être ciblée avec, par exemple,
=python -m copienator annotate-checks Interro/Copies/Copie01.pdf=.
Le mode =--refaire= exige un fichier =refaire.json= et écrit dans
=BRnot=.
OU
2. =python -m copienator annotate-grouped Interro= dans =BGnot=
Ajoute les annotations Gemini, et des checkboxes, et regroupe les
réponses par question.
Enregistrées dans =BGnot=
_Needs_ : label_groups file (made automatically by this function),
qui dit quelles questions regrouper.
3. =python -m copienator export Interro BGnot= (gestion perso)
Cela déplace les groupes dans le dossier configuré par =EXPORT_DIR=
(par défaut =Export=).
Le second argument peut être =BGnot=, =Bnot= ou =Anot= et reste
facultatif (=BGnot= par défaut). =Anot= exporte =Concat.jpg= ; les
deux autres modes exportent =Concat.pdf=. Dans le GUI, seuls les
dossiers présents sont proposés et le mode de la dernière génération
d'annotations est présélectionné.
Il faut ensuite annoter les fichiers dans `EXPORT_DIR` avec une
tablette graphique.
** Lecture de la correction manuscrite
_Before_ : vider le dossier configuré par =IMPORT_DIR= (par défaut
=Import=), puis y copier ou synchroniser les fichiers depuis la tablette.
1. =python -m copienator import Interro BGnot=
Une fois les corrections manuelles appliquées aux fichiers
=Concat.pdf=, il faut enregistrer le fichier annoté au même endroit,
sous le nom =Concat_annotated.pdf=.
Comme pour l'export, le second argument accepte =BGnot=, =Bnot= ou
=Anot=. Le GUI présélectionne le dossier utilisé lors du dernier
export. Pour =Anot=, l'image est importée sous le nom
=Concat_annotated.jpg=.
2. =python -m copienator read-annotations Interro=
Lit les =Concat_annotated= dans =Bnot=, regénère les copies avec
les modifications. Les fichiers générés (=score.json=,
=Concat.jpg=, etc.) sont préparés séparément puis installés
ensemble. Les entrées du dossier =Bnot= restent en place et une
erreur de génération conserve les anciennes sorties.
OU
2. =python -m copienator read-grouped Interro=
Idem, mais pour =BGnot=. Les tâches parallèles remontent leurs
erreurs au processus principal au lieu de les ignorer.
3. =python -m copienator giving-names Interro BGnot=
Crée un dossier =A Rendre= avec des liens symboliques vers
+ La copie à rendre
+ un fichier =score.json= qui contient les notes par question
Si un nom est =Unknown= : renommer à la main le dossier et le fichier dedans.
4. Éventuellement, faire des modifications manuelles aux =score.json=.
Puis
- `python -m copienator read-annotations --update-score Interro`
- `python -m copienator read-grouped --update-score Interro`
pour mettre à jour les scores dans les images.
4. (gestion perso)
+ =gestion_classe ne= pour créer l'interro puis
+ =gestion_classe we= (set barème here)
+ =python -m copienator update-ods Interro=
ou =python -m copienator update-ods Interro --sum= (en l'absence de barème)
+ =gestion_classe re=
+ =gestion_classe wsent=
+ =python -m copienator add-final-score Interro21=
(this makes files in =Server/copies=)
5. (gestion perso)
+ Deploy =miqmacs-copies-assets=, and
+ update the copies from =miqmacs.fr/admin=.
6. (gestion perso) Impression d'une copie. Via Evince » print to pdf.
* Autres
** Recorrection d'une seule copie (peu testé)
!! Attention, refaire ne marchera pas si tu fais une annotation non
groupée into refaire !!
1. Redécoupage
+ =python -m copienator review-labels InterroTest/Copie01.pdf=
+ =python -m copienator split-answers InterroTest/Copie20.pdf=
2. Créer =refaire.json=, avec un contenu comme
: [["Copie02", []],
: ["Copie01", ["Ex 1 : 1)"]]]
3. Appeler =correction= avec --refaire. Il doit créer des groupes
individuels, faire des requêtes, et remplacer les corrections
précédentes (à sauver ailleurs).
Ou non, si tu veux le faire à la main.
4. ?? Si je fais refaire, avant d'avoir créer les annotating with
checks, que se passe-t-il ???
5. Appeler =python -m copienator annotate-checks --refaire --overwrite=
6. =python -m copienator export --refaire Interro24=
6. =python -m copienator import --refaire Interro24=
7. =python -m copienator read-grouped --refaire Interro24=
Avec =--refaire=, =refaire.json= et le dossier =BRnot= sont des
prérequis obligatoires ; leur absence produit le code de sortie 3.
** Exemple de replotting, refaire d'une copie
1. replot it.
2. `python -m copienator split-answers DS09VA/Copies/Copie25.pdf`
this will get rid of old/new.
!! Attention, et si ça dégage un new : bad bad bad.
3. Make `refaire.json`, avec la copie, et les labels à refaire.
4. `python -m copienator correct DS09VA --refaire`
5. `python -m copienator annotate-checks DS09VA --refaire`
6. `python -m copienator import Interro24 --refaire`
+5
View File
@@ -0,0 +1,5 @@
from .dispatcher import main
if __name__ == "__main__":
raise SystemExit(main())
+5
View File
@@ -0,0 +1,5 @@
"""Executable Copienator commands.
Command modules expose a ``main(argv=None)`` entry point and may also expose
their processing functions for reuse and tests.
"""
@@ -7,7 +7,7 @@ from pathlib import Path
import pandas as pd import pandas as pd
from PIL import Image, ImageDraw, ImageFont from PIL import Image, ImageDraw, ImageFont
from config import FINAL_SCORE_FONT_PATH, FINAL_SCORE_ODS_PATH, FINAL_SCORE_OUTPUT_DIR from copienator.configuration import FINAL_SCORE_FONT_PATH, FINAL_SCORE_ODS_PATH, FINAL_SCORE_OUTPUT_DIR
# Configuration constants # Configuration constants
ODS_PATH = Path(FINAL_SCORE_ODS_PATH).expanduser() ODS_PATH = Path(FINAL_SCORE_ODS_PATH).expanduser()
@@ -112,14 +112,17 @@ def process_images(base_dir, output_dir):
shutil.copy(str(pdf_path), str(save_path)) shutil.copy(str(pdf_path), str(save_path))
if __name__ == "__main__": def main(argv=None):
parser = argparse.ArgumentParser(description="Stamp scores on exam copies.") parser = argparse.ArgumentParser(description="Stamp scores on exam copies.")
parser.add_argument("dir", type=Path, help="Root directory containing 'A Rendre' folder") parser.add_argument("dir", type=Path, help="Root directory containing 'A Rendre' folder")
args = parser.parse_args(argv)
args = parser.parse_args()
base_dir = args.dir.expanduser().resolve() base_dir = args.dir.expanduser().resolve()
output_dir = OUTPUT_DIR / base_dir.name output_dir = OUTPUT_DIR / base_dir.name
output_dir.mkdir(parents=True, exist_ok=True) output_dir.mkdir(parents=True, exist_ok=True)
process_images(base_dir, output_dir) process_images(base_dir, output_dir)
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -19,8 +19,8 @@ from matplotlib import pyplot as plt
from pdf2image import convert_from_path from pdf2image import convert_from_path
from PIL import Image, ImageDraw, ImageFont from PIL import Image, ImageDraw, ImageFont
import utils from copienator import utils
from config import LATEX_ANOT_AFTER, LATEX_ANOT_BEFORE from copienator.configuration import LATEX_ANOT_AFTER, LATEX_ANOT_BEFORE
from copienator import ( from copienator import (
EvaluationWorkspace, EvaluationWorkspace,
ExitCode, ExitCode,
@@ -31,7 +31,7 @@ from copienator import (
) )
from copienator.annotation_data import load_annotation_data from copienator.annotation_data import load_annotation_data
from copienator.filesystem import staged_directory from copienator.filesystem import staged_directory
from utils import natural_key from copienator.utils import natural_key
MARGIN_LEFT = 300 MARGIN_LEFT = 300
ANNOT_WIDTH = 600 ANNOT_WIDTH = 600
@@ -10,9 +10,9 @@ from typing import Any
from PIL import Image, ImageDraw from PIL import Image, ImageDraw
from reportlab.pdfgen import canvas from reportlab.pdfgen import canvas
import annotating from copienator.commands import annotating
import annotating_with_checks from copienator.commands import annotating_with_checks
import utils from copienator import utils
from copienator import ( from copienator import (
CliError, CliError,
EvaluationWorkspace, EvaluationWorkspace,
@@ -26,7 +26,7 @@ from copienator import (
) )
from copienator.annotation_data import load_annotation_data from copienator.annotation_data import load_annotation_data
from copienator.filesystem import staged_directory from copienator.filesystem import staged_directory
from utils import natural_key from copienator.utils import natural_key
MAX_HEIGHT_PX = 25000 MAX_HEIGHT_PX = 25000
@@ -125,11 +125,80 @@ def _initial_label_groups(labels: list[str]) -> str:
return "".join(",".join(items) + "\n" for items in groups.values()) return "".join(",".join(items) + "\n" for items in groups.values())
def _gemini_label_groups(
workspace: EvaluationWorkspace, labels: list[str]
) -> list[list[str]] | None:
source = workspace.gemini_exam_items_file
if not source.is_file():
return None
groups: list[list[str]] = []
current: list[str] = []
try:
source_lines = source.read_text(encoding="utf-8").splitlines()
except (OSError, UnicodeError) as exc:
print(f"Warning: could not read Gemini question groups from {source}: {exc}")
return None
for raw_line in source_lines:
line = raw_line.strip()
if not line or line.startswith("#"):
continue
if line == "---":
if current:
groups.append(current)
current = []
continue
if " ### " not in line:
continue
label = line.split(" ### ", 1)[0].strip()
if label and label != "CONTEXT":
current.append(label)
if current:
groups.append(current)
flattened = [label for group in groups for label in group]
known = set(labels)
if (
not flattened
or len(flattened) != len(set(flattened))
or set(flattened) != known
):
missing = sorted(known.difference(flattened), key=natural_key)
unknown = sorted(set(flattened).difference(known), key=natural_key)
details = []
if missing:
details.append("missing: " + ", ".join(missing))
if unknown:
details.append("unknown: " + ", ".join(unknown))
if len(flattened) != len(set(flattened)):
details.append("duplicate labels")
print(
f"Warning: ignoring incompatible Gemini question groups in {source}"
+ (f" ({'; '.join(details)})" if details else "")
)
return None
return groups
def _serialize_label_groups(groups: list[list[str]]) -> str:
return "".join(",".join(group) + "\n" for group in groups)
def _load_label_groups(workspace: EvaluationWorkspace, labels: list[str]) -> list[list[str]]: def _load_label_groups(workspace: EvaluationWorkspace, labels: list[str]) -> list[list[str]]:
label_groups = workspace.root / "label_groups" label_groups = workspace.label_groups_file
if not label_groups.exists(): if not label_groups.exists():
atomic_write_text(label_groups, _initial_label_groups(labels)) gemini_groups = _gemini_label_groups(workspace, labels)
print(f"Created {label_groups}; review the groups before continuing.") if gemini_groups is not None:
initial_content = _serialize_label_groups(gemini_groups)
source_description = "the groups selected in gemini_for_enonce.py"
else:
initial_content = _initial_label_groups(labels)
source_description = "the label-prefix fallback"
atomic_write_text(label_groups, initial_content)
print(
f"Created {label_groups} from {source_description}; "
"review the groups before continuing."
)
utils.edit_file_and_enter(label_groups) utils.edit_file_and_enter(label_groups)
known_labels = set(labels) known_labels = set(labels)
groups: list[list[str]] = [] groups: list[list[str]] = []
@@ -14,8 +14,8 @@ matplotlib.use("Agg")
from PIL import Image, ImageFont from PIL import Image, ImageFont
from reportlab.pdfgen import canvas from reportlab.pdfgen import canvas
import annotating from copienator.commands import annotating
import utils from copienator import utils
from copienator import ( from copienator import (
CliError, CliError,
EvaluationWorkspace, EvaluationWorkspace,
@@ -28,7 +28,7 @@ from copienator import (
) )
from copienator.annotation_data import load_annotation_data from copienator.annotation_data import load_annotation_data
from copienator.filesystem import staged_directory from copienator.filesystem import staged_directory
from utils import natural_key from copienator.utils import natural_key
BOX_SIZE = 30 BOX_SIZE = 30
SCORE_BOX_SIZE = 40 SCORE_BOX_SIZE = 40
@@ -328,3 +328,4 @@ def main(argv: Sequence[str] | None = None) -> int:
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(main()) raise SystemExit(main())
@@ -6,7 +6,7 @@ from pathlib import Path
from google import genai from google import genai
import config from copienator import configuration as config
from copienator import ( from copienator import (
CliError, CliError,
ExitCode, ExitCode,
@@ -134,3 +134,4 @@ def main(argv: Sequence[str] | None = None) -> int:
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(main()) raise SystemExit(main())
@@ -15,9 +15,9 @@ from pathlib import Path
from google import genai from google import genai
import config from copienator import configuration as config
import grouping from copienator.commands import grouping
import prompting from copienator import prompting
from copienator import ( from copienator import (
CliError, CliError,
EvaluationWorkspace, EvaluationWorkspace,
@@ -29,7 +29,7 @@ from copienator import (
target_parser, target_parser,
workspace_from_target, workspace_from_target,
) )
from utils import enonce_total, read_all_labels from copienator.utils import enonce_total, read_all_labels
NB_THREADS = 12 NB_THREADS = 12
@@ -874,7 +874,7 @@ def run_configured(args: argparse.Namespace) -> ExitCode:
+ "\n", + "\n",
) )
print(f"\n[!] Unresolved delayed tasks found! Wrote to {manual_path}.") print(f"\n[!] Unresolved delayed tasks found! Wrote to {manual_path}.")
print(" Please edit it manually, then run `python resolve_manual.py <InputDir>`") print(" Please edit it manually, then run `python -m copienator resolve-manual <InputDir>`")
end_time = time.time() end_time = time.time()
print("Time elapsed : ", end_time - start_time) print("Time elapsed : ", end_time - start_time)
@@ -884,7 +884,7 @@ def run_configured(args: argparse.Namespace) -> ExitCode:
for (err, file) in errors_summary: for (err, file) in errors_summary:
print(err, file=sys.stderr) print(err, file=sys.stderr)
escaped_path = shlex.quote(str(file)) escaped_path = shlex.quote(str(file))
print(f"Run : python correction.py {escaped_path}") print(f"Run : python -m copienator correct {escaped_path}")
return ExitCode.PARTIAL if errors_summary else ExitCode.SUCCESS return ExitCode.PARTIAL if errors_summary else ExitCode.SUCCESS
@@ -341,3 +341,4 @@ def main(argv: Sequence[str] | None = None) -> int:
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(main()) raise SystemExit(main())
@@ -17,8 +17,8 @@ from copienator import (
execute, execute,
workspace_from_args, workspace_from_args,
) )
from platform_utils import WindowsLabelError, validate_windows_labels from copienator.platform import WindowsLabelError, validate_windows_labels
from utils import compile_to_pdf from copienator.utils import compile_to_pdf
def fetch_and_save_sub_text(ex_id, indices, label, text_path): def fetch_and_save_sub_text(ex_id, indices, label, text_path):
@@ -293,3 +293,4 @@ def main(argv: Sequence[str] | None = None) -> int:
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(main()) raise SystemExit(main())
+37 -9
View File
@@ -3,7 +3,7 @@ import sys
from collections.abc import Sequence from collections.abc import Sequence
from pathlib import Path from pathlib import Path
from config import EXPORT_DIR from copienator.configuration import EXPORT_DIR
from copienator import ( from copienator import (
EvaluationWorkspace, EvaluationWorkspace,
ExitCode, ExitCode,
@@ -11,7 +11,9 @@ from copienator import (
execute, execute,
workspace_from_args, workspace_from_args,
) )
from platform_utils import replace_with_link_or_copy from copienator.platform import replace_with_link_or_copy
ANNOTATION_DIRECTORIES = ("BGnot", "Bnot", "Anot")
def export_directory( def export_directory(
@@ -31,12 +33,22 @@ def export_directory(
missing_outputs = 0 missing_outputs = 0
for subdir in subdirs: for subdir in subdirs:
concat_file = subdir / "Concat.pdf" concat_file = next(
if not concat_file.is_file(): (
print(f"Warning: file not found: {concat_file}", file=sys.stderr) candidate
for candidate in (subdir / "Concat.pdf", subdir / "Concat.jpg")
if candidate.is_file()
),
None,
)
if concat_file is None:
print(
f"Warning: no Concat.pdf or Concat.jpg found in {subdir}",
file=sys.stderr,
)
missing_outputs += 1 missing_outputs += 1
continue continue
destination = sync_dir / f"{subdir.name}.pdf" destination = sync_dir / f"{subdir.name}{concat_file.suffix.lower()}"
method = replace_with_link_or_copy(concat_file, destination, prefer="hardlink") method = replace_with_link_or_copy(concat_file, destination, prefer="hardlink")
print(f"Exported: {destination} ({method})") print(f"Exported: {destination} ({method})")
return ExitCode.PARTIAL if missing_outputs else ExitCode.SUCCESS return ExitCode.PARTIAL if missing_outputs else ExitCode.SUCCESS
@@ -44,12 +56,24 @@ def export_directory(
def build_parser() -> argparse.ArgumentParser: def build_parser() -> argparse.ArgumentParser:
parser = evaluation_parser("Export annotated PDFs to the tablet directory.") parser = evaluation_parser("Export annotated PDFs to the tablet directory.")
parser.add_argument(
"annotation_dir",
nargs="?",
choices=ANNOTATION_DIRECTORIES,
default="BGnot",
help="Annotation directory to export (default: BGnot)",
)
parser.add_argument("--refaire", action="store_true", help="Process only copies/labels defined in refaire.json") parser.add_argument("--refaire", action="store_true", help="Process only copies/labels defined in refaire.json")
return parser return parser
def run(workspace: EvaluationWorkspace, *, refaire: bool = False) -> ExitCode: def run(
return export_directory(workspace, "BRnot" if refaire else "BGnot") workspace: EvaluationWorkspace,
*,
annotation_dir: str = "BGnot",
refaire: bool = False,
) -> ExitCode:
return export_directory(workspace, "BRnot" if refaire else annotation_dir)
def main(argv: Sequence[str] | None = None) -> int: def main(argv: Sequence[str] | None = None) -> int:
@@ -57,7 +81,11 @@ def main(argv: Sequence[str] | None = None) -> int:
return execute( return execute(
parser, parser,
argv, argv,
lambda args: run(workspace_from_args(args), refaire=args.refaire), lambda args: run(
workspace_from_args(args),
annotation_dir=args.annotation_dir,
refaire=args.refaire,
),
) )
@@ -5,7 +5,7 @@ from collections.abc import Sequence
from google import genai from google import genai
import config from copienator import configuration as config
from copienator import ( from copienator import (
CliError, CliError,
EvaluationWorkspace, EvaluationWorkspace,
@@ -10,8 +10,8 @@ from google import genai
from google.genai import types from google.genai import types
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
import config from copienator import configuration as config
import utils from copienator import utils
from copienator import ( from copienator import (
CliError, CliError,
EvaluationWorkspace, EvaluationWorkspace,
@@ -21,8 +21,8 @@ from copienator import (
execute, execute,
workspace_from_args, workspace_from_args,
) )
from platform_utils import validate_windows_labels from copienator.platform import validate_windows_labels
from utils import compile_to_pdf from copienator.utils import compile_to_pdf
def get_lcp(s1: str, s2: str) -> str: def get_lcp(s1: str, s2: str) -> str:
@@ -13,7 +13,7 @@ from google import genai
from google.genai import types from google.genai import types
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
import config from copienator import configuration as config
from copienator import ( from copienator import (
CliError, CliError,
EvaluationWorkspace, EvaluationWorkspace,
@@ -24,7 +24,7 @@ from copienator import (
target_parser, target_parser,
workspace_from_target, workspace_from_target,
) )
from utils import natural_key, read_all_labels from copienator.utils import natural_key, read_all_labels
MODEL_ID = config.MODEL_FOR_LABEL_ID MODEL_ID = config.MODEL_FOR_LABEL_ID
api_key = config.API_KEY api_key = config.API_KEY
@@ -422,7 +422,7 @@ def main(argv: Sequence[str] | None = None) -> int:
def handle(args: argparse.Namespace) -> ExitCode: def handle(args: argparse.Namespace) -> ExitCode:
workspace, target = workspace_from_target( workspace, target = workspace_from_target(
args, repository=Path(__file__).resolve().parent args, repository=Path(__file__).resolve().parents[2]
) )
targets = [target] targets = [target]
for additional in args.additional_targets: for additional in args.additional_targets:
@@ -15,7 +15,7 @@ from copienator import (
read_json, read_json,
workspace_from_args, workspace_from_args,
) )
from platform_utils import replace_with_link_or_copy, safe_filename from copienator.platform import replace_with_link_or_copy, safe_filename
ANNOTATION_CHOICES = ("BGnot", "Bnot", "Anot") ANNOTATION_CHOICES = ("BGnot", "Bnot", "Anot")
@@ -163,3 +163,4 @@ def main(argv: Sequence[str] | None = None) -> int:
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(main()) raise SystemExit(main())
@@ -219,7 +219,7 @@ def create_jpg(identifier, group_index, group, root_dir):
print(f"Saved {output_path} with {len(group)} ({os.path.getsize(output_path)/1024/1024:.2f} MB)") print(f"Saved {output_path} with {len(group)} ({os.path.getsize(output_path)/1024/1024:.2f} MB)")
from utils import natural_key from copienator.utils import natural_key
def process_identifier(identifier, files_info, output_dir): def process_identifier(identifier, files_info, output_dir):
@@ -275,3 +275,4 @@ def main(argv: Sequence[str] | None = None) -> int:
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(main()) raise SystemExit(main())
+95
View File
@@ -0,0 +1,95 @@
import argparse
import shutil
import sys
from collections.abc import Sequence
from pathlib import Path
from copienator.configuration import IMPORT_DIR
from copienator import (
EvaluationWorkspace,
ExitCode,
evaluation_parser,
execute,
workspace_from_args,
)
ANNOTATION_DIRECTORIES = ("BGnot", "Bnot", "Anot")
def sync_annotated(
workspace: EvaluationWorkspace,
*,
annotation_dir_name: str,
import_dir: Path,
) -> ExitCode:
workspace.require_directories(annotation_dir_name)
annotation_dir = workspace.root / annotation_dir_name
annotated_dir = Path(import_dir).expanduser()
if not annotated_dir.is_dir():
print(f"Error: directory does not exist: {annotated_dir}", file=sys.stderr)
return ExitCode.INVALID_WORKSPACE
missing_targets = 0
annotated_files = sorted(
(
path
for path in annotated_dir.iterdir()
if path.is_file() and path.suffix.casefold() in {".pdf", ".jpg", ".jpeg"}
),
key=lambda path: path.name.casefold(),
)
for annotated_file in annotated_files:
target_subdir = annotation_dir / annotated_file.stem
if not target_subdir.is_dir():
print(f"Warning: directory not found: {target_subdir}", file=sys.stderr)
missing_targets += 1
else:
suffix = annotated_file.suffix.lower()
dest_file = target_subdir / f"Concat_annotated{suffix}"
print(f"Copying {annotated_file} to {dest_file}")
shutil.copy2(annotated_file, dest_file)
return ExitCode.PARTIAL if missing_targets else ExitCode.SUCCESS
def build_parser() -> argparse.ArgumentParser:
parser = evaluation_parser("Import handwritten annotations from the tablet directory.")
parser.add_argument(
"annotation_dir",
nargs="?",
choices=ANNOTATION_DIRECTORIES,
default="BGnot",
help="Annotation directory receiving imported PDFs (default: BGnot)",
)
parser.add_argument("--refaire", action="store_true", help="Process only copies/labels defined in refaire.json")
return parser
def run(
workspace: EvaluationWorkspace,
*,
annotation_dir: str = "BGnot",
refaire: bool = False,
) -> ExitCode:
return sync_annotated(
workspace,
annotation_dir_name="BRnot" if refaire else annotation_dir,
import_dir=Path(IMPORT_DIR),
)
def main(argv: Sequence[str] | None = None) -> int:
parser = build_parser()
return execute(
parser,
argv,
lambda args: run(
workspace_from_args(args),
annotation_dir=args.annotation_dir,
refaire=args.refaire,
),
)
if __name__ == "__main__":
raise SystemExit(main())
+87
View File
@@ -0,0 +1,87 @@
import json
import sys
import concurrent.futures
from pathlib import Path
from copienator.commands import correction
def get_missing_tasks():
"""
Identifies tasks (groups) where NONE of the student IDs in that group
appear in the existing results for that label in correction.json.
"""
missing = []
# correction.results is already loaded from correction.json during 'from copienator.commands import correction'
# correction.tasks is populated with (filepath, label) during 'from copienator.commands import correction'
for task in correction.tasks:
file_path, label = task
# Find the group metadata file (Group_X.json) to know which IDs are inside
meta_path = Path(file_path).with_suffix('.json')
if not meta_path.exists():
print("Missing meta_path :", meta_path)
continue
with open(meta_path, 'r', encoding="utf-8") as f:
# group_data entries: [pid, ymin, ymax, width_ratio]
group_data = json.load(f)
pids_in_group = [str(item[0]) for item in group_data]
# Check correction.json results for this specific label
label_results = correction.results.get(label, [])
# Collect all student IDs that have already been processed for this label
covered_ids = set()
for result_list in label_results:
for entry in result_list:
covered_ids.add(str(entry.get('id')))
# Logic: Only process if EVERY ID in this group is missing from correction.json
if all(pid not in covered_ids for pid in pids_in_group):
missing.append(task)
return missing
def main(argv=None):
if argv:
print("missing-correction does not accept command-line arguments.", file=sys.stderr)
return 2
missing_tasks = get_missing_tasks()
print("\n Total nb of tasks : ", len(correction.tasks))
if not missing_tasks:
print("All groups are already present in correction.json. Nothing to do.")
return
print("\nThe following groups are missing from correction.json:")
for path, label in missing_tasks:
print(f" - [{label}] {path}")
confirm = input(f"\nFound {len(missing_tasks)} missing groups. Start processing? (y/N): ")
if confirm.lower() != 'y':
print("Aborted.")
return
print(f"Processing {len(missing_tasks)} tasks with {correction.NB_THREADS} threads...")
with concurrent.futures.ThreadPoolExecutor(max_workers=correction.NB_THREADS) as executor:
# Map tasks to the processing function defined in correction.py
futures = {executor.submit(correction.process_single_task, t): t for t in missing_tasks}
for future in concurrent.futures.as_completed(futures):
try:
# Handle potential sub-tasks (like label errors) generated during processing
new_generated_tasks = future.result()
if new_generated_tasks:
for nt in new_generated_tasks:
executor.submit(correction.process_single_task, nt)
except Exception as e:
t = futures[future]
print(f"Error processing {t[0]}: {e}")
print("\nProcessing complete.")
if __name__ == "__main__":
raise SystemExit(main())
@@ -16,7 +16,7 @@ import fitz # PyMuPDF
from PIL import Image, ImageDraw, ImageTk from PIL import Image, ImageDraw, ImageTk
from pypdf import PdfReader, PdfWriter from pypdf import PdfReader, PdfWriter
from config import PAGE_SPLITTER_KB from copienator.configuration import PAGE_SPLITTER_KB
from copienator import ( from copienator import (
CliError, CliError,
EvaluationWorkspace, EvaluationWorkspace,
@@ -25,7 +25,7 @@ from copienator import (
target_parser, target_parser,
workspace_from_target, workspace_from_target,
) )
from platform_utils import launch_pdf_arranger from copienator.platform import launch_pdf_arranger
# --- Constants --- # --- Constants ---
# Conversion factor: 1 cm to points (1 inch = 2.54 cm, 72 points = 1 inch) # Conversion factor: 1 cm to points (1 inch = 2.54 cm, 72 points = 1 inch)
@@ -19,8 +19,8 @@ from copienator import (
target_parser, target_parser,
workspace_from_target, workspace_from_target,
) )
from platform_utils import open_path from copienator.platform import open_path
from utils import natural_key, read_all_labels from copienator.utils import natural_key, read_all_labels
# --- Configuration & Globals --- # --- Configuration & Globals ---
padding = 60 padding = 60
@@ -472,3 +472,4 @@ def main(argv: Sequence[str] | None = None) -> int:
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(main()) raise SystemExit(main())
@@ -16,7 +16,7 @@ from copienator import (
workspace_from_args, workspace_from_args,
) )
WORD_LIST_FILE = Path(__file__).with_name("liste_francais.txt") WORD_LIST_FILE = Path(__file__).resolve().parents[1] / "data" / "liste_francais.txt"
ACCENT_PATTERN = re.compile(r"[éèêëàâäîïôöùûüçœÉÈÊËÀÂÄÎÏÔÖÙÛÜÇŒ]") ACCENT_PATTERN = re.compile(r"[éèêëàâäîïôöùûüçœÉÈÊËÀÂÄÎÏÔÖÙÛÜÇŒ]")
@@ -9,8 +9,8 @@ import numpy as np
from pdf2image import convert_from_path from pdf2image import convert_from_path
from PIL import Image, ImageChops, ImageDraw, ImageFilter from PIL import Image, ImageChops, ImageDraw, ImageFilter
import annotating from copienator.commands import annotating
import utils from copienator import utils
from copienator import ( from copienator import (
EvaluationWorkspace, EvaluationWorkspace,
ExitCode, ExitCode,
@@ -284,3 +284,4 @@ def main(argv: Sequence[str] | None = None) -> int:
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(main()) raise SystemExit(main())
@@ -9,8 +9,8 @@ from typing import Any
from PIL import Image, ImageDraw from PIL import Image, ImageDraw
import annotating from copienator.commands import annotating
import utils from copienator import utils
from copienator import ( from copienator import (
EvaluationWorkspace, EvaluationWorkspace,
ExitCode, ExitCode,
@@ -23,7 +23,7 @@ from copienator import (
from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides
from copienator.annotation_data import AnnotationData, RefaireList, load_annotation_data from copienator.annotation_data import AnnotationData, RefaireList, load_annotation_data
from copienator.filesystem import staged_files from copienator.filesystem import staged_files
from reading_annotations import ( from copienator.commands.reading_annotations import (
concatenate, concatenate,
detect_checks_and_notes, detect_checks_and_notes,
has_significant_notes, has_significant_notes,
@@ -398,3 +398,4 @@ def main(argv: Sequence[str] | None = None) -> int:
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(main()) raise SystemExit(main())
@@ -259,7 +259,7 @@ def resolve_manual(workspace: EvaluationWorkspace) -> ExitCode:
if refaire_tasks: if refaire_tasks:
print( print(
f"File {workspace.refaire_file.name} generated. Run " f"File {workspace.refaire_file.name} generated. Run "
f'`python correction.py "{workspace.command_argument()}" --refaire` ' f'`python -m copienator correct "{workspace.command_argument()}" --refaire` '
"to process updates." "to process updates."
) )
else: else:
@@ -10,7 +10,7 @@ from pathlib import Path
import fitz import fitz
from pypdf import PdfReader, PdfWriter from pypdf import PdfReader, PdfWriter
import utils from copienator import utils
from copienator import ( from copienator import (
CliError, CliError,
EvaluationWorkspace, EvaluationWorkspace,
@@ -264,3 +264,4 @@ def main(argv: Sequence[str] | None = None) -> int:
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(main()) raise SystemExit(main())
@@ -6,7 +6,7 @@ from collections.abc import Sequence
from google import genai from google import genai
from google.genai import types from google.genai import types
import config from copienator import configuration as config
from copienator import ( from copienator import (
CliError, CliError,
EvaluationWorkspace, EvaluationWorkspace,
@@ -6,18 +6,18 @@ from pathlib import Path
import ezodf import ezodf
from config import CURRENT_SCORE_ODS_PATH from copienator.configuration import CURRENT_SCORE_ODS_PATH
from utils import read_all_labels from copienator.utils import read_all_labels
# Configuration # Configuration
ODS_PATH = Path(CURRENT_SCORE_ODS_PATH).expanduser() ODS_PATH = Path(CURRENT_SCORE_ODS_PATH).expanduser()
TARGET_DIR_NAME = "A Rendre" TARGET_DIR_NAME = "A Rendre"
def main(): def main(argv=None):
parser = argparse.ArgumentParser(description="Update ODS with student scores.") parser = argparse.ArgumentParser(description="Update ODS with student scores.")
parser.add_argument("work_dir", nargs="?", default=os.getcwd(), help="Directory to process") parser.add_argument("work_dir", nargs="?", default=os.getcwd(), help="Directory to process")
parser.add_argument("--sum", action="store_true", help="Write only the total sum per student") parser.add_argument("--sum", action="store_true", help="Write only the total sum per student")
args = parser.parse_args() args = parser.parse_args(argv)
work_dir = os.path.abspath(args.work_dir) work_dir = os.path.abspath(args.work_dir)
@@ -169,4 +169,4 @@ def main():
print("Done.") print("Done.")
if __name__ == "__main__": if __name__ == "__main__":
main() raise SystemExit(main())
+82
View File
@@ -0,0 +1,82 @@
from __future__ import annotations
import argparse
import re
import sys
from collections.abc import Sequence
from pathlib import Path
from copienator import (
EvaluationWorkspace,
ExitCode,
evaluation_parser,
execute,
read_json,
workspace_from_args,
)
COPY_PATTERN = re.compile(r"Copie(\d+)")
def build_parser() -> argparse.ArgumentParser:
return evaluation_parser("Verify that every answer PDF appears in group metadata.")
def collect_source_pdfs(copies_dir: Path) -> set[tuple[str, str]]:
source_pdfs: set[tuple[str, str]] = set()
for copy_dir in copies_dir.iterdir():
match = COPY_PATTERN.fullmatch(copy_dir.name)
if not match or not copy_dir.is_dir():
continue
copy_id = match.group(1)
for pdf_path in copy_dir.glob("*.pdf"):
source_pdfs.add((pdf_path.stem, copy_id))
return source_pdfs
def collect_grouped_pdfs(groups_dir: Path) -> tuple[set[tuple[str, str]], int]:
grouped: set[tuple[str, str]] = set()
read_errors = 0
for json_path in groups_dir.glob("*/Group_*.json"):
try:
data = read_json(json_path)
if not isinstance(data, list):
raise TypeError("expected a JSON array")
for entry in data:
grouped.add((str(entry[4]), str(entry[0])))
except (IndexError, OSError, TypeError, ValueError) as exc:
print(f"Error reading {json_path}: {exc}", file=sys.stderr)
read_errors += 1
return grouped, read_errors
def verify_groups(workspace: EvaluationWorkspace) -> ExitCode:
workspace.require_directories("Copies", "Par label")
source_pdfs = collect_source_pdfs(workspace.copies_dir)
grouped_pdfs, read_errors = collect_grouped_pdfs(workspace.groups_dir)
missing = source_pdfs - grouped_pdfs
if missing:
print(f"Verification failed: {len(missing)} files missing from groups:")
for label, copy_id in sorted(missing):
print(f"Copie{copy_id}/{label}.pdf")
return ExitCode.FAILURE
if read_errors:
print("Verification incomplete because some metadata could not be read.")
return ExitCode.PARTIAL
print("Verification successful: all files accounted for.")
return ExitCode.SUCCESS
def run(workspace: EvaluationWorkspace) -> ExitCode:
return verify_groups(workspace)
def main(argv: Sequence[str] | None = None) -> int:
parser = build_parser()
return execute(parser, argv, lambda args: run(workspace_from_args(args)))
if __name__ == "__main__":
raise SystemExit(main())
+36
View File
@@ -0,0 +1,36 @@
from __future__ import annotations
import importlib.util
import os
from pathlib import Path
from types import ModuleType
def _user_config_path() -> Path | None:
explicit = os.environ.get("COPIENATOR_CONFIG", "").strip()
if explicit:
return Path(explicit).expanduser().resolve()
local = Path.cwd() / "config.py"
return local.resolve() if local.is_file() else None
def _load_user_config(path: Path) -> ModuleType:
if not path.is_file():
raise FileNotFoundError(f"Copienator configuration not found: {path}")
spec = importlib.util.spec_from_file_location("_copienator_user_config", path)
if spec is None or spec.loader is None:
raise ImportError(f"Could not load Copienator configuration: {path}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
CONFIG_PATH = _user_config_path()
if CONFIG_PATH is None:
import default_config as _configuration
else:
_configuration = _load_user_config(CONFIG_PATH)
for _name in dir(_configuration):
if not _name.startswith("_"):
globals()[_name] = getattr(_configuration, _name)
File diff suppressed because it is too large Load Diff
+76
View File
@@ -0,0 +1,76 @@
from __future__ import annotations
import importlib
import sys
from collections.abc import Sequence
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class Command:
module: str
description: str
COMMANDS: dict[str, Command] = {
"statement": Command("gemini_for_enonce", "Analyse the exam statement with Gemini"),
"statement-personal": Command("enonce_info", "Generate personal statement metadata"),
"copies": Command("copies_tools", "Rotate or rename scanned copies"),
"page-split": Command("page_splitter", "Split and reorder scanned PDF pages"),
"crop-labels": Command("cutleft", "Crop the label margin from copies"),
"labels": Command("gemini_for_labels", "Detect question labels with Gemini"),
"review-labels": Command("plotting", "Review detected labels interactively"),
"split-answers": Command("splitting_int", "Split copies into answers"),
"group-answers": Command("grouping", "Group answers by question"),
"verify-groups": Command("verify_groups", "Verify grouped answer metadata"),
"correct": Command("correction", "Generate or integrate corrections"),
"batch-submit": Command("submit_batches", "Submit Gemini batch jobs"),
"batch-status": Command("batch_status", "Inspect Gemini batch jobs"),
"batch-fetch": Command("fetch_batched_results", "Fetch Gemini batch results"),
"post-correction": Command("post_correction", "Clean generated correction text"),
"resolve-manual": Command("resolve_manual", "Resolve manual label conflicts"),
"annotate-simple": Command("annotating", "Generate simple annotations"),
"annotate-checks": Command("annotating_with_checks", "Generate checkable annotations"),
"annotate-grouped": Command("annotating_by_label", "Generate grouped annotations"),
"export": Command("export", "Export annotations"),
"import": Command("import_annotations", "Import handwritten annotations"),
"read-annotations": Command("reading_annotations", "Read checkable annotations"),
"read-grouped": Command("reading_grouped_annotations", "Read grouped annotations"),
"giving-names": Command("giving_names", "Name copies and prepare A Rendre"),
"update-ods": Command("update_ods", "Update the configured score spreadsheet"),
"add-final-score": Command("add_final_score", "Stamp final scores on copies"),
"gui": Command("@gui", "Launch the graphical workflow assistant"),
}
def _print_help() -> None:
print("Usage: copienator COMMAND [ARGUMENTS]")
print(" python -m copienator COMMAND [ARGUMENTS]")
print("\nCommands:")
width = max(len(name) for name in COMMANDS)
for name, command in COMMANDS.items():
print(f" {name:<{width}} {command.description}")
print("\nUse 'copienator COMMAND --help' for command-specific help.")
def main(argv: Sequence[str] | None = None) -> int:
arguments = list(sys.argv[1:] if argv is None else argv)
if not arguments or arguments[0] in {"-h", "--help"}:
_print_help()
return 0
command_name = arguments.pop(0)
command = COMMANDS.get(command_name)
if command is None:
print(f"Unknown command: {command_name}", file=sys.stderr)
print("Use 'copienator --help' to list commands.", file=sys.stderr)
return 2
module_name = (
"copienator_gui.__main__"
if command.module == "@gui"
else f"copienator.commands.{command.module}"
)
module = importlib.import_module(module_name)
result = module.main(arguments)
return int(result or 0)
@@ -118,3 +118,4 @@ def replace_with_link_or_copy(
shutil.copy2(source_path, destination_path) shutil.copy2(source_path, destination_path)
return "copy" return "copy"
+3 -2
View File
@@ -1,6 +1,6 @@
from pathlib import Path from pathlib import Path
import io import io
import utils from . import utils
main_prompt = """I'm giving you an image of several written answers to an exam. main_prompt = """I'm giving you an image of several written answers to an exam.
@@ -86,7 +86,7 @@ Here is a possible correct answer :
You are asked to score the question or exercice labeled `<<label>>`, You are asked to score the question or exercice labeled `<<label>>`,
do not score or give feedback to any other question.""" do not score or give feedback to any other question."""
from utils import get_label_text_content, get_label_sol_content, get_label_persp_content from .utils import get_label_text_content, get_label_sol_content, get_label_persp_content
def make_prompt(input_dir,full_label): def make_prompt(input_dir,full_label):
text = get_label_text_content(input_dir, full_label) or "" text = get_label_text_content(input_dir, full_label) or ""
@@ -298,3 +298,4 @@ Here is a list of all possible labels. You need to answer with a list one of the
])] ])]
config = types.GenerateContentConfig(temperature=1.0, response_mime_type="application/json") config = types.GenerateContentConfig(temperature=1.0, response_mime_type="application/json")
return contents, config return contents, config
+3 -3
View File
@@ -2,7 +2,7 @@ import re
from pathlib import Path from pathlib import Path
from copienator import EvaluationWorkspace from copienator import EvaluationWorkspace
from platform_utils import validate_windows_labels from .platform import validate_windows_labels
def natural_key(text): def natural_key(text):
return [int(c) if c.isdigit() else c.lower() for c in re.split(r'(\d+)', str(text))] return [int(c) if c.isdigit() else c.lower() for c in re.split(r'(\d+)', str(text))]
@@ -30,7 +30,7 @@ def enonce_total(base_dir):
import os import os
import shlex import shlex
from platform_utils import open_path from .platform import open_path
def edit_file_and_enter(file): def edit_file_and_enter(file):
editor = os.environ.get("EDITOR") editor = os.environ.get("EDITOR")
@@ -139,7 +139,7 @@ import tempfile
import shutil import shutil
import subprocess import subprocess
from config import LATEX_AFTER, LATEX_BEFORE from .configuration import LATEX_AFTER, LATEX_BEFORE
def compile_to_pdf(text, output_pdf_path): def compile_to_pdf(text, output_pdf_path):
+8
View File
@@ -93,6 +93,14 @@ class EvaluationWorkspace:
def labels_file(self) -> Path: def labels_file(self) -> Path:
return self.root / "labels" return self.root / "labels"
@property
def label_groups_file(self) -> Path:
return self.root / "label_groups"
@property
def gemini_exam_items_file(self) -> Path:
return self.root / "Tmp" / "exam_items.txt"
@property @property
def correction_file(self) -> Path: def correction_file(self) -> Path:
return self.root / "correction.json" return self.root / "correction.json"
+7 -9
View File
@@ -8,25 +8,23 @@ from copienator_gui.app import CopienatorApp
def personal_steps_enabled() -> bool: def personal_steps_enabled() -> bool:
try: try:
from config import SHOW_PERSONAL_STEPS from copienator.configuration import SHOW_PERSONAL_STEPS
except (ImportError, AttributeError): except (ImportError, AttributeError):
try: return False
from default_config import SHOW_PERSONAL_STEPS
except (ImportError, AttributeError):
return False
return bool(SHOW_PERSONAL_STEPS) return bool(SHOW_PERSONAL_STEPS)
def main() -> None: def main(argv=None) -> int:
parser = argparse.ArgumentParser(description="Interface graphique du workflow Copienator") parser = argparse.ArgumentParser(description="Interface graphique du workflow Copienator")
parser.add_argument("evaluation", nargs="?", type=Path, help="Dossier d’évaluation à ouvrir") parser.add_argument("evaluation", nargs="?", type=Path, help="Dossier d’évaluation à ouvrir")
args = parser.parse_args() args = parser.parse_args(argv)
repository = Path(__file__).resolve().parent repository = Path.cwd().resolve()
evaluation = args.evaluation.resolve() if args.evaluation else None evaluation = args.evaluation.resolve() if args.evaluation else None
app = CopienatorApp(repository, personal_steps_enabled(), evaluation) app = CopienatorApp(repository, personal_steps_enabled(), evaluation)
app.mainloop() app.mainloop()
return 0
if __name__ == "__main__": if __name__ == "__main__":
main() raise SystemExit(main())
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+287 -24
View File
@@ -8,7 +8,7 @@ from tkinter import filedialog, messagebox, ttk
from typing import Any from typing import Any
from copienator import ExitCode from copienator import ExitCode
from platform_utils import WindowsLabelError, validate_windows_labels from copienator.platform import WindowsLabelError, open_path, validate_windows_labels
from .diagnostics import collect_diagnostics from .diagnostics import collect_diagnostics
from .runner import ProcessRunner from .runner import ProcessRunner
@@ -35,6 +35,77 @@ STATUS_LABELS = {
"detected": "Détectée", "detected": "Détectée",
} }
DEFAULT_HTTPS_PROXY = "http://10.0.0.1:3128"
ANNOTATION_DIRECTORIES = ("BGnot", "Bnot", "Anot")
ANNOTATION_VARIANT_DIRECTORIES = {
"simple": "Anot",
"checks": "Bnot",
"grouped": "BGnot",
}
def copy_pdf_paths(evaluation: Path) -> list[Path]:
"""List the most relevant version of each scanned copy."""
locations = (evaluation / "Copies", evaluation, evaluation / "Copies Originales")
for location in locations:
if not location.is_dir():
continue
copies = sorted(
(
path
for path in location.glob("*.pdf")
if path.name.casefold() not in {"enonce.pdf", "énoncé.pdf"}
),
key=lambda path: path.name.casefold(),
)
if copies:
return copies
return []
def detected_annotation_directories(evaluation: Path) -> tuple[str, ...]:
return tuple(
name for name in ANNOTATION_DIRECTORIES if (evaluation / name).is_dir()
)
def plotting_shortcut_lines() -> list[str]:
try:
from copienator.configuration import PLOTTING_KB
except (ImportError, AttributeError):
from default_config import PLOTTING_KB
labels = (
("OK", "valider et passer à la suivante"),
("previous", "revenir à la précédente"),
("edit", "éditer le fichier JSON"),
("open pdf", "ouvrir la copie traitée"),
("open original pdf", "ouvrir la copie originale"),
("open eval", "ouvrir l’énoncé"),
)
display_names = {"<Return>": "Entrée", "<Escape>": "Échap"}
lines = [
f"{display_names.get(PLOTTING_KB[action], PLOTTING_KB[action])} : {description}"
for action, description in labels
]
lines.append("Échap : fermer la fenêtre")
return lines
def build_runner_environment(
base: dict[str, str], api_key: str, proxy: str, use_proxy: bool
) -> dict[str, str]:
environment = dict(base)
environment["PYTHONUNBUFFERED"] = "1"
environment["PYTHONIOENCODING"] = "utf-8"
if api_key.strip():
environment["GEMINI_API_KEY"] = api_key.strip()
environment.pop("HTTPS_PROXY", None)
environment.pop("https_proxy", None)
if use_proxy and proxy.strip():
environment["HTTPS_PROXY"] = proxy.strip()
return environment
def process_status(return_code: int, interrupted: bool = False) -> str: def process_status(return_code: int, interrupted: bool = False) -> str:
if interrupted or return_code == ExitCode.INTERRUPTED: if interrupted or return_code == ExitCode.INTERRUPTED:
@@ -46,6 +117,19 @@ def process_status(return_code: int, interrupted: bool = False) -> str:
return "failed" return "failed"
def has_manual_conflicts(path: Path) -> bool:
"""Return whether a manual-resolution file contains an instruction."""
try:
lines = path.read_text(encoding="utf-8").splitlines()
except FileNotFoundError:
return False
except (OSError, UnicodeError):
# If the file cannot be inspected, keep the step visible rather than
# silently claiming that there is nothing to resolve.
return True
return any(line.strip() and not line.lstrip().startswith("###") for line in lines)
class CopienatorApp(tk.Tk): class CopienatorApp(tk.Tk):
def __init__( def __init__(
self, self,
@@ -63,6 +147,7 @@ class CopienatorApp(tk.Tk):
self.active_step_id: str | None = None self.active_step_id: str | None = None
self.current_step: StepDefinition | None = None self.current_step: StepDefinition | None = None
self.arg_vars: dict[str, tk.Variable] = {} self.arg_vars: dict[str, tk.Variable] = {}
self.copy_paths: dict[str, Path] = {}
self._rendering = False self._rendering = False
self.title("Copienator — assistant de correction") self.title("Copienator — assistant de correction")
@@ -72,7 +157,9 @@ class CopienatorApp(tk.Tk):
self.evaluation_var = tk.StringVar() self.evaluation_var = tk.StringVar()
self.api_key_var = tk.StringVar(value=os.environ.get("GEMINI_API_KEY", "")) self.api_key_var = tk.StringVar(value=os.environ.get("GEMINI_API_KEY", ""))
self.proxy_var = tk.StringVar(value=os.environ.get("HTTPS_PROXY", "")) self.proxy_var = tk.StringVar(value=DEFAULT_HTTPS_PROXY)
self.use_proxy_var = tk.BooleanVar(value=False)
self.copy_var = tk.StringVar()
self.variant_var = tk.StringVar() self.variant_var = tk.StringVar()
self.extra_var = tk.StringVar() self.extra_var = tk.StringVar()
self.command_var = tk.StringVar(value="Sélectionnez une étape.") self.command_var = tk.StringVar(value="Sélectionnez une étape.")
@@ -100,13 +187,17 @@ class CopienatorApp(tk.Tk):
top = ttk.Frame(self, padding=(10, 8)) top = ttk.Frame(self, padding=(10, 8))
top.grid(row=0, column=0, sticky="ew") top.grid(row=0, column=0, sticky="ew")
top.columnconfigure(1, weight=1) top.columnconfigure(2, weight=1)
ttk.Label(top, text="Évaluation").grid(row=0, column=0, sticky="w", padx=(0, 8)) ttk.Button(top, text="Charger", command=self._load_evaluation).grid(
row=0, column=0, padx=(0, 8)
)
ttk.Label(top, text="Évaluation").grid(row=0, column=1, sticky="w", padx=(0, 8))
path_entry = ttk.Entry(top, textvariable=self.evaluation_var) path_entry = ttk.Entry(top, textvariable=self.evaluation_var)
path_entry.grid(row=0, column=1, sticky="ew") path_entry.grid(row=0, column=2, sticky="ew")
path_entry.bind("<Return>", lambda _event: self._load_evaluation()) path_entry.bind("<Return>", lambda _event: self._load_evaluation())
ttk.Button(top, text="Parcourir…", command=self._browse_evaluation).grid(row=0, column=2, padx=6) ttk.Button(top, text="Parcourir…", command=self._browse_evaluation).grid(
ttk.Button(top, text="Charger", command=self._load_evaluation).grid(row=0, column=3) row=0, column=3, padx=6
)
ttk.Button(top, text="Diagnostic…", command=self._show_diagnostics).grid(row=0, column=4, padx=(6, 0)) ttk.Button(top, text="Diagnostic…", command=self._show_diagnostics).grid(row=0, column=4, padx=(6, 0))
ttk.Label(top, text="Clé Gemini").grid(row=1, column=0, sticky="w", padx=(0, 8), pady=(7, 0)) ttk.Label(top, text="Clé Gemini").grid(row=1, column=0, sticky="w", padx=(0, 8), pady=(7, 0))
@@ -114,9 +205,15 @@ class CopienatorApp(tk.Tk):
row=1, column=1, sticky="w", pady=(7, 0) row=1, column=1, sticky="w", pady=(7, 0)
) )
environment = ttk.Frame(top) environment = ttk.Frame(top)
environment.grid(row=1, column=2, columnspan=2, sticky="e", pady=(7, 0)) environment.grid(row=1, column=2, columnspan=3, sticky="e", pady=(7, 0))
ttk.Label(environment, text="HTTPS_PROXY").pack(side="left", padx=(0, 6)) ttk.Checkbutton(
ttk.Entry(environment, textvariable=self.proxy_var, width=28).pack(side="left") environment,
text="Utiliser le proxy HTTPS",
variable=self.use_proxy_var,
command=self._toggle_proxy,
).pack(side="left", padx=(0, 6))
self.proxy_entry = ttk.Entry(environment, textvariable=self.proxy_var, width=28, state="disabled")
self.proxy_entry.pack(side="left")
profile = "standard + personnel" if show_personal_steps else "standard" profile = "standard + personnel" if show_personal_steps else "standard"
ttk.Label(environment, text=f"Profil : {profile}").pack(side="left", padx=(12, 0)) ttk.Label(environment, text=f"Profil : {profile}").pack(side="left", padx=(12, 0))
@@ -256,6 +353,9 @@ class CopienatorApp(tk.Tk):
self.evaluation_var.set(selected) self.evaluation_var.set(selected)
self._load_evaluation() self._load_evaluation()
def _toggle_proxy(self) -> None:
self.proxy_entry.configure(state="normal" if self.use_proxy_var.get() else "disabled")
def _load_evaluation(self) -> None: def _load_evaluation(self) -> None:
evaluation = self.evaluation evaluation = self.evaluation
if not evaluation or not evaluation.is_dir(): if not evaluation or not evaluation.is_dir():
@@ -354,6 +454,54 @@ class CopienatorApp(tk.Tk):
self._save_current_form() self._save_current_form()
self.current_step = step self.current_step = step
self._render_step() self._render_step()
self._handle_first_visit(step)
def _handle_first_visit(self, step: StepDefinition) -> None:
if not self.state_store.evaluation:
return
entry = self.state_store.step(step.id)
if entry.get("visited"):
return
# Persist this before scheduling an action so selection callbacks cannot
# trigger the same automatic behavior twice.
self.state_store.update_step(step.id, visited=True)
if step.skip_for_live_correction:
correction = self.state_store.step("correction")
if correction.get("variant", "live") == "live":
self.after_idle(
lambda step_id=step.id: self._automatic_skip(
step_id, "correction immédiate sélectionnée"
)
)
return
if step.skip_without_manual_conflicts:
evaluation = self.evaluation
conflicts = evaluation / "manual_resolutions.txt" if evaluation else None
if conflicts is None or not has_manual_conflicts(conflicts):
self.after_idle(
lambda step_id=step.id: self._automatic_skip(
step_id, "aucun conflit manuel détecté"
)
)
return
if step.auto_start_first_visit and not entry.get("status") and not self._artifacts_exist(step):
self.after_idle(lambda step_id=step.id: self._automatic_start(step_id))
def _automatic_start(self, step_id: str) -> None:
if self.runner.running or not self.current_step or self.current_step.id != step_id:
return
self.info_var.set(f"{self.current_step.title} : démarrage automatique.")
self._run_current_step()
def _automatic_skip(self, step_id: str, reason: str) -> None:
if self.runner.running or not self.current_step or self.current_step.id != step_id:
return
self._mark_step("skipped", automatic=True, reason=reason)
self._move_selection_from(step_id, 1)
def _render_step(self) -> None: def _render_step(self) -> None:
step = self.current_step step = self.current_step
@@ -376,7 +524,7 @@ class CopienatorApp(tk.Tk):
description += "\nPrérequis non détectés : " + ", ".join(missing) description += "\nPrérequis non détectés : " + ", ".join(missing)
self.description_label.configure(text=description) self.description_label.configure(text=description)
row = 0 row = self._render_context_controls(step, 0)
if len(step.variants) > 1: if len(step.variants) > 1:
ttk.Label(self.form, text="Mode").grid(row=row, column=0, sticky="w", pady=4, padx=(0, 8)) ttk.Label(self.form, text="Mode").grid(row=row, column=0, sticky="w", pady=4, padx=(0, 8))
labels = [variant.label for variant in step.variants] labels = [variant.label for variant in step.variants]
@@ -394,6 +542,9 @@ class CopienatorApp(tk.Tk):
if spec.variants and variant.id not in spec.variants: if spec.variants and variant.id not in spec.variants:
continue continue
value = values.get(spec.name, value_for_default(spec.default, evaluation_arg)) value = values.get(spec.name, value_for_default(spec.default, evaluation_arg))
choices = spec.choices
if spec.name == "annotation_dir" and step.id in {"export", "import"}:
choices, value = self._annotation_directory_choices(step.id)
ttk.Label(self.form, text=spec.label).grid(row=row, column=0, sticky="nw", pady=4, padx=(0, 8)) ttk.Label(self.form, text=spec.label).grid(row=row, column=0, sticky="nw", pady=4, padx=(0, 8))
if spec.kind == "bool": if spec.kind == "bool":
variable: tk.Variable = tk.BooleanVar(value=bool(value)) variable: tk.Variable = tk.BooleanVar(value=bool(value))
@@ -401,7 +552,7 @@ class CopienatorApp(tk.Tk):
widget.grid(row=row, column=1, sticky="w", pady=4) widget.grid(row=row, column=1, sticky="w", pady=4)
elif spec.kind == "choice": elif spec.kind == "choice":
variable = tk.StringVar(value=str(value)) variable = tk.StringVar(value=str(value))
widget = ttk.Combobox(self.form, textvariable=variable, values=spec.choices, state="readonly") widget = ttk.Combobox(self.form, textvariable=variable, values=choices, state="readonly")
widget.grid(row=row, column=1, sticky="ew", pady=4) widget.grid(row=row, column=1, sticky="ew", pady=4)
else: else:
variable = tk.StringVar(value=str(value)) variable = tk.StringVar(value=str(value))
@@ -438,6 +589,93 @@ class CopienatorApp(tk.Tk):
self._update_command_preview() self._update_command_preview()
self._update_controls() self._update_controls()
def _render_context_controls(self, step: StepDefinition, row: int) -> int:
if step.id == "review_persp":
ttk.Button(
self.form,
text="Ouvrir le dossier Persp",
command=self._open_persp,
).grid(row=row, column=0, columnspan=2, sticky="w", pady=(0, 8))
row += 1
if step.section == "Prétraitement des copies":
evaluation = self.evaluation
paths = copy_pdf_paths(evaluation) if evaluation else []
self.copy_paths = {path.name: path for path in paths}
names = list(self.copy_paths)
copies = ttk.LabelFrame(self.form, text="Copies détectées", padding=7)
copies.grid(row=row, column=0, columnspan=2, sticky="ew", pady=(0, 8))
copies.columnconfigure(0, weight=1)
summary = (
f"{len(names)} copie(s) : {', '.join(names)}"
if names
else "Aucune copie PDF détectée."
)
ttk.Label(copies, text=summary, wraplength=570, justify="left").grid(
row=0, column=0, columnspan=2, sticky="ew"
)
self.copy_var.set(names[0] if names else "")
selector = ttk.Combobox(
copies,
textvariable=self.copy_var,
values=names,
state="readonly" if names else "disabled",
)
selector.grid(row=1, column=0, sticky="ew", pady=(7, 0))
ttk.Button(
copies,
text="Afficher la copie",
command=self._open_selected_copy,
state="normal" if names else "disabled",
).grid(row=1, column=1, padx=(6, 0), pady=(7, 0))
row += 1
if step.id == "plotting":
shortcuts = ttk.LabelFrame(self.form, text="Raccourcis clavier", padding=7)
shortcuts.grid(row=row, column=0, columnspan=2, sticky="ew", pady=(0, 8))
ttk.Label(
shortcuts,
text="\n".join(plotting_shortcut_lines()),
justify="left",
).grid(row=0, column=0, sticky="w")
row += 1
return row
def _annotation_directory_choices(self, step_id: str) -> tuple[tuple[str, ...], str]:
evaluation = self.evaluation
detected = detected_annotation_directories(evaluation) if evaluation else ()
choices = detected or ANNOTATION_DIRECTORIES
if step_id == "export":
annotation = self.state_store.step("annotation")
variant = annotation.get("last_run_variant", annotation.get("variant", "grouped"))
preferred = ANNOTATION_VARIANT_DIRECTORIES.get(str(variant), "BGnot")
else:
export = self.state_store.step("export")
last_values = export.get("last_run_values", export.get("values", {}))
preferred = (
str(last_values.get("annotation_dir", "BGnot"))
if isinstance(last_values, dict)
else "BGnot"
)
return choices, preferred if preferred in choices else choices[0]
def _open_persp(self) -> None:
evaluation = self.evaluation
self._open_desktop_path(evaluation / "Persp" if evaluation else None, "dossier Persp")
def _open_selected_copy(self) -> None:
self._open_desktop_path(self.copy_paths.get(self.copy_var.get()), "copie")
def _open_desktop_path(self, path: Path | None, label: str) -> None:
if path is None or not path.exists():
messagebox.showerror("Élément introuvable", f"Le {label} nexiste pas encore.")
return
try:
open_path(path)
except (OSError, RuntimeError) as exc:
messagebox.showerror("Ouverture impossible", str(exc))
def _select_variant(self, index: int) -> None: def _select_variant(self, index: int) -> None:
if not self.current_step or index < 0: if not self.current_step or index < 0:
return return
@@ -571,21 +809,27 @@ class CopienatorApp(tk.Tk):
return return
self._save_current_form() self._save_current_form()
run_values = self._values()
ordered_ids = [item.id for item in self.steps] ordered_ids = [item.id for item in self.steps]
self.state_store.invalidate_after(ordered_ids, step.id) self.state_store.invalidate_after(ordered_ids, step.id)
self.state_store.update_step(step.id, status="running", command=command_display(command)) self.state_store.update_step(
step.id,
status="running",
command=command_display(command),
last_run_variant=variant.id,
last_run_values=run_values,
)
self.active_step_id = step.id self.active_step_id = step.id
workspace = self.state_store.workspace workspace = self.state_store.workspace
assert workspace is not None assert workspace is not None
log_path = workspace.log_path(step.id) log_path = workspace.log_path(step.id)
environment = os.environ.copy() environment = build_runner_environment(
environment["PYTHONUNBUFFERED"] = "1" os.environ,
environment["PYTHONIOENCODING"] = "utf-8" self.api_key_var.get(),
if self.api_key_var.get().strip(): self.proxy_var.get(),
environment["GEMINI_API_KEY"] = self.api_key_var.get().strip() self.use_proxy_var.get(),
if self.proxy_var.get().strip(): )
environment["HTTPS_PROXY"] = self.proxy_var.get().strip()
self._append_console(f"\n$ {command_display(command)}\n") self._append_console(f"\n$ {command_display(command)}\n")
try: try:
@@ -601,15 +845,27 @@ class CopienatorApp(tk.Tk):
self._populate_tree() self._populate_tree()
self._update_controls() self._update_controls()
def _mark_step(self, status: str) -> None: def _mark_step(
self, status: str, *, automatic: bool = False, reason: str | None = None
) -> None:
if not self.current_step or not self.state_store.evaluation: if not self.current_step or not self.state_store.evaluation:
return return
self._save_current_form() self._save_current_form()
self.state_store.invalidate_after([item.id for item in self.steps], self.current_step.id) self.state_store.invalidate_after([item.id for item in self.steps], self.current_step.id)
self.state_store.update_step(self.current_step.id, status=status) self.state_store.update_step(self.current_step.id, status=status)
self.state_store.add_history({"step": self.current_step.id, "status": status, "manual": True}) history: dict[str, object] = {
"step": self.current_step.id,
"status": status,
"manual": not automatic,
}
if reason:
history["reason"] = reason
self.state_store.add_history(history)
self._populate_tree() self._populate_tree()
self.info_var.set(f"{self.current_step.title} : {STATUS_LABELS.get(status, status).lower()}.") detail = f" ({reason})" if reason else ""
self.info_var.set(
f"{self.current_step.title} : {STATUS_LABELS.get(status, status).lower()}{detail}."
)
def _skip_step(self) -> None: def _skip_step(self) -> None:
if self.current_step and self.current_step.optional: if self.current_step and self.current_step.optional:
@@ -652,6 +908,8 @@ class CopienatorApp(tk.Tk):
self.active_step_id = None self.active_step_id = None
self._populate_tree() self._populate_tree()
self._update_controls() self._update_controls()
if status == "success":
self.after_idle(lambda completed_id=step_id: self._move_selection_from(completed_id, 1))
def _send_input(self) -> None: def _send_input(self) -> None:
text = self.stdin_var.get() text = self.stdin_var.get()
@@ -703,12 +961,17 @@ class CopienatorApp(tk.Tk):
def _move_selection(self, delta: int) -> None: def _move_selection(self, delta: int) -> None:
if not self.current_step: if not self.current_step:
return return
self._move_selection_from(self.current_step.id, delta)
def _move_selection_from(self, step_id: str, delta: int) -> None:
ids = [step.id for step in self.steps] ids = [step.id for step in self.steps]
try: try:
index = ids.index(self.current_step.id) index = ids.index(step_id)
except ValueError: except ValueError:
return return
target = max(0, min(len(ids) - 1, index + delta)) target = max(0, min(len(ids) - 1, index + delta))
if target == index:
return
self.tree.selection_set(ids[target]) self.tree.selection_set(ids[target])
self.tree.see(ids[target]) self.tree.see(ids[target])
+2 -2
View File
@@ -7,7 +7,7 @@ import sys
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from platform_utils import windows_filename_problems from copienator.platform import windows_filename_problems
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -111,7 +111,7 @@ def collect_diagnostics(
) )
) )
try: try:
from config import CURRENT_SCORE_ODS_PATH, FINAL_SCORE_ODS_PATH from copienator.configuration import CURRENT_SCORE_ODS_PATH, FINAL_SCORE_ODS_PATH
for label, configured_path in ( for label, configured_path in (
("ODS courant", CURRENT_SCORE_ODS_PATH), ("ODS courant", CURRENT_SCORE_ODS_PATH),
+61 -36
View File
@@ -46,6 +46,9 @@ class StepDefinition:
personal: bool = False personal: bool = False
requires: tuple[str, ...] = () requires: tuple[str, ...] = ()
artifacts: tuple[str, ...] = () artifacts: tuple[str, ...] = ()
auto_start_first_visit: bool = False
skip_for_live_correction: bool = False
skip_without_manual_conflicts: bool = False
@property @property
def is_manual(self) -> bool: def is_manual(self) -> bool:
@@ -87,8 +90,8 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Analyser l’énoncé", "Analyser l’énoncé",
"Détecte les questions, leurs groupes, le corrigé et les indications de barème.", "Détecte les questions, leurs groupes, le corrigé et les indications de barème.",
( (
python("gemini", "Analyse avec Gemini", "gemini_for_enonce.py"), python("gemini", "Analyse avec Gemini", "statement"),
python("personal", "Alternative enonce_info.py", "enonce_info.py"), python("personal", "Alternative personnelle", "statement-personal"),
), ),
arguments=( arguments=(
arg_target("Dossier de l’évaluation"), arg_target("Dossier de l’évaluation"),
@@ -106,7 +109,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
StepDefinition( StepDefinition(
"review_persp", "review_persp",
"Prétraitement de l’énoncé", "Prétraitement de l’énoncé",
"Relire les barèmes dans Persp", "Relire les barèmes",
"Étape manuelle facultative : vérifier et modifier les instructions de correction.", "Étape manuelle facultative : vérifier et modifier les instructions de correction.",
(manual("review"),), (manual("review"),),
optional=True, optional=True,
@@ -121,7 +124,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
CommandVariant( CommandVariant(
"rotate", "rotate",
"Rotation", "Rotation",
"copies_tools.py", "copies",
"python", "python",
("rotate",), ("rotate",),
fixed_args_before_positionals=True, fixed_args_before_positionals=True,
@@ -139,7 +142,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
CommandVariant( CommandVariant(
"rename", "rename",
"Renommage", "Renommage",
"copies_tools.py", "copies",
"python", "python",
("rename",), ("rename",),
fixed_args_before_positionals=True, fixed_args_before_positionals=True,
@@ -152,7 +155,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Prétraitement des copies", "Prétraitement des copies",
"Séparer et réordonner les pages", "Séparer et réordonner les pages",
"Ouvre loutil interactif de découpage A3 vers A4. La cible peut être un dossier ou un PDF.", "Ouvre loutil interactif de découpage A3 vers A4. La cible peut être un dossier ou un PDF.",
(python("default", "Séparation des pages", "page_splitter.py"),), (python("default", "Séparation des pages", "page-split"),),
arguments=(arg_target(),), arguments=(arg_target(),),
artifacts=("Copies", "Copies Originales"), artifacts=("Copies", "Copies Originales"),
), ),
@@ -161,7 +164,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Prétraitement des copies", "Prétraitement des copies",
"Découper la marge des labels", "Découper la marge des labels",
"Produit les images de la partie gauche des copies. Une copie précise peut être ciblée.", "Produit les images de la partie gauche des copies. Une copie précise peut être ciblée.",
(python("default", "Découpe", "cutleft.py"),), (python("default", "Découpe", "crop-labels"),),
arguments=( arguments=(
arg_target(), arg_target(),
ArgumentSpec("fullpage", "Toujours utiliser la page entière", "bool", "--fullpage"), ArgumentSpec("fullpage", "Toujours utiliser la page entière", "bool", "--fullpage"),
@@ -174,7 +177,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Labels et regroupement", "Labels et regroupement",
"Détecter les labels avec Gemini", "Détecter les labels avec Gemini",
"Identifie les labels dans les images produites par la découpe gauche.", "Identifie les labels dans les images produites par la découpe gauche.",
(python("default", "Détection des labels", "gemini_for_labels.py"),), (python("default", "Détection des labels", "labels"),),
arguments=( arguments=(
arg_target(), arg_target(),
ArgumentSpec("overwrite", "Régénérer les résultats", "bool", "--overwrite"), ArgumentSpec("overwrite", "Régénérer les résultats", "bool", "--overwrite"),
@@ -187,7 +190,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Labels et regroupement", "Labels et regroupement",
"Vérifier visuellement les labels", "Vérifier visuellement les labels",
"Ouvre la fenêtre de vérification. La cible peut être toute l’évaluation ou une copie.", "Ouvre la fenêtre de vérification. La cible peut être toute l’évaluation ou une copie.",
(python("default", "Vérification", "plotting.py"),), (python("default", "Vérification", "review-labels"),),
arguments=(arg_target(),), arguments=(arg_target(),),
requires=("labels", "Cutleft"), requires=("labels", "Cutleft"),
artifacts=("Copies/Copie*.json",), artifacts=("Copies/Copie*.json",),
@@ -197,27 +200,29 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Labels et regroupement", "Labels et regroupement",
"Découper les réponses par question", "Découper les réponses par question",
"Découpe les copies à partir des coordonnées de labels vérifiées.", "Découpe les copies à partir des coordonnées de labels vérifiées.",
(python("default", "Découpage", "splitting_int.py"),), (python("default", "Découpage", "split-answers"),),
arguments=(arg_target(),), arguments=(arg_target(),),
requires=("Copies",), requires=("Copies",),
artifacts=("Copies/Copie*/*",), artifacts=("Copies/Copie*/*",),
auto_start_first_visit=True,
), ),
StepDefinition( StepDefinition(
"grouping", "grouping",
"Labels et regroupement", "Labels et regroupement",
"Regrouper les réponses", "Regrouper les réponses",
"Regroupe les réponses portant le même label pour préparer les requêtes.", "Regroupe les réponses portant le même label pour préparer les requêtes.",
(python("default", "Regroupement", "grouping.py"),), (python("default", "Regroupement", "group-answers"),),
arguments=(arg_target("Dossier de l’évaluation"),), arguments=(arg_target("Dossier de l’évaluation"),),
requires=("Copies",), requires=("Copies",),
artifacts=("Par label",), artifacts=("Par label",),
auto_start_first_visit=True,
), ),
StepDefinition( StepDefinition(
"verify_groups", "verify_groups",
"Labels et regroupement", "Labels et regroupement",
"Vérifier les groupes produits", "Vérifier les groupes produits",
"Vérifie que chaque réponse PDF apparaît dans les métadonnées des groupes.", "Vérifie que chaque réponse PDF apparaît dans les métadonnées des groupes.",
(python("default", "Vérification des groupes", "verify_groups.py"),), (python("default", "Vérification des groupes", "verify-groups"),),
arguments=(arg_target("Dossier de l’évaluation"),), arguments=(arg_target("Dossier de l’évaluation"),),
optional=True, optional=True,
requires=("Copies", "Par label"), requires=("Copies", "Par label"),
@@ -228,17 +233,17 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Lancer ou intégrer la correction", "Lancer ou intégrer la correction",
"Choisir une correction immédiate, batch, hybride, une recorrection, ou lintégration dun batch.", "Choisir une correction immédiate, batch, hybride, une recorrection, ou lintégration dun batch.",
( (
python("live", "Correction immédiate", "correction.py"), python("live", "Correction immédiate", "correct"),
python("batch", "Préparer toutes les requêtes batch", "correction.py", fixed_args=("--batch",)), python("batch", "Préparer toutes les requêtes batch", "correct", fixed_args=("--batch",)),
python("hybrid", "Batch à partir dun label", "correction.py"), python("hybrid", "Batch à partir dun label", "correct"),
python("refaire", "Recorrection depuis refaire.json", "correction.py", fixed_args=("--refaire",)), python("refaire", "Recorrection depuis refaire.json", "correct", fixed_args=("--refaire",)),
python( python(
"integrate", "integrate",
"Intégrer les résultats batch", "Intégrer les résultats batch",
"correction.py", "correct",
fixed_args=("--deal-with-batched",), fixed_args=("--deal-with-batched",),
), ),
python("reset", "Réinitialiser les corrections", "correction.py", fixed_args=("--reset",), dangerous=True), python("reset", "Réinitialiser les corrections", "correct", fixed_args=("--reset",), dangerous=True),
), ),
arguments=( arguments=(
arg_target("Évaluation ou image Group_X.jpg"), arg_target("Évaluation ou image Group_X.jpg"),
@@ -254,35 +259,38 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Correction", "Correction",
"Envoyer les batchs", "Envoyer les batchs",
"Envoie à Gemini les fichiers JSONL produits par le mode batch.", "Envoie à Gemini les fichiers JSONL produits par le mode batch.",
(python("default", "Envoi", "submit_batches.py"),), (python("default", "Envoi", "batch-submit"),),
arguments=(arg_target("Dossier de l’évaluation"),), arguments=(arg_target("Dossier de l’évaluation"),),
optional=True, optional=True,
artifacts=("batch_jobs.json",), artifacts=("batch_jobs.json",),
skip_for_live_correction=True,
), ),
StepDefinition( StepDefinition(
"batch_status", "batch_status",
"Correction", "Correction",
"Consulter l’état des batchs", "Consulter l’état des batchs",
"Affiche les jobs Gemini en cours. Lidentifiant de téléchargement est facultatif.", "Affiche les jobs Gemini en cours. Lidentifiant de téléchargement est facultatif.",
(python("default", "État des batchs", "batch_status.py"),), (python("default", "État des batchs", "batch-status"),),
arguments=(ArgumentSpec("download", "Télécharger le job", "text", "--download"),), arguments=(ArgumentSpec("download", "Télécharger le job", "text", "--download"),),
optional=True, optional=True,
skip_for_live_correction=True,
), ),
StepDefinition( StepDefinition(
"fetch_batches", "fetch_batches",
"Correction", "Correction",
"Récupérer les résultats batch", "Récupérer les résultats batch",
"Télécharge et rassemble les réponses des jobs terminés.", "Télécharge et rassemble les réponses des jobs terminés.",
(python("default", "Récupération", "fetch_batched_results.py"),), (python("default", "Récupération", "batch-fetch"),),
arguments=(arg_target("Dossier de l’évaluation"),), arguments=(arg_target("Dossier de l’évaluation"),),
optional=True, optional=True,
skip_for_live_correction=True,
), ),
StepDefinition( StepDefinition(
"post_correction", "post_correction",
"Correction", "Correction",
"Nettoyer la correction", "Nettoyer la correction",
"Corrige certains problèmes dencodage et prépare le texte pour LaTeX.", "Corrige certains problèmes dencodage et prépare le texte pour LaTeX.",
(python("default", "Post-correction", "post-correction.py"),), (python("default", "Post-correction", "post-correction"),),
arguments=(arg_target("Dossier de l’évaluation"),), arguments=(arg_target("Dossier de l’évaluation"),),
requires=("correction.json",), requires=("correction.json",),
), ),
@@ -291,10 +299,11 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Correction", "Correction",
"Résoudre les conflits manuels", "Résoudre les conflits manuels",
"Étape conditionnelle, uniquement si manual_resolutions.txt contient des conflits à traiter.", "Étape conditionnelle, uniquement si manual_resolutions.txt contient des conflits à traiter.",
(python("default", "Résolution", "resolve_manual.py"),), (python("default", "Résolution", "resolve-manual"),),
arguments=(arg_target("Dossier de l’évaluation"),), arguments=(arg_target("Dossier de l’évaluation"),),
optional=True, optional=True,
requires=("manual_resolutions.txt", "correction.json"), requires=("manual_resolutions.txt", "correction.json"),
skip_without_manual_conflicts=True,
), ),
StepDefinition( StepDefinition(
"annotation", "annotation",
@@ -302,9 +311,9 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Générer les copies annotées", "Générer les copies annotées",
"Les trois modes sont exclusifs pour un parcours donné.", "Les trois modes sont exclusifs pour un parcours donné.",
( (
python("simple", "Annotations simples (Anot)", "annotating.py"), python("simple", "Annotations simples (Anot)", "annotate-simple"),
python("checks", "Annotations avec cases (Bnot)", "annotating_with_checks.py"), python("checks", "Annotations avec cases (Bnot)", "annotate-checks"),
python("grouped", "Annotations groupées (BGnot)", "annotating_by_label.py"), python("grouped", "Annotations groupées (BGnot)", "annotate-grouped"),
), ),
arguments=( arguments=(
arg_target("Dossier de l’évaluation"), arg_target("Dossier de l’évaluation"),
@@ -317,11 +326,19 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
StepDefinition( StepDefinition(
"export", "export",
"Génération des annotations", "Génération des annotations",
"Exporter vers la tablette", "Exporter",
"Exporte les groupes vers le dossier EXPORT_DIR défini dans config.py.", "Exporte les annotations vers le dossier EXPORT_DIR défini dans config.py.",
(python("default", "Export", "export.py"),), (python("default", "Export", "export"),),
arguments=( arguments=(
arg_target("Dossier de l’évaluation"), arg_target("Dossier de l’évaluation"),
ArgumentSpec(
"annotation_dir",
"Dossier dannotations",
"choice",
default="BGnot",
choices=("BGnot", "Bnot", "Anot"),
positional=True,
),
ArgumentSpec("refaire", "Mode refaire", "bool", "--refaire"), ArgumentSpec("refaire", "Mode refaire", "bool", "--refaire"),
), ),
optional=True, optional=True,
@@ -338,9 +355,17 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Correction manuscrite", "Correction manuscrite",
"Importer les annotations manuscrites", "Importer les annotations manuscrites",
"Copie les PDF présents dans IMPORT_DIR vers l’évaluation.", "Copie les PDF présents dans IMPORT_DIR vers l’évaluation.",
(python("default", "Import", "import.py"),), (python("default", "Import", "import"),),
arguments=( arguments=(
arg_target("Dossier de l’évaluation"), arg_target("Dossier de l’évaluation"),
ArgumentSpec(
"annotation_dir",
"Dossier dannotations",
"choice",
default="BGnot",
choices=("BGnot", "Bnot", "Anot"),
positional=True,
),
ArgumentSpec("refaire", "Mode refaire", "bool", "--refaire"), ArgumentSpec("refaire", "Mode refaire", "bool", "--refaire"),
), ),
), ),
@@ -350,8 +375,8 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Lire les annotations manuscrites", "Lire les annotations manuscrites",
"Le mode doit correspondre au mode choisi lors de la génération des annotations.", "Le mode doit correspondre au mode choisi lors de la génération des annotations.",
( (
python("standard", "Lecture Bnot", "reading_annotations.py"), python("standard", "Lecture Bnot", "read-annotations"),
python("grouped", "Lecture BGnot", "reading_grouped_annotations.py"), python("grouped", "Lecture BGnot", "read-grouped"),
), ),
arguments=( arguments=(
arg_target("Dossier de l’évaluation"), arg_target("Dossier de l’évaluation"),
@@ -364,7 +389,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Finalisation", "Finalisation",
"Attribuer les noms et préparer A Rendre", "Attribuer les noms et préparer A Rendre",
"Crée le dossier A Rendre à partir du dossier dannotations choisi.", "Crée le dossier A Rendre à partir du dossier dannotations choisi.",
(python("default", "Attribution des noms", "giving_names.py"),), (python("default", "Attribution des noms", "giving-names"),),
arguments=( arguments=(
arg_target("Dossier de l’évaluation"), arg_target("Dossier de l’évaluation"),
ArgumentSpec( ArgumentSpec(
@@ -399,7 +424,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Étapes personnelles", "Étapes personnelles",
"Mettre à jour le fichier ODS", "Mettre à jour le fichier ODS",
"Transfère les scores, avec la possibilité de n’écrire que leur somme.", "Transfère les scores, avec la possibilité de n’écrire que leur somme.",
(python("default", "Mise à jour ODS", "update_ods.py"),), (python("default", "Mise à jour ODS", "update-ods"),),
arguments=( arguments=(
arg_target("Dossier de l’évaluation"), arg_target("Dossier de l’évaluation"),
ArgumentSpec("sum", "Écrire seulement la somme", "bool", "--sum"), ArgumentSpec("sum", "Écrire seulement la somme", "bool", "--sum"),
@@ -427,7 +452,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"Étapes personnelles", "Étapes personnelles",
"Ajouter le score final", "Ajouter le score final",
"Génère les fichiers de diffusion avec le score final.", "Génère les fichiers de diffusion avec le score final.",
(python("default", "Score final", "add_final_score.py"),), (python("default", "Score final", "add-final-score"),),
arguments=(arg_target("Dossier de l’évaluation"),), arguments=(arg_target("Dossier de l’évaluation"),),
personal=True, personal=True,
), ),
@@ -469,7 +494,7 @@ def build_command(
program_path = repository / variant.program program_path = repository / variant.program
if variant.kind == "python": if variant.kind == "python":
command = [sys.executable, "-u", str(program_path)] command = [sys.executable, "-u", "-m", "copienator", variant.program]
elif variant.kind == "shell": elif variant.kind == "shell":
command = [str(program_path)] command = [str(program_path)]
else: else:
-65
View File
@@ -1,65 +0,0 @@
import argparse
import shutil
import sys
from collections.abc import Sequence
from pathlib import Path
from config import IMPORT_DIR
from copienator import (
EvaluationWorkspace,
ExitCode,
evaluation_parser,
execute,
workspace_from_args,
)
def sync_annotated(
workspace: EvaluationWorkspace,
*,
refaire: bool = False,
import_dir: Path,
) -> ExitCode:
annotation_name = "BRnot" if refaire else "BGnot"
workspace.require_directories(annotation_name)
annotation_dir = workspace.root / annotation_name
annotated_dir = Path(import_dir).expanduser()
if not annotated_dir.is_dir():
print(f"Error: directory does not exist: {annotated_dir}", file=sys.stderr)
return ExitCode.INVALID_WORKSPACE
missing_targets = 0
for pdf_file in annotated_dir.glob("*.pdf"):
target_subdir = annotation_dir / pdf_file.stem
if not target_subdir.is_dir():
print(f"Warning: directory not found: {target_subdir}", file=sys.stderr)
missing_targets += 1
else:
dest_file = target_subdir / "Concat_annotated.pdf"
print(f"Copying {pdf_file} to {dest_file}")
shutil.copy2(pdf_file, dest_file)
return ExitCode.PARTIAL if missing_targets else ExitCode.SUCCESS
def build_parser() -> argparse.ArgumentParser:
parser = evaluation_parser("Import handwritten annotations from the tablet directory.")
parser.add_argument("--refaire", action="store_true", help="Process only copies/labels defined in refaire.json")
return parser
def run(workspace: EvaluationWorkspace, *, refaire: bool = False) -> ExitCode:
return sync_annotated(workspace, refaire=refaire, import_dir=Path(IMPORT_DIR))
def main(argv: Sequence[str] | None = None) -> int:
parser = build_parser()
return execute(
parser,
argv,
lambda args: run(workspace_from_args(args), refaire=args.refaire),
)
if __name__ == "__main__":
raise SystemExit(main())
+22
View File
@@ -0,0 +1,22 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "copienator"
version = "0.1.0"
description = "Workflow assistant for preparing and correcting scanned exams"
requires-python = ">=3.11"
[project.scripts]
copienator = "copienator.dispatcher:main"
copienator-gui = "copienator_gui.__main__:main"
[tool.setuptools]
py-modules = ["default_config"]
[tool.setuptools.packages.find]
include = ["copienator*", "copienator_gui*"]
[tool.setuptools.package-data]
copienator = ["data/*.txt"]
+287 -29
View File
@@ -1,6 +1,6 @@
from __future__ import annotations from __future__ import annotations
import importlib.util import importlib
import io import io
import json import json
import os import os
@@ -31,13 +31,23 @@ from copienator import (
from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides
from copienator.annotation_data import AnnotationLoadResult, load_annotation_data from copienator.annotation_data import AnnotationLoadResult, load_annotation_data
from copienator.filesystem import staged_directory, staged_files from copienator.filesystem import staged_directory, staged_files
from copienator_gui.app import process_status from copienator.dispatcher import COMMANDS, main as dispatcher_main
from copienator_gui.app import (
CopienatorApp,
DEFAULT_HTTPS_PROXY,
build_runner_environment,
copy_pdf_paths,
detected_annotation_directories,
has_manual_conflicts,
plotting_shortcut_lines,
process_status,
)
from copienator_gui.diagnostics import collect_diagnostics from copienator_gui.diagnostics import collect_diagnostics
from copienator_gui.runner import ProcessRunner from copienator_gui.runner import ProcessRunner
from copienator_gui.state import StateStore from copienator_gui.state import StateStore
from copienator_gui.workflow import build_command, build_workflow, evaluation_argument from copienator_gui.workflow import build_command, build_workflow, evaluation_argument
from copies_tools import rename_all, rotate_all from copienator.commands.copies_tools import rename_all, rotate_all
from platform_utils import ( from copienator.platform import (
WindowsLabelError, WindowsLabelError,
replace_with_link_or_copy, replace_with_link_or_copy,
safe_filename, safe_filename,
@@ -48,18 +58,16 @@ from platform_utils import (
REPOSITORY = Path(__file__).resolve().parents[1] REPOSITORY = Path(__file__).resolve().parents[1]
def load_script_module(filename: str, module_name: str): def load_script_module(filename: str, _module_name: str):
spec = importlib.util.spec_from_file_location(module_name, REPOSITORY / filename) stem = Path(filename).stem.replace("-", "_")
if spec is None or spec.loader is None: if stem == "import":
raise RuntimeError(f"Could not load {filename}") stem = "import_annotations"
module = importlib.util.module_from_spec(spec) return importlib.import_module(f"copienator.commands.{stem}")
sys.modules[module_name] = module
try:
spec.loader.exec_module(module) def command_arguments(command: list[str]) -> list[str]:
except Exception: assert command[2:4] == ["-m", "copienator"]
sys.modules.pop(module_name, None) return command[5:]
raise
return module
class WorkspaceTests(unittest.TestCase): class WorkspaceTests(unittest.TestCase):
@@ -71,6 +79,10 @@ class WorkspaceTests(unittest.TestCase):
self.assertFalse(workspace.metadata_dir.exists()) self.assertFalse(workspace.metadata_dir.exists())
self.assertEqual(workspace.labels_file, root / "labels") self.assertEqual(workspace.labels_file, root / "labels")
self.assertEqual(workspace.label_groups_file, root / "label_groups")
self.assertEqual(
workspace.gemini_exam_items_file, root / "Tmp" / "exam_items.txt"
)
self.assertEqual(workspace.copies_dir, root / "Copies") self.assertEqual(workspace.copies_dir, root / "Copies")
self.assertEqual(workspace.annotation_dir("grouped"), root / "BGnot") self.assertEqual(workspace.annotation_dir("grouped"), root / "BGnot")
self.assertEqual(workspace.state_database, root / ".copienator" / "state.sqlite3") self.assertEqual(workspace.state_database, root / ".copienator" / "state.sqlite3")
@@ -116,6 +128,24 @@ class WorkspaceTests(unittest.TestCase):
self.assertEqual(workspace.names_file(), root / "names") self.assertEqual(workspace.names_file(), root / "names")
class DispatcherTests(unittest.TestCase):
def test_unknown_command_has_argument_error_code(self) -> None:
with redirect_stderr(io.StringIO()):
self.assertEqual(dispatcher_main(["does-not-exist"]), 2)
def test_every_gui_python_command_is_registered(self) -> None:
for step in build_workflow(True):
for variant in step.variants:
if variant.kind == "python":
with self.subTest(step=step.id, command=variant.program):
self.assertIn(variant.program, COMMANDS)
def test_root_contains_only_configuration_python_modules(self) -> None:
root_modules = {path.name for path in REPOSITORY.glob("*.py")}
self.assertIn("default_config.py", root_modules)
self.assertLessEqual(root_modules, {"config.py", "default_config.py"})
class AtomicJsonTests(unittest.TestCase): class AtomicJsonTests(unittest.TestCase):
def test_atomic_binary_round_trip(self) -> None: def test_atomic_binary_round_trip(self) -> None:
with tempfile.TemporaryDirectory() as directory: with tempfile.TemporaryDirectory() as directory:
@@ -169,6 +199,17 @@ class AtomicJsonTests(unittest.TestCase):
persisted = read_json(root / ".copienator-gui.json") persisted = read_json(root / ".copienator-gui.json")
self.assertEqual(persisted["steps"]["labels"]["status"], "success") self.assertEqual(persisted["steps"]["labels"]["status"], "success")
def test_invalidation_preserves_first_visit_marker(self) -> None:
with tempfile.TemporaryDirectory() as directory:
store = StateStore()
store.load(Path(directory))
store.update_step("splitting", status="success", visited=True)
store.invalidate_after(["plotting", "splitting"], "plotting")
self.assertEqual(store.step("splitting")["status"], "stale")
self.assertTrue(store.step("splitting")["visited"])
class AnnotationDataTests(unittest.TestCase): class AnnotationDataTests(unittest.TestCase):
def test_loader_indexes_coordinates_without_mutating_correction(self) -> None: def test_loader_indexes_coordinates_without_mutating_correction(self) -> None:
@@ -435,8 +476,16 @@ class StandardCliTests(unittest.TestCase):
"personal", "personal",
{"target": evaluation}, {"target": evaluation},
), ),
"export": ("export", "default", {"target": evaluation, "refaire": True}), "export": (
"import": ("import", "default", {"target": evaluation, "refaire": True}), "export",
"default",
{"target": evaluation, "annotation_dir": "Bnot", "refaire": True},
),
"import": (
"import",
"default",
{"target": evaluation, "annotation_dir": "Anot", "refaire": True},
),
"giving_names": ( "giving_names": (
"giving_names", "giving_names",
"default", "default",
@@ -529,7 +578,9 @@ class StandardCliTests(unittest.TestCase):
variant = next(item for item in step.variants if item.id == variant_id) variant = next(item for item in step.variants if item.id == variant_id)
command = build_command(REPOSITORY, step, variant, values, evaluation) command = build_command(REPOSITORY, step, variant, values, evaluation)
with self.subTest(script=module_name): with self.subTest(script=module_name):
parsed = self.modules[module_name].build_parser().parse_args(command[3:]) parsed = self.modules[module_name].build_parser().parse_args(
command_arguments(command)
)
parsed_path = getattr(parsed, "evaluation", None) or parsed.target parsed_path = getattr(parsed, "evaluation", None) or parsed.target
self.assertEqual(str(parsed_path), evaluation) self.assertEqual(str(parsed_path), evaluation)
@@ -545,7 +596,7 @@ class StandardCliTests(unittest.TestCase):
) )
with self.subTest(script=f"copies_tools:{step_id}"): with self.subTest(script=f"copies_tools:{step_id}"):
parsed = self.modules["copies_tools"].build_parser().parse_args( parsed = self.modules["copies_tools"].build_parser().parse_args(
command[3:] command_arguments(command)
) )
self.assertEqual(parsed.operation, step_id) self.assertEqual(parsed.operation, step_id)
self.assertEqual(str(parsed.evaluation), evaluation) self.assertEqual(str(parsed.evaluation), evaluation)
@@ -578,6 +629,23 @@ class StandardCliTests(unittest.TestCase):
exported = base / "Export" / "Exam" / "Ex 1.pdf" exported = base / "Export" / "Exam" / "Ex 1.pdf"
self.assertEqual(exported.read_bytes(), b"annotated") self.assertEqual(exported.read_bytes(), b"annotated")
def test_export_accepts_each_annotation_directory(self) -> None:
module = self.modules["export"]
for annotation_dir in ("Anot", "Bnot"):
with self.subTest(annotation_dir=annotation_dir), tempfile.TemporaryDirectory() as directory:
base = Path(directory)
evaluation = base / "Exam"
source = evaluation / annotation_dir / "Copie01"
source.mkdir(parents=True)
suffix = ".jpg" if annotation_dir == "Anot" else ".pdf"
(source / f"Concat{suffix}").write_bytes(annotation_dir.encode())
with patch.object(module, "EXPORT_DIR", base / "Export"):
self.assertEqual(module.main([str(evaluation), annotation_dir]), 0)
self.assertEqual(
(base / "Export" / "Exam" / f"Copie01{suffix}").read_bytes(),
annotation_dir.encode(),
)
def test_import_main_copies_handwritten_annotations(self) -> None: def test_import_main_copies_handwritten_annotations(self) -> None:
module = self.modules["import"] module = self.modules["import"]
with tempfile.TemporaryDirectory() as directory: with tempfile.TemporaryDirectory() as directory:
@@ -594,6 +662,24 @@ class StandardCliTests(unittest.TestCase):
(target / "Concat_annotated.pdf").read_bytes(), b"handwritten" (target / "Concat_annotated.pdf").read_bytes(), b"handwritten"
) )
def test_import_accepts_each_annotation_directory(self) -> None:
module = self.modules["import"]
for annotation_dir in ("Anot", "Bnot"):
with self.subTest(annotation_dir=annotation_dir), tempfile.TemporaryDirectory() as directory:
base = Path(directory)
evaluation = base / "Exam"
target = evaluation / annotation_dir / "Copie01"
target.mkdir(parents=True)
import_dir = base / "Import"
import_dir.mkdir()
suffix = ".jpg" if annotation_dir == "Anot" else ".pdf"
(import_dir / f"Copie01{suffix}").write_bytes(b"handwritten")
with patch.object(module, "IMPORT_DIR", import_dir):
self.assertEqual(module.main([str(evaluation), annotation_dir]), 0)
self.assertEqual(
(target / f"Concat_annotated{suffix}").read_bytes(), b"handwritten"
)
def test_giving_names_main_builds_return_directory(self) -> None: def test_giving_names_main_builds_return_directory(self) -> None:
module = self.modules["giving_names"] module = self.modules["giving_names"]
with tempfile.TemporaryDirectory() as directory: with tempfile.TemporaryDirectory() as directory:
@@ -1337,6 +1423,65 @@ class StandardCliTests(unittest.TestCase):
(previous / "sentinel.txt").read_text(encoding="utf-8"), "old" (previous / "sentinel.txt").read_text(encoding="utf-8"), "old"
) )
def test_grouped_annotations_default_to_gemini_question_groups(self) -> None:
module = self.modules["annotating_by_label"]
with tempfile.TemporaryDirectory() as directory:
evaluation = Path(directory) / "Exam"
items = evaluation / "Tmp" / "exam_items.txt"
items.parent.mkdir(parents=True)
items.write_text(
"# edited Gemini groups\n"
"Ex 1 ### First question\n"
"CONTEXT ### Shared context\n"
"Ex 2 ### Second question\n"
"\n---\n\n"
"Ex 3 ### Third question\n",
encoding="utf-8",
)
workspace = EvaluationWorkspace(evaluation)
with patch.object(module.utils, "edit_file_and_enter") as editor:
groups = module._load_label_groups(
workspace, ["Ex 1", "Ex 2", "Ex 3"]
)
self.assertEqual(groups, [["Ex 1", "Ex 2"], ["Ex 3"]])
self.assertEqual(
workspace.label_groups_file.read_text(encoding="utf-8"),
"Ex 1,Ex 2\nEx 3\n",
)
editor.assert_called_once_with(workspace.label_groups_file)
def test_grouped_annotations_fall_back_when_gemini_was_not_run(self) -> None:
module = self.modules["annotating_by_label"]
with tempfile.TemporaryDirectory() as directory:
workspace = EvaluationWorkspace(Path(directory) / "Exam")
workspace.root.mkdir()
labels = ["Ex 1 : a", "Ex 1 : b", "Ex 2"]
with patch.object(module.utils, "edit_file_and_enter"):
groups = module._load_label_groups(workspace, labels)
self.assertEqual(groups, [["Ex 1 : a", "Ex 1 : b"], ["Ex 2"]])
def test_existing_label_groups_override_gemini_defaults(self) -> None:
module = self.modules["annotating_by_label"]
with tempfile.TemporaryDirectory() as directory:
workspace = EvaluationWorkspace(Path(directory) / "Exam")
workspace.gemini_exam_items_file.parent.mkdir(parents=True)
workspace.gemini_exam_items_file.write_text(
"Ex 1 ### First\nEx 2 ### Second\n", encoding="utf-8"
)
workspace.label_groups_file.write_text(
"Ex 1\nEx 2\n", encoding="utf-8"
)
with patch.object(module.utils, "edit_file_and_enter") as editor:
groups = module._load_label_groups(workspace, ["Ex 1", "Ex 2"])
self.assertEqual(groups, [["Ex 1"], ["Ex 2"]])
editor.assert_not_called()
def test_grouped_batching_does_not_split_one_student(self) -> None: def test_grouped_batching_does_not_split_one_student(self) -> None:
module = self.modules["annotating_by_label"] module = self.modules["annotating_by_label"]
image = Image.new("RGB", (10, 60), "white") image = Image.new("RGB", (10, 60), "white")
@@ -1366,33 +1511,146 @@ class WorkflowTests(unittest.TestCase):
self.assertNotIn("update_ods", standard_ids) self.assertNotIn("update_ods", standard_ids)
self.assertIn("update_ods", personal_ids) self.assertIn("update_ods", personal_ids)
def test_first_visit_automation_is_declared_on_expected_steps(self) -> None:
auto_start = {
step.id for step in self.steps.values() if step.auto_start_first_visit
}
skip_for_live = {
step.id for step in self.steps.values() if step.skip_for_live_correction
}
skip_without_conflicts = {
step.id
for step in self.steps.values()
if step.skip_without_manual_conflicts
}
self.assertEqual(auto_start, {"splitting", "grouping"})
self.assertEqual(
skip_for_live, {"submit_batches", "batch_status", "fetch_batches"}
)
self.assertEqual(skip_without_conflicts, {"manual_resolution"})
def test_review_persp_has_shorter_title(self) -> None:
self.assertEqual(self.steps["review_persp"].title, "Relire les barèmes")
def test_export_has_shorter_title_and_annotation_argument(self) -> None:
self.assertEqual(self.steps["export"].title, "Exporter")
command = self.command(
"export",
"default",
{"target": self.evaluation, "annotation_dir": "Bnot"},
)
self.assertEqual(command_arguments(command), [self.evaluation, "Bnot"])
def test_copy_listing_prefers_processed_copies(self) -> None:
with tempfile.TemporaryDirectory() as directory:
evaluation = Path(directory)
(evaluation / "enonce.pdf").touch()
(evaluation / "scan.pdf").touch()
copies = evaluation / "Copies"
copies.mkdir()
(copies / "Copie02.pdf").touch()
(copies / "Copie01.pdf").touch()
self.assertEqual(
[path.name for path in copy_pdf_paths(evaluation)],
["Copie01.pdf", "Copie02.pdf"],
)
def test_annotation_directory_detection_uses_known_directories(self) -> None:
with tempfile.TemporaryDirectory() as directory:
evaluation = Path(directory)
(evaluation / "Bnot").mkdir()
(evaluation / "Anot").mkdir()
(evaluation / "Other").mkdir()
self.assertEqual(
detected_annotation_directories(evaluation), ("Bnot", "Anot")
)
def test_export_and_import_defaults_follow_previous_runs(self) -> None:
with tempfile.TemporaryDirectory() as directory:
evaluation = Path(directory)
(evaluation / "Bnot").mkdir()
(evaluation / "Anot").mkdir()
store = StateStore()
store.load(evaluation)
store.update_step("annotation", last_run_variant="simple")
app = SimpleNamespace(evaluation=evaluation, state_store=store)
choices, default = CopienatorApp._annotation_directory_choices(
app, "export"
)
self.assertEqual(choices, ("Bnot", "Anot"))
self.assertEqual(default, "Anot")
store.update_step(
"export", last_run_values={"annotation_dir": "Bnot"}
)
_choices, default = CopienatorApp._annotation_directory_choices(
app, "import"
)
self.assertEqual(default, "Bnot")
def test_plotting_shortcuts_describe_open_actions(self) -> None:
rendered = "\n".join(plotting_shortcut_lines())
self.assertIn("ouvrir l’énoncé", rendered)
self.assertIn("ouvrir la copie traitée", rendered)
self.assertIn("ouvrir la copie originale", rendered)
def test_proxy_is_opt_in_and_prefilled(self) -> None:
base = {"HTTPS_PROXY": "http://system-proxy", "OTHER": "kept"}
without_proxy = build_runner_environment(
base, " secret ", DEFAULT_HTTPS_PROXY, False
)
with_proxy = build_runner_environment(base, "", DEFAULT_HTTPS_PROXY, True)
self.assertNotIn("HTTPS_PROXY", without_proxy)
self.assertEqual(without_proxy["GEMINI_API_KEY"], "secret")
self.assertEqual(without_proxy["OTHER"], "kept")
self.assertEqual(with_proxy["HTTPS_PROXY"], DEFAULT_HTTPS_PROXY)
def test_manual_conflicts_ignore_blank_and_comment_lines(self) -> None:
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "manual_resolutions.txt"
self.assertFalse(has_manual_conflicts(path))
path.write_text("\n### Instructions\n ### exemple\n", encoding="utf-8")
self.assertFalse(has_manual_conflicts(path))
path.write_text(
"### Instructions\nCopie01 Ex 1 -> Ex 2\n", encoding="utf-8"
)
self.assertTrue(has_manual_conflicts(path))
def test_live_correction_arguments(self) -> None: def test_live_correction_arguments(self) -> None:
command = self.command( command = self.command(
"correction", "correction",
"live", "live",
{"target": self.evaluation, "overwrite": True, "limit": "0"}, {"target": self.evaluation, "overwrite": True, "limit": "0"},
) )
self.assertEqual(command[3:], [self.evaluation, "--overwrite", "--limit", "0"]) self.assertEqual(
self.assertIn("correction.py", command[2]) command_arguments(command),
[self.evaluation, "--overwrite", "--limit", "0"],
)
self.assertEqual(command[4], "correct")
def test_hybrid_correction_arguments(self) -> None: def test_hybrid_correction_arguments(self) -> None:
command = self.command( command = self.command(
"correction", "hybrid", {"target": self.evaluation, "batch_from": "Ex 4"} "correction", "hybrid", {"target": self.evaluation, "batch_from": "Ex 4"}
) )
self.assertEqual(command[3:], [self.evaluation, "--batch-from", "Ex 4"]) self.assertEqual(
command_arguments(command), [self.evaluation, "--batch-from", "Ex 4"]
)
def test_annotation_variants_are_exclusive_commands(self) -> None: def test_annotation_variants_are_exclusive_commands(self) -> None:
command = self.command( command = self.command(
"annotation", "grouped", {"target": self.evaluation, "overwrite": True} "annotation", "grouped", {"target": self.evaluation, "overwrite": True}
) )
self.assertIn("annotating_by_label.py", command[2]) self.assertEqual(command[4], "annotate-grouped")
self.assertNotIn("annotating.py", command[2])
def test_copy_preparation_commands_are_python(self) -> None: def test_copy_preparation_commands_are_python(self) -> None:
rotate = self.command("rotate", "rotate", {"target": self.evaluation}) rotate = self.command("rotate", "rotate", {"target": self.evaluation})
rename = self.command("rename", "rename", {"target": self.evaluation}) rename = self.command("rename", "rename", {"target": self.evaluation})
self.assertEqual(rotate[2:], [str(REPOSITORY / "copies_tools.py"), "rotate", self.evaluation]) self.assertEqual(rotate[2:], ["-m", "copienator", "copies", "rotate", self.evaluation])
self.assertEqual(rename[2:], [str(REPOSITORY / "copies_tools.py"), "rename", self.evaluation]) self.assertEqual(rename[2:], ["-m", "copienator", "copies", "rename", self.evaluation])
class CrossPlatformFileTests(unittest.TestCase): class CrossPlatformFileTests(unittest.TestCase):
@@ -1430,8 +1688,8 @@ class CrossPlatformFileTests(unittest.TestCase):
source = root / "source.txt" source = root / "source.txt"
destination = root / "destination.txt" destination = root / "destination.txt"
source.write_text("content", encoding="utf-8") source.write_text("content", encoding="utf-8")
with patch("platform_utils.os.link", side_effect=OSError), patch( with patch("copienator.platform.os.link", side_effect=OSError), patch(
"platform_utils.os.symlink", side_effect=OSError "copienator.platform.os.symlink", side_effect=OSError
): ):
method = replace_with_link_or_copy(source, destination) method = replace_with_link_or_copy(source, destination)
self.assertEqual(method, "copy") self.assertEqual(method, "copy")