log_formatter.py

• Pfad: /var/www/tools/ki-protokoll/claude-hook/log_formatter.py
• Namespace: claude-hook
• Zeilen: 70 | Größe: 2,184 Bytes
• Geändert: 2025-12-25 16:57:22 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 95

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

Keine Issues gefunden.

Dependencies 7

• use json
• use os
• use typing.Dict
• use typing.Any
• use typing.Optional
• use log_parser.estimate_tokens
• use log_parser.get_client_ip

Funktionen 8

• format_request_to_json() Zeile 13
• format_response_to_json() Zeile 18
• create_response_data() Zeile 25
• calculate_tokens() Zeile 36
• get_client_metadata() Zeile 49
• get_model_name() Zeile 57
• format_tool_response() Zeile 62
• should_suppress_output() Zeile 67

Code

#!/usr/bin/env python3
"""
Log Formatter Module
Verantwortlich für die Formatierung von Log-Daten zu JSON und Response-Strukturierung
"""

import json
import os
from typing import Dict, Any, Optional
from log_parser import estimate_tokens, get_client_ip


def format_request_to_json(request_data: Dict[str, Any]) -> str:
    """Konvertiert Request-Daten zu JSON-String"""
    return json.dumps(request_data, ensure_ascii=False)


def format_response_to_json(response_data: Optional[Dict[str, Any]]) -> Optional[str]:
    """Konvertiert Response-Daten zu JSON-String"""
    if response_data is None:
        return None
    return json.dumps(response_data, ensure_ascii=False)


def create_response_data(event_name: str, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """Erstellt Response-Daten basierend auf Event-Typ"""
    if event_name in ['Stop', 'SubagentStop']:
        return {'status': 'completed'}

    if event_name in ['SessionStart', 'SessionEnd']:
        return {'status': 'logged'}

    return None


def calculate_tokens(request_str: str, response_str: Optional[str] = None) -> Dict[str, int]:
    """Berechnet Token-Counts für Request und Response"""
    tokens_input = estimate_tokens(request_str)
    tokens_output = estimate_tokens(response_str) if response_str else 0
    tokens_total = tokens_input + tokens_output

    return {
        'tokens_input': tokens_input,
        'tokens_output': tokens_output,
        'tokens_total': tokens_total
    }


def get_client_metadata() -> Dict[str, str]:
    """Sammelt Client-Metadaten für Logging"""
    return {
        'client_ip': get_client_ip(),
        'client_name': os.environ.get('USER', 'unknown')
    }


def get_model_name() -> str:
    """Gibt den verwendeten Model-Namen zurück"""
    return 'claude-sonnet-4-20250514'


def format_tool_response(tool_response: Any) -> str:
    """Formatiert Tool-Response für Speicherung"""
    return json.dumps({'tool_response': tool_response}, ensure_ascii=False)


def should_suppress_output(event_name: str) -> bool:
    """Bestimmt, ob Output für dieses Event unterdrückt werden soll"""
    return event_name in ['UserPromptSubmit', 'SessionStart']