oc-p10-content-recommendation/streamlit_app.py

import streamlit as st
import requests
import json
import time

# Configuration de la page
st.set_page_config(
    page_title="Système de Recommandation",
    page_icon="🎯",
    layout="wide"
)

# Configuration de l'API
API_BASE_URL = "http://localhost:7071/api"

def call_recommendation_api(user_id, base_url=API_BASE_URL):
    """
    Appelle l'API Azure Function pour obtenir des recommandations.
    """
    try:
        url = f"{base_url}/recommend"
        params = {"user_id": user_id}
        
        start_time = time.time()
        response = requests.get(url, params=params, timeout=30)
        response_time = time.time() - start_time
        
        if response.status_code == 200:
            return response.json(), response_time, None
        else:
            error_msg = f"Erreur {response.status_code}: {response.text}"
            return None, response_time, error_msg
            
    except requests.exceptions.Timeout:
        return None, None, "Timeout: L'API a mis trop de temps à répondre"
    except requests.exceptions.ConnectionError:
        return None, None, "Erreur de connexion: Vérifiez que l'Azure Function est démarrée"
    except Exception as e:
        return None, None, f"Erreur inattendue: {str(e)}"

def main():
    # Titre principal
    st.title("🎯 Système de Recommandation de Contenu")
    st.markdown("---")
    
    # Configuration dans la sidebar
    with st.sidebar:
        st.header("⚙️ Configuration")
        
        # URL de l'API
        api_url = st.text_input(
            "URL de l'API", 
            value=API_BASE_URL,
            help="URL de base de votre Azure Function"
        )
        
        # Test de connexion
        if st.button("🔍 Tester la connexion"):
            with st.spinner("Test en cours..."):
                try:
                    test_response = requests.get(f"{api_url}/recommend?user_id=0", timeout=5)
                    if test_response.status_code in [200, 400, 404]:  # API répond
                        st.success("✅ API accessible")
                    else:
                        st.error(f"❌ API répond avec le code {test_response.status_code}")
                except Exception as e:
                    st.error(f"❌ Impossible de joindre l'API: {str(e)}")
        
        st.markdown("---")
        st.markdown("**Instructions:**")
        st.markdown("1. Assurez-vous que `func start` est lancé")
        st.markdown("2. Entrez un ID utilisateur")
        st.markdown("3. Cliquez sur 'Obtenir des recommandations'")
    
    # Interface principale
    col1, col2 = st.columns([1, 2])
    
    with col1:
        st.header("👤 Sélection utilisateur")
        
        # Input pour l'ID utilisateur
        user_id = st.number_input(
            "ID de l'utilisateur",
            min_value=0,
            max_value=10000,
            value=0,
            step=1,
            help="Entrez l'ID de l'utilisateur pour lequel obtenir des recommandations"
        )
        
        # Bouton pour lancer la recommandation
        if st.button("🚀 Obtenir des recommandations", type="primary"):
            with st.spinner(f"Génération des recommandations pour l'utilisateur {user_id}..."):
                result, response_time, error = call_recommendation_api(user_id, api_url)
                
                # Stocker les résultats dans la session
                st.session_state['last_result'] = result
                st.session_state['last_response_time'] = response_time
                st.session_state['last_error'] = error
                st.session_state['last_user_id'] = user_id
    
    with col2:
        st.header("📊 Résultats")
        
        # Affichage des résultats
        if 'last_result' in st.session_state:
            if st.session_state['last_error']:
                st.error(f"❌ {st.session_state['last_error']}")
            elif st.session_state['last_result']:
                result = st.session_state['last_result']
                
                # Métriques
                col_metrics1, col_metrics2 = st.columns(2)
                with col_metrics1:
                    st.metric("👤 Utilisateur", st.session_state['last_user_id'])
                with col_metrics2:
                    if st.session_state['last_response_time']:
                        st.metric("⏱️ Temps de réponse", f"{st.session_state['last_response_time']:.2f}s")
                
                # Recommandations
                if 'recommendations' in result:
                    st.subheader("🎯 Articles recommandés")
                    recommendations = result['recommendations']
                    
                    # Affichage sous forme de liste numérotée
                    for i, article_id in enumerate(recommendations, 1):
                        st.write(f"**{i}.** Article ID: `{article_id}`")
                    
                    # JSON brut (collapsible)
                    with st.expander("📝 Réponse JSON complète"):
                        st.json(result)
                else:
                    st.warning("⚠️ Aucune recommandation trouvée dans la réponse")
        else:
            st.info("👆 Sélectionnez un utilisateur et cliquez sur 'Obtenir des recommandations' pour commencer")
    
    # Section de test avancé
    st.markdown("---")
    st.header("🧪 Tests avancés")
    
    col_test1, col_test2 = st.columns(2)
    
    with col_test1:
        st.subheader("📈 Test de performance")
        if st.button("Tester avec plusieurs utilisateurs"):
            test_users = [0, 1, 2, 5, 10]
            progress_bar = st.progress(0)
            results = []
            
            for i, uid in enumerate(test_users):
                result, resp_time, error = call_recommendation_api(uid, api_url)
                results.append({
                    'user_id': uid,
                    'success': error is None,
                    'response_time': resp_time,
                    'error': error
                })
                progress_bar.progress((i + 1) / len(test_users))
            
            # Affichage des résultats
            success_count = sum(1 for r in results if r['success'])
            avg_time = sum(r['response_time'] for r in results if r['response_time']) / len([r for r in results if r['response_time']])
            
            st.success(f"✅ {success_count}/{len(test_users)} appels réussis")
            st.info(f"⏱️ Temps moyen: {avg_time:.2f}s")
    
    with col_test2:
        st.subheader("🔍 Test d'erreur")
        if st.button("Tester utilisateur inexistant"):
            with st.spinner("Test en cours..."):
                result, resp_time, error = call_recommendation_api(99999, api_url)
                if error:
                    st.warning(f"Erreur attendue: {error}")
                elif result:
                    st.info("Réponse reçue (comportement inattendu)")
                    st.json(result)

if __name__ == "__main__":
    main()