| import datetime |
| import json |
| import os |
| from typing import Optional |
|
|
| from utils.sensory_config_utils import extract_sensory_preferences |
|
|
| RESOURCE_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "resources") |
| USER_PROFILES_FILE = os.path.join(RESOURCE_DIR, os.pardir, "database", "user_profiles.json") |
| os.makedirs(os.path.dirname(USER_PROFILES_FILE), exist_ok=True) |
|
|
|
|
| def load_user_profiles(): |
| """Carica i profili utente dal file JSON""" |
| if os.path.exists(USER_PROFILES_FILE): |
| with open(USER_PROFILES_FILE, "r", encoding="utf-8") as f: |
| profiles = json.load(f) |
| return {profile["user_id"]: profile for profile in profiles} |
| return {} |
|
|
|
|
| def save_user_profiles(profiles): |
| """Salva i profili utente nel file JSON""" |
| with open(USER_PROFILES_FILE, "w", encoding="utf-8") as f: |
| json.dump(list(profiles.values()), f, indent=2, ensure_ascii=False) |
|
|
|
|
| def get_user_profile(user_id): |
| """Ottiene il profilo di un utente specifico""" |
| profiles = load_user_profiles() |
| return profiles.get( |
| user_id, |
| { |
| "user_metadata": {"created_at": None, "last_updated": None}, |
| "selected_pois": [], |
| "sensory_aversions": {}, |
| }, |
| ) |
|
|
|
|
| def update_user_selected_pois(user_id, selected_pois): |
| """Aggiorna i POI selezionati per un utente""" |
| profiles = load_user_profiles() |
| current_time = datetime.datetime.now().isoformat() |
|
|
| if user_id not in profiles: |
| profiles[user_id] = { |
| "user_metadata": { |
| "created_at": current_time, |
| "last_updated": current_time, |
| "profile_complete": False, |
| }, |
| "selected_pois": [], |
| "sensory_aversions": {}, |
| } |
|
|
| profiles[user_id]["selected_pois"] = list(selected_pois) |
| profiles[user_id]["user_metadata"]["last_updated"] = current_time |
|
|
| save_user_profiles(profiles) |
|
|
|
|
| def update_user_sensory_aversions(user_id, *radio_values): |
| """Aggiorna le aversioni sensoriali per un utente""" |
| profiles = load_user_profiles() |
| current_time = datetime.datetime.now().isoformat() |
| sensory_aversions = extract_sensory_preferences(radio_values) |
|
|
| if user_id not in profiles: |
| profiles[user_id] = { |
| "user_metadata": { |
| "created_at": current_time, |
| "last_updated": current_time, |
| "profile_complete": False, |
| }, |
| "selected_pois": [], |
| "sensory_aversions": {}, |
| } |
|
|
| profiles[user_id]["sensory_aversions"] = sensory_aversions |
| profiles[user_id]["user_metadata"]["last_updated"] = current_time |
|
|
| save_user_profiles(profiles) |
|
|
|
|
| def mark_profile_complete(user_id): |
| """Marca il profilo di un utente come completo""" |
| profiles = load_user_profiles() |
| current_time = datetime.datetime.now().isoformat() |
|
|
| if user_id in profiles: |
| profiles[user_id]["user_metadata"]["profile_complete"] = True |
| profiles[user_id]["user_metadata"]["last_updated"] = current_time |
| save_user_profiles(profiles) |
| return profiles[user_id] |
|
|
| return None |
|
|
|
|
| def create_new_user_profile(user_id: str) -> dict: |
| """Crea un nuovo profilo utente""" |
| current_time = datetime.datetime.now().isoformat() |
| return { |
| "user_id": user_id.strip(), |
| "user_metadata": { |
| "created_at": current_time, |
| "last_updated": current_time, |
| "profile_complete": False, |
| }, |
| "selected_pois": [], |
| "sensory_aversions": {}, |
| } |
|
|
|
|
| def check_user_exists(user_id: str) -> tuple[bool, Optional[dict]]: |
| """Controlla se un utente esiste e restituisce il suo profilo""" |
| if not user_id or not user_id.strip(): |
| return False, None |
|
|
| profiles = load_user_profiles() |
| user_profile = profiles.get(user_id.strip()) |
|
|
| if user_profile: |
| return True, user_profile |
| else: |
| return False, None |
|
|
|
|
| def login_user(user_id: str) -> tuple[bool, str, str, dict]: |
| """Gestisce il login dell'utente""" |
| if not user_id or not user_id.strip(): |
| return False, "⚠️ Inserisci un ID utente valido", "", {} |
|
|
| user_id = user_id.strip() |
| user_exists, user_profile = check_user_exists(user_id) |
|
|
| if user_exists: |
| profile_complete = user_profile.get("user_metadata", {}).get( |
| "profile_complete", False |
| ) |
| last_updated = user_profile.get("user_metadata", {}).get("last_updated", "N/A") |
|
|
| if profile_complete: |
| message = f"✅ Benvenuto/a {user_id}! Profilo completo trovato. Reindirizzamento al sistema di raccomandazione..." |
| return True, message, "existing_complete" |
| else: |
| message = f"📝 Utente {user_id} trovato ma profilo incompleto. Completare preferenze e aversioni sensoriali..." |
| return True, message, "existing_incomplete" |
| else: |
| message = f"🆕 Nuovo utente {user_id}. Creazione profilo in corso..." |
| new_profile = create_new_user_profile(user_id) |
|
|
| |
| profiles = load_user_profiles() |
| profiles[user_id] = new_profile |
| save_user_profiles(profiles) |
|
|
| return True, message, "new_user" |
|
|