{
"event": "PreToolUse",
"tool_name": "Read",
"tool_input": {
"file_path": "\/var\/www\/migration\/content-pipeline\/src\/json_utils.py"
}
}
{
"tool_response": {
"type": "text",
"file": {
"filePath": "\/var\/www\/migration\/content-pipeline\/src\/json_utils.py",
"content": "#!\/usr\/bin\/env python3\n\"\"\"\nRobuste JSON-Extraktion für LLM-Responses.\n\nBehandelt häufige Probleme:\n- Mehrere JSON-Blöcke (nimmt den ersten)\n- Trailing Commas\n- Unescaped Quotes in Strings\n- Markdown Code-Blöcke\n\"\"\"\n\nimport json\nimport re\nfrom typing import Any\n\n\ndef extract_json(text: str) -> dict | None:\n \"\"\"\n Extrahiert erstes gültiges JSON-Objekt aus Text.\n\n Args:\n text: LLM-Response mit JSON\n\n Returns:\n Parsed dict oder None bei Fehler\n \"\"\"\n if not text:\n return None\n\n # 1. Markdown Code-Blöcke entfernen\n text = re.sub(r\"```json\\s*\", \"\", text)\n text = re.sub(r\"```\\s*\", \"\", text)\n\n # 2. Ersten JSON-Block finden (Brace-Matching)\n start = text.find(\"{\")\n if start < 0:\n return None\n\n depth = 0\n end = start\n in_string = False\n escape_next = False\n\n for i, char in enumerate(text[start:], start):\n if escape_next:\n escape_next = False\n continue\n\n if char == \"\\\\\":\n escape_next = True\n continue\n\n if char == '\"' and not escape_next:\n in_string = not in_string\n continue\n\n if in_string:\n continue\n\n if char == \"{\":\n depth += 1\n elif char == \"}\":\n depth -= 1\n if depth == 0:\n end = i + 1\n break\n\n if end <= start:\n return None\n\n json_str = text[start:end]\n\n # 3. Versuche direkt zu parsen\n try:\n return json.loads(json_str)\n except json.JSONDecodeError:\n pass\n\n # 4. JSON reparieren und erneut versuchen\n json_str = repair_json(json_str)\n\n try:\n return json.loads(json_str)\n except json.JSONDecodeError:\n return None\n\n\ndef repair_json(json_str: str) -> str:\n \"\"\"\n Repariert häufige JSON-Fehler von LLMs.\n\n Args:\n json_str: Möglicherweise fehlerhafter JSON-String\n\n Returns:\n Reparierter JSON-String\n \"\"\"\n # Trailing Commas vor } oder ] entfernen\n json_str = re.sub(r\",\\s*}\", \"}\", json_str)\n json_str = re.sub(r\",\\s*]\", \"]\", json_str)\n\n # Single Quotes zu Double Quotes (außerhalb von Strings)\n # Vorsicht: nur wenn es eindeutig ist\n if \"'\" in json_str and '\"' not in json_str:\n json_str = json_str.replace(\"'\", '\"')\n\n # Fehlende Quotes um Werte (simple Fälle)\n # z.B. {key: value} -> {\"key\": \"value\"}\n json_str = re.sub(r\"{\\s*(\\w+)\\s*:\", r'{\"\\1\":', json_str)\n json_str = re.sub(r\",\\s*(\\w+)\\s*:\", r', \"\\1\":', json_str)\n\n # Unescaped Newlines in Strings ersetzen\n # Zwischen Quotes: \\n -> \\\\n\n def escape_newlines(match: re.Match) -> str:\n content = match.group(1)\n content = content.replace(\"\\n\", \"\\\\n\")\n content = content.replace(\"\\r\", \"\\\\r\")\n content = content.replace(\"\\t\", \"\\\\t\")\n return f'\"{content}\"'\n\n # Strings mit Newlines finden und escapen\n json_str = re.sub(r'\"([^\"]*(?:\\n|\\r)[^\"]*)\"', escape_newlines, json_str)\n\n return json_str\n\n\ndef safe_get(data: dict, key: str, default: Any = None, valid_values: set = None) -> Any:\n \"\"\"\n Sicherer Zugriff auf dict-Werte mit Validierung.\n\n Args:\n data: Source dict\n key: Schlüssel\n default: Fallback-Wert\n valid_values: Erlaubte Werte (optional)\n\n Returns:\n Validierter Wert oder Default\n \"\"\"\n value = data.get(key, default)\n\n # Liste -> erstes Element\n if isinstance(value, list):\n value = value[0] if value else default\n\n # String normalisieren\n if isinstance(value, str):\n value = value.lower().strip()\n\n # Validierung\n if valid_values and value not in valid_values:\n return default\n\n return value\n",
"numLines": 155,
"startLine": 1,
"totalLines": 155
}
}
}