Spaces:
Running
Running
| 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() |