Spaces:
Running
Running
| """ | |
| sentiment.py — Análisis de sentimiento via HuggingFace Inference API. | |
| Usa el modelo pysentimiento/robertuito-sentiment-analysis, entrenado | |
| específicamente para español rioplatense y de Twitter. Corre en la nube, | |
| sin modelos locales. | |
| Salida por texto: {"label": "positive"|"negative"|"neutral", "score": float} | |
| El modelo devuelve etiquetas POS/NEG/NEU que se normalizan a positive/negative/neutral. | |
| """ | |
| import logging | |
| import re | |
| import time | |
| import requests | |
| from supabase import create_client | |
| import sys | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).parent.parent)) | |
| from config import ( | |
| HUGGINGFACE_API_KEY, | |
| HF_API_URL_SENTIMENT, | |
| HF_BATCH_SIZE, | |
| HF_RETRY_DELAY, | |
| HF_RATELIMIT_DELAY, | |
| SCRAPING_MAX_RETRIES, | |
| SUPABASE_URL, | |
| SUPABASE_KEY, | |
| SUPABASE_SERVICE_KEY, | |
| LOG_FORMAT, | |
| LOG_DATE_FORMAT, | |
| LOG_LEVEL, | |
| ) | |
| from processing.sanitizer import limpiar_texto_para_nlp | |
| logging.basicConfig(format=LOG_FORMAT, datefmt=LOG_DATE_FORMAT) | |
| logger = logging.getLogger("sentiment") | |
| logger.setLevel(getattr(logging, LOG_LEVEL, logging.INFO)) | |
| # Etiqueta vacía que se devuelve cuando un texto falla | |
| _VACIO: dict = {"label": "neutral", "score": 0.0} | |
| # Robertuito usa POS/NEG/NEU — normalizar a los valores esperados por el resto del sistema | |
| _LABEL_MAP: dict[str, str] = { | |
| "pos": "positive", | |
| "neg": "negative", | |
| "neu": "neutral", | |
| # Por si el modelo cambia a formato largo: | |
| "positive": "positive", | |
| "negative": "negative", | |
| "neutral": "neutral", | |
| } | |
| # ── Diccionario de preprocesamiento para español rioplatense/político ────────── | |
| # Términos que el modelo tiende a clasificar mal por su forma superficial | |
| # (palabras que suenan positivas pero son insultos, o negativas pero son neutras). | |
| # Reemplazamos por paráfrasis más estándar que el modelo interpreta correctamente. | |
| _PREPROCESO_RIOPLATENSE: dict[str, str] = { | |
| # Insultos que suenan neutros o positivos para el modelo | |
| "la casta": "los políticos corruptos", | |
| "casta política": "clase política corrupta", | |
| "planero": "receptor de subsidios estatales", | |
| "planera": "receptora de subsidios estatales", | |
| "planeros": "receptores de subsidios estatales", | |
| "ñoqui": "empleado innecesario del estado", | |
| "ñoquis": "empleados innecesarios del estado", | |
| "trucho": "falso o fraudulento", | |
| "trucha": "falsa o fraudulenta", | |
| "truchos": "falsos o fraudulentos", | |
| "curro": "negocio corrupto", | |
| "curros": "negocios corruptos", | |
| "currar": "estafar o robar", | |
| "gorila": "antiperonista extremo", | |
| "gorilas": "antiperonistas extremos", | |
| "zurdo": "persona de izquierda radical", | |
| "zurda": "persona de izquierda radical", | |
| "zurdos": "personas de izquierda radical", | |
| "bardear": "insultar o agredir verbalmente", | |
| "bardeando": "insultando", | |
| "bardeó": "insultó", | |
| "forro": "persona despreciable", | |
| "pelotudo": "persona estúpida", | |
| "pelotuda": "persona estúpida", | |
| "cagón": "cobarde", | |
| "cagada": "error grave o fracaso", | |
| "chorro": "ladrón", | |
| "chorros": "ladrones", | |
| # Términos políticos con carga que el modelo puede malinterpretar | |
| "mileista": "seguidor de Milei", | |
| "mileísmo": "movimiento político de Milei", | |
| "kirchnerista": "seguidor del kirchnerismo", | |
| "kirchnerismo": "movimiento político kirchnerista", | |
| "libertario": "seguidor del partido La Libertad Avanza", | |
| "libertarios": "seguidores del partido La Libertad Avanza", | |
| "nac&pop": "nacional y popular de orientación peronista", | |
| "nac and pop": "nacional y popular de orientación peronista", | |
| # Ironía política argentina frecuente en replies | |
| "viva la libertad": "apoyo al gobierno actual", | |
| "carajo": "expresión de enojo", | |
| } | |
| def preprocesar_para_sentimiento(texto: str) -> str: | |
| """ | |
| Preprocesa texto en español rioplatense antes de enviarlo al modelo. | |
| Reemplaza lunfardo y jerga política que el modelo puede clasificar mal. | |
| No modifica la polaridad del texto — traduce a términos estándar. | |
| """ | |
| if not texto: | |
| return texto | |
| texto_proc = texto | |
| for termino, reemplazo in _PREPROCESO_RIOPLATENSE.items(): | |
| pattern = re.compile(re.escape(termino), re.IGNORECASE) | |
| texto_proc = pattern.sub(reemplazo, texto_proc) | |
| return texto_proc | |
| def _headers() -> dict: | |
| """Construye los headers HTTP para la API de HuggingFace.""" | |
| headers = {"Content-Type": "application/json"} | |
| if HUGGINGFACE_API_KEY: | |
| headers["Authorization"] = f"Bearer {HUGGINGFACE_API_KEY}" | |
| return headers | |
| def _fetch_sentiment_batch(textos: list[str]) -> list[dict]: | |
| """ | |
| Envía un batch de textos a HuggingFace y devuelve el sentimiento de cada uno. | |
| Maneja reintentos ante 503 (modelo cargando) y 429 (rate limit). | |
| Devuelve lista de dicts vacíos si todos los intentos fallan. | |
| """ | |
| payload = {"inputs": textos, "options": {"wait_for_model": True}} | |
| for intento in range(1, SCRAPING_MAX_RETRIES + 1): | |
| try: | |
| resp = requests.post( | |
| HF_API_URL_SENTIMENT, | |
| headers=_headers(), | |
| json=payload, | |
| timeout=60, | |
| ) | |
| if resp.status_code == 200: | |
| raw = resp.json() | |
| resultados = [] | |
| for item in raw: | |
| # La API devuelve lista de candidatos por texto — tomamos el mayor score | |
| if isinstance(item, list): | |
| mejor = max(item, key=lambda x: x.get("score", 0)) | |
| label_raw = mejor.get("label", "neutral").lower() | |
| label = _LABEL_MAP.get(label_raw, "neutral") | |
| resultados.append({ | |
| "label": label, | |
| "score": round(mejor.get("score", 0.0), 4), | |
| }) | |
| else: | |
| resultados.append(_VACIO.copy()) | |
| return resultados | |
| elif resp.status_code == 503: | |
| logger.warning( | |
| f"[sentiment] 503 modelo cargando (intento {intento}/" | |
| f"{SCRAPING_MAX_RETRIES}) — esperando {HF_RETRY_DELAY}s" | |
| ) | |
| time.sleep(HF_RETRY_DELAY) | |
| elif resp.status_code == 429: | |
| logger.warning(f"[sentiment] 429 rate limit — esperando {HF_RATELIMIT_DELAY}s") | |
| time.sleep(HF_RATELIMIT_DELAY) | |
| else: | |
| logger.error(f"[sentiment] error {resp.status_code}: {resp.text[:200]}") | |
| return [_VACIO.copy() for _ in textos] | |
| except requests.exceptions.Timeout: | |
| logger.warning(f"[sentiment] timeout en intento {intento}/{SCRAPING_MAX_RETRIES}") | |
| if intento < SCRAPING_MAX_RETRIES: | |
| time.sleep(HF_RETRY_DELAY) | |
| except Exception as e: | |
| logger.error(f"[sentiment] error inesperado: {e}") | |
| return [_VACIO.copy() for _ in textos] | |
| return [_VACIO.copy() for _ in textos] | |
| def analyze_sentiment(texts: list[str]) -> list[dict]: | |
| """ | |
| Analiza el sentimiento de una lista de textos. | |
| Procesa en batches de HF_BATCH_SIZE. Textos vacíos devuelven neutral/0.0. | |
| Parámetros: | |
| texts: lista de strings a analizar | |
| Devuelve: | |
| Lista de dicts {"label": str, "score": float} en el mismo orden que la entrada. | |
| """ | |
| if not texts: | |
| return [] | |
| resultados: list[dict] = [] | |
| for i in range(0, len(texts), HF_BATCH_SIZE): | |
| batch = texts[i: i + HF_BATCH_SIZE] | |
| textos_validos_idx = [j for j, t in enumerate(batch) if t and t.strip()] | |
| textos_validos = [batch[j] for j in textos_validos_idx] | |
| batch_resultado = [_VACIO.copy() for _ in batch] | |
| if textos_validos: | |
| sentimientos = _fetch_sentiment_batch(textos_validos) | |
| for pos, sent in zip(textos_validos_idx, sentimientos): | |
| batch_resultado[pos] = sent | |
| resultados.extend(batch_resultado) | |
| logger.info( | |
| f"[sentiment] batch {i // HF_BATCH_SIZE + 1}: " | |
| f"{len(textos_validos)} textos analizados" | |
| ) | |
| return resultados | |
| def calcular_sentimiento_replies_por_actor(actor_ids: list[str]) -> dict[str, float]: | |
| """ | |
| Calcula el sentimiento promedio de los replies recibidos por cada actor. | |
| Lee posts con tipo_interaccion='reply_recibido' de Supabase, aplica | |
| preprocesamiento rioplatense y analiza con robertuito via HF API. | |
| Fórmula del score numérico: | |
| positive → +confidence_score | |
| negative → −confidence_score | |
| neutral → 0.0 | |
| Promedio sobre todos los replies del actor. | |
| Parámetros: | |
| actor_ids: lista de actor_ids a procesar | |
| Devuelve: | |
| Dict {actor_id: sentimiento_promedio} con valores entre −1 y +1. | |
| Solo incluye actores que tienen replies disponibles. | |
| """ | |
| if not actor_ids or not SUPABASE_URL or not SUPABASE_KEY: | |
| return {} | |
| try: | |
| client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) | |
| resultados: dict[str, float] = {} | |
| for actor_id in actor_ids: | |
| resp = ( | |
| client.table("posts") | |
| .select("titulo") | |
| .eq("actor_id", actor_id) | |
| .eq("tipo_interaccion", "reply_recibido") | |
| .limit(500) | |
| .execute() | |
| ) | |
| posts = resp.data or [] | |
| if not posts: | |
| continue | |
| textos = [ | |
| preprocesar_para_sentimiento(p.get("titulo", "") or "") | |
| for p in posts | |
| ] | |
| textos = [t for t in textos if t.strip()] | |
| if not textos: | |
| continue | |
| sentimientos = analyze_sentiment(textos) | |
| scores_numericos = [] | |
| for s in sentimientos: | |
| label = s.get("label", "neutral") | |
| score = s.get("score", 0.0) | |
| if label == "positive": | |
| scores_numericos.append(score) | |
| elif label == "negative": | |
| scores_numericos.append(-score) | |
| else: | |
| scores_numericos.append(0.0) | |
| if scores_numericos: | |
| avg = sum(scores_numericos) / len(scores_numericos) | |
| resultados[actor_id] = round(avg, 4) | |
| logger.info( | |
| f"[sentiment] {actor_id}: sentimiento_promedio={avg:+.4f} " | |
| f"({len(scores_numericos)} replies analizados)" | |
| ) | |
| return resultados | |
| except Exception as e: | |
| logger.error(f"[sentiment] error calculando sentimiento de replies: {e}") | |
| return {} | |
| def get_actor_sentiment_summary(actor_id: str) -> dict: | |
| """ | |
| Calcula el resumen de sentimiento para un actor a partir de sus posts en Supabase. | |
| Lee los posts del actor, analiza el sentimiento de cada título+resumen, | |
| y agrega los resultados en porcentajes. | |
| Parámetros: | |
| actor_id: ID del actor en seed_actors.json | |
| Devuelve: | |
| Dict con claves: positivo (%), negativo (%), neutral (%), total_analizado (int). | |
| Si no hay posts devuelve todos en 0. | |
| """ | |
| vacio = {"positivo": 0.0, "negativo": 0.0, "neutral": 0.0, "total_analizado": 0} | |
| if not SUPABASE_URL or not SUPABASE_KEY: | |
| logger.warning("[sentiment] Supabase no configurado — sin datos para analizar") | |
| return vacio | |
| try: | |
| client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) | |
| resp = ( | |
| client.table("posts") | |
| .select("titulo, resumen") | |
| .eq("actor_id", actor_id) | |
| .limit(200) | |
| .execute() | |
| ) | |
| posts = resp.data or [] | |
| except Exception as e: | |
| logger.error(f"[sentiment] error leyendo posts de Supabase para {actor_id}: {e}") | |
| return vacio | |
| if not posts: | |
| logger.info(f"[sentiment] {actor_id}: sin posts en Supabase") | |
| return vacio | |
| # Combinar título + resumen, normalizar y detectar injection | |
| textos = [ | |
| limpiar_texto_para_nlp( | |
| f"{p.get('titulo', '')} {p.get('resumen', '')}".strip(), | |
| actor_id=actor_id, | |
| ) | |
| for p in posts | |
| ] | |
| textos = [t for t in textos if t] # descartar los que quedaron vacíos | |
| sentimientos = analyze_sentiment(textos) | |
| conteo = {"positive": 0, "negative": 0, "neutral": 0} | |
| for s in sentimientos: | |
| label = s.get("label", "neutral") | |
| if label in conteo: | |
| conteo[label] += 1 | |
| else: | |
| conteo["neutral"] += 1 | |
| total = len(sentimientos) | |
| resumen = { | |
| "positivo": round(conteo["positive"] / total * 100, 1), | |
| "negativo": round(conteo["negative"] / total * 100, 1), | |
| "neutral": round(conteo["neutral"] / total * 100, 1), | |
| "total_analizado": total, | |
| } | |
| logger.info( | |
| f"[sentiment] {actor_id}: {total} posts analizados — " | |
| f"pos={resumen['positivo']}% neg={resumen['negativo']}% neu={resumen['neutral']}%" | |
| ) | |
| return resumen | |