first script
This commit is contained in:
@@ -0,0 +1 @@
|
||||
D:/dev/caya-clean/.venv/Scripts/python.exe
|
||||
@@ -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
|
||||
@@ -0,0 +1,5 @@
|
||||
param(
|
||||
[string]$Folder = (Get-Location).Path
|
||||
)
|
||||
|
||||
pdm run caya-clean $Folder
|
||||
@@ -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<date>\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()
|
||||
@@ -0,0 +1,5 @@
|
||||
{
|
||||
"umlaut_exceptions": [
|
||||
"eue"
|
||||
]
|
||||
}
|
||||
Generated
+11
@@ -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"
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user