403 lines
10 KiB
Python
403 lines
10 KiB
Python
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
|
|
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 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
|
|
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 and validate configuration.json."""
|
|
if not CONFIG_FILE.exists():
|
|
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:
|
|
raise ValueError(
|
|
f"Invalid JSON in configuration file: {error}"
|
|
) from error
|
|
|
|
umlaut_exceptions = config.get("umlaut_exceptions", [])
|
|
|
|
if not isinstance(umlaut_exceptions, list):
|
|
raise ValueError("'umlaut_exceptions' must be a JSON array.")
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
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 ä/ö/ü.
|
|
|
|
Exception strings are matched case-insensitively and protected
|
|
before the replacements are applied.
|
|
"""
|
|
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 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.
|
|
|
|
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 ä/ö/ü.
|
|
remainder = replace_umlauts(
|
|
remainder,
|
|
umlaut_exceptions,
|
|
)
|
|
|
|
# Apply configured replacements such as GMBH -> GmbH.
|
|
remainder = apply_replacements(
|
|
remainder,
|
|
replacements,
|
|
)
|
|
|
|
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,
|
|
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:
|
|
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,
|
|
dry_run: bool,
|
|
):
|
|
"""Rename a file or log the planned action in dry-run mode."""
|
|
if dry_run:
|
|
log_message(
|
|
folder,
|
|
f"[DRY-RUN] 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():
|
|
arguments = parse_arguments()
|
|
|
|
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"[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; dry_run={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 0
|
|
|
|
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()
|
|
|
|
# Prefer an already-clean filename for identical files.
|
|
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, dry_run)
|
|
|
|
# Rename remaining Caya original files.
|
|
for path in files_to_keep:
|
|
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: "
|
|
f"{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 "
|
|
f"{path.name} and {target.name}: {error}",
|
|
)
|
|
continue
|
|
|
|
if source_hash == target_hash:
|
|
log_message(
|
|
folder,
|
|
f"[DUPLICATE] {path.name} matches "
|
|
f"existing {target.name}",
|
|
)
|
|
safely_delete(path, folder, dry_run)
|
|
else:
|
|
log_message(
|
|
folder,
|
|
"[CONFLICT] Target already exists with "
|
|
f"different content: {path.name} -> {target.name}",
|
|
)
|
|
|
|
continue
|
|
|
|
safely_rename(path, target, folder, dry_run)
|
|
|
|
log_message(folder, "[END] Processing completed")
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|