file_backup_hook.py

Code Hygiene Score: 80

Keine Issues gefunden.

Dependencies 11

Funktionen 6

Code

#!/usr/bin/env python3
"""
File Backup Hook für Claude Code PreToolUse Events

Sichert Dateien automatisch in file_backup_history BEVOR sie von
Claude Code (Edit/Write Tools) geändert werden.

Trigger: PreToolUse (Edit, Write)
Target: ki_protokoll.file_backup_history

Version: 1.0
Erstellt: 2025-12-20
"""

import json
import os
import sys
import hashlib
import pymysql
from pathlib import Path
from typing import Dict, Any, Optional
from datetime import datetime

# .env aus Hook-Verzeichnis laden
from dotenv import load_dotenv
load_dotenv(Path(__file__).parent / '.env')

# =============================================================================
# KONFIGURATION
# =============================================================================

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', 'root'),
    'password': os.environ.get('CLAUDE_DB_PASSWORD', ''),
    'database': os.environ.get('CLAUDE_DB_NAME', 'ki_dev'),
    'charset': 'utf8mb4'
}

# Welche Verzeichnisse sollen gesichert werden?
BACKUP_DIRS = [
    '/var/www/dev.campus.systemische-tools.de/src',
    '/var/www/dev.campus.systemische-tools.de/public',
    '/var/www/dev.campus.systemische-tools.de/scripts',
    '/var/www/dev.campus.systemische-tools.de/includes',
    '/var/www/dev.campus.systemische-tools.de/Views',
    '/var/www/dev.campus.systemische-tools.de/config',
    '/var/www/dev.campus.systemische-tools.de/adm',
    '/var/www/prod.campus.systemische-tools.de/src',
    '/var/www/prod.campus.systemische-tools.de/public',
    '/var/www/prod.campus.systemische-tools.de/scripts',
    '/var/www/prod.campus.systemische-tools.de/includes',
    '/var/www/prod.campus.systemische-tools.de/Views',
    '/var/www/prod.campus.systemische-tools.de/config',
]

# Dateien die NICHT gesichert werden (Patterns)
EXCLUDE_PATTERNS = [
    '/vendor/',
    '/node_modules/',
    '/.git/',
    '/backups/',
    '/tmp/',
    '/logs/',
    '/cache/',
    '.log',
    '.cache',
    '.tmp'
]

# Maximale Dateigröße für Backup (10 MB)
MAX_FILE_SIZE = 10 * 1024 * 1024

# =============================================================================
# HILFSFUNKTIONEN
# =============================================================================

def should_backup(file_path: str) -> bool:
    """Prüft, ob die Datei gesichert werden soll"""
    if not file_path:
        return False

    # Existiert die Datei?
    if not os.path.isfile(file_path):
        return False

    # Liegt sie in einem Backup-Verzeichnis?
    in_backup_dir = any(file_path.startswith(d) for d in BACKUP_DIRS)
    if not in_backup_dir:
        return False

    # Ist sie ausgeschlossen?
    for pattern in EXCLUDE_PATTERNS:
        if pattern in file_path:
            return False

    # Dateigröße prüfen
    try:
        if os.path.getsize(file_path) > MAX_FILE_SIZE:
            return False
    except OSError:
        return False

    return True


def calculate_hash(content: str) -> str:
    """Berechnet SHA256 Hash des Inhalts"""
    return hashlib.sha256(content.encode('utf-8', errors='ignore')).hexdigest()


def get_next_version(cursor, file_path: str) -> int:
    """Ermittelt die nächste Versionsnummer für eine Datei"""
    cursor.execute(
        "SELECT MAX(version) FROM file_backup_history WHERE file_path = %s",
        (file_path,)
    )
    result = cursor.fetchone()
    current_max = result[0] if result and result[0] else 0
    return current_max + 1


