fichiers oubliés

copienator_gui/giving_names.py

@@ -0,0 +1,228 @@
+from __future__ import annotations
+
+import os
+import re
+import tkinter as tk
+from collections import Counter
+from dataclasses import dataclass
+from pathlib import Path
+from tkinter import messagebox, ttk
+from typing import Callable
+
+from copienator import EvaluationWorkspace
+from copienator.platform import open_path, safe_filename
+
+
+RETURN_FOLDER_PATTERN = re.compile(r"^(?P<name>.+) \((?P<copy_id>\d+)\)$")
+
+
+@dataclass(frozen=True, slots=True)
+class NameIssue:
+    folder: Path
+    name: str
+    copy_id: str
+    reasons: tuple[str, ...]
+
+
+def find_name_issues(evaluation: Path) -> list[NameIssue]:
+    """Return unknown and duplicated names currently present in A Rendre."""
+    return_dir = EvaluationWorkspace(evaluation).return_dir
+    parsed: list[tuple[Path, str, str]] = []
+    if return_dir.is_dir():
+        for folder in return_dir.iterdir():
+            if not folder.is_dir() or folder.is_symlink():
+                continue
+            match = RETURN_FOLDER_PATTERN.fullmatch(folder.name)
+            if match:
+                parsed.append(
+                    (folder, match.group("name"), match.group("copy_id"))
+                )
+
+    counts = Counter(name.casefold() for _folder, name, _copy_id in parsed)
+    issues = []
+    for folder, name, copy_id in parsed:
+        reasons = []
+        if name.casefold() == "unknown":
+            reasons.append("nom inconnu")
+        if counts[name.casefold()] > 1:
+            reasons.append(f"nom attribué à {counts[name.casefold()]} copies")
+        if reasons:
+            issues.append(NameIssue(folder, name, copy_id, tuple(reasons)))
+    return sorted(issues, key=lambda issue: (int(issue.copy_id), issue.name.casefold()))
+
+
+def original_copy_path(evaluation: Path, copy_id: str) -> Path | None:
+    workspace = EvaluationWorkspace(evaluation)
+    filename = f"Copie{copy_id}.pdf"
+    for candidate in (
+        workspace.original_copies_dir / filename,
+        workspace.copies_dir / filename,
+        workspace.root / filename,
+    ):
+        if candidate.is_file():
+            return candidate
+    return None
+
+
+def rename_return_copy(folder: Path, requested_name: str) -> Path:
+    """Rename one return folder and its top-level JPG/PDF, preserving its ID."""
+    match = RETURN_FOLDER_PATTERN.fullmatch(folder.name)
+    if not folder.is_dir() or folder.is_symlink() or match is None:
+        raise ValueError(f"Dossier de copie invalide : {folder}")
+    if folder.parent.name != "A Rendre":
+        raise ValueError("Le dossier à renommer doit se trouver dans A Rendre.")
+
+    entered_name = requested_name.strip()
+    if not entered_name:
+        raise ValueError("Saisissez un nom.")
+    name = safe_filename(entered_name, fallback="")
+    if not name:
+        raise ValueError("Le nom ne contient aucun caractère utilisable.")
+
+    copy_id = match.group("copy_id")
+    destination = folder.with_name(f"{name} ({copy_id})")
+    if destination != folder and os.path.lexists(destination):
+        raise FileExistsError(f"Le dossier {destination.name!r} existe déjà.")
+
+    assets: dict[str, Path] = {}
+    for child in folder.iterdir():
+        suffix = child.suffix.casefold()
+        if suffix not in {".jpg", ".pdf"}:
+            continue
+        if suffix in assets:
+            raise ValueError(
+                f"Plusieurs fichiers {suffix} sont présents dans {folder.name}."
+            )
+        assets[suffix] = child
+
+    renames: list[tuple[Path, Path]] = []
+    for suffix, source in assets.items():
+        target = source.with_name(f"{name}{suffix}")
+        if target != source and os.path.lexists(target):
+            raise FileExistsError(f"Le fichier {target.name!r} existe déjà.")
+        if target != source:
+            renames.append((source, target))
+
+    completed: list[tuple[Path, Path]] = []
+    try:
+        for source, target in renames:
+            source.rename(target)
+            completed.append((source, target))
+        if destination != folder:
+            folder.rename(destination)
+    except OSError:
+        for source, target in reversed(completed):
+            if os.path.lexists(target) and not os.path.lexists(source):
+                target.rename(source)
+        raise
+    return destination
+
+
+class GivingNamesPanel(ttk.LabelFrame):
+    def __init__(
+        self,
+        master: tk.Misc,
+        evaluation: Path,
+        *,
+        opener: Callable[[Path], None] = open_path,
+    ) -> None:
+        super().__init__(master, text="Vérification des noms", padding=7)
+        self.evaluation = evaluation
+        self.opener = opener
+        self.columnconfigure(0, weight=1)
+
+        actions = ttk.Frame(self)
+        actions.grid(row=0, column=0, sticky="w", pady=(0, 7))
+        ttk.Button(
+            actions,
+            text="Ouvrir A Rendre",
+            command=self.open_return_folder,
+        ).pack(side="left")
+        ttk.Button(actions, text="Recharger", command=self.reload).pack(
+            side="left", padx=6
+        )
+
+        self.issue_frame = ttk.Frame(self)
+        self.issue_frame.grid(row=1, column=0, sticky="ew")
+        self.issue_frame.columnconfigure(1, weight=1)
+        self.reload()
+
+    def open_return_folder(self) -> None:
+        path = EvaluationWorkspace(self.evaluation).return_dir
+        if not path.is_dir():
+            messagebox.showerror(
+                "Dossier introuvable",
+                "Exécutez d’abord l’attribution des noms pour créer A Rendre.",
+            )
+            return
+        try:
+            self.opener(path)
+        except (OSError, RuntimeError) as exc:
+            messagebox.showerror("Ouverture impossible", str(exc))
+
+    def open_original(self, copy_id: str) -> None:
+        path = original_copy_path(self.evaluation, copy_id)
+        if path is None:
+            messagebox.showerror(
+                "Copie introuvable", f"Copie{copy_id}.pdf est introuvable."
+            )
+            return
+        try:
+            self.opener(path)
+        except (OSError, RuntimeError) as exc:
+            messagebox.showerror("Ouverture impossible", str(exc))
+
+    def rename_issue(self, issue: NameIssue, variable: tk.StringVar) -> None:
+        try:
+            destination = rename_return_copy(issue.folder, variable.get())
+        except (OSError, ValueError) as exc:
+            messagebox.showerror("Renommage impossible", str(exc))
+            return
+        self.reload()
+        messagebox.showinfo("Copie renommée", f"Nouveau dossier : {destination.name}")
+
+    def reload(self) -> None:
+        for child in self.issue_frame.winfo_children():
+            child.destroy()
+        issues = find_name_issues(self.evaluation)
+        if not issues:
+            return_dir = EvaluationWorkspace(self.evaluation).return_dir
+            text = (
+                "Aucun nom inconnu ou dupliqué."
+                if return_dir.is_dir()
+                else "A Rendre n’a pas encore été généré."
+            )
+            ttk.Label(self.issue_frame, text=text).grid(
+                row=0, column=0, columnspan=4, sticky="w"
+            )
+            return
+
+        ttk.Label(self.issue_frame, text="Copie / problème").grid(
+            row=0, column=0, sticky="w", padx=(0, 6)
+        )
+        ttk.Label(self.issue_frame, text="Nouveau nom").grid(
+            row=0, column=1, sticky="w"
+        )
+        for row, issue in enumerate(issues, start=1):
+            ttk.Label(
+                self.issue_frame,
+                text=f"Copie{issue.copy_id} — {', '.join(issue.reasons)}",
+            ).grid(row=row, column=0, sticky="w", padx=(0, 6), pady=2)
+            variable = tk.StringVar()
+            ttk.Entry(self.issue_frame, textvariable=variable).grid(
+                row=row, column=1, sticky="ew", pady=2
+            )
+            original = original_copy_path(self.evaluation, issue.copy_id)
+            ttk.Button(
+                self.issue_frame,
+                text="Ouvrir l’originale",
+                command=lambda ident=issue.copy_id: self.open_original(ident),
+                state="normal" if original else "disabled",
+            ).grid(row=row, column=2, padx=6, pady=2)
+            ttk.Button(
+                self.issue_frame,
+                text="Renommer",
+                command=lambda item=issue, value=variable: self.rename_issue(
+                    item, value
+                ),
+            ).grid(row=row, column=3, pady=2)

