""" Gradio web interface for sentiment analysis. This module provides a modern, responsive web interface using Gradio for human interaction with the sentiment analysis system, including real-time analysis, confidence visualization, and history tracking. """ import asyncio import logging import json <<<<<<< HEAD import os ======= >>>>>>> 6c0a877e212b959072c8948b934d212c97a3c597 from typing import Dict, Any, List, Tuple, Optional from datetime import datetime import pandas as pd import plotly.graph_objects as go import plotly.express as px try: import gradio as gr GRADIO_AVAILABLE = True except ImportError: GRADIO_AVAILABLE = False logging.error("Gradio not available. Install with: pip install gradio") from .sentiment_analyzer import get_analyzer, SentimentResult, SentimentLabel <<<<<<< HEAD from .tools import list_tools ======= >>>>>>> 6c0a877e212b959072c8948b934d212c97a3c597 class SentimentHistory: """Manages sentiment analysis history.""" def __init__(self, max_entries: int = 100): self.max_entries = max_entries self.entries: List[Dict[str, Any]] = [] self.logger = logging.getLogger(__name__) def add_entry(self, text: str, result: SentimentResult, backend: str) -> None: entry = { "timestamp": datetime.now().isoformat(), "text": text[:100] + "..." if len(text) > 100 else text, "full_text": text, "label": result.label.value, "confidence": result.confidence, "backend": backend, "raw_scores": result.raw_scores } self.entries.append(entry) if len(self.entries) > self.max_entries: self.entries = self.entries[-self.max_entries:] def get_recent_entries(self, count: int = 10) -> List[Dict[str, Any]]: return self.entries[-count:] if self.entries else [] def get_statistics(self) -> Dict[str, Any]: if not self.entries: return { "total_analyses": 0, "label_distribution": {}, "average_confidence": 0.0, "backend_usage": {} } labels = [entry["label"] for entry in self.entries] confidences = [entry["confidence"] for entry in self.entries] backends = [entry["backend"] for entry in self.entries] label_counts = { "positive": labels.count("positive"), "negative": labels.count("negative"), "neutral": labels.count("neutral") } backend_counts = {} for backend in backends: backend_counts[backend] = backend_counts.get(backend, 0) + 1 return { "total_analyses": len(self.entries), "label_distribution": label_counts, "average_confidence": sum(confidences) / len(confidences), "backend_usage": backend_counts } class GradioInterface: """Gradio web interface for sentiment analysis.""" def __init__(self, title: str = "Sentiment Analysis Server", description: str = "Analyze text sentiment using TextBlob or Transformers"): self.title = title self.description = description self.logger = logging.getLogger(__name__) self.history = SentimentHistory() self.interface = None self._setup_interface() def _setup_interface(self) -> None: if not GRADIO_AVAILABLE: raise RuntimeError("Gradio not available") with gr.Blocks( theme=gr.themes.Soft(), title=self.title ) as interface: gr.Markdown(f"# {self.title}") gr.Markdown(f"*{self.description}*") with gr.Tabs(): with gr.TabItem("Sentiment Analysis"): with gr.Row(): with gr.Column(scale=2): text_input = gr.Textbox( label="Text to Analyze", placeholder="Enter text here to analyze its sentiment...", lines=4 ) with gr.Row(): backend_choice = gr.Dropdown( choices=["auto", "textblob", "transformers"], value="auto", label="Analysis Backend" ) analyze_btn = gr.Button( "Analyze Sentiment", variant="primary" ) with gr.Column(scale=1): result_display = gr.HTML( value="

Enter text and click 'Analyze Sentiment' to see results.

" ) confidence_plot = gr.Plot(visible=False) gr.Markdown("### Quick Examples") with gr.Row(): pos_btn = gr.Button("😊 Positive", size="sm") neu_btn = gr.Button("😐 Neutral", size="sm") neg_btn = gr.Button("😞 Negative", size="sm") mix_btn = gr.Button("📝 Mixed", size="sm") with gr.TabItem("Batch Analysis"): with gr.Row(): with gr.Column(): batch_input = gr.Textbox( label="Texts to Analyze (one per line)", placeholder="Enter multiple texts, one per line...", lines=8 ) with gr.Row(): batch_backend = gr.Dropdown( choices=["auto", "textblob", "transformers"], value="auto", label="Analysis Backend" ) batch_analyze_btn = gr.Button( "Analyze Batch", variant="primary" ) with gr.Column(): batch_results = gr.DataFrame( label="Batch Results", headers=["Text", "Sentiment", "Confidence"] ) batch_summary_plot = gr.Plot(visible=False) with gr.TabItem("Analysis History"): with gr.Row(): refresh_history_btn = gr.Button("Refresh History", variant="secondary") clear_history_btn = gr.Button("Clear History", variant="stop") with gr.Row(): with gr.Column(scale=2): history_table = gr.DataFrame( label="Recent Analyses", headers=["Time", "Text", "Sentiment", "Confidence", "Backend"] ) with gr.Column(scale=1): stats_display = gr.HTML(value="

No analyses yet.

") history_plot = gr.Plot(visible=False) with gr.TabItem("Settings & Info"): with gr.Row(): with gr.Column(): gr.Markdown("### Backend Information") backend_info = gr.HTML(value="

Loading backend information...

") refresh_info_btn = gr.Button("Refresh Info", variant="secondary") with gr.Column(): gr.Markdown("### Usage Tips") gr.Markdown(""" - **Auto**: Automatically selects the best available backend - **TextBlob**: Fast, simple sentiment analysis - **Transformers**: More accurate, AI-powered analysis - **Batch Analysis**: Process multiple texts at once - **History**: Track your analysis results over time """) # Event handlers def analyze_sentiment(text: str, backend: str) -> Tuple[str, gr.Plot]: return asyncio.run(self._analyze_sentiment_async(text, backend)) def analyze_batch(texts: str, backend: str) -> Tuple[pd.DataFrame, gr.Plot]: return asyncio.run(self._analyze_batch_async(texts, backend)) def refresh_history() -> Tuple[pd.DataFrame, str, gr.Plot]: return self._get_history_data() def clear_history() -> Tuple[pd.DataFrame, str, gr.Plot]: self.history.entries.clear() return self._get_history_data() def get_backend_info() -> str: return asyncio.run(self._get_backend_info_async()) <<<<<<< HEAD def get_mcp_schema() -> str: """Get MCP tools schema as JSON.""" return asyncio.run(self._get_mcp_schema_async()) ======= >>>>>>> 6c0a877e212b959072c8948b934d212c97a3c597 # Example texts examples = [ "I absolutely love this new feature! It's incredible and makes everything so much easier.", "The weather is okay today, nothing particularly special about it.", "This is terrible and frustrating. I hate how complicated this has become.", "The movie had great visuals but the plot was disappointing. Mixed feelings overall." ] # Wire up events analyze_btn.click( analyze_sentiment, inputs=[text_input, backend_choice], outputs=[result_display, confidence_plot] ) batch_analyze_btn.click( analyze_batch, inputs=[batch_input, batch_backend], outputs=[batch_results, batch_summary_plot] ) refresh_history_btn.click( refresh_history, outputs=[history_table, stats_display, history_plot] ) clear_history_btn.click( clear_history, outputs=[history_table, stats_display, history_plot] ) refresh_info_btn.click( get_backend_info, outputs=[backend_info] ) # Example buttons pos_btn.click(lambda: examples[0], outputs=[text_input]) neu_btn.click(lambda: examples[1], outputs=[text_input]) neg_btn.click(lambda: examples[2], outputs=[text_input]) mix_btn.click(lambda: examples[3], outputs=[text_input]) # Load initial data interface.load(get_backend_info, outputs=[backend_info]) interface.load(refresh_history, outputs=[history_table, stats_display, history_plot]) self.interface = interface async def _analyze_sentiment_async(self, text: str, backend: str) -> Tuple[str, gr.Plot]: try: if not text.strip(): return "

Please enter some text to analyze.

", gr.Plot(visible=False) analyzer = await get_analyzer(backend) result = await analyzer.analyze(text) self.history.add_entry(text, result, analyzer.backend) sentiment_class = f"sentiment-{result.label.value}" confidence_class = ( "confidence-high" if result.confidence > 0.7 else "confidence-medium" if result.confidence > 0.4 else "confidence-low" ) html_result = f"""

Analysis Result

Sentiment: {result.label.value.title()}

Confidence: {result.confidence:.2%}

Backend: {analyzer.backend}

Text Length: {len(text)} characters

""" plot = self._create_confidence_plot(result) return html_result, plot except Exception as e: self.logger.error(f"Analysis failed: {e}") error_html = f"""

Analysis Error

Error: {str(e)}

Please try again or check your input.

""" return error_html, gr.Plot(visible=False) async def _analyze_batch_async(self, texts: str, backend: str) -> Tuple[pd.DataFrame, gr.Plot]: try: if not texts.strip(): return pd.DataFrame(), gr.Plot(visible=False) text_list = [t.strip() for t in texts.split('\n') if t.strip()] if not text_list: return pd.DataFrame(), gr.Plot(visible=False) analyzer = await get_analyzer(backend) results = await analyzer.analyze_batch(text_list) data = [] for text, result in zip(text_list, results): self.history.add_entry(text, result, analyzer.backend) data.append({ "Text": text[:50] + "..." if len(text) > 50 else text, "Sentiment": result.label.value.title(), "Confidence": f"{result.confidence:.2%}" }) df = pd.DataFrame(data) plot = self._create_batch_summary_plot(results) return df, plot except Exception as e: self.logger.error(f"Batch analysis failed: {e}") return pd.DataFrame([{"Error": str(e)}]), gr.Plot(visible=False) def _create_confidence_plot(self, result: SentimentResult) -> gr.Plot: try: fig = go.Figure(go.Indicator( mode="gauge+number", value=result.confidence * 100, domain={'x': [0, 1], 'y': [0, 1]}, title={'text': f"Confidence - {result.label.value.title()}"}, gauge={ 'axis': {'range': [None, 100]}, 'bar': {'color': "darkblue"}, 'steps': [ {'range': [0, 40], 'color': "lightgray"}, {'range': [40, 70], 'color': "yellow"}, {'range': [70, 100], 'color': "green"} ] } )) fig.update_layout(height=300, margin=dict(l=20, r=20, t=40, b=20)) return gr.Plot(value=fig, visible=True) except Exception as e: self.logger.error(f"Failed to create confidence plot: {e}") return gr.Plot(visible=False) def _create_batch_summary_plot(self, results: List[SentimentResult]) -> gr.Plot: try: labels = [result.label.value for result in results] label_counts = { "Positive": labels.count("positive"), "Negative": labels.count("negative"), "Neutral": labels.count("neutral") } fig = px.pie( values=list(label_counts.values()), names=list(label_counts.keys()), title="Sentiment Distribution", color_discrete_map={ "Positive": "#22c55e", "Negative": "#ef4444", "Neutral": "#6b7280" } ) fig.update_layout(height=300, margin=dict(l=20, r=20, t=40, b=20)) return gr.Plot(value=fig, visible=True) except Exception as e: self.logger.error(f"Failed to create batch summary plot: {e}") return gr.Plot(visible=False) def _get_history_data(self) -> Tuple[pd.DataFrame, str, gr.Plot]: try: entries = self.history.get_recent_entries(20) if not entries: empty_df = pd.DataFrame(columns=["Time", "Text", "Sentiment", "Confidence", "Backend"]) return empty_df, "

No analyses yet.

", gr.Plot(visible=False) data = [] for entry in reversed(entries): data.append({ "Time": entry["timestamp"][:19].replace("T", " "), "Text": entry["text"], "Sentiment": entry["label"].title(), "Confidence": f"{entry['confidence']:.2%}", "Backend": entry["backend"] }) df = pd.DataFrame(data) stats = self.history.get_statistics() stats_html = f"""

📊 Analysis Statistics

Total Analyses: {stats['total_analyses']}

Average Confidence: {stats['average_confidence']:.2%}

Sentiment Distribution:

""" plot = self._create_history_plot(stats) if stats['total_analyses'] > 0 else gr.Plot(visible=False) return df, stats_html, plot except Exception as e: self.logger.error(f"Failed to get history data: {e}") error_df = pd.DataFrame([{"Error": str(e)}]) return error_df, f"

Error loading history: {e}

", gr.Plot(visible=False) def _create_history_plot(self, stats: Dict[str, Any]) -> gr.Plot: try: labels = list(stats['label_distribution'].keys()) values = list(stats['label_distribution'].values()) fig = px.bar( x=[label.title() for label in labels], y=values, title="Historical Sentiment Distribution", color=labels, color_discrete_map={ "positive": "#22c55e", "negative": "#ef4444", "neutral": "#6b7280" } ) fig.update_layout(height=300, margin=dict(l=20, r=20, t=40, b=20), showlegend=False) return gr.Plot(value=fig, visible=True) except Exception as e: self.logger.error(f"Failed to create history plot: {e}") return gr.Plot(visible=False) async def _get_backend_info_async(self) -> str: try: analyzer = await get_analyzer("auto") info = analyzer.get_info() html = f"""

🔧 Backend Information

Current Backend: {info['backend']}

Model Loaded: {'Yes' if info['model_loaded'] else 'No'}

TextBlob Available: {'Yes' if info['textblob_available'] else 'No'}

Transformers Available: {'Yes' if info['transformers_available'] else 'No'}

CUDA Available: {'Yes' if info.get('cuda_available', False) else 'No'}

{f"

Model Name: {info['model_name']}

" if info.get('model_name') else ""}
""" return html except Exception as e: self.logger.error(f"Failed to get backend info: {e}") return f"""

❌ Backend Error

Failed to load backend information: {str(e)}

""" <<<<<<< HEAD async def _get_mcp_schema_async(self) -> str: """Get MCP tools schema as formatted JSON.""" try: tools = await list_tools() schema = { "mcp_version": "2024-11-05", "server_info": { "name": "sentiment-analyzer", "version": "1.0.0", "description": "Sentiment analysis server using TextBlob and Transformers" }, "tools": tools, "total_tools": len(tools) } return json.dumps(schema, indent=2) except Exception as e: self.logger.error(f"Failed to get MCP schema: {e}") return json.dumps({ "error": str(e), "error_type": type(e).__name__ }, indent=2) ======= >>>>>>> 6c0a877e212b959072c8948b934d212c97a3c597 def launch(self, **kwargs) -> None: if not self.interface: raise RuntimeError("Interface not initialized") <<<<<<< HEAD # Check for MCP server mode from environment variable or parameter mcp_server_enabled = ( kwargs.get("mcp_server", False) or os.getenv("GRADIO_MCP_SERVER", "").lower() in ("true", "1", "yes", "on") ) ======= >>>>>>> 6c0a877e212b959072c8948b934d212c97a3c597 launch_params = { "server_name": "0.0.0.0", "server_port": 7860, "share": False, "debug": False, "show_error": True, "quiet": False } <<<<<<< HEAD # Add MCP server parameter if enabled if mcp_server_enabled: launch_params["mcp_server"] = True self.logger.info("MCP server functionality enabled for Gradio interface") launch_params.update(kwargs) self.logger.info(f"Launching Gradio interface on {launch_params['server_name']}:{launch_params['server_port']}") if mcp_server_enabled: self.logger.info("Gradio interface will also serve as MCP server with API endpoints") ======= launch_params.update(kwargs) self.logger.info(f"Launching Gradio interface on {launch_params['server_name']}:{launch_params['server_port']}") >>>>>>> 6c0a877e212b959072c8948b934d212c97a3c597 try: self.interface.launch(**launch_params) except Exception as e: self.logger.error(f"Failed to launch interface: {e}") raise def create_gradio_interface(**kwargs) -> GradioInterface: if not GRADIO_AVAILABLE: raise RuntimeError("Gradio not available. Install with: pip install gradio") return GradioInterface(**kwargs) async def main() -> None: logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) interface = create_gradio_interface() interface.launch(debug=True) if __name__ == "__main__": asyncio.run(main())