96 lines
No EOL
3.7 KiB
Python
96 lines
No EOL
3.7 KiB
Python
from prometheus_client import Counter, generate_latest, CONTENT_TYPE_LATEST, CollectorRegistry
|
|
from typing import Dict, Tuple
|
|
import json
|
|
import os
|
|
from models import DrinkType, MateType
|
|
|
|
|
|
class DrinkSalesMetrics:
|
|
def __init__(self, stats_file: str = "drink_stats.json"):
|
|
self.stats_file = stats_file
|
|
self.stats: Dict[Tuple[str, str], int] = self._load_stats()
|
|
|
|
# Erstelle eine eigene Registry ohne Standard-Metriken
|
|
self.registry = CollectorRegistry()
|
|
|
|
# Prometheus Counter für verkaufte Getränke
|
|
self.drinks_sold_counter = Counter(
|
|
'tschunk_drinks_sold_total',
|
|
'Total number of drinks sold',
|
|
['drink_type', 'mate_type'],
|
|
registry=self.registry
|
|
)
|
|
|
|
# Initialisiere Counter mit gespeicherten Werten
|
|
self._initialize_counters()
|
|
|
|
def _load_stats(self) -> Dict[Tuple[str, str], int]:
|
|
"""Lädt die gespeicherten Statistiken aus der JSON-Datei."""
|
|
if os.path.exists(self.stats_file):
|
|
try:
|
|
with open(self.stats_file, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
# Konvertiere String-Tupel zurück zu Tuple
|
|
return {tuple(k.split('|')): v for k, v in data.items()}
|
|
except (json.JSONDecodeError, KeyError, ValueError) as e:
|
|
print(f"Fehler beim Laden der Statistiken: {e}")
|
|
return {}
|
|
return {}
|
|
|
|
def _save_stats(self):
|
|
"""Speichert die Statistiken in eine JSON-Datei."""
|
|
try:
|
|
# Konvertiere Tuple zu String für JSON-Serialisierung
|
|
data = {f"{k[0]}|{k[1]}": v for k, v in self.stats.items()}
|
|
with open(self.stats_file, 'w', encoding='utf-8') as f:
|
|
json.dump(data, f, indent=2, ensure_ascii=False)
|
|
except Exception as e:
|
|
print(f"Fehler beim Speichern der Statistiken: {e}")
|
|
|
|
def _initialize_counters(self):
|
|
"""Initialisiert die Prometheus-Counter mit den gespeicherten Werten."""
|
|
for (drink_type, mate_type), count in self.stats.items():
|
|
# Setze den Counter auf den gespeicherten Wert
|
|
# Da Counter nur inkrementiert werden können, müssen wir das manuell machen
|
|
for _ in range(count):
|
|
self.drinks_sold_counter.labels(
|
|
drink_type=drink_type,
|
|
mate_type=mate_type
|
|
).inc()
|
|
|
|
def record_drinks(self, drinks):
|
|
"""Zeichnet verkaufte Getränke auf und aktualisiert die Statistiken."""
|
|
for drink in drinks:
|
|
drink_type = drink.drink_type.value
|
|
mate_type = drink.mate_type.value
|
|
quantity = drink.quantity
|
|
|
|
# Aktualisiere lokale Statistiken
|
|
key = (drink_type, mate_type)
|
|
self.stats[key] = self.stats.get(key, 0) + quantity
|
|
|
|
# Aktualisiere Prometheus-Counter
|
|
self.drinks_sold_counter.labels(
|
|
drink_type=drink_type,
|
|
mate_type=mate_type
|
|
).inc(quantity)
|
|
|
|
# Speichere Statistiken persistent
|
|
self._save_stats()
|
|
|
|
def get_stats(self) -> Dict[str, Dict[str, int]]:
|
|
"""Gibt die aktuellen Statistiken zurück."""
|
|
result = {}
|
|
for (drink_type, mate_type), count in self.stats.items():
|
|
if drink_type not in result:
|
|
result[drink_type] = {}
|
|
result[drink_type][mate_type] = count
|
|
return result
|
|
|
|
def get_metrics(self):
|
|
"""Gibt die Prometheus-Metriken zurück."""
|
|
return generate_latest(self.registry)
|
|
|
|
|
|
# Globale Instanz der Metriken
|
|
drink_metrics = DrinkSalesMetrics() |