#!/usr/bin/env python3 """ Quick Sorter v9.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 zipfile import xml.etree.ElementTree as ET import urllib.request import unicodedata 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") DEFAULT_FALLBACK_MODEL = os.getenv("QUICK_SORTER_LLM_FALLBACK_MODEL", "architect") 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 = "" name_title: str = "" name_subtitle: str = "" collision: str = "" # exact duplicate / different-content variant handling duplicate_of: str = "" # full-SHA-256 match found by the global pre-pass source_context: str = "" # provenance: source-root name + relative source folder @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, *, allow_legacy_yy_mm_dd: bool = True) -> Optional[str]: """Conservative parser: known filename dates beat filesystem metadata. Legacy ``YY-MM-DD`` remains supported for real filenames because older files in this archive use it. Parent-folder parsing disables that ambiguous form unless surrounding wording (for example German ``Archiv bis DD-MM-YY``) makes the intended interpretation clear. """ stem = Path(name).stem # ISO date already present m = re.search(r"(? first day of range m = re.search(r"(? 20YY-MM-DD. This form is ambiguous with DD-MM-YY, # so parent-folder parsing deliberately disables it unless context resolves it. if allow_legacy_yy_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, root: Optional[Path] = None) -> Optional[str]: """Use a parent-folder date only when it plausibly dates the document. When ``root`` is supplied, only parent folders *inside* that scan root are considered. The root directory itself and every ancestor above it are deliberately excluded. This prevents staging/workspace names such as ``FULL_STAGING_2026-09-14`` from leaking into document dates. Snapshot/cutoff folders such as ``Archiv bis 25-10-16`` describe the archive boundary, not every document inside it. Their date must therefore never become the document date. Bare two-digit hyphen dates in other parent folders remain ambiguous and are not treated as verified dates. """ if root is not None: try: parts = path.parent.relative_to(root).parts except ValueError: # A caller supplied a boundary that does not contain this file. # Be conservative: do not inspect arbitrary ancestors outside it. return None else: # Backwards-compatible behaviour for direct helper calls that do not # provide a scan root. Production scanning always passes the root. parts = path.parent.parts for part in reversed(parts): normalised = _normalise_alias_text(part) # Archive cutoff/snapshot metadata is provenance, not a document date. # Example: "Archiv bis 25-10-16" can contain documents from 2023-2025. if "archiv" in normalised.split() and "bis" in normalised.split(): continue # Four-digit and dotted dates remain available, but ambiguous legacy # YY-MM-DD is disabled for parent folders. parsed = date_from_filename(part, allow_legacy_yy_mm_dd=False) if parsed: return parsed return None def best_date(path: Path, root: Optional[Path] = None) -> tuple[str, str]: # Established priority for this archive: filename -> trustworthy date-folder -> mtime. # Parent-folder lookup is root-bounded so workspace/staging names can never # become document metadata. from_name = date_from_filename(path.name) if from_name: return from_name, "filename" from_folder = date_from_parent_folders(path, root=root) 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) # Some DOCX files keep visible text in drawing/text-box XML that # python-docx does not expose through doc.paragraphs or tables. # Only use this raw XML path as a fallback to avoid duplicating # ordinary paragraph/table text. if not text: xml_parts = [] try: with zipfile.ZipFile(path) as zf: for name in zf.namelist(): if not (name.startswith("word/") and name.endswith(".xml")): continue try: root = ET.fromstring(zf.read(name)) except ET.ParseError: continue for node in root.iter(): if node.tag.endswith("}t") and node.text and node.text.strip(): xml_parts.append(node.text.strip()) if sum(len(x) for x in xml_parts) >= MAX_EXCERPT: break if sum(len(x) for x in xml_parts) >= MAX_EXCERPT: break except (OSError, zipfile.BadZipFile): xml_parts = [] text = _clip(xml_parts) return text, f"docx:{len(text)}" if text else "docx:no-text" if ext == ".pdf": try: from pypdf import PdfReader from pypdf.errors import DependencyError except ImportError: return "", "missing:pypdf" try: 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" except DependencyError as exc: if "cryptography" in str(exc).lower(): return "", "missing:cryptography" raise 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 = [] # python-pptx Slides supports integer indexing but slicing can # raise AttributeError. Iterate normally and stop after 5 slides. for slide_index, slide in enumerate(prs.slides): if slide_index >= 5: break 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__}" FILENAME_PREFIXES = { "Allgemein": "Allgemein", "Cafe_Bach": "CB", "Checkpoint": "Checkpoint", "CSD-Kampagnen": "CSD", "Ehrenamt": "Ehrenamt", "FAK": "FAK", "SP5-Treffen": "SP5", "Vor-Ort-Arbeit": "VOA", } def _filename_prefix(destination: str) -> str: branch = destination.split("/", 1)[0].strip() return FILENAME_PREFIXES.get(branch, branch.replace("_", "-") or "Datei") GENERIC_FILENAME_LABELS = { "dokument", "document", "datei", "file", "unterlagen", "sonstiges", "misc", "info", "information", } CATEGORY_NOISE_LABELS = { "allgemein", "anleitungen", "kommunikation", "materialien", "planung", "protokolle", "veranstaltung", "veranstaltungen", "verwaltung", "vorlagen", } PREFIX_ALIAS_PHRASES = { "Cafe_Bach": ("Café Bach", "Cafe Bach", "Cafe_Bach", "cafebach", "CB"), "Checkpoint": ("Checkpoint", "CP"), "CSD-Kampagnen": ("CSD-Kampagnen", "CSD Kampagnen", "CSD"), "Ehrenamt": ("Ehrenamt",), "FAK": ("FAK",), "SP5-Treffen": ("SP5-Treffen", "SP5 Treffen", "SP5"), "Vor-Ort-Arbeit": ("Vor-Ort-Arbeit", "Vor Ort Arbeit", "VOA"), "Allgemein": ("Allgemein",), } def _filename_component(text: str, *, title_case_first: bool = False) -> str: """Return one Windows/Linux-safe filename component while preserving Unicode. German spelling/capitalization is kept. Underscores remain reserved for the schema separators. Cosmetic punctuation such as commas/semicolons is removed because it adds noise without helping identification. """ value = unicodedata.normalize("NFC", str(text or "")).strip() value = re.sub(r"^\d{4}[-_. ]\d{1,2}[-_. ]\d{1,2}[-_ ]*", "", value) value = re.sub(r"\.[A-Za-z0-9]{1,6}$", "", value) value = value.replace("_", " ") value = re.sub(r'[<>:"/\\|?*\x00-\x1f]', " ", value) value = re.sub(r"[,;]+", " ", value) value = re.sub(r"\s+", "-", value) value = re.sub(r"-+", "-", value).strip("-. ") if title_case_first and value and value[0].islower(): value = value[0].upper() + value[1:] return value def _semantic_key(text: str) -> str: folded = unicodedata.normalize("NFKD", str(text or "")).casefold() folded = "".join(ch for ch in folded if not unicodedata.combining(ch)) folded = folded.replace("ß", "ss") return re.sub(r"[^a-z0-9]+", "", folded) def _branch_name(destination: str) -> str: return destination.split("/", 1)[0].strip() def _prefix_alias_keys(destination: str) -> set[str]: branch = _branch_name(destination) values = set(PREFIX_ALIAS_PHRASES.get(branch, ())) values.add(_filename_prefix(destination)) values.add(branch) return {_semantic_key(x) for x in values if x} def _strip_redundant_date_text(text: str, date_guess: str) -> str: """Remove a full/date-fragment that merely repeats the leading date. We deliberately do not remove a bare year: `März 2024` and `Inventur 2023` can carry useful semantic meaning even when mtime happens to be in that year. """ value = str(text or "") m = re.fullmatch(r"(\d{4})-(\d{2})-(\d{2})", str(date_guess or "")) if not m: return value y, mo, d = m.groups() yy = y[-2:] patterns = [ rf"(? str: return _filename_component( _strip_redundant_date_text(text, rec.date_guess), title_case_first=title_case_first, ) def _fallback_title(rec: Record) -> str: for value in (rec.document_type, rec.subtopic, rec.detail): candidate = _component_for_record(value, rec, title_case_first=True) if candidate and _semantic_key(candidate) not in GENERIC_FILENAME_LABELS: return candidate stem = _component_for_record(Path(rec.filename).stem, rec, title_case_first=True) if stem and _semantic_key(stem) not in GENERIC_FILENAME_LABELS: return stem return "Dokument" def _source_remainder(rec: Record, *blocked_values: str) -> str: """Recover useful detail from the original filename when model fields are noisy. This is intentionally conservative: it is only used as a fallback subtitle. Examples: `Getränkeliste Verzehrkarte` -> `Verzehrkarte`; `Inventur Café Bach 2023` -> `2023`. """ raw = _strip_redundant_date_text(Path(rec.filename).stem, rec.date_guess) # Remove the branch/prefix phrase before token-level subtraction. for phrase in sorted(PREFIX_ALIAS_PHRASES.get(_branch_name(rec.destination), ()), key=len, reverse=True): raw = re.sub(rf"(?= 3): continue kept.append(candidate) return _filename_component(" ".join(kept)) def _fallback_subtitle(rec: Record, title: str, prefix: str) -> str: blocked = {_semantic_key(title), _semantic_key(prefix), _semantic_key(rec.topic)} blocked |= _prefix_alias_keys(rec.destination) blocked |= CATEGORY_NOISE_LABELS for value in (rec.subtopic, rec.detail): candidate = _component_for_record(value, rec) key = _semantic_key(candidate) if candidate and key and key not in blocked and key not in GENERIC_FILENAME_LABELS: return candidate return _source_remainder(rec, title, prefix, rec.topic) def build_structured_filename(rec: Record) -> str: """Build YYYY-MM-DD_Topic_Subtopic_DocumentType_Detail[_vN].ext. Destination folders are deliberately NOT encoded in filenames. The filename describes what the document is; the folder describes where it lives. """ parts = [rec.date_guess] seen = {_semantic_key(rec.date_guess)} for value in ( rec.topic, rec.subtopic, rec.document_type, rec.detail, ): token = _filename_component(value, title_case_first=True) key = _semantic_key(token) if not token or not key: continue if key in seen: continue if key in CATEGORY_NOISE_LABELS: continue parts.append(token) seen.add(key) ver = _filename_component(rec.version) if ver: if re.fullmatch(r"\d+", ver): ver = f"v{ver}" elif not re.fullmatch(r"v\d+", ver, flags=re.I): ver = f"v{ver}" if _semantic_key(ver) not in seen: parts.append(ver) # If the model supplied no useful semantic information, preserve the # original filename instead of inventing a meaningless new one. if len(parts) == 1: return rec.filename stem = "_".join(parts) # Leave headroom for destination path and possible collision suffixes. if len(stem) > 180: stem = stem[:180].rstrip("-_. ") return stem + 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) + " " source_text = " " + _normalise_alias_text(rec.source_context) + " " 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 source_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.source_context} {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: relative_parent = str(p.parent.relative_to(root)) if allow_stay_outside_policy: current = ( "01_INBOX" if relative_parent == "." else f"01_INBOX/{relative_parent}" ) else: current = relative_parent if current == ".": current = "" # Keep provenance separate from destination logic. This matters when an # incoming backup root itself carries useful ownership context (for # example a Cafe-Bach backup) that would otherwise disappear when paths # are made relative to the source root. context_root = "01_INBOX" if allow_stay_outside_policy else root.name source_context = context_root relative_parent = str(p.parent.relative_to(root)) if relative_parent != ".": source_context = f"{context_root}/{relative_parent}" d, source = best_date(p, root=root) 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 ""), source_context=source_context, ) 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, title, subtitle, version, confidence, reason Rules: - destination MUST be one of that item's candidates. Never invent a folder. - In 01_INBOX mode, classify the PURPOSE/ROLE of the document before its owner or topic. - Top-level branch meaning: * 03_PRIVAT = personal-life documents and records: health, private administration, household, personal correspondence, Chrystal, personal paperwork. * 04_ARBEIT = operational documents for Aidshilfe Köln, Cafe Bach, Checkpoint, Vor-Ort-Arbeit, work events, colleagues, work administration, work planning or work communication. * 05_VAULT = reusable knowledge/reference: technical guides, setup notes, AI prompts, manuals, documentation, Linux, shell, Syncthing, VPS, LLM, software or general instructions. Choose Vault even when the knowledge was created while solving a work or private task. * 99_ARCHIV = intentionally historical, retired or obsolete material; never choose it merely because a file is old. * 01_INBOX = quarantine only when the document cannot be classified safely from the available evidence. - Technical/reference material must NOT become 03_PRIVAT or 04_ARBEIT merely because a private or work context is mentioned. - After choosing the correct top-level branch, use owner/work-area evidence to choose the most specific candidate inside that branch. - For 05_VAULT, strongly prefer the most specific suitable Vault candidate instead of the 05_VAULT root: * 05_VAULT/Prompts = prompts, system prompts, start prompts, reusable LLM instructions or prompt templates. * 05_VAULT/Tools = tool-specific configuration, commands, helper-tool documentation or operational tool notes. * 05_VAULT/Wissen/IT = Linux, VPS, Syncthing, networking, hardware, software, LLM setup, system administration, technical guides and technical documentation. * 05_VAULT/Wissen/Organisation = productivity methods, workflows, planning systems and organisational knowledge. * 05_VAULT/Wissen = reusable knowledge that belongs in Wissen but has no clearly suitable more-specific Wissen candidate. * 05_VAULT = use only when no more-specific Vault candidate reasonably fits. - A document that IS itself a prompt belongs in 05_VAULT/Prompts even when its subject is IT, work, ADHD or another domain. - A technical guide ABOUT configuring or understanding a system normally belongs in 05_VAULT/Wissen/IT, not 05_VAULT/Tools. - If uncertain, keep the current Inbox 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. - title: concise German document label for the filename, e.g. Anschreiben, Inventur, Rechnung, Getränkeliste, Checkliste. - subtitle: concise distinguishing subject/entity/event/year, e.g. Phönix-Sauna, 2023, CB2428-Aidshilfe-Köln; empty if unnecessary. - title/subtitle MUST NOT contain the date, destination/prefix, extension, or version and MUST NOT repeat each other. - Prefer normal readable German spelling and capitalization in title/subtitle; do not slugify or translate into English. - 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. - source_context is provenance from the incoming source root/folder. Use it as supporting evidence for the primary owner/work area; generic backup labels are not destinations by themselves. - 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. - Within the already chosen purpose branch, 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 work sub-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. - Use an informative original filename as evidence for title/subtitle, but do not preserve its whole formatting; Python will normalize the final name. - 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. Do not invent a different date. - 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 _model_generation_limits(model: str) -> tuple[int, int]: """Return a generous output-token ceiling and HTTP timeout for one file. Reasoning models spend part of the completion budget before emitting final JSON. These are ceilings, not targets: models can stop much earlier. """ name = model.strip().lower() if "debug" in name: return 6000, 300 if "architect" in name: return 2500, 180 if "daily" in name: return 1000, 120 return 2000, 180 def _result_confidence(result: Optional[dict]) -> float: if not isinstance(result, dict): return 0.0 try: return max(0.0, min(1.0, float(result.get("confidence", 0)))) except (TypeError, ValueError): return 0.0 def _call_llm_one(rec: Record, endpoint: str, model: str) -> dict: item = { "id": 0, "filename": rec.filename, "current_folder": rec.current_folder, "source_context": rec.source_context, "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([item], ensure_ascii=False) max_tokens, timeout = _model_generation_limits(model) payload = { "model": model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.0, "max_tokens": max_tokens, } req = urllib.request.Request( endpoint, data=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(req, timeout=timeout) as resp: data = json.loads(resp.read().decode("utf-8")) choice = data["choices"][0] text = choice.get("message", {}).get("content") or "" parsed = extract_json_array(text) if not isinstance(parsed, list) or not parsed or not isinstance(parsed[0], dict): raise ValueError("model did not return a non-empty JSON array") result = dict(parsed[0]) result["id"] = 0 return result def _needs_fallback(rec: Record, result: Optional[dict]) -> bool: """Escalate only uncertain/fragile cases; do not pay Architect cost always.""" if not isinstance(result, dict): return True dest = str(result.get("destination", "")).strip() if dest not in rec.candidates: return True if _result_confidence(result) < 0.85: return True if is_generic_filename(rec.filename): return True return False def llm_batches( records: list[Record], endpoint: str, model: str, batch_size: int = 1, fallback_model: str = "", ): """Analyse exactly one file per LLM request. batch_size is retained for CLI/backward compatibility but intentionally ignored: one-file requests prevent cross-file influence and isolate failures. When a fallback model is configured, it is used only for a primary error, an invalid destination, confidence below 0.85, or a generic filename. If primary and fallback disagree on destination, the primary proposal is kept but confidence is capped at 0.69 so the GUI forces human review. """ del batch_size fallback_model = fallback_model.strip() for rec in records: batch = [rec] primary = None primary_error = None try: primary = _call_llm_one(rec, endpoint, model) except Exception as exc: primary_error = str(exc) use_fallback = ( bool(fallback_model) and fallback_model != model and (primary_error is not None or _needs_fallback(rec, primary)) ) if not use_fallback: if primary_error is not None: yield batch, None, primary_error else: yield batch, [primary], None continue try: fallback = _call_llm_one(rec, endpoint, fallback_model) except Exception as exc: fallback_error = str(exc) if primary is not None: # A usable primary answer is better than throwing it away merely # because the optional second opinion failed. yield batch, [primary], None else: yield batch, None, f"primary={primary_error}; fallback={fallback_error}" continue if primary is None: chosen = dict(fallback) chosen["reason"] = f"{fallback_model} fallback: {chosen.get('reason', '')}".strip() yield batch, [chosen], None continue primary_dest = str(primary.get("destination", "")).strip() fallback_dest = str(fallback.get("destination", "")).strip() if primary_dest == fallback_dest: # For the cases deliberately escalated, prefer the second opinion # when both models agree on the actual destination. chosen = dict(fallback) chosen["reason"] = f"{fallback_model} fallback: {chosen.get('reason', '')}".strip() yield batch, [chosen], None else: # Do not silently choose between disagreeing models. Keep the fast # primary proposal visible, but force explicit human review. chosen = dict(primary) chosen["confidence"] = min(_result_confidence(primary), 0.69) chosen["reason"] = ( f"Model disagreement: {model}={primary_dest}; " f"{fallback_model}={fallback_dest}" ) yield batch, [chosen], None 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() rec.name_title = str(result.get("title", "")).strip() rec.name_subtitle = str(result.get("subtitle", "")).strip() rec.proposed_name = build_structured_filename(rec) # In 01_INBOX mode, keeping a file in Inbox is a quarantine/review # outcome, not a successfully classified final destination. if dest == "01_INBOX" or dest.startswith("01_INBOX/"): conf = min(conf, 0.69) # 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) # Every V7 proposal follows the structured naming scheme. If its date # comes only from filesystem mtime, the date is useful but historically # unverified, so require explicit human review before apply. if rec.date_source == "mtime": conf = min(conf, 0.69) 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, fallback_model: str = "", 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), one request per file…") 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, allow_stay_outside_policy=inbox_mode, ) for batch, results, error in llm_batches(chosen, endpoint, model, batch_size, fallback_model=fallback_model): 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"), ("PDF AES", "cryptography", "cryptography"), ("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("--fallback-model", default=DEFAULT_FALLBACK_MODEL, help="Second-opinion model for uncertain cases; default: architect") ap.add_argument("--no-fallback", action="store_true", help="Disable the second-opinion model") ap.add_argument("--batch-size", type=int, default=1, help="Compatibility option; V7 always sends one file per LLM request") 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)}") fallback_model = "" if args.no_fallback else args.fallback_model.strip() print(f"Primary LLM model: {args.model}") print(f"Fallback LLM model: {fallback_model or 'disabled'}") 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), fallback_model=fallback_model, aliases=aliases, inbox_mode=inbox_mode) if __name__ == "__main__": main()