dejanandClaude Opus 5.5
5a65b5444c
Ist der Port belegt, bricht der Server mit einem Hinweis ab, statt auszuweichen. So stimmt der Link aus dem Marketing-Dashboard immer. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
389 lines
16 KiB
Python
Executable File
389 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Static server for the post generator, plus a save and a state endpoint.
|
|
|
|
The browser download route turned out to be unreliable here: with a data: URL it
|
|
opened a blank tab, with a blob: URL it opened the image in a tab. Both times the
|
|
download attribute was ignored. Since this only ever runs locally, the generator
|
|
now POSTs the rendered bytes to /save and the file is written straight to the
|
|
Downloads folder. No browser download machinery involved.
|
|
|
|
Der Projektstand liegt nicht mehr nur im localStorage des Browsers, sondern
|
|
unter state/<projekt>.json im Repo. Nur so ueberlebt er einen Rechnerwechsel:
|
|
gepusht und gezogen hat die andere Maschine denselben Stand. Der localStorage
|
|
bleibt als Zwischenspeicher, die Datei ist die Wahrheit.
|
|
|
|
Eigene Bilder kommen ueber /upload herein und landen unverfaelscht als Datei
|
|
unter assets/eigene/. Der Weg ueber den Projektstand waere der falsche: ein
|
|
2-MB-PNG als Base64 blaeht state/post1.json auf und macht jeden Diff
|
|
unlesbar. So steht im Stand nur der Pfad, und das Bild liegt mit im Repo --
|
|
dasselbe Prinzip wie bei allen anderen Assets.
|
|
"""
|
|
|
|
import http.server
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import socketserver
|
|
import subprocess
|
|
import threading
|
|
import urllib.parse
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
PORT = 4321
|
|
ROOT = Path(__file__).resolve().parent
|
|
TARGET = Path.home() / "Downloads"
|
|
STATE_DIR = ROOT / "state"
|
|
EIGENE_DIR = ROOT / "assets" / "eigene"
|
|
SAFE_NAME = re.compile(r"^[A-Za-z0-9._-]{1,120}$")
|
|
SAFE_PROJECT = re.compile(r"^[a-z0-9][a-z0-9-]{0,39}$")
|
|
MAX_STATE = 4 * 1024 * 1024
|
|
MAX_BILD = 40 * 1024 * 1024
|
|
# Nur verlustfreie oder unveraenderte Rasterbilder. Die Bytes werden nicht
|
|
# angefasst, deshalb bleibt ein PNG genau das PNG, das der Finder geliefert hat.
|
|
BILD_ENDUNGEN = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".avif"}
|
|
BILD_ORDNER = ("assets/screens", "assets/eigene")
|
|
ANIM_DIR = ROOT / "animation"
|
|
SAFE_KEY = re.compile(r"^[A-Za-z][A-Za-z0-9-]{0,39}$")
|
|
ANIM_FEHLER = re.compile(r"^[A-Za-z]*Error: ")
|
|
# Laufende und abgeschlossene Videobauten, nach Kennung. Der Bau dauert je
|
|
# nach Laenge ein bis drei Minuten -- zu lang fuer eine offene Anfrage, und
|
|
# animation.mjs holt sich waehrenddessen selbst Seiten von diesem Server.
|
|
# Deshalb laeuft er in einem eigenen Faden und die Seite fragt den Stand ab.
|
|
ANIM_LAEUFE = {}
|
|
ANIM_SPERRE = threading.Lock()
|
|
|
|
|
|
class Handler(http.server.SimpleHTTPRequestHandler):
|
|
def __init__(self, *a, **kw):
|
|
super().__init__(*a, directory=str(ROOT), **kw)
|
|
|
|
def _project(self):
|
|
"""Projektname aus der Anfrage, oder None wenn er nicht taugt."""
|
|
q = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
|
|
name = (q.get("project") or [""])[0]
|
|
return name if SAFE_PROJECT.match(name) else None
|
|
|
|
def _state_path(self, name):
|
|
# Der Name ist geprueft, aber der Pfad wird trotzdem gegengerechnet:
|
|
# ein Schreibziel ausserhalb von state/ darf es nicht geben.
|
|
p = (STATE_DIR / f"{name}.json").resolve()
|
|
return p if p.parent == STATE_DIR.resolve() else None
|
|
|
|
def do_DELETE(self):
|
|
if urllib.parse.urlparse(self.path).path != "/state":
|
|
self.send_error(404)
|
|
return
|
|
name = self._project()
|
|
if not name:
|
|
self._json(400, '{"error":"ungueltiges Projekt"}')
|
|
return
|
|
p = self._state_path(name)
|
|
if p and p.exists():
|
|
p.unlink()
|
|
print(f"Stand verworfen: {p.name}")
|
|
self._json(200, '{"ok":true}')
|
|
|
|
def _put_state(self):
|
|
name = self._project()
|
|
if not name:
|
|
self._json(400, '{"error":"ungueltiges Projekt"}')
|
|
return
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
if length <= 0 or length > MAX_STATE:
|
|
self._json(400, '{"error":"ungueltige Groesse"}')
|
|
return
|
|
raw = self.rfile.read(length)
|
|
try:
|
|
data = json.loads(raw)
|
|
except ValueError:
|
|
self._json(400, '{"error":"kein JSON"}')
|
|
return
|
|
p = self._state_path(name)
|
|
if p is None:
|
|
self._json(400, '{"error":"ungueltiger Pfad"}')
|
|
return
|
|
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
|
# Eingerueckt und mit sortierten Schluesseln, damit git einen lesbaren
|
|
# Unterschied zeigt statt einer einzigen langen Zeile. Erst in eine
|
|
# Nebendatei, dann umbenennen: ein Abbruch mittendrin darf keinen
|
|
# halben Stand hinterlassen.
|
|
tmp = p.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(data, indent=1, sort_keys=True,
|
|
ensure_ascii=False) + "\n", encoding="utf-8")
|
|
tmp.replace(p)
|
|
self._json(200, '{"ok":true}')
|
|
|
|
def do_GET(self):
|
|
"""Vor der Dateiauslieferung nur die Bilderliste abfangen."""
|
|
zerlegt = urllib.parse.urlparse(self.path)
|
|
if zerlegt.path == "/":
|
|
# Die Wurzel fuehrt zur Uebersicht aller Projekte statt zur
|
|
# Dateiliste. So reicht ein Link auf http://127.0.0.1:4321/,
|
|
# zum Beispiel aus dem Marketing-Dashboard.
|
|
self.send_response(302)
|
|
self.send_header("Location", "/uebersicht.html")
|
|
self.end_headers()
|
|
return
|
|
if zerlegt.path == "/bilder":
|
|
self._liste_bilder()
|
|
return
|
|
if zerlegt.path == "/anim":
|
|
self._anim_stand(urllib.parse.parse_qs(zerlegt.query).get("id", [""])[0])
|
|
return
|
|
super().do_GET()
|
|
|
|
def _anim_stand(self, kennung):
|
|
"""Stand eines Videobaus. Unbekannte Kennung ist kein Serverfehler,
|
|
sondern eine Antwort: der Lauf ist weg, mehr steht dazu nicht an."""
|
|
with ANIM_SPERRE:
|
|
lauf = ANIM_LAEUFE.get(kennung)
|
|
if lauf is None:
|
|
self._json(404, json.dumps({"laeuft": False, "fehler": "Lauf unbekannt"}))
|
|
return
|
|
self._json(200, json.dumps(lauf, ensure_ascii=False))
|
|
|
|
def _anim_bauen(self):
|
|
"""Ein einzelnes MP4 bauen lassen. Getrennt vom Bildexport: die
|
|
Animation ist ein eigenes Element im Carousel und soll auch einzeln
|
|
herauskommen."""
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
if length <= 0 or length > 4096:
|
|
self._json(400, '{"error":"ungueltige Anfrage"}')
|
|
return
|
|
try:
|
|
wunsch = json.loads(self.rfile.read(length))
|
|
projekt, key = wunsch["projekt"], wunsch["key"]
|
|
except Exception:
|
|
self._json(400, '{"error":"kein gueltiges JSON"}')
|
|
return
|
|
if not SAFE_PROJECT.match(str(projekt)) or not SAFE_KEY.match(str(key)):
|
|
self._json(400, '{"error":"unzulaessiger Name"}')
|
|
return
|
|
if not (ROOT / (projekt + ".html")).is_file():
|
|
self._json(404, '{"error":"Projekt nicht gefunden"}')
|
|
return
|
|
|
|
kennung = uuid.uuid4().hex[:12]
|
|
with ANIM_SPERRE:
|
|
ANIM_LAEUFE[kennung] = {"laeuft": True, "zeile": "startet …"}
|
|
threading.Thread(target=_anim_lauf, args=(kennung, projekt, key),
|
|
daemon=True).start()
|
|
self._json(200, json.dumps({"id": kennung}))
|
|
|
|
def _liste_bilder(self):
|
|
"""Alle verwendbaren Bilder, damit die Auswahl im Generator nicht im
|
|
Code gepflegt werden muss und importierte sofort auftauchen."""
|
|
raus = []
|
|
for ordner in BILD_ORDNER:
|
|
verz = ROOT / ordner
|
|
if not verz.is_dir():
|
|
continue
|
|
for f in sorted(verz.iterdir()):
|
|
if f.is_file() and f.suffix.lower() in BILD_ENDUNGEN:
|
|
raus.append({"pfad": "./" + ordner + "/" + f.name,
|
|
"name": f.stem, "ordner": ordner.split("/")[-1]})
|
|
self._json(200, json.dumps(raus, ensure_ascii=False))
|
|
|
|
def _upload(self):
|
|
"""Ein eigenes Bild aufnehmen. Die Bytes gehen unveraendert auf die
|
|
Platte: nicht neu kodiert, nicht skaliert, nicht umbenannt ausser um
|
|
gefaehrliche Zeichen. Verlustfrei heisst hier woertlich unangetastet."""
|
|
name = os.path.basename(urllib.parse.unquote(self.headers.get("X-Filename", "")))
|
|
stamm, endung = os.path.splitext(name)
|
|
endung = endung.lower()
|
|
if endung not in BILD_ENDUNGEN:
|
|
self._json(400, '{"error":"kein Bildformat"}')
|
|
return
|
|
# Aus dem Stamm wird ein unverfaenglicher Dateiname. Umlaute werden
|
|
# umschrieben statt weggeworfen, sonst wird aus "pruefbild" ein
|
|
# "pr-fbild" und der Name sagt nichts mehr.
|
|
for a, b in (("ä", "ae"), ("ö", "oe"), ("ü", "ue"), ("Ä", "Ae"),
|
|
("Ö", "Oe"), ("Ü", "Ue"), ("ß", "ss")):
|
|
stamm = stamm.replace(a, b)
|
|
stamm = re.sub(r"[^A-Za-z0-9._-]+", "-", stamm).strip("-.") or "bild"
|
|
stamm = stamm[:80]
|
|
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
if length <= 0 or length > MAX_BILD:
|
|
self._json(400, '{"error":"ungueltige Groesse"}')
|
|
return
|
|
daten = self.rfile.read(length)
|
|
|
|
EIGENE_DIR.mkdir(parents=True, exist_ok=True)
|
|
ziel = (EIGENE_DIR / (stamm + endung)).resolve()
|
|
if ziel.parent != EIGENE_DIR.resolve():
|
|
self._json(400, '{"error":"ungueltiger Pfad"}')
|
|
return
|
|
# Gleicher Inhalt, gleicher Name: nicht doppelt ablegen.
|
|
n = 1
|
|
while ziel.exists() and ziel.read_bytes() != daten:
|
|
ziel = EIGENE_DIR / (stamm + "-" + str(n) + endung)
|
|
n += 1
|
|
ziel.write_bytes(daten)
|
|
pfad = "./assets/eigene/" + ziel.name
|
|
print(f"importiert: {ziel.name} ({len(daten)/1024:.0f} KB)")
|
|
self._json(200, json.dumps({"pfad": pfad, "name": ziel.stem}))
|
|
|
|
def do_POST(self):
|
|
path = urllib.parse.urlparse(self.path).path
|
|
if path == "/state":
|
|
self._put_state()
|
|
return
|
|
if path == "/upload":
|
|
self._upload()
|
|
return
|
|
if path == "/anim":
|
|
self._anim_bauen()
|
|
return
|
|
if path != "/save":
|
|
self.send_error(404)
|
|
return
|
|
|
|
name = self.headers.get("X-Filename", "")
|
|
# The filename comes from the page, so it is never trusted: no paths, no
|
|
# traversal, plain basename only.
|
|
name = os.path.basename(name)
|
|
if not SAFE_NAME.match(name):
|
|
self._json(400, '{"error":"ungueltiger Dateiname"}')
|
|
return
|
|
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
if length <= 0 or length > 80 * 1024 * 1024:
|
|
self._json(400, '{"error":"ungueltige Groesse"}')
|
|
return
|
|
|
|
data = self.rfile.read(length)
|
|
TARGET.mkdir(parents=True, exist_ok=True)
|
|
dest = TARGET / name
|
|
stem, suffix = dest.stem, dest.suffix
|
|
n = 1
|
|
while dest.exists():
|
|
dest = TARGET / f"{stem}-{n}{suffix}"
|
|
n += 1
|
|
dest.write_bytes(data)
|
|
print(f"gespeichert: {dest} ({len(data)/1024:.0f} KB)")
|
|
self._json(200, '{"path":"%s"}' % dest)
|
|
|
|
def _json(self, code, body):
|
|
payload = body.encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
|
|
def end_headers(self):
|
|
# Werkzeug, kein Auslieferungsserver: ein Asset, das unter demselben
|
|
# Namen ausgetauscht wird, muss sofort ankommen.
|
|
#
|
|
# Dafuer stand hier no-store auf allem. Das Ziel war richtig, der
|
|
# Preis zu hoch: no-store verbietet dem Browser die Kopie ueberhaupt,
|
|
# und damit lud jeder Reload die vollen rund 6 MB neu -- 3 MB
|
|
# Screenshots, 3,3 MB Mockups, die Schriften, die Bibliotheken. Am
|
|
# Ende wartete man beim Neuladen auf Dateien, die sich seit Wochen
|
|
# nicht geaendert hatten.
|
|
#
|
|
# no-cache leistet dasselbe billiger: der Browser darf die Kopie
|
|
# behalten, muss aber vor jeder Benutzung nachfragen. Eine
|
|
# ausgetauschte Datei kommt damit genauso sofort an, eine
|
|
# unveraenderte kostet ein 304 ohne Rumpf. Geprueft: der Handler
|
|
# beantwortet If-Modified-Since bereits mit 304 und 0 Bytes.
|
|
#
|
|
# Seite, Stand und Auskuenfte bleiben bei no-store. Die sind klein,
|
|
# und sie duerfen nie aus einer Kopie kommen.
|
|
pfad = urllib.parse.urlparse(self.path).path
|
|
if pfad.startswith(("/assets/", "/vendor/", "/animation/")):
|
|
self.send_header("Cache-Control", "no-cache")
|
|
else:
|
|
self.send_header("Cache-Control", "no-store, must-revalidate")
|
|
self.send_header("Pragma", "no-cache")
|
|
self.send_header("Expires", "0")
|
|
super().end_headers()
|
|
|
|
def log_message(self, fmt, *args):
|
|
# args[0] ist normalerweise die Anfragezeile, bei send_error aber ein
|
|
# HTTPStatus. Ohne str() flog hier ein TypeError, der den Handler-Thread
|
|
# abbrach: der Browser wartete dann ewig auf eine Antwort, die nie kam,
|
|
# und die Seite blieb bei "Lade Screens ..." stehen.
|
|
first = str(args[0]) if args else ""
|
|
if "POST /save" in first or "DELETE" in first:
|
|
super().log_message(fmt, *args)
|
|
elif not first.startswith(("GET", "POST", "HEAD")):
|
|
# Fehler und Warnungen gehoeren gesehen, sonst sucht man lange.
|
|
super().log_message(fmt, *args)
|
|
|
|
|
|
def _anim_lauf(kennung, projekt, key):
|
|
"""animation.mjs in einem eigenen Faden. Seine Ausgabe wandert zeilenweise
|
|
in den Stand, damit die Seite den Fortschritt zeigen kann statt zwei
|
|
Minuten lang nichts."""
|
|
def melden(**neu):
|
|
with ANIM_SPERRE:
|
|
ANIM_LAEUFE.setdefault(kennung, {}).update(neu)
|
|
|
|
try:
|
|
proc = subprocess.Popen(
|
|
["node", "animation.mjs", projekt + ".html", key],
|
|
cwd=str(ROOT), stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
text=True, bufsize=1)
|
|
ausgabe = []
|
|
for zeile in proc.stdout:
|
|
zeile = zeile.strip()
|
|
if zeile:
|
|
ausgabe.append(zeile)
|
|
melden(zeile=zeile)
|
|
code = proc.wait()
|
|
if code != 0:
|
|
# Die letzte Zeile ist bei node die Versionsangabe und sagt nichts,
|
|
# und der Auszug aus dem Quelltext steht vor der eigentlichen
|
|
# Meldung. Gesucht ist die Zeile, die mit dem Fehlernamen anfaengt.
|
|
grund = next((z for z in ausgabe if ANIM_FEHLER.match(z)), "")
|
|
melden(laeuft=False,
|
|
fehler=grund or (ausgabe[-1] if ausgabe else "node endete mit " + str(code)))
|
|
return
|
|
ziel = ANIM_DIR / (projekt + "-" + key + ".mp4")
|
|
if not ziel.is_file():
|
|
melden(laeuft=False, fehler="kein MP4 entstanden")
|
|
return
|
|
groesse = ziel.stat().st_size
|
|
# Zum Hochladen liegt die Datei am selben Ort wie die Bilder. Im Repo
|
|
# bleibt sie trotzdem: dort ist sie versioniert und hat auf der anderen
|
|
# Maschine nach einem Pull denselben Stand. Ueberschrieben wird in
|
|
# Downloads nichts, genau wie beim Bildexport -- eine Datei, die
|
|
# vielleicht schon hochgeladen wurde, gehoert nicht ersetzt.
|
|
TARGET.mkdir(parents=True, exist_ok=True)
|
|
abgabe = TARGET / ziel.name
|
|
n = 1
|
|
while abgabe.exists():
|
|
abgabe = TARGET / f"{ziel.stem}-{n}{ziel.suffix}"
|
|
n += 1
|
|
shutil.copy2(ziel, abgabe)
|
|
melden(laeuft=False, pfad="~/Downloads/" + abgabe.name,
|
|
groesse=f"{groesse/1024/1024:.1f} MB · auch animation/{ziel.name}",
|
|
zeile="fertig")
|
|
print(f"gespeichert: {abgabe} ({groesse/1024/1024:.1f} MB)")
|
|
except FileNotFoundError:
|
|
melden(laeuft=False, fehler="node nicht gefunden")
|
|
except Exception as e: # der Faden darf den Server nicht mitnehmen
|
|
melden(laeuft=False, fehler=str(e))
|
|
|
|
|
|
class Server(socketserver.ThreadingTCPServer):
|
|
allow_reuse_address = True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Der Port ist fest, damit Links von aussen (Marketing-Dashboard,
|
|
# Lesezeichen) immer stimmen. Ist er belegt, laeuft PostGrid meist schon.
|
|
try:
|
|
httpd = Server(("127.0.0.1", PORT), Handler)
|
|
except OSError:
|
|
print(f"Port {PORT} ist belegt. Laeuft PostGrid schon? Dann einfach http://127.0.0.1:{PORT}/ oeffnen.")
|
|
raise SystemExit(1)
|
|
print(f"PostGrid: http://127.0.0.1:{PORT}/")
|
|
print(f"Speichert nach: {TARGET}")
|
|
with httpd:
|
|
httpd.serve_forever()
|