FileToFolderMover.py

"""Mass folder creator / file organizer.

For every file with a chosen extension inside a folder (searched
recursively), create a same-named subfolder and move the file into it,
along with any sidecar files that share the same name (for example a
.srt or .nfo file next to a .mkv). Handy for tidying up flat folders,
such as a movie library with one file (or file group) per title.

Every real run writes a manifest of the moves it made. Pass the
manifest to --undo to reverse a run.

Usage (non-interactive):
    python FileToFolderMover.py --folder ./Movies --extension .mkv
    python FileToFolderMover.py --folder ./Movies --extension .mkv,.mp4
    python FileToFolderMover.py --folder ./Movies --extension .mkv --exclude "*.sample.*"
    python FileToFolderMover.py --folder ./Movies --extension .mkv --min-age-days 1
    python FileToFolderMover.py --folder ./Movies --extension .mkv --dry-run
    python FileToFolderMover.py --folder ./Movies --extension .mkv --watch 3600 --yes
    python FileToFolderMover.py --undo ./Movies/.mass-folder-creator-manifest-20260804-120000.json
"""

from __future__ import annotations

import argparse
import fnmatch
import json
import logging
import shutil
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Set

log = logging.getLogger("mass-folder-creator")

SECONDS_PER_DAY = 86400

# Folders that are never a legitimate target for this script, even
# though it only moves files rather than deleting them. Resolved
# case-insensitively on Windows.
DANGEROUS_NAMES = {
    "windows", "system32", "program files", "program files (x86)",
    "programdata", "boot",
    "etc", "usr", "bin", "sbin", "lib", "lib64", "var", "opt",
    "system", "library",
}


def configure_logging(verbose: bool, log_file: Optional[str]) -> None:
    log.setLevel(logging.DEBUG if verbose else logging.INFO)
    log.handlers.clear()

    console = logging.StreamHandler()
    console.setLevel(logging.DEBUG if verbose else logging.INFO)
    console.setFormatter(logging.Formatter("%(message)s"))
    log.addHandler(console)

    if log_file:
        file_handler = logging.FileHandler(log_file, encoding="utf-8")
        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(
            logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
        )
        log.addHandler(file_handler)


def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument("--folder", help="Folder to organize")
    parser.add_argument(
        "--extension",
        help="File extension(s) to organize, comma-separated, e.g. .mkv,.mp4",
    )
    parser.add_argument(
        "--exclude",
        action="append",
        default=[],
        help="Glob pattern to exclude (matched against filename and path "
        "segments). Can be passed multiple times.",
    )
    parser.add_argument(
        "--min-age-days",
        type=float,
        default=None,
        help="Only organize files last modified at least this many days ago",
    )
    parser.add_argument(
        "--follow-symlinks",
        action="store_true",
        help="Include symlinks (skipped by default for safety)",
    )
    parser.add_argument(
        "--force",
        action="store_true",
        help="Allow running against a folder that looks like a system/home directory",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Show what would be moved without moving anything",
    )
    parser.add_argument(
        "--yes", action="store_true", help="Skip the confirmation prompt"
    )
    parser.add_argument(
        "--watch",
        type=float,
        default=None,
        metavar="SECONDS",
        help="Run repeatedly, waiting this many seconds between runs, "
        "until Ctrl+C. Requires --yes or --dry-run.",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Show debug-level detail, including files skipped by "
        "--exclude/--min-age-days/symlink filtering",
    )
    parser.add_argument(
        "--log-file",
        metavar="PATH",
        help="Also write a timestamped log of this run to PATH",
    )
    parser.add_argument(
        "--manifest",
        metavar="PATH",
        help="Where to write the move manifest (default: a timestamped "
        "file inside the target folder)",
    )
    parser.add_argument(
        "--undo",
        metavar="MANIFEST_PATH",
        help="Reverse a previous run using its manifest file, instead of organizing",
    )
    args = parser.parse_args(argv)

    if args.watch is not None and not (args.yes or args.dry_run):
        parser.error("--watch requires --yes or --dry-run (it runs unattended)")
    if args.watch is not None and args.undo:
        parser.error("--watch cannot be combined with --undo")

    return args


def prompt_for_folder() -> Path:
    while True:
        raw = input("Paste or type the folder path: ").strip()
        folder = Path(raw)
        if folder.is_dir():
            return folder
        print("This is not a valid folder path! Try again.")


def dangerous_path_reason(folder: Path) -> Optional[str]:
    """Return a reason string if the folder looks unsafe to run against."""
    resolved = folder.resolve()

    if resolved.parent == resolved:
        return f"{resolved} is a filesystem root"

    try:
        if resolved == Path.home().resolve():
            return f"{resolved} is your home directory"
    except RuntimeError:
        pass

    if resolved.name.lower() in DANGEROUS_NAMES:
        return f"{resolved} looks like a system directory"

    return None


