app.py

"""
app.py — NPS AI Inference API
HuggingFace Space (Docker SDK) — expose un endpoint REST pour l'Automation 2 SFMC

Endpoints :
  GET  /health    → statut + modèles chargés
  POST /predict   → score IA + description CRM

Déploiement :
  1. Créer un Space HuggingFace (SDK: Docker)
  2. Uploader ce dossier (app.py, modeling_nps_score.py, Dockerfile, requirements.txt)
  3. Ajouter le secret HF_TOKEN dans les Settings du Space
"""

import os
import re
import logging
from contextlib import asynccontextmanager
from typing import Optional

import torch
import numpy as np
from fastapi import FastAPI, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
from transformers import AutoTokenizer, AutoModelForSeq2SeqLM

from modeling_nps_score import NPSScoreModel, NPSScoreConfig

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

from huggingface_hub import login
import os

# ─── Auth HuggingFace ─────────────────────────────────────────────────────────
hf_token = os.getenv("HF_TOKEN")
if hf_token:
    login(token=hf_token)

# ─── Config ──────────────────────────────────────────────────────────────────
SCORE_REPO = os.getenv("SCORE_REPO", "nada-05/nps-score-xlm-roberta")
DESC_REPO  = os.getenv("DESC_REPO",  "nada-05/nps-description-generator-mt5")
API_SECRET = os.getenv("API_SECRET", "")          # Secret à définir dans le Space HF
DEVICE     = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# ─── Seuils NPS ──────────────────────────────────────────────────────────────
def score_to_segment(score: float) -> str:
    s = round(score)
    if s <= 6: return "Detractor"
    if s <= 8: return "Passive"
    return "Promoter"


def estimate_urgency(ai_score: float, clicked_score: float, comments: str) -> str:
    avg = (ai_score + clicked_score) / 2
    txt = comments.lower()
    urgent_kw = [
        "humiliant", "ignoré", "abîmé", "défectueux", "menti", "retour impossible",
        "damaged", "refused", "40 minutes", "30 minutes", "waiting"
    ]
    has_urgent = any(k in txt for k in urgent_kw)
    if avg <= 3 or (avg <= 5 and has_urgent): return "high"
    if avg <= 5: return "medium"
    if avg <= 6 and has_urgent: return "medium"
    if avg <= 6: return "low"
    return "none"


# ─── Nettoyage texte ─────────────────────────────────────────────────────────
def clean_text(text: str) -> str:
    if not isinstance(text, str) or not text.strip():
        return ""
    text = re.sub(r"Q\d\s*:", "", text)
    text = text.replace("|", " ")
    text = re.sub(r"\b(really|hm|hm!)\b", "", text, flags=re.IGNORECASE)
    return re.sub(r"\s+", " ", text).strip()


def build_input_text(comments: str, improvements: str) -> str:
    c = clean_text(comments)
    parts = [c] if c else []
    if improvements.strip():
        parts.append(f"Points signalés : {improvements.strip()}")
    return " ".join(parts) if parts else "[aucun commentaire]"


def build_desc_prompt(comments: str, improvements: str, ai_score: float,
                      segment: str, urgency: str, language: str) -> str:
    c = clean_text(comments)
    lang_hint = f"[lang:{language}] " if language else ""
    prompt = (
        f"generate nps description: {lang_hint}"
        f"score:{ai_score:.0f}/10 segment:{segment} urgency:{urgency} "
        f"comments: {c}"
    )
    if improvements.strip():
        prompt += f" improvements: {improvements.strip()}"
    return prompt


