import os import json import logging from datetime import datetime # Setup logging logger = logging.getLogger(__name__) # Supabase integration try: from supabase import create_client, Client SUPABASE_URL = os.getenv("SUPABASE_URL") SUPABASE_KEY = os.getenv("SUPABASE_KEY") if SUPABASE_URL and SUPABASE_KEY: supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY) else: supabase = None logger.warning("Supabase URL or Key not set. Supabase feedback saving will be disabled.") except ImportError: supabase = None logger.warning("supabase-py not installed. Supabase feedback saving will be disabled.") class FeedbackManager: """Handles the storage and retrieval of user feedback for chatbot responses.""" def __init__(self, feedback_dir="feedback"): """Initialize the feedback manager with a directory to store feedback files.""" self.feedback_dir = feedback_dir # Create feedback directory if it doesn't exist if not os.path.exists(feedback_dir): os.makedirs(feedback_dir) logger.info(f"Created feedback directory: {feedback_dir}") def save_feedback(self, feedback_data): """ Save feedback to a JSON file. Args: feedback_data (dict): Dictionary containing feedback information: - rating: "positive" or "negative" - category: (optional) The category of negative feedback - comment: (optional) Additional comments provided by the user - query: The user's original question - response: The assistant's response that received feedback - timestamp: When the feedback was given """ # Generate a unique filename based on timestamp timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"feedback_{timestamp}.json" filepath = os.path.join(self.feedback_dir, filename) # Add timestamp to the feedback data feedback_data['timestamp'] = datetime.now().isoformat() try: with open(filepath, 'w') as f: json.dump(feedback_data, f, indent=2) logger.info(f"Feedback saved to {filepath}") return True except Exception as e: logger.error(f"Error saving feedback: {str(e)}") return False def get_all_feedback(self): """Retrieve all feedback data as a list of dictionaries.""" feedback_list = [] try: for filename in os.listdir(self.feedback_dir): if filename.endswith('.json'): filepath = os.path.join(self.feedback_dir, filename) with open(filepath, 'r') as f: feedback_list.append(json.load(f)) return feedback_list except Exception as e: logger.error(f"Error retrieving feedback: {str(e)}") return [] def save_feedback_supabase(self, feedback_data): """ Save feedback to Supabase table 'feedback'. Args: feedback_data (dict): Dictionary containing feedback information matching the table schema. Returns: bool: True if feedback was saved successfully, False otherwise """ if not supabase: logger.error("Supabase client not available. Feedback not saved to Supabase.") return False try: logger.info(f"Attempting to insert feedback: {feedback_data}") # Debug outgoing data data, count = supabase.table('feedback').insert(feedback_data).execute() logger.info(f"Feedback saved to Supabase: {data}") return True except Exception as e: logger.error(f"Error saving feedback to Supabase: {str(e)}") return False # Create a singleton instance feedback_manager = FeedbackManager()