src/features.py

"""Construcción de variables para entrenamiento e inferencia."""
from __future__ import annotations

import math
import re
from datetime import datetime, timezone
from typing import Any, Dict, Iterable

import numpy as np
import pandas as pd

from .config import BENEFIT_TERMS, CTA_TERMS, PRICE_TERMS, PROMO_TERMS, TRUST_TERMS, URGENCY_TERMS


def safe_float(value: Any, default: float = 0.0) -> float:
    """Convierte a float aceptando None, strings y NaN sin fallar."""
    try:
        if value is None or value == "":
            return default
        f = float(value)
        if math.isnan(f) or math.isinf(f):
            return default
        return f
    except Exception:
        return default


def safe_int(value: Any, default: int = 0) -> int:
    return int(round(safe_float(value, default=default)))


def contains_any(text: str, terms: Iterable[str]) -> int:
    lower = (text or "").lower()
    return int(any(term in lower for term in terms))


def uppercase_ratio(text: str) -> float:
    letters = [c for c in text or "" if c.isalpha()]
    if not letters:
        return 0.0
    return round(sum(1 for c in letters if c.isupper()) / len(letters), 4)


def count_words(text: str) -> int:
    return len(re.findall(r"\w+", text or "", flags=re.UNICODE))




STOPWORDS_RELEVANCE = {
    "de", "la", "el", "los", "las", "un", "una", "unos", "unas", "y", "o",
    "en", "a", "por", "para", "con", "sin", "del", "al", "que", "como",
    "más", "mas", "muy", "este", "esta", "estos", "estas", "eso", "esa",
    "es", "son", "ser", "fue", "hay", "se", "su", "sus", "tu", "tus",
    "mi", "mis", "lo", "le", "les", "video", "youtube", "canal",
    "comentario", "comentarios", "contenido", "link", "http", "https",
    "www", "com", "reel", "short", "shorts", "post", "publicacion", "publicación"
}

HUMOR_TERMS = {
    "jaja", "jajaja", "jeje", "risa", "humor", "broma", "chiste", "meme",
    "parodia", "sketch", "comedia", "gracioso", "graciosa", "sarcasmo",
    "irónico", "ironico", "pov", "cuando", "fail", "fails", "absurdo", "random"
}
EDU_TERMS = {
    "tutorial", "aprende", "aprender", "guía", "guia", "paso a paso", "cómo",
    "como hacer", "explico", "explicar", "consejos", "tips", "clase", "curso"
}
COMMERCIAL_TERMS = {
    "compra", "comprar", "precio", "descuento", "oferta", "promoción", "promocion",
    "cupón", "cupon", "envío", "envio", "agenda", "reserva", "servicio",
    "producto", "tienda", "cliente", "clientes", "venta", "ventas", "gratis",
    "whatsapp", "cotiza", "cotizar", "link en bio"
}
MUSIC_TERMS = {
    "música", "musica", "canción", "cancion", "letra", "lyrics", "cover",
    "artista", "cantante", "banda", "álbum", "album", "single", "concierto",
    "verso", "coro", "beat", "remix", "videoclip"
}


def relevance_tokens(text: str) -> set[str]:
    """Tokens útiles para medir si el texto OCR conecta con el mensaje."""
    clean = (text or "").lower()
    clean = re.sub(r"https?://\S+", " ", clean)
    clean = re.sub(r"www\.\S+", " ", clean)
    clean = re.sub(r"[^a-záéíóúüñ0-9\s]", " ", clean, flags=re.UNICODE)
    clean = re.sub(r"\s+", " ", clean).strip()
    return {
        token for token in clean.split()
        if len(token) >= 3 and token not in STOPWORDS_RELEVANCE
    }


def _count_custom_terms(text: str, terms: Iterable[str]) -> int:
    low = (text or "").lower()
    return sum(1 for term in terms if str(term).lower() in low)


