| 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"] |
| 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) |
| |
| 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) |
