frank.conrads/pptx-image-compress
1
0
Fork 0
Code Actions Packages Activity

In Batch reflect subfolder /bin, revert script to old due to error in display

Browse Source
This commit is contained in:
Frank Conrads
2025-09-10 12:14:54 +02:00
parent cd24151050
commit e0a8aad59d
2 changed files with 96 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

127
pptx_image_compress.py
View File

@@ -28,16 +28,21 @@ from datetime import timedelta
from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Lock
# -------------------- Version --------------------
__version__ = "1.0.0"
ALLOWED_EXT = {".jpg", ".jpeg", ".png", ".webp", ".gif"}
CAESIUM_QUALITY = 90
# -------------------- Konfiguration --------------------
ALLOWED_EXT = {".jpg", ".jpeg", ".png", ".webp", ".gif"} # GIF wird übersprungen
CAESIUM_QUALITY = 90 # -q 90
PROGRESS_BAR_LEN = 40
TEMP_PREFIX = "pptx_compress_"
# -------------------- Utilities --------------------
def human_mb(nbytes: int) -> float:
return round(nbytes / (1024 * 1024), 2)
def ensure_clean_file(path: Path):
if path.exists():
try:
@@ -48,6 +53,7 @@ def ensure_clean_file(path: Path):
except Exception:
pass
def cleanup_old_temps():
tmp_root = Path(tempfile.gettempdir())
for p in tmp_root.glob(f"{TEMP_PREFIX}*"):
@@ -59,13 +65,15 @@ def cleanup_old_temps():
except Exception:
pass
def print_progress(i: int, total: int):
if total <= 0:
return
done = int(PROGRESS_BAR_LEN * i / total)
bar = "█" * done + "-" * (PROGRESS_BAR_LEN - done)
pct = int(i * 100 / total)
print(f"
print(f"\rBilder: |{bar}| {i}/{total} ({pct}%)", end="", flush=True)
def zip_dir_to_pptx(src_dir: Path, out_pptx: Path):
with zipfile.ZipFile(out_pptx, "w", compression=zipfile.ZIP_DEFLATED) as z:
@@ -75,37 +83,53 @@ def zip_dir_to_pptx(src_dir: Path, out_pptx: Path):
rel = full.relative_to(src_dir)
z.write(full, arcname=str(rel))
def which(cmd: str) -> str | None:
return shutil.which(cmd)
def compress_with_caesium(original: Path, out_dir: Path, caesium_threads: int | None) -> Path | None:
"""
Ruft caesiumclt auf, um eine komprimierte Version zu erzeugen.
Output wird ins out_dir geschrieben (gleicher Filename).
Gibt Pfad zur erzeugten Datei zurück oder None bei Fehler.
"""
exe = which("caesiumclt")
if not exe:
if not exe:
raise RuntimeError(
"❌ 'caesiumclt' wurde nicht gefunden. Bitte CaesiumCLT installieren und in PATH verfügbar machen."
)
out_dir.mkdir(parents=True, exist_ok=True)
# Nur Formate an Caesium geben, die es unterstützt: JPG/JPEG, PNG, WEBP
ext = original.suffix.lower()
if ext not in {".jpg", ".jpeg", ".png", ".webp"}:
if ext not in {".jpg", ".jpeg", ".png", ".webp"}:
return None
return None # GIF & andere werden übersprungen
cmd = [
exe,
"-q", str(CAESIUM_QUALITY),
"-O", "bigger", # <<< nur überschreiben, wenn Ziel größer ist
"-o", str(out_dir),
]
if caesium_threads is not None:
cmd += ["--threads", str(caesium_threads)]
cmd += [str(original)]
try:
r = subprocess.run(cmd, capture_output=True, text=True)
if r.returncode != 0:
if r.returncode != 0:
sys.stderr.write(f"
[caesiumclt] Fehler bei {original.name}:
{r.stderr}
sys.stderr.write(f"\n[caesiumclt] Fehler bei {original.name}:\n{r.stderr}\n")
return None
out_file = out_dir / original.name
return out_file if out_file.exists() else None
except Exception as ex:
except Exception as ex:
sys.stderr.write(f"
[caesiumclt] Ausnahme bei {original.name}: {ex}
sys.stderr.write(f"\n[caesiumclt] Ausnahme bei {original.name}: {ex}\n")
return None
def format_duration(seconds: float) -> str:
total_ms = int(round(seconds * 1000))
td = timedelta(milliseconds=total_ms)
@@ -115,14 +139,27 @@ def format_duration(seconds: float) -> str:
return f"{hms}.{frac[:2]}"
return base
def main():
start_time = time.perf_counter()
parser = argparse.ArgumentParser(
description="PPTX Grafik-Komprimier-Tool (nur CaesiumCLT, Multi-Thread, sauberes Cleanup)",
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
parser.add_argument("-i", "--input", help="Input-PPTX", required=False)
parser.add_argument("-o", "--output", help="Output-PPTX", required=False)
parser.add_argument("-o", "--output", help="Output-PPTX", required=False)
parser.add_argument("-t", "--threads", type=int, default=min(32, os.cpu_count() or 4), help="Anzahl paralleler Threads für die Bildverarbeitung")
parser.add_argument(
"-t", "--threads",
type=int,
default=min(32, os.cpu_count() or 4),
help="Anzahl paralleler Threads für die Bildverarbeitung"
)
parser.add_argument(
"--version",
action="version",
version=f"%(prog)s {__version__}"
)
args = parser.parse_args()
@@ -135,22 +172,28 @@ def main():
print("❌ Eingabedatei existiert nicht oder ist keine .pptx")
sys.exit(2)
if args.output:
output_pptx = Path(args.output).resolve()
else:
output_pptx = input_pptx.with_name(f"{input_pptx.stem}_compressed.pptx")
# Vorherige Temp-Files & existierendes Output löschen
cleanup_old_temps()
ensure_clean_file(output_pptx)
work_dir = Path(tempfile.mkdtemp(prefix=TEMP_PREFIX + "work_"))
# --- Zwei getrennte Temp-Verzeichnisse ---
work_dir = Path(tempfile.mkdtemp(prefix=TEMP_PREFIX + "work_")) # entpackte PPTX
scratch_dir = Path(tempfile.mkdtemp(prefix=TEMP_PREFIX + "scratch_")) # Caesium-Ausgaben (außerhalb!)
# Logdatei neben Output
log_file = output_pptx.with_suffix(".log")
ensure_clean_file(log_file)
ensure_clean_file(log_file)
log_lines = ["image_name,size_before,size_after,saving,saving_percent
log_lines = ["image_name,size_before,size_after,saving,saving_percent\n"]
size_before = input_pptx.stat().st_size
try:
# Entpacken
with zipfile.ZipFile(input_pptx, "r") as z:
z.extractall(work_dir)
@@ -165,13 +208,15 @@ def main():
print(f"🔧 Finde Bilder in {media_dir} ... {total} Kandidaten")
print_progress(0, total)
# Vorab prüfen, ob caesiumclt verfügbar ist
if not which("caesiumclt"):
if not which("caesiumclt"):
print("
print("\n❌ 'caesiumclt' nicht gefunden. Bitte installieren und in PATH verfügbar machen.")
sys.exit(3)
# Oversubscription vermeiden: viele Python-Threads -> caesium intern 1 Thread
caesium_threads = 1 if args.threads and args.threads > 1 else None
# Thread-sichere Fortschritts- & Log-Verwaltung
lock = Lock()
done_count = 0
@@ -179,43 +224,52 @@ def main():
nonlocal done_count
ext = img_path.suffix.lower()
orig_size = img_path.stat().st_size
# GIF überspringen
if ext == ".gif":
with lock:
done_count += 1
done_count += 1
log_lines.append(f"{img_path.name},{orig_size},{orig_size},0,0.0
log_lines.append(f"{img_path.name},{orig_size},{orig_size},0,0.0\n")
print_progress(done_count, total)
return
chosen_size = orig_size
try:
# Eigener Output-Unterordner pro Bild, um Kollisionen zu vermeiden
out_sub = scratch_dir / f"img_{idx:06d}"
caesium_out = compress_with_caesium(img_path, out_sub, caesium_threads)
if caesium_out and caesium_out.exists():
s = caesium_out.stat().st_size
if s < orig_size:
# kleineren ersetzen (atomar)
tmp_target = img_path.with_suffix(img_path.suffix + ".tmp")
shutil.copy2(caesium_out, tmp_target)
tmp_target.replace(img_path)
chosen_size = s
except Exception:
except Exception:
chosen_size = orig_size # Original beibehalten
finally:
saving = orig_size - chosen_size
saving_percent = round((saving / orig_size) * 100, 2) if orig_size > 0 else 0.0
with lock:
with lock:
log_lines.append(f"{img_path.name},{orig_size},{chosen_size},{saving},{saving_percent}
log_lines.append(f"{img_path.name},{orig_size},{chosen_size},{saving},{saving_percent}\n")
done_count += 1
print_progress(done_count, total)
# Parallel ausführen
if total > 0:
with ThreadPoolExecutor(max_workers=max(1, args.threads)) as ex:
futures = [ex.submit(worker, i, p) for i, p in enumerate(images, start=1)]
for _ in as_completed(futures):
for _ in as_completed(futures):
pass # Fortschritt wird im Worker gezeichnet
print() # newline nach Progressbar
# --- Safety-Cleanup innerhalb des Arbeitsverzeichnisses ---
# 1) Entferne evtl. vorhandene caesium*-Ordner (aus alten Runs)
for p in work_dir.rglob("*"):
try:
if p.is_dir() and p.name.lower().startswith("caesium"):
@@ -223,6 +277,7 @@ def main():
except Exception:
pass
# 2) Lösche eventuelle .tmp-Dateien in ppt/media
media_dir = work_dir / "ppt" / "media"
if media_dir.exists():
for f in media_dir.iterdir():
@@ -232,23 +287,24 @@ def main():
except Exception:
pass
# Neue PPTX bauen (nur work_dir -> scratch_dir liegt außerhalb und ist damit sicher ausgeschlossen)
zip_dir_to_pptx(work_dir, output_pptx)
size_after = output_pptx.stat().st_size
# Log schreiben
try:
with open(log_file, "w", encoding="utf-8") as f:
f.writelines(log_lines)
except Exception:
pass
pass
# Summary
savings_pct = 0.0
if size_before > 0:
savings_pct = round(100.0 * (size_before - size_after) / size_before, 2)
elapsed = time.perf_counter() - start_time
print("
print("\n✅ Fertig!")
print("Summary")
print("-------")
print(f"Version: {__version__}")
@@ -260,6 +316,7 @@ def main():
print(f"Log-Datei: {log_file}")
finally:
# Aufräumen ALLER temporären Dateien/Ordner
try:
shutil.rmtree(work_dir, ignore_errors=True)
except Exception:
@@ -268,7 +325,9 @@ def main():
shutil.rmtree(scratch_dir, ignore_errors=True)
except Exception:
pass
# Zusätzlich: ältere Reste entfernen
cleanup_old_temps()
if __name__ == "__main__":
main()