diff --git a/README.md b/README.md index 91d89e0..d273196 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ Ein Python-Skript, das den TI-Status überwacht und neue Meldungen über Apprise - Vermeidet Duplikate durch lokale Statusverfolgung - Debug-Ausgaben für bessere Transparenz - Umfassendes Test-Tool +- [Statistische Erfassung der Störungen](./README_STATISTICS.md) ## Installation diff --git a/README_STATISTICS.md b/README_STATISTICS.md new file mode 100644 index 0000000..4aa6a3a --- /dev/null +++ b/README_STATISTICS.md @@ -0,0 +1,149 @@ +# TI-Status Checker - Statistik-Funktionalität + +## Übersicht + +Der TI-Status Checker wurde um umfangreiche Statistik-Funktionalität erweitert, die automatisch alle Störungen und deren Dauer aufzeichnet. + +## Neue Features + +### 1. Automatische Störungsaufzeichnung +- **Störungsbeginn**: Wird automatisch erkannt und aufgezeichnet +- **Störungsende**: Wird automatisch erkannt und die Dauer berechnet +- **Störungstypen**: Unterscheidung zwischen "full" und "partial" Störungen + +### 2. Detaillierte Statistiken +- **Service-spezifische Statistiken**: Für jeden betroffenen Service +- **Gesamtstatistiken**: Über alle Störungen hinweg +- **Trend-Analysen**: Durchschnittswerte der letzten 30 Tage +- **Zeitreihen**: Chronologische Aufzeichnung aller Störungen + +### 3. Statistik-Berichte +- **Automatische Berichte**: Über Apprise versendbar +- **Markdown-Format**: Gut lesbar in Chat-Systemen +- **Zusammenfassungen**: Übersichtliche Darstellung der wichtigsten Kennzahlen + +## Verwendung + +### Normale Ausführung (Störungen überwachen) +```bash +python ti_status_checker.py +``` + +### Statistik-Bericht senden +```bash +python ti_status_checker.py --stats +# oder +python ti_status_checker.py -s +``` + +### Debug-Modus aktivieren +```bash +python ti_status_checker.py --debug +# oder +python ti_status_checker.py -d +``` + +## Statistik-Daten + +### Gespeicherte Informationen +- **Störungsbeginn**: ISO-Zeitstempel +- **Störungsende**: ISO-Zeitstempel +- **Dauer**: In Minuten +- **Service-Name**: Betroffener Dienst +- **Störungstyp**: "full" oder "partial" +- **Status**: "active" oder "resolved" + +### Berechnete Kennzahlen +- Gesamtanzahl Störungen +- Gesamtdauer aller Störungen +- Durchschnittsdauer pro Störung +- Längste Störung +- Am stärksten betroffener Service +- Tägliche Durchschnittswerte +- Trends der letzten 30 Tage + +## Dateien + +### Neue Dateien +- `ti_statistics.py`: Statistik-Funktionen und -Klasse +- `ti_outage_statistics.json`: Gespeicherte Statistik-Daten + +### Erweiterte Dateien +- `ti_status_checker.py`: Hauptscript mit Statistik-Integration + +## Beispiel-Statistik-Bericht + +``` +📊 **TI-Status Ausfall-Statistiken** +================================================== +**Zusammenfassung:** +• Gesamte Störungen: 15 +• Aktive Störungen: 2 +• Gesamtdauer: 1245 Minuten +• Durchschnittsdauer: 83.0 Minuten +• Längste Störung: 180 Minuten +• Am stärksten betroffen: konnektor + +**Service-Statistiken:** +• **KONNEKTOR**: + - Störungen: 8 + - Gesamtdauer: 720 Min + - Durchschnitt: 90.0 Min + - Längste: 180 Min + +**Trends (letzte 30 Tage):** +• Durchschnitt Störungen/Tag: 0.5 +• Durchschnitt Dauer/Tag: 41.5 Min +• Tage mit Störungen: 12 +• Max. Störungen an einem Tag: 3 + +**Letzte Störungen (7 Tage):** +• **KONNEKTOR** (partial) + - 15.01. 14:30 - 15.01. 16:45 (135 Min) +``` + +## Konfiguration + +Die Statistik-Funktionalität verwendet dieselben Apprise-Konfigurationen wie der normale Status-Checker. Stellen Sie sicher, dass in Ihrer `.env` Datei mindestens eine Apprise-URL konfiguriert ist: + +```env +APPRISE_URL_MATTERMOST=mattermost://... +APPRISE_URL_SLACK=slack://... +# etc. +``` + +## Automatisierung + +### Cron-Job für regelmäßige Statistiken +```bash +# Täglich um 9:00 Uhr Statistik-Bericht senden +0 9 * * * cd /path/to/script && python ti_status_checker.py --stats +``` + +### Windows Task Scheduler +Erstellen Sie eine geplante Aufgabe, die regelmäßig `python ti_status_checker.py --stats` ausführt. + +## Fehlerbehebung + +### Häufige Probleme +1. **Keine Apprise-URLs konfiguriert**: Prüfen Sie Ihre `.env` Datei +2. **Berechtigungsfehler**: Stellen Sie sicher, dass das Script Schreibrechte im Verzeichnis hat +3. **JSON-Fehler**: Bei korrupten Statistik-Daten wird automatisch eine neue Datei erstellt + +### Debug-Modus +Verwenden Sie `--debug` für detaillierte Ausgaben und Fehlerdiagnose. + +## Erweiterte Verwendung + +### Statistik-Daten exportieren +Die Statistik-Daten werden in `ti_outage_statistics.json` gespeichert und können für weitere Analysen verwendet werden. + +### Benutzerdefinierte Berichte +Die `OutageStatistics` Klasse kann in eigene Scripts importiert werden, um maßgeschneiderte Berichte zu erstellen. + +## Support + +Bei Fragen oder Problemen mit der Statistik-Funktionalität: +1. Aktivieren Sie den Debug-Modus mit `--debug` +2. Prüfen Sie die Logs auf Fehlermeldungen +3. Stellen Sie sicher, dass alle Abhängigkeiten installiert sind \ No newline at end of file diff --git a/ti_statistics.py b/ti_statistics.py new file mode 100644 index 0000000..1ef4e67 --- /dev/null +++ b/ti_statistics.py @@ -0,0 +1,240 @@ +import json +import os +from datetime import datetime, timedelta +from collections import defaultdict, Counter +import statistics + +STATS_FILE = "ti_outage_statistics.json" + +class OutageStatistics: + def __init__(self): + self.stats_file = STATS_FILE + self.stats = self.load_statistics() + + def load_statistics(self): + """Lädt gespeicherte Statistiken aus der JSON-Datei""" + try: + if os.path.exists(self.stats_file): + with open(self.stats_file, "r", encoding="utf-8") as f: + return json.load(f) + except (json.JSONDecodeError, IOError) as e: + print(f"Warnung: Konnte Statistiken nicht laden: {e}") + + # Standard-Struktur für neue Statistiken + return { + "outages": [], + "services": {}, + "summary": { + "total_outages": 0, + "total_duration_minutes": 0, + "average_duration_minutes": 0, + "longest_outage_minutes": 0, + "most_affected_service": "", + "last_updated": "" + } + } + + def save_statistics(self): + """Speichert Statistiken in die JSON-Datei""" + try: + with open(self.stats_file, "w", encoding="utf-8") as f: + json.dump(self.stats, f, ensure_ascii=False, indent=2) + except IOError as e: + print(f"Fehler beim Speichern der Statistiken: {e}") + + def record_outage_start(self, service_name, outage_type, timestamp=None): + """Zeichnet den Beginn einer Störung auf""" + if timestamp is None: + timestamp = datetime.now().isoformat() + + # Prüfe ob bereits eine aktive Störung für diesen Service existiert + for outage in self.stats["outages"]: + if (outage["service"] == service_name and + outage["status"] == "active" and + outage["type"] == outage_type): + # Störung bereits aktiv, nicht erneut aufzeichnen + return + + outage_record = { + "service": service_name, + "type": outage_type, + "start_time": timestamp, + "end_time": None, + "status": "active", + "duration_minutes": None + } + + self.stats["outages"].append(outage_record) + self.update_service_stats(service_name, "start") + self.save_statistics() + + def record_outage_end(self, service_name, timestamp=None): + """Zeichnet das Ende einer Störung auf""" + if timestamp is None: + timestamp = datetime.now().isoformat() + + # Finde die aktive Störung für diesen Service + for outage in self.stats["outages"]: + if outage["service"] == service_name and outage["status"] == "active": + outage["end_time"] = timestamp + outage["status"] = "resolved" + + # Berechne Dauer + start_time = datetime.fromisoformat(outage["start_time"]) + end_time = datetime.fromisoformat(timestamp) + duration = end_time - start_time + outage["duration_minutes"] = int(duration.total_seconds() / 60) + + self.update_service_stats(service_name, "end", outage["duration_minutes"]) + break + + self.save_statistics() + + def update_service_stats(self, service_name, event_type, duration_minutes=None): + """Aktualisiert die Service-spezifischen Statistiken""" + if service_name not in self.stats["services"]: + self.stats["services"][service_name] = { + "total_outages": 0, + "total_duration_minutes": 0, + "average_duration_minutes": 0, + "longest_outage_minutes": 0, + "last_outage": None + } + + service_stats = self.stats["services"][service_name] + + if event_type == "start": + service_stats["total_outages"] += 1 + service_stats["last_outage"] = datetime.now().isoformat() + elif event_type == "end" and duration_minutes: + service_stats["total_duration_minutes"] += duration_minutes + service_stats["average_duration_minutes"] = ( + service_stats["total_duration_minutes"] / service_stats["total_outages"] + ) + if duration_minutes > service_stats["longest_outage_minutes"]: + service_stats["longest_outage_minutes"] = duration_minutes + + def update_summary_stats(self): + """Aktualisiert die Zusammenfassungs-Statistiken""" + active_outages = [o for o in self.stats["outages"] if o["status"] == "active"] + resolved_outages = [o for o in self.stats["outages"] if o["status"] == "resolved"] + + total_duration = sum(o.get("duration_minutes", 0) for o in resolved_outages) + durations = [o.get("duration_minutes", 0) for o in resolved_outages if o.get("duration_minutes")] + + self.stats["summary"] = { + "total_outages": len(resolved_outages), + "active_outages": len(active_outages), + "total_duration_minutes": total_duration, + "average_duration_minutes": statistics.mean(durations) if durations else 0, + "longest_outage_minutes": max(durations) if durations else 0, + "most_affected_service": self.get_most_affected_service(), + "last_updated": datetime.now().isoformat() + } + + def get_most_affected_service(self): + """Ermittelt den am stärksten betroffenen Service""" + if not self.stats["services"]: + return "" + + most_affected = max( + self.stats["services"].items(), + key=lambda x: x[1]["total_outages"] + ) + return most_affected[0] + + def get_recent_outages(self, days=30): + """Gibt Störungen der letzten X Tage zurück""" + cutoff_date = datetime.now() - timedelta(days=days) + recent_outages = [] + + for outage in self.stats["outages"]: + if outage["status"] == "resolved" and outage["end_time"]: + end_time = datetime.fromisoformat(outage["end_time"]) + if end_time >= cutoff_date: + recent_outages.append(outage) + + return sorted(recent_outages, key=lambda x: x["end_time"], reverse=True) + + def get_outage_trends(self, days=30): + """Analysiert Trends in den Störungen""" + recent_outages = self.get_recent_outages(days) + + # Gruppiere nach Datum + daily_outages = defaultdict(list) + for outage in recent_outages: + date = outage["end_time"][:10] # YYYY-MM-DD + daily_outages[date].append(outage) + + # Berechne Durchschnitt pro Tag + daily_counts = [len(outages) for outages in daily_outages.values()] + daily_durations = [] + for outages in daily_outages.values(): + daily_duration = sum(o.get("duration_minutes", 0) for o in outages) + daily_durations.append(daily_duration) + + return { + "daily_average_outages": statistics.mean(daily_counts) if daily_counts else 0, + "daily_average_duration": statistics.mean(daily_durations) if daily_durations else 0, + "total_days_with_outages": len(daily_outages), + "most_outages_in_day": max(daily_counts) if daily_counts else 0 + } + + def generate_statistics_report(self): + """Generiert einen detaillierten Statistik-Bericht""" + self.update_summary_stats() + + report = [] + report.append("📊 **TI-Status Ausfall-Statistiken**") + report.append("=" * 50) + + # Zusammenfassung + summary = self.stats["summary"] + report.append(f"**Zusammenfassung:**") + report.append(f"• Gesamte Störungen: {summary['total_outages']}") + report.append(f"• Aktive Störungen: {summary['active_outages']}") + report.append(f"• Gesamtdauer: {summary['total_duration_minutes']} Minuten") + report.append(f"• Durchschnittsdauer: {summary['average_duration_minutes']:.1f} Minuten") + report.append(f"• Längste Störung: {summary['longest_outage_minutes']} Minuten") + report.append(f"• Am stärksten betroffen: {summary['most_affected_service']}") + report.append("") + + # Service-spezifische Statistiken + if self.stats["services"]: + report.append("**Service-Statistiken:**") + for service, stats in sorted( + self.stats["services"].items(), + key=lambda x: x[1]["total_outages"], + reverse=True + )[:10]: # Top 10 Services + report.append(f"• **{service.upper()}**:") + report.append(f" - Störungen: {stats['total_outages']}") + report.append(f" - Gesamtdauer: {stats['total_duration_minutes']} Min") + report.append(f" - Durchschnitt: {stats['average_duration_minutes']:.1f} Min") + report.append(f" - Längste: {stats['longest_outage_minutes']} Min") + report.append("") + + # Trends der letzten 30 Tage + trends = self.get_outage_trends(30) + report.append("**Trends (letzte 30 Tage):**") + report.append(f"• Durchschnitt Störungen/Tag: {trends['daily_average_outages']:.1f}") + report.append(f"• Durchschnitt Dauer/Tag: {trends['daily_average_duration']:.1f} Min") + report.append(f"• Tage mit Störungen: {trends['total_days_with_outages']}") + report.append(f"• Max. Störungen an einem Tag: {trends['most_outages_in_day']}") + report.append("") + + # Letzte Störungen + recent_outages = self.get_recent_outages(7) # Letzte 7 Tage + if recent_outages: + report.append("**Letzte Störungen (7 Tage):**") + for outage in recent_outages[:5]: # Top 5 + duration = outage.get("duration_minutes", 0) + start_time = datetime.fromisoformat(outage["start_time"]).strftime("%d.%m. %H:%M") + end_time = datetime.fromisoformat(outage["end_time"]).strftime("%d.%m. %H:%M") + report.append(f"• **{outage['service'].upper()}** ({outage['type']})") + report.append(f" - {start_time} - {end_time} ({duration} Min)") + + report.append("") + report.append(f"_Bericht generiert am {datetime.now().strftime('%d.%m.%Y %H:%M:%S')}_") + + return "\n".join(report) \ No newline at end of file diff --git a/ti_status_checker.py b/ti_status_checker.py index 2679913..49008e6 100644 --- a/ti_status_checker.py +++ b/ti_status_checker.py @@ -6,6 +6,8 @@ import os from dotenv import load_dotenv import apprise import time +import argparse +from ti_statistics import OutageStatistics # Lade Umgebungsvariablen aus .env Datei load_dotenv() @@ -277,7 +279,66 @@ def send_notification(message): print(f"⏳ Warte {delay} Sekunden...") time.sleep(delay) +def send_statistics_notification(): + """Sendet Statistik-Bericht über Apprise""" + print("📊 Generiere und sende Statistik-Bericht...") + + # Erstelle Statistik-Objekt + stats = OutageStatistics() + + # Generiere Bericht + report = stats.generate_statistics_report() + + # 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 = "📊 TI-Status Ausfall-Statistiken" + body = report + + if is_debug_mode(): + print(f"📤 Sende Statistik-Bericht an {len(urls)} Endpunkt(e)") + print(f"Bericht-Länge: {len(report)} Zeichen") + + # Sende die Nachricht + try: + result = apobj.notify( + title=title, + body=body, + body_format=apprise.NotifyFormat.MARKDOWN + ) + if result: + print("✅ Statistik-Bericht erfolgreich gesendet") + else: + print("❌ Fehler beim Senden des Statistik-Berichts") + except Exception as e: + print(f"❌ Fehler beim Senden des Statistik-Berichts: {e}") + def main(): + # Parse Kommandozeilen-Argumente + parser = argparse.ArgumentParser(description="TI-Status Checker mit Statistik-Funktionalität") + parser.add_argument("--stats", "-s", action="store_true", + help="Generiert und sendet Statistik-Bericht über Apprise") + parser.add_argument("--debug", "-d", action="store_true", + help="Aktiviert Debug-Modus") + + args = parser.parse_args() + + # Wenn --stats gesetzt ist, sende nur Statistiken + if args.stats: + send_statistics_notification() + return + # Prüfe Konfiguration urls = get_apprise_urls() if not urls: @@ -285,7 +346,7 @@ def main(): print("Bitte erstelle eine .env Datei basierend auf env.example") return - if is_debug_mode(): + if is_debug_mode() or args.debug: print("🔧 Debug-Modus aktiviert") print(f"📋 Benachrichtigungslevel: {get_notification_level()}") print(f"🔍 Filter: {get_notification_filters()}") @@ -297,6 +358,9 @@ def main(): last_status = state.get("last_status", {}) print("Prüfe TI-Status-API auf neue Meldungen...") + # Erstelle Statistik-Objekt für die Aufzeichnung von Störungen + stats = OutageStatistics() + try: messages = fetch_status_messages() # Extrahiere aktuelle und vorherige Störungen @@ -307,13 +371,25 @@ def main(): current_outages = extract_outages(app_status) previous_outages = extract_outages(last_status.get("appStatus", {})) if last_status else set() - # Entwarnungen erkennen + # Neue Störungen aufzeichnen + new_outages = current_outages - previous_outages + for dienst in new_outages: + # Bestimme den Störungstyp + outage_type = "full" if app_status.get(dienst, {}).get("outage") == "full" else "partial" + stats.record_outage_start(dienst, outage_type) + if is_debug_mode() or args.debug: + print(f"📊 Neue Störung aufgezeichnet: {dienst} ({outage_type})") + + # Entwarnungen erkennen und aufzeichnen resolved = previous_outages - current_outages for dienst in resolved: + stats.record_outage_end(dienst) msg = f"✅ Entwarnung: Die Störung bei {dienst.upper()} wurde behoben." print(msg) send_notification(msg) known_messages.add(msg) + if is_debug_mode() or args.debug: + print(f"📊 Störung beendet aufgezeichnet: {dienst}") # Normale neue Meldungen new_messages = [m for m in messages if m not in known_messages]