Update app.py

app.py

@@ -1,337 +1,69 @@
 import gradio as gr
 import requests
-import json
+from datetime import datetime
 import os
-import base64
-from PIL import Image
-import soundfile as sf
-import mimetypes
-import logging
-from io import BytesIO
-import tempfile
 
-# Set up logging
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
-logger = logging.getLogger(__name__)
+# GitHub image URL as chatbot icon
+BOT_ICON = "https://raw.githubusercontent.com/GeminiFan207/Secialix_Chatbot/main/IMG_0528-removebg-preview.png"
 
-# Hugging Face API configuration
-HF_API_URL = os.getenv("HF_API_URL")  # URL of your Hugging Face model endpoint
-HF_API_TOKEN = os.getenv("HF_API_TOKEN")  # Hugging Face API token
+# Load API token from Hugging Face Spaces secrets
+API_TOKEN = os.getenv("HF_API_TOKEN")  # Set this in Spaces Secrets
+if not API_TOKEN:
+    raise ValueError("HF_API_TOKEN not set in Spaces Secrets. Please configure it in the Space settings.")
+API_URL = "https://api-inference.huggingface.co/models/ZeppelinCorp/Charm_15"
+headers = {"Authorization": f"Bearer {API_TOKEN}"}
 
-# Default parameter values
-default_max_tokens = 4096
-default_temperature = 0.7
-default_top_p = 0.9
-default_presence_penalty = 0.0
-default_frequency_penalty = 0.0
-
-# Initialize MIME types
-mimetypes.init()
-
-def call_hf_endpoint(payload, api_url, api_token, params=None):
-    """Call Hugging Face Inference API with the given payload."""
-    # Set parameters from the UI inputs or use defaults
-    if params is None:
-        params = {
-            "max_tokens": default_max_tokens,
-            "temperature": default_temperature,
-            "top_p": default_top_p,
-            "presence_penalty": default_presence_penalty,
-            "frequency_penalty": default_frequency_penalty
-        }
-    
-    # Add parameters to the payload
-    if "parameters" not in payload:
-        payload["parameters"] = params
-    
-    # Set up headers
-    headers = {
-        "Authorization": f"Bearer {api_token}",
-        "Content-Type": "application/json"
-    }
-    
-    try:
-        logger.info(f"Sending request to {api_url}")
-        logger.info(f"Using parameters: {params}")
-        response = requests.post(api_url, headers=headers, json=payload)
-        response.raise_for_status()
-        result = response.json()
-        logger.info("Received response successfully")
-        return result
-    except requests.exceptions.HTTPError as error:
-        logger.error(f"Request failed with status code: {error.response.status_code}")
-        logger.error(f"Error message: {error.response.text}")
-        return {"error": error.response.text}
-
-def improved_fetch_audio_from_url(url):
-    """Fetch audio data from URL and convert to base64."""
-    try:
-        logger.info(f"Fetching audio from URL: {url}")
-        response = requests.get(url, timeout=30)
-        response.raise_for_status()
-        
-        # Determine MIME type based on URL
-        file_extension = os.path.splitext(url)[1].lower()
-        mime_type = None
-        
-        if file_extension == '.wav':
-            mime_type = "audio/wav"
-        elif file_extension == '.mp3':
-            mime_type = "audio/mpeg"
-        elif file_extension == '.flac':
-            mime_type = "audio/flac"
-        elif file_extension in ['.m4a', '.aac']:
-            mime_type = "audio/aac"
-        elif file_extension == '.ogg':
-            mime_type = "audio/ogg"
-        else:
-            # Try to detect the MIME type from headers
-            content_type = response.headers.get('Content-Type', '')
-            if content_type.startswith('audio/'):
-                mime_type = content_type
-            else:
-                mime_type = "audio/wav"  # Default to WAV
-        
-        logger.info(f"Detected MIME type: {mime_type}")
-        
-        # Convert to base64
-        base64_audio = base64.b64encode(response.content).decode('utf-8')
-        logger.info(f"Successfully encoded audio to base64, length: {len(base64_audio)}")
-        
-        return mime_type, base64_audio
-    except Exception as e:
-        logger.error(f"Error fetching audio from URL: {e}", exc_info=True)
-        return None, None
-
-def fetch_image_from_url(url):
-    """Fetch image data from URL and convert to base64."""
-    try:
-        logger.info(f"Fetching image from URL: {url}")
-        response = requests.get(url)
-        response.raise_for_status()
-        
-        # Determine MIME type based on URL
-        file_extension = os.path.splitext(url)[1].lower()
-        if file_extension in ['.jpg', '.jpeg']:
-            mime_type = "image/jpeg"
-        elif file_extension == '.png':
-            mime_type = "image/png"
-        elif file_extension == '.gif':
-            mime_type = "image/gif"
-        elif file_extension in ['.bmp', '.tiff', '.webp']:
-            mime_type = f"image/{file_extension[1:]}"
-        else:
-            mime_type = "image/jpeg"  # Default to JPEG
-        
-        # Convert to base64
-        base64_image = base64.b64encode(response.content).decode('utf-8')
-        
-        logger.info(f"Successfully fetched and encoded image, mime type: {mime_type}")
-        return mime_type, base64_image
-    except Exception as e:
-        logger.error(f"Error fetching image from URL: {e}")
-        return None, None
-
-def encode_base64_from_file(file_path):
-    """Encode file content to base64 string and determine MIME type."""
-    file_extension = os.path.splitext(file_path)[1].lower()
-    
-    # Map file extensions to MIME types
-    if file_extension in ['.jpg', '.jpeg']:
-        mime_type = "image/jpeg"
-    elif file_extension == '.png':
-        mime_type = "image/png"
-    elif file_extension == '.gif':
-        mime_type = "image/gif"
-    elif file_extension in ['.bmp', '.tiff', '.webp']:
-        mime_type = f"image/{file_extension[1:]}"
-    elif file_extension == '.flac':
-        mime_type = "audio/flac"
-    elif file_extension == '.wav':
-        mime_type = "audio/wav"
-    elif file_extension == '.mp3':
-        mime_type = "audio/mpeg"
-    elif file_extension in ['.m4a', '.aac']:
-        mime_type = "audio/aac"
-    elif file_extension == '.ogg':
-        mime_type = "audio/ogg"
+def query_huggingface(payload):
+    response = requests.post(API_URL, headers=headers, json=payload)
+    if response.status_code == 200:
+        return response.json()[0]["generated_text"]
     else:
