Spaces:
Runtime error
Runtime error
| """Single-file Gradio app for Multilingual ABSA. | |
| Loads the ONNX model (INT8 → FP32) with a rule-based fallback, runs inference | |
| via ``absa.pipeline.ABSAPipeline``, persists predictions to SQLite, and serves | |
| a Gradio (gr.Blocks) UI. Run with ``python app.py``. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import os | |
| import sys | |
| import time | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Any | |
| import gradio as gr | |
| from sqlalchemy import Column, DateTime, Float, Integer, String, create_engine | |
| from sqlalchemy.orm import declarative_base, sessionmaker | |
| # Make the src-layout `absa` package importable when run directly | |
| # (e.g. `python app.py`) without a prior `pip install -e .`. All `absa` | |
| # imports below are lazy, so this bootstrap runs before any of them. | |
| _SRC_DIR = Path(__file__).resolve().parent / "src" | |
| if str(_SRC_DIR) not in sys.path: | |
| sys.path.insert(0, str(_SRC_DIR)) | |
| logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") | |
| _logger = logging.getLogger("absa.app") | |
| # ── SQLAlchemy / SQLite ─────────────────────────────────────────────────── | |
| DB_PATH = "absa.db" | |
| engine = create_engine( | |
| f"sqlite:///{DB_PATH}", | |
| connect_args={"check_same_thread": False}, | |
| ) | |
| Base = declarative_base() | |
| Session = sessionmaker(bind=engine) | |
| def _utcnow() -> datetime: | |
| return datetime.now(timezone.utc) | |
| class Prediction(Base): | |
| """One aspect-sentiment pair extracted from a single review.""" | |
| __tablename__ = "predictions" | |
| id = Column(Integer, primary_key=True) | |
| text = Column(String, nullable=False) | |
| language = Column(String, nullable=False) | |
| aspect = Column(String, nullable=False) | |
| sentiment = Column(String, nullable=False) | |
| confidence = Column(Float, nullable=False) | |
| created_at = Column(DateTime, default=_utcnow) | |
| Base.metadata.create_all(engine) | |
| # ── Lazy, failure-tolerant model loading ────────────────────────────────── | |
| # Module-level singleton (cached for the app lifetime). If the ONNX models | |
| # are missing the pipeline degrades to the rule-based engine — never crashes. | |
| DEFAULT_MODEL_PATH = Path(os.getenv("MODEL_PATH", "models/onnx")) | |
| def _load_pipeline() -> Any: | |
| """Import the ABSA pipeline lazily and attempt custom ONNX model load.""" | |
| try: | |
| from absa.pipeline.absa_pipeline import ABSAPipeline | |
| except Exception as exc: # pragma: no cover - import-time guard | |
| _logger.error("Could not import ABSA pipeline: %s", exc) | |
| raise | |
| pipelin = ABSAPipeline() | |
| try: | |
| pipelin.load_models() | |
| except Exception as exc: | |
| _logger.warning("Custom model load failed (%s); using rule-based engine.", exc) | |
| return pipelin | |
| PIPELINE = _load_pipeline() | |
| def _inference_mode() -> str: | |
| """Report which engine is active: ONNX INT8 / ONNX FP32 / rule-based.""" | |
| if PIPELINE.aspect_model is not None: | |
| if (DEFAULT_MODEL_PATH / "aspect_extraction_int8").exists(): | |
| return "ONNX INT8" | |
| if (DEFAULT_MODEL_PATH / "aspect_extraction").exists(): | |
| return "ONNX FP32" | |
| return "ONNX" | |
| return "rule-based" | |
| MODE = _inference_mode() | |
| def get_count() -> int: | |
| """Total predictions in SQLite.""" | |
| try: | |
| session = Session() | |
| return session.query(Prediction).count() | |
| except Exception: | |
| return 0 | |
| finally: | |
| session.close() | |
| def get_history_rows() -> list[list]: | |
| """Return last 50 predictions as DataFrame rows.""" | |
| try: | |
| session = Session() | |
| rows = session.query(Prediction).order_by(Prediction.id.desc()).limit(50).all() | |
| return [ | |
| [ | |
| r.created_at.strftime("%Y-%m-%d %H:%M") if r.created_at else "—", | |
| r.language or "—", | |
| r.aspect or "—", | |
| r.sentiment or "—", | |
| round(r.confidence, 3) if r.confidence else 0.0, | |
| (r.text[:60] + "...") if r.text and len(r.text) > 60 else (r.text or "—"), | |
| ] | |
| for r in rows | |
| ] | |
| except Exception: | |
| return [] | |
| finally: | |
| session.close() | |
| def get_history_md() -> str: | |
| return f"Showing last 50 predictions · Total: **{get_count()}**" | |
| def clear_history() -> tuple[str, list]: | |
| """Wipe predictions table.""" | |
| try: | |
| session = Session() | |
| session.query(Prediction).delete() | |
| session.commit() | |
| except Exception: | |
| pass | |
| finally: | |
| session.close() | |
| return "History cleared · Total: **0**", [] | |
| def on_analyze(text: str) -> tuple[str, str, str, Any, list]: | |
| """Run inference, save to DB, return formatted UI outputs.""" | |
| if not text or not text.strip(): | |
| return ( | |
| '<div class="meta-badges"><span class="badge">Detected Language: <strong>—</strong></span><span class="badge">Latency: <strong>—</strong></span></div>', | |
| '<div class="summary-vis empty">Please enter a review to analyze.</div>', | |
| '<div class="empty-state">No analysis yet.</div>', | |
| gr.update(visible=False, value={}) | |
| ) | |
| start = time.time() | |
| try: | |
| response = PIPELINE.predict(text) | |
| aspects = response.aspects if hasattr(response, "aspects") else response.get("aspects", []) | |
| language = response.detected_language if hasattr(response, "detected_language") else response.get("detected_language", "unknown") | |
| elapsed_ms = (time.time() - start) * 1000 | |
| except Exception as e: | |
| return ( | |
| '<div class="meta-badges"><span class="badge">Detected Language: <strong>Error</strong></span><span class="badge">Latency: <strong>—</strong></span></div>', | |
| '<div class="summary-vis empty">Error during analysis.</div>', | |
| f'<div class="empty-state" style="color:#DC2626;">Unable to analyze this review: {str(e)[:120]}</div>', | |
| gr.update(visible=False, value={}) | |
| ) | |
| lang_display = { | |
| "en": "English", "english": "English", | |
| "hi": "Hindi", "hindi": "Hindi", | |
| "hinglish": "Hinglish", "hi-en": "Hinglish", | |
| }.get(str(language).lower(), str(language).title()) | |
| meta_md = f'<div class="meta-badges"><span class="badge">Detected Language: <strong>{lang_display}</strong></span><span class="badge">Latency: <strong>{elapsed_ms:.0f} ms</strong></span></div>' | |
| pos_count = neg_count = neu_count = 0 | |
| table_rows = "" | |
| for a in aspects: | |
| aspect_text = a.get("aspect", "") if isinstance(a, dict) else getattr(a, "aspect", "") | |
| sentiment = a.get("sentiment", "neutral") if isinstance(a, dict) else getattr(a, "sentiment", "neutral") | |
| confidence = float(a.get("confidence", 0.0) if isinstance(a, dict) else getattr(a, "confidence", 0.0)) | |
| if sentiment == "positive": | |
| pos_count += 1 | |
| sent_color = "#16A34A" | |
| sent_label = "Positive" | |
| elif sentiment == "negative": | |
| neg_count += 1 | |
| sent_color = "#DC2626" | |
| sent_label = "Negative" | |
| else: | |
| neu_count += 1 | |
| sent_color = "#64748B" | |
| sent_label = "Neutral" | |
| conf_pct = int(confidence * 100) | |
| table_rows += f""" | |
| <tr> | |
| <td class="aspect-col">{aspect_text}</td> | |
| <td class="sentiment-col"> | |
| <span class="sent-dot" style="background-color: {sent_color};"></span> | |
| <span style="color: {sent_color}; font-weight: 600;">{sent_label}</span> | |
| </td> | |
| <td class="conf-col"> | |
| <div class="conf-bar-bg"><div class="conf-bar-fill" style="width: {conf_pct}%;"></div></div> | |
| <span class="conf-val">{conf_pct}%</span> | |
| </td> | |
| </tr> | |
| """ | |
| if not aspects: | |
| results_html = '<div class="empty-state">No aspects detected in this review.</div>' | |
| else: | |
| results_html = f""" | |
| <table class="results-table"> | |
| <thead> | |
| <tr> | |
| <th>Aspect</th> | |
| <th>Sentiment</th> | |
| <th>Confidence</th> | |
| </tr> | |
| </thead> | |
| <tbody> | |
| {table_rows} | |
| </tbody> | |
| </table> | |
| """ | |
| total = len(aspects) | |
| # Visualization blocks | |
| vis_pos = f'Positive <span class="vis-block pos">{"█" * pos_count}</span> {pos_count}' if pos_count else 'Positive <span class="vis-block empty-block"></span> 0' | |
| vis_neg = f'Negative <span class="vis-block neg">{"█" * neg_count}</span> {neg_count}' if neg_count else 'Negative <span class="vis-block empty-block"></span> 0' | |
| vis_neu = f'Neutral <span class="vis-block neu">{"█" * neu_count}</span> {neu_count}' if neu_count else 'Neutral <span class="vis-block empty-block"></span> 0' | |
| summary_md = f""" | |
| <div class="summary-vis"> | |
| <div class="summary-title">{total} Aspects</div> | |
| <div class="vis-row">{vis_pos}</div> | |
| <div class="vis-row">{vis_neg}</div> | |
| <div class="vis-row">{vis_neu}</div> | |
| </div> | |
| """ | |
| # Save to SQLite | |
| try: | |
| session = Session() | |
| for a in aspects: | |
| aspect_text = a.get("aspect", "") if isinstance(a, dict) else getattr(a, "aspect", "") | |
| sentiment = a.get("sentiment", "neutral") if isinstance(a, dict) else getattr(a, "sentiment", "neutral") | |
| confidence = a.get("confidence", 0.0) if isinstance(a, dict) else getattr(a, "confidence", 0.0) | |
| session.add(Prediction( | |
| text=text, | |
| language=lang_display, | |
| aspect=aspect_text, | |
| sentiment=sentiment, | |
| confidence=float(confidence or 0.0), | |
| )) | |
| session.commit() | |
| except Exception: | |
| pass | |
| finally: | |
| session.close() | |
| return (meta_md, summary_md, results_html, gr.update(visible=True, value=aspects)) | |
| # ── Gradio UI (gr.Blocks) ───────────────────────────────────────────────── | |
| custom_theme = gr.themes.Soft( | |
| primary_hue=gr.themes.colors.blue, | |
| secondary_hue=gr.themes.colors.slate, | |
| neutral_hue=gr.themes.colors.slate, | |
| font=gr.themes.GoogleFont("Inter"), | |
| ).set( | |
| body_background_fill="#F8FAFC", | |
| body_background_fill_dark="#F8FAFC", | |
| body_text_color="#0F172A", | |
| body_text_color_dark="#0F172A", | |
| block_background_fill="#FFFFFF", | |
| block_background_fill_dark="#FFFFFF", | |
| block_border_color="#E2E8F0", | |
| block_border_color_dark="#E2E8F0", | |
| block_label_background_fill="#FFFFFF", | |
| block_label_background_fill_dark="#FFFFFF", | |
| block_label_text_color="#0F172A", | |
| block_label_text_color_dark="#0F172A", | |
| button_primary_background_fill="#2563EB", | |
| button_primary_background_fill_dark="#2563EB", | |
| button_primary_text_color="#FFFFFF", | |
| button_primary_text_color_dark="#FFFFFF", | |
| button_primary_border_color="#2563EB", | |
| button_primary_border_color_dark="#2563EB", | |
| button_secondary_background_fill="#FFFFFF", | |
| button_secondary_background_fill_dark="#FFFFFF", | |
| button_secondary_text_color="#0F172A", | |
| button_secondary_text_color_dark="#0F172A", | |
| button_secondary_border_color="#CBD5E1", | |
| button_secondary_border_color_dark="#CBD5E1", | |
| input_background_fill="#FFFFFF", | |
| input_background_fill_dark="#FFFFFF", | |
| input_border_color="#CBD5E1", | |
| input_border_color_dark="#CBD5E1", | |
| panel_background_fill="#FFFFFF", | |
| panel_background_fill_dark="#FFFFFF", | |
| table_even_background_fill="#F8FAFC", | |
| table_even_background_fill_dark="#F8FAFC", | |
| table_odd_background_fill="#FFFFFF", | |
| table_odd_background_fill_dark="#FFFFFF", | |
| table_border_color="#E2E8F0", | |
| table_border_color_dark="#E2E8F0", | |
| border_color_primary="#E2E8F0", | |
| border_color_primary_dark="#E2E8F0", | |
| color_accent_soft="#EFF6FF", | |
| color_accent_soft_dark="#EFF6FF", | |
| ) | |
| custom_css = """ | |
| /* 60/30/10 Design System */ | |
| /* Layout & Base */ | |
| .gradio-container { max-width: 1200px !important; margin: auto; font-family: 'Inter', sans-serif; padding: 24px !important; } | |
| /* Typography */ | |
| .section-title { font-size: 1.1rem; font-weight: 600; color: #0F172A; margin-bottom: 4px; } | |
| .section-sub { font-size: 0.9rem; color: #64748B; margin-bottom: 16px; } | |
| /* Top Header */ | |
| .top-header { border-bottom: 1px solid #E2E8F0; padding-bottom: 16px; margin-bottom: 24px; align-items: center; } | |
| .header-content h1 { font-size: 1.6rem; font-weight: 600; color: #0F172A; margin: 0 0 4px 0; letter-spacing: -0.02em; } | |
| .header-subtitle { font-size: 0.9rem; color: #64748B; } | |
| /* Header Metrics */ | |
| .header-metrics { display: flex; gap: 24px; justify-content: flex-end; flex-wrap: wrap; } | |
| .metric { display: flex; flex-direction: column; } | |
| .m-label { font-size: 0.65rem; font-weight: 600; color: #64748B; text-transform: uppercase; letter-spacing: 0.05em; margin-bottom: 4px; } | |
| .m-val { font-size: 0.85rem; font-weight: 500; color: #0F172A; } | |
| .status-dot { color: #16A34A; margin-right: 4px; } | |
| /* Panels */ | |
| .workspace-row { gap: 32px !important; align-items: flex-start !important; } | |
| .panel { background: #FFFFFF; border: 1px solid #E2E8F0; border-radius: 8px; padding: 24px; box-shadow: 0 1px 2px rgba(0,0,0,0.02); } | |
| /* Review Input */ | |
| .review-input textarea { border: 1px solid #CBD5E1 !important; border-radius: 6px !important; padding: 12px !important; font-size: 0.95rem !important; line-height: 1.5 !important; background: #FFFFFF !important; color: #0F172A !important; transition: border-color 0.2s; box-shadow: none !important; } | |
| .review-input textarea:focus { border-color: #2563EB !important; ring: 1px solid #2563EB !important; } | |
| .analyze-btn { background: #2563EB !important; color: #FFFFFF !important; font-weight: 500 !important; border-radius: 6px !important; margin-top: 16px !important; padding: 10px 0 !important; border: none !important; transition: background 0.2s !important; } | |
| .analyze-btn:hover { background: #1D4ED8 !important; } | |
| /* Examples */ | |
| .compact-examples .gallery { gap: 8px !important; } | |
| .compact-examples button { border: 1px solid #E2E8F0 !important; border-radius: 6px !important; padding: 8px 12px !important; font-size: 0.8rem !important; background: #F8FAFC !important; color: #475569 !important; text-align: left !important; } | |
| .compact-examples button:hover { background: #F1F5F9 !important; border-color: #CBD5E1 !important; } | |
| /* Meta Badges */ | |
| .meta-badges { display: flex; gap: 12px; margin-bottom: 24px; } | |
| .badge { font-size: 0.75rem; background: #F1F5F9; color: #475569; padding: 4px 10px; border-radius: 4px; border: 1px solid #E2E8F0; } | |
| .badge strong { color: #0F172A; font-weight: 600; margin-left: 4px; } | |
| /* Summary Vis */ | |
| .summary-vis { margin-bottom: 24px; font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace; font-size: 0.8rem; } | |
| .summary-title { font-family: 'Inter', sans-serif; font-size: 0.9rem; font-weight: 600; color: #0F172A; margin-bottom: 12px; border-bottom: 1px solid #E2E8F0; padding-bottom: 8px; } | |
| .vis-row { margin-bottom: 6px; display: flex; align-items: center; gap: 12px; color: #475569; } | |
| .vis-block { letter-spacing: -1px; font-size: 0.7rem; } | |
| .vis-block.pos { color: #16A34A; } | |
| .vis-block.neg { color: #DC2626; } | |
| .vis-block.neu { color: #64748B; } | |
| .vis-block.empty-block { visibility: hidden; width: 10px; } | |
| .summary-vis.empty { font-family: 'Inter', sans-serif; color: #64748B; font-style: italic; } | |
| /* Results Table */ | |
| .results-table { width: 100%; border-collapse: collapse; margin-bottom: 16px; font-size: 0.85rem; } | |
| .results-table th { text-align: left; padding: 8px 12px; border-bottom: 1px solid #E2E8F0; color: #64748B; font-weight: 600; font-size: 0.75rem; text-transform: uppercase; letter-spacing: 0.05em; } | |
| .results-table td { padding: 12px; border-bottom: 1px solid #F1F5F9; color: #0F172A; } | |
| .aspect-col { font-weight: 500; } | |
| .sentiment-col { display: flex; align-items: center; gap: 6px; } | |
| .sent-dot { width: 8px; height: 8px; border-radius: 50%; display: inline-block; } | |
| /* Confidence Bar */ | |
| .conf-col { min-width: 120px; } | |
| .conf-bar-bg { width: 60px; height: 6px; background: #E2E8F0; border-radius: 3px; display: inline-block; vertical-align: middle; margin-right: 8px; overflow: hidden; } | |
| .conf-bar-fill { height: 100%; background: #2563EB; border-radius: 3px; } | |
| .conf-val { font-family: ui-monospace, monospace; font-size: 0.8rem; color: #475569; } | |
| .empty-state { padding: 32px 0; text-align: center; color: #64748B; font-size: 0.9rem; border: 1px dashed #CBD5E1; border-radius: 6px; } | |
| /* Raw Output Accordion */ | |
| .raw-accordion { margin-top: 24px !important; border: 1px solid #E2E8F0 !important; border-radius: 6px !important; overflow: hidden !important; background: #FFFFFF !important; } | |
| .raw-accordion .label-wrap { padding: 12px 16px !important; font-size: 0.85rem !important; font-weight: 600 !important; color: #0F172A !important; background: #F8FAFC !important; border-bottom: 1px solid #E2E8F0 !important; } | |
| .raw-accordion .raw-json { background: #0F172A !important; padding: 16px !important; margin: 0 !important; } | |
| .raw-accordion .raw-json * { color: #F8FAFC !important; font-family: ui-monospace, SFMono-Regular, monospace !important; font-size: 0.8rem !important; } | |
| /* History Table */ | |
| .history-table { border: 1px solid #E2E8F0 !important; border-radius: 8px !important; overflow: hidden !important; margin-top: 16px !important; } | |
| .history-table th { background: #F8FAFC !important; color: #475569 !important; font-size: 0.75rem !important; text-transform: uppercase !important; font-weight: 600 !important; } | |
| /* Mobile Stacking */ | |
| @media (max-width: 768px) { | |
| .workspace-row { flex-direction: column !important; } | |
| .panel { width: 100% !important; } | |
| .header-metrics { justify-content: flex-start; margin-top: 16px; } | |
| } | |
| """ | |
| with gr.Blocks(title="Multilingual ABSA", theme=custom_theme, css=custom_css) as demo: | |
| # TOP HEADER | |
| with gr.Row(elem_classes="top-header"): | |
| with gr.Column(scale=3, elem_classes="header-title-col"): | |
| gr.Markdown(""" | |
| <div class="header-content"> | |
| <h1>Multilingual ABSA</h1> | |
| <div class="header-subtitle">Aspect-Based Sentiment Analysis for English, Hindi, and Hinglish</div> | |
| </div> | |
| """) | |
| with gr.Column(scale=2, elem_classes="header-metrics-col"): | |
| gr.Markdown(f""" | |
| <div class="header-metrics"> | |
| <div class="metric"><span class="m-label">MODE</span><span class="m-val">{MODE}</span></div> | |
| <div class="metric"><span class="m-label">LANGUAGES</span><span class="m-val">English · Hindi · Hinglish</span></div> | |
| <div class="metric"><span class="m-label">PREDICTIONS</span><span class="m-val">{get_count()}</span></div> | |
| <div class="metric"><span class="m-label">STATUS</span><span class="m-val"><span class="status-dot">●</span> Ready</span></div> | |
| </div> | |
| """) | |
| # WORKSPACE | |
| with gr.Tabs(elem_classes="main-tabs"): | |
| with gr.Tab("Workspace"): | |
| with gr.Row(elem_classes="workspace-row"): | |
| # LEFT PANEL | |
| with gr.Column(scale=1, elem_classes="panel panel-left"): | |
| gr.Markdown('<div class="section-title">Analyze Review</div><div class="section-sub">Enter a product review to identify aspects and their sentiment.</div>') | |
| inp = gr.Textbox( | |
| label="", | |
| placeholder="Example: The battery life is amazing, but the camera quality is poor.", | |
| lines=6, | |
| max_lines=15, | |
| elem_classes="review-input", | |
| show_label=False, | |
| ) | |
| btn = gr.Button("Analyze Review", variant="primary", elem_classes="analyze-btn") | |
| gr.Markdown('<div class="section-title" style="margin-top: 32px; font-size: 0.85rem;">Examples</div>') | |
| gr.Examples( | |
| examples=[ | |
| ["The battery life is amazing but the camera quality is poor."], | |
| ["बैटरी बहुत अच्छी है लेकिन कैमरा क्वालिटी खराब है।"], | |
| ["Battery life mast hai lekin camera quality bahut bekar hai."], | |
| ], | |
| inputs=inp, | |
| label="" | |
| ) | |
| # RIGHT PANEL | |
| with gr.Column(scale=1, elem_classes="panel panel-right"): | |
| gr.Markdown('<div class="section-title">Analysis Results</div>') | |
| meta_md = gr.Markdown('<div class="meta-badges"><span class="badge">Detected Language: <strong>Waiting</strong></span><span class="badge">Latency: <strong>—</strong></span></div>') | |
| summary_md = gr.Markdown('<div class="summary-vis empty">Waiting for analysis...</div>') | |
| results_html = gr.HTML('<div class="empty-state">No analysis yet.</div>') | |
| with gr.Accordion("Raw Model Output", open=False, elem_classes="raw-accordion"): | |
| json_out = gr.JSON(label="", visible=False, elem_classes="raw-json") | |
| btn.click( | |
| fn=lambda: ( | |
| '<div class="meta-badges"><span class="badge">Detected Language: <strong>Analyzing...</strong></span><span class="badge">Latency: <strong>—</strong></span></div>', | |
| '<div class="summary-vis empty">Analyzing review...</div>', | |
| '<div class="empty-state">Analyzing...</div>' | |
| ), | |
| outputs=[meta_md, summary_md, results_html] | |
| ).then( | |
| fn=on_analyze, | |
| inputs=inp, | |
| outputs=[meta_md, summary_md, results_html, json_out] | |
| ) | |
| # HISTORY TAB | |
| with gr.Tab("History"): | |
| with gr.Row(): | |
| refresh_btn = gr.Button("Refresh Data", size="sm", variant="secondary") | |
| clear_btn = gr.Button("Clear History", size="sm", variant="secondary") | |
| history_df = gr.Dataframe( | |
| headers=["Timestamp", "Language", "Aspect", "Sentiment", "Confidence", "Review Preview"], | |
| datatype=["str", "str", "str", "str", "number", "str"], | |
| value=get_history_rows(), | |
| interactive=False, | |
| wrap=True, | |
| elem_classes="history-table" | |
| ) | |
| # Now we can attach the `.then` to update history_df | |
| btn.click(fn=lambda: None).then(fn=get_history_rows, outputs=history_df) | |
| refresh_btn.click(fn=get_history_rows, outputs=history_df) | |
| clear_btn.click(fn=clear_history, outputs=[history_df]).then(fn=get_history_rows, outputs=history_df) | |
| if __name__ == "__main__": | |
| demo.launch(ssr_mode=False) | |