1469 lines
60 KiB
Python
1469 lines
60 KiB
Python
#!/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
|
||
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"(?<!\d)(20\d{2})[-_.](\d{1,2})[-_.](\d{1,2})(?!\d)", stem)
|
||
if m:
|
||
value = _valid(int(m[1]), int(m[2]), int(m[3]))
|
||
if value:
|
||
return value
|
||
|
||
# YYYYMMDD
|
||
m = re.search(r"(?<!\d)(20\d{2})(\d{2})(\d{2})(?!\d)", stem)
|
||
if m:
|
||
value = _valid(int(m[1]), int(m[2]), int(m[3]))
|
||
if value:
|
||
return value
|
||
|
||
# DD.-DD.MM.YYYY or DD-DD-MM-YYYY -> first day of range
|
||
m = re.search(r"(?<!\d)(\d{1,2})\.?[-–](\d{1,2})[.\-_](\d{1,2})[.\-_](20\d{2})(?!\d)", stem)
|
||
if m:
|
||
value = _valid(int(m[4]), int(m[3]), int(m[1]))
|
||
if value:
|
||
return value
|
||
|
||
# DD.MM.YYYY, DD-MM-YYYY, DD_MM_YYYY
|
||
m = re.search(r"(?<!\d)(\d{1,2})[.\-_](\d{1,2})[.\-_](20\d{2})(?!\d)", stem)
|
||
if m:
|
||
value = _valid(int(m[3]), int(m[2]), int(m[1]))
|
||
if value:
|
||
return value
|
||
|
||
# Legacy YY-MM-DD -> 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"(?<!\d)(2\d)[-_](\d{1,2})[-_](\d{1,2})(?!\d)", stem)
|
||
if m:
|
||
value = _valid(2000 + int(m[1]), int(m[2]), int(m[3]))
|
||
if value:
|
||
return value
|
||
|
||
# DD.MM.YY only with dots (avoids interpreting time-like underscore strings)
|
||
m = re.search(r"(?<!\d)(\d{1,2})\.(\d{1,2})\.(2\d)(?!\d)", stem)
|
||
if m:
|
||
value = _valid(2000 + int(m[3]), int(m[2]), int(m[1]))
|
||
if value:
|
||
return value
|
||
|
||
return None
|
||
|
||
|
||
def _document_created(path: Path) -> 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 plausibly dates the document.
|
||
|
||
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.
|
||
"""
|
||
for part in reversed(path.parent.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) -> 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) + " "
|
||
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:
|
||
current = str(p.parent.relative_to(root))
|
||
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.
|
||
source_context = root.name
|
||
relative_parent = str(p.parent.relative_to(root))
|
||
if relative_parent != ".":
|
||
source_context = f"{root.name}/{relative_parent}"
|
||
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 ""),
|
||
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, 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.
|
||
- 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.
|
||
- 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. Never reduce an informative original filename to a generic name such as "allgemein_dokument".
|
||
- 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,
|
||
"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(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()
|
||
structured_name = build_structured_filename(rec)
|
||
|
||
# Filesystem mtime is provenance metadata, not a trustworthy document
|
||
# date. If the incoming filename already carries useful semantics, keep
|
||
# it instead of replacing it with an apparently precise mtime-based name.
|
||
# We still allow the LLM to suggest the destination and semantic fields.
|
||
informative_existing_name = not is_generic_filename(rec.filename)
|
||
generic_model_type = sanitise_token(rec.document_type) in {
|
||
"dokument", "document", "datei", "file", "unterlagen", "sonstiges", "misc"
|
||
}
|
||
preserve_existing_name = (
|
||
rec.date_source == "mtime" and informative_existing_name
|
||
) or (
|
||
generic_model_type and informative_existing_name
|
||
)
|
||
name = rec.filename if preserve_existing_name else structured_name
|
||
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)
|
||
|
||
# If the only available date is mtime and a generic filename still needs
|
||
# a generated rename, require explicit human review. An informative
|
||
# existing filename is preserved above, so no unverified date is injected.
|
||
if rec.date_source == "mtime" and name != rec.filename:
|
||
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, 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("<Double-1>", 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("<Return>", 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("<Return>", commit)
|
||
editor.bind("<FocusOut>", commit)
|
||
editor.bind("<Escape>", 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()
|