198 lines
No EOL
8 KiB
Python
198 lines
No EOL
8 KiB
Python
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']
|
|
|
|
# Vérifier si on a le format enrichi (avec scores et catégories)
|
|
if recommendations and isinstance(recommendations[0], dict) and 'similarity_score_percent' in recommendations[0]:
|
|
# Format enrichi - affichage en tableau
|
|
import pandas as pd
|
|
|
|
df_data = []
|
|
for i, rec in enumerate(recommendations, 1):
|
|
df_data.append({
|
|
'Rang': i,
|
|
'Article ID': rec['article_id'],
|
|
'Similarité (%)': f"{rec['similarity_score_percent']:.1f}%",
|
|
'Catégorie': rec.get('category_id', 'N/A')
|
|
})
|
|
|
|
df = pd.DataFrame(df_data)
|
|
st.dataframe(df, hide_index=True, use_container_width=True)
|
|
else:
|
|
# Format simple - affichage basique pour compatibilité
|
|
for i, article_id in enumerate(recommendations, 1):
|
|
if isinstance(article_id, dict):
|
|
article_id = article_id.get('article_id', article_id)
|
|
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() |