Upload 5 files

utils/api_handler.py

@@ -0,0 +1,183 @@
+"""
+utils/api_handler.py - API-Kommunikationsmodul für den Dr. Franz Psychochatbot
+
+Dieses Modul verwaltet die Kommunikation mit der HuggingFace Inference API:
+- Senden von Anfragen
+- Verarbeiten von Antworten
+- Fehlerbehandlung und Wiederholungslogik
+"""
+
+import requests
+import time
+import random
+import json
+from typing import Dict, Any, Optional, List, Union
+
+# Importieren der Konfiguration
+import config
+
+class ApiHandler:
+    """Klasse zur Verwaltung der API-Kommunikation"""
+    
+    def __init__(self, api_token: str = config.API_TOKEN, model_id: str = config.MODEL_ID):
+        """
+        Initialisiert den API-Handler
+        
+        Args:
+            api_token: Der API-Token für HuggingFace
+            model_id: Die ID des zu verwendenden Modells
+        """
+        self.api_token = api_token
+        self.model_id = model_id
+        self.api_url = f"{config.API_BASE_URL}{model_id}"
+        self.headers = {"Authorization": f"Bearer {api_token}"}
+        self.max_retries = 3
+        self.retry_delay = 2  # Sekunden
+    
+    def generate_text(self, prompt: str, params: Optional[Dict[str, Any]] = None) -> str:
+        """
+        Generiert Text über die HuggingFace Inference API
+        
+        Args:
+            prompt: Der Prompt für die Textgenerierung
+            params: Optional, Parameter für die Textgenerierung
+            
+        Returns:
+            Der generierte Text
+        """
+        # Standardparameter verwenden, wenn keine angegeben sind
+        if params is None:
+            params = {
+                "max_new_tokens": config.MAX_NEW_TOKENS,
+                "do_sample": True,
+                "temperature": config.TEMPERATURE,
+                "top_p": config.TOP_P,
+                "top_k": config.TOP_K,
+                "repetition_penalty": config.REPETITION_PENALTY
+            }
+        
+        payload = {
+            "inputs": prompt,
+            "parameters": params
+        }
+        
+        return self._send_api_request(payload)
+    
+    def _send_api_request(self, payload: Dict[str, Any]) -> str:
+        """
+        Sendet eine Anfrage an die API mit Wiederholungslogik
+        
+        Args:
+            payload: Die Anfragedaten
+            
+        Returns:
+            Die generierte Antwort
+        """
+        for attempt in range(self.max_retries):
+            try:
+                response = requests.post(
+                    self.api_url, 
+                    headers=self.headers, 
+                    json=payload,
+                    timeout=config.API_TIMEOUT
+                )
+                
+                # Prüfen auf erfolgreiche Antwort
+                if response.status_code == 200:
+                    return self._process_successful_response(response.json())
+                
+                # Behandlung von Fehlercodes
+                if response.status_code == 401:
+                    if config.DEBUG_MODE:
+                        print("API-Fehler: Ungültiger API-Token")
+                    return self._get_fallback_response("authentication_error")
+                
+                if response.status_code == 503 or response.status_code == 429:
+                    # Server überlastet oder Rate-Limit erreicht, erneuter Versuch
+                    if config.DEBUG_MODE:
+                        print(f"API-Fehler: Server überlastet (Versuch {attempt+1}/{self.max_retries})")
+                    time.sleep(self.retry_delay * (attempt + 1))  # Exponentielles Backoff
+                    continue
+                
+                # Andere Fehler
+                if config.DEBUG_MODE:
+                    print(f"API-Fehler: Statuscode {response.status_code}")
+                return self._get_fallback_response("general_error")
+                
+            except requests.exceptions.Timeout:
+                if config.DEBUG_MODE:
+                    print(f"API-Timeout (Versuch {attempt+1}/{self.max_retries})")
+                if attempt < self.max_retries - 1:
+                    time.sleep(self.retry_delay * (attempt + 1))
+                else:
+                    return self._get_fallback_response("timeout")
+                    
+            except Exception as e:
+                if config.DEBUG_MODE:
+                    print(f"API-Anfragefehler: {e}")
+                return self._get_fallback_response("general_error")
+        
+        # Alle Versuche fehlgeschlagen
+        return self._get_fallback_response("max_retries")
+    
+    def _process_successful_response(self, response_data: Union[List[Dict[str, Any]], Dict[str, Any]]) -> str:
+        """
+        Verarbeitet eine erfolgreiche API-Antwort
+        
+        Args:
+            response_data: Die Antwortdaten von der API
+            
+        Returns:
+            Der extrahierte generierte Text
+        """
+        try:
+            if isinstance(response_data, list) and len(response_data) > 0 and "generated_text" in response_data[0]:
+                full_response = response_data[0]["generated_text"]
+                # Extrahieren der Antwort nach "Dr. Franz:"
+                if "Dr. Franz:" in full_response:
+                    reply = full_response.split("Dr. Franz:")[-1].strip()
+                    return reply
+                else:
+                    return full_response
+            elif isinstance(response_data, dict) and "generated_text" in response_data:
+                return response_data["generated_text"]
+            else:
+                if config.DEBUG_MODE:
+                    print(f"Unerwartetes API-Antwortformat: {response_data}")
+                return self._get_fallback_response("unexpected_format")
+        except Exception as e:
+            if config.DEBUG_MODE:
+                print(f"Fehler bei der Verarbeitung der API-Antwort: {e}")
+            return self._get_fallback_response("processing_error")
+    
+    def _get_fallback_response(self, error_type: str) -> str:
+        """
+        Gibt eine Fallback-Antwort basierend auf dem Fehlertyp zurück
+        
+        Args:
+            error_type: Der Typ des aufgetretenen Fehlers
+            
+        Returns:
+            Eine passende Fallback-Antwort
+        """
+        # Spezifische Antworten je nach Fehlertyp
+        error_responses = {
+            "authentication_error": [
+                "Es scheint ein Problem mit meiner Verbindung zu geben. Erzählen Sie mir mehr über Ihre Gedanken.",
+                "Ich muss kurz nachdenken. Was meinen Sie genau mit Ihrer letzten Aussage?"
+            ],
+            "timeout": [
+                "Ihre Aussage ist komplex und bedarf tieferer Analyse. Könnten Sie Ihren Gedanken weiter ausführen?",
+                "Interessant, dass Sie das so formulieren. Ich brauche einen Moment, um das zu verarbeiten."
+            ],
+            "max_retries": [
+                "Ihre Gedankengänge sind bemerkenswert verschlungen. Lassen Sie uns einen Schritt zurückgehen.",
+                "Ich spüre Widerstand in unserer Kommunikation. Vielleicht sollten wir einen anderen Ansatz versuchen."
+            ]
+        }
+        
+        # Spezifische Antwort für den Fehlertyp oder allgemeine Fallback-Antwort
+        if error_type in error_responses:
+            return random.choice(error_responses[error_type])
+        else:
+            return random.choice(config.FALLBACK_RESPONSES)

