Spaces:

Hasitha16
/

churnsight-ai

Running

File size: 9,108 Bytes

from fastapi import FastAPI, Request, Header, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse
from fastapi.openapi.utils import get_openapi
from fastapi.openapi.docs import get_swagger_ui_html
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.get("/docs", include_in_schema=False)
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",
    )

@app.get("/", response_class=HTMLResponse)
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>
    """

# --- Models ---
class ReviewInput(BaseModel):
    text: str
    model: str = "distilbert-base-uncased-finetuned-sst-2-english"
    industry: str = "Generic"
    aspects: bool = False
    follow_up: str = None
    product_category: str = None
    device: 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"

# --- Auth & Logging ---
VALID_API_KEY = "my-secret-key"
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

# --- Load Models Once ---
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")
}

# --- Analyze (Bulk) ---
@app.post("/bulk/")
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": data.product_category[i] if data.product_category else None,
            "device": data.device[i] if data.device else None,
            "industry": data.industry[i] if data.industry else None,
        }
        results.append(result)

    return {"results": results}

@app.post("/analyze/")
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": data.product_category,
        "device": data.device,
        "industry": data.industry
    }
# --- Translate ---
@app.post("/translate/")
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"]}

# --- LLM Agent Chat ---
@app.post("/chat/")
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()

# --- Custom OpenAPI ---
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