def infer_content_intent_light(title: str = "", description: str = "", transcript_text: str = "", ocr_text: str = "") -> str:
    """Detecta intención básica para no aplicar lógica comercial a todo.

    Retorna una etiqueta simple usada por OCR/guion: humor, comercial,
    educativo, informativo o general.
    """
    text = " ".join([title or "", description or "", transcript_text or "", ocr_text or ""]).lower()
    humor = _count_custom_terms(text, HUMOR_TERMS)
    edu = _count_custom_terms(text, EDU_TERMS)
    music = _count_custom_terms(text, MUSIC_TERMS)
    commercial = _count_custom_terms(text, COMMERCIAL_TERMS) + contains_any(text, CTA_TERMS) + contains_any(text, PROMO_TERMS) + contains_any(text, PRICE_TERMS)
    if music >= 1 and commercial < 3:
        return "musical"
    if humor >= 1 and commercial < 3:
        return "humor/entretenimiento"
    if commercial >= 3:
        return "comercial/promocional"
    if edu >= 2:
        return "educativo/tutorial"
    if any(x in text for x in ["noticia", "actualidad", "denuncia", "entrevista", "comunicado"]):
        return "informativo/noticioso"
    return "general/branding"


def _ocr_role_for_intent(ocr_text: str, intent: str, overlap: float, marketing_signal: float, humor_signal: float) -> str:
    if not (ocr_text or "").strip():
        return "sin_texto_detectado"
    if intent.startswith("musical"):
        return "letra_musical_o_rotulo" if overlap >= 0.15 else "texto_musical_poco_conectado"
    if intent.startswith("humor"):
        if humor_signal > 0 or overlap >= 0.25:
            return "apoya_contexto_o_remate"
        return "texto_desconectado_del_chiste"
    if intent.startswith("comercial"):
        if marketing_signal >= 0.34:
            return "refuerza_oferta_cta_o_beneficio"
        if overlap >= 0.25:
            return "acompaña_mensaje_comercial"
        return "texto_comercial_poco_claro"
    if intent.startswith("educativo"):
        return "apoyo_explicativo" if overlap >= 0.20 else "texto_explicativo_poco_conectado"
    return "contexto_visual" if overlap >= 0.20 else "texto_visible_sin_funcion_clara"


def compute_ocr_relevance_score(
    ocr_text: str,
    title: str = "",
    description: str = "",
    transcript_text: str = "",
    ocr_metrics: Dict[str, Any] | None = None,
) -> float:
    """Score 0-1 de relevancia del texto en pantalla.

    La versión anterior trataba casi todo como pieza comercial. Esta versión es
    sensible al tipo de contenido: humor/entretenimiento, educativo, comercial,
    informativo o general. Por eso no exige CTA ni producto cuando el video no
    está vendiendo nada.
    """
    ocr_metrics = ocr_metrics or {}
    ocr_text = ocr_text or ""
    if not ocr_text.strip():
        return 0.0

    intent = str(ocr_metrics.get("ocr_llm_content_intent") or infer_content_intent_light(title, description, transcript_text, ocr_text))
    ocr_tokens = relevance_tokens(ocr_text)
    context_tokens = relevance_tokens(" ".join([title or "", description or "", transcript_text or ""]))
    overlap = (len(ocr_tokens & context_tokens) / max(len(ocr_tokens), 1)) if ocr_tokens and context_tokens else 0.0

    commercial_flags = [
        contains_any(ocr_text, CTA_TERMS), contains_any(ocr_text, BENEFIT_TERMS),
        contains_any(ocr_text, URGENCY_TERMS), contains_any(ocr_text, TRUST_TERMS),
        contains_any(ocr_text, PROMO_TERMS), contains_any(ocr_text, PRICE_TERMS),
        _count_custom_terms(ocr_text, COMMERCIAL_TERMS) > 0,
    ]
    marketing_signal = sum(bool(x) for x in commercial_flags) / max(len(commercial_flags), 1)
    humor_signal = min(_count_custom_terms(ocr_text, HUMOR_TERMS) / 2, 1.0)
    edu_signal = min(_count_custom_terms(ocr_text, EDU_TERMS) / 2, 1.0)

    coverage = min(safe_float(ocr_metrics.get("ocr_frame_coverage", 0)), 1.0)
    wc = count_words(ocr_text)
    if 2 <= wc <= 18:
        density_fit = 1.0
    elif 19 <= wc <= 40:
        density_fit = 0.65
    elif wc > 40:
        density_fit = 0.35
    else:
        density_fit = 0.45

    if intent.startswith("musical"):
        # En música el texto puede ser letra, título de canción, artista o rótulo;
        # no se exige CTA ni venta.
        score = 0.42 * overlap + 0.25 * coverage + 0.23 * density_fit + 0.10 * max(humor_signal, edu_signal)
    elif intent.startswith("humor"):
        # En humor el texto puede ser caption, setup, meme/subtítulo o remate;
        # no tiene que tener CTA ni beneficio comercial.
        score = 0.45 * overlap + 0.20 * humor_signal + 0.20 * coverage + 0.15 * density_fit
    elif intent.startswith("educativo"):
        score = 0.45 * overlap + 0.20 * edu_signal + 0.20 * coverage + 0.15 * density_fit
    elif intent.startswith("comercial"):
        score = 0.38 * overlap + 0.34 * marketing_signal + 0.18 * coverage + 0.10 * density_fit
    else:
        score = 0.50 * overlap + 0.20 * coverage + 0.20 * density_fit + 0.10 * max(marketing_signal, humor_signal, edu_signal)

    return round(max(0.0, min(score, 1.0)), 4)


