File Organization skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "File Organization"
  short_description: "Rename and dedupe thousands of files, reversibly"
  default_prompt: "$file-organization Clean up my Downloads folder and sort everything by file type"

#!/usr/bin/env python3
"""Apply a reviewed rename/move plan from a CSV with columns: src,dst.

Dry run by default. With --apply it moves files, refuses to overwrite anything, creates
parent folders, and appends every completed move to an undo log so the batch can be
reversed with --undo.

Usage:
    python3 apply_plan.py plan.csv                 # validate and show what would happen
    python3 apply_plan.py plan.csv --apply         # do it, writing plan.csv.undo.csv
    python3 apply_plan.py plan.csv.undo.csv --apply # reverse a previous batch
Exit codes: 0 = ok, 1 = some moves failed, 2 = plan invalid (nothing was moved).
"""
from __future__ import annotations

import argparse
import csv
import os
import shutil
import sys
import unicodedata
from pathlib import Path


def norm(p: Path) -> str:
    # Case- and Unicode-insensitive key: catches collisions on macOS/Windows volumes.
    return unicodedata.normalize("NFC", str(p.resolve(strict=False))).casefold()


def load(plan: Path) -> list[tuple[Path, Path]]:
    with plan.open(newline="", encoding="utf-8") as f:
        rows = list(csv.DictReader(f))
    if rows and not {"src", "dst"} <= set(rows[0]):
        raise SystemExit("plan must have src,dst columns")
    return [(Path(r["src"]), Path(r["dst"])) for r in rows if r["src"] != r["dst"]]


def validate(moves: list[tuple[Path, Path]]) -> list[str]:
    problems: list[str] = []
    sources = {norm(s) for s, _ in moves}
    targets: dict[str, Path] = {}
    for src, dst in moves:
        if not src.exists():
            problems.append(f"missing source: {src}")
        key = norm(dst)
        if key in targets:
            problems.append(f"two files map to {dst} (also {targets[key]})")
        targets[key] = src
        case_only = norm(src) == key
        if dst.exists() and not case_only and key not in sources:
            problems.append(f"target exists: {dst}")
        if key in sources and not case_only:
            problems.append(f"target {dst} is also a source in this plan; split into two passes")
    return problems


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Apply a src,dst move plan safely.")
    ap.add_argument("plan", type=Path)
    ap.add_argument("--apply", action="store_true")
    args = ap.parse_args(argv)

    moves = load(args.plan)
    problems = validate(moves)
    if problems:
        print(f"plan invalid, nothing moved ({len(problems)} problems):")
        for p in problems[:50]:
            print(f"  {p}")
        return 2
    if not args.apply:
        for src, dst in moves[:30]:
            print(f"WOULD MOVE {src} -> {dst}")
        print(f"{len(moves)} moves validated (dry run). Re-run with --apply.")
        return 0

    undo_path = args.plan.with_name(args.plan.name + ".undo.csv")
    failed = 0
    with undo_path.open("a", newline="", encoding="utf-8") as uf:
        undo = csv.writer(uf)
        if uf.tell() == 0:
            undo.writerow(["src", "dst"])
        for src, dst in moves:
            try:
                dst.parent.mkdir(parents=True, exist_ok=True)
                if norm(src) == norm(dst):                 # case-only rename
                    tmp = src.with_name(src.name + ".renaming")
                    os.rename(src, tmp)
                    os.rename(tmp, dst)
                else:
                    if dst.exists():
                        raise FileExistsError(dst)
                    shutil.move(str(src), str(dst))
                undo.writerow([str(dst), str(src)])
                uf.flush()
            except OSError as exc:
                failed += 1
                print(f"FAILED {src} -> {dst}: {exc}", file=sys.stderr)
    print(f"moved {len(moves) - failed}/{len(moves)}; undo log: {undo_path}")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Report duplicate files under one or more folders. Read-only: never deletes anything.

Groups files by size, then by a 64 KiB head hash, then by full SHA-256, so most files
are never read in full. Writes a CSV with one row per file in a duplicate group.

