renamephotos-code (Python script for batch-renaming photo files)

renameall.py

"""Preview and journal photo renames without replacing existing destinations."""from __future__ import annotationsimport argparseimport ctypesfrom dataclasses import asdict, dataclassimport errnofrom functools import lru_cacheimport jsonimport osfrom pathlib import Pathfrom random import SystemRandomimport statimport stringimport sysfrom typing import Callable, TextIOfrom uuid import uuid4EXTENSIONS = {".jpg", ".jpeg", ".png"}ALPHABET = string.ascii_lowercase + string.digits@dataclass(frozen=True)class Rename:    source: str    destination: str    identity: tuple[int, int, int, int]@dataclass(frozen=True)class Plan:    folder: Path    entries: tuple[Rename, ...]def _identity(path: Path) -> tuple[int, int, int, int]:    info = path.lstat()    if not stat.S_ISREG(info.st_mode):        raise ValueError(f"Not a regular file: {path.name!r}")    return info.st_dev, info.st_ino, info.st_size, info.st_mtime_nsdef _new_stem() -> str:    return "".join(SystemRandom().sample(ALPHABET, 20))def build_plan(folder: str | Path) -> Plan:    root = Path(folder).expanduser().resolve(strict=True)    if not root.is_dir():        raise NotADirectoryError(str(root))    children = sorted(root.iterdir(), key=lambda path: (path.name.casefold(), path.name))    reserved = {path.name.casefold() for path in children}    entries = []    for source in children:        if source.suffix.lower() not in EXTENSIONS or not stat.S_ISREG(source.lstat().st_mode):            continue        identity = _identity(source)        for _ in range(100):            name = _new_stem() + source.suffix            if name.casefold() not in reserved:                reserved.add(name.casefold())                entries.append(Rename(source.name, name, identity))                break        else:            raise FileExistsError("Could not reserve a unique photo name; no files were renamed")    return Plan(root, tuple(entries))def _validate(plan: Plan) -> None:    if plan.folder.resolve(strict=True) != plan.folder or not plan.folder.is_dir():        raise ValueError("The selected folder changed after planning")    existing = {path.name.casefold() for path in plan.folder.iterdir()}    sources: set[str] = set()    targets: set[str] = set()    for entry in plan.entries:        for name in (entry.source, entry.destination):            if Path(name).name != name or name in {"", ".", ".."} or "/" in name or "\\" in name:                raise ValueError("Rename entries must stay directly inside the selected folder")        if entry.source.casefold() in sources or entry.destination.casefold() in targets:            raise ValueError("Duplicate entry in rename plan")        if entry.destination.casefold() in existing:            raise FileExistsError(f"Destination already exists: {entry.destination!r}")        if _identity(plan.folder / entry.source) != entry.identity:            raise ValueError(f"Source changed after planning: {entry.source!r}")        sources.add(entry.source.casefold())        targets.add(entry.destination.casefold())@lru_cache(maxsize=1)def _rename_operation() -> Callable[[Path, Path], None]:    if os.name == "nt":        # Python's Windows rename always refuses an existing destination.        return os.rename    library = ctypes.CDLL(None, use_errno=True)    if sys.platform.startswith("linux") and hasattr(library, "renameat2"):        function = library.renameat2        function.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]        function.restype = ctypes.c_int        def invoke(source: bytes, destination: bytes) -> int:            return function(-100, source, -100, destination, 1)  # AT_FDCWD, RENAME_NOREPLACE    elif sys.platform == "darwin" and hasattr(library, "renamex_np"):        function = library.renamex_np        function.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_uint]        function.restype = ctypes.c_int        def invoke(source: bytes, destination: bytes) -> int:            return function(source, destination, 4)  # RENAME_EXCL    else:        raise OSError(errno.ENOTSUP, "Atomic rename without replacement is unavailable on this platform")    def rename(source: Path, destination: Path) -> None:        if invoke(os.fsencode(source), os.fsencode(destination)) != 0:            code = ctypes.get_errno()            raise OSError(code, os.strerror(code), str(destination))    return renamedef _record(journal: TextIO, event: dict) -> None:    journal.write(json.dumps(event) + "\n")    journal.flush()    os.fsync(journal.fileno())def execute_plan(plan: Plan) -> Path | None:    """Stop on the first failure, retaining a prewritten mapping for recovery."""    _validate(plan)    if not plan.entries:        return None    rename = _rename_operation()    journal_path = plan.folder / f".renamephotos-{uuid4().hex}.jsonl"    descriptor = os.open(journal_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)    with os.fdopen(descriptor, "w", encoding="utf-8") as journal:        _record(journal, {"version": 1, "folder": str(plan.folder), "plan": [asdict(entry) for entry in plan.entries]})        for entry in plan.entries:            try:                if _identity(plan.folder / entry.source) != entry.identity:                    raise ValueError(f"Source changed after planning: {entry.source!r}")                rename(plan.folder / entry.source, plan.folder / entry.destination)                _record(journal, {"status": "renamed", "source": entry.source, "destination": entry.destination})            except (OSError, ValueError, KeyboardInterrupt) as exc:                try:                    _record(journal, {"status": "stopped", "source": entry.source, "error": type(exc).__name__})                except OSError:                    pass  # The complete mapping was flushed before the first rename.                raise        _record(journal, {"status": "complete", "renamed": len(plan.entries)})    return journal_pathdef main(argv: list[str] | None = None) -> int:    parser = argparse.ArgumentParser(description=__doc__)    parser.add_argument("folder", nargs="?", help="Folder containing the photos (not recursive)")    mode = parser.add_mutually_exclusive_group()    mode.add_argument("--dry-run", action="store_true", help="Preview without renaming or writing a journal")    mode.add_argument("--yes", action="store_true", help="Apply the displayed plan without a confirmation prompt")    args = parser.parse_args(argv)    try:        folder = args.folder if args.folder is not None else input("Where are the photos: ")        if not folder.strip():            raise ValueError("A folder path is required")        plan = build_plan(folder)        if not plan.entries:            print("No supported regular images found.")            return 0        for entry in plan.entries:            print(f"{entry.source!r} -> {entry.destination}")        if args.dry_run:            print(f"Preview only: {len(plan.entries)} files; nothing changed.")            return 0        if not args.yes and input(f"Rename {len(plan.entries)} files? [y/N] ").strip().lower() != "y":            print("Cancelled; nothing changed.")            return 0        journal = execute_plan(plan)        print(f"Renamed {len(plan.entries)} files. Keep the name mapping: {journal}")        return 0    except (OSError, ValueError, EOFError) as exc:        print(f"Stopped: {exc}. If execution began, keep the .renamephotos-*.jsonl mapping in that folder.", file=sys.stderr)        return 1    except KeyboardInterrupt:        print("Interrupted. Keep any .renamephotos-*.jsonl mapping before resuming.", file=sys.stderr)        return 130if __name__ == "__main__":    raise SystemExit(main())

