๐ง Welcome to NeuroPulse AI
Smarter AI feedback analysis โ Summarization, Sentiment, Emotion, Aspects, LLM Q&A, and Metadata Tags.
๐ Swagger UI
๐ ReDoc
"""
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
@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": 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}
@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": auto_fill(data.product_category),
"device": auto_fill(data.device, "Web"),
"industry": auto_fill(data.industry)
}
@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"]}
@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()
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title=app.title,
version=app.version,
description="""