Upload model.py

model.py

@@ -0,0 +1,143 @@
+import os
+os.environ["TRANSFORMERS_CACHE"] = "/tmp/hf-cache"
+os.environ["HF_HOME"] = "/tmp/hf-home"
+
+import nltk
+nltk.download("punkt", download_dir="/tmp/nltk_data")
+
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.cluster import KMeans
+from sklearn.metrics.pairwise import cosine_similarity
+from nltk.tokenize import sent_tokenize
+from transformers import pipeline
+import numpy as np
+import logging
+
+# === Pipelines ===
+summarizer = pipeline("summarization", model="sshleifer/distilbart-cnn-12-6")
+qa_pipeline = pipeline("question-answering", model="distilbert-base-cased-distilled-squad")
+emotion_pipeline = pipeline("text-classification", model="bhadresh-savani/distilbert-base-uncased-emotion", top_k=1)
+
+# === Brief Summarization ===
+def summarize_review(text, max_len=80, min_len=20):
+    try:
+        return summarizer(text, max_length=max_len, min_length=min_len, do_sample=False)[0]["summary_text"]
+    except Exception as e:
+        logging.warning(f"Summarization fallback used: {e}")
+        return text
+
+# === Smart Summarization with Clustering ===
+def smart_summarize(text, n_clusters=1):
+    try:
+        sentences = sent_tokenize(text)
+        if len(sentences) <= 1:
+            return text
+        tfidf = TfidfVectorizer(stop_words="english")
+        tfidf_matrix = tfidf.fit_transform(sentences)
+        if len(sentences) <= n_clusters:
+            return " ".join(sentences)
+        kmeans = KMeans(n_clusters=n_clusters, random_state=42).fit(tfidf_matrix)
+        summary_sentences = []
+        for i in range(n_clusters):
+            idx = np.where(kmeans.labels_ == i)[0]
+            if not len(idx):
+                continue
+            avg_vector = np.asarray(tfidf_matrix[idx].mean(axis=0))
+            sim = cosine_similarity(avg_vector, tfidf_matrix[idx].toarray())
+            most_representative = sentences[idx[np.argmax(sim)]]
+            summary_sentences.append(most_representative)
+        return " ".join(sorted(summary_sentences, key=sentences.index))
+    except Exception as e:
+        logging.error(f"Smart summarize error: {e}")
+        return text
+
+# === Emotion Detection ===
+def detect_emotion(text):
+    try:
+        result = emotion_pipeline(text)[0]
+        return result["label"]
+    except Exception as e:
+        logging.warning(f"Emotion detection failed: {e}")
+        return "neutral"
+
+# === Follow-up Q&A (Flexible for list or str) ===
+def answer_followup(text, question, verbosity="brief"):
+    try:
+        if isinstance(question, list):
+            answers = []
+            for q in question:
+                response = qa_pipeline({"question": q, "context": text})
+                ans = response.get("answer", "")
+                if verbosity.lower() == "detailed":
+                    answers.append(f"**{q}** → {ans}")
+                else:
+                    answers.append(ans)
+            return answers
+        else:
+            response = qa_pipeline({"question": question, "context": text})
+            ans = response.get("answer", "")
+            return f"**{question}** → {ans}" if verbosity.lower() == "detailed" else ans
+    except Exception as e:
+        logging.warning(f"Follow-up error: {e}")
+        return "Sorry, I couldn't generate a follow-up answer."
+
+# === Fast follow-up (used for direct /followup route) ===
+def answer_only(text, question):
+    try:
+        if not question:
+            return "No question provided."
+        return qa_pipeline({"question": question, "context": text}).get("answer", "No answer found.")
+    except Exception as e:
+        logging.warning(f"Answer-only failed: {e}")
+        return "Q&A failed."
+
+# === Optional Explanation Generator ===
+def generate_explanation(text):
+    try:
+        explanation = summarizer(text, max_length=60, min_length=20, do_sample=False)[0]["summary_text"]
+        return f"🧠 This review can be explained as: {explanation}"
+    except Exception as e:
+        logging.warning(f"Explanation failed: {e}")
+        return "⚠️ Explanation could not be generated."
+
+# === Industry Detector ===
+def detect_industry(text):
+    text = text.lower()
+    if any(k in text for k in ["doctor", "hospital", "health", "pill", "med"]):
+        return "Healthcare"
+    if any(k in text for k in ["flight", "hotel", "trip", "booking"]):
+        return "Travel"
+    if any(k in text for k in ["bank", "loan", "credit", "payment"]):
+        return "Banking"
+    if any(k in text for k in ["gym", "trainer", "fitness", "workout"]):
+        return "Fitness"
+    if any(k in text for k in ["movie", "series", "stream", "video"]):
+        return "Entertainment"
+    if any(k in text for k in ["game", "gaming", "console"]):
+        return "Gaming"
+    if any(k in text for k in ["food", "delivery", "restaurant", "order"]):
+        return "Food Delivery"
+    if any(k in text for k in ["school", "university", "teacher", "course"]):
+        return "Education"
+    if any(k in text for k in ["insurance", "policy", "claim"]):
+        return "Insurance"
+    if any(k in text for k in ["property", "rent", "apartment", "house"]):
+        return "Real Estate"
+    if any(k in text for k in ["shop", "buy", "product", "phone", "amazon", "flipkart"]):
+        return "E-commerce"
+    return "Generic"
+
+# === Product Category Detector ===
+def detect_product_category(text):
+    text = text.lower()
+    if any(k in text for k in ["mobile", "smartphone", "iphone", "samsung", "phone"]):
+        return "Mobile Devices"
+    if any(k in text for k in ["laptop", "macbook", "notebook", "chromebook"]):
+        return "Laptops"
+    if any(k in text for k in ["tv", "refrigerator", "microwave", "washer"]):
+        return "Home Appliances"
+    if any(k in text for k in ["watch", "band", "fitbit", "wearable"]):
+        return "Wearables"
+    if any(k in text for k in ["app", "portal", "site", "website"]):
+        return "Web App"
+    return "General"