From fcb6cd79959cc01c2d4fa49aa9f7c9d4b56f17df Mon Sep 17 00:00:00 2001 From: Markus Busche Date: Tue, 29 Jul 2025 10:22:04 +0200 Subject: [PATCH] =?UTF-8?q?Mock-API=20angepasst:=20/lageapi/v2/tilage=20mi?= =?UTF-8?q?t=20St=C3=B6rung,=20.gitignore=20f=C3=BCr=20*.json=20und=20=5F?= =?UTF-8?q?=5Fpycache=5F=5F/=20aktualisiert?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 5 ++- api_dynamic.py | 70 +++++++++++++++++++++++++++++ requirements_api_mock.txt | 2 + ti_status_checker.py | 93 +++++++++++++++++++++++++++------------ 4 files changed, 142 insertions(+), 28 deletions(-) create mode 100644 api_dynamic.py create mode 100644 requirements_api_mock.txt diff --git a/.gitignore b/.gitignore index 71b98ca..17cdafb 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,5 @@ .venv/ -.env \ No newline at end of file +.env +test* +*.json +__pycache__/ diff --git a/api_dynamic.py b/api_dynamic.py new file mode 100644 index 0000000..6eb57c8 --- /dev/null +++ b/api_dynamic.py @@ -0,0 +1,70 @@ +from fastapi import FastAPI +from pydantic import BaseModel, Field +from typing import List, Optional, Any + +app = FastAPI() + +# Pydantic-Modelle für die Struktur +class AffectedFunction(BaseModel): + function: str + critical: int + impactDesc: str + outage: str # "none"|"partial"|"full" + hasMaintenance: bool + +class AppStatusEntry(BaseModel): + outage: str = "none" + hasMaintenance: bool = False + hasSubComponentMaintenance: bool = False + affectedFunctions: List[AffectedFunction] = Field(default_factory=list) + +class AppStatus(BaseModel): + erezept: Optional[AppStatusEntry] = None + epa: Optional[AppStatusEntry] = None + kim: Optional[AppStatusEntry] = None + wanda: Optional[AppStatusEntry] = None + ogd: Optional[AppStatusEntry] = None + vsdm: Optional[AppStatusEntry] = None + tianschluss: Optional[AppStatusEntry] = None + +class StatusModel(BaseModel): + appStatus: AppStatus + cause: List[Any] = Field(default_factory=list) + +# Initialer Status mit einer Störung bei erezept +status_data = StatusModel( + appStatus=AppStatus( + erezept=AppStatusEntry( + outage="partial", + hasMaintenance=False, + hasSubComponentMaintenance=False, + affectedFunctions=[ + AffectedFunction( + function="Signatur", + critical=1, + impactDesc="Signatur ist zeitweise nicht möglich", + outage="partial", + hasMaintenance=False + ) + ] + ), + epa=AppStatusEntry(), + kim=AppStatusEntry(), + wanda=AppStatusEntry(), + ogd=AppStatusEntry(), + vsdm=AppStatusEntry(), + tianschluss=AppStatusEntry() + ), + cause=[] +) + +@app.get("/lageapi/v2/tilage", response_model=StatusModel) +def get_tilage(): + return status_data + +@app.post("/lageapi/v2/tilage", response_model=StatusModel) +def set_tilage(new_status: StatusModel): + global status_data + status_data = new_status + return status_data + \ No newline at end of file diff --git a/requirements_api_mock.txt b/requirements_api_mock.txt new file mode 100644 index 0000000..f0615cf --- /dev/null +++ b/requirements_api_mock.txt @@ -0,0 +1,2 @@ +fastapi +uvicorn \ No newline at end of file diff --git a/ti_status_checker.py b/ti_status_checker.py index 035e1a6..74dd1ab 100644 --- a/ti_status_checker.py +++ b/ti_status_checker.py @@ -10,7 +10,17 @@ import time # Lade Umgebungsvariablen aus .env Datei load_dotenv() -TI_API_URL = "https://ti-lage.prod.ccs.gematik.solutions/lageapi/v2/tilage" +def is_debug_mode(): + """Prüft ob Debug-Modus aktiviert ist""" + return os.getenv('DBG_MODE', 'false').lower() == 'true' + +def get_ti_api_url(): + """Gibt die API-URL zurück, im Debug-Modus ggf. die lokale Test-API""" + if is_debug_mode(): + return os.getenv("TI_API_URL_DEBUG", "http://localhost:8000/lageapi/v2/tilage") + return os.getenv("TI_API_URL", "https://ti-lage.prod.ccs.gematik.solutions/lageapi/v2/tilage") + +TI_API_URL = get_ti_api_url() STATE_FILE = "ti_status_state.json" # Apprise Konfiguration aus Umgebungsvariablen @@ -46,10 +56,6 @@ def get_apprise_urls(): return urls -def is_debug_mode(): - """Prüft ob Debug-Modus aktiviert ist""" - return os.getenv('DEBUG_MODE', 'false').lower() == 'true' - def get_notification_level(): """Holt das konfigurierte Benachrichtigungslevel""" return os.getenv('NOTIFICATION_LEVEL', 'all').lower() @@ -128,6 +134,7 @@ def fetch_status_messages(): print(f"API-Antwort erhalten. Anzahl Meldungen: {len(data.get('meldungen', []))}") messages = [] + # 1. Bisherige Meldungen for meldung in data.get("meldungen", []): zeit = meldung.get("zeitpunkt", "") titel = meldung.get("titel", "") @@ -135,10 +142,29 @@ def fetch_status_messages(): link = meldung.get("link", "") msg = f"{zeit}\n- {titel}: {beschreibung}\n{link}".strip() messages.append(msg) - if is_debug_mode(): print(f"Verarbeite Meldung: {titel[:50]}...") - + # 2. Neue Auswertung von appStatus + app_status = data.get("appStatus", {}) + for dienst, status in app_status.items(): + outage = status.get("outage", "none") + if outage in ("partial", "full"): + has_maintenance = status.get("hasMaintenance", False) + sub_maintenance = status.get("hasSubComponentMaintenance", False) + affected = status.get("affectedFunctions", []) + # Baue eine verständliche Meldung + msg = f"Störung bei {dienst.upper()} ({'Wartung' if has_maintenance else 'Störung'}): Status: {outage}" + if affected: + for func in affected: + func_name = func.get("function", "Unbekannte Funktion") + impact = func.get("impactDesc", "") + func_outage = func.get("outage", outage) + msg += f"\n- {func_name}: {impact} (Status: {func_outage})" + else: + msg += "\n- Keine weiteren Details." + messages.append(msg) + if is_debug_mode(): + print(f"Erkannte Störung: {msg[:80]}...") if is_debug_mode(): print(f"Insgesamt {len(messages)} Meldungen verarbeitet") return messages @@ -175,48 +201,61 @@ def markdownify_message(message): def send_notification(message): """Sendet Benachrichtigungen an alle konfigurierten Endpunkte""" - + # Debug-Ausgabe: Apprise-Version und URL aus .env + if is_debug_mode(): + print(f"[DEBUG] Apprise-Version: {apprise.__version__}") + print(f"[DEBUG] APPRISE_URL_MATTERMOST aus .env: {os.getenv('APPRISE_URL_MATTERMOST')}") # Prüfe ob Benachrichtigung gesendet werden soll if not should_send_notification(message): if is_debug_mode(): print("🚫 Benachrichtigung wird nicht gesendet (Regeln)") return - + md_message = markdownify_message(message) - + # Hole alle konfigurierten URLs urls = get_apprise_urls() if not urls: print("❌ Keine Apprise URLs konfiguriert!") return - + # Erstelle Apprise Objekt apobj = apprise.Apprise() - + # Füge alle URLs hinzu for url in urls: apobj.add(url) - + # Erstelle die Nachricht title = "Neue TI-Status-Meldung" body = f"{md_message}\n\n[Zur Statusseite](https://fachportal.gematik.de/ti-status)\n_Gemeldet am {datetime.now().strftime('%d.%m.%Y %H:%M:%S')}_" - + if is_debug_mode(): print(f"📤 Sende Benachrichtigung an {len(urls)} Endpunkt(e)") - - # Sende die Nachricht - result = apobj.notify( - title=title, - body=body, - body_format=apprise.NotifyFormat.MARKDOWN - ) - - if result: + print(f"Verwendete Apprise-URLs: {urls}") + print(f"Nachrichtentitel: {title}") + print(f"Nachrichtentext: {body[:200]}...") + + # Sende die Nachricht mit Fehlerausgabe + try: + result = apobj.notify( + title=title, + body=body, + body_format=apprise.NotifyFormat.MARKDOWN + ) + if result: + if is_debug_mode(): + print("✅ Benachrichtigung erfolgreich gesendet") + else: + print("❌ Fehler beim Senden der Benachrichtigung") + if is_debug_mode(): + print("[DEBUG] Apprise notify() Rückgabewert: False") + print("[DEBUG] Prüfe, ob die Webhook-URL korrekt ist, der Zielserver erreichbar ist und keine Authentifizierungsprobleme bestehen.") + except Exception as e: + print("❌ Fehler beim Senden der Benachrichtigung (Exception)") if is_debug_mode(): - print("✅ Benachrichtigung erfolgreich gesendet") - else: - print("❌ Fehler beim Senden der Benachrichtigung") - + print(f"[DEBUG] Exception: {e}") + # Verzögerung zwischen Benachrichtigungen delay = get_notification_delay() if delay > 0: