Project 10 added.

This commit is contained in:
Grégoire Mureau 2025-09-08 17:47:06 +02:00
commit b97406cbbc
19 changed files with 2677 additions and 0 deletions

6
.vscode/extensions.json vendored Normal file
View file

@ -0,0 +1,6 @@
{
"recommendations": [
"ms-azuretools.vscode-azurefunctions",
"ms-python.python"
]
}

15
.vscode/launch.json vendored Normal file
View file

@ -0,0 +1,15 @@
{
"version": "0.2.0",
"configurations": [
{
"name": "Attach to Python Functions",
"type": "debugpy",
"request": "attach",
"connect": {
"host": "localhost",
"port": 9091
},
"preLaunchTask": "func: host start"
}
]
}

9
.vscode/settings.json vendored Normal file
View file

@ -0,0 +1,9 @@
{
"azureFunctions.deploySubpath": "azure-app",
"azureFunctions.scmDoBuildDuringDeployment": true,
"azureFunctions.pythonVenv": ".venv",
"azureFunctions.projectLanguage": "Python",
"azureFunctions.projectRuntime": "~4",
"debug.internalConsoleOptions": "neverOpen",
"azureFunctions.projectLanguageModel": 2
}

33
.vscode/tasks.json vendored Normal file
View file

@ -0,0 +1,33 @@
{
"version": "2.0.0",
"tasks": [
{
"type": "func",
"label": "func: host start",
"command": "host start",
"problemMatcher": "$func-python-watch",
"isBackground": true,
"dependsOn": "pip install (functions)",
"options": {
"cwd": "${workspaceFolder}/azure-app"
}
},
{
"label": "pip install (functions)",
"type": "shell",
"osx": {
"command": "${config:azureFunctions.pythonVenv}/bin/python -m pip install -r requirements.txt"
},
"windows": {
"command": "${config:azureFunctions.pythonVenv}\\Scripts\\python -m pip install -r requirements.txt"
},
"linux": {
"command": "${config:azureFunctions.pythonVenv}/bin/python -m pip install -r requirements.txt"
},
"problemMatcher": [],
"options": {
"cwd": "${workspaceFolder}/azure-app"
}
}
]
}

9
azure-app/.funcignore Normal file
View file

@ -0,0 +1,9 @@
.git*
.vscode
__azurite_db*__.json
__blobstorage__
__queuestorage__
local.settings.json
test
.venv
test_local.py

135
azure-app/.gitignore vendored Normal file
View file

@ -0,0 +1,135 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
pip-wheel-metadata/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don’t work, or not
# install all needed dependencies.
#Pipfile.lock
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# Azure Functions artifacts
bin
obj
appsettings.json
local.settings.json
# Azurite artifacts
__blobstorage__
__queuestorage__
__azurite_db*__.json
.python_packages

6
azure-app/.vscode/extensions.json vendored Normal file
View file

@ -0,0 +1,6 @@
{
"recommendations": [
"ms-azuretools.vscode-azurefunctions",
"ms-python.python"
]
}

15
azure-app/.vscode/launch.json vendored Normal file
View file

@ -0,0 +1,15 @@
{
"version": "0.2.0",
"configurations": [
{
"name": "Attach to Python Functions",
"type": "debugpy",
"request": "attach",
"connect": {
"host": "localhost",
"port": 9091
},
"preLaunchTask": "func: host start"
}
]
}

9
azure-app/.vscode/settings.json vendored Normal file
View file

@ -0,0 +1,9 @@
{
"azureFunctions.deploySubpath": ".",
"azureFunctions.scmDoBuildDuringDeployment": true,
"azureFunctions.pythonVenv": ".venv",
"azureFunctions.projectLanguage": "Python",
"azureFunctions.projectRuntime": "~4",
"debug.internalConsoleOptions": "neverOpen",
"azureFunctions.projectLanguageModel": 2
}

27
azure-app/.vscode/tasks.json vendored Normal file
View file

@ -0,0 +1,27 @@
{
"version": "2.0.0",
"tasks": [
{
"type": "func",
"label": "func: host start",
"command": "host start",
"problemMatcher": "$func-python-watch",
"isBackground": true,
"dependsOn": "pip install (functions)"
},
{
"label": "pip install (functions)",
"type": "shell",
"osx": {
"command": "${config:azureFunctions.pythonVenv}/bin/python -m pip install -r requirements.txt"
},
"windows": {
"command": "${config:azureFunctions.pythonVenv}\\Scripts\\python -m pip install -r requirements.txt"
},
"linux": {
"command": "${config:azureFunctions.pythonVenv}/bin/python -m pip install -r requirements.txt"
},
"problemMatcher": []
}
]
}

File diff suppressed because it is too large Load diff

73
azure-app/function_app.py Normal file
View file