# ─── État global des modèles ──────────────────────────────────────────────────
models = {}


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Chargement des modèles au démarrage (une seule fois)."""
    logger.info(f"Chargement des modèles sur {DEVICE}...")

    logger.info(f"  Score model : {SCORE_REPO}")
    models["score_tokenizer"] = AutoTokenizer.from_pretrained(
        SCORE_REPO, trust_remote_code=True
    )
    models["score_model"] = NPSScoreModel.from_pretrained(
        SCORE_REPO, trust_remote_code=True
    ).to(DEVICE)
    models["score_model"].eval()

    logger.info(f"  Description model : {DESC_REPO}")
    models["desc_tokenizer"] = AutoTokenizer.from_pretrained(DESC_REPO)
    models["desc_model"]     = AutoModelForSeq2SeqLM.from_pretrained(DESC_REPO).to(DEVICE)
    models["desc_model"].eval()

    logger.info("✓ Modèles prêts")
    yield
    models.clear()


# ─── App FastAPI ──────────────────────────────────────────────────────────────
app = FastAPI(
    title="NPS AI Inference API",
    description="Score sentimental NPS + génération de descriptions CRM",
    version="1.0.0",
    lifespan=lifespan,
)

security = HTTPBearer(auto_error=False)


def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
    """Vérifie le Bearer token si API_SECRET est défini."""
    if API_SECRET and (not credentials or credentials.credentials != API_SECRET):
        raise HTTPException(status_code=401, detail="Token invalide")


# ─── Schémas ─────────────────────────────────────────────────────────────────
class NPSRequest(BaseModel):
    comments:    str   = Field(...,  description="AllComments de NPS_Responses")
    improvements: str  = Field("",  description="Improvements de NPS_Responses")
    score_nps:   float = Field(5.0, ge=0, le=10, description="Score cliqué par le client")
    language:    str   = Field("fr", description="'fr' ou 'en'")


class NPSResponse(BaseModel):
    # Score
    score_nps_clicked: float
    segment_clicked:   str
    ai_score:          float
    segment_ai:        str

    # CRM
    description:  str    # Texte complet pour DE_Tasks.Description (≤ 4000 chars)
    ai_reason:    str    # Version courte pour DE_Tasks.AIReason (≤ 48 chars)
    ai_reason_full: str  # Version longue pour NPS_CloseTheLoop_History.AIReason (≤ 498 chars)
    urgency_ai:   str    # high / medium / low / none
    needs_task:   bool   # Recommandation IA de créer une tâche

    # Méta
    model_version: str = "xlm-roberta-v1"


# ─── Endpoints ───────────────────────────────────────────────────────────────
@app.get("/health")
def health():
    return {
        "status": "ok",
        "models_loaded": list(models.keys()),
        "device": str(DEVICE),
        "score_repo": SCORE_REPO,
        "desc_repo":  DESC_REPO,
    }


@app.post("/predict", response_model=NPSResponse)
def predict(req: NPSRequest, _=Security(verify_token)):
    if not models:
        raise HTTPException(status_code=503, detail="Modèles non chargés")

    # ── 1. Score IA ─────────────────────────────────────────────
    input_text = build_input_text(req.comments, req.improvements)
    enc = models["score_tokenizer"](
        input_text, return_tensors="pt",
        max_length=256, truncation=True, padding="max_length",
    ).to(DEVICE)

    with torch.no_grad():
        out = models["score_model"](**enc)
    ai_score    = round(float(out.logits.item()) * 10, 1)
    ai_score    = max(0.0, min(10.0, ai_score))
    segment_ai  = score_to_segment(ai_score)
    seg_clicked = score_to_segment(req.score_nps)
    urgency_ai  = estimate_urgency(ai_score, req.score_nps, req.comments)
    needs_task  = segment_ai == "Detractor"

    # ── 2. Description (seulement si détracteur IA) ─────────────
    description = ""
    if needs_task:
        prompt = build_desc_prompt(
            req.comments, req.improvements,
            ai_score, segment_ai, urgency_ai, req.language
        )
        enc_desc = models["desc_tokenizer"](
            prompt, return_tensors="pt", max_length=256, truncation=True
        ).to(DEVICE)
        with torch.no_grad():
            out_desc = models["desc_model"].generate(
                **enc_desc,
                max_new_tokens=200,
                num_beams=4,
                no_repeat_ngram_size=3,
                early_stopping=True,
            )
        description = models["desc_tokenizer"].decode(out_desc[0], skip_special_tokens=True)

    # ── 3. Versions tronquées pour les DE ────────────────────────
    ai_reason       = description[:48]  if description else ""   # DE_Tasks.AIReason (50)
    ai_reason_full  = description[:498] if description else ""   # History.AIReason (500)

    return NPSResponse(
        score_nps_clicked = req.score_nps,
        segment_clicked   = seg_clicked,
        ai_score          = ai_score,
        segment_ai        = segment_ai,
        description       = description[:4000],
        ai_reason         = ai_reason,
        ai_reason_full    = ai_reason_full,
        urgency_ai        = urgency_ai,
        needs_task        = needs_task,
    )


# ─── Lancement local ─────────────────────────────────────────────────────────
if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False)