Protokoll #1029

{
    "event": "PreToolUse",
    "tool_name": "Read",
    "tool_input": {
        "file_path": "\/var\/www\/tools\/ki-protokoll\/claude-hook\/file_backup_hook.py"
    }
}

{
    "tool_response": {
        "type": "text",
        "file": {
            "filePath": "\/var\/www\/tools\/ki-protokoll\/claude-hook\/file_backup_hook.py",
            "content": "#!\/usr\/bin\/env python3\n\"\"\"\nFile Backup Hook für Claude Code PreToolUse Events\n\nSichert Dateien automatisch in file_backup_history BEVOR sie von\nClaude Code (Edit\/Write Tools) geändert werden.\n\nTrigger: PreToolUse (Edit, Write)\nTarget: ki_protokoll.file_backup_history\n\nVersion: 1.0\nErstellt: 2025-12-20\n\"\"\"\n\nimport json\nimport os\nimport sys\nimport hashlib\nimport pymysql\nfrom pathlib import Path\nfrom typing import Dict, Any, Optional\nfrom datetime import datetime\n\n# .env aus Hook-Verzeichnis laden\nfrom dotenv import load_dotenv\nload_dotenv(Path(__file__).parent \/ '.env')\n\n# =============================================================================\n# KONFIGURATION\n# =============================================================================\n\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# Welche Verzeichnisse sollen gesichert werden?\nBACKUP_DIRS = [\n    '\/var\/www\/dev.campus.systemische-tools.de\/src',\n    '\/var\/www\/dev.campus.systemische-tools.de\/public',\n    '\/var\/www\/dev.campus.systemische-tools.de\/scripts',\n    '\/var\/www\/dev.campus.systemische-tools.de\/includes',\n    '\/var\/www\/dev.campus.systemische-tools.de\/Views',\n    '\/var\/www\/dev.campus.systemische-tools.de\/config',\n    '\/var\/www\/dev.campus.systemische-tools.de\/adm',\n    '\/var\/www\/prod.campus.systemische-tools.de\/src',\n    '\/var\/www\/prod.campus.systemische-tools.de\/public',\n    '\/var\/www\/prod.campus.systemische-tools.de\/scripts',\n    '\/var\/www\/prod.campus.systemische-tools.de\/includes',\n    '\/var\/www\/prod.campus.systemische-tools.de\/Views',\n    '\/var\/www\/prod.campus.systemische-tools.de\/config',\n]\n\n# Dateien die NICHT gesichert werden (Patterns)\nEXCLUDE_PATTERNS = [\n    '\/vendor\/',\n    '\/node_modules\/',\n    '\/.git\/',\n    '\/backups\/',\n    '\/tmp\/',\n    '\/logs\/',\n    '\/cache\/',\n    '.log',\n    '.cache',\n    '.tmp'\n]\n\n# Maximale Dateigröße für Backup (10 MB)\nMAX_FILE_SIZE = 10 * 1024 * 1024\n\n# =============================================================================\n# HILFSFUNKTIONEN\n# =============================================================================\n\ndef should_backup(file_path: str) -> bool:\n    \"\"\"Prüft, ob die Datei gesichert werden soll\"\"\"\n    if not file_path:\n        return False\n\n    # Existiert die Datei?\n    if not os.path.isfile(file_path):\n        return False\n\n    # Liegt sie in einem Backup-Verzeichnis?\n    in_backup_dir = any(file_path.startswith(d) for d in BACKUP_DIRS)\n    if not in_backup_dir:\n        return False\n\n    # Ist sie ausgeschlossen?\n    for pattern in EXCLUDE_PATTERNS:\n        if pattern in file_path:\n            return False\n\n    # Dateigröße prüfen\n    try:\n        if os.path.getsize(file_path) > MAX_FILE_SIZE:\n            return False\n    except OSError:\n        return False\n\n    return True\n\n\ndef calculate_hash(content: str) -> str:\n    \"\"\"Berechnet SHA256 Hash des Inhalts\"\"\"\n    return hashlib.sha256(content.encode('utf-8', errors='ignore')).hexdigest()\n\n\ndef get_next_version(cursor, file_path: str) -> int:\n    \"\"\"Ermittelt die nächste Versionsnummer für eine Datei\"\"\"\n    cursor.execute(\n        \"SELECT MAX(version) FROM file_backup_history WHERE file_path = %s\",\n        (file_path,)\n    )\n    result = cursor.fetchone()\n    current_max = result[0] if result and result[0] else 0\n    return current_max + 1\n\n\ndef get_last_hash(cursor, file_path: str) -> Optional[str]:\n    \"\"\"Holt den Hash der letzten Version\"\"\"\n    cursor.execute(\n        \"SELECT content_hash FROM file_backup_history WHERE file_path = %s ORDER BY version DESC LIMIT 1\",\n        (file_path,)\n    )\n    result = cursor.fetchone()\n    return result[0] if result else None\n\n\ndef create_backup(file_path: str, tool_name: str) -> Optional[Dict[str, Any]]:\n    \"\"\"\n    Erstellt ein Backup der Datei in file_backup_history\n\n    Returns:\n        Dict mit Backup-Info oder None bei Fehler\/Skip\n    \"\"\"\n    try:\n        # Dateiinhalt lesen\n        with open(file_path, 'r', encoding='utf-8', errors='replace') as f:\n            content = f.read()\n\n        file_size = len(content.encode('utf-8'))\n        content_hash = calculate_hash(content)\n\n        # Datenbankverbindung\n        connection = pymysql.connect(**DB_CONFIG)\n\n        with connection.cursor() as cursor:\n            # Prüfen ob sich die Datei geändert hat\n            last_hash = get_last_hash(cursor, file_path)\n            if last_hash == content_hash:\n                # Keine Änderung seit letztem Backup\n                return {'skipped': True, 'reason': 'unchanged'}\n\n            # Nächste Version ermitteln\n            version = get_next_version(cursor, file_path)\n\n            # Backup erstellen\n            reason = f\"Claude Code Pre-Hook Backup vor {tool_name}-Operation\"\n\n            sql = \"\"\"\n                INSERT INTO file_backup_history (\n                    file_path, file_content, content_hash, file_size,\n                    version, change_type, changed_by, reason\n                ) VALUES (\n                    %s, %s, %s, %s, %s, %s, %s, %s\n                )\n            \"\"\"\n\n            cursor.execute(sql, (\n                file_path,\n                content,\n                content_hash,\n                file_size,\n                version,\n                'modified',\n                'claude-code-hook',\n                reason\n            ))\n\n            connection.commit()\n            backup_id = cursor.lastrowid\n\n            return {\n                'success': True,\n                'id': backup_id,\n                'version': version,\n                'file_path': file_path,\n                'file_size': file_size,\n                'hash': content_hash[:16] + '...'\n            }\n\n    except Exception as e:\n        print(f\"Backup error: {e}\", file=sys.stderr)\n        return None\n\n    finally:\n        if 'connection' in locals():\n            connection.close()\n\n\n# =============================================================================\n# MAIN HOOK LOGIC\n# =============================================================================\n\ndef main():\n    \"\"\"Hauptfunktion des Hook-Skripts\"\"\"\n    try:\n        # JSON von Claude Code einlesen\n        input_data = json.load(sys.stdin)\n\n        event_name = input_data.get('hook_event_name', '')\n        tool_name = input_data.get('tool_name', '')\n\n        # Nur PreToolUse für Edit\/Write verarbeiten\n        if event_name != 'PreToolUse':\n            print(json.dumps({\"continue\": True}))\n            sys.exit(0)\n\n        if tool_name not in ['Edit', 'Write']:\n            print(json.dumps({\"continue\": True}))\n            sys.exit(0)\n\n        # Dateipfad aus tool_input extrahieren\n        tool_input = input_data.get('tool_input', {})\n        file_path = tool_input.get('file_path', '')\n\n        if not file_path:\n            print(json.dumps({\"continue\": True}))\n            sys.exit(0)\n\n        # Soll diese Datei gesichert werden?\n        if not should_backup(file_path):\n            print(json.dumps({\"continue\": True}))\n            sys.exit(0)\n\n        # Backup erstellen\n        result = create_backup(file_path, tool_name)\n\n        if result:\n            if result.get('skipped'):\n                print(f\"[Backup] Skipped (unchanged): {file_path}\", file=sys.stderr)\n            else:\n                print(f\"[Backup] Created v{result['version']}: {file_path}\", file=sys.stderr)\n\n        # Immer fortfahren (Hook blockiert nicht)\n        print(json.dumps({\"continue\": True}))\n        sys.exit(0)\n\n    except json.JSONDecodeError as e:\n        print(f\"Invalid JSON input: {e}\", file=sys.stderr)\n        print(json.dumps({\"continue\": True}))\n        sys.exit(0)\n\n    except Exception as e:\n        print(f\"Hook execution error: {e}\", file=sys.stderr)\n        print(json.dumps({\"continue\": True}))\n        sys.exit(0)\n\n\nif __name__ == '__main__':\n    main()\n",
            "numLines": 267,
            "startLine": 1,
            "totalLines": 267
        }
    }
}