{
"tool_response": {
"filePath": "\/var\/www\/tools\/ki-protokoll\/claude-hook\/task_hook.py",
"oldString": "'user': os.environ.get('CLAUDE_DB_USER', 'claude_code'),",
"newString": "'user': os.environ.get('CLAUDE_DB_USER', 'root'),",
"originalFile": "#!\/usr\/bin\/env python3\n\"\"\"\nTask Hook für Claude Code\nErstellt automatisch Tasks aus bestimmten Patterns in User-Prompts\n\nErkannte Patterns:\n- TODO: <text>\n- TASK: <text>\n- @task <text>\n- #task <text>\n\"\"\"\n\nimport json\nimport os\nimport sys\nimport re\nimport uuid\nimport pymysql\nfrom pathlib import Path\nfrom datetime import datetime\nfrom typing import Dict, Any, Optional, List\n\n# .env aus Hook-Verzeichnis laden\nfrom dotenv import load_dotenv\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', 'claude_code'),\n 'password': os.environ.get('CLAUDE_DB_PASSWORD', ''),\n 'database': os.environ.get('CLAUDE_DB_NAME', 'ki_protokoll'),\n 'charset': 'utf8mb4'\n}\n\n# Task-Erkennungsmuster\nTASK_PATTERNS = [\n (re.compile(r'TODO:\\s*(.+?)(?:\\n|$)', re.IGNORECASE), 'todo'),\n (re.compile(r'TASK:\\s*(.+?)(?:\\n|$)', re.IGNORECASE), 'task'),\n (re.compile(r'@task\\s+(.+?)(?:\\n|$)', re.IGNORECASE), 'mention'),\n (re.compile(r'#task\\s+(.+?)(?:\\n|$)', re.IGNORECASE), 'hashtag'),\n]\n\n\ndef get_connection():\n \"\"\"Erstellt Datenbankverbindung\"\"\"\n return pymysql.connect(**DB_CONFIG)\n\n\ndef generate_uuid() -> str:\n \"\"\"Generiert eine UUID\"\"\"\n return str(uuid.uuid4())\n\n\ndef extract_tasks(text: str) -> List[Dict[str, str]]:\n \"\"\"Extrahiert Tasks aus Text\"\"\"\n tasks = []\n\n for pattern, source in TASK_PATTERNS:\n matches = pattern.findall(text)\n for match in matches:\n title = match.strip()\n if title and len(title) >= 3: # Mindestlänge\n tasks.append({\n 'title': title[:255], # Max 255 Zeichen\n 'source': source\n })\n\n return tasks\n\n\ndef create_task(title: str, description: str, created_by: str) -> Optional[int]:\n \"\"\"Erstellt einen neuen Task in der Datenbank\"\"\"\n try:\n connection = get_connection()\n\n with connection.cursor() as cursor:\n task_uuid = generate_uuid()\n now = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')\n\n sql = \"\"\"\n INSERT INTO tasks (\n uuid, title, description, type, priority, status,\n created_by, created_by_type, created_at, updated_at\n ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)\n \"\"\"\n\n cursor.execute(sql, (\n task_uuid,\n title,\n description,\n 'ai_task',\n 'medium',\n 'pending',\n created_by,\n 'ai',\n now,\n now\n ))\n\n task_id = cursor.lastrowid\n\n # Kommentar hinzufügen\n comment_sql = \"\"\"\n INSERT INTO task_comments (\n task_id, author, author_type, comment_type, content, metadata, created_at\n ) VALUES (%s, %s, %s, %s, %s, %s, %s)\n \"\"\"\n\n cursor.execute(comment_sql, (\n task_id,\n 'claude-code-hook',\n 'system',\n 'note',\n 'Task automatisch aus User-Prompt erstellt',\n json.dumps({'source': 'task_hook', 'pattern': 'auto-detect'}),\n now\n ))\n\n connection.commit()\n return task_id\n\n except Exception as e:\n print(f\"Task creation error: {e}\", file=sys.stderr)\n return None\n finally:\n if 'connection' in locals():\n connection.close()\n\n\ndef process_event(data: Dict[str, Any]) -> None:\n \"\"\"Verarbeitet ein Claude Code Event\"\"\"\n event_name = data.get('hook_event_name', '')\n\n # Nur UserPromptSubmit verarbeiten\n if event_name != 'UserPromptSubmit':\n return\n\n # Prompt aus verschiedenen Stellen extrahieren\n prompt = ''\n\n if 'prompt' in data:\n prompt = data['prompt']\n elif 'tool_input' in data and isinstance(data['tool_input'], dict):\n prompt = data['tool_input'].get('prompt', '')\n prompt += ' ' + data['tool_input'].get('message', '')\n\n if not prompt:\n return\n\n # Tasks extrahieren\n tasks = extract_tasks(prompt)\n\n if not tasks:\n return\n\n # User ermitteln\n created_by = os.environ.get('USER', 'claude-code-hook')\n\n # Tasks erstellen\n created_count = 0\n for task in tasks:\n task_id = create_task(\n title=task['title'],\n description=f\"Auto-created from {task['source']} pattern in Claude Code prompt\",\n created_by=created_by\n )\n if task_id:\n created_count += 1\n print(f\"Task #{task_id} created: {task['title'][:50]}...\", file=sys.stderr)\n\n if created_count > 0:\n print(f\"Created {created_count} task(s) from prompt\", file=sys.stderr)\n\n\ndef main():\n \"\"\"Hauptfunktion\"\"\"\n try:\n # Input von stdin lesen\n input_data = json.load(sys.stdin)\n\n # Event verarbeiten\n process_event(input_data)\n\n # Immer continue zurückgeben (non-blocking)\n print(json.dumps({\"continue\": True}))\n sys.exit(0)\n\n except json.JSONDecodeError as e:\n print(f\"JSON decode error: {e}\", file=sys.stderr)\n print(json.dumps({\"continue\": True}))\n sys.exit(0)\n except Exception as e:\n print(f\"Hook error: {e}\", file=sys.stderr)\n print(json.dumps({\"continue\": True}))\n sys.exit(0)\n\n\nif __name__ == '__main__':\n main()\n",
"structuredPatch": [
{
"oldStart": 28,
"oldLines": 7,
"newStart": 28,
"newLines": 7,
"lines": [
" DB_CONFIG = {",
" 'host': os.environ.get('CLAUDE_DB_HOST', 'localhost'),",
" 'port': int(os.environ.get('CLAUDE_DB_PORT', '3306')),",
"- 'user': os.environ.get('CLAUDE_DB_USER', 'claude_code'),",
"+ 'user': os.environ.get('CLAUDE_DB_USER', 'root'),",
" 'password': os.environ.get('CLAUDE_DB_PASSWORD', ''),",
" 'database': os.environ.get('CLAUDE_DB_NAME', 'ki_protokoll'),",
" 'charset': 'utf8mb4'"
]
}
],
"userModified": false,
"replaceAll": false
}
}