def get_last_hash(cursor, file_path: str) -> Optional[str]:
    """Holt den Hash der letzten Version"""
    cursor.execute(
        "SELECT content_hash FROM file_backup_history WHERE file_path = %s ORDER BY version DESC LIMIT 1",
        (file_path,)
    )
    result = cursor.fetchone()
    return result[0] if result else None


def create_backup(file_path: str, tool_name: str) -> Optional[Dict[str, Any]]:
    """
    Erstellt ein Backup der Datei in file_backup_history

    Returns:
        Dict mit Backup-Info oder None bei Fehler/Skip
    """
    try:
        # Dateiinhalt lesen
        with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
            content = f.read()

        file_size = len(content.encode('utf-8'))
        content_hash = calculate_hash(content)

        # Datenbankverbindung
        connection = pymysql.connect(**DB_CONFIG)

        with connection.cursor() as cursor:
            # Prüfen ob sich die Datei geändert hat
            last_hash = get_last_hash(cursor, file_path)
            if last_hash == content_hash:
                # Keine Änderung seit letztem Backup
                return {'skipped': True, 'reason': 'unchanged'}

            # Nächste Version ermitteln
            version = get_next_version(cursor, file_path)

            # Backup erstellen
            reason = f"Claude Code Pre-Hook Backup vor {tool_name}-Operation"

            sql = """
                INSERT INTO file_backup_history (
                    file_path, file_content, content_hash, file_size,
                    version, change_type, changed_by, reason
                ) VALUES (
                    %s, %s, %s, %s, %s, %s, %s, %s
                )
            """

            cursor.execute(sql, (
                file_path,
                content,
                content_hash,
                file_size,
                version,
                'modified',
                'claude-code-hook',
                reason
            ))

            connection.commit()
            backup_id = cursor.lastrowid

            return {
                'success': True,
                'id': backup_id,
                'version': version,
                'file_path': file_path,
                'file_size': file_size,
                'hash': content_hash[:16] + '...'
            }

    except Exception as e:
        print(f"Backup error: {e}", file=sys.stderr)
        return None

    finally:
        if 'connection' in locals():
            connection.close()


# =============================================================================
# MAIN HOOK LOGIC
# =============================================================================

def main():
    """Hauptfunktion des Hook-Skripts"""
    try:
        # JSON von Claude Code einlesen
        input_data = json.load(sys.stdin)

        event_name = input_data.get('hook_event_name', '')
        tool_name = input_data.get('tool_name', '')

        # Nur PreToolUse für Edit/Write verarbeiten
        if event_name != 'PreToolUse':
            print(json.dumps({"continue": True}))
            sys.exit(0)

        if tool_name not in ['Edit', 'Write']:
            print(json.dumps({"continue": True}))
            sys.exit(0)

        # Dateipfad aus tool_input extrahieren
        tool_input = input_data.get('tool_input', {})
        file_path = tool_input.get('file_path', '')

        if not file_path:
            print(json.dumps({"continue": True}))
            sys.exit(0)

        # Soll diese Datei gesichert werden?
        if not should_backup(file_path):
            print(json.dumps({"continue": True}))
            sys.exit(0)

        # Backup erstellen
        result = create_backup(file_path, tool_name)

        if result:
            if result.get('skipped'):
                print(f"[Backup] Skipped (unchanged): {file_path}", file=sys.stderr)
            else:
                print(f"[Backup] Created v{result['version']}: {file_path}", file=sys.stderr)

        # Immer fortfahren (Hook blockiert nicht)
        print(json.dumps({"continue": True}))
        sys.exit(0)

    except json.JSONDecodeError as e:
        print(f"Invalid JSON input: {e}", file=sys.stderr)
        print(json.dumps({"continue": True}))
        sys.exit(0)

    except Exception as e:
        print(f"Hook execution error: {e}", file=sys.stderr)
        print(json.dumps({"continue": True}))
        sys.exit(0)


if __name__ == '__main__':
    main()
← Übersicht