Protokoll #16255

{
    "event": "PreToolUse",
    "tool_name": "Edit",
    "tool_input": {
        "file_path": "\/var\/www\/tools\/ki-protokoll\/claude-hook\/log_storage.py",
        "old_string": "def _get_pending_prompts(cursor) -> List:\n    \"\"\"Holt pending UserPromptSubmit-Einträge\"\"\"\n    cursor.execute(\"\"\"\n        SELECT id, JSON_UNQUOTE(JSON_EXTRACT(request, '$.prompt')) as prompt\n        FROM protokoll\n        WHERE status = 'pending'\n          AND JSON_EXTRACT(request, '$.event') = 'UserPromptSubmit'\n          AND timestamp > DATE_SUB(NOW(), INTERVAL 1 HOUR)\n        ORDER BY id ASC\n    \"\"\")\n    return cursor.fetchall()\n\n\ndef _find_matching_response(db_prompt: str, pairs: List[Dict[str, str]]) -> Optional[str]:\n    \"\"\"Findet matching Response für einen Prompt\"\"\"\n    for pair in pairs:\n        if pair['prompt'].strip() == db_prompt.strip():\n            return pair['response']\n    return None\n\n\ndef _update_prompt_with_response(cursor, db_id: int, response: str, current_time) -> None:\n    \"\"\"Updated einen pending Prompt mit seiner Response\"\"\"\n    from log_parser import estimate_tokens\n\n    tokens_output = estimate_tokens(response)\n    response_json = json.dumps({'assistant_response': response}, ensure_ascii=False)\n\n    cursor.execute(\"\"\"\n        UPDATE protokoll\n        SET response = %s,\n            response_timestamp = %s,\n            tokens_output = %s,\n            tokens_total = tokens_input + %s,\n            duration_ms = ROUND(TIMESTAMPDIFF(MICROSECOND, request_timestamp, %s) \/ 1000, 3),\n            status = 'completed'\n        WHERE id = %s\n    \"\"\", (response_json, current_time, tokens_output, tokens_output, current_time, db_id))",
        "new_string": "def _get_pending_prompts(cursor) -> List:\n    \"\"\"Holt pending UserPromptSubmit-Einträge\"\"\"\n    cursor.execute(\"\"\"SELECT id, JSON_UNQUOTE(JSON_EXTRACT(request, '$.prompt')) as prompt\n        FROM protokoll WHERE status = 'pending' AND JSON_EXTRACT(request, '$.event') = 'UserPromptSubmit'\n        AND timestamp > DATE_SUB(NOW(), INTERVAL 1 HOUR) ORDER BY id ASC\"\"\")\n    return cursor.fetchall()\n\n\ndef _find_matching_response(db_prompt: str, pairs: List[Dict[str, str]]) -> Optional[str]:\n    \"\"\"Findet matching Response für einen Prompt\"\"\"\n    for pair in pairs:\n        if pair['prompt'].strip() == db_prompt.strip():\n            return pair['response']\n    return None\n\n\ndef _update_prompt_with_response(cursor, db_id: int, response: str, current_time) -> None:\n    \"\"\"Updated einen pending Prompt mit seiner Response\"\"\"\n    from log_parser import estimate_tokens\n    tokens = estimate_tokens(response)\n    resp_json = json.dumps({'assistant_response': response}, ensure_ascii=False)\n    cursor.execute(\"\"\"UPDATE protokoll SET response = %s, response_timestamp = %s, tokens_output = %s,\n        tokens_total = tokens_input + %s,\n        duration_ms = ROUND(TIMESTAMPDIFF(MICROSECOND, request_timestamp, %s) \/ 1000, 3),\n        status = 'completed' WHERE id = %s\"\"\", (resp_json, current_time, tokens, tokens, current_time, db_id))"
    }
}

