diff --git a/caya-clean/.gitignore b/caya-clean/.gitignore new file mode 100644 index 0000000..ed8ebf5 --- /dev/null +++ b/caya-clean/.gitignore @@ -0,0 +1 @@ +__pycache__ \ No newline at end of file diff --git a/caya-clean/caya-clean.py b/caya-clean/caya_clean.py similarity index 58% rename from caya-clean/caya-clean.py rename to caya-clean/caya_clean.py index 699f682..22cd2b3 100644 --- a/caya-clean/caya-clean.py +++ b/caya-clean/caya_clean.py @@ -1,14 +1,11 @@ +import argparse import hashlib import json import re -import sys from datetime import datetime from pathlib import Path -# Set to False after checking the output of a test run. -DRY_RUN = True - SCRIPT_DIR = Path(__file__).resolve().parent CONFIG_FILE = SCRIPT_DIR / "configuration.json" LOG_FILENAME = "caya-clean.log" @@ -19,6 +16,30 @@ DATE_PATTERN = re.compile(r"^(?P\d{8})T\d{4}Z\s+") CAYA_ID_PATTERN = re.compile(r"\s+[a-z0-9]{20,35}$", re.IGNORECASE) +def parse_arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Clean up and rename downloaded Caya letter files." + ) + ) + + parser.add_argument( + "folder", + nargs="?", + type=Path, + default=Path.cwd(), + help="Folder containing the letter files.", + ) + + parser.add_argument( + "--dry-run", + action="store_true", + help="Show planned changes without modifying or deleting files.", + ) + + return parser.parse_args() + + def log_message(folder: Path, message: str): """Append a timestamped message to the folder logfile.""" log_file = folder / LOG_FILENAME @@ -31,31 +52,44 @@ def log_message(folder: Path, message: str): def load_config() -> dict: - """Load the configuration from configuration.json.""" + """Load and validate configuration.json.""" if not CONFIG_FILE.exists(): - print(f"Configuration file not found: {CONFIG_FILE}") - sys.exit(1) + raise FileNotFoundError( + f"Configuration file not found: {CONFIG_FILE}" + ) try: with CONFIG_FILE.open("r", encoding="utf-8") as file: config = json.load(file) except json.JSONDecodeError as error: - print(f"Invalid JSON in configuration file: {error}") - sys.exit(1) + raise ValueError( + f"Invalid JSON in configuration file: {error}" + ) from error - exceptions = config.get("umlaut_exceptions", []) + umlaut_exceptions = config.get("umlaut_exceptions", []) - if not isinstance(exceptions, list): - print("'umlaut_exceptions' must be a JSON array.") - sys.exit(1) + if not isinstance(umlaut_exceptions, list): + raise ValueError("'umlaut_exceptions' must be a JSON array.") - config["umlaut_exceptions"] = { - str(word).lower() - for word in exceptions + replacements = config.get("replacements", {}) + + if not isinstance(replacements, dict): + raise ValueError("'replacements' must be a JSON object.") + + for source, target in replacements.items(): + if not isinstance(source, str) or not isinstance(target, str): + raise ValueError( + "Replacement keys and values must both be strings." + ) + + return { + "umlaut_exceptions": { + str(word).lower() + for word in umlaut_exceptions + }, + "replacements": replacements, } - return config - def file_hash(path: Path) -> str: """Calculate the SHA-256 hash of a file.""" @@ -68,10 +102,15 @@ def file_hash(path: Path) -> str: return hasher.hexdigest() -def replace_umlauts(text: str, umlaut_exceptions: set[str]) -> str: +def replace_umlauts( + text: str, + umlaut_exceptions: set[str], +) -> str: """ Replace ae/oe/ue with ä/ö/ü. - Words from the exception list are protected temporarily. + + Exception strings are matched case-insensitively and protected + before the replacements are applied. """ protected_words = {} @@ -98,7 +137,30 @@ def replace_umlauts(text: str, umlaut_exceptions: set[str]) -> str: return text -def clean_stem(stem: str, umlaut_exceptions: set[str]) -> str | None: +def apply_replacements(text: str, replacements: dict[str, str]) -> str: + """Apply configured replacements case-insensitively.""" + # Replace longer keys first to avoid partial matches. + replacement_items = sorted( + replacements.items(), + key=lambda item: len(item[0]), + reverse=True, + ) + + for source, target in replacement_items: + if not source: + continue + + pattern = re.compile(re.escape(source), re.IGNORECASE) + text = pattern.sub(lambda _: target, text) + + return text + + +def clean_stem( + stem: str, + umlaut_exceptions: set[str], + replacements: dict[str, str], +) -> str | None: """ Create a cleaned filename stem. @@ -121,8 +183,17 @@ def clean_stem(stem: str, umlaut_exceptions: set[str]) -> str | None: # Normalize whitespace and remove unnecessary separators. remainder = re.sub(r"\s+", " ", remainder).strip(" -_.") - # Convert ae/oe/ue to ä/ö/ü, respecting configured exceptions. - remainder = replace_umlauts(remainder, umlaut_exceptions) + # Convert ae/oe/ue to ä/ö/ü. + remainder = replace_umlauts( + remainder, + umlaut_exceptions, + ) + + # Apply configured replacements such as GMBH -> GmbH. + remainder = apply_replacements( + remainder, + replacements, + ) if not remainder: return formatted_date @@ -135,10 +206,14 @@ def is_clean_filename(path: Path) -> bool: return bool(re.match(r"^\d{4}-\d{2}-\d{2}\b", path.stem)) -def safely_delete(path: Path, folder: Path): - """Delete a file or log the planned action in test mode.""" - if DRY_RUN: - log_message(folder, f"[TEST] DELETE {path.name}") +def safely_delete( + path: Path, + folder: Path, + dry_run: bool, +): + """Delete a file or log the planned action in dry-run mode.""" + if dry_run: + log_message(folder, f"[DRY-RUN] DELETE {path.name}") return try: @@ -151,12 +226,17 @@ def safely_delete(path: Path, folder: Path): ) -def safely_rename(old_path: Path, new_path: Path, folder: Path): - """Rename a file or log the planned action in test mode.""" - if DRY_RUN: +def safely_rename( + old_path: Path, + new_path: Path, + folder: Path, + dry_run: bool, +): + """Rename a file or log the planned action in dry-run mode.""" + if dry_run: log_message( folder, - f"[TEST] RENAME {old_path.name} -> {new_path.name}", + f"[DRY-RUN] RENAME {old_path.name} -> {new_path.name}", ) return @@ -175,18 +255,27 @@ def safely_rename(old_path: Path, new_path: Path, folder: Path): def main(): - config = load_config() - umlaut_exceptions = config["umlaut_exceptions"] + arguments = parse_arguments() - folder = Path(sys.argv[1]) if len(sys.argv) > 1 else Path.cwd() + try: + config = load_config() + except (FileNotFoundError, ValueError) as error: + print(f"[ERROR] {error}") + return 1 + + folder = arguments.folder if not folder.is_dir(): - print(f"Folder not found: {folder}") - sys.exit(1) + print(f"[ERROR] Folder not found: {folder}") + return 1 + + dry_run = arguments.dry_run + umlaut_exceptions = config["umlaut_exceptions"] + replacements = config["replacements"] log_message( folder, - f"[START] Processing folder; test_mode={DRY_RUN}", + f"[START] Processing folder; dry_run={dry_run}", ) files = [ @@ -202,7 +291,7 @@ def main(): if not files: log_message(folder, "[INFO] No files found") log_message(folder, "[END] Processing completed") - return + return 0 log_message(folder, f"[INFO] Found {len(files)} file(s)") @@ -222,7 +311,7 @@ def main(): files_to_keep = set() files_to_delete = set() - # For identical files, prefer a file that already has a clean filename. + # Prefer an already-clean filename for identical files. for group in hash_groups.values(): if len(group) == 1: files_to_keep.add(group[0]) @@ -248,16 +337,21 @@ def main(): # Delete duplicate files. for path in files_to_delete: - safely_delete(path, folder) + safely_delete(path, folder, dry_run) # Rename remaining Caya original files. for path in files_to_keep: - new_stem = clean_stem(path.stem, umlaut_exceptions) + new_stem = clean_stem( + path.stem, + umlaut_exceptions, + replacements, + ) if new_stem is None: log_message( folder, - f"[SKIP] Filename does not match Caya format: {path.name}", + f"[SKIP] Filename does not match Caya format: " + f"{path.name}", ) continue @@ -276,30 +370,33 @@ def main(): except OSError as error: log_message( folder, - f"[ERROR] Could not compare {path.name} " - f"and {target.name}: {error}", + f"[ERROR] Could not compare " + f"{path.name} and {target.name}: {error}", ) continue if source_hash == target_hash: log_message( folder, - f"[DUPLICATE] {path.name} matches existing {target.name}", + f"[DUPLICATE] {path.name} matches " + f"existing {target.name}", ) - safely_delete(path, folder) + safely_delete(path, folder, dry_run) else: log_message( folder, - "[CONFLICT] Target already exists with different " - f"content: {path.name} -> {target.name}", + "[CONFLICT] Target already exists with " + f"different content: {path.name} -> {target.name}", ) continue - safely_rename(path, target, folder) + safely_rename(path, target, folder, dry_run) log_message(folder, "[END] Processing completed") + return 0 + if __name__ == "__main__": - main() + raise SystemExit(main()) diff --git a/caya-clean/configuration.json b/caya-clean/configuration.json index 5505376..84315f3 100644 --- a/caya-clean/configuration.json +++ b/caya-clean/configuration.json @@ -1,5 +1,11 @@ { "umlaut_exceptions": [ + "aktuell", "eue" - ] + ], + "replacements": { + "TOMTEC IMAGING SYSTEMS": "TOMTEC Imaging Systems", + "OPTIK VITZTHUM": "Optik Vitzthum", + "GMBH": "GmbH" + } }