log_to_db.py

• Pfad: /var/www/tools/ki-protokoll/claude-hook/log_to_db.py
• Namespace: claude-hook
• Zeilen: 132 | Größe: 3,883 Bytes
• Geändert: 2025-12-25 16:58:51 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 75

• Dependencies: 0 (25%)
• LOC: 100 (20%)
• Methods: 100 (20%)
• Secrets: 100 (15%)
• Classes: 100 (10%)
• Magic Numbers: 100 (10%)

Issues 1

Dependencies 21

• use json
• use sys
• use typing.Dict
• use typing.Any
• use typing.Optional
• use log_parser.parse_event_data
• use log_parser.sanitize_data
• use log_parser.get_conversation_pairs_from_session
• use log_formatter.format_request_to_json
• use log_formatter.format_response_to_json
• use log_formatter.create_response_data
• use log_formatter.calculate_tokens
• use log_formatter.get_client_metadata
• use log_formatter.get_model_name
• use log_formatter.format_tool_response
• use log_formatter.should_suppress_output
• use log_storage.find_matching_request
• use log_storage.save_pending_request
• use log_storage.insert_log_entry
• use log_storage.update_request_with_response
• use log_storage.close_pending_user_prompts

Funktionen 2

• log_to_database() Zeile 36
• main() Zeile 110

Code

#!/usr/bin/env python3
"""
Claude Code Hook Logger v3 - Modularisiert
Orchestriert Parsing, Formatting und Storage von Hook-Events
"""

import json
import sys
from typing import Dict, Any, Optional

# Importiere modularisierte Komponenten
from log_parser import (
    parse_event_data,
    sanitize_data,
    get_conversation_pairs_from_session
)
from log_formatter import (
    format_request_to_json,
    format_response_to_json,
    create_response_data,
    calculate_tokens,
    get_client_metadata,
    get_model_name,
    format_tool_response,
    should_suppress_output
)
from log_storage import (
    find_matching_request,
    save_pending_request,
    insert_log_entry,
    update_request_with_response,
    close_pending_user_prompts
)


def log_to_database(data: Dict[str, Any]) -> Optional[int]:
    """
    Hauptfunktion: Orchestriert das Logging eines Events

    Args:
        data: Event-Daten vom Hook

    Returns:
        Database ID des Log-Eintrags oder None bei Fehler
    """
    try:
        event_name = data.get('hook_event_name', 'Unknown')
        session_id = data.get('session_id', '')

        # Prüfe auf matching Request für Response-Events (PostToolUse)
        if event_name == 'PostToolUse':
            matching_request_id = find_matching_request(data)
            if matching_request_id:
                # Update existing request mit response
                tool_response = sanitize_data(data.get('tool_response', {}))
                response_str = format_tool_response(tool_response)
                tokens = calculate_tokens('', response_str)

                update_request_with_response(
                    matching_request_id,
                    response_str,
                    tokens['tokens_output']
                )
                return matching_request_id

        # Bei Stop-Events: Schließe alle pending UserPrompts
        if event_name in ['Stop', 'SubagentStop'] and session_id:
            pairs = get_conversation_pairs_from_session(session_id)
            close_pending_user_prompts(session_id, pairs)

        # Parse Event-Daten
        request_data = parse_event_data(data)

        # Erstelle Response-Daten (falls direkt verfügbar)
        response_data = create_response_data(event_name, data)

        # Formatiere zu JSON
        request_str = format_request_to_json(request_data)
        response_str = format_response_to_json(response_data)

        # Berechne Tokens
        tokens = calculate_tokens(request_str, response_str)

        # Hole Client-Metadaten
        client_meta = get_client_metadata()
        model_name = get_model_name()

        # Insert in DB
        db_id = insert_log_entry(
            request_str=request_str,
            client_ip=client_meta['client_ip'],
            client_name=client_meta['client_name'],
            response_str=response_str,
            tokens_input=tokens['tokens_input'],
            tokens_output=tokens['tokens_output'],
            model_name=model_name
        )

        # Speichere pending Requests für spätere Response-Zuordnung
        if db_id and not response_str:
            save_pending_request(data, db_id)

        return db_id

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


def main():
    """Hauptfunktion des Hook-Skripts"""
    try:
        input_data = json.load(sys.stdin)
        log_to_database(input_data)

        event_name = input_data.get('hook_event_name', '')
        if should_suppress_output(event_name):
            print(json.dumps({"suppressOutput": True}))

        sys.exit(0)

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

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


if __name__ == '__main__':
    main()