Pascal MichelandClaude Opus 5
fca08de1c5
- .vault-tools/trading-plattform-pruefung/LIESMICH.md - .vault-tools/trading-plattform-pruefung/ntfy_lesen.py Erzeugt vom PostToolUse-Hook .vault-tools/auto-push/auto_push.py. Co-Authored-By: Claude Opus 5 <[email protected]>
108 lines
4.2 KiB
Python
108 lines
4.2 KiB
Python
"""Liest zurueck, was bei ntfy unter dem Thema der Maschine ankam, mit maskierten Ziffern.
|
|
|
|
Koordinator-Skript, nicht Teil des Repos. Schickt nichts. Das Thema wird nie ausgegeben -- weder
|
|
hier noch in einer Fehlermeldung. Aufruf aus dem Projektordner (WSL):
|
|
python -B ntfy_lesen.py [--seit 12h | --seit 2026-10-07T20:00]
|
|
Exit 0 gelesen, 1 Netzfehler, 3 kein Thema, 4 Weiterleitung oder Fehlerstatus.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from zoneinfo import ZoneInfo
|
|
|
|
sys.path.insert(0, str(Path.cwd()))
|
|
|
|
# Ohne Login-Shell fehlt die Variable (LIESMICH, Abschnitt zu den Geheimnissen); der Pfad selbst
|
|
# ist kein Geheimnis, sein Inhalt schon.
|
|
os.environ.setdefault("TRADING_GEHEIMNISSE", "/home/claude/.config/trading-plattform/secrets.toml")
|
|
|
|
import requests
|
|
|
|
from trading import geheimnisse
|
|
from trading.paper.melder import lade_melder_config
|
|
from trading.paper.melder import lies_ntfy_geheimnisse
|
|
|
|
BERLIN = ZoneInfo("Europe/Berlin")
|
|
# ntfy nimmt fuer since eine Dauer (12h, 30m) oder einen Unix-Zeitstempel (docs.ntfy.sh/subscribe/api).
|
|
DAUER = re.compile(r"^\d+[smhd]$")
|
|
|
|
|
|
def seit_parameter(wert: str) -> str:
|
|
"""Eine Dauer bleibt, wie sie ist; eine Ortszeit ohne Zone gilt als Berliner Zeit."""
|
|
if DAUER.match(wert):
|
|
return wert
|
|
zeit = datetime.fromisoformat(wert)
|
|
if zeit.tzinfo is None:
|
|
zeit = zeit.replace(tzinfo=BERLIN)
|
|
return str(int(zeit.timestamp()))
|
|
|
|
|
|
def maskiert(text: object) -> str:
|
|
"""Jede Ziffernfolge als #, weil eine Meldung einen Betrag tragen kann."""
|
|
return re.sub(r"\d+", "#", str(text))
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
teiler = argparse.ArgumentParser(description="ntfy zuruecklesen, ohne das Thema zu zeigen.")
|
|
teiler.add_argument(
|
|
"--seit", default="12h",
|
|
help="Dauer wie 12h oder Berliner Ortszeit JJJJ-MM-TTTHH:MM; ntfy.sh haelt Meldungen nur Stunden.",
|
|
)
|
|
args = teiler.parse_args(argv)
|
|
try:
|
|
seit = seit_parameter(args.seit)
|
|
except ValueError:
|
|
print(f"--seit nicht lesbar: {args.seit}")
|
|
return 2
|
|
try:
|
|
# Beide lesen die Geheimnisdatei; fehlt sie, wirft schon die Konfiguration.
|
|
config = lade_melder_config("config/paper.toml", secrets_pfad=None)
|
|
thema, token = lies_ntfy_geheimnisse()
|
|
except OSError as ausnahme:
|
|
print(f"Geheimnisdatei nicht lesbar ({type(ausnahme).__name__}): {geheimnisse.beschreibung()}")
|
|
return 3
|
|
if not thema:
|
|
# Mit leerem Thema hiesse die Adresse .../json; genau das endete am 2026-10-07 mit HTTP 307.
|
|
print(f"Kein Thema in der Geheimnisdatei: {geheimnisse.beschreibung()}")
|
|
return 3
|
|
kopf = {"Authorization": f"Bearer {token}"} if token else {}
|
|
try:
|
|
antwort = requests.get(
|
|
f"{config.ntfy_url}/{thema}/json", params={"poll": "1", "since": seit},
|
|
headers=kopf, timeout=10, allow_redirects=False,
|
|
)
|
|
except Exception as ausnahme:
|
|
# Nur der Typ: Die Ausnahme von requests traegt die Adresse und damit das Thema.
|
|
print(f"Ruecklesen gescheitert: {type(ausnahme).__name__}")
|
|
return 1
|
|
print(f"Ruecklesen seit {args.seit}: HTTP {antwort.status_code}")
|
|
if antwort.status_code >= 300:
|
|
# Einer Weiterleitung folgen hiesse, das Token an ein anderes Ziel zu schicken.
|
|
print("Kein Ergebnis: Weiterleitung oder Fehlerstatus, nicht gefolgt.")
|
|
return 4
|
|
anzahl = 0
|
|
for zeile in antwort.text.splitlines():
|
|
try:
|
|
eintrag = json.loads(zeile)
|
|
except ValueError:
|
|
continue
|
|
if eintrag.get("event") != "message":
|
|
continue
|
|
anzahl += 1
|
|
zeit = datetime.fromtimestamp(eintrag.get("time", 0), BERLIN).strftime("%Y-%m-%d %H:%M:%S")
|
|
erste = (str(eintrag.get("message", "")).splitlines() or [""])[0]
|
|
# Ohne Angabe fuehrt ntfy die Prioritaet 3 (default) und laesst das Feld weg.
|
|
print(f" {zeit} Prioritaet {eintrag.get('priority', 3)} Titel '{maskiert(eintrag.get('title'))}' "
|
|
f"erste Zeile '{maskiert(erste)[:70]}'")
|
|
print(f"{anzahl} Meldung(en).")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|