app.py

import os
import json
import re
import gradio as gr
import torch
import pandas as pd
from scipy.sparse import csr_matrix
from transformers import AutoModel, AutoProcessor
from sklearn.feature_extraction.text import TfidfVectorizer

# 📌 Vérifier si CUDA est disponible
device = "cuda" if torch.cuda.is_available() else "cpu"
print(f"🔹 Utilisation du périphérique : {device}")

# 📌 Chargement du modèle Marqo Embeddings
model_name = "Marqo/marqo-ecommerce-embeddings-L"
print(f"🔄 Chargement du modèle {model_name}...")

model = AutoModel.from_pretrained(model_name, trust_remote_code=True).to(device)
processor = AutoProcessor.from_pretrained(model_name, trust_remote_code=True)

print("✅ Modèle chargé avec succès !")

# 📌 Définition des fichiers JSON
PRODUCTS_FILE = "products.json"
QA_FILE = "qa_sequences_output.json"

# 📌 Fonction pour charger les fichiers JSON
def safe_load_json(file_path):
    if not os.path.exists(file_path):
        print(f"⛔ Fichier introuvable : {file_path}")
        return []
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            data = json.load(f)
            return data.get("products", []) if "products" in data else data
    except json.JSONDecodeError:
        print(f"⚠️ Erreur de décodage JSON dans {file_path}")
        return []

products_data = safe_load_json(PRODUCTS_FILE)
qa_data = safe_load_json(QA_FILE)

# 📌 Générer des embeddings pour les produits
def get_text_embeddings(texts):
    """Génère des embeddings à partir du modèle Marqo"""
    with torch.no_grad():
        processed_texts = processor(text=texts, return_tensors="pt", truncation=True, max_length=64, padding=True).to(device)
        embeddings = model(**processed_texts).last_hidden_state.mean(dim=1)
    return embeddings.cpu().numpy()

# Création des embeddings pour tous les produits
print("🛠️ Génération des embeddings des produits...")
product_embeddings = get_text_embeddings([prod["title"] + " " + prod["description"] for prod in products_data])
print("✅ Embeddings générés et sauvegardés !")

# 📌 TF-IDF Vectorizer pour une recherche hybride
vectorizer = TfidfVectorizer(stop_words="english")
tfidf_matrix = vectorizer.fit_transform([prod["title"] + " " + prod["description"] for prod in products_data])

# 📌 Recherche hybride avec Marqo embeddings + TF-IDF
def search_products(query, category, min_price, max_price, weight_tfidf=0.5, weight_marqo=0.5):
    if not query.strip():
        return "❌ Veuillez entrer un terme de recherche valide."

    min_price = float(min_price) if isinstance(min_price, (int, float)) else 0
    max_price = float(max_price) if isinstance(max_price, (int, float)) else float("inf")

    # 📌 Génération de l'embedding de la requête utilisateur
    query_embedding = get_text_embeddings([query])[0]

    # 📌 Calcul de similarité cosinus entre la requête et les produits
    marqo_scores = (product_embeddings @ query_embedding).tolist()

    # 📌 TF-IDF Similarité
    query_vector_sparse = csr_matrix(vectorizer.transform([query]))
    tfidf_scores = (tfidf_matrix * query_vector_sparse.T).toarray().flatten()

    # 📌 Normalisation des scores Marqo
    if len(marqo_scores) > 0 and max(marqo_scores) != min(marqo_scores):
        marqo_scores = (pd.Series(marqo_scores) - min(marqo_scores)) / (max(marqo_scores) - min(marqo_scores) + 1e-6)
    else:
        marqo_scores = [1] * len(products_data)

    # 📌 Normalisation des scores TF-IDF
    if len(tfidf_scores) > 0 and max(tfidf_scores) != min(tfidf_scores):
        tfidf_scores_norm = (tfidf_scores - min(tfidf_scores)) / (max(tfidf_scores) - min(tfidf_scores) + 1e-6)
    else:
        tfidf_scores_norm = [1] * len(products_data)

    # 📌 Fusionner les scores TF-IDF et Marqo embeddings
    final_scores = weight_tfidf * tfidf_scores_norm[:len(products_data)] + weight_marqo * marqo_scores

    # 📌 Création d'un DataFrame avec les scores finaux
    results_df = pd.DataFrame(products_data)
    results_df["score"] = final_scores

    # 📌 Filtrage des résultats par prix et disponibilité
    results_df = results_df[
        (results_df["price"].fillna(0).astype(float) >= min_price) &
        (results_df["price"].fillna(0).astype(float) <= max_price) &
        (results_df["availability"].fillna("").str.lower() == "in stock")
    ]

    if category and category != "Toutes":
        results_df = results_df[results_df["category"].str.contains(category, case=False, na=False)]
    
    return results_df.sort_values(by="score", ascending=False).head(20)

# 📌 Interface Gradio
app = gr.Interface(
    fn=search_products,
    inputs=[
        gr.Textbox(label="Rechercher un produit"), 
        gr.Textbox(label="Catégorie"), 
        gr.Number(label="Prix min"), 
        gr.Number(label="Prix max")
    ],
    outputs=[
        gr.Dataframe(headers=["ID", "Titre", "Description", "Prix", "Disponibilité", "Score"], 
                     datatype=["str", "str", "str", "number", "str", "number"])
    ]
)

app.launch()