posts/serve.py

#!/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 socketserver
import urllib.parse
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")


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."""
        if urllib.parse.urlparse(self.path).path == "/bilder":
            self._liste_bilder()
            return
        super().do_GET()

    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 != "/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: Assets werden hier laufend
        # ausgetauscht, und ein gecachter Screenshot unter gleichem Dateinamen
        # kostet mehr Zeit als jeder erneute Ladevorgang.
        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)


class Server(socketserver.ThreadingTCPServer):
    allow_reuse_address = True


if __name__ == "__main__":
    print(f"Generator:  http://127.0.0.1:{PORT}/post1.html")
    print(f"Speichert nach: {TARGET}")
    with Server(("127.0.0.1", PORT), Handler) as httpd:
        httpd.serve_forever()