#!/usr/bin/env python3 """Sort PDFs into the Dewey decimal folder skeleton using the Claude API. Usage: python3 sort_pdfs.py # classify loose PDFs (library root + category roots) python3 sort_pdfs.py --all # re-verify every PDF in the library python3 sort_pdfs.py some/folder # classify PDFs under a specific folder python3 sort_pdfs.py --dry-run # show the plan, move nothing Requires: pip install anthropic; poppler-utils (pdftotext/pdfinfo); ANTHROPIC_API_KEY in the environment (or an `ant auth login` profile). Every move is appended to sort_moves_.csv in the library root so it can be undone. Exact-duplicate collisions are quarantined in _duplicates_to_delete/. """ import argparse import csv import hashlib import json import subprocess import sys from datetime import date from pathlib import Path import anthropic LIBRARY = Path("/mnt/library/Public") MODEL = "claude-opus-5" BATCH_SIZE = 20 EXCLUDE_DIRS = {"@eaDir", "_duplicates_to_delete"} QUARANTINE = LIBRARY / "_duplicates_to_delete" INSTRUCTIONS = """You classify PDFs from a personal digital library into an existing Dewey-decimal folder skeleton (for the owner's website library page). You will receive a numbered list of PDFs, each with its filename, current folder, PDF metadata title, and a snippet of first-page text. For each one, pick the single best destination folder from the skeleton list provided. You MUST pick destinations exactly as written in the skeleton — never invent a folder. The current location is where the owner roughly dropped the file; it may be the right category but unsorted, or the wrong category entirely. Owner's conventions: - Synthesizers, keyboards, samplers, MIDI gear, music software -> 700_Arts/786.* and 781.* - A/V receivers, hi-fi, recorders, speakers, pro-audio -> 600_Technology/621.389_Audio_Engineering - Electronics datasheets/app notes -> 600_Technology/621.380_Electronics_General or 621.3817_Electronic_Components - Programming/OS/networking docs -> 000_Computer_Science/004.* and 005.* - Company reports, budgets, finance -> 300_Social_Sciences/33* - Choose the most specific folder that clearly fits; fall back to a general xxx.000 folder only when nothing specific fits. Confidence: high = clearly correct; medium = reasonable judgment call; low = guessing from thin evidence (the owner will review these).""" OUTPUT_SCHEMA = { "type": "object", "properties": { "results": { "type": "array", "items": { "type": "object", "properties": { "index": {"type": "integer"}, "destination": {"type": "string"}, "confidence": {"type": "string", "enum": ["high", "medium", "low"]}, "reason": {"type": "string"}, }, "required": ["index", "destination", "confidence", "reason"], "additionalProperties": False, }, } }, "required": ["results"], "additionalProperties": False, } def skeleton_dirs() -> list[str]: dirs = [] for top in sorted(LIBRARY.iterdir()): if top.is_dir() and top.name[:1].isdigit(): for sub in sorted(top.rglob("*")): if sub.is_dir() and not (set(sub.parts) & EXCLUDE_DIRS): dirs.append(str(sub.relative_to(LIBRARY))) return dirs def find_pdfs(scope: str, paths: list[Path]) -> list[Path]: def wanted(p: Path) -> bool: return p.suffix.lower() == ".pdf" and not (set(p.parts) & EXCLUDE_DIRS) if paths: found = [] for base in paths: found += [p for p in base.rglob("*") if wanted(p)] if base.is_dir() else [base] return sorted(found) if scope == "all": return sorted(p for p in LIBRARY.rglob("*") if wanted(p)) # default: loose files at the library root or directly inside a top-level category loose = [p for p in LIBRARY.glob("*.pdf") if wanted(p)] for top in LIBRARY.iterdir(): if top.is_dir() and top.name[:1].isdigit(): loose += [p for p in top.glob("*.pdf") if wanted(p)] return sorted(loose) def run(cmd: list[str], timeout: int = 25) -> str: try: return subprocess.run( cmd, capture_output=True, text=True, timeout=timeout, errors="replace" ).stdout except subprocess.TimeoutExpired: return "" def extract(pdf: Path) -> dict: title = "" for line in run(["pdfinfo", str(pdf)]).splitlines(): if line.startswith("Title:"): title = line.split(":", 1)[1].strip() break text = run(["pdftotext", "-f", "1", "-l", "2", "-q", str(pdf), "-"]) snippet = " ".join(text.split())[:400] return {"path": pdf, "title": title, "snippet": snippet} def classify_batch(client, system_blocks, items: list[dict]) -> list[dict]: lines = [] for i, it in enumerate(items): rel = it["path"].relative_to(LIBRARY) lines.append( f"[{i}] filename: {it['path'].name}\n" f" current folder: {rel.parent}\n" f" pdf title: {it['title'] or '(none)'}\n" f" first-page text: {it['snippet'] or '(no extractable text)'}" ) response = client.messages.create( model=MODEL, max_tokens=8000, system=system_blocks, output_config={ "effort": "medium", "format": {"type": "json_schema", "schema": OUTPUT_SCHEMA}, }, messages=[{"role": "user", "content": "Classify these PDFs:\n\n" + "\n".join(lines)}], ) if response.stop_reason == "refusal": print(" ! model declined this batch; leaving those files in place", file=sys.stderr) return [None] * len(items) text = next(b.text for b in response.content if b.type == "text") results = {r["index"]: r for r in json.loads(text)["results"]} return [results.get(i) for i in range(len(items))] def md5(path: Path) -> str: h = hashlib.md5() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): h.update(chunk) return h.hexdigest() def main() -> None: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("paths", nargs="*", type=Path, help="specific files/folders to sort") ap.add_argument("--all", action="store_true", help="re-verify every PDF in the library") ap.add_argument("--dry-run", action="store_true", help="print the plan without moving") args = ap.parse_args() skeleton = skeleton_dirs() valid = set(skeleton) pdfs = find_pdfs("all" if args.all else "loose", args.paths) if not pdfs: print("Nothing to sort.") return print(f"{len(pdfs)} PDFs to classify against {len(skeleton)} folders.") client = anthropic.Anthropic() # Skeleton goes in the system prompt with a cache breakpoint: after the first # request each later batch reads it from cache at ~10% of input price. system_blocks = [ {"type": "text", "text": INSTRUCTIONS}, { "type": "text", "text": "Valid destination folders (relative to the library root):\n" + "\n".join(skeleton), "cache_control": {"type": "ephemeral"}, }, ] log_path = LIBRARY / f"sort_moves_{date.today().isoformat()}.csv" new_log = not log_path.exists() moved = kept = flagged = 0 with open(log_path, "a", newline="") as logf: log = csv.writer(logf) if new_log and not args.dry_run: log.writerow(["old_path", "new_path", "confidence", "reason"]) for start in range(0, len(pdfs), BATCH_SIZE): batch = [extract(p) for p in pdfs[start : start + BATCH_SIZE]] print(f"Classifying {start + 1}-{start + len(batch)} of {len(pdfs)}...") for item, result in zip(batch, classify_batch(client, system_blocks, batch)): src = item["path"] rel_parent = str(src.relative_to(LIBRARY).parent) if result is None or result["destination"] not in valid: print(f" ? unresolved, left in place: {src.name}") flagged += 1 continue dest_dir = result["destination"] if dest_dir == rel_parent: kept += 1 continue target = LIBRARY / dest_dir / src.name tag = f"[{result['confidence']}] {src.name} -> {dest_dir} ({result['reason']})" if args.dry_run: print(f" DRY {tag}") continue if target.exists(): if md5(src) == md5(target): QUARANTINE.mkdir(exist_ok=True) qtarget = QUARANTINE / src.name src.rename(qtarget) log.writerow([src, qtarget, "high", "exact duplicate quarantined"]) print(f" = duplicate quarantined: {src.name}") else: print(f" ! name collision (different content), left in place: {src.name}") flagged += 1 continue src.rename(target) log.writerow([src, target, result["confidence"], result["reason"]]) moved += 1 print(f" > {tag}") print(f"\nDone. Moved {moved}, already correct {kept}, needs your review {flagged}.") if not args.dry_run: print(f"Undo log: {log_path}") if __name__ == "__main__": main()