def _is_excluded(path: Path, root: Path, exclude_patterns: List[str]) -> bool:
    rel_parts = path.relative_to(root).parts
    return any(
        fnmatch.fnmatch(path.name, pat) or any(fnmatch.fnmatch(part, pat) for part in rel_parts)
        for pat in exclude_patterns
    )


def find_matching_files(
    folder: Path,
    extensions: List[str],
    exclude_patterns: List[str],
    follow_symlinks: bool,
    min_age_days: Optional[float],
) -> List[Path]:
    """Find primary files (recursively) matching any of the given extensions."""
    matches: Set[Path] = set()
    for extension in extensions:
        pattern = f"*{extension}"
        matches.update(p for p in folder.rglob(pattern) if p.is_file() or p.is_symlink())

    skipped_symlinks = 0
    cutoff = time.time() - min_age_days * SECONDS_PER_DAY if min_age_days else None

    results = []
    for path in matches:
        if path.is_symlink() and not follow_symlinks:
            skipped_symlinks += 1
            log.debug("Skipping symlink: %s", path)
            continue

        if _is_excluded(path, folder, exclude_patterns):
            log.debug("Excluded by --exclude: %s", path)
            continue

        if cutoff is not None and path.stat().st_mtime > cutoff:
            log.debug("Too recent for --min-age-days: %s", path)
            continue

        results.append(path)

    if skipped_symlinks:
        log.info(
            "Skipping %d symlink(s); use --follow-symlinks to include them.",
            skipped_symlinks,
        )

    return sorted(results)


def find_sidecars(
    primary: Path,
    extensions_lower: Set[str],
    root: Path,
    exclude_patterns: List[str],
    follow_symlinks: bool,
) -> List[Path]:
    """Find files next to `primary` that share its name but not its extension.

    A file is only ever a sidecar if its own extension is not one of the
    tracked extensions; otherwise it is a primary in its own right.
    """
    stem = primary.stem
    sidecars = []
    for sibling in sorted(primary.parent.iterdir()):
        if sibling == primary or sibling.stem != stem:
            continue
        if not (sibling.is_file() or sibling.is_symlink()):
            continue
        if sibling.suffix.lower() in extensions_lower:
            continue
        if sibling.is_symlink() and not follow_symlinks:
            continue
        if _is_excluded(sibling, root, exclude_patterns):
            continue
        sidecars.append(sibling)
    return sidecars


def unique_destination(folder: Path, name: str) -> Path:
    """Avoid clobbering an existing folder by adding a numeric suffix."""
    candidate = folder / name
    counter = 1
    while candidate.exists():
        candidate = folder / f"{name} ({counter})"
        counter += 1
    return candidate


def perform_moves(
    matches: List[Path],
    extensions_lower: Set[str],
    root: Path,
    exclude_patterns: List[str],
    follow_symlinks: bool,
) -> tuple[int, int, List[Dict[str, str]], List[str]]:
    moved, failed = 0, 0
    manifest_moves: List[Dict[str, str]] = []
    created_dirs: List[str] = []

    for primary in matches:
        sidecars = find_sidecars(primary, extensions_lower, root, exclude_patterns, follow_symlinks)
        dest_dir = unique_destination(primary.parent, primary.stem)
        try:
            dest_dir.mkdir()
        except OSError as exc:
            log.error("Could not create folder %s: %s", dest_dir, exc)
            failed += 1 + len(sidecars)
            continue

        created_dirs.append(str(dest_dir))
        for member, label in [(primary, "primary")] + [(s, "sidecar") for s in sidecars]:
            destination = dest_dir / member.name
            try:
                shutil.move(str(member), str(destination))
                manifest_moves.append({"from": str(member), "to": str(destination)})
                moved += 1
                log.info("Moved (%s): %s -> %s", label, member, destination)
            except OSError as exc:
                failed += 1
                log.error("Could not move %s: %s", member, exc)

    return moved, failed, manifest_moves, created_dirs


def write_manifest(path: Path, folder: Path, moves: List[Dict[str, str]], created_dirs: List[str]) -> None:
    data = {
        "created_at": datetime.now().isoformat(timespec="seconds"),
        "folder": str(folder),
        "moves": moves,
        "created_dirs": created_dirs,
    }
    path.write_text(json.dumps(data, indent=2), encoding="utf-8")