def describe_ocr_relevance(
    ocr_text: str,
    title: str = "",
    description: str = "",
    transcript_text: str = "",
    ocr_metrics: Dict[str, Any] | None = None,
) -> Dict[str, Any]:
    """Devuelve explicación del OCR sensible al objetivo del video."""
    ocr_metrics = ocr_metrics or {}
    intent = str(ocr_metrics.get("ocr_llm_content_intent") or infer_content_intent_light(title, description, transcript_text, ocr_text))
    ocr_tokens = relevance_tokens(ocr_text)
    context_tokens = relevance_tokens(" ".join([title or "", description or "", transcript_text or ""]))
    overlap = (len(ocr_tokens & context_tokens) / max(len(ocr_tokens), 1)) if ocr_tokens and context_tokens else 0.0
    commercial_flags = [
        contains_any(ocr_text, CTA_TERMS), contains_any(ocr_text, BENEFIT_TERMS),
        contains_any(ocr_text, URGENCY_TERMS), contains_any(ocr_text, TRUST_TERMS),
        contains_any(ocr_text, PROMO_TERMS), contains_any(ocr_text, PRICE_TERMS),
        _count_custom_terms(ocr_text, COMMERCIAL_TERMS) > 0,
    ]
    marketing_signal = sum(bool(x) for x in commercial_flags) / max(len(commercial_flags), 1)
    humor_signal = min(_count_custom_terms(ocr_text, HUMOR_TERMS) / 2, 1.0)
    score = compute_ocr_relevance_score(ocr_text, title, description, transcript_text, ocr_metrics)
    role = str(ocr_metrics.get("ocr_llm_role") or _ocr_role_for_intent(ocr_text, intent, overlap, marketing_signal, humor_signal))
    if not (ocr_text or "").strip():
        interpretation = "No se detectó texto en pantalla; la lectura visual depende del audio, imagen y ritmo."
    elif intent.startswith("musical"):
        interpretation = "El texto en pantalla se evalúa como letra, rótulo, título, artista o apoyo visual del videoclip; no como CTA comercial."
    elif intent.startswith("humor"):
        interpretation = "El texto en pantalla se evalúa como soporte del chiste, contexto o remate; no como CTA de venta."
    elif intent.startswith("comercial"):
        interpretation = "El texto en pantalla se evalúa por su capacidad de reforzar oferta, beneficio, prueba o acción."
    elif intent.startswith("educativo"):
        interpretation = "El texto en pantalla se evalúa como apoyo explicativo: debe aclarar conceptos, pasos o ideas clave."
    else:
        interpretation = "El texto en pantalla se evalúa por su conexión con el tema central y por si ayuda a entender el video."
    return {
        "content_intent": intent,
        "ocr_role": role,
        "ocr_context_overlap": round(overlap, 4),
        "ocr_marketing_signal": round(marketing_signal, 4),
        "ocr_humor_signal": round(humor_signal, 4),
        "ocr_relevance_score": score,
        "ocr_interpretation": interpretation,
    }

def parse_date(value: Any) -> datetime | None:
    if not value:
        return None
    try:
        parsed = pd.to_datetime(value, utc=True, errors="coerce")
        if pd.isna(parsed):
            return None
        return parsed.to_pydatetime()
    except Exception:
        return None


def days_since_publication(published_at: Any) -> float:
    dt = parse_date(published_at)
    if not dt:
        return 1.0
    now = datetime.now(timezone.utc)
    days = (now - dt).total_seconds() / 86400
    return max(days, 1.0)