@ -0,0 +1,73 @@
import azure.functions as func
import logging
import json
import os
import sys
# Ajouter le dossier utils au path pour les imports
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from utils.data_loader import load_embeddings, load_user_data
from utils.content_based import get_recommendations
app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)
@app.route(route="recommend")
def recommend(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Recommendation request received.')
# Récupérer le user_id depuis les paramètres de requête
user_id = req.params.get('user_id')
if not user_id:
try:
req_body = req.get_json()
user_id = req_body.get('user_id') if req_body else None
except ValueError:
pass
if not user_id:
return func.HttpResponse(
json.dumps({"error": "user_id parameter required"}),
status_code=400,
mimetype="application/json"
)
try:
# Convertir en entier
user_id = int(user_id)
# Générer les recommandations
recommendations = get_recommendations(user_id)
if recommendations:
return func.HttpResponse(
json.dumps({
"user_id": user_id,
"recommendations": recommendations
}),
status_code=200,
mimetype="application/json"
)
else:
return func.HttpResponse(
json.dumps({
"error": f"No recommendations found for user {user_id}"
}),
status_code=404,
mimetype="application/json"
)
except ValueError:
return func.HttpResponse(
json.dumps({"error": "user_id must be a valid integer"}),
status_code=400,
mimetype="application/json"
)
except Exception as e:
logging.error(f"Error generating recommendations: {str(e)}")
return func.HttpResponse(
json.dumps({"error": "Internal server error"}),
status_code=500,
mimetype="application/json"
)

15
azure-app/host.json Normal file
View file

@ -0,0 +1,15 @@
{
"version": "2.0",
"logging": {
"applicationInsights": {
"samplingSettings": {
"isEnabled": true,
"excludedTypes": "Request"
}
}
},
"extensionBundle": {
"id": "Microsoft.Azure.Functions.ExtensionBundle",
"version": "[4.*, 5.0.0)"
}
}

View file

@ -0,0 +1,4 @@
azure-functions
numpy
scikit-learn
pandas

92
azure-app/test_local.py Normal file
View file

@ -0,0 +1,92 @@
import logging
import sys
import os
# Configuration du logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
# Ajouter le dossier utils au path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from utils.content_based import get_recommendations
from utils.data_loader import get_available_users, check_data_files, load_embeddings
def test_data_loading():
"""Test du chargement des données."""
print("🔄 Test du chargement des données...")
if not check_data_files():
print("❌ Fichiers de données manquants")
return False
try:
# Test chargement embeddings
embeddings = load_embeddings()
print(f"✅ Embeddings chargés: {embeddings.shape}")
# Test chargement utilisateurs
users = get_available_users()
print(f"✅ Utilisateurs chargés: {len(users)} utilisateurs")
print(f" Premiers utilisateurs: {users[:5]}")
return True
except Exception as e:
print(f"❌ Erreur lors du chargement: {e}")
return False
def test_recommendations():
"""Test des recommandations."""
print("\n🔄 Test des recommandations...")
users = get_available_users()
test_users = users[:3] if len(users) >= 3 else users
for user_id in test_users:
print(f"\n👤 Test pour utilisateur {user_id}:")
try:
recommendations = get_recommendations(user_id, n_recommendations=5)
if recommendations:
print(f"✅ {len(recommendations)} recommandations générées")
print(f" Articles recommandés: {recommendations}")
else:
print("⚠️ Aucune recommandation générée")
except Exception as e:
print(f"❌ Erreur pour utilisateur {user_id}: {e}")
def test_edge_cases():
"""Test des cas limites."""
print("\n🔄 Test des cas limites...")
# Test utilisateur inexistant
print("Test utilisateur inexistant (ID: 99999):")
result = get_recommendations(99999)
if result is None:
print("✅ Utilisateur inexistant géré correctement")
else:
print(f"⚠️ Résultat inattendu: {result}")
def main():
"""Fonction principale de test."""
print("🚀 Démarrage des tests locaux\n")
# Test 1: Chargement des données
if not test_data_loading():
print("\n❌ Tests interrompus - problème de chargement")
return
# Test 2: Recommandations
test_recommendations()
# Test 3: Cas limites
test_edge_cases()
print("\n✨ Tests terminés")
if __name__ == "__main__":
main()

View file

View file

@ -0,0 +1,79 @@
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from typing import List, Optional, Tuple
import logging
from .data_loader import load_embeddings, get_user_articles
def _calculate_mean_embedding(embeddings: np.ndarray, article_ids: List[int]) -> np.ndarray:
"""Calcule la moyenne des embeddings pour une liste d'articles."""
valid_ids = [aid for aid in article_ids if aid < len(embeddings)]
if not valid_ids:
raise ValueError("No valid article IDs found")
articles_embeddings = embeddings[valid_ids]
return np.mean(articles_embeddings, axis=0).reshape(1, -1)
def _compute_similarities(reference_embedding: np.ndarray, embeddings: np.ndarray) -> np.ndarray:
"""Calcule la similarité cosinus entre l'embedding de référence et tous les embeddings."""
return cosine_similarity(reference_embedding, embeddings)[0]
def _filter_candidates(similarities: np.ndarray, clicked_articles: List[int]) -> List[Tuple[int, float]]:
"""Filtre les candidats en excluant les articles déjà vus."""
candidates = []
for article_id, similarity in enumerate(similarities):
if article_id not in clicked_articles:
candidates.append((article_id, similarity))
return candidates
def _get_top_recommendations(candidates: List[Tuple[int, float]], n_recommendations: int) -> List[int]:
"""Trie les candidats et retourne le top N."""
candidates.sort(key=lambda x: x[1], reverse=True)
return [article_id for article_id, _ in candidates[:n_recommendations]]
def _validate_user_articles(clicked_articles: List[int], embeddings: np.ndarray) -> List[int]:
"""Valide et filtre les articles dans les limites des embeddings."""
valid_articles = [art for art in clicked_articles if 0 <= art < len(embeddings)]
if not valid_articles:
raise ValueError("No valid articles found for user")
return valid_articles
def get_recommendations(user_id: int, n_recommendations: int = 5) -> Optional[List[int]]:
"""
Génère des recommandations pour un utilisateur donné.
Args:
user_id: ID de l'utilisateur
n_recommendations: Nombre de recommandations à retourner
Returns:
Liste des article_ids recommandés ou None si erreur
"""
try:
# Charger les données
embeddings = load_embeddings()
clicked_articles = get_user_articles(user_id)
if not clicked_articles:
logging.warning(f"No articles found for user {user_id}")
return None
# Valider les articles
valid_articles = _validate_user_articles(clicked_articles, embeddings)
# Calculer l'embedding de référence (stratégie mean)
reference_embedding = _calculate_mean_embedding(embeddings, valid_articles)
# Calculer similarités
similarities = _compute_similarities(reference_embedding, embeddings)
# Filtrer et trier
candidates = _filter_candidates(similarities, clicked_articles)
recommendations = _get_top_recommendations(candidates, n_recommendations)
logging.info(f"Generated {len(recommendations)} recommendations for user {user_id}")
return recommendations
except Exception as e:
logging.error(f"Error generating recommendations for user {user_id}: {str(e)}")
return None

View file

@ -0,0 +1,87 @@
import pickle
import pandas as pd
import os
import logging
from typing import Dict, List, Optional
import numpy as np
# Chemins des fichiers
DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'data')
EMBEDDINGS_PATH = os.path.join(DATA_DIR, 'embeddings_azure_50d.pkl')
CLICKS_PATH = os.path.join(DATA_DIR, 'clicks_sample.csv')
# Cache global
_embeddings_cache: Optional[np.ndarray] = None
_user_data_cache: Optional[Dict[int, List[int]]] = None
def _load_pickle_file(file_path: str) -> np.ndarray:
"""Charge un fichier pickle."""
with open(file_path, 'rb') as f:
return pickle.load(f)
def _load_csv_file(file_path: str) -> pd.DataFrame:
"""Charge un fichier CSV."""
return pd.read_csv(file_path)
def _process_clicks_data(clicks_df: pd.DataFrame) -> Dict[int, List[int]]:
"""Convertit le DataFrame en dictionnaire user->articles."""
user_data = {}
for _, row in clicks_df.iterrows():
user_id = int(row['user_id'])
article_id = int(row['click_article_id'])
if user_id not in user_data:
user_data[user_id] = []
user_data[user_id].append(article_id)
return user_data
def load_embeddings() -> np.ndarray:
"""Charge les embeddings avec cache."""
global _embeddings_cache
if _embeddings_cache is None:
try:
_embeddings_cache = _load_pickle_file(EMBEDDINGS_PATH)
logging.info(f"Embeddings loaded: {_embeddings_cache.shape}")
except Exception as e:
logging.error(f"Error loading embeddings: {str(e)}")
raise
return _embeddings_cache
def load_user_data() -> Dict[int, List[int]]:
"""Charge les données utilisateur avec cache."""
global _user_data_cache
if _user_data_cache is None:
try:
clicks_df = _load_csv_file(CLICKS_PATH)
_user_data_cache = _process_clicks_data(clicks_df)
logging.info(f"User data loaded: {len(_user_data_cache)} users")
except Exception as e:
logging.error(f"Error loading user data: {str(e)}")
raise
return _user_data_cache
def get_user_articles(user_id: int) -> Optional[List[int]]:
"""Récupère les articles d'un utilisateur."""
user_data = load_user_data()
return user_data.get(user_id)
def check_data_files() -> bool:
"""Vérifie que tous les fichiers de données existent."""
files_to_check = [EMBEDDINGS_PATH, CLICKS_PATH]
for file_path in files_to_check:
if not os.path.exists(file_path):
logging.error(f"Missing data file: {file_path}")
return False
return True
def get_available_users() -> List[int]:
"""Retourne la liste de tous les user_ids disponibles."""
user_data = load_user_data()
return list(user_data.keys())

179
streamlit_app.py Normal file
View file

@ -0,0 +1,179 @@
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()