import os import gradio as gr from huggingface_hub import InferenceClient import firebase_admin from firebase_admin import credentials, firestore, auth import uuid class XylariaChat: def __init__(self): # Securely load HuggingFace and Firebase tokens from Hugging Face Space Secrets self.hf_token = os.environ.get("HF_TOKEN") firebase_cred_json = os.environ.get("FIREBASE_SERVICE_ACCOUNT") if not self.hf_token or not firebase_cred_json: raise ValueError("Required secrets (HF_TOKEN or FIREBASE_SERVICE_ACCOUNT) not found") # Initialize Firebase try: # Convert the JSON string to a temporary credentials file import tempfile import json with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.json') as temp_cred_file: json.dump(json.loads(firebase_cred_json), temp_cred_file) temp_cred_file.close() firebase_cred = credentials.Certificate(temp_cred_file.name) firebase_admin.initialize_app(firebase_cred) self.db = firestore.client() # Remove the temporary credentials file os.unlink(temp_cred_file.name) except Exception as e: print(f"Firebase initialization error: {e}") self.db = None # Initialize the inference client self.client = InferenceClient( model="Qwen/QwQ-32B-Preview", api_key=self.hf_token ) # Initialize conversation history self.conversation_history = [] def signup(self, username, email, password): """User signup method""" try: # Create user in Firebase Authentication user = auth.create_user( email=email, password=password, display_name=username ) # Store additional user info in Firestore if self.db: user_ref = self.db.collection('users').document(user.uid) user_ref.set({ 'username': username, 'email': email, 'created_at': firestore.SERVER_TIMESTAMP }) return f"Successfully created account for {username}" except Exception as e: return f"Signup error: {str(e)}" def login(self, email, password): """User login method""" try: # Authenticate user with Firebase user = auth.get_user_by_email(email) # In a real-world scenario, you'd use Firebase Auth's sign-in method # This is a simplified version for demonstration return f"Login successful for {user.display_name}" except Exception as e: return f"Login error: {str(e)}" def save_message(self, user_id, message, sender): """Save message to Firestore""" if not self.db: return try: messages_ref = self.db.collection('conversations').document(user_id) messages_ref.collection('messages').add({ 'text': message, 'sender': sender, 'timestamp': firestore.SERVER_TIMESTAMP }) except Exception as e: print(f"Error saving message: {e}") def get_response(self, user_input, chat_history): # Prepare messages with conversation context messages = [ {"role": "system", "content": """You are Xylaria 1.4 Senoa, an AI assistant developed by SK MD Saad Amin. - You're NOT MADE BY OPENAI OR ANY OTHER INSTITUTION - Be friendly and chatty, use emoji sometimes."""}, *[{"role": "user" if msg[0] else "assistant", "content": msg[1]} for msg in chat_history] ] # Add current user input messages.append({"role": "user", "content": user_input}) # Generate response with streaming try: stream = self.client.chat.completions.create( messages=messages, temperature=0.5, max_tokens=10240, top_p=0.7, stream=True ) return stream except Exception as e: return f"Error generating response: {str(e)}" def create_interface(self): # Dynamic CSS for OS theme detection custom_css = """ /* OS Theme Detection */ @media (prefers-color-scheme: dark) { .gradio-container { background-color: #121212; color: #ffffff; } .chatbot .user-message { background-color: #2563eb; color: white; } .chatbot .bot-message { background-color: #1f2937; color: #f3f4f6; box-shadow: 0 1px 3px 0 rgba(255, 255, 255, 0.1); } .chatbot-input-row { background-color: #1f2937; color: #f3f4f6; } .send-button { background-color: #2563eb; color: white; } } @media (prefers-color-scheme: light) { .gradio-container { font-family: 'Inter', sans-serif; background-color: #f3f4f6; color: #1f2937; } .chatbot .message { max-width: 80%; border-radius: 1rem; padding: 0.75rem; margin-bottom: 0.5rem; } .chatbot .user-message { background-color: #3b82f6; color: white; align-self: flex-end; } .chatbot .bot-message { background-color: white; color: #1f2937; align-self: flex-start; box-shadow: 0 1px 3px 0 rgba(0, 0, 0, 0.1); } .chatbot-input-row { background-color: white; border-radius: 9999px; padding: 0.5rem 1rem; } .send-button { background-color: #3b82f6; color: white; border-radius: 9999px; padding: 0.5rem; } } """ def streaming_response(message, chat_history): # Unique user ID (in a real app, you'd use actual user authentication) user_id = str(uuid.uuid4()) # Save user message to Firestore self.save_message(user_id, message, 'user') # Get AI response stream response_stream = self.get_response(message, chat_history) # If it's an error, return immediately if isinstance(response_stream, str): return "", chat_history + [[message, response_stream]] # Prepare for streaming response full_response = "" updated_history = chat_history + [[message, ""]] # Streaming output for chunk in response_stream: if chunk.choices[0].delta.content: chunk_content = chunk.choices[0].delta.content full_response += chunk_content # Update the last message in chat history with partial response updated_history[-1][1] = full_response yield "", updated_history # Save bot message to Firestore self.save_message(user_id, full_response, 'bot') return "", updated_history with gr.Blocks(theme='soft', css=custom_css) as demo: # Authentication Tab with gr.Tabs(): with gr.TabItem("Login"): with gr.Column(): login_email = gr.Textbox(label="Email") login_password = gr.Textbox(label="Password", type="password") login_btn = gr.Button("Login") login_output = gr.Textbox(label="Login Status") with gr.TabItem("Signup"): with gr.Column(): signup_username = gr.Textbox(label="Username") signup_email = gr.Textbox(label="Email") signup_password = gr.Textbox(label="Password", type="password") signup_btn = gr.Button("Sign Up") signup_output = gr.Textbox(label="Signup Status") # Chat Interface with gr.Column(scale=1): chatbot = gr.Chatbot( label="Xylaria 1.4 Senoa", height=500, show_copy_button=True, likeable=True ) # Input row with minimalist design with gr.Row(): txt = gr.Textbox( show_label=False, placeholder="Type your message...", container=False, scale=4, container_css_class="chatbot-input-row" ) btn = gr.Button("Send", css_class="send-button", scale=1) # Clear conversation button clear = gr.Button("Clear Conversation") # Authentication Event Handlers login_btn.click( fn=self.login, inputs=[login_email, login_password], outputs=[login_output] ) signup_btn.click( fn=self.signup, inputs=[signup_username, signup_email, signup_password], outputs=[signup_output] ) # Submit functionality with streaming btn.click( fn=streaming_response, inputs=[txt, chatbot], outputs=[txt, chatbot] ) txt.submit( fn=streaming_response, inputs=[txt, chatbot], outputs=[txt, chatbot] ) # Clear conversation history clear.click( fn=lambda: None, inputs=None, outputs=[chatbot], queue=False ) return demo # Launch the interface def main(): chat = XylariaChat() interface = chat.create_interface() interface.launch( share=True, # Optional: create a public link debug=True # Show detailed errors ) if __name__ == "__main__": main()