{
    "tool_response": {
        "filePath": "\/var\/www\/tools\/ki-protokoll\/claude-hook\/log_storage.py",
        "oldString": "def _get_pending_prompts(cursor) -> List:\n    \"\"\"Holt pending UserPromptSubmit-Einträge\"\"\"\n    cursor.execute(\"\"\"\n        SELECT id, JSON_UNQUOTE(JSON_EXTRACT(request, '$.prompt')) as prompt\n        FROM protokoll\n        WHERE status = 'pending'\n          AND JSON_EXTRACT(request, '$.event') = 'UserPromptSubmit'\n          AND timestamp > DATE_SUB(NOW(), INTERVAL 1 HOUR)\n        ORDER BY id ASC\n    \"\"\")\n    return cursor.fetchall()\n\n\ndef _find_matching_response(db_prompt: str, pairs: List[Dict[str, str]]) -> Optional[str]:\n    \"\"\"Findet matching Response für einen Prompt\"\"\"\n    for pair in pairs:\n        if pair['prompt'].strip() == db_prompt.strip():\n            return pair['response']\n    return None\n\n\ndef _update_prompt_with_response(cursor, db_id: int, response: str, current_time) -> None:\n    \"\"\"Updated einen pending Prompt mit seiner Response\"\"\"\n    from log_parser import estimate_tokens\n\n    tokens_output = estimate_tokens(response)\n    response_json = json.dumps({'assistant_response': response}, ensure_ascii=False)\n\n    cursor.execute(\"\"\"\n        UPDATE protokoll\n        SET response = %s,\n            response_timestamp = %s,\n            tokens_output = %s,\n            tokens_total = tokens_input + %s,\n            duration_ms = ROUND(TIMESTAMPDIFF(MICROSECOND, request_timestamp, %s) \/ 1000, 3),\n            status = 'completed'\n        WHERE id = %s\n    \"\"\", (response_json, current_time, tokens_output, tokens_output, current_time, db_id))",
        "newString": "def _get_pending_prompts(cursor) -> List:\n    \"\"\"Holt pending UserPromptSubmit-Einträge\"\"\"\n    cursor.execute(\"\"\"SELECT id, JSON_UNQUOTE(JSON_EXTRACT(request, '$.prompt')) as prompt\n        FROM protokoll WHERE status = 'pending' AND JSON_EXTRACT(request, '$.event') = 'UserPromptSubmit'\n        AND timestamp > DATE_SUB(NOW(), INTERVAL 1 HOUR) ORDER BY id ASC\"\"\")\n    return cursor.fetchall()\n\n\ndef _find_matching_response(db_prompt: str, pairs: List[Dict[str, str]]) -> Optional[str]:\n    \"\"\"Findet matching Response für einen Prompt\"\"\"\n    for pair in pairs:\n        if pair['prompt'].strip() == db_prompt.strip():\n            return pair['response']\n    return None\n\n\ndef _update_prompt_with_response(cursor, db_id: int, response: str, current_time) -> None:\n    \"\"\"Updated einen pending Prompt mit seiner Response\"\"\"\n    from log_parser import estimate_tokens\n    tokens = estimate_tokens(response)\n    resp_json = json.dumps({'assistant_response': response}, ensure_ascii=False)\n    cursor.execute(\"\"\"UPDATE protokoll SET response = %s, response_timestamp = %s, tokens_output = %s,\n        tokens_total = tokens_input + %s,\n        duration_ms = ROUND(TIMESTAMPDIFF(MICROSECOND, request_timestamp, %s) \/ 1000, 3),\n        status = 'completed' WHERE id = %s\"\"\", (resp_json, current_time, tokens, tokens, current_time, db_id))",
        "originalFile": "#!\/usr\/bin\/env python3\n\"\"\"\nLog Storage Module\nVerantwortlich für die Speicherung und Aktualisierung von Logs in der Datenbank\n\"\"\"\n\nimport os\nimport json\nimport datetime\nimport pymysql\nimport sys\nimport tempfile\nfrom pathlib import Path\nfrom typing import Dict, Any, Optional, List\nfrom dotenv import load_dotenv\n\n# .env aus Hook-Verzeichnis laden\nload_dotenv(Path(__file__).parent \/ '.env')\n\n# Konfiguration aus Environment-Variablen\nDB_CONFIG = {\n    'host': os.environ.get('CLAUDE_DB_HOST', 'localhost'),\n    'port': int(os.environ.get('CLAUDE_DB_PORT', '3306')),\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    'charset': 'utf8mb4'\n}\n\n# Session-Tracking im temporären Verzeichnis\nTEMP_DIR = Path(tempfile.gettempdir()) \/ \"claude_hooks\"\nTEMP_DIR.mkdir(exist_ok=True)\n\n\ndef get_session_tracking_key(data: Dict[str, Any]) -> str:\n    \"\"\"Erstellt einen eindeutigen Key für Session-Tracking\"\"\"\n    sid = data.get('session_id', '')\n    evt = data.get('hook_event_name', '')\n    tool = data.get('tool_name', '')\n    if evt in ['PreToolUse', 'PostToolUse'] and tool:\n        return f\"{sid}_{tool}_{evt}\"\n    return f\"{sid}_{evt}\"\n\n\ndef save_pending_request(data: Dict[str, Any], db_id: int) -> None:\n    \"\"\"Speichert pending Request für spätere Response-Zuordnung\"\"\"\n    try:\n        key = get_session_tracking_key(data)\n        tracking = {'db_id': db_id, 'timestamp': datetime.datetime.now().isoformat(),\n                    'event': data.get('hook_event_name'), 'tool_name': data.get('tool_name', ''),\n                    'session_id': data.get('session_id', '')}\n        (TEMP_DIR \/ f\"{key}.json\").write_text(json.dumps(tracking))\n    except Exception as e:\n        print(f\"Session tracking save error: {e}\", file=sys.stderr)\n\n\ndef find_matching_request(data: Dict[str, Any]) -> Optional[int]:\n    \"\"\"Findet matching Request für Response-Event\"\"\"\n    try:\n        if data.get('hook_event_name', '') != 'PostToolUse':\n            return None\n        search = dict(data)\n        search['hook_event_name'] = 'PreToolUse'\n        key = get_session_tracking_key(search)\n        track_file = TEMP_DIR \/ f\"{key}.json\"\n        if track_file.exists():\n            track_data = json.loads(track_file.read_text())\n            track_file.unlink()\n            return track_data['db_id']\n    except Exception as e:\n        print(f\"Session tracking find error: {e}\", file=sys.stderr)\n    return None\n\n\ndef insert_log_entry(\n    request_str: str,\n    client_ip: str,\n    client_name: str,\n    response_str: Optional[str],\n    tokens_input: int,\n    tokens_output: int,\n    model_name: str\n) -> Optional[int]:\n    \"\"\"Fügt einen neuen Log-Eintrag in die Datenbank ein\"\"\"\n    try:\n        connection = pymysql.connect(**DB_CONFIG)\n        with connection.cursor() as cursor:\n            current_time = datetime.datetime.now()\n            tokens_total = tokens_input + tokens_output\n            status = 'completed' if response_str else 'pending'\n            response_timestamp = current_time if response_str else None\n\n            cursor.execute(\"\"\"\n                INSERT INTO protokoll (\n                    timestamp, request_ip, client_name, request, request_timestamp,\n                    response, response_timestamp, duration_ms, tokens_input,\n                    tokens_output, tokens_total, model_name, status\n                ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)\n            \"\"\", (\n                current_time, client_ip, client_name, request_str, current_time,\n                response_str, response_timestamp, 0, tokens_input, tokens_output,\n                tokens_total, model_name, status\n            ))\n\n            connection.commit()\n            return cursor.lastrowid\n    except Exception as e:\n        print(f\"Database insert error: {e}\", file=sys.stderr)\n        return None\n    finally:\n        if 'connection' in locals():\n            connection.close()\n\n\ndef update_request_with_response(db_id: int, response_data: str, tokens_output: int) -> None:\n    \"\"\"Updated existing Request mit Response-Daten und berechnet Duration\"\"\"\n    try:\n        connection = pymysql.connect(**DB_CONFIG)\n        with connection.cursor() as cursor:\n            current_time = datetime.datetime.now()\n\n            cursor.execute(\"\"\"\n                UPDATE protokoll\n                SET response = %s, response_timestamp = %s, tokens_output = %s,\n                    tokens_total = tokens_input + %s, status = 'completed'\n                WHERE id = %s\n            \"\"\", (response_data, current_time, tokens_output, tokens_output, db_id))\n\n            cursor.execute(\"\"\"\n                UPDATE protokoll\n                SET duration_ms = ROUND(TIMESTAMPDIFF(MICROSECOND, request_timestamp, response_timestamp) \/ 1000, 3)\n                WHERE id = %s\n            \"\"\", (db_id,))\n\n            connection.commit()\n    except Exception as e:\n        print(f\"Database update error: {e}\", file=sys.stderr)\n    finally:\n        if 'connection' in locals():\n            connection.close()\n\n\ndef _get_pending_prompts(cursor) -> List:\n    \"\"\"Holt pending UserPromptSubmit-Einträge\"\"\"\n    cursor.execute(\"\"\"\n        SELECT id, JSON_UNQUOTE(JSON_EXTRACT(request, '$.prompt')) as prompt\n        FROM protokoll\n        WHERE status = 'pending'\n          AND JSON_EXTRACT(request, '$.event') = 'UserPromptSubmit'\n          AND timestamp > DATE_SUB(NOW(), INTERVAL 1 HOUR)\n        ORDER BY id ASC\n    \"\"\")\n    return cursor.fetchall()\n\n\ndef _find_matching_response(db_prompt: str, pairs: List[Dict[str, str]]) -> Optional[str]:\n    \"\"\"Findet matching Response für einen Prompt\"\"\"\n    for pair in pairs:\n        if pair['prompt'].strip() == db_prompt.strip():\n            return pair['response']\n    return None\n\n\ndef _update_prompt_with_response(cursor, db_id: int, response: str, current_time) -> None:\n    \"\"\"Updated einen pending Prompt mit seiner Response\"\"\"\n    from log_parser import estimate_tokens\n\n    tokens_output = estimate_tokens(response)\n    response_json = json.dumps({'assistant_response': response}, ensure_ascii=False)\n\n    cursor.execute(\"\"\"\n        UPDATE protokoll\n        SET response = %s,\n            response_timestamp = %s,\n            tokens_output = %s,\n            tokens_total = tokens_input + %s,\n            duration_ms = ROUND(TIMESTAMPDIFF(MICROSECOND, request_timestamp, %s) \/ 1000, 3),\n            status = 'completed'\n        WHERE id = %s\n    \"\"\", (response_json, current_time, tokens_output, tokens_output, current_time, db_id))\n\n\ndef close_pending_user_prompts(session_id: str, conversation_pairs: List[Dict[str, str]]) -> None:\n    \"\"\"Schließt offene UserPromptSubmit-Einträge mit ihren zugehörigen Responses\"\"\"\n    if not conversation_pairs:\n        return\n\n    try:\n        connection = pymysql.connect(**DB_CONFIG)\n        with connection.cursor() as cursor:\n            pending_prompts = _get_pending_prompts(cursor)\n            current_time = datetime.datetime.now()\n\n            for db_id, db_prompt in pending_prompts:\n                if db_prompt:\n                    matching_response = _find_matching_response(db_prompt, conversation_pairs)\n                    if matching_response:\n                        _update_prompt_with_response(cursor, db_id, matching_response, current_time)\n\n            connection.commit()\n    except Exception as e:\n        print(f\"Close pending prompts error: {e}\", file=sys.stderr)\n    finally:\n        if 'connection' in locals():\n            connection.close()\n",
        "structuredPatch": [
            {
                "oldStart": 142,
                "oldLines": 14,
                "newStart": 142,
                "newLines": 9,
                "lines": [
                    " ",
                    " def _get_pending_prompts(cursor) -> List:",
                    "     \"\"\"Holt pending UserPromptSubmit-Einträge\"\"\"",
                    "-    cursor.execute(\"\"\"",
                    "-        SELECT id, JSON_UNQUOTE(JSON_EXTRACT(request, '$.prompt')) as prompt",
                    "-        FROM protokoll",
                    "-        WHERE status = 'pending'",
                    "-          AND JSON_EXTRACT(request, '$.event') = 'UserPromptSubmit'",
                    "-          AND timestamp > DATE_SUB(NOW(), INTERVAL 1 HOUR)",
                    "-        ORDER BY id ASC",
                    "-    \"\"\")",
                    "+    cursor.execute(\"\"\"SELECT id, JSON_UNQUOTE(JSON_EXTRACT(request, '$.prompt')) as prompt",
                    "+        FROM protokoll WHERE status = 'pending' AND JSON_EXTRACT(request, '$.event') = 'UserPromptSubmit'",
                    "+        AND timestamp > DATE_SUB(NOW(), INTERVAL 1 HOUR) ORDER BY id ASC\"\"\")",
                    "     return cursor.fetchall()",
                    " ",
                    " "
                ]
            },
            {
                "oldStart": 164,
                "oldLines": 22,
                "newStart": 159,
                "newLines": 14,
                "lines": [
                    " def _update_prompt_with_response(cursor, db_id: int, response: str, current_time) -> None:",
                    "     \"\"\"Updated einen pending Prompt mit seiner Response\"\"\"",
                    "     from log_parser import estimate_tokens",
                    "+    tokens = estimate_tokens(response)",
                    "+    resp_json = json.dumps({'assistant_response': response}, ensure_ascii=False)",
                    "+    cursor.execute(\"\"\"UPDATE protokoll SET response = %s, response_timestamp = %s, tokens_output = %s,",
                    "+        tokens_total = tokens_input + %s,",
                    "+        duration_ms = ROUND(TIMESTAMPDIFF(MICROSECOND, request_timestamp, %s) \/ 1000, 3),",
                    "+        status = 'completed' WHERE id = %s\"\"\", (resp_json, current_time, tokens, tokens, current_time, db_id))",
                    " ",
                    "-    tokens_output = estimate_tokens(response)",
                    "-    response_json = json.dumps({'assistant_response': response}, ensure_ascii=False)",
                    " ",
                    "-    cursor.execute(\"\"\"",
                    "-        UPDATE protokoll",
                    "-        SET response = %s,",
                    "-            response_timestamp = %s,",
                    "-            tokens_output = %s,",
                    "-            tokens_total = tokens_input + %s,",
                    "-            duration_ms = ROUND(TIMESTAMPDIFF(MICROSECOND, request_timestamp, %s) \/ 1000, 3),",
                    "-            status = 'completed'",
                    "-        WHERE id = %s",
                    "-    \"\"\", (response_json, current_time, tokens_output, tokens_output, current_time, db_id))",
                    "-",
                    "-",
                    " def close_pending_user_prompts(session_id: str, conversation_pairs: List[Dict[str, str]]) -> None:",
                    "     \"\"\"Schließt offene UserPromptSubmit-Einträge mit ihren zugehörigen Responses\"\"\"",
                    "     if not conversation_pairs:"
                ]
            }
        ],
        "userModified": false,
        "replaceAll": false
    }
}