-        mime_type = "application/octet-stream"
-    
-    # Read and encode file content
-    with open(file_path, "rb") as file:
-        encoded_string = base64.b64encode(file.read()).decode('utf-8')
-    
-    return encoded_string, mime_type
-
-def process_message(history, message, conversation_state):
-    """Process user message and update both history and internal state."""
-    # Extract text and files
-    text_content = message["text"] if message["text"] else ""
-    
-    image_files = []
-    audio_files = []
-    
-    # Create content array for internal state
-    content_items = []
-    
-    # Add text if available
-    if text_content:
-        content_items.append({"type": "text", "text": text_content})
-    
-    # Process and immediately convert files to base64
-    if message["files"] and len(message["files"]) > 0:
-        for file_path in message["files"]:
-            file_extension = os.path.splitext(file_path)[1].lower()
-            file_name = os.path.basename(file_path)
-            
-            # Convert the file to base64 immediately
-            base64_content, mime_type = encode_base64_from_file(file_path)
-            
-            # Add to content items for the API
-            if mime_type.startswith("image/"):
-                content_items.append({
-                    "type": "image_url",
-                    "image_url": {
-                        "url": f"data:{mime_type};base64,{base64_content}"
-                    }
-                })
-                image_files.append(file_path)
-            elif mime_type.startswith("audio/"):
-                content_items.append({
-                    "type": "audio_url",
-                    "audio_url": {
-                        "url": f"data:{mime_type};base64,{base64_content}"
-                    }
-                })
-                audio_files.append(file_path)
-    
-    # Only proceed if we have content
-    if content_items:
-        # Add to Gradio chatbot history (for display)
-        history.append({"role": "user", "content": text_content})
-
-        # Add file messages if present
-        for file_path in image_files + audio_files:
-            history.append({"role": "user", "content": {"path": file_path}})
-            
-        logger.info(f"Updated history with user message. Current conversation has {len(image_files)} images and {len(audio_files)} audio files")
-        
-        # Add to internal conversation state (with base64 data)
-        conversation_state.append({
-            "role": "user",
-            "content": content_items
-        })
-    
-    return history, gr.MultimodalTextbox(value=None, interactive=False), conversation_state
+        return f"Error: API request failed with status {response.status_code}"
 
-def bot_response(history, conversation_state):
-    """Generate bot response based on conversation state."""
-    if not conversation_state:
-        return history, conversation_state
+def chatbot_response(user_input):
+    timestamp = datetime.now().strftime("%H:%M:%S")
     
-    # Create the payload
+    # Query the Charm_15 model via Hugging Face API
     payload = {
-        "inputs": conversation_state
+        "inputs": user_input,
+        "parameters": {"max_length": 100, "num_return_sequences": 1}
     }
+    bot_response = query_huggingface(payload)
     
