573 lines
19 KiB
Python
Executable file
573 lines
19 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
PROJECT = Path("/data/02_DEV/13_document_sorter")
|
|
SORTER = PROJECT / "2026-09-15_quick-sorter_destination-normalization_v18.py"
|
|
POLICY = PROJECT / "sorter_policy.json"
|
|
DATA_ROOT = Path("/data")
|
|
INBOX = DATA_ROOT / "01_INBOX"
|
|
RUNS_ROOT = DATA_ROOT / "90_SORT_DAILY_INBOX"
|
|
STATE_DIR = Path.home() / ".local" / "state" / "inbox-sort"
|
|
STATE_FILE = STATE_DIR / "deferred.json"
|
|
HISTORY_FILE = STATE_DIR / "history.log"
|
|
ENDPOINT = "http://100.105.210.21:8080/v1/chat/completions"
|
|
MODEL = "architect"
|
|
|
|
def hash_file(path: Path) -> str:
|
|
h = hashlib.sha256()
|
|
with path.open("rb") as f:
|
|
for chunk in iter(lambda: f.read(1024 * 1024), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
def ignored(rel: Path) -> bool:
|
|
if ".stfolder" in rel.parts:
|
|
return True
|
|
name = rel.name
|
|
if name in {".directory", ".quick_sorter_log.csv"}:
|
|
return True
|
|
if name.startswith(".sync-conflict-") and name.endswith(".directory"):
|
|
return True
|
|
return False
|
|
|
|
def scan_inbox() -> dict[str, dict[str, object]]:
|
|
out: dict[str, dict[str, object]] = {}
|
|
if not INBOX.is_dir():
|
|
raise SystemExit(f"STOP: Inbox does not exist: {INBOX}")
|
|
for p in INBOX.rglob("*"):
|
|
if not p.is_file():
|
|
continue
|
|
rel = p.relative_to(INBOX)
|
|
if ignored(rel):
|
|
continue
|
|
out[str(rel)] = {
|
|
"sha256": hash_file(p),
|
|
"size": p.stat().st_size,
|
|
}
|
|
return out
|
|
|
|
def load_state() -> dict[str, dict[str, object]]:
|
|
if not STATE_FILE.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(STATE_FILE.read_text(encoding="utf-8"))
|
|
if isinstance(data, dict) and "deferred" in data:
|
|
data = data["deferred"]
|
|
if not isinstance(data, dict):
|
|
return {}
|
|
return data
|
|
except Exception as exc:
|
|
raise SystemExit(f"STOP: cannot read deferred state: {exc}")
|
|
|
|
def save_state(deferred: dict[str, dict[str, object]]) -> None:
|
|
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
|
payload = {
|
|
"version": 1,
|
|
"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
"deferred": dict(sorted(deferred.items())),
|
|
}
|
|
tmp = STATE_FILE.with_suffix(".tmp")
|
|
tmp.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
|
os.replace(tmp, STATE_FILE)
|
|
|
|
def load_sorter_module():
|
|
if not SORTER.is_file():
|
|
raise SystemExit(f"STOP: sorter missing: {SORTER}")
|
|
spec = importlib.util.spec_from_file_location("inbox_sorter_v18", SORTER)
|
|
if spec is None or spec.loader is None:
|
|
raise SystemExit("STOP: could not load sorter V18")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
def allowed_destinations() -> list[str]:
|
|
module = load_sorter_module()
|
|
policy = module.load_policy(POLICY)
|
|
return module.allowed_destinations(INBOX, policy, "01_INBOX")
|
|
|
|
def destination_roots(destinations: list[str]) -> list[str]:
|
|
roots = sorted({d.split("/", 1)[0] for d in destinations if d != "01_INBOX"})
|
|
bad = [r for r in roots if r not in {"03_PRIVAT", "04_ARBEIT", "05_VAULT", "99_ARCHIV"}]
|
|
if bad:
|
|
raise SystemExit(f"STOP: unexpected destination root(s) in policy: {', '.join(bad)}")
|
|
return roots
|
|
|
|
def copy_file_preserving(src: Path, dst: Path) -> None:
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(src, dst)
|
|
|
|
def stage_baseline(target: Path, roots: list[str]) -> None:
|
|
print("\nBuilding safe staging baseline...")
|
|
for root in roots:
|
|
src = DATA_ROOT / root
|
|
dst = target / root
|
|
if not src.is_dir():
|
|
raise SystemExit(f"STOP: destination root missing: {src}")
|
|
dst.mkdir(parents=True, exist_ok=True)
|
|
cmd = [
|
|
"rsync", "-a",
|
|
"--exclude=.stfolder/",
|
|
"--exclude=.directory",
|
|
"--exclude=.sync-conflict-*.directory",
|
|
f"{src}/", f"{dst}/",
|
|
]
|
|
subprocess.run(cmd, check=True)
|
|
(target / "01_INBOX").mkdir(parents=True, exist_ok=True)
|
|
|
|
def parse_log(source: Path, target: Path) -> dict[str, Path]:
|
|
logfile = source / ".quick_sorter_log.csv"
|
|
out: dict[str, Path] = {}
|
|
if not logfile.is_file():
|
|
return out
|
|
with logfile.open("r", encoding="utf-8", newline="") as f:
|
|
for row in csv.reader(f):
|
|
if len(row) < 3:
|
|
continue
|
|
try:
|
|
src = Path(row[1])
|
|
dst = Path(row[2])
|
|
rel = src.relative_to(source)
|
|
except (ValueError, OSError):
|
|
continue
|
|
out[str(rel)] = dst
|
|
return out
|
|
|
|
def target_hash_index(target: Path) -> dict[tuple[int, str], list[Path]]:
|
|
out: dict[tuple[int, str], list[Path]] = {}
|
|
for p in target.rglob("*"):
|
|
if not p.is_file():
|
|
continue
|
|
rel = p.relative_to(target)
|
|
if rel.parts and rel.parts[0] == "01_INBOX":
|
|
continue
|
|
size = p.stat().st_size
|
|
digest = hash_file(p)
|
|
out.setdefault((size, digest), []).append(p)
|
|
return out
|
|
|
|
def under(child: Path, parent: Path) -> bool:
|
|
try:
|
|
child.relative_to(parent)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
def verify_target_against_live(target: Path, roots: list[str]):
|
|
additions: list[tuple[Path, Path]] = []
|
|
collisions: list[Path] = []
|
|
for root in roots:
|
|
base = target / root
|
|
if not base.exists():
|
|
continue
|
|
for p in base.rglob("*"):
|
|
if not p.is_file():
|
|
continue
|
|
rel = p.relative_to(target)
|
|
live = DATA_ROOT / rel
|
|
if not live.exists():
|
|
additions.append((p, live))
|
|
continue
|
|
if not live.is_file():
|
|
collisions.append(rel)
|
|
continue
|
|
if p.stat().st_size != live.stat().st_size or hash_file(p) != hash_file(live):
|
|
collisions.append(rel)
|
|
return additions, collisions
|
|
|
|
def show_dry_run(target: Path, roots: list[str]) -> None:
|
|
print("\n=== REAL IMPORT DRY RUN ===")
|
|
for root in roots:
|
|
base = target / root
|
|
if not base.exists():
|
|
continue
|
|
cmd = [
|
|
"rsync", "-aivn",
|
|
"--ignore-existing",
|
|
"--omit-dir-times",
|
|
f"{base}/", f"{DATA_ROOT / root}/",
|
|
]
|
|
result = subprocess.run(cmd, text=True, capture_output=True, check=True)
|
|
lines = [line for line in result.stdout.splitlines() if line.startswith(">f+++++++++")]
|
|
if lines:
|
|
print(f"\n[{root}]")
|
|
for line in lines:
|
|
print(line)
|
|
|
|
def append_history(message: str) -> None:
|
|
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
|
with HISTORY_FILE.open("a", encoding="utf-8") as f:
|
|
f.write(message.rstrip() + "\n")
|
|
|
|
def clean_empty_dirs() -> None:
|
|
dirs = [p for p in INBOX.rglob("*") if p.is_dir() and ".stfolder" not in p.parts]
|
|
for d in sorted(dirs, key=lambda p: len(p.parts), reverse=True):
|
|
try:
|
|
d.rmdir()
|
|
except OSError:
|
|
pass
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
prog="inbox-sort",
|
|
description="Safely stage, review and import new/changed files from /data/01_INBOX.",
|
|
)
|
|
parser.add_argument(
|
|
"--review-deferred",
|
|
action="store_true",
|
|
help="include unchanged deferred files in this run",
|
|
)
|
|
parser.add_argument(
|
|
"--status",
|
|
action="store_true",
|
|
help="show new/changed and deferred counts without changing anything",
|
|
)
|
|
parser.add_argument(
|
|
"--defer-current",
|
|
action="store_true",
|
|
help="one-time: mark every current Inbox file as deferred without sorting",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
if not POLICY.is_file():
|
|
raise SystemExit(f"STOP: sorter policy missing: {POLICY}")
|
|
|
|
current = scan_inbox()
|
|
deferred = load_state()
|
|
|
|
unchanged_deferred = {
|
|
rel for rel, info in current.items()
|
|
if rel in deferred
|
|
and deferred[rel].get("sha256") == info["sha256"]
|
|
and deferred[rel].get("size") == info["size"]
|
|
}
|
|
|
|
if args.review_deferred:
|
|
selected = set(current)
|
|
else:
|
|
selected = set(current) - unchanged_deferred
|
|
|
|
print("=== INBOX SORT ===")
|
|
print(f"Inbox files: {len(current)}")
|
|
print(f"New/changed for this run: {len(selected)}")
|
|
print(f"Deferred unchanged: {len(unchanged_deferred)}")
|
|
if args.review_deferred:
|
|
print("Mode: review deferred")
|
|
|
|
if not STATE_FILE.exists():
|
|
print("\nNote: no deferred state exists yet.")
|
|
print("This first run treats every current Inbox file as new.")
|
|
print("Anything you leave unresolved will become deferred.")
|
|
|
|
if args.status:
|
|
return 0
|
|
|
|
if args.defer_current:
|
|
print("\nWARNING: this marks every current Inbox file as already reviewed/deferred.")
|
|
answer = input("Type DEFER CURRENT to continue: ").strip()
|
|
if answer != "DEFER CURRENT":
|
|
print("Aborted.")
|
|
return 0
|
|
save_state(current)
|
|
print(f"Deferred baseline saved: {len(current)} file(s).")
|
|
return 0
|
|
|
|
if not selected:
|
|
print("\nNothing new or changed to sort.")
|
|
return 0
|
|
|
|
destinations = allowed_destinations()
|
|
roots = destination_roots(destinations)
|
|
|
|
stamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")
|
|
run_dir = RUNS_ROOT / f"{stamp}_{os.getpid()}"
|
|
source = run_dir / "SOURCE"
|
|
target = run_dir / "TARGET"
|
|
source.mkdir(parents=True)
|
|
target.mkdir(parents=True)
|
|
|
|
manifest = {rel: current[rel] for rel in sorted(selected)}
|
|
(run_dir / "manifest.json").write_text(
|
|
json.dumps(manifest, indent=2, ensure_ascii=False) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
print(f"\nStaging run: {run_dir}")
|
|
print("Copying new/changed Inbox files...")
|
|
for rel in sorted(selected):
|
|
copy_file_preserving(INBOX / rel, source / rel)
|
|
|
|
stage_baseline(target, roots)
|
|
|
|
print("\nOpening V18.")
|
|
print("Review the suggestions in the GUI.")
|
|
print("- Confirm + Apply files you want sorted.")
|
|
print("- Leave uncertain files unapplied, or route them to 01_INBOX.")
|
|
print("- Exact non-empty duplicates are resolved automatically by SHA-256.")
|
|
print("- Close the GUI when you are finished.\n")
|
|
|
|
cmd = [
|
|
sys.executable, str(SORTER),
|
|
str(source),
|
|
"--profile", "01_INBOX",
|
|
"--target-root", str(target),
|
|
"--endpoint", ENDPOINT,
|
|
"--model", MODEL,
|
|
"--no-fallback",
|
|
]
|
|
result = subprocess.run(cmd, cwd=PROJECT)
|
|
if result.returncode != 0:
|
|
print(f"\nSTOP: sorter exited with code {result.returncode}.")
|
|
print(f"Staging retained: {run_dir}")
|
|
return 1
|
|
|
|
log_map = parse_log(source, target)
|
|
index = target_hash_index(target)
|
|
|
|
unresolved: set[str] = set()
|
|
resolved: dict[str, Path] = {}
|
|
problems: list[str] = []
|
|
|
|
inbox_target = target / "01_INBOX"
|
|
|
|
for rel, info in manifest.items():
|
|
staged_source = source / rel
|
|
log_dest = log_map.get(rel)
|
|
|
|
# Explicitly routed back to Inbox => defer.
|
|
if log_dest is not None and under(log_dest, inbox_target):
|
|
unresolved.add(rel)
|
|
continue
|
|
|
|
key = (int(info["size"]), str(info["sha256"]))
|
|
matches = index.get(key, [])
|
|
|
|
# If the source still exists but a non-empty exact canonical duplicate
|
|
# exists, resolve it automatically. Zero-byte files are never auto-deduped.
|
|
if staged_source.exists():
|
|
if int(info["size"]) > 0 and matches:
|
|
resolved[rel] = matches[0]
|
|
else:
|
|
unresolved.add(rel)
|
|
continue
|
|
|
|
# Applied/moved file: prefer its logged destination if it verifies.
|
|
evidence = None
|
|
if log_dest is not None and under(log_dest, target) and not under(log_dest, inbox_target):
|
|
if log_dest.is_file():
|
|
if (
|
|
log_dest.stat().st_size == int(info["size"])
|
|
and hash_file(log_dest) == str(info["sha256"])
|
|
):
|
|
evidence = log_dest
|
|
|
|
if evidence is None and matches:
|
|
evidence = matches[0]
|
|
|
|
if evidence is None:
|
|
problems.append(f"{rel}: removed from staging SOURCE but no verified canonical copy found")
|
|
else:
|
|
resolved[rel] = evidence
|
|
|
|
if problems:
|
|
print("\nSTOP: post-GUI accounting failed.")
|
|
for problem in problems:
|
|
print(" -", problem)
|
|
print(f"Staging retained: {run_dir}")
|
|
return 1
|
|
|
|
additions, collisions = verify_target_against_live(target, roots)
|
|
if collisions:
|
|
print("\nSTOP: staging TARGET differs from an existing live destination path:")
|
|
for rel in collisions:
|
|
print(" -", rel)
|
|
print("No live changes were made.")
|
|
print(f"Staging retained: {run_dir}")
|
|
return 1
|
|
|
|
addition_roots = sorted({p.relative_to(target).parts[0] for p, _ in additions})
|
|
affected = list(addition_roots)
|
|
if resolved:
|
|
affected = sorted(set(affected) | {"01_INBOX"})
|
|
|
|
print("\n=== REVIEW RESULT ===")
|
|
print(f"Resolved from Inbox: {len(resolved)}")
|
|
print(f"Deferred after this run: {len(unresolved)}")
|
|
print(f"New destination files: {len(additions)}")
|
|
|
|
show_dry_run(target, addition_roots)
|
|
|
|
# If nothing will change live, just record the deferred decisions.
|
|
if not resolved and not additions:
|
|
latest = scan_inbox()
|
|
new_state: dict[str, dict[str, object]] = {}
|
|
for rel, info in latest.items():
|
|
old = deferred.get(rel)
|
|
if old and old.get("sha256") == info["sha256"] and old.get("size") == info["size"]:
|
|
new_state[rel] = info
|
|
elif rel in unresolved and rel in manifest:
|
|
if info == manifest[rel]:
|
|
new_state[rel] = info
|
|
save_state(new_state)
|
|
shutil.rmtree(run_dir)
|
|
print("\nNo live file changes were needed.")
|
|
print(f"Deferred state updated: {len(new_state)} file(s).")
|
|
return 0
|
|
|
|
print("\nAffected Syncthing folders:")
|
|
for name in affected:
|
|
print(" -", name)
|
|
|
|
answer = input(
|
|
"\nPause those folders manually in Syncthing now.\n"
|
|
"Type PAUSED when they are paused, or press Enter to abort: "
|
|
).strip()
|
|
if answer != "PAUSED":
|
|
print("\nAborted. No live changes made.")
|
|
print(f"Staging retained: {run_dir}")
|
|
return 0
|
|
|
|
confirm = input(
|
|
"\nApply the verified additions and remove resolved originals from 01_INBOX? [y/N]: "
|
|
).strip().lower()
|
|
if confirm != "y":
|
|
print("\nAborted. Resume the folders you paused.")
|
|
print(f"Staging retained: {run_dir}")
|
|
return 0
|
|
|
|
# Pre-flight check: every resolved Inbox original must still be unchanged.
|
|
live_now = scan_inbox()
|
|
changed = []
|
|
for rel, info in manifest.items():
|
|
if rel not in resolved:
|
|
continue
|
|
if rel not in live_now or live_now[rel] != info:
|
|
changed.append(rel)
|
|
if changed:
|
|
print("\nSTOP: resolved Inbox file(s) changed or disappeared during the run:")
|
|
for rel in changed:
|
|
print(" -", rel)
|
|
print("No live import/deletion was performed.")
|
|
print("Resume Syncthing and rerun inbox-sort.")
|
|
print(f"Staging retained: {run_dir}")
|
|
return 1
|
|
|
|
# Additive imports.
|
|
for root in addition_roots:
|
|
subprocess.run(
|
|
[
|
|
"rsync", "-aiv",
|
|
"--ignore-existing",
|
|
"--omit-dir-times",
|
|
f"{target / root}/", f"{DATA_ROOT / root}/",
|
|
],
|
|
check=True,
|
|
)
|
|
|
|
# Verify every new destination path exactly.
|
|
import_failures = []
|
|
for staged_file, live_file in additions:
|
|
if not live_file.is_file():
|
|
import_failures.append(f"missing after import: {live_file}")
|
|
continue
|
|
if (
|
|
staged_file.stat().st_size != live_file.stat().st_size
|
|
or hash_file(staged_file) != hash_file(live_file)
|
|
):
|
|
import_failures.append(f"content mismatch after import: {live_file}")
|
|
|
|
# Verify every resolved Inbox file has exact canonical evidence in live data.
|
|
for rel, evidence in resolved.items():
|
|
evidence_rel = evidence.relative_to(target)
|
|
live_evidence = DATA_ROOT / evidence_rel
|
|
info = manifest[rel]
|
|
if not live_evidence.is_file():
|
|
import_failures.append(f"{rel}: verified destination missing: {live_evidence}")
|
|
continue
|
|
if (
|
|
live_evidence.stat().st_size != int(info["size"])
|
|
or hash_file(live_evidence) != str(info["sha256"])
|
|
):
|
|
import_failures.append(f"{rel}: verified destination mismatch: {live_evidence}")
|
|
|
|
if import_failures:
|
|
print("\nSTOP: post-import verification failed.")
|
|
for item in import_failures:
|
|
print(" -", item)
|
|
print("Inbox originals were NOT deleted.")
|
|
print("Keep Syncthing paused and inspect this run:")
|
|
print(run_dir)
|
|
return 1
|
|
|
|
# Re-check all resolved originals immediately before deletion.
|
|
live_now = scan_inbox()
|
|
changed = [
|
|
rel for rel, info in manifest.items()
|
|
if rel in resolved and (rel not in live_now or live_now[rel] != info)
|
|
]
|
|
if changed:
|
|
print("\nSTOP: Inbox changed after import; originals were NOT deleted:")
|
|
for rel in changed:
|
|
print(" -", rel)
|
|
print("Destination imports are verified and additive.")
|
|
print("Keep Syncthing paused and inspect before continuing.")
|
|
print(run_dir)
|
|
return 1
|
|
|
|
for rel in sorted(resolved):
|
|
(INBOX / rel).unlink()
|
|
|
|
clean_empty_dirs()
|
|
|
|
# Update deferred state. Previously deferred unchanged files stay deferred;
|
|
# unresolved files from this run are added. New files arriving mid-run are not.
|
|
latest = scan_inbox()
|
|
new_state: dict[str, dict[str, object]] = {}
|
|
for rel, info in latest.items():
|
|
old = deferred.get(rel)
|
|
if old and old.get("sha256") == info["sha256"] and old.get("size") == info["size"]:
|
|
new_state[rel] = info
|
|
elif rel in unresolved and rel in manifest and info == manifest[rel]:
|
|
new_state[rel] = info
|
|
|
|
save_state(new_state)
|
|
|
|
append_history(
|
|
f"{datetime.now().astimezone().isoformat(timespec='seconds')} "
|
|
f"resolved={len(resolved)} additions={len(additions)} "
|
|
f"deferred={len(new_state)} affected={','.join(affected)}"
|
|
)
|
|
|
|
print("\n=== LOCAL RESULT CLEAN ===")
|
|
print(f"Resolved Inbox files removed: {len(resolved)}")
|
|
print(f"Verified destination additions: {len(additions)}")
|
|
print(f"Deferred unchanged for future runs: {len(new_state)}")
|
|
print("\nResume these Syncthing folders:")
|
|
for name in affected:
|
|
print(" -", name)
|
|
|
|
input("\nWait until all of them show Up to Date, then press Enter to finish: ")
|
|
|
|
shutil.rmtree(run_dir)
|
|
try:
|
|
RUNS_ROOT.rmdir()
|
|
except OSError:
|
|
pass
|
|
|
|
print("\nDone. Daily use is simply: inbox-sort")
|
|
return 0
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|