Refactor shared script helpers
This commit is contained in:
ajp_anton
+22
-99
@@ -6,137 +6,60 @@ Drag files or directories onto this script, or run:
|
||||
python3 grouping.py path/to/file/or/directory [...]
|
||||
|
||||
Files are sorted by their current path and moved into one folder named
|
||||
[first stem]-[last stem]. Multiple files/folders are grouped as selected.
|
||||
[first stem]-[last stem] inside the working directory.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ORDERED_TIMESTAMP_STEM_RE = re.compile(r"^(?P<timestamp>\d{8}_\d{6})-[1-9]\d*$")
|
||||
from tools.console import pause_if_interactive
|
||||
from tools.filenames import group_stem
|
||||
from tools.filesystem import collect_files, determine_working_dir, unique_path
|
||||
|
||||
|
||||
def group_stem(path: Path) -> str:
|
||||
stem = path.stem
|
||||
match = ORDERED_TIMESTAMP_STEM_RE.match(stem)
|
||||
if match:
|
||||
return match.group("timestamp")
|
||||
return stem
|
||||
def collect_group_files(args: list[str]) -> list[Path]:
|
||||
return collect_files(args, sort_by_name=True)
|
||||
|
||||
|
||||
def unique_path(path: Path) -> Path:
|
||||
if not path.exists():
|
||||
return path
|
||||
|
||||
for index in range(1, 10000):
|
||||
candidate = path.with_name(f"{path.stem}-{index}{path.suffix}")
|
||||
if not candidate.exists():
|
||||
return candidate
|
||||
|
||||
raise RuntimeError(f"Could not find an available name for {path}")
|
||||
|
||||
|
||||
def existing_input_paths(args: list[str]) -> list[Path]:
|
||||
paths: list[Path] = []
|
||||
for arg in args:
|
||||
path = Path(arg).expanduser()
|
||||
if path.exists():
|
||||
paths.append(path.resolve())
|
||||
else:
|
||||
print(f"Skipping missing path: {arg}")
|
||||
return paths
|
||||
|
||||
|
||||
def resolve_single_directory(path: Path) -> list[Path]:
|
||||
current = path
|
||||
while current.is_dir():
|
||||
children = sorted(current.iterdir(), key=lambda item: item.name.lower())
|
||||
if not children:
|
||||
return []
|
||||
if len(children) == 1 and children[0].is_dir():
|
||||
current = children[0]
|
||||
continue
|
||||
return [child.resolve() for child in children]
|
||||
return [current.resolve()]
|
||||
|
||||
|
||||
def resolve_group_items(args: list[str]) -> list[Path]:
|
||||
paths = existing_input_paths(args)
|
||||
if len(paths) == 1:
|
||||
path = paths[0]
|
||||
if path.is_dir():
|
||||
return resolve_single_directory(path)
|
||||
return [path]
|
||||
return sorted(paths, key=lambda item: item.name.lower())
|
||||
|
||||
|
||||
def working_dir_for_items(items: list[Path]) -> Path:
|
||||
if len(items) == 1:
|
||||
item = items[0]
|
||||
return item if item.is_dir() else item.parent
|
||||
|
||||
parents = {item.parent for item in items}
|
||||
if len(parents) == 1:
|
||||
return next(iter(parents))
|
||||
|
||||
try:
|
||||
common = Path(os.path.commonpath([str(item) for item in items]))
|
||||
except ValueError as exc:
|
||||
raise RuntimeError("Input paths do not have a common ancestor.") from exc
|
||||
|
||||
if str(common) == common.anchor:
|
||||
raise RuntimeError("Input paths only share the filesystem root; choose items closer together.")
|
||||
return common
|
||||
|
||||
|
||||
def group_items(items: list[Path]) -> Path | None:
|
||||
if not items:
|
||||
def group_files(files: list[Path], working_dir: Path) -> Path | None:
|
||||
if not files:
|
||||
return None
|
||||
|
||||
sorted_items = sorted(items, key=lambda item: item.name.lower())
|
||||
first = group_stem(sorted_items[0])
|
||||
last = group_stem(sorted_items[-1])
|
||||
target_dir = unique_path(working_dir_for_items(sorted_items) / f"{first}-{last}")
|
||||
target_dir.mkdir(parents=True)
|
||||
sorted_files = sorted(files, key=lambda item: item.name.lower())
|
||||
first = group_stem(sorted_files[0])
|
||||
last = group_stem(sorted_files[-1])
|
||||
target_dir = working_dir / f"{first}-{last}"
|
||||
target_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
for source in sorted_items:
|
||||
for source in sorted_files:
|
||||
destination = unique_path(target_dir / source.name)
|
||||
shutil.move(str(source), str(destination))
|
||||
|
||||
return target_dir
|
||||
|
||||
|
||||
def pause_if_interactive() -> None:
|
||||
if sys.stdin.isatty():
|
||||
try:
|
||||
input("\nPress Enter to close...")
|
||||
except EOFError:
|
||||
pass
|
||||
|
||||
|
||||
def run(argv: list[str]) -> int:
|
||||
if len(argv) < 2:
|
||||
print("Usage: python3 grouping.py path/to/file/or/directory [...]")
|
||||
return 2
|
||||
|
||||
items = resolve_group_items(argv[1:])
|
||||
if not items:
|
||||
print("No items to group.")
|
||||
return 0
|
||||
|
||||
try:
|
||||
target_dir = group_items(items)
|
||||
working_dir = determine_working_dir(argv[1:])
|
||||
except RuntimeError as exc:
|
||||
print(exc)
|
||||
return 1
|
||||
|
||||
files = collect_group_files(argv[1:])
|
||||
if not files:
|
||||
print("No files to group.")
|
||||
return 0
|
||||
|
||||
target_dir = group_files(files, working_dir)
|
||||
if target_dir is not None:
|
||||
print(f"Moved {len(items)} item(s) into {target_dir}")
|
||||
print(f"Moved {len(files)} file(s) into {target_dir}")
|
||||
return 0
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user