90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
import requests
|
|
import time
|
|
import json
|
|
from bs4 import BeautifulSoup
|
|
from datetime import datetime
|
|
import re
|
|
|
|
TI_STATUS_URL = "https://fachportal.gematik.de/ti-status"
|
|
WEBHOOK_URL = "https://mattermost.medisoftware.org/hooks/i67zgcgajifxxxtfwjxcxace7a"
|
|
STATE_FILE = "ti_status_state.json"
|
|
CHECK_INTERVAL = 300 # 5 Minuten
|
|
|
|
def fetch_status_messages():
|
|
resp = requests.get(TI_STATUS_URL)
|
|
resp.raise_for_status()
|
|
soup = BeautifulSoup(resp.text, "html.parser")
|
|
# Annahme: Meldungen stehen in bestimmten HTML-Elementen (z.B. <div class="c-status-list__item">)
|
|
# Dies muss ggf. angepasst werden, wenn die Struktur anders ist!
|
|
messages = []
|
|
for item in soup.find_all("div", class_="c-status-list__item"):
|
|
title = item.get_text(strip=True)
|
|
if title:
|
|
messages.append(title)
|
|
# Fallback: Wenn keine spezielle Klasse, alle sichtbaren Meldungen im Hauptbereich
|
|
if not messages:
|
|
for p in soup.find_all("p"):
|
|
text = p.get_text(strip=True)
|
|
if text and len(text) > 30:
|
|
messages.append(text)
|
|
return messages
|
|
|
|
def load_state():
|
|
try:
|
|
with open(STATE_FILE, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
return {"messages": []}
|
|
|
|
def save_state(state):
|
|
with open(STATE_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(state, f, ensure_ascii=False, indent=2)
|
|
|
|
def markdownify_message(message):
|
|
# Datumsangaben fett hervorheben (z.B. 23.06.2025)
|
|
message = re.sub(r"(\d{2}\.\d{2}\.\d{4})", r"**\1**", message)
|
|
# URLs als Links darstellen
|
|
message = re.sub(r"(https?://\S+)", r"[\1](\1)", message)
|
|
# Listenpunkte erkennen (z.B. mit - oder *)
|
|
lines = message.splitlines()
|
|
md_lines = []
|
|
for line in lines:
|
|
line = line.strip()
|
|
if line.startswith("-") or line.startswith("*"):
|
|
md_lines.append(f"- {line[1:].strip()}")
|
|
else:
|
|
md_lines.append(line)
|
|
# Absätze durch doppelte Zeilenumbrüche trennen
|
|
md_message = "\n\n".join([l for l in md_lines if l])
|
|
return md_message
|
|
|
|
def send_to_mattermost(message):
|
|
md_message = markdownify_message(message)
|
|
payload = {
|
|
"text": f"#### Neue TI-Status-Meldung\n\n{md_message}\n\n[Zur Statusseite]({TI_STATUS_URL})\n_Gemeldet am {datetime.now().strftime('%d.%m.%Y %H:%M:%S')}_"
|
|
}
|
|
resp = requests.post(WEBHOOK_URL, json=payload)
|
|
resp.raise_for_status()
|
|
|
|
def main():
|
|
state = load_state()
|
|
known_messages = set(state.get("messages", []))
|
|
print("Starte Überwachung der TI-Status-Seite...")
|
|
while True:
|
|
try:
|
|
messages = fetch_status_messages()
|
|
new_messages = [m for m in messages if m not in known_messages]
|
|
for msg in new_messages:
|
|
print(f"Neue Meldung gefunden: {msg}")
|
|
send_to_mattermost(msg)
|
|
known_messages.add(msg)
|
|
if new_messages:
|
|
save_state({"messages": list(known_messages)})
|
|
else:
|
|
print(f"Keine neuen Meldungen ({datetime.now().strftime('%H:%M:%S')})")
|
|
except Exception as e:
|
|
print(f"Fehler: {e}")
|
|
time.sleep(CHECK_INTERVAL)
|
|
|
|
if __name__ == "__main__":
|
|
main() |