app.py

import os
import logging
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from functools import partial
import time

import gradio as gr
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

@dataclass
class Config:
    """Application configuration."""
    DEFAULT_BOT_URL: str = "http://localhost:1212/api/v1/chat"
    BOT_URL: str = os.getenv("RWID_BOT_URL", DEFAULT_BOT_URL)
    HEALTH_CHECK_URL: str = BOT_URL.replace("/api/v1/chat", "/")
    REQUEST_TIMEOUT: int = 60
    MAX_RETRIES: int = 3
    EXAMPLES: List[List[str]] = None

    def __post_init__(self):
        self.EXAMPLES = [
            ["Apa itu RWID?"],
            ["Jika saya sorang pemula apakah saya bisa bergabung dengan RWID?"],
            ["Siapa itu founder RWID?"],
            ["kenapa saya harus join rwid ? dan berapa biaya pendaftaran rwid ? lalu setelah join rwid saya harus gimana ?"],
        ]

class APIClient:
    """Handle API communication with retry logic."""
    
    def __init__(self, config: Config):
        self.config = config
        self.session = self._create_session()
    
    def _create_session(self) -> requests.Session:
        """Create session with retry logic."""
        session = requests.Session()
        retry_strategy = Retry(
            total=self.config.MAX_RETRIES,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("http://", adapter)
        session.mount("https://", adapter)
        return session
    
    def health_check(self) -> str:
        """Check server health status."""
        try:
            response = self.session.get(
                self.config.HEALTH_CHECK_URL,
                timeout=self.config.REQUEST_TIMEOUT
            )
            if response.status_code == 200:
                return "✅ Server is healthy and responding!"
            return f"⚠️ Server returned status {response.status_code}"
        except requests.exceptions.RequestException as e:
            logger.exception("Healthcheck failed")
            return "❌ Server connection failed."
    
    def send_message(self, message: str, history: List[Dict[str, Any]]) -> str:
        """Send message to API and handle response."""
        try:
            payload = self._prepare_payload(message, history)
            response = self.session.post(
                self.config.BOT_URL,
                json=payload,
                timeout=self.config.REQUEST_TIMEOUT
            )
            return self._handle_response(response)
        except requests.exceptions.RequestException as e:
            logger.exception("Network error while contacting API")
            return "Connection error. Please check your network connection."
    
    def _prepare_payload(self, message: str, history: List[Dict[str, Any]]) -> Dict:
        """Prepare API request payload."""
        return {
            "messages": [
                *[msg for msg in history if isinstance(msg, dict)],
                {"role": "user", "content": message}
            ]
        }
    
    def _handle_response(self, response: requests.Response) -> str:
        """Process and extract assistant message from API response."""
        try:
            response.raise_for_status()
            return response.json()["choices"][0]["message"]["content"]
        except requests.exceptions.HTTPError:
            logger.exception(f"API HTTP error. Status Code: {response.status_code}")
        except (KeyError, IndexError):
            logger.exception("Unexpected response format")
        except Exception as e:
            logger.exception(f"Unexpected error: {str(e)}")
        return "Apologies, I'm experiencing technical difficulties. Please try again later."

class ChatInterface:
    """Manage chat interface components and interactions."""
    
    def __init__(self, api_client: APIClient, config: Config):
        self.api_client = api_client
        self.config = config
    
    def create_interface(self) -> gr.Blocks:
        """Create and configure Gradio chat interface."""
        with gr.Blocks(
            title="RWID Bot Assistant 🤖",
            theme=gr.themes.Soft(primary_hue="blue"),
            css="""
                .gradio-container {max-width: 800px; margin: auto}
                .message-wrap {padding: 10px}
                .assistant-message {background-color: #f0f7ff}
                .user-message {background-color: #f5f5f5}
            """
        ) as interface:
            self._create_header()
            self._create_chat_component()
            self._create_footer()
        return interface

    def _create_header(self):
        """Create header section."""
        gr.Markdown("""
# RWID Bot Assistant 🌐
**Your Expert Guide to Distributed Teams and Digital Collaboration**
        """)
        health_check = self.api_client.health_check()
        gr.Markdown(f"**Server Status:** {health_check}")

    def _create_chat_component(self):
        """Create main chat interface component."""
        gr.ChatInterface(
            fn=self.api_client.send_message,
            examples=self.config.EXAMPLES,
            chatbot=gr.Chatbot(
                label="Chat History",
                show_copy_button=True,
                height=500,
                type="messages",
                bubble_full_width=False,
                show_label=True,
                # avatar_images=["👤", "🤖"]
            ),
            textbox=gr.Textbox(
                placeholder="Type your question about remote work...",
                autofocus=True,
                container=False,
                scale=7,
                show_label=False
            ),
            submit_btn=gr.Button("Send Message", variant="primary", size="lg"),
            additional_inputs=None
        )

    def _create_footer(self):
        """Create footer section."""
        gr.Markdown("""
---
### Example Topics 📚
- Remote team management strategies
- Digital nomad lifestyle tips
- Async communication best practices
- Project collaboration tools
- Remote work-life balance

*This bot is continuously learning and improving. Your feedback helps us enhance the experience.*
        """)

def main():
    """Application entry point."""
    config = Config()
    api_client = APIClient(config)
    chat_interface = ChatInterface(api_client, config)
    
    logger.info("Starting RWID Bot Assistant interface...")
    interface = chat_interface.create_interface()
    interface.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False
    )

if __name__ == "__main__":
    main()