From c8ce04cf50985233234cacd64a197ef924529a7e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Miquel?= Date: Mon, 21 Sep 2026 16:07:56 +0200 Subject: [PATCH] Remove enter binding, ; hide batched correction steps by default --- copienator/commands/page_splitter.py | 6 +- copienator_gui/app.py | 62 ++++++++++++++----- default_config.py | 1 - tests/test_gui_convenience.py | 92 +++++++++++++++++++++++++++- 4 files changed, 139 insertions(+), 22 deletions(-) diff --git a/copienator/commands/page_splitter.py b/copienator/commands/page_splitter.py index a02b28d..38775a6 100644 --- a/copienator/commands/page_splitter.py +++ b/copienator/commands/page_splitter.py @@ -187,7 +187,6 @@ class PDFPreviewer: action_map = { "move_left": self.move_line_left, "move_right": self.move_line_right, - "confirm_next": self.confirm_and_next_page, "rotate_page": self.rotate_page, "rotate_all_pages": self.rotate_all_pages, "rotate_all_files": self.rotate_all_files, @@ -209,7 +208,6 @@ class PDFPreviewer: # self.master.bind("", self.move_line_left) # self.master.bind("", self.move_line_right) - # self.master.bind("", self.confirm_and_next_page) # self.master.bind("c", self.rotate_page) # self.master.bind("C", self.rotate_all_pages) # self.master.bind(",", self.rotate_all_files) @@ -260,7 +258,9 @@ class PDFPreviewer: return page = self.doc.load_page(self.current_page_index) - self.page_label.config(text=f"Page {self.current_page_index + 1} of {len(self.doc)}") + self.page_label.config( + text=f"{self.pdf_path.name} — Page {self.current_page_index + 1} of {len(self.doc)}" + ) # --- Calculate Scaling --- canvas_width = self.canvas.winfo_width() diff --git a/copienator_gui/app.py b/copienator_gui/app.py index df5d299..0f45ea8 100644 --- a/copienator_gui/app.py +++ b/copienator_gui/app.py @@ -591,7 +591,10 @@ class CopienatorApp(tk.Tk): opened = {self.tree.item(item, "text"): self.tree.item(item, "open") for item in self.tree.get_children()} self.tree.delete(*self.tree.get_children()) section_items: dict[str, str] = {} + show_batches = self._has_batch_correction() for step in self.steps: + if step.skip_for_live_correction and not show_batches: + continue if step.section not in section_items: section_id = f"section:{len(section_items)}" section_items[step.section] = section_id @@ -607,6 +610,25 @@ class CopienatorApp(tk.Tk): ) if selected and self.tree.exists(selected): self.tree.selection_set(selected) + elif selected and self.tree.exists("correction"): + self.tree.selection_set("correction") + + def _has_batch_correction(self) -> bool: + evaluation = self.state_store.evaluation + if evaluation is None: + return False + correction = self.state_store.step("correction") + if correction.get("batch_started") or correction.get("last_run_variant") in {"batch", "hybrid", "integrate"}: + return True + for name in ("batch_requests_flash.jsonl", "batch_requests_pro.jsonl", + "batch_jobs.json", "batched_correction_result.jsonl"): + path = evaluation / name + try: + if path.is_file() and path.stat().st_size > 0: + return True + except OSError: + continue + return False def _step_status(self, step: StepDefinition) -> str: if self.state_store.evaluation: @@ -721,16 +743,6 @@ class CopienatorApp(tk.Tk): # 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 @@ -1217,7 +1229,9 @@ class CopienatorApp(tk.Tk): def _progression_ids(self, step_id: str) -> list[str]: redo = step_id.startswith("refaire_") - return [step.id for step in self.steps if step.id.startswith("refaire_") == redo] + show_batches = self._has_batch_correction() + return [step.id for step in self.steps if step.id.startswith("refaire_") == redo + and (not step.skip_for_live_correction or show_batches)] def _refaire_selection(self) -> list[list]: evaluation = self.evaluation @@ -1473,6 +1487,8 @@ class CopienatorApp(tk.Tk): self._append_console(f"\n$ {command_display(command)}\n") try: self.runner.start(command, self.repository, environment, log_path) + if step.id == "correction" and variant.id in {"batch", "hybrid"}: + self.state_store.update_step("correction", batch_started=True) except (OSError, RuntimeError) as exc: self.state_store.update_step(step.id, status="failed") self.state_store.add_history( @@ -1614,15 +1630,18 @@ class CopienatorApp(tk.Tk): self._populate_tree() message = f"{evaluation.name} : tous les résultats batch sont prêts à récupérer." self.info_var.set(message) - try: - notify_desktop("Copienator — batchs terminés", message) - except (OSError, RuntimeError) as exc: - self.bell() - self._append_console(f"Notification de bureau indisponible : {exc}\n{message}\n") + self._notify_completion("Copienator — batchs terminés", message) if (self.current_step and self.current_step.id == "batch_status" and not self.active_step_id and not self.runner.running): self._move_selection_from("batch_status", 1) + def _notify_completion(self, title: str, message: str) -> None: + try: + notify_desktop(title, message) + except (OSError, RuntimeError) as exc: + self.bell() + self._append_console(f"Notification de bureau indisponible : {exc}\n{message}\n") + def _finish_process(self, return_code: int, interrupted: bool) -> None: step_id = self.active_step_id if not step_id: @@ -1709,6 +1728,17 @@ class CopienatorApp(tk.Tk): "sur « Marquer terminée »." ) return + if step_id in {"labels", "correction"} and self.state_store.evaluation: + completion = ( + "détection des labels terminée" + if step_id == "labels" else "étape de correction terminée" + ) + outcome = STATUS_LABELS[status].lower() + self._notify_completion( + f"Copienator — {completion} ({outcome})", + f"{self.state_store.evaluation.name} : {completion} — {outcome} " + f"(code {return_code}).", + ) if status == "success": if step_id == "batch_status" and self.batch_monitor.active: self.batch_monitor.stop() diff --git a/default_config.py b/default_config.py index b1f1f26..af648ef 100644 --- a/default_config.py +++ b/default_config.py @@ -44,7 +44,6 @@ MODEL_PRO_ID = "gemini-3.8-flash" PAGE_SPLITTER_KB = { "move_left": "", "move_right": "", - "confirm_next": "", "rotate_page": "r", "rotate_all_pages": "R", "rotate_all_files": ",", diff --git a/tests/test_gui_convenience.py b/tests/test_gui_convenience.py index ad447af..ab335fc 100644 --- a/tests/test_gui_convenience.py +++ b/tests/test_gui_convenience.py @@ -133,7 +133,8 @@ class GuiConvenienceTests(unittest.TestCase): self.assertNotEqual(self.app.current_step.id, "giving_names") def test_batch_status_waits_until_all_jobs_are_ready(self): - self.app.state_store.update_step("correction", variant="batch") + self.app.state_store.update_step("correction", variant="batch", last_run_variant="batch") + self.app._populate_tree() self.app.state_store.update_step("batch_status", visited=True) self.app.tree.selection_set("batch_status") self.app.update() @@ -150,7 +151,8 @@ class GuiConvenienceTests(unittest.TestCase): self.assertEqual(self.app.current_step.id, "fetch_batches") def test_batch_watch_buttons_use_loaded_evaluation_and_stop_on_reload(self): - self.app.state_store.update_step("correction", variant="batch") + self.app.state_store.update_step("correction", variant="batch", last_run_variant="batch") + self.app._populate_tree() self.app.tree.selection_set("batch_status") self.app.update() with patch.object(self.app.batch_monitor, "start") as start: @@ -175,6 +177,92 @@ class GuiConvenienceTests(unittest.TestCase): self.assertEqual(self.app.current_step.id, "inputs") self.assertEqual(self.app.state_store.step("batch_status")["status"], "success") + def test_batch_steps_hidden_until_a_correction_is_started(self): + batch_ids = ("submit_batches", "batch_status", "fetch_batches") + self.app.state_store.update_step("correction", variant="batch") + self.app._populate_tree() + for step_id in batch_ids: + self.assertFalse(self.app.tree.exists(step_id)) + self.assertNotIn(step_id, self.app._progression_ids("correction")) + self.app.state_store.update_step("correction", last_run_variant="hybrid", status="running") + self.app._populate_tree() + for step_id in batch_ids: + self.assertTrue(self.app.tree.exists(step_id)) + self.assertIn(step_id, self.app._progression_ids("correction")) + + def test_existing_batch_artifacts_reveal_steps_even_in_live_mode(self): + for filename in ("batch_requests_flash.jsonl", "batch_requests_pro.jsonl", + "batch_jobs.json", "batched_correction_result.jsonl"): + with self.subTest(filename=filename): + path = self.evaluation / filename + path.write_text("") + self.app._populate_tree() + self.assertFalse(self.app.tree.exists("batch_status")) + path.write_text('{"jobs":{"flash":{"name":"batches/1"}}}\n') + self.app._populate_tree() + self.assertTrue(self.app.tree.exists("batch_status")) + self.app._handle_first_visit(self.app.step_by_id["batch_status"]) + self.app.update() + self.assertNotEqual(self.app.state_store.step("batch_status").get("status"), "skipped") + path.unlink() + + def test_started_batches_remain_visible_after_mode_change_and_reload(self): + self.app.state_store.update_step("correction", batch_started=True, last_run_variant="live") + self.app.state_store.load(self.evaluation) + self.app._populate_tree() + self.assertTrue(self.app.tree.exists("batch_status")) + fresh = self.repository / "Fresh" + fresh.mkdir() + self.app.state_store.load(fresh) + self.app._populate_tree() + self.assertFalse(self.app.tree.exists("batch_status")) + + def test_navigation_skips_hidden_batch_steps(self): + with patch.object(self.app.tree, "selection_set") as select: + self.app._move_selection_from("correction", 1) + select.assert_called_once_with("manual_resolution") + + def test_labels_and_correction_notify_once_on_success(self): + for step_id in ("labels", "correction"): + with self.subTest(step=step_id), patch("copienator_gui.app.notify_desktop") as notify: + self.app.active_step_id = step_id + self.app._finish_process(0, False) + self.app._finish_process(0, False) + notify.assert_called_once() + self.assertIn(self.evaluation.name, notify.call_args.args[1]) + self.assertIn("labels" if step_id == "labels" else "correction", + notify.call_args.args[1]) + + def test_incomplete_failed_and_interrupted_steps_notify_the_outcome_once(self): + for step_id in ("labels", "correction"): + for code, interrupted, outcome in ( + (4, False, "partielle"), (1, False, "échouée"), + (0, True, "interrompue"), (130, False, "interrompue"), + ): + with self.subTest(step=step_id, code=code, interrupted=interrupted), \ + patch("copienator_gui.app.notify_desktop") as notify: + self.app.active_step_id = step_id + self.app._finish_process(code, interrupted) + self.app._finish_process(code, interrupted) + notify.assert_called_once() + title, message = notify.call_args.args + self.assertIn(outcome, title) + self.assertIn(outcome, message) + self.assertIn(self.evaluation.name, message) + self.assertIn(f"code {code}", message) + self.assertNotIn("réussie", message) + + def test_notification_failure_keeps_success_and_uses_existing_fallback(self): + self.app.active_step_id = "labels" + with patch("copienator_gui.app.notify_desktop", side_effect=RuntimeError("unavailable")), \ + patch.object(self.app, "bell") as bell, \ + patch.object(self.app, "_append_console") as console: + self.app._finish_process(0, False) + bell.assert_called_once() + self.assertTrue(any("Notification de bureau indisponible" in call.args[0] + for call in console.call_args_list)) + self.assertEqual(self.app.state_store.step("labels")["status"], "success") + def test_optional_blank_crop_can_target_a_copy_or_be_skipped(self): copies = self.evaluation/"Copies" copies.mkdir()