def duration_fit_score(duration_seconds: float) -> float:
    """Score heurístico de ajuste de duración para anuncios y contenido corto."""
    d = safe_float(duration_seconds)
    if d <= 0:
        return 0.5
    if d <= 7:
        return 0.95
    if d <= 15:
        return 0.90
    if d <= 30:
        return 0.80
    if d <= 60:
        return 0.70
    if d <= 180:
        return 0.55
    return 0.40


def compute_text_power_score(text: str, ocr_metrics: Dict[str, Any] | None = None) -> float:
    """Score 0-1 de potencia comunicacional por señales básicas."""
    ocr_metrics = ocr_metrics or {}
    score = 0.0
    score += 0.20 * contains_any(text, CTA_TERMS)
    score += 0.18 * contains_any(text, BENEFIT_TERMS)
    score += 0.15 * contains_any(text, URGENCY_TERMS)
    score += 0.12 * contains_any(text, TRUST_TERMS)
    score += 0.10 * contains_any(text, PROMO_TERMS)
    score += 0.08 * contains_any(text, PRICE_TERMS)
    wc = count_words(text)
    if 8 <= wc <= 120:
        score += 0.10
    elif wc > 120:
        score += 0.04
    if safe_float(ocr_metrics.get("ocr_frame_coverage", 0)) > 0:
        score += 0.07
    return round(min(score, 1.0), 4)


def compute_engagement_metrics(views: Any, likes: Any, comments: Any, published_at: Any = None) -> Dict[str, float]:
    views_f = max(safe_float(views), 0.0)
    likes_f = max(safe_float(likes), 0.0)
    comments_f = max(safe_float(comments), 0.0)
    denom = max(views_f, 1.0)
    days = days_since_publication(published_at)
    return {
        "views": views_f,
        "likes": likes_f,
        "comments": comments_f,
        "engagement_rate": round((likes_f + comments_f) / denom, 6),
        "like_rate": round(likes_f / denom, 6),
        "comment_rate": round(comments_f / denom, 6),
        "views_per_day": round(views_f / days, 6),
        "log_views": round(float(np.log1p(views_f)), 6),
        "log_likes": round(float(np.log1p(likes_f)), 6),
        "log_comments": round(float(np.log1p(comments_f)), 6),
    }


def build_text_features(title: str, description: str, transcript_text: str, ocr_text: str) -> Dict[str, Any]:
    title = title or ""
    description = description or ""
    transcript_text = transcript_text or ""
    ocr_text = ocr_text or ""
    text_total = " ".join([title, description, transcript_text, ocr_text]).strip()
    return {
        "title": title,
        "description": description,
        "transcript_text": transcript_text,
        "ocr_text": ocr_text,
        "text_total": text_total,
        "title_len": len(title),
        "description_len": len(description),
        "transcript_len": len(transcript_text),
        "ocr_text_len": len(ocr_text),
        "text_total_len": len(text_total),
        "title_word_count": count_words(title),
        "description_word_count": count_words(description),
        "transcript_word_count": count_words(transcript_text),
        "ocr_word_count": count_words(ocr_text),
        "text_total_word_count": count_words(text_total),
        "exclamation_count": text_total.count("!"),
        "question_count": text_total.count("?"),
        "uppercase_ratio": uppercase_ratio(text_total),
        "digit_count": sum(c.isdigit() for c in text_total),
        "cta_flag": contains_any(text_total, CTA_TERMS),
        "urgency_flag": contains_any(text_total, URGENCY_TERMS),
        "trust_flag": contains_any(text_total, TRUST_TERMS),
        "promo_flag": contains_any(text_total, PROMO_TERMS),
        "benefit_flag": contains_any(text_total, BENEFIT_TERMS),
        "price_flag": contains_any(text_total, PRICE_TERMS),
    }


