Spaces:
Running
Running
File size: 3,854 Bytes
b5ad127 7ea2e4f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 |
from typing import List, Optional
from pydantic import BaseModel
from transformers import pipeline
import nltk.data
# ✅ Extra: Smart Summarization Imports
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans
from nltk.tokenize import sent_tokenize
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
import os
os.environ["TRANSFORMERS_CACHE"] = "/tmp/huggingface"
# 📄 Load HuggingFace Pipelines
summarizer = pipeline("summarization", model="sshleifer/distilbart-cnn-12-6")
sentiment_analyzer = pipeline("sentiment-analysis")
# 🧠 Basic Summarization (Abstractive)
def summarize_review(text):
return summarizer(text, max_length=60, min_length=10, do_sample=False, no_repeat_ngram_size=3)[0]["summary_text"]
# 🧠 Smart Summarization (Clustered Key Sentences)
def smart_summarize(text, n_clusters=1):
"""Improved summarization using clustering on sentence embeddings"""
tokenizer = nltk.tokenize.PunktSentenceTokenizer() # ✅ Use default trained Punkt tokenizer
sentences = tokenizer.tokenize(text)
if len(sentences) <= 1:
return text
vectorizer = TfidfVectorizer(stop_words="english")
tfidf_matrix = vectorizer.fit_transform(sentences)
if len(sentences) <= n_clusters:
return " ".join(sentences)
kmeans = KMeans(n_clusters=n_clusters, random_state=42)
kmeans.fit(tfidf_matrix)
avg = []
for i in range(n_clusters):
idx = np.where(kmeans.labels_ == i)[0]
if len(idx) == 0:
continue
avg_vector = tfidf_matrix[idx].mean(axis=0).A1.reshape(1, -1) # Convert np.matrix to ndarray
sim = cosine_similarity(avg_vector, tfidf_matrix[idx])
most_representative_idx = idx[np.argmax(sim)]
avg.append(sentences[most_representative_idx])
return " ".join(sorted(avg, key=sentences.index))
# 📊 Sentiment Detection
def analyze_sentiment(text):
result = sentiment_analyzer(text)[0]
label = result["label"]
score = result["score"]
if "star" in label:
stars = int(label[0])
if stars <= 2:
label = "NEGATIVE"
elif stars == 3:
label = "NEUTRAL"
else:
label = "POSITIVE"
return {
"label": label,
"score": score
}
# 🔥 Emotion Detection (heuristic-based)
def detect_emotion(text):
text_lower = text.lower()
if "angry" in text_lower or "hate" in text_lower:
return "anger"
elif "happy" in text_lower or "love" in text_lower:
return "joy"
elif "sad" in text_lower or "disappointed" in text_lower:
return "sadness"
elif "confused" in text_lower or "unclear" in text_lower:
return "confusion"
else:
return "neutral"
# 🧩 Aspect-Based Sentiment (mock)
def extract_aspect_sentiment(text, aspects: list):
results = {}
text_lower = text.lower()
for asp in aspects:
label = "positive" if asp in text_lower and "not" not in text_lower else "neutral"
results[asp] = {
"label": label,
"confidence": 0.85
}
return results
# ✅ Pydantic Schemas for FastAPI
class ReviewInput(BaseModel):
text: str
model: str = "distilbert-base-uncased-finetuned-sst-2-english"
industry: str = "Generic"
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: str = "Generic"
aspects: bool = False
product_category: Optional[str] = None
device: Optional[str] = None
class TranslationInput(BaseModel):
text: str
target_lang: str = "fr"
class ChatInput(BaseModel):
question: str
context: str |