Spaces:
Running
Running
| from fastapi import FastAPI, Request, Header, HTTPException, UploadFile, File | |
| from fastapi.responses import HTMLResponse, JSONResponse | |
| from fastapi.openapi.utils import get_openapi | |
| from fastapi.openapi.docs import get_swagger_ui_html | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from pydantic import BaseModel | |
| from transformers import pipeline | |
| from io import StringIO | |
| import os, csv, logging | |
| from openai import OpenAI | |
| from model import summarize_review, smart_summarize | |
| from typing import Optional | |
| app = FastAPI( | |
| title="🧠 NeuroPulse AI", | |
| description="Multilingual GenAI for smarter feedback — summarization, sentiment, emotion, aspects, Q&A and tags.", | |
| version="2025.1.0", | |
| openapi_url="/openapi.json", | |
| docs_url=None, | |
| redoc_url="/redoc" | |
| ) | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| def custom_swagger_ui(): | |
| return get_swagger_ui_html( | |
| openapi_url=app.openapi_url, | |
| title="🧠 Swagger UI - NeuroPulse AI", | |
| swagger_favicon_url="https://cdn-icons-png.flaticon.com/512/3794/3794616.png", | |
| swagger_js_url="https://cdn.jsdelivr.net/npm/[email protected]/swagger-ui-bundle.js", | |
| swagger_css_url="https://cdn.jsdelivr.net/npm/[email protected]/swagger-ui.css", | |
| ) | |
| def root(): | |
| return """ | |
| <html> | |
| <head> | |
| <title>NeuroPulse AI</title> | |
| <style> | |
| body { | |
| font-family: 'Segoe UI', sans-serif; | |
| background: linear-gradient(135deg, #f0f4ff, #fef3c7); | |
| margin: 0; | |
| padding: 60px; | |
| text-align: center; | |
| color: #1f2937; | |
| } | |
| .container { | |
| background: white; | |
| padding: 40px; | |
| border-radius: 16px; | |
| max-width: 800px; | |
| margin: auto; | |
| box-shadow: 0 10px 30px rgba(0,0,0,0.08); | |
| animation: fadeIn 1s ease-in-out; | |
| } | |
| @keyframes fadeIn { | |
| from {opacity: 0; transform: translateY(20px);} | |
| to {opacity: 1; transform: translateY(0);} | |
| } | |
| h1 { | |
| font-size: 36px; | |
| margin-bottom: 12px; | |
| color: #4f46e5; | |
| } | |
| p { | |
| font-size: 18px; | |
| margin-bottom: 32px; | |
| } | |
| .btn { | |
| display: inline-block; | |
| margin: 8px; | |
| padding: 14px 24px; | |
| border-radius: 8px; | |
| font-weight: 600; | |
| color: white; | |
| text-decoration: none; | |
| background: linear-gradient(90deg, #4f46e5, #6366f1); | |
| transition: all 0.3s ease; | |
| } | |
| .btn:hover { | |
| transform: translateY(-2px); | |
| box-shadow: 0 4px 12px rgba(0,0,0,0.1); | |
| } | |
| .btn.red { | |
| background: linear-gradient(90deg, #dc2626, #ef4444); | |
| } | |
| </style> | |
| </head> | |
| <body> | |
| <div class="container"> | |
| <h1>🧠 Welcome to <strong>NeuroPulse AI</strong></h1> | |
| <p>Smarter AI feedback analysis — Summarization, Sentiment, Emotion, Aspects, LLM Q&A, and Metadata Tags.</p> | |
| <a class="btn" href="/docs">📘 Swagger UI</a> | |
| <a class="btn red" href="/redoc">📕 ReDoc</a> | |
| </div> | |
| </body> | |
| </html> | |
| """ | |
| class ReviewInput(BaseModel): | |
| text: str | |
| model: str = "distilbert-base-uncased-finetuned-sst-2-english" | |
| industry: Optional[str] = None | |
| aspects: bool = False | |
| follow_up: Optional[str] = None | |
| product_category: Optional[str] = None | |
| device: Optional[str] = None | |
| class BulkReviewInput(BaseModel): | |
| reviews: list[str] | |
| model: str = "distilbert-base-uncased-finetuned-sst-2-english" | |
| industry: Optional[list[str]] = None | |
| aspects: bool = False | |
| product_category: Optional[list[str]] = None | |
| device: Optional[list[str]] = None | |
| class ChatInput(BaseModel): | |
| question: str | |
| context: str | |
| class TranslationInput(BaseModel): | |
| text: str | |
| target_lang: str = "fr" | |
| VALID_API_KEY = "my-secret-key" | |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") | |
| summarizer = pipeline("summarization", model="sshleifer/distilbart-cnn-12-6") | |
| emotion_model = pipeline("text-classification", model="j-hartmann/emotion-english-distilroberta-base", top_k=1) | |
| sentiment_pipelines = { | |
| "distilbert-base-uncased-finetuned-sst-2-english": pipeline("sentiment-analysis", model="distilbert-base-uncased-finetuned-sst-2-english"), | |
| "nlptown/bert-base-multilingual-uncased-sentiment": pipeline("sentiment-analysis", model="nlptown/bert-base-multilingual-uncased-sentiment") | |
| } | |
| def auto_fill(value: Optional[str], default: str = "Generic") -> str: | |
| if not value or value.strip().lower() == "auto-detect": | |
| return default | |
| return value | |
| async def bulk(data: BulkReviewInput, x_api_key: str = Header(None)): | |
| if x_api_key != VALID_API_KEY: | |
| raise HTTPException(status_code=401, detail="Invalid or missing API key") | |
| sentiment_pipeline = sentiment_pipelines[data.model] | |
| summaries = summarizer(data.reviews, max_length=80, min_length=20, truncation=True) | |
| sentiments = sentiment_pipeline(data.reviews) | |
| emotions = emotion_model(data.reviews) | |
| results = [] | |
| for i, review in enumerate(data.reviews): | |
| label = sentiments[i]["label"] | |
| if "star" in label: | |
| stars = int(label[0]) | |
| label = "NEGATIVE" if stars <= 2 else "NEUTRAL" if stars == 3 else "POSITIVE" | |
| result = { | |
| "review": review, | |
| "summary": summaries[i]["summary_text"], | |
| "sentiment": label, | |
| "emotion": emotions[i][0]["label"], | |
| "aspects": [], | |
| "product_category": auto_fill(data.product_category[i]) if data.product_category else None, | |
| "device": auto_fill(data.device[i], "Web") if data.device else None, | |
| "industry": auto_fill(data.industry[i]) if data.industry else None, | |
| } | |
| results.append(result) | |
| return {"results": results} | |
| async def analyze(request: Request, data: ReviewInput, x_api_key: str = Header(None), download: str = None): | |
| if x_api_key != VALID_API_KEY: | |
| raise HTTPException(status_code=401, detail="Invalid or missing API key") | |
| sentiment_pipeline = sentiment_pipelines.get(data.model) | |
| summary = smart_summarize(data.text) if request.query_params.get("smart") == "1" else summarize_review(data.text) | |
| sentiment = sentiment_pipeline(data.text)[0] | |
| label = sentiment["label"] | |
| if "star" in label: | |
| stars = int(label[0]) | |
| label = "NEGATIVE" if stars <= 2 else "NEUTRAL" if stars == 3 else "POSITIVE" | |
| emotion = emotion_model(data.text)[0][0]["label"] | |
| aspects_list = [] | |
| if data.aspects: | |
| for asp in ["battery", "price", "camera"]: | |
| if asp in data.text.lower(): | |
| asp_result = sentiment_pipeline(asp + " " + data.text)[0] | |
| aspects_list.append({ | |
| "aspect": asp, | |
| "sentiment": asp_result["label"], | |
| "score": asp_result["score"] | |
| }) | |
| follow_up_response = chat_llm(data.follow_up, data.text) if data.follow_up else None | |
| return { | |
| "summary": summary, | |
| "sentiment": {"label": label, "score": sentiment["score"]}, | |
| "emotion": emotion, | |
| "aspects": aspects_list, | |
| "follow_up": follow_up_response, | |
| "product_category": auto_fill(data.product_category), | |
| "device": auto_fill(data.device, "Web"), | |
| "industry": auto_fill(data.industry) | |
| } | |
| async def translate(data: TranslationInput): | |
| translator = pipeline("translation", model=f"Helsinki-NLP/opus-mt-en-{data.target_lang}") | |
| return {"translated_text": translator(data.text)[0]["translation_text"]} | |
| async def chat(input: ChatInput, x_api_key: str = Header(None)): | |
| if x_api_key != VALID_API_KEY: | |
| raise HTTPException(status_code=401, detail="Invalid or missing API key") | |
| return {"response": chat_llm(input.question, input.context)} | |
| def chat_llm(question, context): | |
| client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) | |
| res = client.chat.completions.create( | |
| model="gpt-3.5-turbo", | |
| messages=[ | |
| {"role": "system", "content": "You are a helpful AI review analyst."}, | |
| {"role": "user", "content": f"Context: {context}\nQuestion: {question}"} | |
| ] | |
| ) | |
| return res.choices[0].message.content.strip() | |
| def custom_openapi(): | |
| if app.openapi_schema: | |
| return app.openapi_schema | |
| openapi_schema = get_openapi( | |
| title=app.title, | |
| version=app.version, | |
| description=""" | |
| <b><span style='color:#4f46e5'>NeuroPulse AI</span></b> · Smart GenAI Feedback Engine<br> | |
| Summarize reviews, detect sentiment/emotion, extract aspects, tag metadata, and ask GPT follow-ups. | |
| """, | |
| routes=app.routes | |
| ) | |
| openapi_schema["openapi"] = "3.0.0" | |
| app.openapi_schema = openapi_schema | |
| return app.openapi_schema | |
| app.openapi = custom_openapi | |