def build_feature_row(
    title: str = "",
    description: str = "",
    transcript_text: str = "",
    ocr_text: str = "",
    category_id: Any = "",
    duration_seconds: Any = 0,
    views: Any = 0,
    likes: Any = 0,
    comments: Any = 0,
    published_at: Any = None,
    video_type: str = "auto",
    extra: Dict[str, Any] | None = None,
) -> Dict[str, Any]:
    """Construye una fila de features para inferencia."""
    extra = extra or {}
    text_features = build_text_features(title, description, transcript_text, ocr_text)
    engagement = compute_engagement_metrics(views, likes, comments, published_at)
    d = safe_float(duration_seconds)
    ocr_relevance_info = describe_ocr_relevance(
        ocr_text=ocr_text,
        title=title,
        description=description,
        transcript_text=transcript_text,
        ocr_metrics=extra,
    )
    ocr_relevance = safe_float(ocr_relevance_info.get("ocr_relevance_score", 0))
    content_intent = ocr_relevance_info.get("content_intent") or infer_content_intent_light(title, description, transcript_text, ocr_text)
    text_power = compute_text_power_score(text_features["text_total"], extra)
    vt = video_type or "auto"
    if vt == "auto":
        vt = "sin narrador" if d and d <= 8 else "con narrador/desconocido"
    return {
        **text_features,
        **engagement,
        "category_id": str(category_id or "unknown"),
        "duration_seconds": d,
        "duration_fit_score": duration_fit_score(d),
        "duration_bucket": "short" if d <= 15 else "medium" if d <= 60 else "long",
        "video_type": vt,
        "content_intent": content_intent,
        "text_power_score": text_power,
        "ocr_relevance_score": ocr_relevance,
        **ocr_relevance_info,
        **extra,
    }


def metricas_block(features: Dict[str, Any]) -> Dict[str, Any]:
    """Subdiccionario `metricas` para el contrato de salida final."""
    return {
        "views": int(safe_float(features.get("views", 0))),
        "likes": int(safe_float(features.get("likes", 0))),
        "comments": int(safe_float(features.get("comments", 0))),
        "engagement_rate": round(safe_float(features.get("engagement_rate", 0)), 6),
        "like_rate": round(safe_float(features.get("like_rate", 0)), 6),
        "comment_rate": round(safe_float(features.get("comment_rate", 0)), 6),
        "views_per_day": round(safe_float(features.get("views_per_day", 0)), 4),
        "duration_seconds": round(safe_float(features.get("duration_seconds", 0)), 2),
        "duration_fit_score": round(safe_float(features.get("duration_fit_score", 0)), 4),
        "text_power_score": round(safe_float(features.get("text_power_score", 0)), 4),
        "ocr_relevance_score": round(safe_float(features.get("ocr_relevance_score", 0)), 4),
        "title_word_count": int(safe_float(features.get("title_word_count", 0))),
        "description_word_count": int(safe_float(features.get("description_word_count", 0))),
        "transcript_word_count": int(safe_float(features.get("transcript_word_count", 0))),
        "ocr_word_count": int(safe_float(features.get("ocr_word_count", 0))),
        "ocr_frame_coverage": round(safe_float(features.get("ocr_frame_coverage", 0)), 4),
        "visual_text_density": features.get("visual_text_density", "no_disponible"),
        "video_type": features.get("video_type", "auto"),
        "content_intent": features.get("content_intent", "general/branding"),
    }


def transcripcion_block(features: Dict[str, Any], extra_status: Dict[str, Any] | None = None) -> Dict[str, Any]:
    """Subdiccionario `analisis_transcripcion` para el contrato."""
    extra_status = extra_status or {}
    text = features.get("transcript_text", "") or ""
    return {
        "disponible": bool(text.strip()),
        "fuente": extra_status.get("source", "manual_o_no_ejecutado"),
        "advertencia": extra_status.get("warning", ""),
        "longitud_caracteres": len(text),
        "cantidad_palabras": count_words(text),
        "tiene_cta": bool(features.get("cta_flag")),
        "tiene_beneficio": bool(features.get("benefit_flag")),
        "tiene_urgencia": bool(features.get("urgency_flag")),
        "tiene_confianza": bool(features.get("trust_flag")),
        "tipo_contenido_estimado": features.get("content_intent", "general/branding"),
        "preview": text[:900] + ("..." if len(text) > 900 else ""),
    }


