from fastapi import FastAPI from fastapi.responses import HTMLResponse from jinja2 import Template import requests import json from dateutil import parser, tz app = FastAPI() API_URL = "http://192.168.18.4:8000/results/updatecheck/" READ_API_KEY = "read_key_1" REFERENCE_FILE = "reference_versions.json" # ================= Templates ================== MAIN_TEMPLATE = Template(""" UpdateLog Übersicht

UpdateLog Übersicht

{% for server, content in data.items() %}
🔍 Verlauf anzeigen
{% endfor %} """) HISTORY_TEMPLATE = Template(""" Verlauf: {{ hostname }}

Update-Verlauf für {{ hostname }}

⬅ Zurück zur Übersicht {% for entry in history %}

{{ entry.timestamp }}

{% for pkg, ver in entry.packages.items() %} {% endfor %}
PaketInstalliert
{{ pkg }} {{ ver }}
{% endfor %} """) # ================= Routes ================== @app.get("/", response_class=HTMLResponse) def index(): data = fetch_latest_per_host() return MAIN_TEMPLATE.render(data=data) @app.get("/host/{hostname}", response_class=HTMLResponse) def host_history(hostname: str): history = fetch_history_for_host(hostname) return HISTORY_TEMPLATE.render(hostname=hostname, history=history) # ================= Datenfunktionen ================== def fetch_raw_lines(): try: response = requests.get(API_URL, headers={"Authorization": f"Bearer {READ_API_KEY}"}, timeout=5) response.raise_for_status() return response.json().get("lines", []) except Exception as e: print(f"❌ Fehler beim Abrufen der Logs: {e}") return [] def fetch_reference(): try: with open(REFERENCE_FILE) as f: return json.load(f) except Exception as e: print(f"❌ Fehler beim Laden der Referenzversionen: {e}") return {} def fetch_latest_per_host(): lines = fetch_raw_lines() reference = fetch_reference() latest_by_host = {} for line in lines: parts = line.strip().split(",", 3) if len(parts) < 4: continue timestamp, hostname, _, packages = parts if hostname not in latest_by_host or timestamp > latest_by_host[hostname]["timestamp"]: latest_by_host[hostname] = { "timestamp": timestamp, "raw": packages } result = {} for hostname, entry in latest_by_host.items(): # Zeitstempel umrechnen try: dt_utc = parser.isoparse(entry["timestamp"]) dt_local = dt_utc.astimezone(tz.gettz("Europe/Berlin")) display_time = dt_local.strftime("%Y-%m-%d %H:%M:%S") except Exception: display_time = entry["timestamp"] package_str = entry["raw"] result[hostname] = { "timestamp": display_time, "packages": {} } for item in package_str.strip(";").split(";"): if "=" not in item: continue pkg, version = item.split("=", 1) expected = reference.get(pkg) if expected: result[hostname]["packages"][pkg] = { "current": version, "expected": expected } return result def fetch_history_for_host(hostname): lines = fetch_raw_lines() history = [] for line in lines: parts = line.strip().split(",", 3) if len(parts) < 4: continue timestamp, host, _, packages = parts if host != hostname: continue parsed = {} for item in packages.strip(";").split(";"): if "=" not in item: continue pkg, version = item.split("=", 1) parsed[pkg] = version history.append({ "timestamp": timestamp, "packages": parsed }) history.sort(key=lambda x: x["timestamp"], reverse=True) return history