utils/constants.py

@@ -0,0 +1,21 @@
+# Configuration constants
+MAX_RESPONSE_LENGTH = 2000
+DEFAULT_PROMPT = "Dr. Franz Psychochatbot:"
+
+# Response types
+RESPONSE_TYPES = {
+    "NORMAL": "normal",
+    "ANALYSIS": "analysis",
+    "ADVICE": "advice",
+    "EMOTIONAL": "emotional"
+}
+
+# Emotion categories
+EMOTION_CATEGORIES = [
+    "angry",
+    "sad",
+    "happy",
+    "neutral",
+    "confused",
+    "excited"
+]

utils/helpers.py

@@ -0,0 +1,15 @@
+import os
+from typing import Dict, Any
+
+def load_config() -> Dict[str, Any]:
+    """Load configuration from config.py"""
+    from config import CONFIG
+    return CONFIG
+
+def get_asset_path(asset_name: str) -> str:
+    """Get the full path to an asset"""
+    return os.path.join(os.path.dirname(__file__), "../assets", asset_name)
+
+def format_response(text: str) -> str:
+    """Format the bot's response text"""
+    return text.strip().capitalize()

utils/logger.py

@@ -0,0 +1,143 @@
+"""
+utils/logger.py - Logging-Modul für den Dr. Franz Psychochatbot
+
+Dieses Modul stellt Logging-Funktionalitäten bereit:
+- Protokollierung von Aktivitäten
+- Unterstützung beim Debugging
+- Überwachung der Performance
+"""
+
+import logging
+import os
+import time
+from datetime import datetime
+from typing import Optional, Dict, Any
+
+# Importieren der Konfiguration
+import config
+
+class Logger:
+    """Klasse zur Verwaltung des Loggings"""
+    
+    def __init__(self, log_level: str = config.LOG_LEVEL):
+        """
+        Initialisiert den Logger
+        
+        Args:
+            log_level: Das Log-Level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
+        """
+        self.log_level = log_level
+        self.logger = self._setup_logger()
+        self.start_time = time.time()
+        self.interaction_count = 0
+        
+    def _setup_logger(self) -> logging.Logger:
+        """
+        Richtet den Logger ein
+        
+        Returns:
+            Konfigurierter Logger
+        """
+        # Log-Level aus String in Logging-Konstante umwandeln
+        numeric_level = getattr(logging, self.log_level.upper(), logging.INFO)
+        
+        # Logger konfigurieren
+        logger = logging.getLogger("psychobot")
+        logger.setLevel(numeric_level)
+        
+        # Handler für Konsolenausgabe
+        console_handler = logging.StreamHandler()
+        console_handler.setLevel(numeric_level)
+        
+        # Formatter für lesbares Format
+        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+        console_handler.setFormatter(formatter)
+        
+        # Handler hinzufügen, falls noch nicht vorhanden
+        if not logger.handlers:
+            logger.addHandler(console_handler)
+        
+        return logger
+    
+    def log_startup(self) -> None:
+        """Protokolliert den Start der Anwendung"""
+        self.logger.info(f"=== {config.CHATBOT_TITLE} gestartet ===")
+        self.logger.info(f"Verwende Modell: {config.MODEL_ID}")
+        self.logger.info(f"Debug-Modus: {config.DEBUG_MODE}")
+        self.logger.info(f"API-Token vorhanden: {bool(config.API_TOKEN)}")
+    
+    def log_user_input(self, user_input: str) -> None:
+        """
+        Protokolliert eine Nutzereingabe
+        
+        Args:
+            user_input: Die Eingabe des Nutzers
+        """
+        self.interaction_count += 1
+        self.logger.info(f"Nutzeranfrage #{self.interaction_count}: {user_input[:50]}...")
+    
+    def log_bot_response(self, response: str, generation_time: float) -> None:
+        """
+        Protokolliert eine Bot-Antwort
+        
+        Args:
+            response: Die Antwort des Bots
+            generation_time: Die Zeit zur Generierung der Antwort in Sekunden
+        """
+        self.logger.info(f"Bot-Antwort #{self.interaction_count}: {response[:50]}... (Generiert in {generation_time:.2f}s)")
+    
+    def log_api_request(self, model_id: str, prompt_length: int) -> None:
+        """
+        Protokolliert eine API-Anfrage
+        
+        Args:
+            model_id: Die ID des verwendeten Modells
+            prompt_length: Die Länge des Prompts in Zeichen
+        """
+        self.logger.debug(f"API-Anfrage an {model_id} mit Prompt-Länge: {prompt_length} Zeichen")
+    
+    def log_api_response(self, status_code: int, response_time: float) -> None:
+        """
+        Protokolliert eine API-Antwort
+        
+        Args:
+            status_code: Der HTTP-Statuscode
+            response_time: Die Antwortzeit in Sekunden
+        """
+        self.logger.debug(f"API-Antwort erhalten: Status {status_code}, Zeit: {response_time:.2f}s")
+    
+    def log_error(self, error_type: str, error_message: str, details: Optional[Dict[str, Any]] = None) -> None:
+        """
+        Protokolliert einen Fehler
+        
+        Args:
+            error_type: Der Typ des Fehlers
+            error_message: Die Fehlermeldung
+            details: Optional, zusätzliche Details zum Fehler
+        """
+        self.logger.error(f"FEHLER - {error_type}: {error_message}")
+        if details and config.DEBUG_MODE:
+            self.logger.error(f"Details: {details}")
+    
+    def log_performance_stats(self) -> Dict[str, Any]:
+        """
+        Protokolliert Performance-Statistiken
+        
+        Returns:
+            Dictionary mit Performance-Statistiken
+        """
+        uptime = time.time() - self.start_time
+        hours, remainder = divmod(uptime, 3600)
+        minutes, seconds = divmod(remainder, 60)
+        
+        stats = {
+            "uptime": f"{int(hours)}h {int(minutes)}m {int(seconds)}s",
+            "interactions": self.interaction_count,
+            "interactions_per_hour": round(self.interaction_count / (uptime / 3600), 2) if uptime > 0 else 0
+        }
+        
+        self.logger.info(f"Performance-Statistiken: Laufzeit {stats['uptime']}, "
+                        f"{stats['interactions']} Interaktionen, "
+                        f"{stats['interactions_per_hour']} Interaktionen/Stunde")
+        
+        return stats