Usage:
    python3 find_duplicates.py DIR [DIR ...] [--min-size BYTES] [--csv dupes.csv]
Exit codes: 0 = no duplicates, 1 = duplicates found, 2 = bad arguments.
"""
from __future__ import annotations

import argparse
import csv
import hashlib
import os
import sys
from collections import defaultdict
from pathlib import Path

CHUNK = 1 << 20
HEAD = 64 * 1024
SKIP_NAMES = {".DS_Store", "Thumbs.db", "desktop.ini"}


def digest(path: Path, limit: int | None = None) -> str:
    h = hashlib.sha256()
    read = 0
    with path.open("rb") as f:
        while chunk := f.read(CHUNK if limit is None else min(CHUNK, limit - read)):
            h.update(chunk)
            read += len(chunk)
            if limit is not None and read >= limit:
                break
    return h.hexdigest()


def walk(roots: list[Path], min_size: int):
    seen_inodes: set[tuple[int, int]] = set()
    for root in roots:
        for dirpath, dirnames, filenames in os.walk(root):
            dirnames[:] = [d for d in dirnames if not d.startswith(".")]
            for name in filenames:
                if name in SKIP_NAMES:
                    continue
                p = Path(dirpath) / name
                try:
                    st = p.lstat()
                except OSError as exc:
                    print(f"skip {p}: {exc}", file=sys.stderr)
                    continue
                if not p.is_file() or p.is_symlink() or st.st_size < min_size:
                    continue
                key = (st.st_dev, st.st_ino)
                if key in seen_inodes:          # hard link to a file already counted
                    continue
                seen_inodes.add(key)
                yield p, st.st_size


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Report duplicate files (read-only).")
    ap.add_argument("dirs", nargs="+", type=Path)
    ap.add_argument("--min-size", type=int, default=1, help="ignore files smaller than this (bytes)")
    ap.add_argument("--csv", type=Path, help="write groups to this CSV file")
    args = ap.parse_args(argv)
    for d in args.dirs:
        if not d.is_dir():
            print(f"not a directory: {d}", file=sys.stderr)
            return 2

    by_size: dict[int, list[Path]] = defaultdict(list)
    total = 0
    for p, size in walk(args.dirs, args.min_size):
        by_size[size].append(p)
        total += 1

    groups: list[tuple[int, str, list[Path]]] = []
    for size, paths in by_size.items():
        if len(paths) < 2:
            continue
        by_head: dict[str, list[Path]] = defaultdict(list)
        for p in paths:
            by_head[digest(p, HEAD)].append(p)
        for head_paths in by_head.values():
            if len(head_paths) < 2:
                continue
            by_full: dict[str, list[Path]] = defaultdict(list)
            for p in head_paths:
                by_full[digest(p) if size > HEAD else digest(p, HEAD)].append(p)
            for h, same in by_full.items():
                if len(same) > 1:
                    groups.append((size, h, sorted(same)))

    groups.sort(key=lambda g: g[0] * (len(g[2]) - 1), reverse=True)
    wasted = sum(size * (len(paths) - 1) for size, _, paths in groups)
    print(f"scanned {total} files; {len(groups)} duplicate groups; "
          f"{sum(len(g[2]) - 1 for g in groups)} redundant copies; {wasted / 1e6:.1f} MB reclaimable")
    for size, h, paths in groups[:20]:
        print(f"\n{size} bytes  sha256 {h[:12]}")
        for p in paths:
            print(f"  {p}")
    if len(groups) > 20:
        print(f"\n... {len(groups) - 20} more groups (see CSV)")

    if args.csv:
        with args.csv.open("w", newline="", encoding="utf-8") as f:
            w = csv.writer(f)
            w.writerow(["group", "sha256", "size", "path", "mtime"])
            for i, (size, h, paths) in enumerate(groups, start=1):
                for p in paths:
                    w.writerow([i, h, size, str(p), int(p.stat().st_mtime)])
        print(f"\nwrote {args.csv}")
    return 1 if groups else 0


if __name__ == "__main__":
    sys.exit(main())