{
"event": "PreToolUse",
"tool_name": "Edit",
"tool_input": {
"file_path": "\/var\/www\/tools\/ki-protokoll\/claude-hook\/log_parser.py",
"old_string": "def parse_event_data(data: Dict[str, Any]) -> Dict[str, Any]:\n \"\"\"Extrahiert und strukturiert Event-Daten basierend auf Event-Typ\"\"\"\n event_name = data.get('hook_event_name', 'Unknown')\n request_data = {}\n\n if event_name == 'UserPromptSubmit':\n request_data = {\n 'event': event_name,\n 'prompt': sanitize_data(data.get('prompt', ''))\n }\n\n elif event_name == 'PreToolUse':\n tool_name = data.get('tool_name', '')\n tool_input = sanitize_data(data.get('tool_input', {}))\n request_data = {\n 'event': event_name,\n 'tool_name': tool_name,\n 'tool_input': tool_input\n }\n\n elif event_name == 'PostToolUse':\n tool_response = sanitize_data(data.get('tool_response', {}))\n request_data = {\n 'event': event_name,\n 'tool_name': data.get('tool_name', ''),\n 'tool_response': tool_response\n }\n\n elif event_name in ['Stop', 'SubagentStop']:\n request_data = {\n 'event': event_name,\n 'stop_hook_active': data.get('stop_hook_active', False)\n }\n\n elif event_name in ['SessionStart', 'SessionEnd']:\n request_data = {\n 'event': event_name,\n 'source': data.get('source', data.get('reason', ''))\n }\n\n else:\n request_data = {\n 'event': event_name,\n 'raw_data': sanitize_data(data)\n }\n\n return request_data",
"new_string": "def parse_event_data(data: Dict[str, Any]) -> Dict[str, Any]:\n \"\"\"Extrahiert und strukturiert Event-Daten basierend auf Event-Typ\"\"\"\n evt = data.get('hook_event_name', 'Unknown')\n if evt == 'UserPromptSubmit':\n return {'event': evt, 'prompt': sanitize_data(data.get('prompt', ''))}\n elif evt == 'PreToolUse':\n return {'event': evt, 'tool_name': data.get('tool_name', ''),\n 'tool_input': sanitize_data(data.get('tool_input', {}))}\n elif evt == 'PostToolUse':\n return {'event': evt, 'tool_name': data.get('tool_name', ''),\n 'tool_response': sanitize_data(data.get('tool_response', {}))}\n elif evt in ['Stop', 'SubagentStop']:\n return {'event': evt, 'stop_hook_active': data.get('stop_hook_active', False)}\n elif evt in ['SessionStart', 'SessionEnd']:\n return {'event': evt, 'source': data.get('source', data.get('reason', ''))}\n else:\n return {'event': evt, 'raw_data': sanitize_data(data)}"
}
}
{
"tool_response": {
"filePath": "\/var\/www\/tools\/ki-protokoll\/claude-hook\/log_parser.py",
"oldString": "def parse_event_data(data: Dict[str, Any]) -> Dict[str, Any]:\n \"\"\"Extrahiert und strukturiert Event-Daten basierend auf Event-Typ\"\"\"\n event_name = data.get('hook_event_name', 'Unknown')\n request_data = {}\n\n if event_name == 'UserPromptSubmit':\n request_data = {\n 'event': event_name,\n 'prompt': sanitize_data(data.get('prompt', ''))\n }\n\n elif event_name == 'PreToolUse':\n tool_name = data.get('tool_name', '')\n tool_input = sanitize_data(data.get('tool_input', {}))\n request_data = {\n 'event': event_name,\n 'tool_name': tool_name,\n 'tool_input': tool_input\n }\n\n elif event_name == 'PostToolUse':\n tool_response = sanitize_data(data.get('tool_response', {}))\n request_data = {\n 'event': event_name,\n 'tool_name': data.get('tool_name', ''),\n 'tool_response': tool_response\n }\n\n elif event_name in ['Stop', 'SubagentStop']:\n request_data = {\n 'event': event_name,\n 'stop_hook_active': data.get('stop_hook_active', False)\n }\n\n elif event_name in ['SessionStart', 'SessionEnd']:\n request_data = {\n 'event': event_name,\n 'source': data.get('source', data.get('reason', ''))\n }\n\n else:\n request_data = {\n 'event': event_name,\n 'raw_data': sanitize_data(data)\n }\n\n return request_data",
"newString": "def parse_event_data(data: Dict[str, Any]) -> Dict[str, Any]:\n \"\"\"Extrahiert und strukturiert Event-Daten basierend auf Event-Typ\"\"\"\n evt = data.get('hook_event_name', 'Unknown')\n if evt == 'UserPromptSubmit':\n return {'event': evt, 'prompt': sanitize_data(data.get('prompt', ''))}\n elif evt == 'PreToolUse':\n return {'event': evt, 'tool_name': data.get('tool_name', ''),\n 'tool_input': sanitize_data(data.get('tool_input', {}))}\n elif evt == 'PostToolUse':\n return {'event': evt, 'tool_name': data.get('tool_name', ''),\n 'tool_response': sanitize_data(data.get('tool_response', {}))}\n elif evt in ['Stop', 'SubagentStop']:\n return {'event': evt, 'stop_hook_active': data.get('stop_hook_active', False)}\n elif evt in ['SessionStart', 'SessionEnd']:\n return {'event': evt, 'source': data.get('source', data.get('reason', ''))}\n else:\n return {'event': evt, 'raw_data': sanitize_data(data)}",
"originalFile": "#!\/usr\/bin\/env python3\n\"\"\"Log Parser Module - Parsen und Extrahieren von Daten aus Hook-Events\"\"\"\n\nimport json, os, re\nfrom pathlib import Path\nfrom typing import Dict, Any, List, Optional\n\nMAX_FIELD_LENGTH = 10000\nSENSITIVE_KEY_PATTERNS = re.compile(\n r\"(?i)(password|pass|secret|token|apikey|api_key|authorization|auth|bearer|credential)\")\nSENSITIVE_VALUE_PATTERNS = [\n re.compile(r\"(?i)\\bAKIA[0-9A-Z]{16}\\b\"),\n re.compile(r\"(?i)\\b(?:sk|rk|pk)[0-9A-Za-z]{20,}\\b\"),\n re.compile(r\"(?i)\\beyJ[a-zA-Z0-9-]{10,}\\.[a-zA-Z0-9_-]{10,}\\.[a-zA-Z0-9_-]{10,}\\b\")]\n\n\ndef sanitize_data(obj: Any) -> Any:\n \"\"\"Entfernt oder maskiert sensible Daten\"\"\"\n if isinstance(obj, dict):\n result = {}\n for key, value in obj.items():\n if SENSITIVE_KEY_PATTERNS.search(str(key)):\n result[key] = '[REDACTED]'\n else:\n result[key] = sanitize_data(value)\n return result\n elif isinstance(obj, list):\n return [sanitize_data(item) for item in obj]\n elif isinstance(obj, str):\n for pattern in SENSITIVE_VALUE_PATTERNS:\n if pattern.search(obj):\n return '[REDACTED]'\n if len(obj) > MAX_FIELD_LENGTH:\n import hashlib\n hash_value = hashlib.sha256(obj.encode('utf-8', errors='ignore')).hexdigest()[:16]\n return obj[:MAX_FIELD_LENGTH] + f'... [TRUNCATED-{hash_value}]'\n return obj\n return obj\n\n\ndef estimate_tokens(text: str) -> int:\n \"\"\"Grobe Token-Schätzung (4 Zeichen = 1 Token)\"\"\"\n if not text:\n return 0\n return max(1, len(text) \/\/ 4)\n\n\ndef get_client_ip() -> str:\n \"\"\"Ermittelt die Client-IP-Adresse\"\"\"\n ssh_client = os.environ.get('SSH_CLIENT', '')\n if ssh_client:\n return ssh_client.split()[0]\n ssh_connection = os.environ.get('SSH_CONNECTION', '')\n if ssh_connection:\n return ssh_connection.split()[0]\n return '127.0.0.1'\n\n\ndef parse_event_data(data: Dict[str, Any]) -> Dict[str, Any]:\n \"\"\"Extrahiert und strukturiert Event-Daten basierend auf Event-Typ\"\"\"\n event_name = data.get('hook_event_name', 'Unknown')\n request_data = {}\n\n if event_name == 'UserPromptSubmit':\n request_data = {\n 'event': event_name,\n 'prompt': sanitize_data(data.get('prompt', ''))\n }\n\n elif event_name == 'PreToolUse':\n tool_name = data.get('tool_name', '')\n tool_input = sanitize_data(data.get('tool_input', {}))\n request_data = {\n 'event': event_name,\n 'tool_name': tool_name,\n 'tool_input': tool_input\n }\n\n elif event_name == 'PostToolUse':\n tool_response = sanitize_data(data.get('tool_response', {}))\n request_data = {\n 'event': event_name,\n 'tool_name': data.get('tool_name', ''),\n 'tool_response': tool_response\n }\n\n elif event_name in ['Stop', 'SubagentStop']:\n request_data = {\n 'event': event_name,\n 'stop_hook_active': data.get('stop_hook_active', False)\n }\n\n elif event_name in ['SessionStart', 'SessionEnd']:\n request_data = {\n 'event': event_name,\n 'source': data.get('source', data.get('reason', ''))\n }\n\n else:\n request_data = {\n 'event': event_name,\n 'raw_data': sanitize_data(data)\n }\n\n return request_data\n\n\ndef get_conversation_pairs_from_session(session_id: str) -> List[Dict[str, str]]:\n \"\"\"Liest User-Prompt → Assistant-Response Paare aus der Session-JSONL\"\"\"\n pairs = []\n try:\n claude_dir = Path.home() \/ '.claude' \/ 'projects'\n\n for project_dir in claude_dir.iterdir():\n if not project_dir.is_dir():\n continue\n\n session_file = project_dir \/ f\"{session_id}.jsonl\"\n if session_file.exists():\n lines = session_file.read_text().strip().split('\\n')\n\n current_user_prompt = None\n for line in lines:\n try:\n entry = json.loads(line)\n entry_type = entry.get('type', '')\n\n # User-Message gefunden\n if entry_type == 'user':\n message = entry.get('message', {})\n content = message.get('content', '')\n if isinstance(content, str) and content.strip():\n current_user_prompt = content.strip()\n\n # Assistant-Message gefunden - pair mit letztem User-Prompt\n elif entry_type == 'assistant' and current_user_prompt:\n message = entry.get('message', {})\n content_parts = message.get('content', [])\n text_parts = []\n for part in content_parts:\n if isinstance(part, dict) and part.get('type') == 'text':\n text_parts.append(part.get('text', ''))\n elif isinstance(part, str):\n text_parts.append(part)\n\n if text_parts:\n assistant_response = '\\n'.join(text_parts)\n pairs.append({\n 'prompt': current_user_prompt,\n 'response': sanitize_data(assistant_response)\n })\n current_user_prompt = None # Reset für nächstes Paar\n\n except json.JSONDecodeError:\n continue\n\n break # Session gefunden, aufhören zu suchen\n\n return pairs\n\n except Exception:\n return []\n",
"structuredPatch": [
{
"oldStart": 58,
"oldLines": 53,
"newStart": 58,
"newLines": 23,
"lines": [
" ",
" def parse_event_data(data: Dict[str, Any]) -> Dict[str, Any]:",
" \"\"\"Extrahiert und strukturiert Event-Daten basierend auf Event-Typ\"\"\"",
"- event_name = data.get('hook_event_name', 'Unknown')",
"- request_data = {}",
"-",
"- if event_name == 'UserPromptSubmit':",
"- request_data = {",
"- 'event': event_name,",
"- 'prompt': sanitize_data(data.get('prompt', ''))",
"- }",
"-",
"- elif event_name == 'PreToolUse':",
"- tool_name = data.get('tool_name', '')",
"- tool_input = sanitize_data(data.get('tool_input', {}))",
"- request_data = {",
"- 'event': event_name,",
"- 'tool_name': tool_name,",
"- 'tool_input': tool_input",
"- }",
"-",
"- elif event_name == 'PostToolUse':",
"- tool_response = sanitize_data(data.get('tool_response', {}))",
"- request_data = {",
"- 'event': event_name,",
"- 'tool_name': data.get('tool_name', ''),",
"- 'tool_response': tool_response",
"- }",
"-",
"- elif event_name in ['Stop', 'SubagentStop']:",
"- request_data = {",
"- 'event': event_name,",
"- 'stop_hook_active': data.get('stop_hook_active', False)",
"- }",
"-",
"- elif event_name in ['SessionStart', 'SessionEnd']:",
"- request_data = {",
"- 'event': event_name,",
"- 'source': data.get('source', data.get('reason', ''))",
"- }",
"-",
"+ evt = data.get('hook_event_name', 'Unknown')",
"+ if evt == 'UserPromptSubmit':",
"+ return {'event': evt, 'prompt': sanitize_data(data.get('prompt', ''))}",
"+ elif evt == 'PreToolUse':",
"+ return {'event': evt, 'tool_name': data.get('tool_name', ''),",
"+ 'tool_input': sanitize_data(data.get('tool_input', {}))}",
"+ elif evt == 'PostToolUse':",
"+ return {'event': evt, 'tool_name': data.get('tool_name', ''),",
"+ 'tool_response': sanitize_data(data.get('tool_response', {}))}",
"+ elif evt in ['Stop', 'SubagentStop']:",
"+ return {'event': evt, 'stop_hook_active': data.get('stop_hook_active', False)}",
"+ elif evt in ['SessionStart', 'SessionEnd']:",
"+ return {'event': evt, 'source': data.get('source', data.get('reason', ''))}",
" else:",
"- request_data = {",
"- 'event': event_name,",
"- 'raw_data': sanitize_data(data)",
"- }",
"+ return {'event': evt, 'raw_data': sanitize_data(data)}",
" ",
"- return request_data",
" ",
"-",
" def get_conversation_pairs_from_session(session_id: str) -> List[Dict[str, str]]:",
" \"\"\"Liest User-Prompt → Assistant-Response Paare aus der Session-JSONL\"\"\"",
" pairs = []"
]
}
],
"userModified": false,
"replaceAll": false
}
}