def ocr_block(features: Dict[str, Any], extra_status: Dict[str, Any] | None = None) -> Dict[str, Any]:
    """Subdiccionario `analisis_ocr` para el contrato."""
    extra_status = extra_status or {}
    text = features.get("ocr_text", "") or ""
    return {
        "disponible": bool(text.strip()),
        "motor": extra_status.get("engine", "no_disponible"),
        "advertencia": extra_status.get("warning", ""),
        "frames_con_texto": int(safe_float(features.get("ocr_frames_with_text", 0))),
        "frames_totales": int(safe_float(features.get("ocr_frame_count", 0))),
        "cobertura": round(safe_float(features.get("ocr_frame_coverage", 0)), 4),
        "cantidad_palabras": int(safe_float(features.get("ocr_word_count", 0))),
        "densidad": features.get("visual_text_density", "sin_texto"),
        "relevancia_texto_en_pantalla": round(safe_float(features.get("ocr_relevance_score", 0)), 4),
        "relevancia_pct": round(safe_float(features.get("ocr_relevance_score", 0)) * 100, 1),
        "tiene_cta": bool(features.get("ocr_cta_flag")),
        "tiene_promocion": bool(features.get("ocr_promo_flag")),
        "tiene_urgencia": bool(features.get("ocr_urgency_flag")),
        "tiene_confianza": bool(features.get("ocr_trust_flag")),
        "tiene_exceso_texto": features.get("visual_text_density") == "alta",
        "tipo_contenido_estimado": features.get("ocr_llm_content_intent") or features.get("content_intent", "general/branding"),
        "rol_texto_en_pantalla": features.get("ocr_llm_role") or features.get("ocr_role", "sin_funcion_clara"),
        "ocr_text_raw": features.get("ocr_text_raw", ""),
        "ocr_first_result": features.get("ocr_first_result", ""),
        "ocr_llm_source": features.get("ocr_llm_source", ""),
        "ocr_llm_warning": features.get("ocr_llm_warning", ""),
        "ocr_llm_meaning": features.get("ocr_llm_meaning", ""),
        "ocr_llm_confidence": features.get("ocr_llm_confidence", ""),
        "ocr_llm_content_intent": features.get("ocr_llm_content_intent", ""),
        "ocr_llm_role": features.get("ocr_llm_role", ""),
        "overlap_contextual": round(safe_float(features.get("ocr_context_overlap", 0)), 4),
        "senal_comercial": round(safe_float(features.get("ocr_marketing_signal", 0)), 4),
        "senal_humor": round(safe_float(features.get("ocr_humor_signal", 0)), 4),
        "interpretacion_ocr": features.get("ocr_interpretation", ""),
        "preview": text[:900] + ("..." if len(text) > 900 else ""),
    }