requirements.txt

# Standard-library utility; no third-party packages are required.

STANDARDS.md

Repository StandardsThis document is the authoritative list of standards every repository in thisworkspace is expected to follow. New repositories created fromtheprawntemplate inherit the baseline automatically; this file records whatthat baseline is, what is required vs. optional, and the packaging pattern forcontainer repositories.It exists so that humans and automation agents working across the fleet have asingle source of truth to conform to, rather than relying on implicitper-repository conventions.Last reviewed: 2026-09-27.## ScopeApplies to all owned repositories (currently 79: 71 non-container, 7 container,plus one history-managed repository). "Container repository" means a repositorythat builds and publishes one or more container images to GitHub ContainerRegistry (GHCR, ghcr.io).## 1. Security scanning (REQUIRED for every repo)Every repository must carry these GitHub Actions workflows. They are present intheprawntemplate and propagate to new repositories:| Workflow | File | Purpose || --- | --- | --- || CodeQL | .github/workflows/codeql.yml | Static analysis / code scanning || Semgrep | .github/workflows/semgrep.yml | Pattern-based SAST || TruffleHog | .github/workflows/trufflehog.yml | Secret scanning || Dependency Review | .github/workflows/dependency-review.yml | Blocks vulnerable dependency changes in PRs || Scorecard | .github/workflows/scorecard.yml | OpenSSF supply-chain posture || LFS Guard | .github/workflows/lfs-guard.yml | Blocks accidental large/binary commits |A repository is security-compliant when all six are present and enabled.## 2. Automation and CI (REQUIRED)- .github/workflows/ci.yml ΓÇö project checks; adapt commands to the stack.- .github/dependabot.yml + .github/workflows/dependabot-auto-merge.yml and .github/workflows/auto-merge-bots.yml ΓÇö dependency updates with guarded auto-merge.- .github/workflows/labeler.yml, greetings.yml, heartbeat.yml, summary.yml ΓÇö repository hygiene and activity signals.- .github/workflows/deploy.yml ΓÇö deployment (or a clear statement that the repo is not deployed).## 3. Governance and hygiene files (REQUIRED unless noted)| File | Requirement || --- | --- || LICENSE | Required ΓÇö Apache-2.0. || NOTICE | Required ΓÇö accompanies Apache-2.0. || SECURITY.md | Required. Use GitHub private "Report a vulnerability"; never a personal email. || CONTRIBUTING.md | Required. || AGENTS.md | Required ΓÇö instructions for automation agents. || .github/dependabot.yml | Required. || .env.example | Required only when the project reads environment variables. Repos with no env config legitimately omit it. || .github/ISSUE_TEMPLATE/*, .github/pull_request_template.md | Required. || .gitignore, .gitattributes | Required. |## 4. Privacy and portability rules (REQUIRED)These reflect the fleet-wide remediation sweep and must hold for every repo:- No secrets in git. Use environment variables and .env.example. TruffleHog enforces this in CI; do not defeat it.- No personal contact details or private operator references in tracked files (READMEs, SECURITY.md, configs, .agents/). Vulnerability reporting goes through GitHub's private channel, not a personal email.- Windows/Linux portability. Launchers and paths must work on both; avoid hardcoded absolute paths and OS-specific-only launch scripts where a portable equivalent exists.- .agents/ hygiene. Never write secrets or personal details into .agents/STATE.md or .agents/JOURNAL.md.- PRs over direct commits to main. Changes land through reviewed pull requests.## 5. Container / GHCR packaging (REQUIRED for container repos; OPT-IN for others)Container repositories publish images to ghcr.io/<owner>/<repo>/<component>.The standard packaging pattern ΓÇö verified present in all seven current containerrepositories (musicstream, facetracker, FORGE, unifiedcollector, unifiedanalyzer,ticketremaster-b, ctfsolver) ΓÇö is:| File | Purpose || --- | --- || .github/workflows/docker-publish.yml | Builds and pushes image(s) to GHCR on the release trigger. || .github/scripts/ghcr-retention.cjs | Prunes old GHCR package versions to control storage. || Dockerfile (and variants) | Image build definition(s). |Opt-in scaffolding for these files lives under.github/optional/ so a new repository can adopt thepackaging pattern by copying them into place rather than hand-rolling each time.See .github/optional/README.md for enable steps.Package visibility: package visibility (private vs public) is governed by theorganization's package-creation policy (Org Settings ΓåÆ Packages). Individualcontainer package visibility is a deliberate, per-package decision ΓÇö especiallyfor privacy-sensitive images (photo/OSINT/security tooling) whose layers becomeworld-pullable when public.## 6. Compliance statusA live compliance matrix across all owned repositories is maintained out-of-band(see repo-standards-compliance-matrix.md in the workspace release artifacts).As of the last review, 76/79 repositories carry the full security-workflow set;the known exceptions are actively-developed repositories that have not yetadopted the baseline. All seven container repositories carry the complete GHCRpipeline.## How to use this document- Creating a new repo: start from theprawntemplate; you inherit sections 1ΓÇô4 automatically. If it will publish containers, adopt section 5 from .github/optional/.- Auditing an existing repo: check it against sections 1ΓÇô4 (all repos) and section 5 (container repos). Missing required items are compliance gaps to fix via a reviewed PR.- Agents: treat this file as the conformance target for any repository work.

