diff --git a/caya-clean/.pdm-python b/caya-clean/.pdm-python new file mode 100644 index 0000000..68935a3 --- /dev/null +++ b/caya-clean/.pdm-python @@ -0,0 +1 @@ +D:/dev/caya-clean/.venv/Scripts/python.exe \ No newline at end of file diff --git a/caya-clean/README.md b/caya-clean/README.md new file mode 100644 index 0000000..1c69224 --- /dev/null +++ b/caya-clean/README.md @@ -0,0 +1,23 @@ +# caya-clean + +Automatically cleans up and renames downloaded Caya letter files. + +The script: + +- Iterates over all files in a folder +- Converts dates from `YYYYMMDDTHHMMZ` to `YYYY-MM-DD` +- Removes the Caya ID from filenames +- Converts `ae`, `oe`, and `ue` to `ä`, `ö`, and `ü` +- Supports configurable umlaut exceptions +- Detects identical files using SHA-256 hashes +- Deletes duplicate files +- Keeps files that already have a clean filename +- Detects filename conflicts with different file contents +- Preserves file extensions + +## Example + +Input: + +```text +20250424T1227Z Friedhofsgaertnerei Florian Fischer - Rechnung cm9vc7tya00cf0ble1x9e9aso.pdf diff --git a/caya-clean/caya-clean.ps1 b/caya-clean/caya-clean.ps1 new file mode 100644 index 0000000..80bf346 --- /dev/null +++ b/caya-clean/caya-clean.ps1 @@ -0,0 +1,5 @@ +param( + [string]$Folder = (Get-Location).Path +) + +pdm run caya-clean $Folder diff --git a/caya-clean/caya-clean.py b/caya-clean/caya-clean.py new file mode 100644 index 0000000..699f682 --- /dev/null +++ b/caya-clean/caya-clean.py @@ -0,0 +1,305 @@ +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" + +DATE_PATTERN = re.compile(r"^(?P\d{8})T\d{4}Z\s+") + +# Caya ID at the end of the filename, before the file extension. +CAYA_ID_PATTERN = re.compile(r"\s+[a-z0-9]{20,35}$", re.IGNORECASE) + + +def log_message(folder: Path, message: str): + """Append a timestamped message to the folder logfile.""" + log_file = folder / LOG_FILENAME + timestamp = datetime.now().astimezone().isoformat(timespec="seconds") + + with log_file.open("a", encoding="utf-8") as file: + file.write(f"{timestamp} {message}\n") + + print(message) + + +def load_config() -> dict: + """Load the configuration from configuration.json.""" + if not CONFIG_FILE.exists(): + print(f"Configuration file not found: {CONFIG_FILE}") + sys.exit(1) + + 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) + + exceptions = config.get("umlaut_exceptions", []) + + if not isinstance(exceptions, list): + print("'umlaut_exceptions' must be a JSON array.") + sys.exit(1) + + config["umlaut_exceptions"] = { + str(word).lower() + for word in exceptions + } + + return config + + +def file_hash(path: Path) -> str: + """Calculate the SHA-256 hash of a file.""" + hasher = hashlib.sha256() + + with path.open("rb") as file: + for block in iter(lambda: file.read(1024 * 1024), b""): + hasher.update(block) + + return hasher.hexdigest() + + +def replace_umlauts(text: str, umlaut_exceptions: set[str]) -> str: + """ + Replace ae/oe/ue with ä/ö/ü. + Words from the exception list are protected temporarily. + """ + protected_words = {} + + for index, word in enumerate(umlaut_exceptions): + placeholder = f"__UMLAUT_EXCEPTION_{index}__" + pattern = re.compile(re.escape(word), re.IGNORECASE) + + def protect(match, placeholder=placeholder): + protected_words[placeholder] = match.group(0) + return placeholder + + text = pattern.sub(protect, text) + + text = text.replace("AE", "Ä") + text = text.replace("OE", "Ö") + text = text.replace("UE", "Ü") + text = text.replace("ae", "ä") + text = text.replace("oe", "ö") + text = text.replace("ue", "ü") + + for placeholder, original in protected_words.items(): + text = text.replace(placeholder, original) + + return text + + +def clean_stem(stem: str, umlaut_exceptions: set[str]) -> str | None: + """ + Create a cleaned filename stem. + + Return None if the filename does not match the expected + Caya original filename format. + """ + match = DATE_PATTERN.match(stem) + + if not match: + return None + + date = match.group("date") + formatted_date = f"{date[:4]}-{date[4:6]}-{date[6:8]}" + + remainder = stem[match.end():] + + # Remove the Caya ID at the end. + remainder = CAYA_ID_PATTERN.sub("", remainder) + + # 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) + + if not remainder: + return formatted_date + + return f"{formatted_date} {remainder}" + + +def is_clean_filename(path: Path) -> bool: + """Check whether a file already starts with YYYY-MM-DD.""" + 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}") + return + + try: + path.unlink() + log_message(folder, f"[DELETE] {path.name}") + except OSError as error: + log_message( + folder, + f"[ERROR] Could not delete {path.name}: {error}", + ) + + +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: + log_message( + folder, + f"[TEST] RENAME {old_path.name} -> {new_path.name}", + ) + return + + try: + old_path.rename(new_path) + log_message( + folder, + f"[RENAME] {old_path.name} -> {new_path.name}", + ) + except OSError as error: + log_message( + folder, + f"[ERROR] Could not rename " + f"{old_path.name} -> {new_path.name}: {error}", + ) + + +def main(): + config = load_config() + umlaut_exceptions = config["umlaut_exceptions"] + + folder = Path(sys.argv[1]) if len(sys.argv) > 1 else Path.cwd() + + if not folder.is_dir(): + print(f"Folder not found: {folder}") + sys.exit(1) + + log_message( + folder, + f"[START] Processing folder; test_mode={DRY_RUN}", + ) + + files = [ + path + for path in folder.iterdir() + if ( + path.is_file() + and not path.name.startswith("~$") + and path.name != LOG_FILENAME + ) + ] + + if not files: + log_message(folder, "[INFO] No files found") + log_message(folder, "[END] Processing completed") + return + + log_message(folder, f"[INFO] Found {len(files)} file(s)") + + # Calculate hashes and group files with identical content. + hash_groups = {} + + for path in files: + try: + hash_value = file_hash(path) + hash_groups.setdefault(hash_value, []).append(path) + except OSError as error: + log_message( + folder, + f"[ERROR] Could not hash {path.name}: {error}", + ) + + files_to_keep = set() + files_to_delete = set() + + # For identical files, prefer a file that already has a clean filename. + for group in hash_groups.values(): + if len(group) == 1: + files_to_keep.add(group[0]) + continue + + sorted_group = sorted( + group, + key=lambda path: ( + not is_clean_filename(path), + path.name.lower(), + ), + ) + + primary_file = sorted_group[0] + files_to_keep.add(primary_file) + + log_message(folder, "[DUPLICATES]") + log_message(folder, f" KEEP {primary_file.name}") + + for duplicate in sorted_group[1:]: + files_to_delete.add(duplicate) + log_message(folder, f" DELETE {duplicate.name}") + + # Delete duplicate files. + for path in files_to_delete: + safely_delete(path, folder) + + # Rename remaining Caya original files. + for path in files_to_keep: + new_stem = clean_stem(path.stem, umlaut_exceptions) + + if new_stem is None: + log_message( + folder, + f"[SKIP] Filename does not match Caya format: {path.name}", + ) + continue + + new_name = new_stem + path.suffix + target = path.with_name(new_name) + + # The file already has the desired name. + if target == path: + log_message(folder, f"[SKIP] Already clean: {path.name}") + continue + + if target.exists(): + try: + source_hash = file_hash(path) + target_hash = file_hash(target) + except OSError as error: + log_message( + folder, + f"[ERROR] Could not compare {path.name} " + f"and {target.name}: {error}", + ) + continue + + if source_hash == target_hash: + log_message( + folder, + f"[DUPLICATE] {path.name} matches existing {target.name}", + ) + safely_delete(path, folder) + else: + log_message( + folder, + "[CONFLICT] Target already exists with different " + f"content: {path.name} -> {target.name}", + ) + + continue + + safely_rename(path, target, folder) + + log_message(folder, "[END] Processing completed") + + +if __name__ == "__main__": + main() diff --git a/caya-clean/configuration.json b/caya-clean/configuration.json new file mode 100644 index 0000000..5505376 --- /dev/null +++ b/caya-clean/configuration.json @@ -0,0 +1,5 @@ +{ + "umlaut_exceptions": [ + "eue" + ] +} diff --git a/caya-clean/pdm.lock b/caya-clean/pdm.lock new file mode 100644 index 0000000..042e767 --- /dev/null +++ b/caya-clean/pdm.lock @@ -0,0 +1,11 @@ +# This file is @generated by PDM. +# It is not intended for manual editing. + +[metadata] +groups = ["default"] +strategy = ["inherit_metadata"] +lock_version = "4.5.1" +content_hash = "sha256:6e3746c7427353cd94cfd6c1a5a6309fdfd871714e3b1213ded2befc6ef522a0" + +[[metadata.targets]] +requires_python = ">=3.10" diff --git a/caya-clean/pyproject.toml b/caya-clean/pyproject.toml new file mode 100644 index 0000000..b424c20 --- /dev/null +++ b/caya-clean/pyproject.toml @@ -0,0 +1,14 @@ +[project] +name = "caya-clean" +version = "0.1.0" +description = "Automatically cleans up and renames downloaded Caya letter files." +readme = "README.md" +requires-python = ">=3.10" +dependencies = [] + +[project.scripts] +caya-clean = "caya_clean:main" + +[build-system] +requires = ["pdm-backend"] +build-backend = "pdm.backend"