Pascal MichelandClaude Opus 5
246af19683
- .vault-tools/plan-kopie/LIESMICH.md - .vault-tools/plan-kopie/rot_nachweis.py - .vault-tools/trading-plattform-pruefung/LIESMICH.md - .vault-tools/trading-plattform-pruefung/dienste_stand.sh - .vault-tools/trading-plattform-pruefung/ha_einrichten.py - .vault-tools/trading-plattform-pruefung/hole_kerzen.py - .vault-tools/trading-plattform-pruefung/neustart_echtlauf.py - .vault-tools/trading-plattform-pruefung/ntfy_lesen.py - .vault-tools/trading-plattform-pruefung/pause_echtlauf.py - .vault-tools/trading-plattform-pruefung/wiederanlauf_probe.ps1 - AI/Regelwerk/codeworkflow.md - AI/Regelwerk/projekte.md - … und 2 weitere Erzeugt vom PostToolUse-Hook .vault-tools/auto-push/auto_push.py. Co-Authored-By: Claude Opus 5 <[email protected]>
53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
"""Liest nur zurueck, was in der letzten Stunde bei ntfy unter dem Thema der Maschine ankam.
|
|
|
|
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 ntfy_lesen.py
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path.cwd()))
|
|
|
|
import requests
|
|
|
|
from trading.paper.melder import lade_melder_config
|
|
from trading.paper.melder import lies_ntfy_geheimnisse
|
|
|
|
|
|
def main() -> int:
|
|
config = lade_melder_config("config/paper.toml", secrets_pfad="config/secrets.toml")
|
|
thema, token = lies_ntfy_geheimnisse("config/secrets.toml")
|
|
kopf = {"Authorization": f"Bearer {token}"} if token else {}
|
|
try:
|
|
antwort = requests.get(
|
|
f"{config.ntfy_url}/{thema}/json", params={"poll": "1", "since": "1h"},
|
|
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: HTTP {antwort.status_code}")
|
|
anzahl = 0
|
|
for zeile in antwort.text.splitlines():
|
|
try:
|
|
eintrag = json.loads(zeile)
|
|
except ValueError:
|
|
continue
|
|
if eintrag.get("event") != "message":
|
|
continue
|
|
anzahl += 1
|
|
# Ohne Angabe fuehrt ntfy die Prioritaet 3 (default) und laesst das Feld weg.
|
|
print(f" bei ntfy: Titel '{eintrag.get('title')}', Prioritaet {eintrag.get('priority', 3)}, "
|
|
f"Zeit {datetime.fromtimestamp(eintrag.get('time', 0)).isoformat()}, "
|
|
f"erste Zeile '{str(eintrag.get('message', '')).splitlines()[0][:70]}'")
|
|
print(f"{anzahl} Meldung(en) in der letzten Stunde.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|