tests__test_renameall.py

import contextlibfrom dataclasses import replaceimport ioimport jsonimport osfrom pathlib import Pathimport statimport subprocessimport sysimport tempfileimport unittestfrom unittest.mock import patchimport renameallclass RenameTests(unittest.TestCase):    def setUp(self) -> None:        self.temporary = tempfile.TemporaryDirectory(prefix="renamephotos-test-")        self.root = Path(self.temporary.name).resolve()        self.assertTrue(self.root.is_relative_to(Path(tempfile.gettempdir()).resolve()))    def tearDown(self) -> None:        self.assertTrue(self.root.is_relative_to(Path(tempfile.gettempdir()).resolve()))        self.temporary.cleanup()    def photo(self, name: str = "photo.jpg", content: bytes = b"synthetic-photo") -> Path:        path = self.root / name        path.write_bytes(content)        return path    def journal(self) -> list[dict]:        paths = list(self.root.glob(".renamephotos-*.jsonl"))        self.assertEqual(len(paths), 1)        return [json.loads(line) for line in paths[0].read_text(encoding="utf-8").splitlines()]    def test_import_does_not_prompt_or_touch_folder(self) -> None:        self.photo()        environment = dict(os.environ, PYTHONPATH=str(Path(renameall.__file__).parent))        result = subprocess.run([sys.executable, "-c", "import renameall"], cwd=self.root,                                env=environment, stdin=subprocess.DEVNULL, capture_output=True, timeout=20)        self.assertEqual(result.returncode, 0, result.stderr)        self.assertEqual(result.stdout, b"")        self.assertEqual([p.name for p in self.root.iterdir()], ["photo.jpg"])    def test_plan_accepts_mixed_case_extensions_and_skips_directories(self) -> None:        names = ["z.JPG", "a.jpeg", "b.PNG", "c.pNg", "d.JPEG", "e.jpg"]        for name in names:            self.photo(name)        self.photo("notes.txt")        (self.root / "album.jpg").mkdir()        before = sorted(p.name for p in self.root.iterdir())        plan = renameall.build_plan(self.root)        self.assertEqual([entry.source for entry in plan.entries], sorted(names, key=str.casefold))        for entry in plan.entries:            self.assertEqual(Path(entry.source).suffix, Path(entry.destination).suffix)            stem = Path(entry.destination).stem            self.assertEqual(len(stem), 20)            self.assertEqual(len(set(stem)), 20)            self.assertLessEqual(set(stem), set(renameall.ALPHABET))        self.assertEqual(sorted(p.name for p in self.root.iterdir()), before)    def test_symlinks_are_excluded(self) -> None:        target = self.photo("target.txt")        link = self.root / "linked.jpg"        try:            link.symlink_to(target)        except OSError as exc:            self.skipTest(f"Symlink creation unavailable: {exc}")        self.assertEqual(renameall.build_plan(self.root).entries, ())        self.assertTrue(link.is_symlink())    def test_names_are_reserved_case_insensitively_and_across_the_plan(self) -> None:        existing = "abcdefghijklmnopqrst"        first = "bcdefghijklmnopqrstu"        second = "cdefghijklmnopqrstuv"        (self.root / (existing.upper() + ".JPG")).mkdir()        self.photo("a.jpg")        self.photo("b.jpg")        with patch.object(renameall, "_new_stem", side_effect=[existing, first, first, second]):            plan = renameall.build_plan(self.root)        self.assertEqual([entry.destination for entry in plan.entries], [first + ".jpg", second + ".jpg"])    def test_exhausted_name_generation_changes_nothing(self) -> None:        stem = "abcdefghijklmnopqrst"        self.photo(stem + ".jpg")        with patch.object(renameall, "_new_stem", return_value=stem) as generator:            with self.assertRaises(FileExistsError):                renameall.build_plan(self.root)        self.assertEqual(generator.call_count, 100)        self.assertEqual([p.name for p in self.root.iterdir()], [stem + ".jpg"])    def test_execution_preserves_bytes_extensions_and_permissions(self) -> None:        photo = self.photo("family_日本.PNG", b"\x00\xfforiginal-photo-bytes")        mode = stat.S_IMODE(photo.stat().st_mode)        plan = renameall.build_plan(self.root)        journal_path = renameall.execute_plan(plan)        destination = self.root / plan.entries[0].destination        self.assertFalse(photo.exists())        self.assertEqual(destination.read_bytes(), b"\x00\xfforiginal-photo-bytes")        self.assertEqual(stat.S_IMODE(destination.stat().st_mode), mode)        records = self.journal()        self.assertEqual(records[0]["plan"][0]["source"], photo.name)        self.assertEqual(records[0]["plan"][0]["destination"], destination.name)        self.assertEqual(records[-1], {"status": "complete", "renamed": 1})        if os.name != "nt":            self.assertEqual(stat.S_IMODE(journal_path.stat().st_mode), 0o600)    def test_preexisting_destination_aborts_before_any_rename(self) -> None:        first = self.photo("a.jpg", b"first")        second = self.photo("b.jpg", b"second")        plan = renameall.build_plan(self.root)        occupied = self.root / plan.entries[1].destination        occupied.write_bytes(b"retained-destination")        with self.assertRaises(FileExistsError):            renameall.execute_plan(plan)        self.assertEqual(first.read_bytes(), b"first")        self.assertEqual(second.read_bytes(), b"second")        self.assertEqual(occupied.read_bytes(), b"retained-destination")        self.assertFalse(list(self.root.glob(".renamephotos-*.jsonl")))    def test_destination_created_after_preflight_is_not_overwritten(self) -> None:        source = self.photo()        plan = renameall.build_plan(self.root)        operation = renameall._rename_operation()        def racing_rename(src: Path, dst: Path) -> None:            dst.write_bytes(b"concurrent-destination")            operation(src, dst)        with patch.object(renameall, "_rename_operation", return_value=racing_rename):            with self.assertRaises(FileExistsError):                renameall.execute_plan(plan)        self.assertEqual(source.read_bytes(), b"synthetic-photo")        self.assertEqual((self.root / plan.entries[0].destination).read_bytes(), b"concurrent-destination")        self.assertEqual(self.journal()[-1]["status"], "stopped")    def test_changed_source_aborts_whole_plan_before_rename(self) -> None:        first = self.photo("a.jpg")        second = self.photo("b.jpg")        plan = renameall.build_plan(self.root)        second.write_bytes(b"changed-content")        with self.assertRaisesRegex(ValueError, "changed"):            renameall.execute_plan(plan)        self.assertTrue(first.exists())        self.assertEqual(second.read_bytes(), b"changed-content")        self.assertFalse(list(self.root.glob(".renamephotos-*.jsonl")))    def test_missing_source_aborts_before_rename(self) -> None:        source = self.photo()        plan = renameall.build_plan(self.root)        source.rename(self.root / "moved-by-someone-else.jpg")        with self.assertRaises(FileNotFoundError):            renameall.execute_plan(plan)        self.assertEqual((self.root / "moved-by-someone-else.jpg").read_bytes(), b"synthetic-photo")    def test_plan_cannot_escape_folder_or_repeat_entries(self) -> None:        self.photo()        plan = renameall.build_plan(self.root)        for target in ["../outside.jpg", "..\\outside.jpg", str(self.root.parent / "outside.jpg")]:            with self.subTest(target=target), self.assertRaises(ValueError):                renameall.execute_plan(replace(plan, entries=(replace(plan.entries[0], destination=target),)))        with self.assertRaises(ValueError):            renameall.execute_plan(replace(plan, entries=plan.entries * 2))        self.assertTrue((self.root / "photo.jpg").exists())    def test_mapping_is_flushed_before_any_rename(self) -> None:        self.photo()        plan = renameall.build_plan(self.root)        operation = renameall._rename_operation()        def inspect_mapping(src: Path, dst: Path) -> None:            self.assertEqual(self.journal()[0]["plan"][0]["destination"], dst.name)            operation(src, dst)        with patch.object(renameall, "_rename_operation", return_value=inspect_mapping):            renameall.execute_plan(plan)    def test_journal_sync_failure_prevents_first_rename(self) -> None:        source = self.photo()        plan = renameall.build_plan(self.root)        with patch.object(renameall.os, "fsync", side_effect=OSError("disk unavailable")):            with self.assertRaises(OSError):                renameall.execute_plan(plan)        self.assertEqual(source.read_bytes(), b"synthetic-photo")    def test_partial_failure_preserves_completed_mapping_and_remaining_sources(self) -> None:        self.photo("a.jpg", b"a")        self.photo("b.jpg", b"b")        self.photo("c.jpg", b"c")        plan = renameall.build_plan(self.root)        operation = renameall._rename_operation()        def fail_second(src: Path, dst: Path) -> None:            if src.name == "b.jpg":                raise PermissionError("fixture lock")            operation(src, dst)        with patch.object(renameall, "_rename_operation", return_value=fail_second):            with self.assertRaises(PermissionError):                renameall.execute_plan(plan)        self.assertEqual((self.root / plan.entries[0].destination).read_bytes(), b"a")        self.assertEqual((self.root / "b.jpg").read_bytes(), b"b")        self.assertEqual((self.root / "c.jpg").read_bytes(), b"c")        records = self.journal()        self.assertEqual(len(records[0]["plan"]), 3)        self.assertEqual([r.get("status") for r in records[1:]], ["renamed", "stopped"])    def test_result_sync_failure_stops_later_renames(self) -> None:        self.photo("a.jpg", b"a")        self.photo("b.jpg", b"b")        plan = renameall.build_plan(self.root)        with patch.object(renameall.os, "fsync", side_effect=[None, OSError("disk unavailable"), OSError("disk unavailable")]):            with self.assertRaises(OSError):                renameall.execute_plan(plan)        self.assertEqual((self.root / plan.entries[0].destination).read_bytes(), b"a")        self.assertEqual((self.root / "b.jpg").read_bytes(), b"b")        self.assertEqual(len(self.journal()[0]["plan"]), 2)    def test_interruption_after_native_rename_keeps_recovery_mapping(self) -> None:        self.photo()        plan = renameall.build_plan(self.root)        operation = renameall._rename_operation()        def interrupted(src: Path, dst: Path) -> None:            operation(src, dst)            raise KeyboardInterrupt        with patch.object(renameall, "_rename_operation", return_value=interrupted):            with self.assertRaises(KeyboardInterrupt):                renameall.execute_plan(plan)        self.assertEqual((self.root / plan.entries[0].destination).read_bytes(), b"synthetic-photo")        self.assertEqual(self.journal()[0]["plan"][0]["destination"], plan.entries[0].destination)    def test_cli_preview_and_decline_write_nothing(self) -> None:        self.photo()        for args in [[str(self.root), "--dry-run"], [str(self.root)]]:            with patch("builtins.input", return_value="n"), contextlib.redirect_stdout(io.StringIO()):                self.assertEqual(renameall.main(args), 0)            self.assertEqual([p.name for p in self.root.iterdir()], ["photo.jpg"])    def test_cli_yes_executes_and_empty_folder_is_noop(self) -> None:        with contextlib.redirect_stdout(io.StringIO()):            self.assertEqual(renameall.main([str(self.root), "--yes"]), 0)            self.assertEqual(list(self.root.iterdir()), [])            self.photo()            self.assertEqual(renameall.main([str(self.root), "--yes"]), 0)        self.assertEqual(len(list(self.root.glob("*.jpg"))), 1)        self.assertEqual(self.journal()[-1]["status"], "complete")    def test_cli_invalid_input_and_filesystem_failure_return_nonzero(self) -> None:        with contextlib.redirect_stderr(io.StringIO()), patch("builtins.input", return_value=""):            self.assertEqual(renameall.main([]), 1)            self.assertEqual(renameall.main([str(self.root / "missing")]), 1)        self.photo()        with patch.object(renameall, "_rename_operation", side_effect=OSError("unsupported filesystem")), \                contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):            self.assertEqual(renameall.main([str(self.root), "--yes"]), 1)        self.assertTrue((self.root / "photo.jpg").exists())if __name__ == "__main__":    unittest.main()