{
"event": "PreToolUse",
"tool_name": "Read",
"tool_input": {
"file_path": "\/var\/www\/tools\/ki-protokoll\/claude-hook\/log_storage.py",
"limit": 100
}
}
{
"tool_response": {
"type": "text",
"file": {
"filePath": "\/var\/www\/tools\/ki-protokoll\/claude-hook\/log_storage.py",
"content": "#!\/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 session_id = data.get('session_id', '')\n event_name = data.get('hook_event_name', '')\n tool_name = data.get('tool_name', '')\n\n # Für Tool-Events: tool_name-spezifischer Key\n if event_name in ['PreToolUse', 'PostToolUse'] and tool_name:\n return f\"{session_id}_{tool_name}_{event_name}\"\n\n # Für andere Events: event-spezifischer Key\n return f\"{session_id}_{event_name}\"\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_data = {\n 'db_id': db_id,\n 'timestamp': datetime.datetime.now().isoformat(),\n 'event': data.get('hook_event_name'),\n 'tool_name': data.get('tool_name', ''),\n 'session_id': data.get('session_id', '')\n }\n (TEMP_DIR \/ f\"{key}.json\").write_text(json.dumps(tracking_data))\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 event_name = data.get('hook_event_name', '')\n if event_name != 'PostToolUse':\n return None\n\n search_data = dict(data)\n search_data['hook_event_name'] = 'PreToolUse'\n key = get_session_tracking_key(search_data)\n tracking_file = TEMP_DIR \/ f\"{key}.json\"\n\n if tracking_file.exists():\n tracking_data = json.loads(tracking_file.read_text())\n tracking_file.unlink()\n return tracking_data['db_id']\n except Exception as e:\n print(f\"Session tracking find error: {e}\", file=sys.stderr)\n\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()",
"numLines": 100,
"startLine": 1,
"totalLines": 218
}
}
}