Spaces:
Sleeping
Sleeping
| # Database interface for conversation and message storage using HF buckets | |
| # JSONL format for scalability and future database migration | |
| import json | |
| from datetime import datetime | |
| from typing import Optional, List, Dict | |
| from huggingface_hub import HfApi | |
| from api.config import settings | |
| class ConversationDatabase: | |
| """Manages conversation and message storage in HF bucket using JSONL format""" | |
| def __init__(self): | |
| self.api = HfApi() | |
| self.repo_id = settings.bucket.repo_id | |
| self.repo_type = settings.bucket.repo_type | |
| self.conversations_base = settings.bucket.conversations_base | |
| def add_message_to_conversation( | |
| self, user_id: str, conversation_id: str, | |
| role: str, content: str, | |
| language: str = "en", | |
| symptoms: List[str] = [], | |
| quick_replies: List[str] = [] | |
| ) -> bool: | |
| """Add message to conversation JSONL file""" | |
| try: | |
| message = { | |
| "timestamp": datetime.utcnow().isoformat(), | |
| "role": role, # "user" or "assistant" | |
| "content": content, | |
| "language": language, | |
| "symptoms": symptoms or [], | |
| "quick_replies": quick_replies or [] | |
| } | |
| # Read existing messages | |
| messages_file = f"{self.conversations_base}{user_id}/{conversation_id}_messages.jsonl" | |
| existing_content = "" | |
| try: | |
| existing_content = self.api.read_file_content( | |
| repo_id=self.repo_id, | |
| repo_type=self.repo_type, | |
| filename=messages_file | |
| ) | |
| except Exception: | |
| # File doesn't exist yet | |
| pass | |
| # Append new message | |
| new_content = existing_content | |
| if new_content and not new_content.endswith('\n'): | |
| new_content += '\n' | |
| new_content += json.dumps(message) | |
| # Upload updated file | |
| self.api.upload_file( | |
| path_or_fileobj=new_content.encode('utf-8'), | |
| path_in_repo=messages_file, | |
| repo_id=self.repo_id, | |
| repo_type=self.repo_type, | |
| commit_message=f"Add message to conversation {conversation_id}" | |
| ) | |
| # Update conversation metadata with last_updated timestamp | |
| self._update_conversation_timestamp(user_id, conversation_id) | |
| return True | |
| except Exception as e: | |
| print(f"Error adding message: {e}") | |
| return False | |
| def get_conversation_messages(self, user_id: str, conversation_id: str) -> List[Dict]: | |
| """Read all messages from conversation JSONL file""" | |
| try: | |
| messages_file = f"{self.conversations_base}{user_id}/{conversation_id}_messages.jsonl" | |
| content = self.api.read_file_content( | |
| repo_id=self.repo_id, | |
| repo_type=self.repo_type, | |
| filename=messages_file | |
| ) | |
| messages = [] | |
| for line in content.strip().split('\n'): | |
| if line: | |
| messages.append(json.loads(line)) | |
| return messages | |
| except Exception: | |
| # File doesn't exist or error reading | |
| return [] | |
| def save_diagnosis_to_conversation(self, user_id: str, conversation_id: str, diagnosis: Dict, recommendations: str = "") -> bool: | |
| """Save diagnosis results to conversation diagnosis file""" | |
| try: | |
| diagnosis_file = f"{self.conversations_base}{user_id}/{conversation_id}_diagnosis.json" | |
| diagnosis_data = { | |
| "timestamp": datetime.utcnow().isoformat(), | |
| "diagnosis": diagnosis, | |
| "recommendations": recommendations | |
| } | |
| self.api.upload_file( | |
| path_or_fileobj=json.dumps(diagnosis_data, indent=2).encode('utf-8'), | |
| path_in_repo=diagnosis_file, | |
| repo_id=self.repo_id, | |
| repo_type=self.repo_type, | |
| commit_message=f"Add diagnosis to conversation {conversation_id}" | |
| ) | |
| # Update conversation metadata | |
| self._update_conversation_timestamp(user_id, conversation_id) | |
| return True | |
| except Exception as e: | |
| print(f"Error saving diagnosis: {e}") | |
| return False | |
| def get_diagnosis_from_conversation(self, user_id: str, conversation_id: str) -> Optional[Dict]: | |
| """Read diagnosis results from conversation diagnosis file""" | |
| try: | |
| diagnosis_file = f"{self.conversations_base}{user_id}/{conversation_id}_diagnosis.json" | |
| content = self.api.read_file_content( | |
| repo_id=self.repo_id, | |
| repo_type=self.repo_type, | |
| filename=diagnosis_file | |
| ) | |
| return json.loads(content) | |
| except Exception: | |
| # Diagnosis file doesn't exist | |
| return None | |
| def _update_conversation_timestamp(self, user_id: str, conversation_id: str): | |
| """Update the last_updated timestamp in conversation metadata""" | |
| try: | |
| metadata_file = f"{self.conversations_base}{user_id}/{conversation_id}_metadata.json" | |
| metadata = {} | |
| # Try to read existing metadata | |
| try: | |
| content = self.api.read_file_content( | |
| repo_id=self.repo_id, | |
| repo_type=self.repo_type, | |
| filename=metadata_file | |
| ) | |
| metadata = json.loads(content) | |
| except Exception: | |
| pass | |
| # Update timestamp | |
| metadata["last_updated"] = datetime.utcnow().isoformat() | |
| self.api.upload_file( | |
| path_or_fileobj=json.dumps(metadata, indent=2).encode('utf-8'), | |
| path_in_repo=metadata_file, | |
| repo_id=self.repo_id, | |
| repo_type=self.repo_type, | |
| commit_message=f"Update conversation {conversation_id} timestamp" | |
| ) | |
| except Exception as e: | |
| print(f"Error updating conversation timestamp: {e}") | |
| # Singleton instance | |
| conversation_db = ConversationDatabase() | |