#!/usr/bin/env python3 """ Quick Sorter v2.0 Purpose: fast human-reviewed file sorting/renaming with an optional local LLM. One screen, one proposal per file, no spreadsheet round-trip, no per-file Q&A. V2 adds a first-class top-level 01_INBOX workflow: source files can be routed to sibling roots such as 03_PRIVAT, 04_ARBEIT and 05_VAULT. Archives are identified before document analysis and left for the deliberate unpack/repack workflow. Safety: - Scan and LLM analysis are read-only. - Nothing moves until a row is explicitly confirmed. - Preview is always available before Apply. - Existing targets are never overwritten. - Exact target collisions are full-SHA-256 compared: identical sources are removed only on Apply; different content is preserved as a visible _variant-N file. - Every real Apply is logged. """ from __future__ import annotations import argparse import csv import hashlib import shutil import json import os import re import threading import queue import urllib.request import urllib.error from dataclasses import dataclass, field from datetime import datetime, date from pathlib import Path from typing import Optional # ---------------- configuration ---------------- IGNORE_DIR_NAMES = { ".git", ".stfolder", ".stversions", "__pycache__", "node_modules", ".venv", "venv", ".idea", ".vscode", } IGNORE_FILE_PATTERNS = [ re.compile(r"^\.sync-conflict-.*"), re.compile(r"^Thumbs\.db$", re.I), re.compile(r"^\.~lock\..*#$"), re.compile(r"^~\$.*"), re.compile(r"^\.directory$"), re.compile(r"^desktop\.ini$", re.I), re.compile(r"^\.DS_Store$"), ] STANDARD_PREFIX = re.compile(r"^\d{4}-\d{2}-\d{2}_.+") UMLAUT_MAP = {"ä": "ae", "ö": "oe", "ü": "ue", "ß": "ss"} DEFAULT_ENDPOINT = os.getenv( "QUICK_SORTER_LLM_ENDPOINT", "http://localhost:8080/v1/chat/completions", ) DEFAULT_MODEL = os.getenv("QUICK_SORTER_LLM_MODEL", "daily") LOG_FILENAME = ".quick_sorter_log.csv" MAX_EXCERPT = 2400 ARCHIVE_SUFFIXES = ( ".zip", ".7z", ".rar", ".tar", ".tar.gz", ".tgz", ".tar.xz", ".txz", ".tar.bz2", ".tbz2", ".gz", ".bz2", ".xz", ) # ---------------- data model ---------------- @dataclass class Record: path: Path root: Path target_root: Path rel_path: str filename: str ext: str current_folder: str date_guess: str date_source: str proposed_name: str destination: str confidence: float = 0.0 reason: str = "" status: str = "pending" # pending/suggested/review/confirmed/skip/done candidates: list[str] = field(default_factory=list) excerpt: str = "" content_status: str = "not-read" topic: str = "" subtopic: str = "" document_type: str = "" detail: str = "" version: str = "" collision: str = "" # exact duplicate / different-content variant handling duplicate_of: str = "" # full-SHA-256 match found by the global pre-pass @property def target(self) -> Path: return self.target_root / self.destination / self.proposed_name def is_archive_file(path: Path) -> bool: """Recognise common archive/container formats before LLM analysis.""" name = path.name.lower() return any(name.endswith(suffix) for suffix in ARCHIVE_SUFFIXES) # ---------------- date logic ---------------- def _valid(y: int, m: int, d: int) -> Optional[str]: try: return date(y, m, d).isoformat() except ValueError: return None def date_from_filename(name: str) -> Optional[str]: """Conservative parser: known filename dates beat filesystem metadata.""" stem = Path(name).stem # ISO date already present m = re.search(r"(? first day of range m = re.search(r"(? 20YY-MM-DD m = re.search(r"(? Optional[str]: """Optional embedded metadata; never required.""" ext = path.suffix.lower() try: if ext == ".docx": import docx created = docx.Document(str(path)).core_properties.created return created.date().isoformat() if created else None if ext == ".xlsx": import openpyxl wb = openpyxl.load_workbook(str(path), read_only=True, data_only=True) created = wb.properties.created return created.date().isoformat() if created else None if ext == ".pdf": from pypdf import PdfReader created = PdfReader(str(path)).metadata.creation_date return created.date().isoformat() if created else None if ext in {".jpg", ".jpeg", ".tif", ".tiff"}: from PIL import Image, ExifTags image = Image.open(path) exif = image.getexif() tag_id = next((k for k, v in ExifTags.TAGS.items() if v == "DateTimeOriginal"), None) raw = exif.get(tag_id) if tag_id else None if raw: return datetime.strptime(str(raw), "%Y:%m:%d %H:%M:%S").date().isoformat() except Exception: return None return None def date_from_parent_folders(path: Path) -> Optional[str]: """Use a parent folder date only when it is explicitly parseable.""" for part in reversed(path.parent.parts): parsed = date_from_filename(part) if parsed: return parsed return None def best_date(path: Path) -> tuple[str, str]: # Established priority for this archive: filename -> trustworthy date-folder -> mtime. from_name = date_from_filename(path.name) if from_name: return from_name, "filename" from_folder = date_from_parent_folders(path) if from_folder: return from_folder, "folder" return datetime.fromtimestamp(path.stat().st_mtime).date().isoformat(), "mtime" # ---------------- content + token helpers ---------------- def sanitise_token(text: str) -> str: 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 tokens(text: str) -> set[str]: 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} # Generic document-type words are useful as a warning sign, but poor context. # If a filename is mostly made of these words, the model must use the excerpt # to recover the actual project/event/subject before proposing a new name. GENERIC_NAME_WORDS = { "ablaufplan", "plan", "planung", "protokoll", "handout", "praesentation", "präsentation", "rechnung", "abrechnung", "liste", "teilnahmeliste", "feedback", "infos", "info", "programm", "entwurf", "vorlage", "dokument", "scan", "schicht", "treffen", "teamtreffen", "aufgabenliste", "todo", } def is_generic_filename(name: str) -> bool: stem = Path(name).stem # Ignore date/version fragments; judge the semantic remainder. words = [w for w in re.split(r"[^A-Za-zÄÖÜäöüß0-9]+", stem.lower()) if w] words = [w for w in words if not re.fullmatch(r"(?:20)?\d{2,8}|v\d+", w)] if not words: return True generic = sum(1 for w in words if w in GENERIC_NAME_WORDS) # One lone generic title ("Ablaufplan") or a name dominated by generic # document-type words needs content-derived context. return len(words) <= 2 and generic >= 1 or (generic / len(words) >= 0.60) def _clip(parts: list[str]) -> str: text = "\n".join(x.strip() for x in parts if x and x.strip()) return text[:MAX_EXCERPT] def extract_excerpt(path: Path) -> tuple[str, str]: """Return (text, status). Status is deliberately visible in the UI so missing optional libraries or text-less/scanned documents never fail silently. """ ext = path.suffix.lower() try: if ext in {".md", ".txt", ".csv", ".html", ".htm"}: text = path.read_text(encoding="utf-8", errors="ignore")[:MAX_EXCERPT] return text, f"text:{len(text)}" if ext == ".docx": try: import docx except ImportError: return "", "missing:python-docx" doc = docx.Document(str(path)) parts = [] props = doc.core_properties for label, value in ( ("title", props.title), ("subject", props.subject), ("keywords", props.keywords), ("comments", props.comments), ): if value: parts.append(f"{label}: {value}") # Normal paragraphs parts.extend(p.text for p in doc.paragraphs if p.text.strip()) # Tables are crucial for schedules, plans and forms. for table in doc.tables: for row in table.rows: cells = [c.text.strip() for c in row.cells if c.text.strip()] if cells: parts.append(" | ".join(cells)) if sum(len(x) for x in parts) >= MAX_EXCERPT: break if sum(len(x) for x in parts) >= MAX_EXCERPT: break # Headers/footers can carry event/project names. if sum(len(x) for x in parts) < MAX_EXCERPT: for section in doc.sections: parts.extend(p.text for p in section.header.paragraphs if p.text.strip()) parts.extend(p.text for p in section.footer.paragraphs if p.text.strip()) text = _clip(parts) return text, f"docx:{len(text)}" if text else "docx:no-text" if ext == ".pdf": try: from pypdf import PdfReader except ImportError: return "", "missing:pypdf" reader = PdfReader(str(path)) parts = [] meta = reader.metadata if meta: for key in ("/Title", "/Subject", "/Keywords"): value = meta.get(key) if value: parts.append(f"{key[1:].lower()}: {value}") for page in reader.pages[:3]: parts.append(page.extract_text() or "") if sum(len(x) for x in parts) >= MAX_EXCERPT: break text = _clip(parts) return text, f"pdf:{len(text)}" if text else "pdf:no-text-or-scan" if ext == ".xlsx": try: import openpyxl except ImportError: return "", "missing:openpyxl" wb = openpyxl.load_workbook(str(path), read_only=True, data_only=True) parts = [] for ws in wb.worksheets[:2]: parts.append(f"sheet: {ws.title}") 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) if line.strip(): parts.append(line) if sum(len(x) for x in parts) >= MAX_EXCERPT: break if sum(len(x) for x in parts) >= MAX_EXCERPT: break text = _clip(parts) return text, f"xlsx:{len(text)}" if text else "xlsx:no-text" if ext == ".pptx": try: from pptx import Presentation except ImportError: return "", "missing:python-pptx" prs = Presentation(str(path)) parts = [] for slide in prs.slides[:5]: for shape in slide.shapes: text = getattr(shape, "text", "") if text and text.strip(): parts.append(text) if sum(len(x) for x in parts) >= MAX_EXCERPT: break if sum(len(x) for x in parts) >= MAX_EXCERPT: break text = _clip(parts) return text, f"pptx:{len(text)}" if text else "pptx:no-text" if ext == ".odt": try: from odf import text as odf_text from odf.opendocument import load as odf_load from odf.teletype import extractText except ImportError: return "", "missing:odfpy" doc = odf_load(str(path)) parts = [extractText(n) for n in doc.getElementsByType(odf_text.P)] text = _clip(parts) return text, f"odt:{len(text)}" if text else "odt:no-text" return "", f"unsupported:{ext or 'none'}" except Exception as exc: return "", f"error:{type(exc).__name__}" def build_structured_filename(rec: Record) -> str: """Build the filename deterministically from model-derived semantic parts.""" parts = [] for value in (rec.topic, rec.subtopic, rec.document_type, rec.detail): token = sanitise_token(value) if token and token not in parts: parts.append(token) ver = sanitise_token(rec.version) if ver: if re.fullmatch(r"\d+", ver): ver = f"v{ver}" parts.append(ver) if not parts: return rec.filename return "_".join([rec.date_guess] + parts) + rec.ext # ---------------- policy + scan ---------------- def ignored(path: Path, root: Path) -> bool: rel = path.relative_to(root) if any(part in IGNORE_DIR_NAMES or part.startswith(".") for part in rel.parts[:-1]): return True return any(p.match(path.name) for p in IGNORE_FILE_PATTERNS) or path.name == LOG_FILENAME def load_policy(path: Optional[Path]) -> dict: if not path or not path.exists(): return {} return json.loads(path.read_text(encoding="utf-8")) def dynamic_destinations(root: Path) -> list[str]: values = [] for p in root.rglob("*"): if not p.is_dir() or p == root: continue rel = p.relative_to(root) if any(part in IGNORE_DIR_NAMES or part.startswith(".") for part in rel.parts): continue # Avoid presenting the entire historical tree as a flat menu. if len(rel.parts) <= 2 and "Altbestand" not in rel.parts and "01_INBOX" not in rel.parts: values.append(str(rel)) return sorted(set(values)) def allowed_destinations(root: Path, policy: dict, profile: str = "") -> list[str]: key = profile.strip() or root.name configured = policy.get(key) if configured is not None: return configured return dynamic_destinations(root) def _normalise_alias_text(text: str) -> str: folded = text.lower() for src, repl in UMLAUT_MAP.items(): folded = folded.replace(src, repl) return " ".join(t for t in re.split(r"[^a-z0-9]+", folded) if t) def _strong_route_branch(rec: Record, aliases: dict[str, list[str]]) -> str: """Return a deterministic branch only when aliases are unambiguous. A file can legitimately mention more than one work area (e.g. a Cafe Bach refrigerator accounting sheet for a CSD occasion). In that situation, hard-routing would destroy useful context. We therefore narrow the LLM only when exactly ONE branch is matched; otherwise candidate scoring + the LLM decide the primary owner/context. """ if not aliases: return "" filename_text = " " + _normalise_alias_text(rec.filename) + " " folder_text = " " + _normalise_alias_text(rec.current_folder) + " " excerpt_text = " " + _normalise_alias_text(rec.excerpt[:700]) + " " matched_branches = set() for branch, terms in aliases.items(): for term in terms: needle = _normalise_alias_text(term) if not needle: continue token = f" {needle} " if token in filename_text or token in folder_text or token in excerpt_text: matched_branches.add(branch) break if len(matched_branches) == 1: return next(iter(matched_branches)) return "" def choose_candidates(rec: Record, destinations: list[str], limit: int = 14, aliases: Optional[dict[str, list[str]]] = None, allow_stay_outside_policy: bool = False) -> list[str]: if not destinations: return [rec.current_folder] if rec.current_folder else [""] # User-defined strong aliases route obvious domain shorthand before the LLM. branch = _strong_route_branch(rec, aliases or {}) if branch: routed = [d for d in destinations if d == branch or d.startswith(branch + "/")] if routed: destinations = routed text = f"{rec.filename} {rec.current_folder} {rec.excerpt[:400]}" q = tokens(text) scored = [] current_top = rec.current_folder.split("/", 1)[0] if rec.current_folder else "" for folder in destinations: f_tokens = tokens(folder) score = len(q & f_tokens) * 3 if current_top and (folder == current_top or folder.startswith(current_top + "/")): score += 2 score -= max(0, folder.count("/") - 1) scored.append((score, folder)) scored.sort(key=lambda x: (-x[0], x[1])) # In Inbox mode destinations span several sibling roots. Always expose each # top-level area to the LLM, otherwise an unrelated alphabetical slice could # accidentally hide PRIVAT or VAULT from an ambiguous file. cross_root = any(re.match(r"^\d{2}_[^/]+(?:/|$)", d) for d in destinations) chosen = [] if cross_root and not branch: top_names = [] for d in destinations: top = d.split("/", 1)[0] if top not in top_names: top_names.append(top) for top in top_names: if top in destinations and top not in chosen: chosen.append(top) for _, folder in scored: if folder not in chosen: chosen.append(folder) if len(chosen) >= limit: break # For a real 01_INBOX we must always allow "leave it where it is" when the # model cannot know enough. This can be a nested unpack folder as well. if allow_stay_outside_policy and rec.current_folder and rec.current_folder not in chosen: chosen.append(rec.current_folder) return chosen[:limit + (1 if allow_stay_outside_policy else 0)] def scan(root: Path, target_root: Path, destinations: list[str], aliases: Optional[dict[str, list[str]]] = None, allow_stay_outside_policy: bool = False) -> list[Record]: records = [] for p in sorted(root.rglob("*")): if not p.is_file() or ignored(p, root): continue rel = str(p.relative_to(root)) try: current = str(p.parent.relative_to(target_root)) except ValueError: current = str(p.parent.relative_to(root)) if current == ".": current = "" d, source = best_date(p) archive = is_archive_file(p) rec = Record( path=p, root=root, target_root=target_root, rel_path=rel, filename=p.name, ext=p.suffix, current_folder=current, date_guess=d, date_source=source, proposed_name=p.name, destination=current, status="archive" if archive else "pending", content_status="archive" if archive else "not-read", reason=("Archiv: entpacken, Inhalt sortieren, bei Bedarf sauber neu packen" if archive else ""), ) rec.candidates = choose_candidates( rec, destinations, aliases=aliases, allow_stay_outside_policy=allow_stay_outside_policy, ) records.append(rec) return records # ---------------- LLM batching ---------------- SYSTEM_PROMPT = """You are a conservative file-sorting assistant. Your job is to save the human time, not create more decisions. Return one JSON ARRAY only, one object per input item, in the same order. Each object must contain exactly: id, destination, topic, subtopic, document_type, detail, version, confidence, reason Rules: - destination MUST be one of that item's candidates. Never invent a folder. - In 01_INBOX mode, choose the stable owner/domain first; use 99_ARCHIV only for clearly historical/reference material, not merely because a file is old. - If uncertain, keep the current folder when it is a candidate, lower confidence, and say why. - Do NOT write the final filename. Extract semantic naming fields; Python will build the filename consistently. - topic: stable broad context/owner, e.g. cafe-bach, checkpoint, csd, sp5. - subtopic: distinctive event/project/subject, e.g. bach-voices, golden-gays-versammlung, chemsex. - document_type: what the document is, e.g. ablaufplan, protokoll, rechnung, handout. - detail: optional useful distinguishing detail; empty string if unnecessary. - version: explicit version only if supported by filename/content; otherwise empty string. - Use only as many semantic parts as useful; never invent filler. - The filename is NOT the only source of truth. Use the excerpt to recover the actual project, event, organisation, or subject when clearly stated. - If generic_name=true (for example "Ablaufplan", "Protokoll", "Präsentation", "Rechnung"), DO NOT merely preserve/paraphrase the generic filename. Add the distinctive project/event/subject from the content when it is clear. - For destination, prefer the PRIMARY OWNER/WORK AREA of the document. A mentioned event/topic can be secondary context. Example principle: an inventory/accounting document owned by one work area does not switch branches merely because it mentions a campaign/event. - If several work-area aliases occur, do not assume the longest alias is primary; use filename structure + content semantics to decide and lower confidence if genuinely ambiguous. - Preserve a filename that is already clear AND context-specific unless there is a concrete improvement. - Use the supplied date_guess; do not invent a different date from vague content. - If date_source is mtime, treat the date as unverified fallback metadata. Prefer preserving an already-useful filename unless renaming clearly helps. - Correct obvious spelling/transliteration problems when safe. - confidence is 0.0 to 1.0. Use >=0.85 only for genuinely clear cases. - reason: max 12 words, factual and concise. """ def llm_batches(records: list[Record], endpoint: str, model: str, batch_size: int = 10): for start in range(0, len(records), batch_size): batch = records[start:start + batch_size] items = [] for idx, rec in enumerate(batch): items.append({ "id": idx, "filename": rec.filename, "current_folder": rec.current_folder, "date_guess": rec.date_guess, "date_source": rec.date_source, "generic_name": is_generic_filename(rec.filename), "content_status": rec.content_status, "candidates": rec.candidates, "excerpt": rec.excerpt[:MAX_EXCERPT], }) prompt = SYSTEM_PROMPT + "\nINPUT:\n" + json.dumps(items, ensure_ascii=False) payload = { "model": model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, "max_tokens": 1400, } 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=90) as resp: data = json.loads(resp.read().decode("utf-8")) text = data["choices"][0]["message"]["content"] parsed = extract_json_array(text) if not isinstance(parsed, list): raise ValueError("model did not return a JSON array") yield batch, parsed, None except Exception as exc: yield batch, None, str(exc) def extract_json_array(text: str): text = text.strip() if text.startswith("```"): text = re.sub(r"^```(?:json)?\s*", "", text, flags=re.I) text = re.sub(r"\s*```$", "", text) try: return json.loads(text) except json.JSONDecodeError: start, end = text.find("["), text.rfind("]") if start >= 0 and end > start: return json.loads(text[start:end + 1]) raise def apply_llm_results(batch: list[Record], results: list[dict]): by_id = {int(x.get("id", -1)): x for x in results if isinstance(x, dict)} for idx, rec in enumerate(batch): result = by_id.get(idx) if not result: rec.status = "review" rec.reason = "no model result" continue dest = str(result.get("destination", "")).strip() try: conf = float(result.get("confidence", 0)) except (TypeError, ValueError): conf = 0.0 if dest not in rec.candidates: dest = rec.current_folder conf = min(conf, 0.4) rec.destination = dest rec.topic = str(result.get("topic", "")).strip() rec.subtopic = str(result.get("subtopic", "")).strip() rec.document_type = str(result.get("document_type", "")).strip() rec.detail = str(result.get("detail", "")).strip() rec.version = str(result.get("version", "")).strip() name = build_structured_filename(rec) rec.proposed_name = name # If no usable content was extracted from a generic filename, the model # cannot reliably recover a hidden event/project name. Force human review. if is_generic_filename(rec.filename) and (not rec.excerpt.strip() or "no-text" in rec.content_status or rec.content_status.startswith("missing:")): conf = min(conf, 0.69) # mtime is only a fallback, not a verified creation date. If the model # builds a NEW filename from that date, never let the row enter the # one-click high-confidence confirmation group. The human can still # review and confirm it explicitly. if rec.date_source == "mtime" and name != rec.filename: conf = min(conf, 0.79) rec.confidence = max(0.0, min(1.0, conf)) rec.reason = str(result.get("reason", ""))[:120] rec.status = "suggested" if rec.confidence >= 0.70 else "review" # ---------------- planning + apply ---------------- _HASH_CACHE: dict[str, tuple[int, int, str]] = {} def sha256_file(path: Path) -> str: """Full-file SHA-256. Used only when two files compete for the same target. The cache is keyed by path + size + mtime_ns so routine re-planning in the GUI does not repeatedly hash unchanged files. """ st = path.stat() key = str(path) cached = _HASH_CACHE.get(key) sig = (st.st_size, st.st_mtime_ns) if cached and cached[:2] == sig: return cached[2] h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) digest = h.hexdigest() _HASH_CACHE[key] = (sig[0], sig[1], digest) return digest def same_content(a: Path, b: Path) -> bool: """True only for byte-identical files. Any read error is treated as different. This is deliberately conservative: we only auto-delete when equality can be proven. Files that merely *look* equivalent but differ in metadata stay kept. """ try: if a.stat().st_size != b.stat().st_size: return False return sha256_file(a) == sha256_file(b) except OSError: return False def global_duplicate_prepass(records: list[Record], source_root: Path, target_root: Path) -> dict[str, int | bool | str]: """Mark exact duplicates before content extraction or LLM analysis. The pre-pass is deliberately enabled only for *separate* source and target trees. This covers migration/staging imports without changing the established V2 Inbox behaviour where the Inbox is nested inside a broader target root. Full SHA-256 is used for every exact-duplicate decision. Existing target files are authoritative. If no target copy exists, the first source copy becomes the representative and later byte-identical source copies are marked duplicates. """ stats: dict[str, int | bool | str] = { "enabled": False, "target_files": 0, "source_files": 0, "target_duplicates": 0, "source_duplicates": 0, "hash_errors": 0, "reason": "", } source_root = source_root.resolve() target_root = target_root.resolve() if source_root == target_root or source_root.is_relative_to(target_root): stats["reason"] = "source is the target or nested inside it; preserving normal V2 behaviour" return stats if target_root.is_relative_to(source_root): stats["reason"] = "target is nested inside source; unsafe layout for global dedupe" return stats stats["enabled"] = True target_by_hash: dict[str, Path] = {} for p in sorted(target_root.rglob("*")): if p.is_symlink() or not p.is_file() or ignored(p, target_root): continue try: digest = sha256_file(p) except OSError: stats["hash_errors"] += 1 continue stats["target_files"] += 1 target_by_hash.setdefault(digest, p) source_seen: dict[str, Path] = {} for rec in records: # scan() has already applied the normal ignore rules. Archives are hashed # too: an archive that is already byte-identical in TARGET needs no LLM. try: digest = sha256_file(rec.path) except OSError: stats["hash_errors"] += 1 continue stats["source_files"] += 1 target_match = target_by_hash.get(digest) if target_match is not None: rec.duplicate_of = str(target_match) rec.status = "exact-duplicate" rec.collision = "EXACT → already in target" try: display = str(target_match.relative_to(target_root)) except ValueError: display = str(target_match) rec.reason = f"SHA-256 match in target: {display}" stats["target_duplicates"] += 1 continue first_source = source_seen.get(digest) if first_source is not None: rec.duplicate_of = str(first_source) rec.status = "exact-duplicate" rec.collision = "EXACT → duplicate incoming" try: display = str(first_source.relative_to(source_root)) except ValueError: display = str(first_source) rec.reason = f"SHA-256 match in source: {display}" stats["source_duplicates"] += 1 continue source_seen[digest] = rec.path return stats def variant_path(target: Path, n: int) -> Path: """Insert a visible variant marker before the full extension chain.""" suffix = "".join(target.suffixes) stem = target.name[:-len(suffix)] if suffix else target.name return target.with_name(f"{stem}_variant-{n}{suffix}") def _find_variant_or_duplicate(source: Path, target: Path, reserved: dict[str, Path]) -> tuple[Optional[Path], Optional[Path]]: """Find a safe variant target. Returns (free_variant_path, None), or (None, duplicate_of) if an already-existing variant is byte-identical to source. """ n = 2 while True: cand = variant_path(target, n) other = reserved.get(str(cand)) if other is not None: if same_content(source, other): return None, other n += 1 continue if cand.exists(): if cand.is_file() and same_content(source, cand): return None, cand n += 1 continue return cand, None def plan(records: list[Record]) -> list[dict]: """Plan confirmed rows, automatically resolving target collisions. Rules: - exact byte duplicate (full SHA-256) -> keep the already-present/planned copy, remove the redundant source on Apply; - same target but different bytes -> keep both, route the new one to *_variant-2, *_variant-3, ...; - never overwrite an existing file. """ rows = [] reserved: dict[str, Path] = {} # future/existing target path -> content source for rec in records: if rec.status != "confirmed": continue source = rec.path # A global pre-pass duplicate bypasses semantic destination/name planning. # Re-verify the bytes immediately before planning deletion; if the other # copy changed or vanished, fail closed and touch nothing. if rec.duplicate_of: duplicate_of = Path(rec.duplicate_of) if duplicate_of.exists() and same_content(source, duplicate_of): result = "EXACT duplicate -> would remove source" resolved = duplicate_of collision = "EXACT → remove duplicate" else: result = "SKIPPED - pre-pass duplicate no longer matches" resolved = duplicate_of collision = "RECHECK → duplicate changed" rec.collision = collision rows.append({ "old_path": str(source), "new_path": str(resolved), "requested_path": str(rec.target), "result": result, }) continue target = rec.target if target == source: result = "unchanged" resolved = target collision = "" reserved.setdefault(str(target), source) else: occupied_by = reserved.get(str(target)) if occupied_by is not None: if same_content(source, occupied_by): result = "EXACT duplicate -> would remove source" resolved = target collision = "EXACT → remove duplicate" else: variant, dup_of = _find_variant_or_duplicate(source, target, reserved) if dup_of is not None: result = "EXACT duplicate -> would remove source" resolved = dup_of collision = "EXACT → remove duplicate" else: resolved = variant result = "DIFF -> would move as variant" collision = f"DIFF → {resolved.name}" reserved[str(resolved)] = source elif target.exists(): if target.is_file() and same_content(source, target): result = "EXACT duplicate -> would remove source" resolved = target collision = "EXACT → remove duplicate" reserved[str(target)] = target else: # Base name is occupied by different content. Preserve both. reserved.setdefault(str(target), target) variant, dup_of = _find_variant_or_duplicate(source, target, reserved) if dup_of is not None: result = "EXACT duplicate -> would remove source" resolved = dup_of collision = "EXACT → remove duplicate" else: resolved = variant result = "DIFF -> would move as variant" collision = f"DIFF → {resolved.name}" reserved[str(resolved)] = source else: result = "would move" resolved = target collision = "" reserved[str(target)] = source rec.collision = collision rows.append({ "old_path": str(source), "new_path": str(resolved), "requested_path": str(target), "result": result, }) return rows def apply_confirmed(records: list[Record], log_path: Path) -> list[dict]: planned = plan(records) by_old = {str(r.path): r for r in records} timestamp = datetime.now().isoformat(timespec="seconds") out = [] for row in planned: result = row["result"] old = Path(row["old_path"]) new = Path(row["new_path"]) rec = by_old.get(str(old)) if result in ("would move", "DIFF -> would move as variant"): # Plan already proved 'new' is unoccupied. Re-check before touching disk. if new.exists(): result = "SKIPPED - resolved target unexpectedly exists" else: new.parent.mkdir(parents=True, exist_ok=True) if old.parent == new.parent: old.rename(new) else: shutil.move(str(old), str(new)) result = "moved" if row["result"] == "would move" else "moved as variant" if rec: rec.path = new try: rec.rel_path = str(new.relative_to(rec.root)) except ValueError: rec.rel_path = str(new.relative_to(rec.target_root)) rec.current_folder = str(new.parent.relative_to(rec.target_root)) rec.filename = new.name rec.proposed_name = new.name rec.collision = "" rec.status = "done" elif result == "EXACT duplicate -> would remove source": # Only reached after a full SHA-256 equality check in plan(). try: old.unlink() result = "removed exact duplicate" if rec: rec.collision = "" rec.status = "done" except OSError as e: result = f"SKIPPED - could not remove duplicate: {e}" out.append({"timestamp": timestamp, **row, "result": result}) if out: write_header = not log_path.exists() with log_path.open("a", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=out[0].keys()) if write_header: writer.writeheader() writer.writerows(out) return out # ---------------- GUI ---------------- def launch_gui(root: Path, target_root: Path, records: list[Record], destinations: list[str], endpoint: str, model: str, batch_size: int, aliases: Optional[dict[str, list[str]]] = None, inbox_mode: bool = False): import tkinter as tk from tkinter import ttk, messagebox class App: def __init__(self, win): self.win = win self.records = records self.destinations = destinations self.by_iid = {} self.q = queue.Queue() self.analysis_running = False self.active_editor = None win.title(f"Quick Sorter — {root}") win.geometry("1500x760") self._build() self._populate() def _build(self): bar = ttk.Frame(self.win, padding=6) bar.pack(fill="x") ttk.Button(bar, text="Analyse selected", command=self.analyse_selected).pack(side="left", padx=2) if inbox_mode: ttk.Button(bar, text="Rescan Inbox", command=self.rescan).pack(side="left", padx=2) ttk.Button(bar, text="Confirm selected", command=lambda: self.set_status("confirmed")).pack(side="left", padx=2) ttk.Button(bar, text="Confirm ≥ 0.85", command=self.confirm_confident).pack(side="left", padx=2) ttk.Button(bar, text="Skip selected", command=lambda: self.set_status("skip")).pack(side="left", padx=2) ttk.Separator(bar, orient="vertical").pack(side="left", fill="y", padx=8) ttk.Button(bar, text="Set folder…", command=self.set_folder).pack(side="left", padx=2) ttk.Separator(bar, orient="vertical").pack(side="left", fill="y", padx=8) ttk.Button(bar, text="Preview confirmed", command=self.preview).pack(side="left", padx=2) ttk.Button(bar, text="Apply confirmed", command=self.apply).pack(side="left", padx=2) cols = ("status", "file", "date", "source", "content", "destination", "semantic", "name", "confidence", "collision", "reason") self.tree = ttk.Treeview(self.win, columns=cols, show="headings", selectmode="extended") widths = {"status": 85, "file": 300, "date": 95, "source": 70, "content": 120, "destination": 200, "semantic": 250, "name": 360, "confidence": 80, "collision": 260, "reason": 230} for c in cols: self.tree.heading(c, text=c.title(), command=lambda col=c: self.sort(col, False)) self.tree.column(c, width=widths[c], stretch=False) self.tree.pack(fill="both", expand=True) self.tree.bind("", self.edit_cell) self.status = ttk.Label(self.win, padding=6) self.status.pack(fill="x") def values(self, r): semantic = " / ".join(x for x in (r.topic, r.subtopic, r.document_type, r.detail, r.version) if x) return (r.status, r.rel_path, r.date_guess, r.date_source, r.content_status, r.destination, semantic, r.proposed_name, f"{r.confidence:.2f}" if r.confidence else "", r.collision, r.reason) def _populate(self): self.tree.delete(*self.tree.get_children()) self.by_iid.clear() for idx, rec in enumerate(self.records): iid = str(idx) self.by_iid[iid] = rec self.tree.insert("", "end", iid=iid, values=self.values(rec)) self._counts() def refresh(self, iid): rec = self.by_iid[iid] self.tree.item(iid, values=self.values(rec)) def _counts(self): counts = {} for r in self.records: counts[r.status] = counts.get(r.status, 0) + 1 self.status.config(text=" / ".join(f"{k}: {v}" for k, v in sorted(counts.items())) + f" / total: {len(self.records)}") def selected(self): return [(iid, self.by_iid[iid]) for iid in self.tree.selection()] def recompute_collisions(self): # Preserve global pre-pass markers on rows that have not yet been # confirmed. plan() still handles ordinary same-target collisions. for rec in self.records: if not rec.duplicate_of: rec.collision = "" plan(self.records) for iid in self.by_iid: self.refresh(iid) def rescan(self): if self.analysis_running: messagebox.showinfo("Busy", "Analysis is already running.") return old = {str(r.path): r for r in self.records} fresh = scan( root, target_root, self.destinations, aliases=aliases, allow_stay_outside_policy=inbox_mode, ) merged = [] new_count = 0 for rec in fresh: previous = old.get(str(rec.path)) if previous is not None: merged.append(previous) else: merged.append(rec) new_count += 1 self.records = merged global_duplicate_prepass(self.records, root, target_root) self._populate() messagebox.showinfo( "Rescan complete", f"New files: {new_count}\nCurrently in source folder: {len(self.records)}", ) def analyse_selected(self): if self.analysis_running: messagebox.showinfo("Busy", "Analysis is already running.") return selected = [r for _, r in self.selected()] duplicate_count = sum(1 for r in selected if r.duplicate_of) archive_count = sum(1 for r in selected if not r.duplicate_of and (r.status == "archive" or is_archive_file(r.path))) chosen = [ r for r in selected if not r.duplicate_of and r.status != "archive" and not is_archive_file(r.path) ] if not chosen: if duplicate_count: messagebox.showinfo( "Exact duplicate selected", f"{duplicate_count} exact SHA-256 duplicate(s) are already resolved and are not sent to the LLM.\n" "Confirm them only when you want Apply to remove the redundant SOURCE copy.", ) elif archive_count: messagebox.showinfo( "Archive selected", "Archive files are intentionally not sent to the LLM.\n" "Unpack them first, then use Rescan Inbox.", ) else: messagebox.showinfo("Select files", "Select one or more rows first.") return self.analysis_running = True self.status.config(text=f"Analysing {len(chosen)} file(s) in batches…") def worker(): for rec in chosen: if not rec.excerpt: rec.excerpt, rec.content_status = extract_excerpt(rec.path) rec.candidates = choose_candidates(rec, self.destinations, aliases=aliases) for batch, results, error in llm_batches(chosen, endpoint, model, batch_size): self.q.put((batch, results, error)) self.q.put((None, None, "__DONE__")) threading.Thread(target=worker, daemon=True).start() self.win.after(100, self.poll) def poll(self): try: while True: batch, results, error = self.q.get_nowait() if error == "__DONE__": self.analysis_running = False self._counts() return if error: for r in batch: r.status = "review" r.reason = f"LLM error: {error[:80]}" else: apply_llm_results(batch, results) for iid, rec in self.by_iid.items(): if rec in batch: self.refresh(iid) except queue.Empty: pass self.win.after(120, self.poll) def set_status(self, status): for iid, rec in self.selected(): rec.status = status self.refresh(iid) self.recompute_collisions() self._counts() def confirm_confident(self): for iid, rec in self.by_iid.items(): if rec.status == "suggested" and rec.confidence >= 0.85: rec.status = "confirmed" self.refresh(iid) self.recompute_collisions() self._counts() def set_folder(self): chosen = self.selected() if not chosen: return top = tk.Toplevel(self.win) top.title("Set destination") var = tk.StringVar() combo = ttk.Combobox(top, values=self.destinations, textvariable=var, width=55) combo.pack(padx=10, pady=10) combo.focus() def go(): dest = var.get().strip() if dest in self.destinations or dest == "": for iid, rec in chosen: rec.destination = dest rec.status = "review" self.refresh(iid) top.destroy() ttk.Button(top, text="Apply", command=go).pack(pady=(0, 10)) combo.bind("", lambda _e: go()) def edit_cell(self, event): region = self.tree.identify("region", event.x, event.y) if region != "cell": return iid = self.tree.identify_row(event.y) col = int(self.tree.identify_column(event.x)[1:]) - 1 if not iid or col not in (5, 7): # destination/name only return bbox = self.tree.bbox(iid, f"#{col + 1}") if not bbox: return x, y, w, h = bbox rec = self.by_iid[iid] if self.active_editor: self.active_editor.destroy() if col == 5: editor = ttk.Combobox(self.tree, values=self.destinations) editor.set(rec.destination) else: editor = ttk.Entry(self.tree) editor.insert(0, rec.proposed_name) editor.select_range(0, tk.END) editor.place(x=x, y=y, width=w, height=h) editor.focus() self.active_editor = editor def commit(_e=None): value = editor.get().strip() if col == 5 and value in self.destinations: rec.destination = value elif col == 7 and value: rec.proposed_name = Path(value).name rec.status = "review" editor.destroy() self.active_editor = None self.refresh(iid) editor.bind("", commit) editor.bind("", commit) editor.bind("", lambda _e: editor.destroy()) def preview(self): self.recompute_collisions() rows = plan(self.records) if not rows: messagebox.showinfo("Preview", "No confirmed rows.") return text = "\n".join(f"{r['result']}:\n {r['old_path']}\n → {r['new_path']}" for r in rows[:30]) if len(rows) > 30: text += f"\n\n… and {len(rows)-30} more" messagebox.showinfo("Dry-run preview", text) def apply(self): rows = plan(self.records) if not rows: messagebox.showinfo("Apply", "No confirmed rows.") return normal = sum(1 for r in rows if r["result"] == "would move") variants = sum(1 for r in rows if r["result"] == "DIFF -> would move as variant") duplicates = sum(1 for r in rows if r["result"] == "EXACT duplicate -> would remove source") unchanged = sum(1 for r in rows if r["result"] == "unchanged") summary = ( f"Normal move/rename: {normal}\n" f"Different-content collisions kept as variant: {variants}\n" f"Exact SHA-256 duplicates to remove: {duplicates}\n" f"Unchanged: {unchanged}\n\n" "No existing file will be overwritten. Exact duplicates are only " "removed after a full-file SHA-256 match." ) if not messagebox.askyesno("Apply resolved batch?", summary): return result = apply_confirmed(self.records, root / LOG_FILENAME) moved = sum(1 for x in result if x["result"] in ("moved", "moved as variant")) removed = sum(1 for x in result if x["result"] == "removed exact duplicate") skipped = sum(1 for x in result if x["result"].startswith("SKIPPED")) if inbox_mode: self.records = scan( root, target_root, self.destinations, aliases=aliases, allow_stay_outside_policy=True, ) self._populate() messagebox.showinfo( "Done", f"Moved/renamed: {moved}\nRemoved exact duplicates: {removed}\nSkipped: {skipped}\n" f"Remaining in Inbox: {len(self.records) if inbox_mode else 'n/a'}\nLog: {root / LOG_FILENAME}" ) def sort(self, column, reverse): rows = [(self.tree.set(iid, column), iid) for iid in self.tree.get_children("")] rows.sort(key=lambda x: x[0].lower(), reverse=reverse) for pos, (_, iid) in enumerate(rows): self.tree.move(iid, "", pos) self.tree.heading(column, command=lambda: self.sort(column, not reverse)) win = tk.Tk() App(win) win.mainloop() def dependency_report() -> list[tuple[str, str, bool]]: checks = [ ("DOCX", "python-docx", "docx"), ("PDF", "pypdf", "pypdf"), ("XLSX", "openpyxl", "openpyxl"), ("PPTX", "python-pptx", "pptx"), ("ODT", "odfpy", "odf"), ("Images metadata", "Pillow", "PIL"), ] out = [] for label, package, module in checks: try: __import__(module) ok = True except ImportError: ok = False out.append((label, package, ok)) return out # ---------------- CLI ---------------- def main(): ap = argparse.ArgumentParser() ap.add_argument("root", nargs="?", help="Folder to review, e.g. /data/04_ARBEIT") ap.add_argument("--policy", default=str(Path(__file__).with_name("sorter_policy.json"))) ap.add_argument("--profile", default="", help="Policy profile to use even when the test folder has a different name, e.g. 04_ARBEIT or 01_INBOX") ap.add_argument("--target-root", default="", help="Base folder destinations are relative to. For 01_INBOX this defaults to the Inbox parent.") ap.add_argument("--endpoint", default=DEFAULT_ENDPOINT) ap.add_argument("--model", default=DEFAULT_MODEL) ap.add_argument("--batch-size", type=int, default=10) ap.add_argument("--scan-only", action="store_true", help="Read-only summary; do not open GUI") ap.add_argument("--check-deps", action="store_true", help="Show optional document readers available in this Python environment and exit") args = ap.parse_args() if args.check_deps: for label, package, ok in dependency_report(): print(f"{'OK' if ok else 'MISSING':7} {label:16} pip package: {package}") return if not args.root: raise SystemExit("root folder is required unless --check-deps is used") root = Path(args.root).expanduser().resolve() if not root.is_dir(): raise SystemExit(f"Not a directory: {root}") policy = load_policy(Path(args.policy)) profile = args.profile.strip() or root.name inbox_mode = profile == "01_INBOX" or root.name == "01_INBOX" if args.target_root: target_root = Path(args.target_root).expanduser().resolve() elif inbox_mode: target_root = root.parent else: target_root = root if not target_root.is_dir(): raise SystemExit(f"Not a target directory: {target_root}") destinations = allowed_destinations(root, policy, args.profile) aliases = policy.get("_strong_aliases", {}).get(profile, {}) records = scan( root, target_root, destinations, aliases=aliases, allow_stay_outside_policy=inbox_mode, ) dedupe = global_duplicate_prepass(records, root, target_root) print(f"Root: {root}") print(f"Target root: {target_root}") print(f"Policy profile: {profile}") print(f"Files: {len(records)}") print(f"Archives: {sum(r.status == 'archive' for r in records)}") if dedupe["enabled"]: print( "Global SHA-256 pre-pass: " f"target files={dedupe['target_files']}, source files={dedupe['source_files']}, " f"already in target={dedupe['target_duplicates']}, " f"duplicate incoming={dedupe['source_duplicates']}, hash errors={dedupe['hash_errors']}" ) else: print(f"Global SHA-256 pre-pass: disabled ({dedupe['reason']})") print(f"Allowed destinations: {len(destinations)}") print(f"Strong route aliases: {sum(len(v) for v in aliases.values())} across {len(aliases)} branches") print(f"Date from filename: {sum(r.date_source == 'filename' for r in records)}") print(f"Date from dated parent folder: {sum(r.date_source == 'folder' for r in records)}") print(f"Date from mtime fallback: {sum(r.date_source == 'mtime' for r in records)}") if args.scan_only: for r in records[:25]: duplicate = f" -> {r.duplicate_of}" if r.duplicate_of else "" print(f"{r.status:15} {r.date_guess} [{r.date_source:8}] {r.rel_path}{duplicate}") return launch_gui(root, target_root, records, destinations, args.endpoint, args.model, max(1, args.batch_size), aliases=aliases, inbox_mode=inbox_mode) if __name__ == "__main__": main()