claude-ai
466 lines
17 KiB
Python
466 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Image Folder Watcher
|
||
====================
|
||
Überwacht einen Zielordner. Wenn ein Unterordner mit Bildern hineingeschoben wird,
|
||
werden alle Bilder automatisch:
|
||
1. In einen "Backup"-Unterordner kopiert (Originale erhalten)
|
||
2. Umbenannt nach: <Ordnername>_00, <Ordnername>_01, ...
|
||
3. Auf 72 DPI und 1920px Breite skaliert (Höhe proportional)
|
||
|
||
Unterstützte Formate: JPEG, PNG, WEBP, HEIC, HEIF, TIFF, BMP, GIF, AVIF
|
||
|
||
Verwendung:
|
||
python watcher.py /pfad/zum/überwachten/ordner
|
||
|
||
Abhängigkeiten:
|
||
pip install Pillow pillow-heif watchdog
|
||
|
||
Für CachyOS/Arch:
|
||
sudo pacman -S python-pillow python-watchdog
|
||
pip install pillow-heif --break-system-packages (für HEIC/HEIF)
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import re
|
||
import shutil
|
||
import time
|
||
import logging
|
||
import argparse
|
||
import unicodedata
|
||
from pathlib import Path
|
||
from watchdog.observers import Observer
|
||
from watchdog.events import FileSystemEventHandler
|
||
|
||
from PIL import Image
|
||
|
||
# HEIC/HEIF-Support optional laden
|
||
try:
|
||
from pillow_heif import register_heif_opener
|
||
register_heif_opener()
|
||
HEIF_SUPPORT = True
|
||
except ImportError:
|
||
HEIF_SUPPORT = False
|
||
|
||
# ─── Konfiguration ───────────────────────────────────────────────────────────
|
||
|
||
TARGET_WIDTH = 1920
|
||
TARGET_DPI = 72
|
||
BACKUP_DIR_NAME = "Backup"
|
||
OUTPUT_FORMAT = "JPEG" # Ausgabeformat für konvertierte Bilder (JPEG, PNG, WEBP)
|
||
JPEG_QUALITY = 92
|
||
MAX_FILE_SIZE = 1_000_000 # Maximale Dateigröße in Bytes (1 MB)
|
||
JPEG_QUALITY_MIN = 40 # Untergrenze für JPEG-Qualität bei Größenanpassung
|
||
|
||
SUPPORTED_EXTENSIONS = {
|
||
".jpg", ".jpeg", ".png", ".webp", ".tiff", ".tif",
|
||
".bmp", ".gif", ".avif", ".heic", ".heif",
|
||
}
|
||
|
||
# Stabilisierungszeit: Warten bis keine neuen Dateien mehr kommen (Sekunden)
|
||
# Höher setzen bei SMB/Netzwerk-Kopien vom Mac (Standard: 5s)
|
||
STABILIZE_DELAY = 5.0
|
||
|
||
# ─── Logging ──────────────────────────────────────────────────────────────────
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s │ %(levelname)-7s │ %(message)s",
|
||
datefmt="%H:%M:%S",
|
||
)
|
||
log = logging.getLogger("watcher")
|
||
|
||
# ─── Bildverarbeitung ────────────────────────────────────────────────────────
|
||
|
||
|
||
def sanitize_name(name: str) -> str:
|
||
"""Macht einen Datei-/Ordnernamen web-sicher.
|
||
|
||
- Unicode-Zeichen → ASCII-Äquivalent (ä→ae, ö→oe, ü→ue, ß→ss, é→e, ...)
|
||
- Leerzeichen → Bindestrich
|
||
- Sonderzeichen entfernen (nur a-z, 0-9, Bindestrich, Unterstrich)
|
||
- Mehrfache Bindestriche zusammenfassen
|
||
- Alles lowercase
|
||
"""
|
||
# Deutsche Umlaute explizit ersetzen
|
||
replacements = {
|
||
"ä": "ae", "ö": "oe", "ü": "ue", "ß": "ss",
|
||
"Ä": "Ae", "Ö": "Oe", "Ü": "Ue",
|
||
}
|
||
for src, dst in replacements.items():
|
||
name = name.replace(src, dst)
|
||
|
||
# Restliche Unicode-Zeichen → ASCII (é→e, ñ→n, etc.)
|
||
name = unicodedata.normalize("NFKD", name)
|
||
name = name.encode("ascii", "ignore").decode("ascii")
|
||
|
||
# Leerzeichen und Unterstriche → Bindestrich
|
||
name = name.replace(" ", "-").replace("_", "-")
|
||
|
||
# Nur erlaubte Zeichen behalten: a-z, 0-9, Bindestrich
|
||
name = re.sub(r"[^a-zA-Z0-9\-]", "", name)
|
||
|
||
# Mehrfache Bindestriche zusammenfassen
|
||
name = re.sub(r"-{2,}", "-", name)
|
||
|
||
# Führende/trailing Bindestriche entfernen
|
||
name = name.strip("-")
|
||
|
||
# Lowercase
|
||
name = name.lower()
|
||
|
||
return name if name else "image"
|
||
|
||
|
||
def get_image_files(folder: Path) -> list[Path]:
|
||
"""Sammelt alle unterstützten Bilddateien rekursiv (inkl. Unterordner).
|
||
Ignoriert den Backup-Ordner."""
|
||
files = []
|
||
for f in sorted(folder.rglob("*")):
|
||
if f.is_file() and f.suffix.lower() in SUPPORTED_EXTENSIONS:
|
||
# Backup-Ordner überspringen
|
||
if BACKUP_DIR_NAME not in f.relative_to(folder).parts:
|
||
files.append(f)
|
||
return files
|
||
|
||
|
||
def fix_permissions(folder: Path) -> None:
|
||
"""Versucht Schreibrechte rekursiv zu setzen.
|
||
Auf SMB/NFS-Mounts schlägt chmod oft fehl – dann einfach weitermachen,
|
||
da die Berechtigungen dort vom Mount selbst gesteuert werden."""
|
||
try:
|
||
folder.chmod(0o755)
|
||
for item in folder.rglob("*"):
|
||
if item.is_file():
|
||
item.chmod(0o644)
|
||
elif item.is_dir() and item.name != BACKUP_DIR_NAME:
|
||
item.chmod(0o755)
|
||
log.info(f" 🔓 Berechtigungen rekursiv korrigiert")
|
||
except (PermissionError, OSError):
|
||
log.info(f" ℹ️ chmod nicht möglich (Netzwerk-Mount?) – fahre trotzdem fort")
|
||
|
||
|
||
def process_folder(folder: Path) -> None:
|
||
"""Verarbeitet einen neu erkannten Ordner mit Bildern."""
|
||
folder_name = folder.name
|
||
log.info(f"📂 Verarbeite Ordner: {folder_name}")
|
||
|
||
# ── 0. Berechtigungen korrigieren (Mac → Linux) ──
|
||
fix_permissions(folder)
|
||
|
||
images = get_image_files(folder)
|
||
if not images:
|
||
log.warning(f" Keine unterstützten Bilder in '{folder_name}' gefunden. Überspringe.")
|
||
return
|
||
|
||
log.info(f" {len(images)} Bild(er) gefunden.")
|
||
|
||
# ── 1. Backup-Ordner erstellen und Originale sichern (Struktur erhalten) ──
|
||
backup_dir = folder / BACKUP_DIR_NAME
|
||
backup_dir.mkdir(exist_ok=True)
|
||
log.info(f" 💾 Backup-Ordner: {backup_dir}")
|
||
|
||
for img_path in images:
|
||
# Relativen Pfad beibehalten für Unterordner-Struktur im Backup
|
||
rel_path = img_path.relative_to(folder)
|
||
backup_path = backup_dir / rel_path
|
||
backup_path.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.copy2(img_path, backup_path)
|
||
log.info(f" ↳ Backup: {rel_path}")
|
||
|
||
# ── 2. Umbenennen und alle Bilder ins Hauptverzeichnis verschieben ──
|
||
# Ordnername web-sicher machen
|
||
safe_name = sanitize_name(folder_name)
|
||
log.info(f" 🌐 Web-sicherer Name: '{folder_name}' → '{safe_name}'")
|
||
|
||
# Bestimme Ausgabe-Endung
|
||
ext_map = {
|
||
"JPEG": ".jpg",
|
||
"PNG": ".png",
|
||
"WEBP": ".webp",
|
||
}
|
||
out_ext = ext_map.get(OUTPUT_FORMAT, ".jpg")
|
||
|
||
# Erst alle in temporäre Namen verschieben (vermeidet Kollisionen)
|
||
temp_files: list[tuple[Path, Path]] = []
|
||
for idx, img_path in enumerate(images):
|
||
temp_name = f".tmp_{idx:04d}{img_path.suffix}"
|
||
temp_path = folder / temp_name
|
||
shutil.move(str(img_path), str(temp_path))
|
||
temp_files.append((temp_path, folder / f"{safe_name}-{idx:02d}{out_ext}"))
|
||
|
||
# Dann in finale Namen umbenennen
|
||
for temp_path, final_path in temp_files:
|
||
if final_path.exists():
|
||
final_path.unlink()
|
||
temp_path.rename(final_path)
|
||
log.info(f" ✏️ Umbenannt → {final_path.name}")
|
||
|
||
# Leere Unterordner aufräumen (außer Backup)
|
||
for sub in sorted(folder.iterdir(), reverse=True):
|
||
if sub.is_dir() and sub.name != BACKUP_DIR_NAME:
|
||
try:
|
||
sub.rmdir() # Nur wenn leer
|
||
log.info(f" 🗑️ Leerer Unterordner entfernt: {sub.name}")
|
||
except OSError:
|
||
pass # Nicht leer, ignorieren
|
||
|
||
# ── 3. Skalieren auf 1920px Breite, 72 DPI ──
|
||
final_files = get_image_files(folder)
|
||
for img_path in final_files:
|
||
try:
|
||
resize_image(img_path)
|
||
except Exception as e:
|
||
log.error(f" ❌ Fehler bei '{img_path.name}': {e}")
|
||
|
||
log.info(f"✅ Ordner '{folder_name}' fertig verarbeitet!\n")
|
||
|
||
|
||
def resize_image(img_path: Path) -> None:
|
||
"""Skaliert ein Bild auf TARGET_WIDTH px Breite (proportional), setzt 72 DPI,
|
||
und stellt sicher dass die Dateigröße unter MAX_FILE_SIZE bleibt."""
|
||
with Image.open(img_path) as img:
|
||
# EXIF-Rotation anwenden (wichtig bei Handy-Fotos!)
|
||
from PIL import ImageOps
|
||
img = ImageOps.exif_transpose(img)
|
||
|
||
orig_w, orig_h = img.size
|
||
|
||
if orig_w == 0:
|
||
log.warning(f" ⚠️ Überspringe '{img_path.name}' (Breite = 0)")
|
||
return
|
||
|
||
# Neues Seitenverhältnis berechnen
|
||
ratio = TARGET_WIDTH / orig_w
|
||
new_h = round(orig_h * ratio)
|
||
|
||
img_resized = img.resize((TARGET_WIDTH, new_h), Image.LANCZOS)
|
||
|
||
# In RGB konvertieren falls nötig (z.B. RGBA → RGB für JPEG)
|
||
if OUTPUT_FORMAT == "JPEG" and img_resized.mode in ("RGBA", "P", "LA"):
|
||
background = Image.new("RGB", img_resized.size, (255, 255, 255))
|
||
if img_resized.mode == "P":
|
||
img_resized = img_resized.convert("RGBA")
|
||
background.paste(img_resized, mask=img_resized.split()[-1] if "A" in img_resized.mode else None)
|
||
img_resized = background
|
||
|
||
# Speichern mit DPI – bei JPEG iterativ Qualität senken bis < MAX_FILE_SIZE
|
||
save_kwargs = {"dpi": (TARGET_DPI, TARGET_DPI)}
|
||
|
||
if OUTPUT_FORMAT == "JPEG":
|
||
quality = JPEG_QUALITY
|
||
save_kwargs["optimize"] = True
|
||
|
||
while quality >= JPEG_QUALITY_MIN:
|
||
save_kwargs["quality"] = quality
|
||
img_resized.save(img_path, format=OUTPUT_FORMAT, **save_kwargs)
|
||
file_size = img_path.stat().st_size
|
||
|
||
if file_size <= MAX_FILE_SIZE:
|
||
size_kb = file_size / 1024
|
||
log.info(
|
||
f" 📐 {img_path.name}: {orig_w}×{orig_h} → {TARGET_WIDTH}×{new_h} "
|
||
f"@ {TARGET_DPI} DPI, Q{quality}, {size_kb:.0f} KB"
|
||
)
|
||
return
|
||
|
||
# Qualität reduzieren und nochmal versuchen
|
||
quality -= 5
|
||
log.debug(f" {img_path.name}: {file_size/1024:.0f} KB > {MAX_FILE_SIZE/1024:.0f} KB – senke Qualität auf {quality}")
|
||
|
||
# Untergrenze erreicht, trotzdem speichern
|
||
file_size = img_path.stat().st_size
|
||
size_kb = file_size / 1024
|
||
log.warning(
|
||
f" ⚠️ {img_path.name}: {size_kb:.0f} KB – konnte nicht unter "
|
||
f"{MAX_FILE_SIZE/1024:.0f} KB gebracht werden (Q{JPEG_QUALITY_MIN})"
|
||
)
|
||
else:
|
||
img_resized.save(img_path, format=OUTPUT_FORMAT, **save_kwargs)
|
||
file_size = img_path.stat().st_size
|
||
size_kb = file_size / 1024
|
||
log.info(
|
||
f" 📐 {img_path.name}: {orig_w}×{orig_h} → {TARGET_WIDTH}×{new_h} "
|
||
f"@ {TARGET_DPI} DPI, {size_kb:.0f} KB"
|
||
)
|
||
|
||
|
||
# ─── Ordner-Überwachung ──────────────────────────────────────────────────────
|
||
|
||
|
||
class FolderHandler(FileSystemEventHandler):
|
||
"""Erkennt neue Unterordner und verarbeitet sie nach Stabilisierung.
|
||
Filtert temporäre Ordner (macOS SMB: .::TMPNAME:..., .DS_Store, ._*, etc.)"""
|
||
|
||
# Muster für temporäre/versteckte Ordner die ignoriert werden sollen
|
||
IGNORE_PREFIXES = (".", ".::TMPNAME", "._", "__MACOSX")
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self._pending: dict[str, float] = {}
|
||
self._processed: set[str] = set()
|
||
|
||
def _should_ignore(self, dir_path: Path) -> bool:
|
||
"""Prüft ob ein Ordner ignoriert werden soll."""
|
||
name = dir_path.name
|
||
if name == BACKUP_DIR_NAME:
|
||
return True
|
||
for prefix in self.IGNORE_PREFIXES:
|
||
if name.startswith(prefix):
|
||
return True
|
||
return False
|
||
|
||
def on_created(self, event):
|
||
if event.is_directory:
|
||
dir_path = Path(event.src_path)
|
||
if self._should_ignore(dir_path):
|
||
log.debug(f" Ignoriere temporären Ordner: {dir_path.name}")
|
||
return
|
||
if str(dir_path) not in self._processed:
|
||
log.info(f"🔍 Neuer Ordner erkannt: {dir_path.name} – warte auf Stabilisierung...")
|
||
self._pending[str(dir_path)] = time.time()
|
||
|
||
def on_moved(self, event):
|
||
"""Fängt Umbenennungen ab – macOS SMB erstellt .::TMPNAME:... und benennt dann um."""
|
||
if event.is_directory:
|
||
dest_path = Path(event.dest_path)
|
||
src_path = Path(event.src_path)
|
||
|
||
# Temporären Eintrag entfernen falls vorhanden
|
||
src_str = str(src_path)
|
||
if src_str in self._pending:
|
||
del self._pending[src_str]
|
||
|
||
if self._should_ignore(dest_path):
|
||
return
|
||
|
||
if str(dest_path) not in self._processed:
|
||
log.info(f"🔍 Ordner umbenannt: {src_path.name} → {dest_path.name} – warte auf Stabilisierung...")
|
||
self._pending[str(dest_path)] = time.time()
|
||
|
||
def on_modified(self, event):
|
||
"""Auch Modifikationen tracken (Dateien die in bestehenden Ordner kommen)."""
|
||
if event.is_directory:
|
||
dir_path = Path(event.src_path)
|
||
dir_str = str(dir_path)
|
||
if self._should_ignore(dir_path):
|
||
return
|
||
if dir_str in self._pending:
|
||
self._pending[dir_str] = time.time()
|
||
|
||
def check_pending(self):
|
||
"""Prüft ob ausstehende Ordner stabil sind (keine neuen Dateien seit STABILIZE_DELAY)."""
|
||
now = time.time()
|
||
ready = []
|
||
for dir_str, last_change in list(self._pending.items()):
|
||
if now - last_change >= STABILIZE_DELAY:
|
||
ready.append(dir_str)
|
||
|
||
for dir_str in ready:
|
||
del self._pending[dir_str]
|
||
folder = Path(dir_str)
|
||
# Nochmal prüfen: existiert der Ordner noch und ist er kein Temp-Ordner?
|
||
if folder.exists() and folder.is_dir() and not self._should_ignore(folder):
|
||
self._processed.add(dir_str)
|
||
try:
|
||
process_folder(folder)
|
||
except Exception as e:
|
||
log.error(f"❌ Fehler bei Verarbeitung von '{folder.name}': {e}")
|
||
|
||
|
||
# ─── Main ─────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description="Überwacht einen Ordner und verarbeitet neue Bild-Unterordner automatisch.",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
Beispiel:
|
||
python watcher.py ~/Bilder/Eingang
|
||
|
||
Was passiert:
|
||
1. Du schiebst einen Ordner (z.B. "Produkt-Fotos") in ~/Bilder/Eingang
|
||
2. Das Tool wartet kurz bis alle Dateien da sind
|
||
3. Originale werden nach Produkt-Fotos/Backup/ kopiert
|
||
4. Bilder werden umbenannt zu Produkt-Fotos_00.jpg, Produkt-Fotos_01.jpg, ...
|
||
5. Bilder werden auf 1920px Breite / 72 DPI skaliert
|
||
""",
|
||
)
|
||
parser.add_argument(
|
||
"watch_dir",
|
||
type=str,
|
||
help="Pfad zum Ordner der überwacht werden soll",
|
||
)
|
||
parser.add_argument(
|
||
"--width", type=int, default=TARGET_WIDTH,
|
||
help=f"Zielbreite in Pixel (Standard: {TARGET_WIDTH})",
|
||
)
|
||
parser.add_argument(
|
||
"--dpi", type=int, default=TARGET_DPI,
|
||
help=f"Ziel-DPI (Standard: {TARGET_DPI})",
|
||
)
|
||
parser.add_argument(
|
||
"--format", type=str, default=OUTPUT_FORMAT,
|
||
choices=["JPEG", "PNG", "WEBP"],
|
||
help=f"Ausgabeformat (Standard: {OUTPUT_FORMAT})",
|
||
)
|
||
parser.add_argument(
|
||
"--quality", type=int, default=JPEG_QUALITY,
|
||
help=f"JPEG-Qualität 1-100 (Standard: {JPEG_QUALITY})",
|
||
)
|
||
parser.add_argument(
|
||
"--delay", type=float, default=STABILIZE_DELAY,
|
||
help=f"Wartezeit in Sekunden bis Ordner als stabil gilt (Standard: {STABILIZE_DELAY})",
|
||
)
|
||
|
||
args = parser.parse_args()
|
||
|
||
# Konfiguration aus CLI-Argumenten in Modul-Variablen schreiben
|
||
# WICHTIG: Die laufende Modul-Instanz referenzieren (beim direkten Start
|
||
# ist das "__main__", nicht das per Name importierte "watcher"). Sonst
|
||
# landen die CLI-Overrides im falschen Modul und greifen nicht.
|
||
_self = sys.modules[__name__]
|
||
_self.TARGET_WIDTH = args.width
|
||
_self.TARGET_DPI = args.dpi
|
||
_self.OUTPUT_FORMAT = args.format
|
||
_self.JPEG_QUALITY = args.quality
|
||
_self.STABILIZE_DELAY = args.delay
|
||
|
||
watch_path = Path(args.watch_dir).expanduser().resolve()
|
||
if not watch_path.exists():
|
||
log.info(f"📁 Erstelle Überwachungsordner: {watch_path}")
|
||
watch_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
if not HEIF_SUPPORT:
|
||
log.warning("⚠️ pillow-heif nicht installiert – HEIC/HEIF-Dateien werden nicht unterstützt.")
|
||
log.warning(" Installiere mit: pip install pillow-heif --break-system-packages")
|
||
|
||
log.info(f"{'═' * 60}")
|
||
log.info(f" 🖼️ Image Folder Watcher gestartet")
|
||
log.info(f" 📂 Überwache: {watch_path}")
|
||
log.info(f" 📐 Ziel: {TARGET_WIDTH}px Breite, {TARGET_DPI} DPI, {OUTPUT_FORMAT}")
|
||
log.info(f" ⏱️ Stabilisierung: {STABILIZE_DELAY}s")
|
||
log.info(f"{'═' * 60}")
|
||
log.info(f" Warte auf neue Ordner... (Strg+C zum Beenden)\n")
|
||
|
||
handler = FolderHandler()
|
||
observer = Observer()
|
||
observer.schedule(handler, str(watch_path), recursive=False)
|
||
observer.start()
|
||
|
||
try:
|
||
while True:
|
||
handler.check_pending()
|
||
time.sleep(0.5)
|
||
except KeyboardInterrupt:
|
||
log.info("\n👋 Watcher beendet.")
|
||
observer.stop()
|
||
|
||
observer.join()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|