diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e69de29 diff --git a/README.md b/README.md index e69de29..3feb124 100644 --- a/README.md +++ b/README.md @@ -0,0 +1,212 @@ +# vault-renamer — shared core + +This is step one: the shared logic both front ends (GUI and batch/spreadsheet) +will sit on top of. Nothing here touches your real files yet except +`preview_scan.py`, which only *reads*. + +## Layout + +``` +vault-renamer/ +├── core/ +│ ├── config.py # junk patterns to ignore, "already conventioned" regex +│ ├── sanitiser.py # filename-building logic, ported from Dokumenten_Beschrifter.html +│ ├── model.py # FileRecord — one row per file, shared by both front ends +│ ├── scanner.py # walks a folder, builds FileRecords, discovers destination folders live +│ ├── suggest.py # simple keyword-overlap folder suggestion (no ML, no network) +│ └── executor.py # applies a reviewed batch to disk — dry-run by default, always logged +└── preview_scan.py # read-only sanity check, run this first +``` + +## LLM-assisted suggestions (new) — optional + +Uses your local llama-swap instance to suggest topic/subtopic/destination +by actually reading each file, not just its filename. Runs entirely +against `http://localhost:8080` — nothing leaves your machine. + +**Setup, only needed for `.pdf`/`.docx`/`.xlsx` content reading** (`.md`/`.txt` +work with no setup): +```bash +pip install pypdf python-docx openpyxl --break-system-packages +``` +Missing one of these doesn't break anything — that specific format just +falls back to filename-only until it's installed. + +**In the batch GUI:** select rows, click "Suggest via LLM (selected)". +Runs in a background thread — the window stays responsive while it waits +on the model. Topic/subtopic/destination fields fill in as each result +comes back; a file the model couldn't help with is left completely +unchanged, never blanked. A destination the model invents that isn't one +of your real folders is discarded rather than applied. + +**Defaults to the `daily` model** (fast, good for this) **against +`localhost:8080`** (llama-swap). Override either without touching code: +```bash +VAULT_RENAMER_LLM_MODEL=architect VAULT_RENAMER_LLM_ENDPOINT=http://localhost:8080/v1/chat/completions python3 gui/batch_app.py /path +``` + +## Batch GUI (new) — a table, not a spreadsheet round-trip + +```bash +python3 gui/batch_app.py /path/to/folder +``` + +This exists specifically because the CSV workflow's round-trip through +LibreOffice caused real problems (BOM characters, auto-reformatted dates, +autocorrected header casing). Editing happens directly on the same +in-memory data `apply_batch()` acts on — nothing is ever serialized to a +file and re-parsed in between, so none of that class of bug can happen here. + +How it behaves: +- One row per file, all visible at once — not one-at-a-time like `gui/app.py`. +- **Double-click** a Date/Topic/Subtopic/Extra/More/Version/Destination + cell to edit it in place. **Single-click** the "Keep name" cell to toggle it. +- **Click a column header to sort by it** — sort by File to group + everything from one folder together before working through it, the + same trick that helped in the spreadsheet. +- Select one or many rows (ctrl/shift-click, like any file manager) and + use the toolbar to mark them **Confirmed** / **Skip** / **Pending** — + no risk of a typo'd status value, unlike a free-text CSV column. Row + colour reflects status at a glance. +- **Suggest folder (selected)** applies each selected row's own keyword + suggestion individually. **Set folder (selected)…** applies one chosen + folder to every selected row at once — useful after sorting a whole + folder's worth of files together. +- **Preview (dry run)** and **Apply for real** act on every row currently + marked Confirmed. A collision (target already exists) resets that row + back to Pending rather than leaving it in limbo, so it stays visible + and you can fix and retry it. +- Shares `.vault_renamer_state.json` / `.vault_renamer_log.csv` with both + other front ends — confirmed by actually running this after the GUI + and the CSV batch tool on the same folder and checking all three agree. + +**A real limitation worth knowing:** if two files are genuinely +byte-identical (same size, same content — true duplicates, or two empty +placeholder files), they get the same internal identity, because there's +nothing in their content to tell them apart. All three front ends now +warn you when this is detected. If you process one and not the other, +progress-tracking may not distinguish them correctly. The actual fix is +deduplicating true duplicates as a separate pass, not a cleverer identity +scheme — no content-based approach can distinguish files that are +genuinely, byte-for-byte the same. + +## Batch / spreadsheet stager + +Two scripts instead of one app — generate a CSV, edit it in Calc, apply it. + +```bash +python3 batch/stage.py /path/to/folder # writes staging.csv +# ... edit staging.csv in LibreOffice Calc / Excel / whatever ... +python3 batch/apply.py /path/to/folder staging.csv # dry run — just previews +python3 batch/apply.py /path/to/folder staging.csv --apply # actually does it +``` + +How it behaves: +- `stage.py` writes one row per file not yet marked `done`. Files already + `done` (by either front end) are left out entirely; files marked `skip` + are left out too unless you pass `--include-skipped`. It also writes a + `..._folders.txt` alongside the CSV — the full list of real destination + folders, to copy/paste into `dest_folder` cells. +- Every row starts with `status = pending`. Editing name/destination + fields does nothing by itself — a row is only acted on once you change + its `status` to `confirmed`. Set it to `skip` to mark a file as + deliberately left alone without touching it. +- `apply.py` defaults to a **dry run** — it prints and logs exactly what + it would do without touching a single file. Only `--apply` does it for real. +- Rows you leave as `pending` are completely ignored — nothing is lost, + they'll show up in the next `stage.py` run (or in the GUI) unchanged. +- **Shares progress with the GUI** — same `.vault_renamer_state.json`, + same `.vault_renamer_log.csv`, same identity-based matching. A file + processed in one front end shows up as `done` in the other. Confirmed + by actually running one, then the other, on the same folder — not + assumed. +- `identity` and `current_path` columns are there so `apply.py` can find + the right file again — don't edit them. If a row's file can't be found + anymore (e.g. moved by the GUI in the meantime), `apply.py` reports it + and skips that row rather than guessing. + +## GUI (new) + +```bash +python3 gui/app.py /path/to/some/folder +``` + +On Arch/EndeavourOS, Tkinter needs the system `tk` package if it's not +already installed: +```bash +sudo pacman -S tk +``` + +How it behaves: +- One file at a time. The left list shows everything still pending; + click one to load it into the form on the right. +- Every field defaults to "no change" — an untouched file stays exactly + as it is. Files that already match the naming convention default to + **"keep current filename (only move)"** ticked, so opening one just to + relocate it doesn't blank out a name that was already fine. Typing into + any name field automatically unticks that box. +- The preview always shows the exact new name and destination *before* + you click Confirm. +- **Confirm && Move applies immediately** — this isn't a queue you review + later, it's the same as renaming the file yourself, just faster. That's + deliberate: you've already seen the exact result in the preview. +- Progress and every field you've typed are saved automatically to + `.vault_renamer_state.json` in the scanned folder. Close the app + mid-batch, reopen later (even after a restart), and it picks up exactly + where you left off — done files stay done even though they've since + moved to a different path, because progress is tracked by a content + fingerprint (`core/identity.py`), not by location. +- Every move — real or attempted — is appended to + `.vault_renamer_log.csv` in the scanned folder, so there's a full audit + trail even if a session gets interrupted. +- "Skip" marks a file as deliberately left alone (distinct from "haven't + gotten to it yet"). Tick "show skipped too" to bring them back into view. +- If a computed target filename already exists, the file is left + untouched and you get a warning — never silently overwritten. + +Both `.vault_renamer_state.json` and `.vault_renamer_log.csv` are +automatically excluded from the scan itself, so the tool never tries to +rename its own bookkeeping files. + +## Try it first — read-only, changes nothing + +```bash +cd vault-renamer +python3 preview_scan.py /path/to/some/folder +``` + +Start with something small — `03_PRIVAT` rather than the whole vault — and check: +- Does the file count look right (junk correctly excluded — `.stfolder`, + `.sync-conflict-*`, `Thumbs.db`, LibreOffice lock files)? +- Does the filesystem date look like a sane *starting point* (not + necessarily correct — you're always meant to review/override it)? +- Does the "already conventioned" count roughly match what you'd expect? + +This is exactly the kind of small, cheap test worth running before either +front end gets built on top — if the date-guessing or junk-filtering is off +in some way I haven't anticipated, better to find that now on read-only code +than after a GUI is built on it. + +## What's deliberately NOT here yet + +- No folder taxonomy hardcoded anywhere — `discover_folders()` reads your + actual directory tree fresh every run, so it stays correct as your + folders change. +- No GUI, no CSV/spreadsheet staging generator, no actual renaming happening + anywhere — those are the next two pieces, once this core checks out. +- The suggestion engine (`suggest.py`) is intentionally simple: it only + matches shared words between what you type and folder names. It won't + read file *content* to guess a folder — that's a reasonable next upgrade + for `.md`/`.txt` files if the simple version feels too weak once you've + tried it for real, but not worth building before we know it's needed. + +## Safety notes baked in + +- `executor.apply_batch()` defaults to `dry_run=True`. A real run requires + explicitly passing `dry_run=False`. +- Every run — dry or real — writes the same CSV log, so a dry-run's log + is the exact preview of what a real run would do. +- A target path that already exists is skipped with a clear reason, never + silently overwritten. +- Only rows explicitly marked `status = "confirmed"` are touched at all; + everything else is left alone. diff --git a/batch/apply.py b/batch/apply.py new file mode 100644 index 0000000..e206608 --- /dev/null +++ b/batch/apply.py @@ -0,0 +1,207 @@ +#!/usr/bin/env python3 +""" +Reads a CSV produced by stage.py (after you've filled it in) and applies +it: moves/renames every row marked status=confirmed, records status=skip +into the shared progress file, and leaves status=pending rows completely +untouched. + +Defaults to a DRY RUN — prints what it would do and logs it, but touches +nothing on disk. Add --apply to actually do it. + +Usage: + python3 batch/apply.py /path/to/folder staging.csv # dry run + python3 batch/apply.py /path/to/folder staging.csv --apply # for real +""" + +import argparse +import csv +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from core import scan, apply_batch, find_duplicate_identities +from core.config import STATE_FILENAME, LOG_FILENAME + + +REQUIRED_COLUMNS = { + "status", "current_path", "date_override", "topic", "subtopic", + "extra", "more", "version", "dest_folder", "keep_filename", "identity", +} + + +def _read_rows(csv_path: Path) -> list: + """Reads the CSV defensively: strips a UTF-8 BOM if LibreOffice added + one, and normalises header casing — a header accidentally typed or + autocorrected as 'Status' instead of 'status' shouldn't crash the + whole run.""" + with open(csv_path, newline="", encoding="utf-8-sig") as f: + reader = csv.DictReader(f) + rows = [{k.strip().lower(): v for k, v in row.items()} for row in reader] + + if not rows: + print(f"{csv_path} has no data rows.") + sys.exit(1) + + missing = REQUIRED_COLUMNS - set(rows[0].keys()) + if missing: + print(f"{csv_path} is missing expected column(s): {', '.join(sorted(missing))}") + print(f"Columns actually found: {', '.join(rows[0].keys())}") + print(f"This usually means the wrong file got opened/saved in the spreadsheet " + f"(e.g. {LOG_FILENAME} instead of the staging CSV), or a header got " + f"renamed by accident. Check the file and try again.") + sys.exit(1) + + return rows + return str(value).strip().lower() in ("true", "yes", "y", "1") + + +def _load_state(root: Path) -> dict: + p = root / STATE_FILENAME + if p.exists(): + try: + return json.loads(p.read_text(encoding="utf-8")) + except Exception: + return {} + return {} + + +def _save_state(root: Path, state: dict): + (root / STATE_FILENAME).write_text( + json.dumps(state, indent=2, ensure_ascii=False), encoding="utf-8" + ) + + +def _state_entry(rec, status: str) -> dict: + return { + "status": 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, + } + + +def main(): + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("root", help="Folder that was scanned") + ap.add_argument("csv_path", help="The filled-in CSV from stage.py") + ap.add_argument("--apply", action="store_true", + help="Actually move/rename files. Without this: dry-run only.") + args = ap.parse_args() + + root = Path(args.root).expanduser().resolve() + if not root.is_dir(): + print(f"Not a directory: {root}") + sys.exit(1) + + csv_path = Path(args.csv_path) + if not csv_path.is_file(): + print(f"Not a file: {csv_path}") + sys.exit(1) + + rows = _read_rows(csv_path) + print(f"Read {len(rows)} rows from {csv_path}") + + records = scan(root) + duplicates = find_duplicate_identities(records) + if duplicates: + print(f"\nWARNING: {len(duplicates)} group(s) of byte-identical files found — " + f"a CSV row matching one of these may resolve to the wrong twin, since " + f"they can't be told apart by content:") + for recs in duplicates.values(): + print(" " + ", ".join(r.rel_path for r in recs)) + + by_identity = {rec.identity: rec for rec in records} + + to_confirm, to_skip, not_found = [], [], [] + still_pending = 0 + + for row in rows: + rec = by_identity.get(row["identity"]) + if rec is None: + not_found.append(row["current_path"]) + continue + + status = row["status"].strip().lower() + if status == "confirmed": + rec.date_override = row["date_override"] + rec.topic = row["topic"] + rec.subtopic = row["subtopic"] + rec.extra = row["extra"] + rec.more = row["more"] + rec.version = row["version"] + rec.dest_folder = row["dest_folder"] + rec.keep_filename = _parse_bool(row["keep_filename"]) + rec.status = "confirmed" + to_confirm.append(rec) + elif status == "skip": + rec.status = "skip" + to_skip.append(rec) + else: + still_pending += 1 + # "pending" (or anything else/blank): left completely alone + + dry_run = not args.apply + + if not_found: + print(f"\nWARNING: {len(not_found)} row(s) couldn't be matched to a current file — " + f"probably already moved by another tool since this CSV was made. Skipped:") + for p in not_found: + print(f" - {p}") + + print(f"\nMatched to real files: {len(rows) - len(not_found)} / {len(rows)}") + print(f" status=confirmed : {len(to_confirm)}") + print(f" status=skip : {len(to_skip)}") + print(f" status=pending : {still_pending} (left completely alone — not an error)") + + if len(to_confirm) == 0 and len(to_skip) == 0 and still_pending > 0: + print(f"\nNothing to do: every matched row is still status=pending.") + print(f"apply.py only acts on rows you've explicitly set to 'confirmed' (or 'skip').") + print(f"Open {csv_path.name} and check the status column actually got changed — " + f"a common slip is editing the topic/folder cells but leaving status untouched.") + return + + log_rows = apply_batch(to_confirm, dry_run=dry_run, log_path=None) + + log_path = root / LOG_FILENAME + if log_rows: + write_header = not log_path.exists() + with open(log_path, "a", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=list(log_rows[0].keys())) + if write_header: + writer.writeheader() + writer.writerows(log_rows) + + moved = sum(1 for r in log_rows if r["result"] in ("moved", "would move")) + unchanged = sum(1 for r in log_rows if r["result"] == "unchanged") + collisions = sum(1 for r in log_rows if r["result"].startswith("SKIPPED")) + + print(f"\n{'Would move' if dry_run else 'Moved'}: {moved}") + if unchanged: + print(f"Unchanged (new name+folder computed identical to current): {unchanged}") + print(f"Skipped — target already exists: {collisions}") + print(f"Marked as deliberately skipped: {len(to_skip)}") + + if dry_run: + print(f"\nThis was a DRY RUN — nothing was touched, nothing was saved. " + f"Review {LOG_FILENAME}, then re-run with --apply to actually do it.") + return + + state = _load_state(root) + for rec in to_confirm: + if rec.status == "done": + state[rec.identity] = _state_entry(rec, "done") + for rec in to_skip: + state[rec.identity] = _state_entry(rec, "skip") + _save_state(root, state) + print(f"Progress saved to {STATE_FILENAME} — the GUI will see these as done/skipped too.") + + +if __name__ == "__main__": + main() diff --git a/batch/stage.py b/batch/stage.py new file mode 100644 index 0000000..356a072 --- /dev/null +++ b/batch/stage.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 +""" +Generates a CSV you fill in (LibreOffice Calc or similar) to review many +files at once, instead of one-by-one in the GUI. + +Usage: + python3 batch/stage.py /path/to/folder + python3 batch/stage.py /path/to/folder -o mybatch.csv --include-skipped + +Shares progress with the GUI: files already marked "done" (by either +front end) are left out entirely, and anything already "skip"ped is left +out unless you ask for it with --include-skipped. + +Columns you edit: + status leave "pending" to do nothing with a row, set to + "confirmed" for apply.py to act on it, or "skip" to + mark it deliberately left alone + date_override starts pre-filled with the filesystem guess — edit freely + topic / subtopic / extra / more / version + dest_folder starts pre-filled with the file's current folder — + see the matching *_folders.txt for the full list of + real destination folders to copy from + keep_filename TRUE/FALSE — TRUE means "move only, don't rename". + Starts TRUE for files that already match the naming + convention, so you don't have to retype a name that + was already fine just to move the file. + +Columns NOT meant to be edited (used to find the file again in apply.py): + identity, current_path +""" + +import argparse +import csv +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from core import scan, discover_folders +from core.config import STATE_FILENAME + +FIELDNAMES = [ + "status", "current_path", "name_matches_convention", "fs_date", + "date_override", "topic", "subtopic", "extra", "more", "version", + "dest_folder", "keep_filename", "identity", +] + + +def _load_state(root: Path) -> dict: + p = root / STATE_FILENAME + if p.exists(): + try: + return json.loads(p.read_text(encoding="utf-8")) + except Exception: + return {} + return {} + + +def main(): + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("root", help="Folder to scan") + ap.add_argument("-o", "--output", default="staging.csv") + ap.add_argument("--include-skipped", action="store_true", + help="Also include files previously marked 'skip'") + args = ap.parse_args() + + root = Path(args.root).expanduser().resolve() + if not root.is_dir(): + print(f"Not a directory: {root}") + sys.exit(1) + + records = scan(root) + saved = _load_state(root) + + rows = [] + for rec in records: + identity = rec.identity # one read per file, reused below — not recomputed + saved_rec = saved.get(identity) + if saved_rec: + for k, v in saved_rec.items(): + setattr(rec, k, v) + + if rec.status == "done": + continue + if rec.status == "skip" and not args.include_skipped: + continue + + rows.append({ + "status": rec.status, + "current_path": rec.rel_path, + "name_matches_convention": rec.name_matches_convention, + "fs_date": rec.fs_date, + "date_override": rec.date_override or rec.fs_date, + "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, + "identity": identity, + }) + + out_path = Path(args.output) + with open(out_path, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=FIELDNAMES) + writer.writeheader() + writer.writerows(rows) + + folders = discover_folders(root) + folders_path = out_path.with_name(out_path.stem + "_folders.txt") + folders_path.write_text("\n".join(folders), encoding="utf-8") + + print(f"Wrote {len(rows)} rows to {out_path}") + print(f"Wrote {len(folders)} known destination folders to {folders_path} " + f"(reference list only — copy/paste into dest_folder)") + + +if __name__ == "__main__": + main() diff --git a/core/__init__.py b/core/__init__.py new file mode 100644 index 0000000..7bf3067 --- /dev/null +++ b/core/__init__.py @@ -0,0 +1,12 @@ +from .model import FileRecord +from .scanner import scan, discover_folders +from .sanitiser import sanitise, build_filename +from .suggest import suggest_folders +from .executor import apply_batch, plan_new_path +from .identity import find_duplicate_identities + +__all__ = [ + "FileRecord", "scan", "discover_folders", + "sanitise", "build_filename", "suggest_folders", + "apply_batch", "plan_new_path", "find_duplicate_identities", +] diff --git a/core/config.py b/core/config.py new file mode 100644 index 0000000..6c34c3f --- /dev/null +++ b/core/config.py @@ -0,0 +1,48 @@ +""" +Central configuration for the vault renamer. + +Deliberately NOT in here: the folder taxonomy (01_INBOX, 04_ARBEIT/Cafe_Bach, +etc). That's discovered live from disk every run in scanner.py, since the +taxonomy is expected to keep changing — hardcoding it here would mean +editing code every time a folder gets renamed or added. +""" + +import re + +# Directory names to skip entirely when scanning: never listed as files, +# never offered as a destination folder. +IGNORE_DIR_NAMES = { + ".git", ".stfolder", ".stversions", "__pycache__", "node_modules", + ".venv", "venv", ".idea", ".vscode", +} + +# The tool's own bookkeeping files — never scanned, never shown as "needing +# a rename". Without this, the app would try to rename its own state/log +# files sitting right there in the scanned folder. +STATE_FILENAME = ".vault_renamer_state.json" +LOG_FILENAME = ".vault_renamer_log.csv" +TOOL_STATE_FILENAMES = {STATE_FILENAME, LOG_FILENAME} + +# Filename patterns to skip entirely — pure sync/system noise. Never shown +# in any review list, never touched. +IGNORE_FILE_PATTERNS = [ + re.compile(r"^\.sync-conflict-.*"), + re.compile(r"^Thumbs\.db$", re.IGNORECASE), + re.compile(r"^\.~lock\..*#$"), # LibreOffice lock files + re.compile(r"^\.directory$"), + re.compile(r"^desktop\.ini$", re.IGNORECASE), + re.compile(r"^\.DS_Store$"), +] + +# A filename counts as "already conventioned" if it starts with +# YYYY-MM-DD_ followed by at least one more character before the extension. +ALREADY_CONVENTIONED = re.compile(r"^\d{4}-\d{2}-\d{2}_.+") + +# Formats seen in the vault that are *close* to the convention but not +# quite it — flagged separately so they stand out from genuinely untouched +# files rather than being lumped in with everything else. +NEAR_CONVENTION_PATTERNS = [ + re.compile(r"^\d{8}_.+"), # e.g. 20260827_161234 + re.compile(r"^\d{2}-\d{2}-\d{4}.+"), # DD-MM-YYYY style + re.compile(r"^\d{2}_\d{2}_\d{4}.+"), # DD_MM_YYYY style +] diff --git a/core/content_extract.py b/core/content_extract.py new file mode 100644 index 0000000..b2a221f --- /dev/null +++ b/core/content_extract.py @@ -0,0 +1,87 @@ +""" +Extracts a text snippet from a file so an LLM suggestion can be based on +what's actually in the file, not just its name — helps most for the +genuinely opaque cases (scans, "Download (3).jpg"-style names) that a +filename-only guess can't do anything with. + +.md/.txt need nothing extra. .pdf/.docx/.xlsx each need one small +pip-installed library: + + pip install pypdf python-docx openpyxl --break-system-packages + +If a library isn't installed, that format quietly returns None (treated +as "no content available", falls back to filename-only) rather than +crashing anything — one missing library never blocks the others. +""" + +from pathlib import Path +from typing import Optional + +_MAX_CHARS = 2000 # a snippet is enough for a topic guess; keeps the LLM prompt small and fast + + +def extract_text(path: Path) -> Optional[str]: + ext = path.suffix.lower() + try: + if ext in (".md", ".txt", ".csv"): + return _read_plain(path) + if ext == ".pdf": + return _read_pdf(path) + if ext == ".docx": + return _read_docx(path) + if ext == ".xlsx": + return _read_xlsx(path) + except Exception: + return None # a corrupt/unreadable file falls back to filename-only, not a crash + return None # unsupported format + + +def _read_plain(path: Path) -> str: + return path.read_text(encoding="utf-8", errors="ignore")[:_MAX_CHARS] + + +def _read_pdf(path: Path) -> Optional[str]: + try: + from pypdf import PdfReader + except ImportError: + return None + reader = PdfReader(str(path)) + chunks = [] + total = 0 + for page in reader.pages[:3]: # first few pages is plenty for a topic guess + text = page.extract_text() or "" + chunks.append(text) + total += len(text) + if total > _MAX_CHARS: + break + return "".join(chunks)[:_MAX_CHARS] + + +def _read_docx(path: Path) -> Optional[str]: + try: + import docx + except ImportError: + return None + document = docx.Document(str(path)) + text = "\n".join(p.text for p in document.paragraphs) + return text[:_MAX_CHARS] + + +def _read_xlsx(path: Path) -> Optional[str]: + try: + import openpyxl + except ImportError: + return None + wb = openpyxl.load_workbook(str(path), read_only=True, data_only=True) + parts = [] + total = 0 + for ws in wb.worksheets[:2]: + for row in ws.iter_rows(max_row=20, values_only=True): + line = " ".join(str(c) for c in row if c is not None) + parts.append(line) + total += len(line) + if total > _MAX_CHARS: + break + if total > _MAX_CHARS: + break + return "\n".join(parts)[:_MAX_CHARS] diff --git a/core/executor.py b/core/executor.py new file mode 100644 index 0000000..d5c9874 --- /dev/null +++ b/core/executor.py @@ -0,0 +1,66 @@ +""" +Applies a reviewed batch of FileRecords to disk. + +Safety rules, non-negotiable regardless of which front end calls this: + - dry_run=True by default. Nothing is touched unless you explicitly + pass dry_run=False. + - Every run produces a log (same shape for dry-run and real runs), so + a dry-run's log IS the preview of exactly what a real run would do. + - A file is only ever touched if its status is "confirmed" — "pending" + or "skip" rows are left completely alone. + - If the computed target path already exists, the file is skipped with + a clear reason rather than silently overwritten. +""" + +import csv +import shutil +from datetime import datetime +from pathlib import Path +from typing import List, Optional + +from .model import FileRecord + + +def plan_new_path(rec: FileRecord) -> Path: + return rec.root / rec.dest_folder / rec.new_filename_preview + + +def apply_batch(records: List[FileRecord], dry_run: bool = True, + log_path: Optional[Path] = None) -> List[dict]: + log_rows = [] + + for rec in records: + if rec.status != "confirmed": + continue + + new_path = plan_new_path(rec) + row = { + "timestamp": datetime.now().isoformat(timespec="seconds"), + "old_path": str(rec.path), + "new_path": str(new_path), + "dry_run": dry_run, + "result": "", + } + + if new_path == rec.path: + row["result"] = "unchanged" + elif new_path.exists(): + row["result"] = "SKIPPED - target already exists" + elif dry_run: + row["result"] = "would move" + else: + new_path.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(rec.path), str(new_path)) + rec.path = new_path # the record now represents the file at its NEW location — + row["result"] = "moved" # without this, anything using rec.path afterwards + rec.status = "done" # (identity, re-display, a second move) points at nothing. + + log_rows.append(row) + + if log_path and log_rows: + with open(log_path, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=list(log_rows[0].keys())) + writer.writeheader() + writer.writerows(log_rows) + + return log_rows diff --git a/core/identity.py b/core/identity.py new file mode 100644 index 0000000..c3712a6 --- /dev/null +++ b/core/identity.py @@ -0,0 +1,43 @@ +""" +A file's path changes the moment you rename or move it — which means path +can't be used as the key for "have I already handled this file?" across +sessions. A file processed today and reopened next week needs to be +recognised at its NEW location, not just its old one. + +Identity here is (size, partial content hash) — cheap even for large +files, since only the first chunk is read, not the whole thing. +""" + +import hashlib +from pathlib import Path + +_CHUNK_SIZE = 65536 + + +def file_identity(path: Path) -> str: + size = path.stat().st_size + h = hashlib.sha1() + try: + with open(path, "rb") as f: + h.update(f.read(_CHUNK_SIZE)) + except OSError: + pass # fall back to size-only identity if the file can't be read + return f"{size}:{h.hexdigest()}" + + +def find_duplicate_identities(records) -> dict: + """Groups records that share an identity — genuinely byte-identical + files (same size, same content), not files that merely look similar. + + This is a real, unavoidable limitation: content-based identity cannot + tell two identical files apart, because there is nothing to tell apart + — their content really is the same. Progress-tracking (which file was + already handled) can silently apply to the wrong twin when this + happens. Callers should surface this to the person rather than hide + it; the real fix is deduplicating genuinely identical files as a + separate pass, not a cleverer identity function. + """ + groups: dict = {} + for rec in records: + groups.setdefault(rec.identity, []).append(rec) + return {identity: recs for identity, recs in groups.items() if len(recs) > 1} diff --git a/core/llm_suggest.py b/core/llm_suggest.py new file mode 100644 index 0000000..e7c473a --- /dev/null +++ b/core/llm_suggest.py @@ -0,0 +1,109 @@ +""" +Asks a local llama-swap model to suggest a topic, subtopic, and +destination folder for a file, given its filename, current folder, and +(if available) an extracted content excerpt. + +Talks to llama-swap's OpenAI-compatible endpoint over plain HTTP using +only the stdlib — nothing about a file's name or content ever leaves the +machine this endpoint points at. +""" + +import json +import os +import urllib.error +import urllib.request + +DEFAULT_ENDPOINT = os.environ.get("VAULT_RENAMER_LLM_ENDPOINT", "http://localhost:8080/v1/chat/completions") +DEFAULT_MODEL = os.environ.get("VAULT_RENAMER_LLM_MODEL", "daily") +TIMEOUT_SECONDS = 30 + +PROMPT_TEMPLATE = """You are helping organise a personal file vault. Given a file's name, its current folder, an optional content excerpt, and the list of real destination folders that already exist, suggest a short topic, a short subtopic, and which existing folder it belongs in. + +Rules: +- topic and subtopic: a few lowercase words each, no punctuation, describing what the file is actually about — not a restatement of the filename. +- dest_folder: MUST be exactly one of the folders listed below, copied verbatim. If genuinely unsure, pick the closest match rather than inventing a new folder. +- Reply with ONLY a JSON object, nothing else, no markdown fence: {{"topic": "...", "subtopic": "...", "dest_folder": "..."}} + +Current filename: {filename} +Current folder: {current_folder} +Content excerpt (may be empty): {excerpt} + +Real destination folders: +{folders} +""" + + +class LLMSuggestError(Exception): + """Callers should catch this and treat it as 'no suggestion available' + for this file — never let a model/network hiccup crash the batch.""" + + +def suggest(filename: str, current_folder: str, excerpt: str, folders: list, + endpoint: str = DEFAULT_ENDPOINT, model: str = DEFAULT_MODEL) -> dict: + prompt = PROMPT_TEMPLATE.format( + filename=filename, + current_folder=current_folder or "(root)", + excerpt=(excerpt or "")[:1500], + folders="\n".join(folders), + ) + + payload = { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "temperature": 0.2, + "max_tokens": 200, + } + + req = urllib.request.Request( + endpoint, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + + try: + with urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS) as resp: + data = json.loads(resp.read().decode("utf-8")) + except urllib.error.URLError as e: + raise LLMSuggestError(f"Couldn't reach {endpoint} — is llama-swap running? ({e})") + except TimeoutError: + raise LLMSuggestError(f"No response within {TIMEOUT_SECONDS}s — model may still be loading.") + + try: + content = data["choices"][0]["message"]["content"] + except (KeyError, IndexError): + raise LLMSuggestError(f"Unexpected response shape from {endpoint}: {data}") + + parsed = _extract_json(content) + if parsed is None: + raise LLMSuggestError(f"Model didn't reply with valid JSON: {content[:200]!r}") + + for key in ("topic", "subtopic", "dest_folder"): + parsed.setdefault(key, "") + + if parsed["dest_folder"] not in folders: + # The model invented a folder that doesn't exist — never apply a + # destination that isn't real. + parsed["dest_folder"] = "" + + return parsed + + +def _extract_json(text: str): + text = text.strip() + if text.startswith("```"): + text = text.strip("`") + if text.lower().startswith("json"): + text = text[4:] + text = text.strip() + try: + return json.loads(text) + except json.JSONDecodeError: + pass + start, end = text.find("{"), text.rfind("}") + if start != -1 and end != -1 and end > start: + try: + return json.loads(text[start:end + 1]) + except json.JSONDecodeError: + return None + return None diff --git a/core/model.py b/core/model.py new file mode 100644 index 0000000..6c8c3b2 --- /dev/null +++ b/core/model.py @@ -0,0 +1,72 @@ +""" +FileRecord: one row per file under review. Both the GUI and the batch/ +spreadsheet front end read and write the same fields, so nothing about +a file's state is front-end-specific. + +Every editable field defaults to "no change" (current filename, current +folder) when a record is first created — so a file you never touch in +review simply stays exactly as it is. Nothing is lost by omission. +""" + +from dataclasses import dataclass, field +from pathlib import Path +from typing import List + + +@dataclass +class FileRecord: + path: Path # current absolute path on disk + root: Path # vault root this file was scanned under + filename: str = "" # current filename, including extension + ext: str = "" + fs_date: str = "" # best-guess filesystem date — a starting + # point only, never treated as authoritative + name_matches_convention: bool = False + near_miss: bool = False # matched a "close but not quite" date format + + # --- Editable fields --- + date_override: str = "" # what actually goes in the filename + topic: str = "" + subtopic: str = "" + extra: str = "" + more: str = "" + version: str = "" + dest_folder: str = "" # relative to root + keep_filename: bool = False # if True, move only — never rebuild the name. + # Defaults to True for files that already match + # the convention, so opening one to move it + # doesn't blank out a name that was already fine. + + suggested_folders: List[str] = field(default_factory=list) + + # pending -> not yet reviewed | confirmed -> ready for apply_batch + # skip -> deliberately left alone | done -> already applied + status: str = "pending" + + @property + def identity(self) -> str: + """Stable across rename/move — see core.identity. Use this, not + rel_path, as the key for anything that needs to survive a file + actually being moved (e.g. resuming progress across sessions).""" + from .identity import file_identity + return file_identity(self.path) + + @property + def rel_path(self) -> str: + return str(self.path.relative_to(self.root)) + + @property + def current_folder(self) -> str: + return str(self.path.parent.relative_to(self.root)) + + @property + def new_filename_preview(self) -> str: + if self.keep_filename: + return self.filename + from .sanitiser import build_filename + return build_filename( + date=self.date_override or self.fs_date, + topic=self.topic, subtopic=self.subtopic, + extra=self.extra, more=self.more, version=self.version, + ext=self.ext, + ) diff --git a/core/sanitiser.py b/core/sanitiser.py new file mode 100644 index 0000000..cfc9e2f --- /dev/null +++ b/core/sanitiser.py @@ -0,0 +1,39 @@ +""" +Filename token sanitiser — ported line-for-line in behaviour from your +existing Dokumenten_Beschrifter.html tool, so a name built here and a name +built in the browser tool always match. +""" + +import re + +UMLAUT_MAP = { + "ä": "ae", "ö": "oe", "ü": "ue", "ß": "ss", +} + + +def sanitise(text: str) -> str: + """Lowercase, fold umlauts, spaces -> hyphens, strip anything that + isn't a-z/0-9/hyphen, collapse repeats, trim leading/trailing hyphens.""" + if not text: + return "" + result = text.strip().lower() + for src, repl in UMLAUT_MAP.items(): + result = result.replace(src, repl) + result = re.sub(r"\s+", "-", result) + result = re.sub(r"[^a-z0-9\-]", "", result) + result = re.sub(r"-+", "-", result) + return result.strip("-") + + +def build_filename(date: str, topic: str = "", subtopic: str = "", + extra: str = "", more: str = "", version: str = "", + ext: str = "") -> str: + """Assemble the final filename from parts. Empty fields are skipped, + so a file only needs as many fields filled in as makes sense for it — + same rule as the HTML tool ("fill in as many or as few boxes as needed").""" + parts = [sanitise(p) for p in (topic, subtopic, extra, more)] + parts = [p for p in parts if p] + if version and sanitise(version): + parts.append(sanitise(version)) + stem = "_".join([date] + parts) if parts else date + return f"{stem}{ext}" diff --git a/core/scanner.py b/core/scanner.py new file mode 100644 index 0000000..cfe1bf2 --- /dev/null +++ b/core/scanner.py @@ -0,0 +1,89 @@ +""" +Walks a vault root and builds FileRecord objects. + +Two jobs: filter out junk (config.IGNORE_*), and make a best-guess starting +date for each real file. That date is explicitly NOT trusted as correct — +you said the metadata is often wrong, so this only has to be a reasonable +starting point that you can freely overwrite, never a claim of truth. +""" + +from datetime import datetime +from pathlib import Path +from typing import List + +from . import config +from .model import FileRecord + + +def _is_ignored_file(name: str) -> bool: + if name in config.TOOL_STATE_FILENAMES: + return True + return any(p.match(name) for p in config.IGNORE_FILE_PATTERNS) + + +def _is_inside_ignored_dir(rel_parts) -> bool: + return any( + part in config.IGNORE_DIR_NAMES or part.startswith(".") + for part in rel_parts[:-1] # everything except the filename itself + ) + + +def _best_guess_date(stat) -> str: + """True creation time isn't reliably available on ext4 via stat() — + this is a heuristic (earliest of mtime/ctime, or real birthtime if the + platform exposes it), not an authoritative answer.""" + ts = getattr(stat, "st_birthtime", None) + if ts is None: + ts = min(stat.st_mtime, stat.st_ctime) + return datetime.fromtimestamp(ts).strftime("%Y-%m-%d") + + +def discover_folders(root: Path) -> List[str]: + """Every real directory under root, as paths relative to root. This is + the live, always-current list of possible destinations — it updates + automatically as your folder structure changes, nothing hardcoded.""" + folders = [] + for p in root.rglob("*"): + if p.is_dir() and p.name not in config.IGNORE_DIR_NAMES and not p.name.startswith("."): + folders.append(str(p.relative_to(root))) + return sorted(folders) + + +def scan(root: Path) -> List[FileRecord]: + """Scan root recursively and return one FileRecord per real, non-junk file.""" + records = [] + for p in root.rglob("*"): + if p.is_dir(): + continue + + rel_parts = p.relative_to(root).parts + if _is_inside_ignored_dir(rel_parts): + continue + if _is_ignored_file(p.name): + continue + + stat = p.stat() + fs_date = _best_guess_date(stat) + name_ok = bool(config.ALREADY_CONVENTIONED.match(p.name)) + near = (not name_ok) and any( + pat.match(p.name) for pat in config.NEAR_CONVENTION_PATTERNS + ) + + rec = FileRecord( + path=p, + root=root, + filename=p.name, + ext=p.suffix, + fs_date=fs_date, + name_matches_convention=name_ok, + near_miss=near, + ) + # Defaults: no change unless you edit something. Files that already + # match the convention default to "keep name, move only" — see + # FileRecord.keep_filename for why. + rec.date_override = fs_date + rec.dest_folder = rec.current_folder + rec.keep_filename = name_ok + records.append(rec) + + return records diff --git a/core/suggest.py b/core/suggest.py new file mode 100644 index 0000000..4d5d1b0 --- /dev/null +++ b/core/suggest.py @@ -0,0 +1,50 @@ +""" +Folder suggestion: plain token overlap between the text you've typed +(filename, topic, subtopic — whatever's available) and each candidate +destination folder's path segments. + +Deliberately simple and inspectable rather than clever — no ML, nothing +that runs over the network, and you can always see exactly why a folder +was suggested (shared words) rather than trusting a black box. If the +suggestions turn out too weak once you try it on real files, the obvious +next upgrade is peeking at file content for .md/.txt files — but that's +worth trying this simple version against real data first. +""" + +import re +from pathlib import Path +from typing import List + +from .sanitiser import UMLAUT_MAP + + +def _tokens(text: str) -> set: + """Split on ANY run of non-alphanumeric characters (underscore, hyphen, + space, ...) — folder names use underscores (04_ARBEIT), typed text uses + spaces, so the tokeniser has to treat both as word boundaries. This is + deliberately more lenient than sanitise(), which is for building the + final filename, not for comparing words.""" + folded = text.lower() + for src, repl in UMLAUT_MAP.items(): + folded = folded.replace(src, repl) + return {t for t in re.split(r"[^a-z0-9]+", folded) if len(t) > 2} + + +def suggest_folders(query_text: str, all_folders: List[str], top_n: int = 5) -> List[str]: + """Rank candidate folders by shared tokens with query_text. + Returns [] if query_text has no usable tokens — never guesses blind.""" + query_tokens = _tokens(query_text) + if not query_tokens: + return [] + + scored = [] + for folder in all_folders: + folder_tokens = set() + for segment in Path(folder).parts: + folder_tokens |= _tokens(segment) + overlap = len(query_tokens & folder_tokens) + if overlap: + scored.append((overlap, folder)) + + scored.sort(key=lambda pair: (-pair[0], pair[1])) + return [folder for _, folder in scored[:top_n]] diff --git a/gui/app.py b/gui/app.py new file mode 100644 index 0000000..799642a --- /dev/null +++ b/gui/app.py @@ -0,0 +1,368 @@ +""" +Tkinter front end for the vault renamer. + +Behaviour worth knowing before using this: + + - Confirming a file moves/renames it IMMEDIATELY (not queued for later). + You've already seen the exact new name and destination in the preview + before clicking Confirm, so this behaves like doing the rename + yourself by hand, just faster. + - Every move is appended to a running CSV log next to the folder you + scanned, so there's a record of everything even if the app or your + session gets interrupted partway through. + - Progress is saved automatically to a small JSON file in the scanned + folder. Closing the app mid-batch and reopening later picks up + exactly where you left off — reviewed files won't be shown again. + - "Skip" marks a file as deliberately left alone, distinct from just + not having gotten to it yet. Tick "show skipped too" to bring them + back into view. + - "Keep current filename" is pre-ticked for files that already match + the naming convention, so opening one just to move it doesn't blank + out a name that was already correct. Untick it if you actually want + to rename that particular file too. +""" + +import csv +import json +import sys +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.model import FileRecord +from core.config import STATE_FILENAME, LOG_FILENAME + +EDITABLE_FIELDS = ["date_override", "topic", "subtopic", "extra", "more", "version", "dest_folder"] +FIELD_LABELS = { + "date_override": "Date (YYYY-MM-DD)", + "topic": "Main topic", + "subtopic": "Subtopic", + "extra": "Extra", + "more": "More explanation", + "version": "Version (e.g. v1)", +} + + +class VaultRenamerApp: + def __init__(self, win: tk.Tk, vault_root: Path): + self.win = win + self.vault_root = vault_root + self.win.title(f"Vault Renamer — {vault_root}") + self.win.geometry("1150x680") + + self.records: list[FileRecord] = [] + self.folders: list[str] = [] + self.current: FileRecord | None = None + self.show_skipped = tk.BooleanVar(value=False) + self._suppress_trace = False + + self._build_ui() + self._load_and_scan() + + # ---------- persistence ---------- + 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 _load_and_scan(self): + self.records = scan(self.vault_root) + self.folders = discover_folders(self.vault_root) + + saved = self._load_state() + if saved: + 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.", + ) + + self._refresh_list() + + # ---------- UI construction ---------- + def _build_ui(self): + left = ttk.Frame(self.win, padding=8) + left.pack(side="left", fill="y") + + ttk.Label(left, text="Pending files", font=("", 11, "bold")).pack(anchor="w") + self.progress_label = ttk.Label(left, text="") + self.progress_label.pack(anchor="w", pady=(0, 6)) + + self.listbox = tk.Listbox(left, width=50, height=34, exportselection=False) + self.listbox.pack(fill="y", expand=True) + self.listbox.bind("<>", self._on_select) + + ttk.Checkbutton( + left, text="Show skipped too", variable=self.show_skipped, + command=self._refresh_list, + ).pack(anchor="w", pady=(6, 0)) + + right = ttk.Frame(self.win, padding=12) + right.pack(side="left", fill="both", expand=True) + + self.path_label = ttk.Label(right, text="", wraplength=650, font=("", 10, "bold")) + self.path_label.pack(anchor="w") + + self.flag_label = ttk.Label(right, text="", foreground="#555", wraplength=650) + self.flag_label.pack(anchor="w", pady=(0, 10)) + + form = ttk.Frame(right) + form.pack(anchor="w", fill="x") + + self.vars = {key: tk.StringVar() for key in EDITABLE_FIELDS} + self.keep_filename_var = tk.BooleanVar() + + row = 0 + for key in ["date_override", "topic", "subtopic", "extra", "more", "version"]: + ttk.Label(form, text=FIELD_LABELS[key]).grid(row=row, column=0, sticky="w", pady=3) + entry = ttk.Entry(form, textvariable=self.vars[key], width=48) + entry.grid(row=row, column=1, sticky="w", pady=3, padx=(8, 0)) + self.vars[key].trace_add("write", self._on_field_edit) + row += 1 + + ttk.Checkbutton( + form, text="Keep current filename (only move — don't rename)", + variable=self.keep_filename_var, command=self._update_preview, + ).grid(row=row, column=0, columnspan=2, sticky="w", pady=(6, 3)) + row += 1 + + ttk.Label(form, text="Destination folder").grid(row=row, column=0, sticky="w", pady=3) + self.dest_combo = ttk.Combobox(form, textvariable=self.vars["dest_folder"], width=45) + self.dest_combo.grid(row=row, column=1, sticky="w", pady=3, padx=(8, 0)) + self.vars["dest_folder"].trace_add("write", lambda *_: self._update_preview()) + row += 1 + + ttk.Button(right, text="Suggest folders", command=self._show_suggestions).pack(anchor="w", pady=(8, 0)) + self.suggestion_frame = ttk.Frame(right) + self.suggestion_frame.pack(anchor="w", pady=(4, 10)) + + ttk.Separator(right).pack(fill="x", pady=10) + + ttk.Label(right, text="Preview:", font=("", 10, "bold")).pack(anchor="w") + self.preview_label = ttk.Label(right, text="", foreground="#0a6", wraplength=650) + self.preview_label.pack(anchor="w", pady=(0, 14)) + + btns = ttk.Frame(right) + btns.pack(anchor="w") + ttk.Button(btns, text="Confirm && Move", command=self._confirm_move).pack(side="left", padx=(0, 8)) + ttk.Button(btns, text="Skip (leave alone)", command=self._skip).pack(side="left", padx=(0, 8)) + ttk.Button(btns, text="Save progress", command=self._save_state).pack(side="left") + + # ---------- list handling ---------- + def _pending_records(self): + if self.show_skipped.get(): + return [r for r in self.records if r.status in ("pending", "skip")] + return [r for r in self.records if r.status == "pending"] + + def _refresh_list(self): + self.listbox.delete(0, tk.END) + for rec in self._pending_records(): + flag = "OK " if rec.name_matches_convention else ("~ " if rec.near_miss else " ") + skip_mark = " [skipped]" if rec.status == "skip" else "" + self.listbox.insert(tk.END, f"[{flag}] {rec.rel_path}{skip_mark}") + self._update_progress_label() + + def _update_progress_label(self): + total = len(self.records) + done = sum(1 for r in self.records if r.status == "done") + skipped = sum(1 for r in self.records if r.status == "skip") + pending = total - done - skipped + self.progress_label.config(text=f"{pending} pending / {done} done / {skipped} skipped / {total} total") + + def _on_select(self, _event): + sel = self.listbox.curselection() + if not sel: + return + rec = self._pending_records()[sel[0]] + self.current = rec + self._load_record_into_form(rec) + + def _load_record_into_form(self, rec: FileRecord): + self._suppress_trace = True + self.path_label.config(text=rec.rel_path) + + flags = [] + if rec.name_matches_convention: + flags.append("already matches naming convention") + if rec.near_miss: + flags.append("near-miss date format") + flags.append(f"filesystem date guess: {rec.fs_date} (starting point only — often wrong)") + self.flag_label.config(text=" · ".join(flags)) + + self.vars["date_override"].set(rec.date_override or rec.fs_date) + self.vars["topic"].set(rec.topic) + self.vars["subtopic"].set(rec.subtopic) + self.vars["extra"].set(rec.extra) + self.vars["more"].set(rec.more) + self.vars["version"].set(rec.version) + self.vars["dest_folder"].set(rec.dest_folder) + self.keep_filename_var.set(rec.keep_filename) + self.dest_combo["values"] = self.folders + + for child in self.suggestion_frame.winfo_children(): + child.destroy() + + self._suppress_trace = False + self._update_preview() + + # ---------- live preview / suggestions ---------- + def _on_field_edit(self, *_args): + # Typing into a name field while "keep current filename" is ticked + # means you've changed your mind about that file — untick it + # automatically rather than silently ignoring what was typed. + if not self._suppress_trace and self.keep_filename_var.get(): + self.keep_filename_var.set(False) + self._update_preview() + + def _sync_form_to_record(self): + rec = self.current + if not rec: + return + for key in EDITABLE_FIELDS: + setattr(rec, key, self.vars[key].get().strip()) + rec.keep_filename = self.keep_filename_var.get() + + def _update_preview(self): + if not self.current or self._suppress_trace: + return + self._sync_form_to_record() + rec = self.current + new_name = rec.new_filename_preview + new_path = Path(rec.dest_folder) / new_name if rec.dest_folder else Path(new_name) + self.preview_label.config(text=f"{new_name}\n→ {new_path}") + + def _show_suggestions(self): + if not self.current: + return + query = " ".join([self.vars["topic"].get(), self.vars["subtopic"].get(), self.current.filename]) + suggestions = suggest_folders(query, self.folders, top_n=5) + + for child in self.suggestion_frame.winfo_children(): + child.destroy() + + if not suggestions: + ttk.Label(self.suggestion_frame, text="No keyword matches — pick from the dropdown instead.", + foreground="#888").pack(anchor="w") + return + + ttk.Label(self.suggestion_frame, text="Suggestions (click to use):").pack(anchor="w") + for folder in suggestions: + ttk.Button(self.suggestion_frame, text=folder, + command=lambda f=folder: self.vars["dest_folder"].set(f)).pack(anchor="w") + + # ---------- actions ---------- + def _confirm_move(self): + rec = self.current + if not rec: + return + self._sync_form_to_record() + rec.status = "confirmed" + + log_rows = apply_batch([rec], dry_run=False, log_path=None) + self._append_log(log_rows) + + result = log_rows[0]["result"] if log_rows else "no-op" + if result.startswith("SKIPPED"): + messagebox.showwarning("Not moved", f"{result}\n\n{log_rows[0]['new_path']}") + rec.status = "pending" + return + + self._save_state() + self._refresh_list() + self._select_next() + + def _skip(self): + rec = self.current + if not rec: + return + rec.status = "skip" + self._save_state() + self._refresh_list() + self._select_next() + + def _select_next(self): + pending = self._pending_records() + if pending: + self.listbox.selection_set(0) + self._on_select(None) + else: + self.current = None + self.path_label.config(text="") + self.flag_label.config(text="") + self.preview_label.config(text="") + messagebox.showinfo("Done", "No pending files left in this view.") + + 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 main(): + if len(sys.argv) != 2: + print("Usage: python3 gui/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() + VaultRenamerApp(win, vault_root) + win.mainloop() + + +if __name__ == "__main__": + main() diff --git a/gui/batch_app.py b/gui/batch_app.py new file mode 100644 index 0000000..f54093c --- /dev/null +++ b/gui/batch_app.py @@ -0,0 +1,508 @@ +""" +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("", self._on_double_click) + self.tree.bind("", 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("", commit) + editor.bind("", commit) + editor.bind("", 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("", 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() diff --git a/preview_scan.py b/preview_scan.py new file mode 100644 index 0000000..bdc97dc --- /dev/null +++ b/preview_scan.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +""" +Run this first, before the GUI or batch tool exist. + +It only reads your filesystem - it never renames or moves anything. +Point it at a real folder (start small, e.g. just 03_PRIVAT, not the +whole 711-file vault) and check: + - does the file count look right (junk correctly excluded)? + - does the date guess look reasonable as a STARTING point? + - does the "already conventioned" count match what you'd expect? + +Usage: + python3 preview_scan.py /path/to/folder +""" + +import sys +from pathlib import Path + +from core import scan, discover_folders + + +def main(): + if len(sys.argv) != 2: + print("Usage: python3 preview_scan.py /path/to/folder") + sys.exit(1) + + root = Path(sys.argv[1]).expanduser().resolve() + if not root.is_dir(): + print(f"Not a directory: {root}") + sys.exit(1) + + records = scan(root) + folders = discover_folders(root) + + already = sum(r.name_matches_convention for r in records) + near = sum(r.near_miss for r in records) + untouched = len(records) - already - near + + print(f"Scanned: {root}") + print(f"Files found (junk excluded): {len(records)}") + print(f" already conventioned: {already}") + print(f" near-miss date format: {near}") + print(f" untouched: {untouched}") + print(f"Destination folders found: {len(folders)}") + print() + print("First 20 files:") + for r in records[:20]: + flag = "OK " if r.name_matches_convention else ("~ " if r.near_miss else " ") + print(f" [{flag}] {r.fs_date} {r.rel_path}") + + if len(records) > 20: + print(f" ... and {len(records) - 20} more") + + +if __name__ == "__main__": + main()