-    # Log the payload for debugging (without base64 data)
-    debug_payload = json.loads(json.dumps(payload))
-    for item in debug_payload["inputs"]:
-        if "content" in item and isinstance(item["content"], list):
-            for content_item in item["content"]:
-                if "image_url" in content_item:
-                    parts = content_item["image_url"]["url"].split(",")
-                    if len(parts) > 1:
-                        content_item["image_url"]["url"] = parts[0] + ",[BASE64_DATA_REMOVED]"
-                if "audio_url" in content_item:
-                    parts = content_item["audio_url"]["url"].split(",")
-                    if len(parts) > 1:
-                        content_item["audio_url"]["url"] = parts[0] + ",[BASE64_DATA_REMOVED]"
-    
-    logger.info(f"Sending payload: {json.dumps(debug_payload, indent=2)}")
-    
-    # Call Hugging Face Inference API
-    response = call_hf_endpoint(payload, HF_API_URL, HF_API_TOKEN)
-    
-    # Extract text response from the Hugging Face API response
-    try:
-        if isinstance(response, dict):
-            if "generated_text" in response:
-                result = response["generated_text"]
-            elif "error" in response:
-                result = f"Error: {response['error']}"
-            else:
-                result = f"Received response: {json.dumps(response)}"
-        else:
-            result = str(response)
-    except Exception as e:
-        result = f"Error processing response: {str(e)}"
-    
-    # Add bot response to history
-    history.append({"role": "assistant", "content": result})
-    
-    # Add to conversation state
-    conversation_state.append({
-        "role": "assistant",
-        "content": [{"type": "text", "text": result}]
-    })
+    # Fallback if response is empty or an error
+    if not bot_response or "Error" in bot_response:
+        bot_response = f"Failed to generate response. Echo: '{user_input}' at {timestamp}"
+    else:
+        bot_response = f"{bot_response} (Generated at {timestamp})"
     
-    return history, conversation_state
+    return [[user_input, bot_response]]
 
-def enable_input():
-    """Re-enable the input box after bot responds."""
-    return gr.MultimodalTextbox(interactive=True)
+# Custom CSS for circular avatar
+custom_css = """
+.bot-avatar {
+    border-radius: 50% !important;
+    width: 32px !important;
+    height: 32px !important;
+}
+"""
 
-def update_debug(conversation_state):
-    """Update debug output with the last payload that would be sent."""
-    if not conversation_state:
-        return {}
-    
-    # Create a payload from the conversation
-    payload = {
-        "inputs": conversation_state
-    }
-    
-    # Remove base64 data to avoid cluttering the UI
-    sanitized_payload = json.loads(json.dumps(payload))
-    for item in sanitized_payload["inputs"]:
-        if "content" in item and isinstance(item["content"], list):
-            for content_item in item["content"]:
-                if "image_url" in content_item:
-                    parts = content_item["image_url"]["url"].split(",")
-                    if len(parts) > 1:
-                        content_item["image_url"]["url"] = parts[0] + ",[BASE64_DATA_REMOVED]"
-                if "audio_url" in content_item:
-                    parts = content_item["audio_url"]["url"].split(",")
-                    if len(parts) > 1:
-                        content_item["audio_url"]["url"] = parts[0] + ",[BASE64_DATA_REMOVED]"
-    
-    return sanitized_payload
-
-# Gradio interface setup
-with gr.Blocks() as demo:
-    gr.Markdown("# Chatbot with Hugging Face Models")
-    
-    with gr.Tab("Chat"):
-        # Chat history display
-        chat_history = gr.Chatbot(label="Chat")
-        
-        with gr.Row():
-            with gr.Column():
-                prompt_input = gr.MultimodalTextbox(label="Enter your prompt", placeholder="Type your message here...", lines=2)
-                generate_button = gr.Button("Send")
-        
-        # Chat interaction
-        generate_button.click(
-            process_message,
-            inputs=[chat_history, prompt_input],
-            outputs=[chat_history, prompt_input]
-        )
-    
-    # Debug output
-    debug_output = gr.JSON(label="Debug Output")
-    demo.load(update_debug, inputs=[chat_history], outputs=debug_output, every=1)
+# Gradio interface
+with gr.Blocks(css=custom_css) as demo:
+    chatbot = gr.Chatbot(
+        label="Charm_15 Chatbot Playground",
+        avatar_images=[None, BOT_ICON],  # None for user, GitHub image for bot
+        bubble_full_width=False
+    )
+    msg = gr.Textbox(placeholder="Type your message here...")
+    submit_btn = gr.Button("Send")
+    
+    def submit_message(user_input, history=[]):
+        new_conversation = chatbot_response(user_input)
+        return history + new_conversation, ""
+    
+    # Bind submission events
+    msg.submit(submit_message, [msg, chatbot], [chatbot, msg])
+    submit_btn.click(submit_message, [msg, chatbot], [chatbot, msg])
 
-# Launch the interface
+# Launch the app (Spaces handles this automatically)
 demo.launch()