Spaces:
Running
Running
File size: 3,968 Bytes
ad3ba0d f4fc950 ad3ba0d f4fc950 ad3ba0d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 |
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() |