copienator_gui/syncthing.py

@@ -0,0 +1,113 @@
+"""Non-blocking controls for the system Syncthing instance."""
+import queue
+import subprocess
+import threading
+from tkinter import ttk, messagebox
+
+SERVICE = "syncthing@sebastien.service"
+
+
+def service_status():
+    result = subprocess.run(
+        ["systemctl", "show", SERVICE, "--property=ActiveState", "--value"],
+        capture_output=True, text=True, timeout=10,
+    )
+    if result.returncode:
+        raise RuntimeError(result.stderr.strip() or "Statut indisponible")
+    return result.stdout.strip() or "unknown"
+
+
+def control_service(action):
+    if action == "ensure" and service_status() == "active":
+        return "active"
+    command = "start" if action == "ensure" else action
+    result = subprocess.run(
+        ["systemctl", "--no-ask-password", command, SERVICE],
+        capture_output=True, text=True, timeout=60,
+    )
+    if result.returncode:
+        raise RuntimeError(result.stderr.strip() or f"Échec de systemctl {command}")
+    status = service_status()
+    if command == "start" and status != "active":
+        raise RuntimeError(f"Syncthing n’est pas actif : {status}")
+    return status
+
+
+class SyncthingPanel(ttk.Frame):
+    def __init__(self, parent):
+        super().__init__(parent)
+        self.events = queue.Queue()
+        self.busy = False
+        self.pending_export = None
+        self.status = ttk.Label(self, text=f"{SERVICE} : vérification…")
+        self.status.pack(anchor="w")
+        controls = ttk.Frame(self)
+        controls.pack(anchor="w", pady=4)
+        self.start_button = ttk.Button(controls, text="Démarrer Syncthing",
+                                       command=lambda: self.request("start"))
+        self.start_button.pack(side="left")
+        self.stop_button = ttk.Button(controls, text="Arrêter Syncthing",
+                                      command=lambda: self.request("stop"))
+        self.stop_button.pack(side="left", padx=6)
+        self.timer = self.after(100, self.poll)
+        self.refresh_timer = None
+        self.bind("<Destroy>", self.cleanup, add=True)
+        self.request("status")
+
+    def ensure_started(self, callback):
+        self.pending_export = callback
+        if not self.busy:
+            self.request("ensure")
+
+    def request(self, action):
+        if self.busy:
+            return
+        self.busy = True
+        self.start_button.configure(state="disabled")
+        self.stop_button.configure(state="disabled")
+        if action != "status":
+            self.status.configure(text=f"{SERVICE} : opération en cours…")
+
+        def work():
+            try:
+                status = service_status() if action == "status" else control_service(action)
+                self.events.put((action, status, None))
+            except (OSError, RuntimeError, subprocess.TimeoutExpired) as exc:
+                self.events.put((action, None, str(exc)))
+
+        threading.Thread(target=work, daemon=True).start()
+
+    def poll(self):
+        try:
+            action, status, error = self.events.get_nowait()
+        except queue.Empty:
+            pass
+        else:
+            self.busy = False
+            self.start_button.configure(state="normal")
+            self.stop_button.configure(state="normal")
+            self.status.configure(text=f"{SERVICE} : {error or status}")
+            if error and action != "status":
+                self.pending_export = None
+                messagebox.showerror("Syncthing", error)
+            elif action == "ensure":
+                callback, self.pending_export = self.pending_export, None
+                if callback:
+                    callback()
+            elif self.pending_export:
+                self.request("ensure")
+        if self.winfo_exists():
+            self.timer = self.after(100, self.poll)
+            if self.refresh_timer is None:
+                self.refresh_timer = self.after(5000, self.refresh)
+
+    def refresh(self):
+        self.refresh_timer = None
+        self.request("status")
+
+    def cleanup(self, event):
+        if event.widget == self:
+            self.pending_export = None
+            for timer in (self.timer, self.refresh_timer):
+                if timer:
+                    self.after_cancel(timer)