deploy script: ultimo_staende_eingabe

scripts/ultimo_staende_eingabe.py

@@ -0,0 +1,447 @@
+import httpx
+import json
+from datetime import datetime, date
+from typing import List, Dict, Any, Optional
+
+# Globals werden zur Laufzeit bereitgestellt
+# EXTERNAL_BASE_URL: str
+# AUTH_HEADERS: dict
+# PARAMS: dict
+
+def execute_graphql(query: str, variables: Optional[Dict] = None) -> Dict[str, Any]:
+    """Führt eine GraphQL-Abfrage aus"""
+    try:
+        response = httpx.post(
+            f"{EXTERNAL_BASE_URL}/graphql",
+            json={"query": query, "variables": variables or {}},
+            headers=AUTH_HEADERS,
+            timeout=30
+        )
+        response.raise_for_status()
+        return response.json()
+    except Exception as e:
+        return {"errors": [{"message": f"GraphQL-Fehler: {str(e)}"}]}
+
+def get_sensors_for_meter(meter_number: str) -> List[Dict]:
+    """Sucht Sensoren für eine Zählernummer"""
+    query = """
+    query GetSensorsForMeter($meterNumber: String!) {
+      sensorsForMeterNumber(meterNumber: $meterNumber) {
+        sensorId
+        sensorName
+        sensorNameExtern
+        descr
+        measureConcept {
+          id
+          name
+          descr
+        }
+      }
+    }
+    """
+    
+    response = execute_graphql(query, {"meterNumber": meter_number})
+    
+    if "errors" in response:
+        return []
+    
+    return response.get("data", {}).get("sensorsForMeterNumber", [])
+
+def get_available_variables(sensor_id: str) -> List[Dict]:
+    """Holt verfügbare Variablen/Units für einen Sensor"""
+    query = """
+    query GetAvailableVariables($sensorId: ID!) {
+      availableVariableUnits(sensorId: $sensorId) {
+        variableUnitId
+        variableName
+        unitName
+      }
+    }
+    """
+    
+    response = execute_graphql(query, {"sensorId": sensor_id})
+    
+    if "errors" in response:
+        return []
+    
+    return response.get("data", {}).get("availableVariableUnits", [])
+
+def record_ultimo_readings(sensor_id: str, variable_name: str, variable_unit: str, readings: List[Dict]) -> Dict:
+    """Trägt Ultimo-Stände ein"""
+    mutation = """
+    mutation RecordUltimoReadings($input: UltimoReadingsInput!) {
+      recordUltimoReadings(input: $input) {
+        success
+        created {
+          id
+          moment
+          value
+          meterValue
+        }
+        errors {
+          code
+          message
+          details
+        }
+      }
+    }
+    """
+    
+    input_data = {
+        "sensorId": sensor_id,
+        "variableName": variable_name.strip(),
+        "variableUnit": variable_unit.strip(),
+        "readings": readings
+    }
+    
+    response = execute_graphql(mutation, {"input": input_data})
+    
+    if "errors" in response:
+        return {"success": False, "errors": response["errors"]}
+    
+    return response.get("data", {}).get("recordUltimoReadings", {"success": False, "errors": [{"message": "Unbekannter Fehler"}]})
+
+def parse_month_readings(month_data: str) -> List[Dict]:
+    """Parst Monatsdaten im Format 'YYYY-MM: Wert'"""
+    readings = []
+    lines = [line.strip() for line in month_data.strip().split('\n') if line.strip()]
+    
+    for line in lines:
+        if ':' not in line:
+            continue
+        
+        try:
+            month_str, value_str = line.split(':', 1)
+            month = month_str.strip()
+            value = float(value_str.strip().replace(',', '.'))
+            
+            # Validiere Monatsformat
+            datetime.strptime(month + '-01', '%Y-%m-%d')
+            
+            readings.append({
+                "month": month,
+                "meterValue": value
+            })
+        except (ValueError, AttributeError) as e:
+            continue
+    
+    return readings
+
+def render_success_html(created_observations: List[Dict], sensor_name: str, variable_info: str) -> str:
+    """Rendert Erfolgsmeldung als HTML"""
+    obs_count = len(created_observations)
+    
+    obs_html = ""
+    for obs in created_observations:
+        moment = obs.get("moment", "")
+        meter_value = obs.get("meterValue", 0)
+        value = obs.get("value", 0)
+        
+        obs_html += f"""
+        <tr>
+            <td>{moment}</td>
+            <td style="text-align: right;">{meter_value:.2f}</td>
+            <td style="text-align: right;">{value:.2f}</td>
+        </tr>
+        """
+    
+    return f"""
+    <div style="max-width: 800px; margin: 0 auto; font-family: Arial, sans-serif;">
+        <div style="background: var(--color-success); color: white; padding: 16px; border-radius: 8px; margin-bottom: 20px;">
+            <h2 style="margin: 0; display: flex; align-items: center;">
+                <span style="margin-right: 8px;">✅</span>
+                Ultimo-Stände erfolgreich eingetragen
+            </h2>
+        </div>
+        
+        <div style="background: #f5f5f5; padding: 16px; border-radius: 8px; margin-bottom: 20px;">
+            <h3 style="margin-top: 0;">Details</h3>
+            <p><strong>Sensor:</strong> {sensor_name}</p>
+            <p><strong>Variable/Einheit:</strong> {variable_info}</p>
+            <p><strong>Anzahl Messungen:</strong> {obs_count}</p>
+        </div>
+        
+        <div style="background: white; border: 1px solid #ddd; border-radius: 8px; overflow: hidden;">
+            <table style="width: 100%; border-collapse: collapse;">
+                <thead>
+                    <tr style="background: #f8f9fa;">
+                        <th style="padding: 12px; text-align: left; border-bottom: 1px solid #ddd;">Zeitpunkt</th>
+                        <th style="padding: 12px; text-align: right; border-bottom: 1px solid #ddd;">Zählerstand</th>
+                        <th style="padding: 12px; text-align: right; border-bottom: 1px solid #ddd;">Wert</th>
+                    </tr>
+                </thead>
+                <tbody>
+                    {obs_html}
+                </tbody>
+            </table>
+        </div>
+    </div>
+    """
+
+def render_error_html(errors: List[Dict]) -> str:
+    """Rendert Fehlermeldungen als HTML"""
+    error_html = ""
+    for error in errors:
+        code = error.get("code", "UNKNOWN")
+        message = error.get("message", "Unbekannter Fehler")
+        details = error.get("details", "")
+        
+        error_html += f"""
+        <div style="background: #fff3cd; border: 1px solid #ffeaa7; border-radius: 4px; padding: 12px; margin-bottom: 10px;">
+            <strong>{code}:</strong> {message}
+            {f'<br><small>{details}</small>' if details else ''}
+        </div>
+        """
+    
+    return f"""
+    <div style="max-width: 800px; margin: 0 auto; font-family: Arial, sans-serif;">
+        <div style="background: var(--color-danger); color: white; padding: 16px; border-radius: 8px; margin-bottom: 20px;">
+            <h2 style="margin: 0; display: flex; align-items: center;">
+                <span style="margin-right: 8px;">❌</span>
+                Fehler beim Eintragen der Ultimo-Stände
+            </h2>
+        </div>
+        
+        <div style="margin-bottom: 20px;">
+            {error_html}
+        </div>
+    </div>
+    """
+
+# Hauptlogik
+try:
+    meter_number = PARAMS.get("meter_number", "").strip()
+    sensor_id = PARAMS.get("sensor_id", "").strip()
+    
+    if meter_number and not sensor_id:
+        # PHASE 1: Sensoren suchen und Auswahlformular generieren
+        sensors = get_sensors_for_meter(meter_number)
+        
+        if not sensors:
+            result = {
+                "type": "html",
+                "content": f"""
+                <div style="max-width: 600px; margin: 0 auto; font-family: Arial, sans-serif; padding: 20px;">
+                    <div style="background: var(--color-warning); color: white; padding: 16px; border-radius: 8px; text-align: center;">
+                        <h3 style="margin: 0;">⚠️ Keine Sensoren gefunden</h3>
+                        <p style="margin: 10px 0 0 0;">Für die Zählernummer \"{meter_number}\" wurden keine Sensoren gefunden.</p>
+                    </div>
+                </div>
+                """
+            }
+        else:
+            # Sensor-Optionen sammeln
+            sensor_options = []
+            variable_options_by_sensor = {}
+            
+            for sensor in sensors:
+                s_id = sensor.get("sensorId")
+                s_name = sensor.get("sensorName", "").strip()
+                s_extern = sensor.get("sensorNameExtern") or ""
+                mc_name = sensor.get("measureConcept", {}).get("name", "").strip()
+                
+                display_name = f"{s_name}"
+                if s_extern.strip():
+                    display_name += f" ({s_extern.strip()})"
+                if mc_name:
+                    display_name += f" - {mc_name}"
+                
+                sensor_options.append({
+                    "value": s_id,
+                    "label": display_name
+                })
+                
+                # Verfügbare Variablen für diesen Sensor sammeln
+                variables = get_available_variables(s_id)
+                var_opts = []
+                for var in variables:
+                    var_name = var.get("variableName", "").strip()
+                    unit_name = var.get("unitName", "").strip()
+                    combined_value = f"{var_name}|{unit_name}"
+                    var_opts.append({
+                        "value": combined_value,
+                        "label": f"{var_name} ({unit_name})"
+                    })
+                
+                variable_options_by_sensor[s_id] = var_opts
+            
+            # Da wir nur eine statische Form unterstützen, nehmen wir die Variablen des ersten Sensors
+            # In einer echten Anwendung würde man hier eine dynamischere Lösung wählen
+            first_sensor_vars = variable_options_by_sensor.get(sensors[0].get("sensorId"), [])
+            
+            form_definition = {
+                "title": "Ultimo-Stände Eingabe",
+                "description": f"Sensoren für Zählernummer {meter_number} gefunden. Wählen Sie Sensor und Eingabemodus.",
+                "fields": [
+                    {
+                        "name": "sensor_id",
+                        "widget": "dropdown",
+                        "label": "Sensor",
+                        "options": sensor_options,
+                        "validators": [{"type": "required", "error_text": "Sensor auswählen"}]
+                    },
+                    {
+                        "name": "variable_unit",
+                        "widget": "dropdown",
+                        "label": "Variable und Einheit",
+                        "options": first_sensor_vars,
+                        "validators": [{"type": "required", "error_text": "Variable/Einheit auswählen"}],
+                        "helper_text": "Hinweis: Die Variablen werden für den ersten Sensor angezeigt. Nach Sensorauswahl evtl. nochmal aktualisieren."
+                    },
+                    {
+                        "name": "input_mode",
+                        "widget": "segmented_control",
+                        "label": "Eingabemodus",
+                        "options": [
+                            {"value": "single", "label": "Einzelner Monat"},
+                            {"value": "batch", "label": "Mehrere Monate"}
+                        ],
+                        "initial_value": "single"
+                    },
+                    {
+                        "name": "single_month",
+                        "widget": "text_field",
+                        "label": "Monat (YYYY-MM)",
+                        "hint_text": "z.B. 2024-12",
+                        "conditional": {
+                            "field_name": "input_mode",
+                            "operator": "equals",
+                            "value": "single",
+                            "action": "show"
+                        },
+                        "validators": [
+                            {"type": "match", "value": "^\\d{4}-\\d{2}$", "error_text": "Format: YYYY-MM"}
+                        ]
+                    },
+                    {
+                        "name": "single_value",
+                        "widget": "text_field",
+                        "label": "Zählerstand",
+                        "text_field_config": {"keyboard_type": "number"},
+                        "conditional": {
+                            "field_name": "input_mode",
+                            "operator": "equals",
+                            "value": "single",
+                            "action": "show"
+                        },
+                        "validators": [
+                            {"type": "numeric", "error_text": "Numerischer Wert erforderlich"}
+                        ]
+                    },
+                    {
+                        "name": "batch_data",
+                        "widget": "text_field",
+                        "label": "Monatsdaten",
+                        "hint_text": "Eine Zeile pro Monat im Format:\n2024-10: 1500.5\n2024-11: 1650.2\n2024-12: 1800.0",
+                        "text_field_config": {
+                            "max_lines": 10,
+                            "keyboard_type": "multiline"
+                        },
+                        "conditional": {
+                            "field_name": "input_mode",
+                            "operator": "equals",
+                            "value": "batch",
+                            "action": "show"
+                        },
+                        "validators": [
+                            {"type": "required", "error_text": "Monatsdaten erforderlich"}
+                        ]
+                    }
+                ],
+                "submit_label": "Ultimo-Stände eintragen"
+            }
+            
+            result = {
+                "type": "form",
+                "form_definition": form_definition
+            }
+    
+    elif sensor_id:
+        # PHASE 2: Ultimo-Stände eintragen
+        variable_unit = PARAMS.get("variable_unit", "")
+        input_mode = PARAMS.get("input_mode", "single")
+        
+        if not variable_unit or "|" not in variable_unit:
+            result = {
+                "type": "error",
+                "message": "Variable und Einheit müssen ausgewählt werden."
+            }
+        else:
+            variable_name, unit_name = variable_unit.split("|", 1)
+            
+            # Readings sammeln
+            readings = []
+            
+            if input_mode == "single":
+                single_month = PARAMS.get("single_month", "").strip()
+                single_value = PARAMS.get("single_value", "")
+                
+                if not single_month or not single_value:
+                    result = {
+                        "type": "error",
+                        "message": "Monat und Zählerstand müssen angegeben werden."
+                    }
+                else:
+                    try:
+                        value = float(single_value.replace(",", "."))
+                        readings = [{
+                            "month": single_month,
+                            "meterValue": value
+                        }]
+                    except ValueError:
+                        result = {
+                            "type": "error",
+                            "message": "Ungültiger Zählerstand. Bitte numerischen Wert eingeben."
+                        }
+            else:  # batch
+                batch_data = PARAMS.get("batch_data", "").strip()
+                if not batch_data:
+                    result = {
+                        "type": "error",
+                        "message": "Batch-Daten müssen angegeben werden."
+                    }
+                else:
+                    readings = parse_month_readings(batch_data)
+                    if not readings:
+                        result = {
+                            "type": "error",
+                            "message": "Keine gültigen Monatsdaten gefunden. Format: YYYY-MM: Wert"
+                        }
+            
+            # Wenn wir Readings haben, eintragen
+            if "type" not in result and readings:
+                # Sensor-Name für Display holen
+                sensor_name = "Unbekannt"
+                for sensor in get_sensors_for_meter(meter_number):
+                    if sensor.get("sensorId") == sensor_id:
+                        sensor_name = sensor.get("sensorName", "").strip()
+                        break
+                
+                # Ultimo-Stände eintragen
+                ultimo_result = record_ultimo_readings(sensor_id, variable_name, unit_name, readings)
+                
+                if ultimo_result.get("success"):
+                    created = ultimo_result.get("created", [])
+                    variable_info = f"{variable_name} ({unit_name})"
+                    result = {
+                        "type": "html",
+                        "content": render_success_html(created, sensor_name, variable_info)
+                    }
+                else:
+                    errors = ultimo_result.get("errors", [])
+                    result = {
+                        "type": "html",
+                        "content": render_error_html(errors)
+                    }
+    
+    else:
+        result = {
+            "type": "error",
+            "message": "Zählernummer muss angegeben werden."
+        }
+        
+except Exception as e:
+    result = {
+        "type": "error",
+        "message": f"Unerwarteter Fehler: {str(e)}"
+    }