import streamlit as st from datasets import load_dataset import pandas as pd from sentence_transformers import SentenceTransformer from sklearn.metrics.pairwise import cosine_similarity import plotly.express as px from isodate import parse_duration, ISO8601Error import ast import numpy as np from transformers import pipeline import time from functools import lru_cache import warnings import re import json from PIL import Image import requests from io import BytesIO import traceback warnings.filterwarnings('ignore') # ==================== CONFIGURACIÓN ==================== st.set_page_config( page_title="Recetas Saludables IA 🍎", page_icon="🍎", layout="wide", initial_sidebar_state="expanded" ) # CSS mejorado con más estilos para listas # MEJORA: Mejor presentación st.markdown(""" """, unsafe_allow_html=True) # ==================== FUNCIONES AUXILIARES ==================== def parse_ingredient_string(ing_str): """Parsear cadena de ingredientes del formato R a lista Python""" try: if isinstance(ing_str, list): return ing_str if not isinstance(ing_str, str): return [] # Limpiar la cadena # MEJORA: Más robusto contra errores ing_str = ing_str.strip().replace('NA', '').replace('character(0)', '') # Caso 1: Formato R c("item1", "item2") if ing_str.startswith('c('): # Remover c( y el último paréntesis ing_str = ing_str[2:-1] if ing_str.endswith(')') else ing_str[2:] # Reemplazar comillas dobles escapadas ing_str = ing_str.replace('\\"', '"') # Intentar evaluar como lista Python try: result = ast.literal_eval(ing_str) if isinstance(result, list): return [str(item).strip('"\'') for item in result] else: return [str(result).strip('"\'')] except: # Si falla, dividir por comas items = [item.strip().strip('"\'') for item in ing_str.split(',')] return [item for item in items if item] # Caso 2: Lista JSON-like elif ing_str.startswith('[') and ing_str.endswith(']'): try: result = json.loads(ing_str) if isinstance(result, list): return [str(item) for item in result] except: pass # Caso 3: Separado por comas simple items = [item.strip().strip('"\'') for item in ing_str.split(',')] items = [item for item in items if item] return items except Exception as e: st.warning(f"Error al parsear ingredientes: {e}") return [] def parse_instruction_string(instr_str): """Parsear instrucciones del formato R a lista Python""" try: if isinstance(instr_str, list): return instr_str if not isinstance(instr_str, str): return [] instr_str = instr_str.strip().replace('NA', '').replace('character(0)', '') # MEJORA: Limpieza extra # Si es una cadena JSON-like o lista Python if (instr_str.startswith('[') and instr_str.endswith(']')): try: result = json.loads(instr_str) if isinstance(result, list): return [str(item) for item in result] except: pass # Dividir por puntos o números instructions = [] # Patrón para dividir por números (1., 2., etc.) o puntos patterns = [r'\d+\.', r'\d+\)', r'Step \d+:', r'\n'] for pattern in patterns: if re.search(pattern, instr_str): split_instr = re.split(pattern, instr_str) instructions = [instr.strip() for instr in split_instr if instr.strip()] if len(instructions) > 1: break # Si no se pudo dividir, usar toda la cadena como una instrucción if not instructions: instructions = [instr_str] return instructions except Exception as e: st.warning(f"Error al parsear instrucciones: {e}") return [str(instr_str)] def parse_image_string(img_str): """Parsear URLs de imágenes""" try: if isinstance(img_str, list): return img_str if not isinstance(img_str, str): return [] img_str = img_str.strip() # Formato R c("url1", "url2") if img_str.startswith('c('): img_str = img_str[2:-1] if img_str.endswith(')') else img_str[2:] img_str = img_str.replace('\\"', '"') try: result = ast.literal_eval(img_str) if isinstance(result, list): return [str(item).strip('"\'') for item in result] except: pass # Lista JSON elif img_str.startswith('[') and img_str.endswith(']'): try: result = json.loads(img_str) if isinstance(result, list): return [str(item) for item in result] except: pass # URL única if img_str.startswith('http'): return [img_str] return [] except Exception as e: return [] @st.cache_resource(show_spinner="Cargando modelo de traducción...") def load_translator(): return pipeline("translation_en_to_es", model="Helsinki-NLP/opus-mt-en-es") @st.cache_resource(show_spinner="Cargando modelo de embeddings...") def load_embedding_model(): return SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2') # MEJORA: Modelo multilingüe para español y sinónimos @lru_cache(maxsize=1000) def traducir_texto_cached(texto): """Cachea traducciones para mejorar rendimiento""" if not texto or pd.isna(texto) or str(texto).strip() == '': return "" try: texto = str(texto) if len(texto) > 500: texto = texto[:497] + "..." translator = load_translator() return translator(texto, max_length=512)[0]['translation_text'] except Exception: return texto def load_image_from_url(url, max_size=(400, 300)): """Cargar imagen desde URL con manejo de errores""" try: if not url or not isinstance(url, str) or not url.startswith('http'): return None response = requests.get(url, timeout=5) response.raise_for_status() img = Image.open(BytesIO(response.content)) # Convertir a RGB si es necesario if img.mode in ('RGBA', 'LA', 'P'): img = img.convert('RGB') # Redimensionar manteniendo aspecto img.thumbnail(max_size, Image.Resampling.LANCZOS) return img except Exception: return None # ==================== CARGA DE DATOS ==================== @st.cache_data(show_spinner="Cargando y procesando datos de recetas...") def load_and_preprocess_data(): """Carga y preprocesa los datos con validación robusta""" try: # Cargar dataset st.info("Descargando dataset de recetas... Esto puede tomar unos segundos.") ds = load_dataset("AkashPS11/recipes_data_food.com", trust_remote_code=True) df = ds['train'].to_pandas() if 'train' in ds else ds.to_pandas() # Limitar tamaño para mejor rendimiento pero mantener variedad df = df.head(8000).copy() # Procesar ingredientes df['ingredients_parsed'] = df['RecipeIngredientParts'].apply(parse_ingredient_string) df['ingredients_str'] = df['ingredients_parsed'].apply( lambda x: ' '.join([str(i).lower() for i in x]) if x else '' ) # Procesar cantidades df['quantities_parsed'] = df['RecipeIngredientQuantities'].apply(parse_ingredient_string) # Procesar instrucciones df['instructions_parsed'] = df['RecipeInstructions'].apply(parse_instruction_string) # Procesar imágenes df['images_parsed'] = df['Images'].apply(parse_image_string) # Filtrar recetas saludables mask = ( (df['Calories'] < 800) | df['Keywords'].str.contains('healthy|low fat|vegan|low calorie|vegetarian', na=False, case=False, regex=True) ) df = df[mask].copy() # Procesar tiempos con manejo de errores def parse_time(time_str): try: if pd.isna(time_str) or not isinstance(time_str, str): return 0 return parse_duration(time_str).total_seconds() / 60 except (ISO8601Error, AttributeError, TypeError): return 0 df['total_minutes'] = df['TotalTime'].apply(parse_time) df['prep_minutes'] = df['PrepTime'].apply(parse_time) df['cook_minutes'] = df['CookTime'].apply(parse_time) # Limpiar valores NaN numeric_cols = ['Calories', 'FatContent', 'SugarContent', 'ProteinContent', 'AggregatedRating'] for col in numeric_cols: if col in df.columns: df[col] = pd.to_numeric(df[col], errors='coerce').fillna(0) # Pre-calcular embeddings para mejor rendimiento st.info("Calculando embeddings para búsqueda rápida...") model = load_embedding_model() ingredients_texts = df['ingredients_str'].fillna('').tolist() # Calcular embeddings en lotes para evitar memory error # MEJORA: Batch más pequeño para mejor rendimiento batch_size = 50 embeddings = [] for i in range(0, len(ingredients_texts), batch_size): batch = ingredients_texts[i:i+batch_size] batch_embeddings = model.encode(batch, show_progress_bar=False) embeddings.extend(batch_embeddings) df['embedding'] = list(embeddings) st.success(f"✅ Dataset cargado: {len(df)} recetas procesadas") return df except Exception as e: st.error(f"Error crítico al cargar datos: {str(e)}") st.error(traceback.format_exc()) return pd.DataFrame() # ==================== FUNCIONES DE RECOMENDACIÓN ==================== @lru_cache(maxsize=50) # MEJORA: Cache extra para recomendaciones repetidas def recommend_recipes_optimized(user_ingredients, category="", top_k=5, max_cal=500, is_vegan=False, max_time=60): """Función de recomendación optimizada con embeddings pre-calculados""" try: if df.empty: return pd.DataFrame() # Crear embedding de consulta del usuario # MEJORA: Soporte multilingüe, no traduce input model = load_embedding_model() user_text = ' '.join([str(i) for i in user_ingredients]) # No lower, modelo maneja user_embedding = model.encode(user_text) # Calcular similitudes usando embeddings pre-calculados embeddings = np.vstack(df['embedding'].values) similarities = cosine_similarity([user_embedding], embeddings)[0] df['similarity'] = similarities # Aplicar filtros mask = ( (df['Calories'] <= max_cal) & (df['total_minutes'] <= max_time) & (df['similarity'] > 0.1) # Umbral más bajo para más resultados ) if category and category != "": mask &= df['RecipeCategory'].str.contains(category, case=False, na=False) if is_vegan: mask &= df['Keywords'].str.contains('vegan', case=False, na=False) filtered = df[mask].copy() if filtered.empty: # Relajar filtros si no hay resultados st.warning("No se encontraron recetas con esos filtros estrictos. Mostrando recetas más similares...") filtered = df[df['similarity'] > 0.05].copy() # Ordenar y retornar recs = filtered.nlargest(top_k, ['similarity', 'AggregatedRating']) return recs except Exception as e: st.error(f"Error en recomendación: {str(e)}") return pd.DataFrame() def get_similar_recipes(recipe_index, top_n=5): """Obtener recetas similares a una receta específica""" try: if df.empty or recipe_index not in df.index: return pd.DataFrame() # Obtener embedding de la receta de referencia target_embedding = df.loc[recipe_index, 'embedding'].reshape(1, -1) # Calcular similitudes con todas las recetas embeddings = np.vstack(df['embedding'].values) similarities = cosine_similarity(target_embedding, embeddings)[0] # Obtener índices de las recetas más similares (excluyendo la receta misma) similar_indices = np.argsort(similarities)[::-1][1:top_n+1] similar_recipes = df.iloc[similar_indices].copy() similar_recipes['similarity_to_recipe'] = similarities[similar_indices] return similar_recipes except Exception as e: st.error(f"Error al buscar recetas similares: {str(e)}") return pd.DataFrame() # ==================== INTERFAZ STREAMLIT ==================== # Título y descripción st.title("🍎 Generador Inteligente de Recetas Saludables") st.markdown("""
🍎 Generador de Recetas Saludables con IA • Usa modelos de machine learning para encontrar recetas perfectas basadas en tus ingredientes
Powered by Hugging Face 🤗 • Sentence Transformers • Streamlit