def run_once(args: argparse.Namespace, extensions: List[str], folder: Path) -> None:
    extensions_lower = {e.lower() for e in extensions}
    matches = find_matching_files(
        folder, extensions, args.exclude, args.follow_symlinks, args.min_age_days
    )
    if not matches:
        log.info(
            "There are no files matching %s in that folder.", ", ".join(extensions)
        )
        return

    log.info("There are %d file(s) matching %s.", len(matches), ", ".join(extensions))
    for primary in matches:
        log.info("  %s", primary)
        for sidecar in find_sidecars(primary, extensions_lower, folder, args.exclude, args.follow_symlinks):
            log.info("    + %s (sidecar)", sidecar)

    if args.dry_run:
        log.info("Dry run: nothing was moved.")
        return

    if not args.yes:
        answer = input("Do you want to proceed? y/n: ").strip().lower()
        if not answer.startswith("y"):
            log.info("Bye!")
            return

    moved, failed, manifest_moves, created_dirs = perform_moves(
        matches, extensions_lower, folder, args.exclude, args.follow_symlinks
    )

    if manifest_moves:
        manifest_path = Path(args.manifest) if args.manifest else folder / (
            f".mass-folder-creator-manifest-{datetime.now().strftime('%Y%m%d-%H%M%S-%f')}.json"
        )
        write_manifest(manifest_path, folder, manifest_moves, created_dirs)
        log.info("Manifest saved to %s. Use --undo %s to reverse this run.", manifest_path, manifest_path)

    log.info("Done. Moved %d file(s), %d failure(s).", moved, failed)


def run_undo(args: argparse.Namespace) -> None:
    manifest_path = Path(args.undo)
    if not manifest_path.is_file():
        log.error("Manifest not found: %s", manifest_path)
        sys.exit(1)

    data = json.loads(manifest_path.read_text(encoding="utf-8"))
    moves = data.get("moves", [])
    created_dirs = data.get("created_dirs", [])

    if not moves:
        log.info("Nothing to undo: manifest has no recorded moves.")
        return

    log.info("This will reverse %d move(s) from %s.", len(moves), manifest_path)
    for move in moves:
        log.info("  %s -> %s", move["to"], move["from"])

    if args.dry_run:
        log.info("Dry run: nothing was moved.")
        return

    if not args.yes:
        answer = input(f"Reverse these {len(moves)} move(s)? y/n: ").strip().lower()
        if not answer.startswith("y"):
            log.info("Bye!")
            return

    restored, failed = 0, 0
    for move in reversed(moves):
        current = Path(move["to"])
        original = Path(move["from"])
        if not current.exists():
            log.error("Skipping, not found: %s", current)
            failed += 1
            continue
        if original.exists():
            log.error("Skipping, would overwrite existing file: %s", original)
            failed += 1
            continue
        try:
            original.parent.mkdir(parents=True, exist_ok=True)
            shutil.move(str(current), str(original))
            restored += 1
            log.info("Restored: %s -> %s", current, original)
        except OSError as exc:
            failed += 1
            log.error("Could not restore %s: %s", current, exc)

    removed_dirs = 0
    for dir_str in created_dirs:
        dir_path = Path(dir_str)
        try:
            dir_path.rmdir()
            removed_dirs += 1
            log.info("Removed empty folder: %s", dir_path)
        except OSError:
            log.debug("Left in place (not empty or already gone): %s", dir_path)

    log.info(
        "Undo done. Restored %d file(s), %d failure(s), removed %d folder(s).",
        restored, failed, removed_dirs,
    )


def main(argv: Optional[List[str]] = None) -> None:
    args = parse_args(argv)
    configure_logging(args.verbose, args.log_file)
    log.info("Mass folder creator")

    if args.undo:
        run_undo(args)
        return

    folder = Path(args.folder) if args.folder else None
    if folder is None or not folder.is_dir():
        if args.folder:
            log.error("Folder does not exist: %s", args.folder)
        folder = prompt_for_folder()

    reason = dangerous_path_reason(folder)
    if reason and not args.force:
        log.error("Refusing to run: %s. Pass --force to override.", reason)
        sys.exit(1)

    extension_raw = args.extension or input(
        "Type the file extension(s) you want to make folders of, comma-separated (i.e. .txt): "
    )
    extensions = [
        e if e.startswith(".") else f".{e}"
        for e in (e.strip() for e in extension_raw.split(","))
        if e
    ]
    if not extensions:
        log.error("No file extension provided.")
        sys.exit(1)

    if args.watch is None:
        run_once(args, extensions, folder)
        return

    log.info("Watch mode: running every %.0fs. Press Ctrl+C to stop.", args.watch)
    try:
        while True:
            run_once(args, extensions, folder)
            time.sleep(args.watch)
    except KeyboardInterrupt:
        log.info("Stopped.")


if __name__ == "__main__":
    main()