utils/security.py

@@ -0,0 +1,97 @@
+"""
+utils/security.py - Sicherheitsmodul für den Dr. Franz Psychochatbot
+
+Dieses Modul verwaltet die Sicherheitsaspekte des Chatbots:
+- Sichere Verwaltung von API-Tokens
+- Laden von Umgebungsvariablen
+- Schutz sensibler Daten
+"""
+
+import os
+import re
+from typing import Optional, Dict, Any
+
+class Security:
+    """Klasse zur Verwaltung der Sicherheitsaspekte"""
+    
+    def __init__(self):
+        """Initialisiert die Sicherheitskomponente"""
+        self.sensitive_patterns = [
+            r'\b(?:\d[ -]*?){13,16}\b',  # Kreditkartennummern
+            r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',  # E-Mail-Adressen
+            r'\b(?:\+\d{1,3}[ -]?)?(?:\(?\d{2,4}\)?[ -]?)?[\d -]{7,10}\b',  # Telefonnummern
+            r'\bhf_\w+\b',  # HuggingFace API-Tokens
+            r'\bsk-\w+\b'   # OpenAI API-Tokens
+        ]
+    
+    def get_api_token(self, env_var_name: str = "HF_API_TOKEN", default: str = "") -> str:
+        """
+        Holt den API-Token aus den Umgebungsvariablen
+        
+        Args:
+            env_var_name: Name der Umgebungsvariable
+            default: Standardwert, falls die Variable nicht existiert
+            
+        Returns:
+            Der API-Token oder der Standardwert
+        """
+        return os.environ.get(env_var_name, default)
+    
+    def sanitize_user_input(self, text: str) -> str:
+        """
+        Entfernt potenziell sensible Informationen aus der Nutzereingabe
+        
+        Args:
+            text: Die zu bereinigende Nutzereingabe
+            
+        Returns:
+            Die bereinigte Nutzereingabe
+        """
+        sanitized_text = text
+        
+        # Ersetzen sensibler Muster
+        for pattern in self.sensitive_patterns:
+            sanitized_text = re.sub(pattern, "[ENTFERNT]", sanitized_text)
+        
+        return sanitized_text
+    
+    def validate_api_response(self, response: Any) -> bool:
+        """
+        Überprüft, ob eine API-Antwort sicher ist
+        
+        Args:
+            response: Die zu überprüfende API-Antwort
+            
+        Returns:
+            True, wenn die Antwort sicher ist, sonst False
+        """
+        # Konvertieren der Antwort in einen String für die Überprüfung
+        if not isinstance(response, str):
+            try:
+                response_str = str(response)
+            except:
+                return False
+        else:
+            response_str = response
+        
+        # Überprüfen auf sensible Informationen
+        for pattern in self.sensitive_patterns:
+            if re.search(pattern, response_str):
+                return False
+        
+        return True
+    
+    def is_valid_api_token(self, token: str) -> bool:
+        """
+        Überprüft, ob ein API-Token gültig erscheint
+        
+        Args:
+            token: Der zu überprüfende API-Token
+            
+        Returns:
+            True, wenn der Token gültig erscheint, sonst False
+        """
+        # HuggingFace-Tokens beginnen mit "hf_"
+        if token.startswith("hf_") and len(token) > 10:
+            return True
+        return False