json_utils.py
- Pfad:
/var/www/scripts/pipeline/json_utils.py - Namespace: pipeline
- Zeilen: 155 | Größe: 3,705 Bytes
- Geändert: 2025-12-31 03:01:09 | Gescannt: 2025-12-31 10:22:15
Code Hygiene Score: 100
- Dependencies: 100 (25%)
- LOC: 100 (20%)
- Methods: 100 (20%)
- Secrets: 100 (15%)
- Classes: 100 (10%)
- Magic Numbers: 100 (10%)
Keine Issues gefunden.
Dependencies 3
- use json
- use re
- use typing.Any
Funktionen 3
-
extract_json()Zeile 17 -
repair_json()Zeile 88 -
safe_get()Zeile 127
Code
#!/usr/bin/env python3
"""
Robuste JSON-Extraktion für LLM-Responses.
Behandelt häufige Probleme:
- Mehrere JSON-Blöcke (nimmt den ersten)
- Trailing Commas
- Unescaped Quotes in Strings
- Markdown Code-Blöcke
"""
import json
import re
from typing import Any
def extract_json(text: str) -> dict | None:
"""
Extrahiert erstes gültiges JSON-Objekt aus Text.
Args:
text: LLM-Response mit JSON
Returns:
Parsed dict oder None bei Fehler
"""
if not text:
return None
# 1. Markdown Code-Blöcke entfernen
text = re.sub(r"```json\s*", "", text)
text = re.sub(r"```\s*", "", text)
# 2. Ersten JSON-Block finden (Brace-Matching)
start = text.find("{")
if start < 0:
return None
depth = 0
end = start
in_string = False
escape_next = False
for i, char in enumerate(text[start:], start):
if escape_next:
escape_next = False
continue
if char == "\\":
escape_next = True
continue
if char == '"' and not escape_next:
in_string = not in_string
continue
if in_string:
continue
if char == "{":
depth += 1
elif char == "}":
depth -= 1
if depth == 0:
end = i + 1
break
if end <= start:
return None
json_str = text[start:end]
# 3. Versuche direkt zu parsen
try:
return json.loads(json_str)
except json.JSONDecodeError:
pass
# 4. JSON reparieren und erneut versuchen
json_str = repair_json(json_str)
try:
return json.loads(json_str)
except json.JSONDecodeError:
return None
def repair_json(json_str: str) -> str:
"""
Repariert häufige JSON-Fehler von LLMs.
Args:
json_str: Möglicherweise fehlerhafter JSON-String
Returns:
Reparierter JSON-String
"""
# Trailing Commas vor } oder ] entfernen
json_str = re.sub(r",\s*}", "}", json_str)
json_str = re.sub(r",\s*]", "]", json_str)
# Single Quotes zu Double Quotes (außerhalb von Strings)
# Vorsicht: nur wenn es eindeutig ist
if "'" in json_str and '"' not in json_str:
json_str = json_str.replace("'", '"')
# Fehlende Quotes um Werte (simple Fälle)
# z.B. {key: value} -> {"key": "value"}
json_str = re.sub(r"{\s*(\w+)\s*:", r'{"\1":', json_str)
json_str = re.sub(r",\s*(\w+)\s*:", r', "\1":', json_str)
# Unescaped Newlines in Strings ersetzen
# Zwischen Quotes: \n -> \\n
def escape_newlines(match: re.Match) -> str:
content = match.group(1)
content = content.replace("\n", "\\n")
content = content.replace("\r", "\\r")
content = content.replace("\t", "\\t")
return f'"{content}"'
# Strings mit Newlines finden und escapen
json_str = re.sub(r'"([^"]*(?:\n|\r)[^"]*)"', escape_newlines, json_str)
return json_str
def safe_get(data: dict, key: str, default: Any = None, valid_values: set = None) -> Any:
"""
Sicherer Zugriff auf dict-Werte mit Validierung.
Args:
data: Source dict
key: Schlüssel
default: Fallback-Wert
valid_values: Erlaubte Werte (optional)
Returns:
Validierter Wert oder Default
"""
value = data.get(key, default)
# Liste -> erstes Element
if isinstance(value, list):
value = value[0] if value else default
# String normalisieren
if isinstance(value, str):
value = value.lower().strip()
# Validierung
if valid_values and value not in valid_values:
return default
return value