from flask import Flask, request, jsonify from transformers import pipeline import logging import os import re logging.basicConfig(level=logging.INFO, format="%(asctime)s [BERT] %(message)s") app = Flask(__name__) ROMAN_URDU_NORMALIZATION = { r'\bbohat\b': 'bohat', r'\bbahut\b': 'bohat', r'\bbht\b': 'bohat', r'\bacha\b': 'acha', r'\baachi\b': 'acha', r'\bachy\b': 'acha', r'\bachaa\b': 'acha', r'\bbura\b': 'bura', r'\bburra\b': 'bura', r'\bmza\b': 'maza', r'\bmaaza\b': 'maza', r'\bnahi\b': 'nahi', r'\bnhi\b': 'nahi', r'\bhai\b': 'hai', r'\bhy\b': 'hai', r'\bkuch\b': 'kuch', r'\bkch\b': 'kuch', r'\bwr\b': 'aur', } def preprocess_text(text: str) -> str: text = str(text).strip().lower() text = re.sub(r'http\S+|www\.\S+', '', text) text = re.sub(r'[^\w\s.,!?]', ' ', text) for pattern, replacement in ROMAN_URDU_NORMALIZATION.items(): text = re.sub(pattern, replacement, text) return re.sub(r'\s+', ' ', text).strip() HF_MODEL = os.environ.get("MODEL_ID", "campusevents/Bert_sentiment_analysis") logging.info(f"Loading model: {HF_MODEL} …") sentiment_pipeline = pipeline( "text-classification", model=HF_MODEL, tokenizer=HF_MODEL, truncation=True, max_length=128, ) logging.info("Model loaded.") LABEL_SCORE = { "Positive": 1.0, "Neutral": 0.0, "Negative": -1.0, } def label_to_score(label: str) -> float: return LABEL_SCORE.get(label, 0.0) def score_to_label(score: float) -> str: if score >= 0.2: return "Positive" if score <= -0.2: return "Negative" return "Neutral" def rating_description(avg: float) -> str: if avg >= 4.5: return "excellent" if avg >= 3.5: return "good" if avg >= 2.5: return "average" if avg >= 1.5: return "below average" return "poor" def build_summary( sentiment_label: str, avg_rating: float, total: int, positive_count: int, negative_count: int, neutral_count: int, highlights: list, concerns: list, ) -> str: rating_word = rating_description(avg_rating) star_display = f"{avg_rating:.1f} out of 5 stars" lines = [] lines.append( f"A total of {total} student{'s' if total != 1 else ''} submitted feedback for this event. " f"The overall reception was {sentiment_label.lower()}, with an average rating of {star_display} — " f"which is considered {rating_word}." ) parts = [] if positive_count: parts.append(f"{positive_count} response{'s were' if positive_count != 1 else ' was'} positive") if neutral_count: parts.append(f"{neutral_count} {'were' if neutral_count != 1 else 'was'} neutral") if negative_count: parts.append(f"{negative_count} {'were' if negative_count != 1 else 'was'} negative") if parts: lines.append("Sentiment breakdown: " + ", ".join(parts) + ".") if highlights: joined = "; ".join(f'"{h}"' for h in highlights) lines.append(f"Students appreciated aspects of the event, with comments such as: {joined}.") if concerns: joined = "; ".join(f'"{c}"' for c in concerns) lines.append(f"Areas of concern raised by attendees included: {joined}.") if sentiment_label == "Positive": lines.append("Overall, the event was well-received and attendees expressed satisfaction.") elif sentiment_label == "Negative": lines.append("Overall, the event did not meet attendee expectations and improvements are recommended.") else: lines.append( "Overall, attendee responses were mixed. " "Addressing the concerns raised may help improve future events." ) return " ".join(lines) def truncate_for_snippet(text: str, max_words: int = 12) -> str: words = text.split() return text if len(words) <= max_words else " ".join(words[:max_words]) + "…" @app.route("/health", methods=["GET"]) def health(): return jsonify({"status": "ok", "model": HF_MODEL}) @app.route("/sentiment", methods=["POST"]) def sentiment(): data = request.get_json(force=True) texts = data.get("texts", []) if not texts or not isinstance(texts, list): return jsonify({"error": "texts must be a non-empty list"}), 400 safe_texts = [preprocess_text(t)[:512] for t in texts if t] if not safe_texts: return jsonify({"averageScore": 0.0, "overallLabel": "Neutral", "details": []}) try: results = sentiment_pipeline(safe_texts) except Exception as e: logging.error(f"Pipeline error: {e}") return jsonify({"error": str(e)}), 500 details = [] total_score = 0.0 for original, processed, result in zip(texts, safe_texts, results): label = result["label"] # "Positive" | "Neutral" | "Negative" score = label_to_score(label) total_score += score display = str(original) details.append({ "text": display[:80] + "…" if len(display) > 80 else display, "bertLabel": label, "confidence": round(result["score"], 4), "score": score, "label": label, }) avg = total_score / len(safe_texts) return jsonify({ "averageScore": round(avg, 4), "overallLabel": score_to_label(avg), "details": details, }) @app.route("/summarize", methods=["POST"]) def summarize(): data = request.get_json(force=True) comments = data.get("comments", []) ratings = data.get("ratings", []) if not isinstance(comments, list) or not isinstance(ratings, list): return jsonify({"error": "comments and ratings must be arrays"}), 400 total = len(ratings) if total == 0: return jsonify({ "summaryText": "No feedback was submitted for this event.", "overallSentiment": "Neutral", "averageRating": 0, "ratingWord": "N/A", }) valid_ratings = [r for r in ratings if isinstance(r, (int, float)) and 1 <= r <= 5] avg_rating = sum(valid_ratings) / len(valid_ratings) if valid_ratings else 0.0 non_empty = [ (i, preprocess_text(str(c))[:512]) for i, c in enumerate(comments) if c and str(c).strip() ] positive_count = negative_count = neutral_count = 0 highlights: list = [] concerns: list = [] if non_empty: indices, texts_proc = zip(*non_empty) try: results = sentiment_pipeline(list(texts_proc)) except Exception as e: logging.error(f"Pipeline error in /summarize: {e}") return jsonify({"error": str(e)}), 500 scores = [] for (orig_idx, text_proc), result in zip(non_empty, results): label = result["label"] score = label_to_score(label) scores.append(score) # Use original (raw) text for snippets raw_text = str(comments[orig_idx]) if label == "Positive": positive_count += 1 if len(highlights) < 3: highlights.append(truncate_for_snippet(raw_text)) elif label == "Negative": negative_count += 1 if len(concerns) < 3: concerns.append(truncate_for_snippet(raw_text)) else: neutral_count += 1 neutral_count += total - len(non_empty) avg_score = sum(scores) / len(scores) overall_label = score_to_label(avg_score) else: neutral_count = total if avg_rating >= 3.5: overall_label = "Positive"; positive_count = total; neutral_count = 0 elif avg_rating < 2.5: overall_label = "Negative"; negative_count = total; neutral_count = 0 else: overall_label = "Neutral" summary = build_summary( sentiment_label=overall_label, avg_rating=avg_rating, total=total, positive_count=positive_count, negative_count=negative_count, neutral_count=neutral_count, highlights=highlights, concerns=concerns, ) return jsonify({ "summaryText": summary, "overallSentiment": overall_label, "averageRating": round(avg_rating, 2), "ratingWord": rating_description(avg_rating), }) if __name__ == "__main__": port = int(os.environ.get("BERT_PORT", 5001)) logging.info(f"Starting on port {port}") app.run(host="0.0.0.0", port=port, debug=False)