Compare commits
2 Commits
aa7b088a1a
...
fcb6cd7995
| Author | SHA1 | Date | |
|---|---|---|---|
| fcb6cd7995 | |||
| 93c693f481 |
3
.gitignore
vendored
3
.gitignore
vendored
@@ -1,2 +1,5 @@
|
||||
.venv/
|
||||
.env
|
||||
test*
|
||||
*.json
|
||||
__pycache__/
|
||||
|
||||
234
TI-Status-API.postman_collection.json
Normal file
234
TI-Status-API.postman_collection.json
Normal file
@@ -0,0 +1,234 @@
|
||||
{
|
||||
"info": {
|
||||
"_postman_id": "ti-status-api-collection",
|
||||
"name": "TI-Status API",
|
||||
"description": "Collection für die TI-Status-API der gematik\n\nEndpunkte:\n- GET /lageapi/v2/tilage - Aktuelle TI-Lage Meldungen\n\nVerwendung:\n1. Importiere diese Collection in Postman\n2. Teste die API-Endpunkte\n3. Überprüfe die JSON-Antworten",
|
||||
"schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json"
|
||||
},
|
||||
"item": [
|
||||
{
|
||||
"name": "TI-Lage API",
|
||||
"item": [
|
||||
{
|
||||
"name": "GET Aktuelle TI-Lage Meldungen",
|
||||
"request": {
|
||||
"method": "GET",
|
||||
"header": [
|
||||
{
|
||||
"key": "Accept",
|
||||
"value": "application/json",
|
||||
"type": "text"
|
||||
},
|
||||
{
|
||||
"key": "User-Agent",
|
||||
"value": "TI-Status-Checker/1.0",
|
||||
"type": "text"
|
||||
}
|
||||
],
|
||||
"url": {
|
||||
"raw": "https://ti-lage.prod.ccs.gematik.solutions/lageapi/v2/tilage",
|
||||
"protocol": "https",
|
||||
"host": [
|
||||
"ti-lage",
|
||||
"prod",
|
||||
"ccs",
|
||||
"gematik",
|
||||
"solutions"
|
||||
],
|
||||
"path": [
|
||||
"lageapi",
|
||||
"v2",
|
||||
"tilage"
|
||||
]
|
||||
},
|
||||
"description": "Holt alle aktuellen TI-Lage Meldungen von der gematik API"
|
||||
},
|
||||
"response": [
|
||||
{
|
||||
"name": "Erfolgreiche Antwort",
|
||||
"originalRequest": {
|
||||
"method": "GET",
|
||||
"header": [
|
||||
{
|
||||
"key": "Accept",
|
||||
"value": "application/json",
|
||||
"type": "text"
|
||||
}
|
||||
],
|
||||
"url": {
|
||||
"raw": "https://ti-lage.prod.ccs.gematik.solutions/lageapi/v2/tilage",
|
||||
"protocol": "https",
|
||||
"host": [
|
||||
"ti-lage",
|
||||
"prod",
|
||||
"ccs",
|
||||
"gematik",
|
||||
"solutions"
|
||||
],
|
||||
"path": [
|
||||
"lageapi",
|
||||
"v2",
|
||||
"tilage"
|
||||
]
|
||||
}
|
||||
},
|
||||
"status": "OK",
|
||||
"code": 200,
|
||||
"_postman_previewlanguage": "json",
|
||||
"header": [
|
||||
{
|
||||
"key": "Content-Type",
|
||||
"value": "application/json"
|
||||
}
|
||||
],
|
||||
"cookie": [],
|
||||
"body": "{\n \"meldungen\": [\n {\n \"zeitpunkt\": \"2024-01-15T10:30:00Z\",\n \"titel\": \"Beispiel Meldung\",\n \"beschreibung\": \"Dies ist eine Beispiel-Beschreibung für eine TI-Lage Meldung.\",\n \"link\": \"https://fachportal.gematik.de/ti-status\",\n \"kategorie\": \"wartung\",\n \"prioritaet\": \"normal\"\n }\n ],\n \"letzteAktualisierung\": \"2024-01-15T10:30:00Z\",\n \"anzahlMeldungen\": 1\n}"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "GET API Status (Health Check)",
|
||||
"request": {
|
||||
"method": "GET",
|
||||
"header": [
|
||||
{
|
||||
"key": "Accept",
|
||||
"value": "application/json",
|
||||
"type": "text"
|
||||
}
|
||||
],
|
||||
"url": {
|
||||
"raw": "https://ti-lage.prod.ccs.gematik.solutions/lageapi/v2/health",
|
||||
"protocol": "https",
|
||||
"host": [
|
||||
"ti-lage",
|
||||
"prod",
|
||||
"ccs",
|
||||
"gematik",
|
||||
"solutions"
|
||||
],
|
||||
"path": [
|
||||
"lageapi",
|
||||
"v2",
|
||||
"health"
|
||||
]
|
||||
},
|
||||
"description": "Prüft den Status der API (falls verfügbar)"
|
||||
},
|
||||
"response": []
|
||||
}
|
||||
],
|
||||
"description": "Hauptendpunkte der TI-Status-API"
|
||||
},
|
||||
{
|
||||
"name": "Tests & Beispiele",
|
||||
"item": [
|
||||
{
|
||||
"name": "Test mit verschiedenen Headers",
|
||||
"request": {
|
||||
"method": "GET",
|
||||
"header": [
|
||||
{
|
||||
"key": "Accept",
|
||||
"value": "application/json",
|
||||
"type": "text"
|
||||
},
|
||||
{
|
||||
"key": "Cache-Control",
|
||||
"value": "no-cache",
|
||||
"type": "text"
|
||||
},
|
||||
{
|
||||
"key": "X-Requested-With",
|
||||
"value": "XMLHttpRequest",
|
||||
"type": "text"
|
||||
}
|
||||
],
|
||||
"url": {
|
||||
"raw": "https://ti-lage.prod.ccs.gematik.solutions/lageapi/v2/tilage",
|
||||
"protocol": "https",
|
||||
"host": [
|
||||
"ti-lage",
|
||||
"prod",
|
||||
"ccs",
|
||||
"gematik",
|
||||
"solutions"
|
||||
],
|
||||
"path": [
|
||||
"lageapi",
|
||||
"v2",
|
||||
"tilage"
|
||||
]
|
||||
},
|
||||
"description": "Test mit zusätzlichen Headers für bessere Kompatibilität"
|
||||
},
|
||||
"response": []
|
||||
}
|
||||
],
|
||||
"description": "Zusätzliche Tests und Beispiele für die API"
|
||||
}
|
||||
],
|
||||
"event": [
|
||||
{
|
||||
"listen": "prerequest",
|
||||
"script": {
|
||||
"type": "text/javascript",
|
||||
"exec": [
|
||||
"// Pre-request Script für TI-Status API",
|
||||
"console.log('TI-Status API Request gestartet');",
|
||||
"console.log('URL:', pm.request.url.toString());",
|
||||
"console.log('Method:', pm.request.method);"
|
||||
]
|
||||
}
|
||||
},
|
||||
{
|
||||
"listen": "test",
|
||||
"script": {
|
||||
"type": "text/javascript",
|
||||
"exec": [
|
||||
"// Test Script für TI-Status API",
|
||||
"pm.test('Status Code ist 200', function () {",
|
||||
" pm.response.to.have.status(200);",
|
||||
"});",
|
||||
"",
|
||||
"pm.test('Response ist JSON', function () {",
|
||||
" pm.response.to.be.json;",
|
||||
"});",
|
||||
"",
|
||||
"pm.test('Response hat meldungen Array', function () {",
|
||||
" const jsonData = pm.response.json();",
|
||||
" pm.expect(jsonData).to.have.property('meldungen');",
|
||||
" pm.expect(jsonData.meldungen).to.be.an('array');",
|
||||
"});",
|
||||
"",
|
||||
"pm.test('Meldungen haben korrekte Struktur', function () {",
|
||||
" const jsonData = pm.response.json();",
|
||||
" if (jsonData.meldungen && jsonData.meldungen.length > 0) {",
|
||||
" const meldung = jsonData.meldungen[0];",
|
||||
" pm.expect(meldung).to.have.property('zeitpunkt');",
|
||||
" pm.expect(meldung).to.have.property('titel');",
|
||||
" pm.expect(meldung).to.have.property('beschreibung');",
|
||||
" }",
|
||||
"});",
|
||||
"",
|
||||
"// Log Response für Debugging",
|
||||
"console.log('Response Status:', pm.response.status);",
|
||||
"console.log('Response Time:', pm.response.responseTime + 'ms');",
|
||||
"console.log('Response Size:', pm.response.size().body + ' bytes');"
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
"variable": [
|
||||
{
|
||||
"key": "base_url",
|
||||
"value": "https://ti-lage.prod.ccs.gematik.solutions",
|
||||
"type": "string"
|
||||
},
|
||||
{
|
||||
"key": "api_version",
|
||||
"value": "v2",
|
||||
"type": "string"
|
||||
}
|
||||
]
|
||||
}
|
||||
70
api_dynamic.py
Normal file
70
api_dynamic.py
Normal file
@@ -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
|
||||
|
||||
2
requirements_api_mock.txt
Normal file
2
requirements_api_mock.txt
Normal file
@@ -0,0 +1,2 @@
|
||||
fastapi
|
||||
uvicorn
|
||||
@@ -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,7 +201,10 @@ 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():
|
||||
@@ -203,19 +232,29 @@ def send_notification(message):
|
||||
|
||||
if is_debug_mode():
|
||||
print(f"📤 Sende Benachrichtigung an {len(urls)} Endpunkt(e)")
|
||||
print(f"Verwendete Apprise-URLs: {urls}")
|
||||
print(f"Nachrichtentitel: {title}")
|
||||
print(f"Nachrichtentext: {body[:200]}...")
|
||||
|
||||
# Sende die Nachricht
|
||||
# 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(f"[DEBUG] Exception: {e}")
|
||||
|
||||
# Verzögerung zwischen Benachrichtigungen
|
||||
delay = get_notification_delay()
|
||||
|
||||
Reference in New Issue
Block a user