Spaces:
Sleeping
Sleeping
| import re | |
| import json | |
| from bertopic import BERTopic | |
| from sentence_transformers import SentenceTransformer | |
| from transformers import pipeline | |
| # IMPORTANT: must match training embedding model | |
| EMBEDDING_NAME = "paraphrase-multilingual-MiniLM-L12-v2" | |
| # Load embedding + topic model | |
| embedding_model = SentenceTransformer(EMBEDDING_NAME) | |
| topic_model = BERTopic.load("afcon_topic_model", embedding_model=embedding_model) | |
| # Load topic mapping | |
| with open("topic_mapping.json", "r") as f: | |
| topic_mapping = {int(k): v for k, v in json.load(f).items()} | |
| # Sentiment pipeline | |
| sentiment_pipe = pipeline("sentiment-analysis", model="ImaneAz/youtube-roberta-sentiment") | |
| LABEL_MAP = {"LABEL_0": "negative", "LABEL_1": "neutral", "LABEL_2": "positive"} | |
| # Same cleaning used in notebook | |
| def clean_for_topics(text: str) -> str: | |
| if not isinstance(text, str): | |
| return "" | |
| text = text.lower() | |
| text = re.sub(r"http\S+", "", text) | |
| text = re.sub(r"[^a-z0-9\s]", " ", text) | |
| text = re.sub(r"\s+", " ", text).strip() | |
| return text | |
| def analyze_text(text: str): | |
| norm = clean_for_topics(text) | |
| # topic prediction | |
| topic_id, _ = topic_model.transform([norm]) | |
| topic_id = int(topic_id[0]) | |
| topic_final = topic_mapping.get(topic_id, "Other") | |
| # sentiment prediction | |
| try: | |
| s = sentiment_pipe(text[:512])[0]["label"] | |
| sentiment = LABEL_MAP.get(s, s) | |
| except: | |
| sentiment = "UNKNOWN" | |
| return { | |
| "topic_id": topic_id, | |
| "topic_final": topic_final, | |
| "sentiment": sentiment, | |
| "clean_text": norm | |
| } | |