def normalize_training_dataframe(df: pd.DataFrame) -> pd.DataFrame:
    """Normaliza columnas frecuentes de datasets públicos de YouTube.

    Implementación **vectorizada** para que escale a cientos de miles de filas:
    se evita ``df.apply(axis=1)`` y se usan operaciones nativas de NumPy/Pandas.
    """
    rename_candidates = {
        "view_count": "views",
        "views_count": "views",
        "likes_count": "likes",
        "like_count": "likes",
        "comment_count": "comments",
        "comments_count": "comments",
        "publishedAt": "published_at",
        "publish_time": "published_at",
        "categoryId": "category_id",
        "channelTitle": "channel_title",
        "channel": "channel_title",  # archive/daily_trending_videos.csv usa "channel"
    }
    normalized = df.copy()
    for src_col, dst_col in rename_candidates.items():
        if src_col in normalized.columns and dst_col not in normalized.columns:
            normalized = normalized.rename(columns={src_col: dst_col})
    for col in ["title", "description", "tags"]:
        if col not in normalized.columns:
            normalized[col] = ""
        normalized[col] = normalized[col].fillna("").astype(str)
    for col in ["views", "likes", "comments"]:
        if col not in normalized.columns:
            normalized[col] = 0
        normalized[col] = pd.to_numeric(normalized[col], errors="coerce").fillna(0)
    if "published_at" not in normalized.columns:
        normalized["published_at"] = None
    if "category_id" not in normalized.columns:
        normalized["category_id"] = "unknown"
    normalized["category_id"] = normalized["category_id"].fillna("unknown").astype(str)
    if "duration_seconds" not in normalized.columns:
        normalized["duration_seconds"] = 0
    normalized["duration_seconds"] = pd.to_numeric(normalized["duration_seconds"], errors="coerce").fillna(0)

    # Texto combinado para TF-IDF.
    normalized["text_total"] = (
        normalized["title"] + " "
        + normalized["description"] + " "
        + normalized["tags"]
    )

    # Métricas de longitud y palabras (vectorizado).
    normalized["title_len"] = normalized["title"].str.len().fillna(0).astype(int)
    normalized["description_len"] = normalized["description"].str.len().fillna(0).astype(int)
    normalized["text_total_word_count"] = (
        normalized["text_total"].str.findall(r"\w+", flags=re.UNICODE).str.len().fillna(0).astype(int)
    )

    # Engagement vectorizado.
    views = normalized["views"].astype(float).clip(lower=0)
    likes = normalized["likes"].astype(float).clip(lower=0)
    comments = normalized["comments"].astype(float).clip(lower=0)
    denom = views.replace(0, np.nan).fillna(1.0)
    normalized["engagement_rate"] = ((likes + comments) / denom).round(6)
    normalized["like_rate"] = (likes / denom).round(6)
    normalized["comment_rate"] = (comments / denom).round(6)
    normalized["log_views"] = np.log1p(views).round(6)
    normalized["log_likes"] = np.log1p(likes).round(6)
    normalized["log_comments"] = np.log1p(comments).round(6)

    # Días desde publicación, vectorizado con clip a 1 día mínimo.
    pub = pd.to_datetime(normalized["published_at"], errors="coerce", utc=True)
    now = pd.Timestamp.now(tz="UTC")
    days = (now - pub).dt.total_seconds() / 86400.0
    days = days.fillna(1.0).clip(lower=1.0)
    normalized["views_per_day"] = (views / days).round(6)

    # Asegura que las columnas escalares existan en el orden conocido.
    normalized["views"] = views
    normalized["likes"] = likes
    normalized["comments"] = comments

    # Flags textuales vectorizados con regex (más rápido que apply en datasets grandes).
    text_lower = normalized["text_total"].str.lower()

    def _flag(terms):
        pattern = "|".join(re.escape(t) for t in terms)
        return text_lower.str.contains(pattern, regex=True, na=False).astype(int)

    normalized["cta_flag"] = _flag(CTA_TERMS)
    normalized["urgency_flag"] = _flag(URGENCY_TERMS)
    normalized["trust_flag"] = _flag(TRUST_TERMS)
    normalized["promo_flag"] = _flag(PROMO_TERMS)
    normalized["benefit_flag"] = _flag(BENEFIT_TERMS)
    normalized["price_flag"] = _flag(PRICE_TERMS)

    # text_power_score vectorizado replicando los pesos de compute_text_power_score.
    score = (
        0.20 * normalized["cta_flag"]
        + 0.18 * normalized["benefit_flag"]
        + 0.15 * normalized["urgency_flag"]
        + 0.12 * normalized["trust_flag"]
        + 0.10 * normalized["promo_flag"]
        + 0.08 * normalized["price_flag"]
    )
    wc = normalized["text_total_word_count"]
    score = score + np.where((wc >= 8) & (wc <= 120), 0.10, np.where(wc > 120, 0.04, 0.0))
    normalized["text_power_score"] = score.clip(upper=1.0).round(4)

    # duration_fit_score vectorizado.
    d = normalized["duration_seconds"].astype(float)
    fit = pd.Series(0.40, index=normalized.index)
    fit = np.where(d <= 7, 0.95, fit)
    fit = np.where((d > 7) & (d <= 15), 0.90, fit)
    fit = np.where((d > 15) & (d <= 30), 0.80, fit)
    fit = np.where((d > 30) & (d <= 60), 0.70, fit)
    fit = np.where((d > 60) & (d <= 180), 0.55, fit)
    fit = np.where(d <= 0, 0.50, fit)  # duración desconocida
    normalized["duration_fit_score"] = fit
    return normalized


def create_boost_candidate_target(df: pd.DataFrame, percentile: float = 0.70) -> pd.DataFrame:
    """Crea etiqueta boost_candidate a partir de score compuesto."""
    work = df.copy()

    def robust_minmax(series: pd.Series) -> pd.Series:
        s = pd.to_numeric(series, errors="coerce").fillna(0)
        lo, hi = s.quantile(0.02), s.quantile(0.98)
        clipped = s.clip(lo, hi)
        if hi - lo == 0:
            return pd.Series(np.zeros(len(s)), index=s.index)
        return (clipped - lo) / (hi - lo)

    work["organic_performance_score"] = (
        0.35 * robust_minmax(work["views_per_day"])
        + 0.25 * robust_minmax(work["engagement_rate"])
        + 0.15 * robust_minmax(work["like_rate"])
        + 0.10 * robust_minmax(work["comment_rate"])
        + 0.10 * robust_minmax(work["text_power_score"])
        + 0.05 * robust_minmax(work["duration_fit_score"])
    )
    threshold = work["organic_performance_score"].quantile(percentile)
    work["boost_candidate"] = (work["organic_performance_score"] >= threshold).astype(int)
    work["target_threshold"] = threshold
    return work