{
"event": "PreToolUse",
"tool_name": "Edit",
"tool_input": {
"file_path": "\/var\/www\/tools\/ki-protokoll\/claude-hook\/quality\/task_creator.py",
"old_string": "def get_db_connection():\n \"\"\"Erstellt DB-Verbindung zu ki_dev.\"\"\"\n return mysql.connector.connect(\n host=\"localhost\",\n user=\"root\",\n password=os.environ.get(\"MARIADB_ROOT_PASSWORD\", \"\"),\n database=\"ki_dev\"\n )",
"new_string": "def get_db_connection():\n \"\"\"Erstellt DB-Verbindung zu ki_dev.\"\"\"\n return mysql.connector.connect(\n host=os.environ.get(\"CLAUDE_DB_HOST\", \"localhost\"),\n user=os.environ.get(\"CLAUDE_DB_USER\", \"root\"),\n password=os.environ.get(\"CLAUDE_DB_PASSWORD\", \"\"),\n database=os.environ.get(\"CLAUDE_DB_NAME\", \"ki_dev\")\n )"
}
}
{
"tool_response": {
"filePath": "\/var\/www\/tools\/ki-protokoll\/claude-hook\/quality\/task_creator.py",
"oldString": "def get_db_connection():\n \"\"\"Erstellt DB-Verbindung zu ki_dev.\"\"\"\n return mysql.connector.connect(\n host=\"localhost\",\n user=\"root\",\n password=os.environ.get(\"MARIADB_ROOT_PASSWORD\", \"\"),\n database=\"ki_dev\"\n )",
"newString": "def get_db_connection():\n \"\"\"Erstellt DB-Verbindung zu ki_dev.\"\"\"\n return mysql.connector.connect(\n host=os.environ.get(\"CLAUDE_DB_HOST\", \"localhost\"),\n user=os.environ.get(\"CLAUDE_DB_USER\", \"root\"),\n password=os.environ.get(\"CLAUDE_DB_PASSWORD\", \"\"),\n database=os.environ.get(\"CLAUDE_DB_NAME\", \"ki_dev\")\n )",
"originalFile": "#!\/usr\/bin\/env python3\n\"\"\"\nTask Creator - Erstellt Tasks für Quality Violations.\n\nSchreibt direkt in ki_dev.tasks via MariaDB.\n\"\"\"\n\nimport json\nimport os\nimport uuid\nfrom datetime import datetime\nfrom pathlib import Path\nfrom typing import Optional\n\nimport mysql.connector\n\n\ndef get_db_connection():\n \"\"\"Erstellt DB-Verbindung zu ki_dev.\"\"\"\n return mysql.connector.connect(\n host=\"localhost\",\n user=\"root\",\n password=os.environ.get(\"MARIADB_ROOT_PASSWORD\", \"\"),\n database=\"ki_dev\"\n )\n\n\ndef create_violation_task(\n file_path: str,\n rule_id: str,\n message: str,\n severity: str = \"block\" # \"block\" oder \"warn\"\n) -> Optional[int]:\n \"\"\"\n Erstellt einen Task für eine Quality Violation.\n\n Args:\n file_path: Pfad zur betroffenen Datei\n rule_id: Regel-ID (z.B. \"P1.1\", \"W1.2\")\n message: Vollständige Fehlermeldung\n severity: \"block\" (Pre-Hook) oder \"warn\" (Post-Hook)\n\n Returns:\n Task-ID oder None bei Fehler\n \"\"\"\n filename = Path(file_path).name\n\n # Task-Titel\n title = f\"Quality [{rule_id}]: {filename}\"\n\n # Task-Beschreibung mit Struktur\n description = f\"\"\"## Violation\n\n**Datei:** `{file_path}`\n**Regel:** {rule_id}\n**Severity:** {severity.upper()}\n\n**Meldung:**\n{message}\n\n---\n\n## Analyse\n\n*Zu analysieren:*\n- Was ist die Ursache der Violation?\n- Welche Abhängigkeiten sind betroffen?\n- Gibt es ähnliche Violations in anderen Dateien?\n\n---\n\n## Plan\n\n*Zu planen:*\n1. [ ] Violation verstehen\n2. [ ] Lösung entwerfen\n3. [ ] Änderungen implementieren\n4. [ ] Testen\n\n---\n\n## Umsetzung\n\n*Nach Abschluss dokumentieren:*\n- Was wurde geändert?\n- Welche Dateien wurden angepasst?\n- Sind weitere Änderungen nötig?\n\"\"\"\n\n # Metadata\n metadata = {\n \"source\": \"quality-hook\",\n \"file_path\": file_path,\n \"rule_id\": rule_id,\n \"severity\": severity,\n \"message\": message,\n \"detected_at\": datetime.now().isoformat()\n }\n\n try:\n conn = get_db_connection()\n cursor = conn.cursor()\n\n # Prüfe ob bereits ein offener Task für diese Datei+Regel existiert\n cursor.execute(\"\"\"\n SELECT id FROM tasks\n WHERE status IN ('pending', 'in_progress')\n AND JSON_EXTRACT(metadata, '$.file_path') = %s\n AND JSON_EXTRACT(metadata, '$.rule_id') = %s\n LIMIT 1\n \"\"\", (file_path, rule_id))\n\n existing = cursor.fetchone()\n if existing:\n # Task existiert bereits, nicht duplizieren\n cursor.close()\n conn.close()\n return existing[0]\n\n # Neuen Task erstellen\n cursor.execute(\"\"\"\n INSERT INTO tasks (uuid, title, description, type, status, created_by, created_by_type, metadata)\n VALUES (%s, %s, %s, %s, %s, %s, %s, %s)\n \"\"\", (\n str(uuid.uuid4()),\n title,\n description,\n \"ai_task\",\n \"pending\",\n \"quality-hook\",\n \"ai\",\n json.dumps(metadata)\n ))\n\n task_id = cursor.lastrowid\n conn.commit()\n cursor.close()\n conn.close()\n\n return task_id\n\n except Exception as e:\n # Bei DB-Fehler nicht crashen, nur loggen\n import sys\n print(f\"[QUALITY] Task creation failed: {e}\", file=sys.stderr)\n return None\n\n\ndef create_tasks_for_warnings(file_path: str, warnings: list) -> list:\n \"\"\"\n Erstellt Tasks für mehrere Warnungen.\n\n Returns:\n Liste der erstellten Task-IDs\n \"\"\"\n task_ids = []\n\n for warning in warnings:\n # Parse Rule-ID aus Warnung (Format: \"W1.2: ...\")\n parts = warning.split(\":\", 1)\n if len(parts) >= 2:\n rule_id = parts[0].strip()\n message = parts[1].strip()\n else:\n rule_id = \"UNKNOWN\"\n message = warning\n\n task_id = create_violation_task(file_path, rule_id, message, \"warn\")\n if task_id:\n task_ids.append(task_id)\n\n return task_ids\n",
"structuredPatch": [
{
"oldStart": 18,
"oldLines": 10,
"newStart": 18,
"newLines": 10,
"lines": [
" def get_db_connection():",
" \"\"\"Erstellt DB-Verbindung zu ki_dev.\"\"\"",
" return mysql.connector.connect(",
"- host=\"localhost\",",
"- user=\"root\",",
"- password=os.environ.get(\"MARIADB_ROOT_PASSWORD\", \"\"),",
"- database=\"ki_dev\"",
"+ host=os.environ.get(\"CLAUDE_DB_HOST\", \"localhost\"),",
"+ user=os.environ.get(\"CLAUDE_DB_USER\", \"root\"),",
"+ password=os.environ.get(\"CLAUDE_DB_PASSWORD\", \"\"),",
"+ database=os.environ.get(\"CLAUDE_DB_NAME\", \"ki_dev\")",
" )",
" ",
" "
]
}
],
"userModified": false,
"replaceAll": false
}
}