508 lines
20 KiB
Python
508 lines
20 KiB
Python
"""
|
|
Batch GUI: an in-app table for reviewing many files at once, instead of
|
|
one-by-one (gui/app.py) or via an external spreadsheet (batch/stage.py +
|
|
apply.py).
|
|
|
|
This exists specifically to avoid the CSV round-trip problems that hit
|
|
the spreadsheet workflow: LibreOffice adding a BOM, auto-reformatting a
|
|
date-looking column, or autocorrect changing a header's case. None of
|
|
that can happen here because nothing leaves this app as a file until you
|
|
actually apply — editing happens directly on the same in-memory records
|
|
apply_batch() acts on.
|
|
|
|
Behaviour:
|
|
- One row per file. Double-click a Date/Topic/Subtopic/Extra/More/
|
|
Version/Destination cell to edit it in place. Click the "Keep name"
|
|
cell to toggle it.
|
|
- Click a column header to sort by it — e.g. sort by File to group
|
|
everything from the same folder together before working through it,
|
|
the same grouping trick that helped in the spreadsheet workflow.
|
|
- Rows are never acted on individually — select one or many (ctrl/shift
|
|
click) and use the toolbar buttons to mark them confirmed/skip/pending.
|
|
A row's colour reflects its status at a glance.
|
|
- "Preview (dry run)" and "Apply for real" both act on every row
|
|
currently marked confirmed, exactly like the other two front ends.
|
|
- Shares the same .vault_renamer_state.json / .vault_renamer_log.csv as
|
|
the GUI and batch tools — work in any of the three, the others see it.
|
|
"""
|
|
|
|
import csv
|
|
import json
|
|
import queue
|
|
import sys
|
|
import threading
|
|
import tkinter as tk
|
|
from pathlib import Path
|
|
from tkinter import messagebox, ttk
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from core import scan, discover_folders, suggest_folders, apply_batch, find_duplicate_identities
|
|
from core.config import STATE_FILENAME, LOG_FILENAME
|
|
from core.content_extract import extract_text
|
|
from core.llm_suggest import suggest as llm_suggest, LLMSuggestError
|
|
|
|
# (internal field name, column header, starting width)
|
|
COLUMNS = [
|
|
("status", "Status", 80),
|
|
("rel_path", "File", 320),
|
|
("date_override", "Date", 90),
|
|
("topic", "Topic", 110),
|
|
("subtopic", "Subtopic", 130),
|
|
("extra", "Extra", 90),
|
|
("more", "More", 110),
|
|
("version", "Ver", 50),
|
|
("dest_folder", "Destination", 220),
|
|
("keep_filename", "Keep name", 80),
|
|
("preview", "New name preview", 320),
|
|
]
|
|
TEXT_EDITABLE_FIELDS = {"date_override", "topic", "subtopic", "extra", "more", "version"}
|
|
STATUS_COLORS = {
|
|
"pending": "#ffffff",
|
|
"confirmed": "#dff0d8",
|
|
"skip": "#f0f0f0",
|
|
"done": "#cfe2ff",
|
|
}
|
|
|
|
|
|
class BatchApp:
|
|
def __init__(self, win: tk.Tk, vault_root: Path):
|
|
self.win = win
|
|
self.vault_root = vault_root
|
|
self.win.title(f"Vault Renamer — Batch — {vault_root}")
|
|
self.win.geometry("1400x700")
|
|
|
|
self.records = []
|
|
self.by_iid = {} # treeview row id -> FileRecord
|
|
self.folders = []
|
|
self.hide_done = tk.BooleanVar(value=True)
|
|
self._active_editor = None
|
|
self._llm_queue = queue.Queue()
|
|
self._llm_total = 0
|
|
self._llm_completed = 0
|
|
|
|
self._build_ui()
|
|
self._load_and_scan()
|
|
|
|
# ---------- persistence (identical scheme to gui/app.py) ----------
|
|
def _state_path(self) -> Path:
|
|
return self.vault_root / STATE_FILENAME
|
|
|
|
def _log_path(self) -> Path:
|
|
return self.vault_root / LOG_FILENAME
|
|
|
|
def _load_state(self) -> dict:
|
|
p = self._state_path()
|
|
if p.exists():
|
|
try:
|
|
return json.loads(p.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return {}
|
|
return {}
|
|
|
|
def _save_state(self):
|
|
state = {
|
|
rec.identity: {
|
|
"status": rec.status, "date_override": rec.date_override,
|
|
"topic": rec.topic, "subtopic": rec.subtopic, "extra": rec.extra,
|
|
"more": rec.more, "version": rec.version,
|
|
"dest_folder": rec.dest_folder, "keep_filename": rec.keep_filename,
|
|
}
|
|
for rec in self.records if rec.status != "pending"
|
|
}
|
|
self._state_path().write_text(json.dumps(state, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
|
|
def _append_log(self, rows):
|
|
if not rows:
|
|
return
|
|
log_path = self._log_path()
|
|
write_header = not log_path.exists()
|
|
with open(log_path, "a", newline="", encoding="utf-8") as f:
|
|
writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
|
|
if write_header:
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
def _load_and_scan(self):
|
|
self.records = scan(self.vault_root)
|
|
self.folders = discover_folders(self.vault_root)
|
|
saved = self._load_state()
|
|
for rec in self.records:
|
|
saved_rec = saved.get(rec.identity)
|
|
if saved_rec:
|
|
for k, v in saved_rec.items():
|
|
setattr(rec, k, v)
|
|
|
|
duplicates = find_duplicate_identities(self.records)
|
|
if duplicates:
|
|
groups = "\n".join(
|
|
" " + ", ".join(r.rel_path for r in recs)
|
|
for recs in duplicates.values()
|
|
)
|
|
messagebox.showwarning(
|
|
"Identical files found",
|
|
f"{len(duplicates)} group(s) of byte-identical files were found:\n\n{groups}\n\n"
|
|
f"These can't be told apart by content, so progress-tracking may not "
|
|
f"distinguish them correctly if you process one and not the other. "
|
|
f"Consider deduplicating these separately before relying on saved progress "
|
|
f"for them specifically.",
|
|
)
|
|
|
|
self._populate_tree()
|
|
|
|
# ---------- UI ----------
|
|
def _build_ui(self):
|
|
toolbar = ttk.Frame(self.win, padding=6)
|
|
toolbar.pack(side="top", fill="x")
|
|
|
|
ttk.Button(toolbar, text="Mark Confirmed", command=lambda: self._mark_selected("confirmed")).pack(side="left", padx=2)
|
|
ttk.Button(toolbar, text="Mark Skip", command=lambda: self._mark_selected("skip")).pack(side="left", padx=2)
|
|
ttk.Button(toolbar, text="Reset to Pending", command=lambda: self._mark_selected("pending")).pack(side="left", padx=2)
|
|
ttk.Separator(toolbar, orient="vertical").pack(side="left", fill="y", padx=8)
|
|
ttk.Button(toolbar, text="Suggest folder (selected)", command=self._suggest_for_selected).pack(side="left", padx=2)
|
|
ttk.Button(toolbar, text="Set folder (selected)…", command=self._set_folder_dialog).pack(side="left", padx=2)
|
|
ttk.Separator(toolbar, orient="vertical").pack(side="left", fill="y", padx=8)
|
|
ttk.Button(toolbar, text="Suggest via LLM (selected)", command=self._llm_suggest_for_selected).pack(side="left", padx=2)
|
|
ttk.Separator(toolbar, orient="vertical").pack(side="left", fill="y", padx=8)
|
|
ttk.Checkbutton(toolbar, text="Hide already-done rows", variable=self.hide_done,
|
|
command=self._populate_tree).pack(side="left", padx=2)
|
|
|
|
tree_frame = ttk.Frame(self.win)
|
|
tree_frame.pack(side="top", fill="both", expand=True)
|
|
|
|
col_ids = [c[0] for c in COLUMNS]
|
|
self.tree = ttk.Treeview(tree_frame, columns=col_ids, show="headings", selectmode="extended")
|
|
for field, label, width in COLUMNS:
|
|
self.tree.heading(field, text=label, command=lambda f=field: self._sort_by(f, False))
|
|
self.tree.column(field, width=width, stretch=False)
|
|
|
|
vsb = ttk.Scrollbar(tree_frame, orient="vertical", command=self.tree.yview)
|
|
hsb = ttk.Scrollbar(tree_frame, orient="horizontal", command=self.tree.xview)
|
|
self.tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set)
|
|
self.tree.grid(row=0, column=0, sticky="nsew")
|
|
vsb.grid(row=0, column=1, sticky="ns")
|
|
hsb.grid(row=1, column=0, sticky="ew")
|
|
tree_frame.rowconfigure(0, weight=1)
|
|
tree_frame.columnconfigure(0, weight=1)
|
|
|
|
for status, color in STATUS_COLORS.items():
|
|
self.tree.tag_configure(status, background=color)
|
|
|
|
self.tree.bind("<Double-1>", self._on_double_click)
|
|
self.tree.bind("<Button-1>", self._on_single_click, add="+")
|
|
|
|
bottom = ttk.Frame(self.win, padding=6)
|
|
bottom.pack(side="bottom", fill="x")
|
|
self.count_label = ttk.Label(bottom, text="")
|
|
self.count_label.pack(side="left")
|
|
ttk.Button(bottom, text="Preview (dry run)", command=lambda: self._run(dry_run=True)).pack(side="right", padx=4)
|
|
ttk.Button(bottom, text="Apply for real", command=lambda: self._run(dry_run=False)).pack(side="right", padx=4)
|
|
|
|
# ---------- populate / refresh ----------
|
|
def _row_values(self, rec):
|
|
return (
|
|
rec.status, rec.rel_path, rec.date_override or rec.fs_date,
|
|
rec.topic, rec.subtopic, rec.extra, rec.more, rec.version,
|
|
rec.dest_folder, "✓" if rec.keep_filename else "", rec.new_filename_preview,
|
|
)
|
|
|
|
def _populate_tree(self):
|
|
self._destroy_active_editor()
|
|
self.tree.delete(*self.tree.get_children())
|
|
self.by_iid.clear()
|
|
|
|
for rec in self.records:
|
|
if self.hide_done.get() and rec.status == "done":
|
|
continue
|
|
iid = rec.rel_path # unique within a single scan, unlike identity —
|
|
self.tree.insert("", "end", iid=iid, values=self._row_values(rec), tags=(rec.status,))
|
|
self.by_iid[iid] = rec
|
|
|
|
self._update_counts()
|
|
|
|
def _update_counts(self):
|
|
total = len(self.records)
|
|
counts = {"pending": 0, "confirmed": 0, "skip": 0, "done": 0}
|
|
for rec in self.records:
|
|
counts[rec.status] = counts.get(rec.status, 0) + 1
|
|
self.count_label.config(
|
|
text=f"{counts['pending']} pending / {counts['confirmed']} confirmed / "
|
|
f"{counts['skip']} skipped / {counts['done']} done / {total} total"
|
|
)
|
|
|
|
def _refresh_row(self, iid):
|
|
rec = self.by_iid.get(iid)
|
|
if rec is None:
|
|
return
|
|
self.tree.item(iid, values=self._row_values(rec), tags=(rec.status,))
|
|
|
|
# ---------- sorting ----------
|
|
def _sort_by(self, field, reverse):
|
|
items = [(self.tree.set(k, field), k) for k in self.tree.get_children("")]
|
|
items.sort(key=lambda pair: pair[0].lower(), reverse=reverse)
|
|
for index, (_, k) in enumerate(items):
|
|
self.tree.move(k, "", index)
|
|
self.tree.heading(field, command=lambda: self._sort_by(field, not reverse))
|
|
|
|
# ---------- inline editing ----------
|
|
def _destroy_active_editor(self):
|
|
if self._active_editor is not None:
|
|
try:
|
|
self._active_editor.destroy()
|
|
except tk.TclError:
|
|
pass
|
|
self._active_editor = None
|
|
|
|
def _on_single_click(self, event):
|
|
# Clicking the "Keep name" cell toggles it immediately — no need
|
|
# to double-click for a single boolean flip.
|
|
region = self.tree.identify("region", event.x, event.y)
|
|
if region != "cell":
|
|
return
|
|
row_id = self.tree.identify_row(event.y)
|
|
col_id = self.tree.identify_column(event.x)
|
|
if not row_id or not col_id:
|
|
return
|
|
field = self._field_for_column(col_id)
|
|
if field == "keep_filename":
|
|
self._toggle_keep_filename(row_id)
|
|
|
|
def _on_double_click(self, event):
|
|
region = self.tree.identify("region", event.x, event.y)
|
|
if region != "cell":
|
|
return
|
|
row_id = self.tree.identify_row(event.y)
|
|
col_id = self.tree.identify_column(event.x)
|
|
if not row_id or not col_id:
|
|
return
|
|
field = self._field_for_column(col_id)
|
|
if field is None or field == "keep_filename":
|
|
return # keep_filename is single-click toggle, not text-edit
|
|
if field not in TEXT_EDITABLE_FIELDS and field != "dest_folder":
|
|
return # status/rel_path/preview are not directly editable
|
|
|
|
self._start_edit(row_id, col_id, field)
|
|
|
|
def _field_for_column(self, col_id: str):
|
|
index = int(col_id[1:]) - 1
|
|
if 0 <= index < len(COLUMNS):
|
|
return COLUMNS[index][0]
|
|
return None
|
|
|
|
def _start_edit(self, row_id, col_id, field):
|
|
self._destroy_active_editor()
|
|
bbox = self.tree.bbox(row_id, col_id)
|
|
if not bbox:
|
|
return
|
|
x, y, w, h = bbox
|
|
current = self.tree.set(row_id, field)
|
|
|
|
if field == "dest_folder":
|
|
editor = ttk.Combobox(self.tree, values=self.folders)
|
|
editor.set(current)
|
|
else:
|
|
editor = ttk.Entry(self.tree)
|
|
editor.insert(0, current)
|
|
editor.select_range(0, tk.END)
|
|
|
|
editor.place(x=x, y=y, width=w, height=h)
|
|
editor.focus()
|
|
self._active_editor = editor
|
|
|
|
def commit(_event=None):
|
|
self._commit_edit(row_id, field, editor.get())
|
|
self._destroy_active_editor()
|
|
|
|
editor.bind("<Return>", commit)
|
|
editor.bind("<FocusOut>", commit)
|
|
editor.bind("<Escape>", lambda _e: self._destroy_active_editor())
|
|
|
|
def _commit_edit(self, row_id, field, new_value):
|
|
rec = self.by_iid.get(row_id)
|
|
if rec is None:
|
|
return
|
|
setattr(rec, field, new_value.strip())
|
|
self._refresh_row(row_id)
|
|
|
|
def _toggle_keep_filename(self, row_id):
|
|
rec = self.by_iid.get(row_id)
|
|
if rec is None:
|
|
return
|
|
rec.keep_filename = not rec.keep_filename
|
|
self._refresh_row(row_id)
|
|
|
|
# ---------- bulk actions on selection ----------
|
|
def _mark_selected(self, status):
|
|
for row_id in self.tree.selection():
|
|
rec = self.by_iid.get(row_id)
|
|
if rec is None:
|
|
continue
|
|
rec.status = status
|
|
self._refresh_row(row_id)
|
|
self._update_counts()
|
|
|
|
def _suggest_for_selected(self):
|
|
for row_id in self.tree.selection():
|
|
rec = self.by_iid.get(row_id)
|
|
if rec is None:
|
|
continue
|
|
query = " ".join([rec.topic, rec.subtopic, rec.filename])
|
|
suggestions = suggest_folders(query, self.folders, top_n=1)
|
|
if suggestions:
|
|
rec.dest_folder = suggestions[0]
|
|
self._refresh_row(row_id)
|
|
|
|
def _set_folder_dialog(self):
|
|
selected = self.tree.selection()
|
|
if not selected:
|
|
messagebox.showinfo("No selection", "Select one or more rows first.")
|
|
return
|
|
|
|
dialog = tk.Toplevel(self.win)
|
|
dialog.title("Set destination folder")
|
|
dialog.geometry("420x120")
|
|
ttk.Label(dialog, text=f"Apply to {len(selected)} selected row(s):").pack(pady=(12, 4))
|
|
var = tk.StringVar()
|
|
combo = ttk.Combobox(dialog, textvariable=var, values=self.folders, width=45)
|
|
combo.pack(pady=4)
|
|
combo.focus()
|
|
|
|
def apply_folder():
|
|
folder = var.get().strip()
|
|
if folder:
|
|
for row_id in selected:
|
|
rec = self.by_iid.get(row_id)
|
|
if rec is None:
|
|
continue
|
|
rec.dest_folder = folder
|
|
self._refresh_row(row_id)
|
|
dialog.destroy()
|
|
|
|
ttk.Button(dialog, text="Apply", command=apply_folder).pack(pady=8)
|
|
combo.bind("<Return>", lambda _e: apply_folder())
|
|
|
|
# ---------- LLM suggestion (background thread) ----------
|
|
def _llm_suggest_for_selected(self):
|
|
selected = self.tree.selection()
|
|
if not selected:
|
|
messagebox.showinfo("No selection", "Select one or more rows first.")
|
|
return
|
|
|
|
targets = [self.by_iid[row_id] for row_id in selected if row_id in self.by_iid]
|
|
self._llm_total = len(targets)
|
|
self._llm_completed = 0
|
|
self.count_label.config(text=f"Asking the model… 0/{self._llm_total}")
|
|
|
|
thread = threading.Thread(target=self._llm_worker, args=(targets,), daemon=True)
|
|
thread.start()
|
|
self.win.after(100, self._poll_llm_queue)
|
|
|
|
def _llm_worker(self, targets):
|
|
# Runs off the main thread — must never touch Tkinter widgets
|
|
# directly here, only read plain Python data and push results
|
|
# through the thread-safe queue for the main thread to apply.
|
|
for rec in targets:
|
|
try:
|
|
excerpt = extract_text(rec.path) or ""
|
|
result = llm_suggest(rec.filename, rec.current_folder, excerpt, self.folders)
|
|
self._llm_queue.put((rec.rel_path, result, None))
|
|
except LLMSuggestError as e:
|
|
self._llm_queue.put((rec.rel_path, None, str(e)))
|
|
except Exception as e: # a single file's failure must not kill the whole batch
|
|
self._llm_queue.put((rec.rel_path, None, f"Unexpected error: {e}"))
|
|
|
|
def _poll_llm_queue(self):
|
|
errors = []
|
|
try:
|
|
while True:
|
|
row_id, result, error = self._llm_queue.get_nowait()
|
|
self._llm_completed += 1
|
|
if error:
|
|
errors.append(f"{row_id}: {error}")
|
|
elif row_id in self.by_iid:
|
|
rec = self.by_iid[row_id]
|
|
if result.get("topic"):
|
|
rec.topic = result["topic"]
|
|
if result.get("subtopic"):
|
|
rec.subtopic = result["subtopic"]
|
|
if result.get("dest_folder"):
|
|
rec.dest_folder = result["dest_folder"]
|
|
self._refresh_row(row_id)
|
|
except queue.Empty:
|
|
pass
|
|
|
|
if self._llm_completed < self._llm_total:
|
|
self.count_label.config(text=f"Asking the model… {self._llm_completed}/{self._llm_total}")
|
|
self.win.after(150, self._poll_llm_queue)
|
|
else:
|
|
self._update_counts()
|
|
if errors:
|
|
shown = "\n".join(errors[:8])
|
|
more = f"\n… and {len(errors) - 8} more" if len(errors) > 8 else ""
|
|
messagebox.showwarning(
|
|
"Some suggestions failed",
|
|
f"{len(errors)} of {self._llm_total} file(s) got no suggestion:\n\n{shown}{more}",
|
|
)
|
|
|
|
# ---------- run ----------
|
|
def _run(self, dry_run: bool):
|
|
to_confirm = [rec for rec in self.records if rec.status == "confirmed"]
|
|
if not to_confirm:
|
|
messagebox.showinfo("Nothing to do", "No rows are marked confirmed.")
|
|
return
|
|
|
|
if not dry_run:
|
|
proceed = messagebox.askyesno(
|
|
"Apply for real?",
|
|
f"{len(to_confirm)} file(s) will be moved/renamed now. This cannot be "
|
|
f"undone automatically. Continue?",
|
|
)
|
|
if not proceed:
|
|
return
|
|
|
|
log_rows = apply_batch(to_confirm, dry_run=dry_run, log_path=None)
|
|
self._append_log(log_rows)
|
|
|
|
moved = sum(1 for r in log_rows if r["result"] in ("moved", "would move"))
|
|
collisions = [r for r in log_rows if r["result"].startswith("SKIPPED")]
|
|
|
|
if not dry_run:
|
|
# A collision leaves status at "confirmed" (apply_batch only ever advances
|
|
# it to "done" on success) — reset it to "pending" so the row stays visible
|
|
# and actionable instead of silently vanishing from every front end's view.
|
|
for rec, row in zip(to_confirm, log_rows):
|
|
if row["result"].startswith("SKIPPED"):
|
|
rec.status = "pending"
|
|
|
|
summary = f"{'Would move' if dry_run else 'Moved'}: {moved}\n"
|
|
if collisions:
|
|
summary += f"Skipped (target already exists): {len(collisions)}\n"
|
|
summary += "\n".join(f" - {r['old_path']}" for r in collisions[:10])
|
|
|
|
for rec, row in zip(to_confirm, log_rows):
|
|
self._refresh_row(rec.identity)
|
|
|
|
messagebox.showinfo("Dry run result" if dry_run else "Apply result", summary)
|
|
|
|
if not dry_run:
|
|
self._save_state()
|
|
if self.hide_done.get():
|
|
self._populate_tree()
|
|
self._update_counts()
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) != 2:
|
|
print("Usage: python3 gui/batch_app.py /path/to/vault-root")
|
|
sys.exit(1)
|
|
vault_root = Path(sys.argv[1]).expanduser().resolve()
|
|
if not vault_root.is_dir():
|
|
print(f"Not a directory: {vault_root}")
|
|
sys.exit(1)
|
|
win = tk.Tk()
|
|
BatchApp(win, vault_root)
|
|
win.mainloop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|