#!/usr/home/jozapf/public_html/toolenv/bin/python
# -*- coding: utf-8 -*-
"""
autoindex-slides — Slideshow für ein Verzeichnis (SVG · JPG/PNG/GIF/WebP · PDF)
Version: 1.5.2
Ein einzelnes Python-CGI. Es scannt bei jedem Aufruf das Medienverzeichnis,
sortiert die Funde natürlich nach Dateiname und liefert:
* ohne Parameter -> die HTML-Seite (Jinja2, slides.tmpl.html)
* ?manifest=1 -> die Slide-Liste als JSON (Cache-Control: no-store)
* ?search=1 -> der Suchbestand als JSON (ETag, meist 304)
Neue Dateien erscheinen ohne Konfiguration und ohne Neustart: die Seite fragt
das Manifest regelmäßig neu ab, deshalb tauchen sie auch in einer bereits
offenen Slideshow auf.
Bilder werden serverseitig aufbereitet (Pillow), PDF-Seiten zu Bildern
gerastert (poppler / pdftoppm) — jede PDF-Seite wird eine eigene Slide. Der
Client bekommt dadurch ausschließlich <img>-Elemente, was auf dem Handy
zuverlässig funktioniert, wo ein eingebetteter PDF-Viewer es nicht tut. Das
Template lädt dafür genau eine eigene Skriptdatei (slides.js); die CSP der
Toolbox lässt mit script-src 'self' ohnehin nur solche zu.
Aufbereitet wird in Häppchen: pro Aufruf nur so lange, wie RENDER_BUDGET_SECONDS
erlaubt — ein 200-seitiges PDF kann so keinen CGI-Timeout auslösen. Damit die
Show trotzdem sofort brauchbar ist, kommt zuerst die erste Slide *jeder* Datei
an die Reihe, danach die Folgeseiten, und dabei bevorzugt die Umgebung der
Position, an der der Betrachter gerade steht (?near=). Solange etwas offen ist,
fragt der Client in kurzem Abstand nach (POLL_BUSY_SECONDS).
Die Ableitungen liegen unter unratbaren Namen im Cache-Verzeichnis; die
Originale werden nur verlinkt, wenn MEDIA_URL gesetzt ist.
"""
# Der Marker macht dieses Skript für verify.py zu einem zulässigen Rücksprung-
# ziel nach der Bot-Challenge (nur ausdrücklich markierte Werkzeuge sind das)
# und zur Kachel auf der Toolbox-Übersicht. Die "downloads" nennen die Quell-
# dateien, die die Toolbox einzeln zum Ansehen und Herunterladen anbietet —
# bewusst ohne media/ und cache/: nur was hier steht, wird ausgeliefert.
# TOOLBOX-TILE: {"title": "Autoindex Slides", "desc": "Slideshow für ein Verzeichnis — SVG, Fotos und PDF, mit Volltextsuche über Karteninhalt, PDF-Seitentext und Datei-Metadaten. Quelltext einzeln herunterladbar, ohne Medien.", "icon": "🖼️", "type": "web", "downloads": [{"file": "index.py", "label": "index.py"}, {"file": "slides.tmpl.html", "label": "Template"}, {"file": "slides.css", "label": "CSS"}, {"file": "slides.js", "label": "JS"}, {"file": ".htaccess", "label": ".htaccess"}]}
# jozapf.de toolbox · rev jzt-7c3f9a2e
# ════════════════════════════════════════════════════════════════════════════
# KONFIGURATION — alles Veränderbare steht hier.
# Beim Übertragen auf ein anderes Verzeichnis oder einen anderen Server ist
# nur dieser Block anzufassen; darunter muss nichts angepasst werden.
# ════════════════════════════════════════════════════════════════════════════
# ── Beschriftung ────────────────────────────────────────────────────────────
TITLE = "Slides"
SUBTITLE = "" # optionale Zeile unter dem Titel ("" = keine)
EMPTY_HINT = "Noch nichts abgelegt. Dateien ins Medienverzeichnis kopieren — sie erscheinen von selbst."
FOOTER_NOTE = "Bei Bedarf übernehme ich gerne die Umsetzung — jozapf.de"
FOOTER_URL = "https://jozapf.de" # Ziel des Hinweises; FOOTER_NOTE = "" blendet ihn aus
# ── Verzeichnisse ───────────────────────────────────────────────────────────
# Pfade: relativ zum Skript ODER absolut. MEDIA_DIR darf außerhalb des
# Web-Roots liegen (dann MEDIA_URL = None setzen) — die Originale sind dann
# über HTTP überhaupt nicht erreichbar, nur die Ableitungen im Cache.
MEDIA_DIR = "media" # hier werden die Inhalte abgelegt
CACHE_DIR = "cache" # abgeleitete Bilder; MUSS im Web-Root liegen
# URL-Pfade, relativ zur Seite. MEDIA_URL = None -> Originale nie verlinken.
MEDIA_URL = "media"
CACHE_URL = "cache"
# ── Welche Dateien werden aufgenommen ───────────────────────────────────────
RASTER_EXTENSIONS = (".jpg", ".jpeg", ".png", ".gif", ".webp", ".avif", ".bmp", ".tif", ".tiff")
VECTOR_EXTENSIONS = (".svg",) # werden als <img> gezeigt, nie ins DOM inline
DOCUMENT_EXTENSIONS = (".pdf",) # jede Seite wird eine eigene Slide
RECURSIVE = True # Unterverzeichnisse mitnehmen
MAX_FILES = 500 # Obergrenze, damit ein Versehen nichts sprengt
# ── Reihenfolge ─────────────────────────────────────────────────────────────
NATURAL_SORT = True # "2" vor "10" (statt alphabetisch "10" vor "2")
SORT_REVERSE = False # True = absteigend
# ── Aufbereitung ────────────────────────────────────────────────────────────
PDF_DPI = 150 # Rasterauflösung für PDF-Seiten
PDF_MAX_PAGES = 200 # Seiten je PDF, darüber wird abgeschnitten
DISPLAY_MAX_EDGE = 2400 # große Bilder verkleinern (0 = Original behalten)
THUMB_MAX_EDGE = 400 # Kantenlänge der Vorschaubilder im Raster
JPEG_QUALITY = 82
RENDER_BUDGET_SECONDS = 18.0 # Rechenzeit je Aufruf; der Rest folgt beim
# nächsten Nachfragen — nie ein Timeout
PROBE_BUDGET_SECONDS = 4.0 # davon höchstens für pdfinfo-Aufrufe
PRIORITY_FIRST_SLIDE = True # erst eine Slide je Datei, dann die Folgeseiten
PRUNE_ORPHANS = True # Cache-Dateien gelöschter Quellen aufräumen
# ── Verhalten der Oberfläche ────────────────────────────────────────────────
POLL_SECONDS = 45 # Abstand, in dem nach neuen Dateien gesehen wird
POLL_BUSY_SECONDS = 2 # Abstand, solange noch etwas aufbereitet wird
AUTOPLAY_SECONDS = 8 # Standard-Intervall der Diaschau (0 = kein Autoplay)
SHOW_CAPTIONS = True # Dateiname unter der Slide einblenden
SHOW_DOWNLOAD = False # Link auf das Original anbieten (braucht MEDIA_URL)
# ── Suche ───────────────────────────────────────────────────────────────────
# Volltextsuche über das, was in den Slides steht, und über die Metadaten der
# Dateien. Je Datei durchsucht: Dateiname und Änderungsdatum (immer, das Datum
# als JJJJ-MM-TT — "2026-07" findet die Neuzugänge des Monats), bei SVG der
# enthaltene Text samt <metadata>-Block, bei Rasterbildern die EXIF-Felder
# (Beschreibung, Stichwörter, Urheber, Aufnahmedatum), bei PDF die Info-Felder
# (Titel, Thema, Stichwörter) und der Seitentext (pdftotext) — jede Seite mit
# Text wird ein eigener Treffer und führt direkt auf die richtige Seite.
SEARCH_ENABLED = True # False = Feld ausblenden, Endpunkt antwortet 404
SEARCH_MAX_CHARS = 6000 # je Datei gespeicherter Text, darüber gekappt
SEARCH_BUDGET_SECONDS = 6.0 # Rechenzeit je Aufruf des Such-Endpunkts
# ── Bot-Schutz (toolbox_guard) ──────────────────────────────────────────────
GUARD_TOOL_ID = "slides" # Kennung in der Statistik; None = Guard aus
GUARD_IMPORT_DIR = ".." # wo toolbox_guard.py liegt (relativ zum Skript)
# ── Externe Programme ───────────────────────────────────────────────────────
PDFTOPPM = "pdftoppm" # poppler-utils
PDFINFO = "pdfinfo"
PDFTOTEXT = "pdftotext"
# ════════════════════════════════════════════════════════════════════════════
# Ab hier ist nichts mehr zu konfigurieren.
# ════════════════════════════════════════════════════════════════════════════
import hashlib
import json
import os
import re
import subprocess
import sys
import time
from pathlib import Path
from urllib.parse import parse_qs, quote
SCRIPT_DIR = Path(__file__).resolve().parent
TEMPLATE = "slides.tmpl.html"
_START = time.monotonic()
def _resolve(p):
"""Konfigurierten Pfad auflösen (relativ zum Skript oder absolut)."""
path = Path(os.path.expanduser(str(p)))
return path if path.is_absolute() else (SCRIPT_DIR / path)
MEDIA_PATH = _resolve(MEDIA_DIR)
CACHE_PATH = _resolve(CACHE_DIR)
ALL_EXTENSIONS = tuple(RASTER_EXTENSIONS) + tuple(VECTOR_EXTENSIONS) + tuple(DOCUMENT_EXTENSIONS)
# ── Bot-Schutz ──────────────────────────────────────────────────────────────
# Fällt bewusst offen aus: ein Importproblem darf die Slideshow nicht offline
# nehmen. Der Guard sitzt auf dem Seiteneinstieg (UA-Filter, JS-Challenge,
# Rate-Limit). Das Manifest prüft nur die bestehende Sitzung, weil das Polling
# sonst das eigene Rate-Limit aufbrauchen würde.
sys.path.insert(0, str(_resolve(GUARD_IMPORT_DIR)))
try:
from toolbox_guard import guard as _guard, valid_session, valid_admin
except Exception: # pragma: no cover
def _guard(_tool):
return True
def valid_session(_cookie=None):
return True
def valid_admin(_cookie=None):
return True
# ── Sortierung ──────────────────────────────────────────────────────────────
_NUM_RE = re.compile(r"(\d+)")
def natural_key(name):
"""Sortierschlüssel: Ziffernblöcke numerisch, Text klein geschrieben.
Alle Bestandteile sind gleich geformte Tupel, damit der Vergleich nie auf
gemischte Typen läuft ('01' < '2' < '10')."""
if not NATURAL_SORT:
return [(0, 0, name.lower())]
return [
(1, int(part), "") if part.isdigit() else (0, 0, part)
for part in _NUM_RE.split(name.lower())
]
# ── Zeitbudget ──────────────────────────────────────────────────────────────
def _elapsed():
return time.monotonic() - _START
def _budget_left(limit=None):
return _elapsed() < (RENDER_BUDGET_SECONDS if limit is None else limit)
# ── Cache ───────────────────────────────────────────────────────────────────
def _salt():
"""Zufälliges Salt für die Cache-Namen; einmalig angelegt (0600).
Es macht die Namen der Ableitungen unratbar — ohne Manifest kommt niemand
an eine PDF-Seite, auch wenn er den Original-Dateinamen kennt."""
CACHE_PATH.mkdir(parents=True, exist_ok=True)
path = CACHE_PATH / ".salt"
try:
return path.read_bytes().strip()
except OSError:
value = os.urandom(16).hex().encode()
try:
fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(fd, "wb") as fh:
fh.write(value)
return value
except OSError: # Wettlauf: der andere war da
return path.read_bytes().strip()
def cache_key(rel, stat):
"""Schlüssel aus Pfad, Änderungszeit und Größe: ändert sich die Datei,
entsteht ein neuer Schlüssel und die Ableitung wird neu erzeugt."""
raw = b"%s|%s|%d|%d" % (_salt(), rel.encode("utf-8"), int(stat.st_mtime), stat.st_size)
return hashlib.sha256(raw).hexdigest()[:20]
def _meta_load(key):
try:
return json.loads((CACHE_PATH / ("%s.json" % key)).read_text("utf-8"))
except (OSError, ValueError):
return {}
def _meta_save(key, meta):
path = CACHE_PATH / ("%s.json" % key)
tmp = CACHE_PATH / ("%s.json.tmp" % key)
try:
tmp.write_text(json.dumps(meta), "utf-8")
os.replace(str(tmp), str(path))
except OSError:
pass
# ── Quellen einsammeln ──────────────────────────────────────────────────────
def scan_sources():
"""Alle aufzunehmenden Dateien, natürlich sortiert nach relativem Pfad."""
if not MEDIA_PATH.is_dir():
return []
pattern = "**/*" if RECURSIVE else "*"
found = []
for path in MEDIA_PATH.glob(pattern):
try:
if not path.is_file() or path.name.startswith("."):
continue
if path.suffix.lower() not in ALL_EXTENSIONS:
continue
rel = path.relative_to(MEDIA_PATH).as_posix()
found.append((rel, path, path.stat()))
except OSError:
continue
found.sort(key=lambda item: natural_key(item[0]), reverse=SORT_REVERSE)
return found[:MAX_FILES]
# ── Ableitungen: Rasterbilder ───────────────────────────────────────────────
def _pillow():
from PIL import Image, ImageOps
return Image, ImageOps
def _save(img, path, keep_alpha):
"""Erst in eine Nebendatei schreiben, dann umbenennen: ein abgebrochener
Aufruf hinterlässt so nie ein halbes Bild, das als fertig gilt."""
tmp = Path(str(path) + ".tmp")
if keep_alpha:
img.save(tmp, "PNG", optimize=True)
else:
if img.mode not in ("RGB", "L"):
img = img.convert("RGB")
img.save(tmp, "JPEG", quality=JPEG_QUALITY, optimize=True, progressive=True)
os.replace(str(tmp), str(path))
def derive_raster(src, key, meta):
"""Anzeige- und Vorschauvariante eines Rasterbildes erzeugen."""
Image, ImageOps = _pillow()
with Image.open(src) as opened:
animated = getattr(opened, "n_frames", 1) > 1
img = ImageOps.exif_transpose(opened) or opened # Handy-Fotos drehen
meta["w"], meta["h"] = img.size
alpha = img.mode in ("RGBA", "LA") or (
img.mode == "P" and "transparency" in img.info)
if animated:
# Ableiten würde die Animation zerstören — Original anzeigen.
meta["display"] = None
elif DISPLAY_MAX_EDGE and max(img.size) > DISPLAY_MAX_EDGE:
out = img.copy()
out.thumbnail((DISPLAY_MAX_EDGE, DISPLAY_MAX_EDGE), Image.LANCZOS)
name = "%s-d.%s" % (key, "png" if alpha else "jpg")
_save(out, CACHE_PATH / name, alpha)
meta["display"] = name
meta["w"], meta["h"] = out.size
elif src.suffix.lower() in (".jpg", ".jpeg", ".png", ".webp", ".gif"):
meta["display"] = None # klein und web-tauglich
else:
name = "%s-d.%s" % (key, "png" if alpha else "jpg")
_save(img.copy(), CACHE_PATH / name, alpha)
meta["display"] = name
thumb = CACHE_PATH / ("%s-t.jpg" % key)
if not thumb.exists():
tn = img.copy()
tn.thumbnail((THUMB_MAX_EDGE, THUMB_MAX_EDGE), Image.LANCZOS)
_save(tn, thumb, False)
meta["ready"] = True
def raster_ready(key, meta):
if not meta.get("ready"):
return False
if not (CACHE_PATH / ("%s-t.jpg" % key)).exists():
return False
display = meta.get("display")
return True if display is None else (CACHE_PATH / display).exists()
# ── Ableitungen: PDF ────────────────────────────────────────────────────────
def pdf_page_count(src):
try:
out = subprocess.run([PDFINFO, str(src)], capture_output=True, timeout=20,
check=False).stdout.decode("utf-8", "replace")
except (OSError, subprocess.SubprocessError):
return 0
match = re.search(r"^Pages:\s+(\d+)", out, re.M)
return min(int(match.group(1)), PDF_MAX_PAGES) if match else 0
def pdf_page_file(key, page):
return CACHE_PATH / ("%s-p%03d.jpg" % (key, page))
def render_pdf_page(src, key, page):
"""Eine einzelne PDF-Seite rastern (-singlefile hängt .jpg selbst an).
Erst unter einem Nebennamen rendern, dann umbenennen: eine abgebrochene
Rasterung — Timeout, oder zwei Aufrufe gleichzeitig auf derselben Seite —
hinterließe sonst ein halbes Bild unter dem endgültigen Namen. Fertig ist
hier gleichbedeutend mit „Datei existiert", also würde es nie wieder
erneuert und dauerhaft kaputt ausgeliefert."""
final = pdf_page_file(key, page)
prefix = CACHE_PATH / ("%s-p%03d.part%d" % (key, page, os.getpid()))
produced = Path(str(prefix) + ".jpg")
cmd = [PDFTOPPM, "-jpeg", "-jpegopt", "quality=%d" % JPEG_QUALITY,
"-r", str(PDF_DPI), "-f", str(page), "-l", str(page),
"-singlefile", str(src), str(prefix)]
try:
result = subprocess.run(cmd, capture_output=True, timeout=90, check=False)
except (OSError, subprocess.SubprocessError):
result = None
if result is None or result.returncode != 0 or not produced.exists():
try:
produced.unlink()
except OSError:
pass
return False
os.replace(str(produced), str(final))
_thumb_from(final, CACHE_PATH / ("%s-p%03dt.jpg" % (key, page)))
return True
def _thumb_from(image_path, thumb_path):
if thumb_path.exists():
return
try:
Image, _ = _pillow()
with Image.open(image_path) as img:
tn = img.copy()
tn.thumbnail((THUMB_MAX_EDGE, THUMB_MAX_EDGE), Image.LANCZOS)
_save(tn, thumb_path, False)
except Exception:
pass
# ── Arbeitsplanung ──────────────────────────────────────────────────────────
# Eine "Slot" ist eine künftige Slide: entweder ein Bild oder eine PDF-Seite.
# Erst wird geplant (billig, nur Dateiprüfungen), dann priorisiert, dann so
# viel abgearbeitet, wie das Zeitbudget hergibt.
def plan_slots(sources):
"""Slots aller Quellen in Anzeigereihenfolge, je mit Fertig-Kennzeichen."""
slots = []
for order, (rel, path, stat) in enumerate(sources):
key = cache_key(rel, stat)
meta = _meta_load(key)
suffix = path.suffix.lower()
common = {"key": key, "rel": rel, "path": path, "meta": meta, "order": order}
if suffix in VECTOR_EXTENSIONS:
if MEDIA_URL: # ohne URL nicht darstellbar
slots.append(dict(common, kind="svg", page=1, index=0, ready=True))
elif suffix in DOCUMENT_EXTENSIONS:
pages = meta.get("pages")
if pages is None:
if not _budget_left(PROBE_BUDGET_SECONDS):
slots.append(dict(common, kind="probe", page=0, index=0, ready=False))
continue
pages = pdf_page_count(path)
meta["pages"] = pages
_meta_save(key, meta)
for page in range(1, pages + 1):
slots.append(dict(common, kind="pdf", page=page, index=page - 1,
ready=pdf_page_file(key, page).exists()))
else:
slots.append(dict(common, kind="raster", page=1, index=0,
ready=raster_ready(key, meta)))
return slots
def work_order(slots, near_id):
"""Reihenfolge der offenen Arbeit: erst eine Slide je Datei (damit die Show
sofort vollständig besetzt ist), dann nach Nähe zur betrachteten Stelle."""
# Was schon einmal gescheitert ist, bleibt liegen: eine kaputte 10-MB-Datei
# würde sonst bei jedem Aufruf aufs Neue Rechenzeit verbrennen. Ein neuer
# Stand der Datei bekommt einen neuen Schlüssel und damit einen neuen Versuch.
todo = [s for s in slots if not s["ready"] and not s["meta"].get("error")]
if not todo:
return todo
anchor = 0
for position, slot in enumerate(slots):
if slot_id(slot) == near_id:
anchor = position
break
position_of = {id(s): p for p, s in enumerate(slots)}
def priority(slot):
first = 0 if (PRIORITY_FIRST_SLIDE and slot["index"] == 0) else 1
return (first, abs(position_of[id(slot)] - anchor), slot["order"], slot["index"])
return sorted(todo, key=priority)
def do_work(todo):
"""Offene Slots abarbeiten, bis das Zeitbudget aufgebraucht ist."""
touched = {}
for slot in todo:
if not _budget_left():
break
try:
if slot["kind"] == "raster":
derive_raster(slot["path"], slot["key"], slot["meta"])
slot["ready"] = raster_ready(slot["key"], slot["meta"])
touched[slot["key"]] = slot["meta"]
elif slot["kind"] == "pdf":
slot["ready"] = render_pdf_page(slot["path"], slot["key"], slot["page"])
elif slot["kind"] == "probe":
slot["meta"]["pages"] = pdf_page_count(slot["path"])
touched[slot["key"]] = slot["meta"] # Seiten folgen beim nächsten Mal
except Exception as exc:
# Eine kaputte Datei darf nie die ganze Show kippen: merken,
# überspringen, nicht erneut versuchen.
slot["meta"]["error"] = str(exc)[:200]
slot["meta"]["ready"] = True
slot["ready"] = False
touched[slot["key"]] = slot["meta"]
for key, meta in touched.items():
_meta_save(key, meta)
# ── Manifest ────────────────────────────────────────────────────────────────
def slot_id(slot):
return "%s-p%03d" % (slot["key"], slot["page"]) if slot["kind"] == "pdf" else slot["key"]
def _media_url(rel, key=""):
"""URL eines Originals — mit Kennung des Dateistands.
Originale werden unter ihrem eigenen Namen ausgeliefert und dürfen wie die
Ableitungen lange im Browser-Cache liegen. Ohne die Kennung zeigte der
Browser nach dem Ersetzen einer Datei tagelang weiter das alte Bild: der
Schlüssel steckt Änderungszeit und Größe ein, die URL ändert sich also
genau dann, wenn sich die Datei ändert."""
if not MEDIA_URL:
return None
url = "%s/%s" % (MEDIA_URL.rstrip("/"), quote(rel))
return "%s?v=%s" % (url, key) if key else url
def _cache_url(name):
return "%s/%s" % (CACHE_URL.rstrip("/"), quote(name))
def slot_to_slide(slot):
"""Fertigen Slot in einen Manifest-Eintrag übersetzen."""
key, rel, meta = slot["key"], slot["rel"], slot["meta"]
title = Path(rel).stem
original = _media_url(rel, key) if SHOW_DOWNLOAD else None
if slot["kind"] == "pdf":
pages = meta.get("pages") or 1
return {
"id": slot_id(slot), "kind": "pdf", "page": slot["page"], "source": rel,
"title": title if pages == 1 else "%s · S. %d" % (title, slot["page"]),
"src": _cache_url("%s-p%03d.jpg" % (key, slot["page"])),
"thumb": _cache_url("%s-p%03dt.jpg" % (key, slot["page"])),
"original": original,
}
if slot["kind"] == "svg":
url = _media_url(rel, key)
return {"id": key, "kind": "svg", "page": 1, "source": rel, "title": title,
"src": url, "thumb": url, "original": original}
display = meta.get("display")
src = _cache_url(display) if display else _media_url(rel, key)
if not src: # Original nicht erreichbar
return None
return {"id": key, "kind": "image", "page": 1, "source": rel, "title": title,
"src": src, "thumb": _cache_url("%s-t.jpg" % key),
"w": meta.get("w"), "h": meta.get("h"), "original": original}
def build_manifest(near_id=""):
CACHE_PATH.mkdir(parents=True, exist_ok=True)
slots = plan_slots(scan_sources())
do_work(work_order(slots, near_id))
slides = [s for s in (slot_to_slide(x) for x in slots if x["ready"]) if s]
pending = sum(1 for s in slots if not s["ready"] and not s["meta"].get("error"))
failed = sum(1 for s in slots if s["meta"].get("error"))
if PRUNE_ORPHANS and not pending and _budget_left():
prune_orphans({s["key"] for s in slots})
return {"version": 1, "count": len(slides), "pending": pending, "failed": failed,
"poll": POLL_BUSY_SECONDS if pending else POLL_SECONDS,
"slides": slides}
_CACHE_FILE_RE = re.compile(r"^([0-9a-f]{20})[-.]")
def prune_orphans(live_keys):
"""Ableitungen entfernen, deren Quelle gelöscht oder verändert wurde.
Läuft nur, wenn nichts mehr offen ist — sonst würde gerade entstehende
Arbeit weggeräumt."""
try:
entries = list(CACHE_PATH.iterdir())
except OSError:
return
for entry in entries:
match = _CACHE_FILE_RE.match(entry.name)
if match and match.group(1) not in live_keys:
try:
entry.unlink()
except OSError:
pass
# ── Suche ───────────────────────────────────────────────────────────────────
# Ein SVG ist XML: sein Text steht lesbar darin und lässt sich ohne Bilderkennung
# einsammeln. Dasselbe gilt für die Metadaten der anderen Formate: EXIF-Felder
# eines Fotos, Info-Felder und Seitentext eines PDF (pdftotext). Kein
# Volltextindex neben den Dateien, sondern das Gelesene im selben Meta-Eintrag,
# den die Ableitungen schon benutzen. Der Schlüssel darüber (cache_key) steckt
# Änderungszeit und Größe ein: eine geänderte Datei bekommt einen neuen Schlüssel
# und wird dadurch von selbst neu gelesen. Es gibt deshalb nichts, was veralten
# könnte — außer der Extraktion selbst; deshalb trägt der Eintrag eine
# Schema-Nummer, und ein alter Stand wird bei nächster Gelegenheit neu gelesen.
SEARCH_SCHEMA = 2 # bei geänderter Extraktion hochzählen: alle
# Meta-Einträge werden dann von selbst neu gelesen
SVG_MAX_BYTES = 8 * 1024 * 1024 # größere Vektordateien nicht einlesen
_TEXT_TAGS = ("text", "title", "desc")
def _collapse(value):
return re.sub(r"\s+", " ", value).strip()
def extract_svg_text(path):
"""Text eines SVG einsammeln: <title>, <desc>, alle <text>-Blöcke — und den
<metadata>-Block, in dem Werkzeuge wie Inkscape Titel, Beschreibung und
Stichwörter ablegen (RDF/Dublin Core; die Tag-Namen darin sind egal, der
Suche genügt der reine Text).
`itertext()` nimmt die Kinder mit — die Zeilen einer Karte stehen meist als
<tspan> innerhalb eines <text>, und die gehören zusammen."""
try:
if path.stat().st_size > SVG_MAX_BYTES:
return ""
raw = path.read_text("utf-8", "replace")
except OSError:
return ""
parts, meta_parts = [], []
try:
from xml.etree import ElementTree
skip = set() # ein <title> IN <metadata> zählt
for element in ElementTree.fromstring(raw).iter(): # nicht doppelt
tag = element.tag.rsplit("}", 1)[-1] if isinstance(element.tag, str) else ""
if tag == "metadata":
meta_parts.append(" ".join(element.itertext()))
skip.update(id(child) for child in element.iter())
elif id(element) not in skip and tag in _TEXT_TAGS:
parts.append("".join(element.itertext()))
except Exception:
# Nicht wohlgeformt: dann eben grob. Ein paar Zeilen Text sind mehr wert
# als eine Karte, die sich nicht finden lässt.
from html import unescape
found = re.findall(r"<(text|title|desc|metadata)\b[^>]*>(.*?)</\1\s*>", raw, re.S | re.I)
parts = [unescape(re.sub(r"<[^>]+>", " ", body)) for _tag, body in found]
return _collapse(" ".join(parts + meta_parts))[:SEARCH_MAX_CHARS]
# EXIF-Felder mit lesbarem Text: Beschreibung, Urheber, Copyright und die
# XP*-Felder, in die der Windows-Explorer Titel/Stichwörter/Kommentar schreibt.
_EXIF_TAGS = (270, 315, 33432, 40091, 40092, 40093, 40094, 40095)
def _exif_str(value):
"""EXIF-Werte sind mal str, mal Bytes: die XP*-Felder sind UTF-16-kodiert,
UserComment trägt eine 8-Byte-Kennung vor dem Text."""
if isinstance(value, (tuple, list)): # manche Pillow-Stände: Byte-Tupel
try:
value = bytes(bytearray(part & 0xFF for part in value))
except (TypeError, ValueError):
return ""
if isinstance(value, bytes):
try:
if value[:8] == b"ASCII\x00\x00\x00":
return value[8:].decode("ascii", "replace")
if value[:8] == b"UNICODE\x00":
return value[8:].decode("utf-16", "replace")
return value.decode("utf-16-le", "replace")
except Exception:
return ""
return str(value) if value is not None else ""
def extract_raster_text(path):
"""Metadaten eines Rasterbildes: EXIF-Beschreibung, Stichwörter, Urheber,
Aufnahmedatum. Liest nur den Dateikopf, nie die Pixel — das ist billig
genug, um es im Suchbudget zu erledigen."""
parts = []
try:
Image, _ = _pillow()
with Image.open(path) as img:
exif = img.getexif()
for tag in _EXIF_TAGS:
if tag in exif:
parts.append(_exif_str(exif.get(tag)))
detail = exif.get_ifd(0x8769) # DateTimeOriginal, UserComment
taken = _exif_str(detail.get(36867))
if taken:
# "2026:07:28 10:22:33" -> "2026-07-28 10:22:33", damit das
# Aufnahmedatum dieselbe Schreibweise hat wie das Dateidatum
parts.append(taken.replace(":", "-", 2))
parts.append(_exif_str(detail.get(37510)))
except Exception:
return ""
return _collapse(" ".join(p.replace("\x00", " ") for p in parts if p))[:SEARCH_MAX_CHARS]
def extract_pdf_search(path):
"""Info-Felder und Seitentext eines PDF.
pdftotext trennt Seiten mit Seitenvorschub (\\f) — je Seite entsteht so ein
eigener Sucheintrag, und ein Treffer führt direkt auf die richtige Seite.
SEARCH_MAX_CHARS gilt je Datei: die Seiten werden der Reihe nach
aufgenommen, bis das Kontingent aufgebraucht ist. Ein PDF ohne Textebene
(gescannt) liefert nichts und bleibt über den Dateinamen auffindbar."""
info = []
try:
out = subprocess.run([PDFINFO, str(path)], capture_output=True, timeout=20,
check=False).stdout.decode("utf-8", "replace")
for field in ("Title", "Subject", "Keywords", "Author"):
match = re.search(r"^%s:\s*(.+)$" % field, out, re.M)
value = _collapse(match.group(1)) if match else ""
if value and value.lower() != path.stem.lower(): # Titel = Dateiname
info.append(value) # trägt nichts bei
except (OSError, subprocess.SubprocessError):
pass
pages = []
try:
result = subprocess.run(
[PDFTOTEXT, "-q", "-enc", "UTF-8", "-l", str(PDF_MAX_PAGES), str(path), "-"],
capture_output=True, timeout=30, check=False)
if result.returncode == 0:
quota = SEARCH_MAX_CHARS
for page_text in result.stdout.decode("utf-8", "replace").split("\f"):
text = _collapse(page_text)[:max(quota, 0)]
quota -= len(text)
pages.append(text)
except (OSError, subprocess.SubprocessError):
pass
while pages and not pages[-1]: # pdftotext endet mit \f
pages.pop()
return " ".join(info), pages
def extract_search(path):
"""Suchtext einer Datei nach ihrem Typ — Fehler machen die Datei nicht
unauffindbar, sie bleibt es über den Namen."""
found = {"v": SEARCH_SCHEMA, "text": ""}
suffix = path.suffix.lower()
try:
if suffix in VECTOR_EXTENSIONS:
found["text"] = extract_svg_text(path)
elif suffix in DOCUMENT_EXTENSIONS:
found["text"], found["pages"] = extract_pdf_search(path)
else:
found["text"] = extract_raster_text(path)
except Exception:
pass
return found
def build_search_index():
"""Suchbares je Quelle: Dateiname und Änderungsdatum immer, dazu, was die
Datei selbst hergibt — SVG-Text samt <metadata>, EXIF-Felder, PDF-Info und
PDF-Seitentext. Ein PDF liefert ab Seite 2 je Seite mit Text einen eigenen
Eintrag unter der Kennung der Slide — ein Treffer führt so direkt auf die
richtige Seite; Seite 1 gehört zum Dateieintrag, dessen Schlüssel ohnehin
dorthin führt.
Der Eintrag trägt denselben Schlüssel wie die Ableitungen und liegt im
selben Meta-Eintrag. Schreiben hier und Aufbereiten im Manifest können sich
kreuzen; der letzte gewinnt, und was dabei verloren geht, wird beim
nächsten Aufruf an seinem Fehlen erkannt und neu erzeugt."""
entries, touched, pending = [], {}, 0
for rel, path, stat in scan_sources():
key = cache_key(rel, stat)
stem = Path(rel).stem
meta = _meta_load(key)
found = meta.get("search")
if not (isinstance(found, dict) and found.get("v") == SEARCH_SCHEMA):
# "oder noch gar nichts geschafft": ein zu knapp gesetztes Budget
# ließe sonst jeden Aufruf ohne Fortschritt enden — der Client
# fragte dann ewig nach. So kommt in jedem Aufruf mindestens eine
# Datei hinzu, und der Bestand wird in endlich vielen Runden voll.
if _budget_left(SEARCH_BUDGET_SECONDS) or not touched:
found = meta["search"] = extract_search(path)
meta.pop("text", None) # Altbestand vor der Schema-Nummer
touched[key] = meta
else:
pending += 1 # der nächste Aufruf liest es nach
found = {}
pages = found.get("pages") or []
parts = [found.get("text", "")] + pages[:1]
parts.append(time.strftime("%Y-%m-%d", time.localtime(stat.st_mtime)))
entries.append({"k": key, "t": stem, "s": rel,
"x": _collapse(" ".join(parts))})
for number, page_text in enumerate(pages[1:PDF_MAX_PAGES], 2):
if page_text:
entries.append({"k": "%s-p%03d" % (key, number),
"t": "%s · S. %d" % (stem, number),
"s": rel, "x": page_text})
for key, meta in touched.items():
_meta_save(key, meta)
return {"version": 2, "pending": pending, "entries": entries}
# ── Ausgabe ─────────────────────────────────────────────────────────────────
def _write(header, body):
sys.stdout.write(header)
sys.stdout.flush()
sys.stdout.buffer.write(body.encode("utf-8"))
def serve_manifest(params):
"""JSON-Liste. Nie zwischenspeichern, sonst erscheinen neue Dateien nicht.
Ohne gültige Sitzung 401 — die Seite lädt dann neu und durchläuft die
Challenge; das Nachfragen selbst zählt nicht aufs Rate-Limit, sonst
sperrte sich eine offene Slideshow nach wenigen Minuten selbst aus."""
if GUARD_TOOL_ID and not (valid_session() or valid_admin()):
_write("Status: 401 Unauthorized\r\n"
"Content-Type: application/json; charset=utf-8\r\n"
"Cache-Control: no-store\r\n\r\n", json.dumps({"error": "session"}))
return
near = re.sub(r"[^0-9a-fp-]", "", params.get("near", [""])[0])[:32]
data = build_manifest(near)
_write("Content-Type: application/json; charset=utf-8\r\n"
"Cache-Control: no-store, max-age=0\r\n"
"X-Robots-Tag: noindex, nofollow\r\n\r\n",
json.dumps(data, ensure_ascii=False))
def serve_search():
"""Suchbestand als JSON — einmal geholt, dann filtert der Client selbst.
Bewusst nicht im Manifest: das wird alle POLL_SECONDS neu geholt und trüge
den Text dann bei jedem Durchgang erneut durch die Leitung. Hier reicht
Nachfragen mit ETag, und in aller Regel antwortet das mit 304."""
if not SEARCH_ENABLED:
_write("Status: 404 Not Found\r\n"
"Content-Type: application/json; charset=utf-8\r\n\r\n",
json.dumps({"error": "disabled"}))
return
if GUARD_TOOL_ID and not (valid_session() or valid_admin()):
_write("Status: 401 Unauthorized\r\n"
"Content-Type: application/json; charset=utf-8\r\n"
"Cache-Control: no-store\r\n\r\n", json.dumps({"error": "session"}))
return
body = json.dumps(build_search_index(), ensure_ascii=False)
etag = '"%s"' % hashlib.sha256(body.encode("utf-8")).hexdigest()[:24]
header = ('ETag: %s\r\n'
'Cache-Control: private, no-cache\r\n'
'X-Robots-Tag: noindex, nofollow\r\n' % etag)
if os.environ.get("HTTP_IF_NONE_MATCH", "").strip() == etag:
_write("Status: 304 Not Modified\r\n" + header + "\r\n", "")
return
_write("Content-Type: application/json; charset=utf-8\r\n" + header + "\r\n", body)
def serve_page():
from jinja2 import Environment, FileSystemLoader, select_autoescape
env = Environment(loader=FileSystemLoader(str(SCRIPT_DIR)),
autoescape=select_autoescape(["html", "xml"]))
html = env.get_template(TEMPLATE).render(
title=TITLE, subtitle=SUBTITLE, empty_hint=EMPTY_HINT,
footer_note=FOOTER_NOTE, footer_url=FOOTER_URL,
poll_seconds=POLL_SECONDS, poll_busy_seconds=POLL_BUSY_SECONDS,
autoplay_seconds=AUTOPLAY_SECONDS,
show_captions=SHOW_CAPTIONS, show_download=SHOW_DOWNLOAD,
show_search=SEARCH_ENABLED,
)
_write("Content-Type: text/html; charset=utf-8\r\n"
"Cache-Control: no-store\r\n"
"X-Robots-Tag: noindex, nofollow\r\n\r\n", html)
def main():
params = parse_qs(os.environ.get("QUERY_STRING", ""))
if params.get("manifest"):
serve_manifest(params)
return
if params.get("search"):
serve_search()
return
# Seiteneinstieg: voller Guard (UA-Filter, JS-Challenge, Rate-Limit).
if GUARD_TOOL_ID and not _guard(GUARD_TOOL_ID):
return # Guard hat schon geantwortet
serve_page()
if __name__ == "__main__":
main()