diff --git a/scripts/sensor_list_report.py b/scripts/sensor_list_report.py index 2febd2c..2a5ea77 100644 --- a/scripts/sensor_list_report.py +++ b/scripts/sensor_list_report.py @@ -1,9 +1,10 @@ -import httpx -import json +import html from datetime import datetime -# GraphQL Query für alle Sensoren -query = ''' +import httpx + + +QUERY = """ query GetAllSensors { sensors { id @@ -17,47 +18,65 @@ query GetAllSensors { } } } -''' - -# API Aufruf -with httpx.Client() as client: - response = client.post( - f"{EXTERNAL_BASE_URL}/graphql", - json={"query": query}, - headers=AUTH_HEADERS, - timeout=30.0 +""".strip() + + +def escape_text(value, default=""): + text = default if value is None else str(value) + return html.escape(text, quote=True) + + +def build_error(message): + return ( + "
" + f"{escape_text(message)}" + "
" ) - + + +def build_report(): + with httpx.Client() as client: + response = client.post( + f"{EXTERNAL_BASE_URL}/graphql", + json={"query": QUERY}, + headers=AUTH_HEADERS, + timeout=30.0, + ) + if response.status_code != 200: - result = f"
Fehler beim Abrufen der Sensoren: {response.status_code}
" - else: - data = response.json() - sensors = data.get('data', {}).get('sensors', []) - - # Sortiere Sensoren nach MeasureConcept Name - sensors.sort(key=lambda s: (s['measureConcept']['name'].strip() if s['measureConcept']['name'] else '', s['name'].strip())) - - # Gruppiere nach MeasureConcept - measure_concepts = {} - for sensor in sensors: - mc_id = sensor['measureConcept']['id'] - mc_name = sensor['measureConcept']['name'].strip() if sensor['measureConcept']['name'] else 'Unbekannt' - mc_desc = sensor['measureConcept']['description'] or '' - - if mc_id not in measure_concepts: - measure_concepts[mc_id] = { - 'name': mc_name, - 'description': mc_desc, - 'sensors': [] - } - measure_concepts[mc_id]['sensors'].append(sensor) - - # Statistiken - total_sensors = len(sensors) - total_concepts = len(measure_concepts) - - # HTML generieren - html = f''' + return build_error(f"Fehler beim Abrufen der Sensoren: {response.status_code}") + + data = response.json() + sensors = data.get("data", {}).get("sensors", []) + + sensors.sort( + key=lambda s: ( + (s.get("measureConcept") or {}).get("name", "").strip(), + (s.get("name") or "").strip(), + ) + ) + + measure_concepts = {} + for sensor in sensors: + measure_concept = sensor.get("measureConcept") or {} + mc_id = str(measure_concept.get("id") or "unknown") + mc_name = (measure_concept.get("name") or "Unbekannt").strip() + mc_desc = measure_concept.get("description") or "" + + if mc_id not in measure_concepts: + measure_concepts[mc_id] = { + "name": mc_name, + "description": mc_desc, + "sensors": [], + } + + measure_concepts[mc_id]["sensors"].append(sensor) + + total_sensors = len(sensors) + total_concepts = len(measure_concepts) + + html_output = f""" @@ -257,7 +276,7 @@ with httpx.Client() as client:
-

🔧 Sensor Übersicht

+

Sensor Uebersicht

{total_sensors}
@@ -269,127 +288,127 @@ with httpx.Client() as client:
- +
- +
- +
- +
-''' - - # MeasureConcepts und Sensoren ausgeben - for mc_id, mc_data in sorted(measure_concepts.items(), key=lambda x: x[1]['name']): - mc_name = mc_data['name'] - mc_desc = mc_data['description'] - sensors_in_concept = mc_data['sensors'] - - html += f''' -
-
+""" + + for mc_id, mc_data in sorted(measure_concepts.items(), key=lambda item: item[1]["name"]): + safe_mc_id = escape_text(mc_id) + mc_name = escape_text(mc_data["name"], "Unbekannt") + mc_desc = escape_text(mc_data["description"]) + sensors_in_concept = mc_data["sensors"] + + html_output += f""" +
+

- 📊 {mc_name} + {mc_name} {len(sensors_in_concept)} Sensoren

-''' - if mc_desc: - html += f'
{mc_desc}
\n' - - html += ''' +""" + if mc_desc: + html_output += f'
{mc_desc}
\n' + + html_output += f"""
-
+
-''' - - for sensor in sensors_in_concept: - sensor_id = sensor['id'] - sensor_name = sensor['name'].strip() if sensor['name'] else 'Unbekannter Name' - sensor_extern = sensor.get('nameExtern', '') or '' - sensor_desc = sensor.get('description', '') or '' - - html += f''' +""" + + for sensor in sensors_in_concept: + sensor_id = escape_text(sensor.get("id"), "Unbekannt") + sensor_name = escape_text((sensor.get("name") or "").strip(), "Unbekannter Name") + sensor_extern = escape_text(sensor.get("nameExtern") or "") + sensor_desc = escape_text(sensor.get("description") or "") + + html_output += f"""
ID: {sensor_id}
{sensor_name}
-''' - - if sensor_extern and sensor_extern.strip() != '-': - html += f'
Extern: {sensor_extern}
\n' - - if sensor_desc: - html += f'
{sensor_desc}
\n' - else: - html += '
Keine Beschreibung verfügbar
\n' - - html += '
\n' - - html += ''' +""" + + if sensor_extern and sensor_extern != "-": + html_output += f'
Extern: {sensor_extern}
\n' + + if sensor_desc: + html_output += f'
{sensor_desc}
\n' + else: + html_output += '
Keine Beschreibung verfuegbar
\n' + + html_output += "
\n" + + html_output += """
-''' - - # Timestamp und JavaScript - current_time = datetime.now().strftime('%d.%m.%Y um %H:%M:%S') - html += f''' +""" + + current_time = datetime.now().strftime("%d.%m.%Y um %H:%M:%S") + html_output += f"""
- +
- 📅 Erstellt am {current_time} + Erstellt am {current_time}
- + -''' - - result = html +""" + + return html_output + + +result = build_report() +print(result)