claude-regelwerk/.vault-tools/trading-plattform-pruefung/ntfy_probe_server.py

"""Nimmt ntfy-Meldungen lokal entgegen und schreibt sie mit, statt sie zu verschicken.

Koordinator-Werkzeug, nicht fuers Repo. Frage (Herkunft: Anforderung Paket D vom 2026-09-18, "eigener Lauf mit
echten Daten ohne echtes Thema"): Welche Meldungen schickt der Paper-Runner oder der Stille-Waechter wirklich
hinaus -- wie viele, mit welchem Titel, welcher Prioritaet, welchem Text? Der Server lauscht nur auf 127.0.0.1,
der Melder erlaubt http genau fuer diese Adresse.
Mitgeschrieben wird je Anfrage eine JSONL-Zeile: Zeit, Methode, Pfad, Title, Priority, Tags, Textkoerper und vom
Authorization-Kopf nur, ob er da ist, sein Schema und seine Laenge -- nie der Wert.
Was er nicht sieht: ob ntfy.sh dieselbe Anfrage annaehme (Themenregeln, Groessengrenzen, Ratenbegrenzung).
Regressionsnachweis: --status 403 laesst jede Anfrage scheitern; der Melder muss dann einen Fehlschlag melden
und der Runner trotzdem mit demselben Exit-Code enden.

Mit --status-folge 403,403,200 antwortet er der Reihe nach mit diesen Codes und bleibt beim letzten; so zeigt
sich, ob der Stille-Waechter eine gescheiterte Meldung wiederholt, bis sie draussen ist. Die Mitschrift nennt
je Zeile den Status, mit dem geantwortet wurde.

Aufruf: python ntfy_probe_server.py <port> <mitschrift.jsonl> [--status 200] [--status-folge 403,200] [--verzoegerung 0.0]
Beenden mit Strg+C oder SIGTERM.
"""

import json
import sys
import threading
import time
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


def baue_handler(mitschrift: str, folge: list[int], verzoegerung: float):
    sperre = threading.Lock()

    def naechster_status() -> int:
        with sperre:
            return folge.pop(0) if len(folge) > 1 else folge[0]

    class Handler(BaseHTTPRequestHandler):
        def _nimm(self) -> None:
            laenge = int(self.headers.get("Content-Length") or 0)
            koerper = self.rfile.read(laenge) if laenge else b""
            auth = self.headers.get("Authorization")
            status = naechster_status()
            zeile = {
                "zeit": datetime.now(timezone.utc).isoformat(),
                "methode": self.command,
                "pfad": self.path,
                "title": self.headers.get("Title"),
                "priority": self.headers.get("Priority"),
                "tags": self.headers.get("Tags"),
                "text": koerper.decode("utf-8", errors="replace"),
                "authorization_da": auth is not None,
                "authorization_schema": auth.split(" ", 1)[0] if auth else None,
                "authorization_laenge": len(auth) if auth else 0,
                "status": status,
            }
            with open(mitschrift, "a", encoding="utf-8") as datei:
                datei.write(json.dumps(zeile, ensure_ascii=False) + "\n")
            if verzoegerung:
                time.sleep(verzoegerung)
            antwort = json.dumps({"id": "probe", "event": "message"}).encode("utf-8")
            self.send_response(status)
            if 300 <= status < 400:
                # Eine echte Weiterleitung auf denselben Server: Folgt der Melder ihr, steht eine zweite Zeile in der Mitschrift.
                self.send_header("Location", f"http://127.0.0.1:{self.server.server_address[1]}/weitergeleitet")
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(antwort)))
            self.end_headers()
            self.wfile.write(antwort)

        do_POST = _nimm
        do_PUT = _nimm

        def log_message(self, *_args) -> None:  # die Mitschrift ist das Protokoll
            return

    return Handler


def main() -> int:
    argumente = sys.argv[1:]
    if len(argumente) < 2:
        print(__doc__)
        return 3
    port, mitschrift = int(argumente[0]), argumente[1]
    status = int(argumente[argumente.index("--status") + 1]) if "--status" in argumente else 200
    folge = [status]
    if "--status-folge" in argumente:
        folge = [int(teil) for teil in argumente[argumente.index("--status-folge") + 1].split(",")]
    verzoegerung = float(argumente[argumente.index("--verzoegerung") + 1]) if "--verzoegerung" in argumente else 0.0
    server = ThreadingHTTPServer(("127.0.0.1", port), baue_handler(mitschrift, folge, verzoegerung))
    print(f"ntfy-Probe lauscht auf 127.0.0.1:{port}, Statusfolge {folge}, Mitschrift {mitschrift}", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    return 0


if __name__ == "__main__":
    sys.exit(main())