import base64 import json import asyncio import re import uuid import random import traceback from datetime import datetime from typing import Dict, List, Optional, Tuple from fastapi import HTTPException from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from openai import OpenAI from dotenv import load_dotenv import os from fitness_app.data_manager import ( Meal, MealPlanResponse, SwapMealResponse, FoodScanResponse, HarmfulIngredient, SafeIngredient, NeutralIngredient, Ingredient, Macronutrients, DailyNutritionResponse, SwapMealRequest, FoodTypeInfo, food_types_for_scan, ) from fitness_app.vector_store import MealGenVectorStore, ScannerVectorStore load_dotenv() LLM_SEMAPHORE = asyncio.Semaphore(500) IMAGE_SEMAPHORE = asyncio.Semaphore(100) # Cooking adjectives the LLM prepends to ingredient names — strip these _COOKING_ADJ = re.compile( r'\b(grilled|roasted|baked|fried|steamed|fresh|smoked|saut[eé]ed|' r'pan[-\s]seared|boiled|raw|cooked|diced|sliced|chopped|minced|' r'shredded|whole|dried|canned|frozen|organic|lean|light|dark)\s+', re.IGNORECASE, ) # ───────────────────────────────────────────────────────────────────────────── # Normalisation helpers # ───────────────────────────────────────────────────────────────────────────── def _normalize_blood_type(blood_type: str) -> str: if not blood_type: return "O" cleaned = blood_type.strip().upper().split('+')[0].split('-')[0].strip() return cleaned if cleaned in ("O", "A", "B", "AB") else "O" def _normalize_diet_type(diet_type: str) -> str: if not diet_type: return "standard" mapping = { "classic": "standard", "standard": "standard", "vegan": "vegan", "pescatarian": "pescatarian", "vegetarian": "vegetarian", "carnivore": "carnivore", } return mapping.get(diet_type.strip().lower(), "standard") def _food_names(items: List, max_items: int = 80) -> str: """Format food names from str list or dict list for prompt injection.""" if not items: return "None provided" names = [] for item in items[:max_items]: if isinstance(item, dict): name = item.get("food_name") or item.get("name") or "" else: name = str(item) if name: names.append(name) return ", ".join(names) if names else "None provided" def format_blood_group_rules( blood_type: str, beneficial=None, neutral=None, avoid=None, client_rules: Optional[str] = None, ) -> str: """ Build blood-group diet rules for LLM prompts. If the client already sent blood_group_rules (FormData / query), use that. Otherwise build from vector-store beneficial / neutral / avoid lists. """ if client_rules and str(client_rules).strip(): return str(client_rules).strip() return ( f"Blood Type {blood_type} diet rules (ground truth from our food database):\n" f"- Beneficial (prefer / safe): {_food_names(beneficial or [])}\n" f"- Neutral (allowed): {_food_names(neutral or [])}\n" f"- Avoid (do not recommend / harmful): {_food_names(avoid or [])}\n" f"Prefer beneficial foods, allow neutral, never use avoid foods for blood type {blood_type}." ) # ───────────────────────────────────────────────────────────────────────────── # Ingredient name normalisation & DB matching # ───────────────────────────────────────────────────────────────────────────── def _strip_cooking_adjectives(name: str) -> str: """'grilled Chicken Breast' → 'Chicken Breast'""" return _COOKING_ADJ.sub('', name).strip() def _build_name_lookup(food_dicts: List[Dict]) -> Dict[str, str]: """ Build {lowercase_name: exact_db_name} from a list of food dicts. Used for post-generation normalisation of LLM ingredient output. """ lookup: Dict[str, str] = {} for f in food_dicts: name = f.get("food_name", "") if name: lookup[name.lower()] = name return lookup def _match_to_db_name(ingredient: str, lookup: Dict[str, str]) -> Optional[str]: """ Try to map an LLM-output ingredient string to the exact DB name. 1. Strip cooking adjectives, exact match. 2. Substring match (DB name inside ingredient, or ingredient inside DB name). Returns exact DB name if found, else None. """ cleaned = _strip_cooking_adjectives(ingredient).lower().strip() # exact if cleaned in lookup: return lookup[cleaned] # DB name is a substring of the cleaned ingredient (e.g. "salmon fillet" → "salmon") for lower, original in lookup.items(): if lower in cleaned: return original # cleaned ingredient is a substring of a DB name (e.g. "rice" → "Brown Rice") for lower, original in lookup.items(): if cleaned in lower: return original return None def _normalise_meal_ingredients( meal_dict: Dict, lookup: Dict[str, str], ) -> Dict: """ Walk through a meal's ingredient list and replace each name with the exact DB name where possible. Quantity and icon are preserved. """ new_ingredients = [] for ing in meal_dict.get("ingredients", []): raw_name = ing.get("name", "") db_name = _match_to_db_name(raw_name, lookup) new_ingredients.append({ "name": db_name if db_name else raw_name, "quantity": ing.get("quantity", ""), "icon": ing.get("icon", "🍽️"), }) meal_dict["ingredients"] = new_ingredients return meal_dict # ───────────────────────────────────────────────────────────────────────────── # Priority-weighted sampling # ───────────────────────────────────────────────────────────────────────────── PRIORITY_WEIGHTS: Dict[str, int] = {"High": 70, "Medium": 30, "Low": 10} MEAL_TIME_ELIGIBLE = {"High", "Medium", "Low"} # Foods with this meal_time value in DB are eligible for that meal def _is_eligible_for_meal(food: Dict, meal_time: str) -> bool: # Only High and Medium are considered "eligible" in the first pass return food.get(meal_time.lower(), "Medium") in {"High", "Medium"} def _weighted_sample_no_replace( foods: List[Dict], n: int, meal_time: str, used_names: Optional[set] = None, ) -> List[Dict]: if used_names is None: used_names = set() # 1. Strict eligibility (High + Medium), honouring used_names eligible = [ f for f in foods if _is_eligible_for_meal(f, meal_time) and f.get("food_name", "").lower() not in used_names ] # 2. If not enough, relax to include Low priority foods (still honouring used_names) if len(eligible) < n: eligible = [ f for f in foods if f.get(meal_time.lower(), "Medium") in {"High", "Medium", "Low"} and f.get("food_name", "").lower() not in used_names ] # 3. If still not enough, drop the used‑names constraint if len(eligible) < n: eligible = [ f for f in foods if f.get(meal_time.lower(), "Medium") in {"High", "Medium", "Low"} ] # Split the final list by priority high_med = [f for f in eligible if f.get("priority", "Medium") in {"High", "Medium"}] low = [f for f in eligible if f.get("priority", "Medium") == "Low"] result = [] # Use all high/medium first, weighted if len(high_med) >= n: pool = list(high_med) weights = [PRIORITY_WEIGHTS.get(f["priority"], 30) for f in pool] while pool and len(result) < n: idx = random.choices(range(len(pool)), weights=weights, k=1)[0] result.append(pool.pop(idx)) weights.pop(idx) else: result.extend(high_med) remaining = n - len(high_med) # Fill remaining with Low priority foods if low and remaining > 0: pool = list(low) weights = [PRIORITY_WEIGHTS["Low"]] * len(pool) for _ in range(min(remaining, len(pool))): idx = random.choices(range(len(pool)), weights=weights, k=1)[0] result.append(pool.pop(idx)) weights.pop(idx) return result[:n] # ───────────────────────────────────────────────────────────────────────────── # Category-aware food selection per meal time # ───────────────────────────────────────────────────────────────────────────── _CATEGORY_ALIASES: Dict[str, List[str]] = { # Maps logical category names to the actual values in the DB "category" field. "Protein": ["Protein", "Seafood", "Dairy / Protein", "Carb / Protein", "Legume / Protein", "Dairy"], "Carbs": ["Carb", "Carb / Legume", "Carb / Nut", "Legume / Carb", "Carb / Vegetable", "Vegetable / Carb", "Carb / Protein"], "Vegetables": ["Vegetable", "Vegetable / Carb", "Carb / Vegetable", "Vegetable / corn"], "Fruits": ["Fruit", "Fruit / Seasoning"], "Fats": ["Fat / Nut", "Fat / Seed", "Fat / Spread", "Fat / Oil", "Fat / Dairy"], } _MEAL_CATEGORIES: Dict[str, List[str]] = { "breakfast": ["Protein", "Carbs", "Fruits", "Fats"], "lunch": ["Protein", "Carbs", "Vegetables", "Fats"], "dinner": ["Protein", "Carbs", "Vegetables", "Fats"], } # Per-category pool sizes sent to the LLM _POOL_SIZES = { "Protein": 25, "Carbs": 20, "Vegetables": 15, "Fruits": 14, "Fats": 13 } # Per-category pool sizes for swap-meal (larger for 3 distinct alternatives) _SWAP_POOL_SIZES: Dict[str, int] = { "Protein": 10, "Carbs": 8, "Vegetables": 8, "Fruits": 6, "Fats": 4, } def _filter_by_logical_category(foods: List[Dict], logical: str) -> List[Dict]: """Return foods whose DB category matches any alias for the logical category.""" aliases = set(_CATEGORY_ALIASES.get(logical, [logical])) return [f for f in foods if f.get("category", "") in aliases] def _select_pool_for_meal( beneficial: List[Dict], neutral: List[Dict], meal_time: str, pool_sizes: Dict[str, int], used_globally: set, ) -> Tuple[Dict[str, List[Dict]], set]: """ For each required category, draw foods using priority-weighted sampling. Returns (selected_by_category, updated_used_names). 70% drawn from beneficial, 30% from neutral. """ categories = _MEAL_CATEGORIES.get(meal_time.lower(), ["Protein", "Carbs", "Vegetables", "Fats"]) selected: Dict[str, List[Dict]] = {} used_this_meal = set(used_globally) ben_names = {f.get("food_name", "").lower() for f in beneficial} for cat in categories: total_n = pool_sizes.get(cat, 5) n_ben = max(1, round(total_n * 0.70)) n_neu = max(1, total_n - n_ben) ben_cat = _filter_by_logical_category(beneficial, cat) neu_cat = _filter_by_logical_category(neutral, cat) if len(ben_cat) < n_ben: n_neu += n_ben - len(ben_cat) n_ben = len(ben_cat) sel_ben = _weighted_sample_no_replace(ben_cat, n_ben, meal_time, used_this_meal) sel_neu = _weighted_sample_no_replace(neu_cat, n_neu, meal_time, used_this_meal) combined = sel_ben + sel_neu # tag each food so the prompt is clear for f in combined: f["_compat"] = "Beneficial" if f.get("food_name", "").lower() in ben_names else "Neutral" selected[cat] = combined for f in combined: used_this_meal.add(f.get("food_name", "").lower()) return selected, used_this_meal def _format_pool_for_prompt( selected: Dict[str, List[Dict]], meal_time: str, ) -> str: """ Format the pre-selected pool into a compact, unambiguous prompt block. Lists EXACT food names from the DB. LLM must use these exact names. """ lines = [ f"=== PRE-SELECTED FOOD POOL FOR {meal_time.upper()} ===", "RULE: Use ONLY the EXACT food names listed below in your ingredient lists.", " ★★ = High priority (use as primary ingredient, ~70% of calorie contribution)", " ★ = Medium priority (supporting/variety role, ~30%)", " [B] = Beneficial for blood type [N] = Neutral", "", ] for cat, foods in selected.items(): if not foods: continue lines.append(f" [{cat.upper()}]") for f in foods: name = f.get("food_name", "") priority = f.get("priority", "Medium") compat = f.get("_compat", "Neutral") tag = "★★" if priority == "High" else "★ " ct = "B" if compat == "Beneficial" else "N" lines.append(f" {tag} [{ct}] {name}") lines.append("") return "\n".join(lines) def _build_full_plan_database( beneficial: List[Dict], neutral: List[Dict], ) -> Tuple[str, Dict[str, str]]: """ Build the food database string for the full 27-meal plan prompt. Also returns a name lookup dict for post-generation normalisation. """ sections = [] used_globally: set = set() for meal_time in ("breakfast", "lunch", "dinner"): selected, used_after = _select_pool_for_meal( beneficial, neutral, meal_time, _POOL_SIZES, used_globally ) sections.append(_format_pool_for_prompt(selected, meal_time)) # Promote used High-priority foods to global exclusion for cat, foods in selected.items(): for f in foods: if f.get("priority") == "High": used_globally.add(f.get("food_name", "").lower()) all_foods = beneficial + neutral name_lookup = _build_name_lookup(all_foods) return "\n\n".join(sections), name_lookup def _build_swap_database( beneficial: List[Dict], neutral: List[Dict], meal_time: str, ) -> Tuple[str, Dict[str, str]]: """ Build the food database for a swap-meal prompt (single meal time). Returns (database_string, name_lookup). """ selected, _ = _select_pool_for_meal( beneficial, neutral, meal_time, _SWAP_POOL_SIZES, set() ) all_foods = beneficial + neutral name_lookup = _build_name_lookup(all_foods) return _format_pool_for_prompt(selected, meal_time), name_lookup # ───────────────────────────────────────────────────────────────────────────── # Legacy helpers kept for daily-nutrition prompt # ───────────────────────────────────────────────────────────────────────────── def _format_food_list_bullets(food_dicts: List[Dict], max_items: int = 50) -> str: if not food_dicts: return "None" return "\n".join( f"• {f['food_name']} ({f.get('category', '')})" if isinstance(f, dict) else f"• {f}" for f in food_dicts[:max_items] ) # ───────────────────────────────────────────────────────────────────────────── # MealImageGenerator # ───────────────────────────────────────────────────────────────────────────── class MealImageGenerator: def __init__(self): self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) def generate_meal_image(self, meal_name, description, ingredients, sub_category=""): ingredient_list = ", ".join( f"{i.get('quantity','some')} {i.get('name','')}" for i in ingredients ) prompt = ( f"Professional food photography of: {meal_name}. " f"Description: {description}. Ingredients: {ingredient_list}. " "Style: High-quality restaurant presentation, natural lighting, appetizing. " "No text, labels, or logos." ) try: return self._generate_image(prompt) except Exception as e: print(f"Image gen failed for {meal_name}: {e}") return None def _generate_image(self, prompt): r = self.client.images.generate( model="gpt-image-1.5", prompt=prompt, size="1024x1024", quality="low" ) return r.data[0].b64_json def generate_category_image(self, category, prompt, meals_dict): try: enhanced = ( f"{prompt} Category: {category}. " "Style: Professional food photography, clean presentation. " "No text, labels, or logos." ) r = self.client.images.generate( model="gpt-image-1.5", prompt=enhanced[:1000], size="1024x1024", quality="low" ) return r.data[0].b64_json except Exception as e: print(f"Category image failed for {category}: {e}") return self.generate_fallback_category_image(category) def generate_fallback_category_image(self, category): prompts = { "breakfast": "Beautiful breakfast spread, clean white background, professional food photography", "lunch": "Appetizing lunch plate, clean presentation, professional food photography", "dinner": "Delicious dinner, elegant plating, professional restaurant presentation", } prompt = prompts.get(category.lower(), "Professional food photography of a balanced meal") try: r = self.client.images.generate( model="gpt-image-1.5", prompt=prompt, size="1024x1024", quality="low" ) return r.data[0].b64_json except Exception as e: print(f"Fallback image failed: {e}") return None # ───────────────────────────────────────────────────────────────────────────── # EnhancedMealPlanGenerator # ───────────────────────────────────────────────────────────────────────────── class EnhancedMealPlanGenerator: def __init__(self, vector_store: Optional[MealGenVectorStore] = None): self.llm = ChatOpenAI( model="gpt-4o", temperature=0.7, api_key=os.getenv("OPENAI_API_KEY"), ) self.vector_store = vector_store or MealGenVectorStore() self.meal_image_generator = MealImageGenerator() # ── category image prompt ───────────────────────────────────────── self.category_image_prompt = ChatPromptTemplate.from_template(""" You are an expert food photographer. Create a single image prompt representing {category} meals. MEAL SUMMARY: {meal_options_summary} USER: Blood Type {blood_type}, Diet {diet_type}, Country {country} Return ONLY the image prompt text. """) # ── daily nutrition prompt ──────────────────────────── self.daily_nutrition_prompt = ChatPromptTemplate.from_template(""" You are an expert nutritionist specialising in blood type diets. Calculate the optimal daily calorie intake, macronutrient distribution, and meal calorie distribution. USER PROFILE: - Blood Type: {blood_type} - Diet Type: {diet_type} - Country: {country} - Age: {age} years - Current Weight: {weight} kg - Desired Weight: {desired_weight} kg - Height: {height} cm - Main Goal: {main_goal} CALORIE DISTRIBUTION GUIDELINES: Breakfast 25%, Lunch 40%, Dinner 35% BLOOD TYPE MACRONUTRIENT GUIDELINES: - O: Protein 30-40%, Fat 30-35%, Carbs 25-35% - A: Protein 15-25%, Fat 25-30%, Carbs 45-60% - B: Balanced (Carbs 30-40%, Protein 20-30%, Fat 30-35%) - AB: Similar to A but slightly more protein (20-30%) DIET ADJUSTMENTS: - Vegan: plant-based protein sources - Carnivore: Protein 50-60%, Fat 40-50%, minimal carbs - Pescatarian: moderate protein from fish, balanced macros - Vegetarian: plant-based protein with dairy/eggs If desired_weight is provided, adjust calories for 0.5-1 kg/week loss or 0.25-0.5 kg/week gain. Respond ONLY with valid JSON: {{ "total_daily_calories": number, "total_daily_macronutrients": {{"carbohydrates": number, "protein": number, "fat": number}}, "meal_calorie_distribution": {{"breakfast": number, "lunch": number, "dinner": number}} }} carbs*4 + protein*4 + fat*9 ≈ total_daily_calories. """) # ── meal plan prompt ────────────────────────────────────────────── self.meal_plan_prompt = ChatPromptTemplate.from_template(""" You are an expert nutritionist specialising in blood type diets. Generate a personalised daily meal plan with 3 sub‑categories per meal × 3 options = 27 meals total. USER PROFILE: - Blood Type: {blood_type} | Diet: {diet_type} | Country: {country} - Age: {age} | Weight: {weight} kg | Height: {height} cm - Dislikes: {food_dislikes} | Allergies: {allergies} - Target Daily Calories: {target_calories} | Macros: {total_daily_macronutrients} - Goal: {main_goal} ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ BLOOD GROUP RULES (must follow) ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ {blood_group_rules} ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ BACKEND PRE‑SELECTED FOOD POOL ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ {food_database} ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ═══════════════════════════════════════════════════════ ABSOLUTE RULES (must be followed for every single meal) ═══════════════════════════════════════════════════════ 1. INGREDIENT RULES a. Use ONLY the EXACT food names from the pool above. Copy them character‑for‑case. b. DO NOT add any ingredient not in the pool – no salt, pepper, oil, water, spices, herbs, sauces, stock, etc., unless it is explicitly listed in the pool. c. List EVERY ingredient you use in the `ingredients` array. d. The 3 meals within a sub‑category must use a different primary protein/carb source. Do not repeat the exact same ingredient combination. 2. PRIORITY RULES ★★ High priority → primary ingredient (~70 % of the meal's calories). ★ Medium priority → supporting role (~30 % of calories). You may repeat a high‑priority protein across different meals if it naturally fits; aim for variety in carb sources, vegetables, cooking styles and meal names. 3. CALORIE RANGES (VARY NATURALLY) Breakfast meals: each between {breakfast_cal_min} and {breakfast_cal_max} kcal(use any number which is within the range and look natural). Lunch meals: each between {lunch_cal_min} and {lunch_cal_max} kcal(use any number which is within the range and look natural). Dinner meals: each between {dinner_cal_min} and {dinner_cal_max} kcal(use any number which is within the range and look natural). ⚠ Do NOT make all meals in a meal‑time identical in calories. Vary them naturally across the 9 options, keeping each within its range. The numbers needs to vary across the 3 sub‑categories and 3 options and needs to use variety of different calorie values within each one range, not the same repeated. 4. CALORIE MATHS total_calories = (carbohydrates × 4) + (protein × 4) + (fat × 9) Must equal the stated calorie value. Every ingredient must have a quantity in grams (g) or millilitres (ml). The macronutrient totals must be the sum of the listed ingredients. 5. PORTION ADJUSTMENT ORDER (when tuning to hit target) Adjust in this strict order: ① carbs → ② protein → ③ fats/oils/seeds. Never add an ingredient that is not already present just to hit numbers. 6. COUNTRY CONTEXT {country_guidance} ═══════════════════════════════════════════════════════ BANNED MEAL TYPES (NEVER generate these) ═══════════════════════════════════════════════════════ Do NOT generate any of the following meal types under any circumstances: - Wraps, Burritos, Tacos - Rice Bowls, Quinoa Bowls, Comfort Bowls, Power Bowls, Protein Bowls, Noodle Bowls - Any meal named or structured as a "Bowl" of any kind The goal is realistic plated meals, not trendy bowl/wrap formats. ═══════════════════════════════════════════════════════ MEAL NAMING RULES ═══════════════════════════════════════════════════════ Meal names must describe the actual food being served. ✓ GOOD names: "Grilled Salmon with Rice & Broccoli", "Steak with Sweet Potato & Asparagus", "Shrimp Stir Fry", "Blueberry Oatmeal", "Scrambled Eggs with Spinach" ✗ BAD names: "Comfort Bowl", "Power Bowl", "Protein Packed Bowl", "Energy Bowl", "Light & Fresh Bowl", "Nourish Bowl", "Wellness Plate" The name must reflect the protein, cooking method, and key ingredients. ═══════════════════════════════════════════════════════ BREAKFAST TEMPLATES – CHOOSE ONE STYLE PER MEAL ═══════════════════════════════════════════════════════ Every breakfast meal must belong to exactly ONE of the four templates below. NEVER mix ingredients from different templates in the same meal. Target breakfast distribution across the 9 breakfast meals: ~30% Oatmeal (≈3 meals) | ~30% Yogurt Bowls (≈3 meals) | ~30% Egg-Based (≈3 meals) | ~10% Other Approved ■ OATMEAL BREAKFAST Must contain: oats + at least 1 fruit + at least 1 approved topping. Approved toppings (only if in pool): Honey, Cinnamon, Walnuts, Almonds, Chia Seeds, Pumpkin Seeds, Flax Seeds. FORBIDDEN: fish, meat, eggs, rice, noodles, savoury proteins, random oils. Example: Oats + Blueberries + Walnuts + Honey → "Blueberry Walnut Oatmeal" ■ YOGURT BOWL BREAKFAST Must contain: yogurt + at least 1 fruit + at least 1 approved topping. Approved toppings (only if in pool): Honey, Walnuts, Almonds, Chia Seeds, Pumpkin Seeds, Flax Seeds, Granola. FORBIDDEN: fish, meat, eggs, rice, noodles, savoury proteins. Example: Greek Yogurt + Strawberries + Chia Seeds + Honey → "Strawberry Yogurt Bowl" ■ EGG BREAKFAST Allowed: eggs + vegetables + optional approved carb (e.g. potato, bread if in pool) + optional approved fat/oil. FORBIDDEN: fruit, berries, yogurt, kefir, honey, oats. Example: scrambled eggs + spinach + whole‑grain toast + olive oil. ■ SAVOURY BREAKFAST Allowed: eggs, fish/tofu (ONLY if breakfast priority is Medium or High), rice/potato/vegetables (ONLY if breakfast priority Medium or High). FORBIDDEN: oats, berries, fruit, yogurt, kefir, honey, sweet ingredients. Example: smoked salmon + scrambled eggs + sautéed vegetables. ═══════════════════════════════════════════════════════ LUNCH & DINNER MEAL TYPES & DISTRIBUTION ═══════════════════════════════════════════════════════ Across the 9 lunch meals and 9 dinner meals, use this distribution: • 70% Standard Plated Meals (≈6 meals per meal time) • 20% Stir Fries (≈2 meals per meal time) • 10% Salads (≈1 meal per meal time) ■ STANDARD PLATED MEAL (most common) Structure: Protein + Carbohydrate + Vegetable Every meal must contain: • a clear main protein (100‑180 g cooked) • a carb (rice/pasta/noodles 60‑90 g dry; potatoes 150‑250 g) • vegetables (50‑150 g) • optional fats (5‑15 g) All quantities in grams or millilitres; no cups or tablespoons. Examples: "Grilled Salmon with Rice & Broccoli", "Steak with Sweet Potato & Asparagus", "Turkey with Potatoes & Kale", "Cod with Rice & Green Beans" ■ STIR FRY Structure: Protein + Vegetables + Optional Carbohydrate Must contain a clear protein source and at least 2 vegetables. Carbohydrate is optional (e.g. rice noodles or rice on the side). Examples: "Chicken & Vegetable Stir Fry", "Shrimp Stir Fry with Broccoli & Peppers" ■ SALAD Structure: Protein + Salad Vegetables + Optional Carb or Fat Must contain a clear protein source; not just leaves. Examples: "Grilled Chicken Salad", "Tuna Salad with Mixed Greens" ═══════════════════════════════════════════════════════ SEASONING & FLAVOUR RULES ═══════════════════════════════════════════════════════ Meals must not be plain. Add suitable seasonings from the pool where available. If the following are in the pool, apply them to matching proteins: • Fish dishes → Lemon, Garlic, Dill, Parsley • Beef dishes → Garlic, Rosemary, Black Pepper • Turkey dishes → Garlic, Rosemary, Thyme • Vegetables → Herbs, Garlic, Black Pepper ═══════════════════════════════════════════════════════ PROTEIN ROTATION & VARIETY RULES ═══════════════════════════════════════════════════════ • Do NOT use the same primary protein in consecutive meals within a meal time. • Rotate proteins across all 9 meals in each meal time – avoid repeating the same protein more than twice across lunch options or more than twice across dinner options. • Rotate fruits, vegetables, carbohydrates, and breakfast toppings for variety. • The same meal must not appear more than once across the 27 generated meals. ═══════════════════════════════════════════════════════ BANNED PAIRINGS (never appear together in ANY meal) ═══════════════════════════════════════════════════════ - fish + berries / fruit / oats / yogurt / kefir - meat + berries / fruit - rice + yogurt / kefir - noodles for breakfast - oil as a main ingredient - random fruit in savoury lunch/dinner meals - sweet ingredients (fruit, berries, yogurt, kefir, honey) in any lunch/dinner ═══════════════════════════════════════════════════════ MEAL COMPLEXITY DISTRIBUTION ═══════════════════════════════════════════════════════ Target: 80% Simple Meals | 20% Complex/Specialty Meals Simple = Protein + Vegetable + Carbohydrate (easy to prepare) Complex = Stir Fries or specialty preparations Keep most meals straightforward and realistic to cook. ═══════════════════════════════════════════════════════ SUB‑CATEGORY DEFINITIONS (3 per meal time) ═══════════════════════════════════════════════════════ 1. PROTEIN‑PACKED - Feels high‑protein, filling, gym/performance focused. - Protein clearly dominates the meal (>30 g). - Larger approved protein portions, moderate carbs, balanced fats. - Avoid: fruit‑heavy, low‑protein, snack‑style meals. - Calorie feel: substantial, higher than other sub‑categories. 2. LIGHT & FRESH - Feels lighter, refreshing, cleaner, easier to digest. - Lighter approved protein portions, more fruit/vegetables, simpler combinations. - Avoid: heavy oily meals, dense carbs, oversized portions, greasy meals. - Calorie feel: lighter, lower than Protein‑Packed. 3. HEALTHY & COMFORTING - Feels warm, balanced, comforting, realistic, healthy but satisfying. - Balanced protein/carb/fat, softer comforting foods, everyday combinations. - Avoid: random ingredient combos, ultra‑light or extremely heavy meals. - Calorie feel: balanced, moderate. ═══════════════════════════════════════════════════════ MEAL QUALITY TEST ═══════════════════════════════════════════════════════ Before finalising each meal, ask: "Would a normal person realistically cook, order, or eat this?" If NO → reject and regenerate. Requirements: clear protein, realistic ingredient combinations, descriptive meal name, proper cooking method in description. ═══════════════════════════════════════════════════════ OUTPUT FORMAT (valid JSON only, no extra text) ═══════════════════════════════════════════════════════ {{ "breakfast_options": {{ "protein_packed": [ meal_obj, meal_obj, meal_obj ], "light_fresh": [ meal_obj, meal_obj, meal_obj ], "healthy_comforting": [ meal_obj, meal_obj, meal_obj ] }}, "lunch_options": {{ ... }}, "dinner_options": {{ ... }} }} Where each meal_obj is: {{ "meal_name": "string", "category": "Breakfast|Lunch|Dinner", "sub_category": "Protein-Packed|Light & Fresh|Healthy & Comforting", "total_calories": number, "macronutrients": {{"carbohydrates": number, "protein": number, "fat": number}}, "description": "string", "ingredients": [{{"name": "EXACT_POOL_NAME", "quantity": "Xg or Xml", "icon": "emoji"}}], "number_of_servings": 1 }} """) # ── swap meal prompt ────────────────────────────────────────────────────────────────── self.swap_meal_prompt = ChatPromptTemplate.from_template(""" You are an expert nutritionist. Generate 3 COMPLETELY DISTINCT alternative {category} meals for the {sub_category} sub‑category. USER: Blood Type {blood_type} | Diet {diet_type} | Country {country} Target: {target_calories} kcal ± 5 | Allergies: {allergies} | Dislikes: {food_dislikes} ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ BACKEND PRE‑SELECTED FOOD POOL FOR {category_upper} ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ {food_database} ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ═══════════════════════════════════════════════════════ ABSOLUTE RULES (must be followed for every single meal) ═══════════════════════════════════════════════════════ 1. INGREDIENT RULES a. Use ONLY the EXACT food names from the pool above. Copy them character‑for‑case. b. DO NOT add any ingredient not in the pool – no salt, pepper, oil, water, spices, herbs, sauces, stock, etc., unless it is explicitly listed in the pool. c. List EVERY ingredient you use in the `ingredients` array. d. The 3 meals within a sub‑category must use a different primary protein/carb source. Do not repeat the exact same ingredient combination. 2. PRIORITY RULES ★★ High priority → primary ingredient (~70 % of the meal's calories). ★ Medium priority → supporting role (~30 % of calories). You may repeat a high‑priority protein across different meals if it naturally fits; aim for variety in carb sources, vegetables, cooking styles and meal names. 3. CALORIE MATHS total_calories = (carbs×4)+(protein×4)+(fat×9). Must match {target_calories} ±5 kcal. Every ingredient must have a quantity in g or ml. Adjust in order: carbs → protein → fats. 4. PORTION ADJUSTMENT ORDER (when tuning to hit target) Adjust in this strict order: ① carbs → ② protein → ③ fats/oils/seeds. Never add an ingredient that is not already present just to hit numbers. 5. COUNTRY CONTEXT {country_guidance} ═══════════════════════════════════════════════════════ BANNED MEAL TYPES (NEVER generate these) ═══════════════════════════════════════════════════════ Do NOT generate any of the following meal types: - Wraps, Burritos, Tacos - Rice Bowls, Quinoa Bowls, Comfort Bowls, Power Bowls, Protein Bowls, Noodle Bowls - Any meal named or structured as a "Bowl" of any kind The goal is realistic plated meals, not trendy bowl/wrap formats. ═══════════════════════════════════════════════════════ MEAL NAMING RULES ═══════════════════════════════════════════════════════ Meal names must describe the actual food. ✓ GOOD: "Grilled Salmon with Rice & Broccoli", "Shrimp Stir Fry", "Blueberry Oatmeal" ✗ BAD: "Comfort Bowl", "Power Bowl", "Protein Packed Bowl", "Energy Bowl", "Nourish Bowl" The name must reflect the protein, cooking method, and key ingredients. ═══════════════════════════════════════════════════════ BREAKFAST TEMPLATES – CHOOSE ONE STYLE PER MEAL ═══════════════════════════════════════════════════════ Every breakfast meal must belong to exactly ONE of the four templates below. NEVER mix ingredients from different templates in the same meal. ■ OATMEAL BREAKFAST Must contain: oats + at least 1 fruit + at least 1 approved topping. Approved toppings (only if in pool): Honey, Cinnamon, Walnuts, Almonds, Chia Seeds, Pumpkin Seeds, Flax Seeds. FORBIDDEN: fish, meat, eggs, rice, noodles, savoury proteins, random oils. ■ YOGURT BOWL BREAKFAST Must contain: yogurt + at least 1 fruit + at least 1 approved topping. Approved toppings (only if in pool): Honey, Walnuts, Almonds, Chia Seeds, Pumpkin Seeds, Flax Seeds, Granola. FORBIDDEN: fish, meat, eggs, rice, noodles, savoury proteins. ■ EGG BREAKFAST Allowed: eggs + vegetables + optional approved carb (e.g. potato, bread if in pool) + optional approved fat/oil. FORBIDDEN: fruit, berries, yogurt, kefir, honey, oats. Example: scrambled eggs + spinach + whole‑grain toast + olive oil. ■ SAVOURY BREAKFAST Allowed: eggs, fish/tofu (ONLY if breakfast priority is Medium or High), rice/potato/vegetables (ONLY if breakfast priority Medium or High). FORBIDDEN: oats, berries, fruit, yogurt, kefir, honey, sweet ingredients. Example: smoked salmon + scrambled eggs + sautéed vegetables. ═══════════════════════════════════════════════════════ LUNCH & DINNER MEAL TYPES ═══════════════════════════════════════════════════════ ■ STANDARD PLATED MEAL: Protein + Carbohydrate + Vegetable • protein (100‑180 g cooked), carb (60‑90 g dry), vegetables (50‑150 g), optional fats (5‑15 g) • All quantities in grams or millilitres; no cups or tablespoons. • Examples: "Grilled Salmon with Rice & Broccoli", "Steak with Sweet Potato & Asparagus" ■ STIR FRY: Protein + Vegetables + Optional Carbohydrate • Must contain a clear protein source and at least 2 vegetables. • Examples: "Chicken & Vegetable Stir Fry", "Shrimp Stir Fry with Broccoli & Peppers" ■ SALAD: Protein + Salad Vegetables + Optional Carb or Fat • Must contain a clear protein source; not just leaves. • Examples: "Grilled Chicken Salad", "Tuna Salad with Mixed Greens" ═══════════════════════════════════════════════════════ SEASONING & FLAVOUR RULES ═══════════════════════════════════════════════════════ Meals must not be plain. Add suitable seasonings from the pool where available. • Fish dishes → Lemon, Garlic, Dill, Parsley • Beef dishes → Garlic, Rosemary, Black Pepper • Turkey dishes → Garlic, Rosemary, Thyme • Vegetables → Herbs, Garlic, Black Pepper ═══════════════════════════════════════════════════════ BANNED PAIRINGS (never appear together in ANY meal) ═══════════════════════════════════════════════════════ - fish + berries / fruit / oats / yogurt / kefir - meat + berries / fruit - rice + yogurt / kefir - noodles for breakfast - oil as a main ingredient - random fruit in savoury lunch/dinner meals - sweet ingredients (fruit, berries, yogurt, kefir, honey) in any lunch/dinner ═══════════════════════════════════════════════════════ SUB‑CATEGORY DEFINITIONS ═══════════════════════════════════════════════════════ 1. PROTEIN‑PACKED - Feels high‑protein, filling, gym/performance focused. - Protein clearly dominates the meal (>30 g). - Larger approved protein portions, moderate carbs, balanced fats. - Avoid: fruit‑heavy, low‑protein, snack‑style meals. - Calorie feel: substantial, higher than other sub‑categories. 2. LIGHT & FRESH - Feels lighter, refreshing, cleaner, easier to digest. - Lighter approved protein portions, more fruit/vegetables, simpler combinations. - Avoid: heavy oily meals, dense carbs, oversized portions, greasy meals. - Calorie feel: lighter, lower than Protein‑Packed. 3. HEALTHY & COMFORTING - Feels warm, balanced, comforting, realistic, healthy but satisfying. - Balanced protein/carb/fat, softer comforting foods, everyday combinations. - Avoid: random ingredient combos, ultra‑light or extremely heavy meals. - Calorie feel: balanced, moderate. ═══════════════════════════════════════════════════════ MEAL QUALITY TEST ═══════════════════════════════════════════════════════ Before finalising each meal, ask: "Would a normal person realistically cook, order, or eat this?" If NO → reject and regenerate. Requirements: clear protein, realistic combinations, descriptive meal name, proper cooking method. VARIETY: the 3 meals must differ in protein, carb, vegetable, cooking style, and cultural flavour. Return ONLY valid JSON: {{ "alternatives": [ {{ "meal_name": "string", "category": "{category}", "sub_category": "{sub_category}", "total_calories": number, "macronutrients": {{"carbohydrates": number, "protein": number, "fat": number}}, "description": "string", "ingredients": [{{"name": "EXACT_POOL_NAME", "quantity": "Xg or Xml", "icon": "emoji"}}], "number_of_servings": 1 }}, {{...}}, {{...}} ] }} """) # ── static helpers ──────────────────────────────────────────────────── @staticmethod def get_country_guidance(country: str) -> str: g = { "china": "Traditional Chinese cooking: steaming, stir-frying, braising. Ginger, garlic, tofu, vegetables.", "south_korea": "Korean traditions: fermented foods, gochujang, sesame oil, kimchi, varied vegetables.", } if not country: return "Create balanced meals using generally available ingredients." k = country.strip().lower() if k in ("china", "chinese"): return g["china"] if k in ("south korea", "south korean", "korea", "korean"): return g["south_korea"] return f"Create balanced meals incorporating local cuisine from {country} when appropriate." @staticmethod def _meal_structure_text(category: str) -> str: if category.lower() == "breakfast": return "Light protein + carb or fruit + optional fats. No heavy meats." return f"{category}: protein 100-180g cooked + carb 60-90g dry + vegetables 50-150g + optional fats 5-15g." @staticmethod def calculate_target_calories(age, weight, height) -> int: if age and weight and height: bmr = 10 * weight + 6.25 * height - 5 * age + 5 return max(1500, min(4000, int(bmr * 1.55))) return 2000 @staticmethod def _format_blood_type_foods(btf: Dict) -> str: lines = [] for cat, foods in btf.items(): if foods: names = [f["food_name"] if isinstance(f, dict) else f for f in foods] lines.append(f"{cat.replace('_',' ').title()}: {', '.join(names)}") return "\n".join(lines) # ── category image ──────────────────────────────────────────────────── async def generate_category_image_prompt(self, category, meals_dict, user_data): try: summaries = [] if meals_dict: for sub, meals in meals_dict.items(): for i, m in enumerate(meals[:2]): keys = ", ".join(ing.name for ing in m.ingredients[:3]) summaries.append(f"{sub.replace('_',' ').title()} {i+1}: {m.meal_name} — {keys}") summary = "\n".join(summaries) or f"{category} protein-packed, light & fresh, healthy & comforting" chain = self.category_image_prompt | self.llm async with IMAGE_SEMAPHORE: r = await chain.ainvoke({ "category": category, "meal_options_summary": summary, "blood_type": user_data.get("blood_type",""), "diet_type": user_data.get("diet_type","standard"), "country": user_data.get("country",""), }) p = r.content.strip() return p.split("```")[1].strip() if p.startswith("```") else p except Exception as e: print(f"Category image prompt error: {e}") return f"Professional food photography of {category} meals, balanced, appetising, clean background" async def generate_single_category_image(self, category, meals_dict, user_data): try: prompt = await self.generate_category_image_prompt(category, meals_dict, user_data) b64 = self.meal_image_generator.generate_category_image(category, prompt, meals_dict) return (b64, str(uuid.uuid4())[:8]) if b64 else (None, None) except Exception as e: print(f"Category image error {category}: {e}") return (None, None) async def generate_category_images(self, meal_plan, user_data): cats = {"breakfast": meal_plan.breakfast_options, "lunch": meal_plan.lunch_options, "dinner": meal_plan.dinner_options} out = {} for name, md in cats.items(): out[name] = await self.generate_single_category_image(name, md, user_data) if md else (None, None) return out # ── daily nutrition ─────────────────────────────────────────────────── async def calculate_daily_nutrition(self, user_data: Dict) -> DailyNutritionResponse: blood_type = _normalize_blood_type(user_data.get("blood_type", "O")) diet_type = _normalize_diet_type(user_data.get("diet_type", "standard")) country = user_data.get("country", "") chain = self.daily_nutrition_prompt | self.llm async with LLM_SEMAPHORE: r = await chain.ainvoke({ "blood_type": blood_type, "diet_type": diet_type, "country": country or "Not specified", "age": user_data.get("age", "Not specified"), "weight": user_data.get("weight", "Not specified"), "desired_weight": user_data.get("desired_weight", "Not specified"), "height": user_data.get("height", "Not specified"), "main_goal": user_data.get("main_goal", "Stay Fit"), }) content = re.sub(r'^```json\s*', '', r.content.strip()) content = re.sub(r'\s*```$', '', content).strip() try: data = json.loads(content) except json.JSONDecodeError as e: raise HTTPException(500, f"Failed to parse nutrition response: {e}") total = data.get("total_daily_calories", 2000) macros = data.get("total_daily_macronutrients", {}) meal_dist = data.get("meal_calorie_distribution") or { "breakfast": int(total * 0.30), "lunch": int(total * 0.40), "dinner": int(total * 0.30), } return DailyNutritionResponse( total_daily_calories=total, total_daily_macronutrients=Macronutrients( carbohydrates=macros.get("carbohydrates", 250), protein=macros.get("protein", 100), fat=macros.get("fat", 67), ), user_id=user_data.get("user_id", "unknown"), calculation_timestamp=datetime.now(), meal_calorie_distribution=meal_dist, ) # ── generate meal plan ──────────────────────────────────────────────── async def generate_meal_plan( self, user_data: Dict, daily_nutrition: Optional[DailyNutritionResponse] = None, generate_images: bool = False, ) -> MealPlanResponse: blood_type = _normalize_blood_type(user_data.get("blood_type", "O")) diet_type = _normalize_diet_type(user_data.get("diet_type", "standard")) country = user_data.get("country", "") # 1. Fetch full pools ( backend-controlled selection to ensure quality and relevance ) ben = await self.vector_store.get_beneficial_foods(blood_type, diet_type, limit=500) neu = await self.vector_store.get_neutral_foods(blood_type, diet_type, limit=200) print(f"[MealPlan] Pools — beneficial: {len(ben)}, neutral: {len(neu)}") blood_group_rules = format_blood_group_rules( blood_type=blood_type, beneficial=ben, neutral=neu, avoid=[], client_rules=user_data.get("blood_group_rules"), ) # 2. Backend-controlled weighted selection + build prompt database food_database, name_lookup = _build_full_plan_database(ben, neu) # 3. Calorie targets if daily_nutrition: target = daily_nutrition.total_daily_calories macros = daily_nutrition.total_daily_macronutrients dist = daily_nutrition.meal_calorie_distribution or {} else: target = self.calculate_target_calories( user_data.get("age"), user_data.get("weight"), user_data.get("height")) macros = None dist = {} b_cal = dist.get("breakfast", int(target * 0.25)) l_cal = dist.get("lunch", int(target * 0.40)) d_cal = dist.get("dinner", int(target * 0.35)) breakfast_min = b_cal - 5 breakfast_max = b_cal + 5 lunch_min = l_cal - 5 lunch_max = l_cal + 5 dinner_min = d_cal - 5 dinner_max = d_cal + 5 macro_str = json.dumps({ "carbohydrates": macros.carbohydrates, "protein": macros.protein, "fat": macros.fat, }) if macros else "Not specified" # 4. LLM call chain = self.meal_plan_prompt | self.llm async with LLM_SEMAPHORE: r = await chain.ainvoke({ "blood_type": blood_type, "diet_type": diet_type, "country": country or "Not specified", "age": user_data.get("age", "Not specified"), "weight": user_data.get("weight", "Not specified"), "height": user_data.get("height", "Not specified"), "food_dislikes": user_data.get("food_dislikes", "None"), "allergies": user_data.get("allergies", "None"), "target_calories": target, "total_daily_macronutrients": macro_str, "main_goal": user_data.get("main_goal", "Stay Fit"), "meal_calorie_distribution": json.dumps(dist), "breakfast_cal_min": breakfast_min, "breakfast_cal_max": breakfast_max, "lunch_cal_min": lunch_min, "lunch_cal_max": lunch_max, "dinner_cal_min": dinner_min, "dinner_cal_max": dinner_max, "food_database": food_database, "blood_group_rules": blood_group_rules, "country_guidance": self.get_country_guidance(country), }) content = re.sub(r'^```json\s*', '', r.content.strip()) content = re.sub(r'\s*```$', '', content).strip() try: data = json.loads(content) except json.JSONDecodeError as e: print(f"JSON error: {e}\n{content[:400]}") return self._create_fallback_meal_plan(user_data, daily_nutrition) # 5. Validate structure req_cats = ["breakfast_options", "lunch_options", "dinner_options"] req_subs = ["protein_packed", "light_fresh", "healthy_comforting"] for cat in req_cats: if cat not in data: return self._create_fallback_meal_plan(user_data, daily_nutrition) for sub in req_subs: if sub not in data[cat] or len(data[cat][sub]) != 3: return self._create_fallback_meal_plan(user_data, daily_nutrition) # 6. Post-generation ingredient name normalisation # Replace LLM-modified names ("grilled X") with exact DB names ("X") for cat in req_cats: for sub in req_subs: data[cat][sub] = [ _normalise_meal_ingredients(meal, name_lookup) for meal in data[cat][sub] ] # 7. Convert to Meal objects meal_plan = MealPlanResponse( breakfast_options=self._convert_subcats(data["breakfast_options"]), lunch_options= self._convert_subcats(data["lunch_options"]), dinner_options= self._convert_subcats(data["dinner_options"]), user_id= user_data.get("user_id", "unknown"), blood_type= blood_type, diet_type= diet_type, meal_calorie_distribution=data.get("meal_calorie_distribution"), ) if daily_nutrition: meal_plan.total_daily_calories = daily_nutrition.total_daily_calories meal_plan.total_daily_macronutrients = daily_nutrition.total_daily_macronutrients # 8. Optional images if generate_images: imgs = await self.generate_category_images(meal_plan, user_data) meal_plan.breakfast_image = imgs.get("breakfast",(None,None))[0] meal_plan.breakfast_image_id = imgs.get("breakfast",(None,None))[1] meal_plan.lunch_image = imgs.get("lunch", (None,None))[0] meal_plan.lunch_image_id = imgs.get("lunch", (None,None))[1] meal_plan.dinner_image = imgs.get("dinner", (None,None))[0] meal_plan.dinner_image_id = imgs.get("dinner", (None,None))[1] return meal_plan # ── swap meal ───────────────────────────────────────────────────────── async def generate_swap_meals( self, swap_request: SwapMealRequest, generate_images: bool = False, ) -> SwapMealResponse: blood_type = _normalize_blood_type(swap_request.blood_type or "O") diet_type = _normalize_diet_type(swap_request.diet_type or "standard") country = swap_request.country or "" category = swap_request.category or "Lunch" meal_key = category.lower() ben = await self.vector_store.get_beneficial_foods(blood_type, diet_type, limit=300) neu = await self.vector_store.get_neutral_foods(blood_type, diet_type, limit=150) print(f"[SwapMeal] Pools — beneficial: {len(ben)}, neutral: {len(neu)}") food_database, name_lookup = _build_swap_database(ben, neu, meal_key) chain = self.swap_meal_prompt | self.llm async with LLM_SEMAPHORE: r = await chain.ainvoke({ "category": category, "category_upper": category.upper(), "sub_category": swap_request.sub_category or "General", "blood_type": blood_type, "diet_type": diet_type, "country": country or "Not specified", "target_calories": swap_request.current_calories or 600, "allergies": swap_request.allergies or "None", "food_dislikes": swap_request.food_dislikes or "None", "food_database": food_database, "meal_structure": self._meal_structure_text(category), "country_guidance": self.get_country_guidance(country), }) content = re.sub(r'^```json\s*', '', r.content.strip()) content = re.sub(r'\s*```$', '', content).strip() try: data = json.loads(content) except json.JSONDecodeError as e: raise HTTPException(500, f"Failed to parse swap meal response: {e}") if "alternatives" not in data or len(data["alternatives"]) != 3: raise HTTPException(500, "Expected exactly 3 alternative meals") # Normalise ingredient names data["alternatives"] = [ _normalise_meal_ingredients(m, name_lookup) for m in data["alternatives"] ] meals = self._convert_meal_list(data["alternatives"]) if generate_images: for m in meals: m.image = self.meal_image_generator.generate_meal_image( meal_name=m.meal_name, description=m.description, ingredients=[{"name": i.name, "quantity": i.quantity} for i in m.ingredients], sub_category=m.sub_category, ) return SwapMealResponse( alternatives=meals, original_category=category, original_sub_category=swap_request.sub_category or "General", target_calories=swap_request.current_calories or 600, ) # ── fallback ────────────────────────────────────────────────────────── def _create_fallback_meal_plan(self, user_data, daily_nutrition=None): print("Using fallback meal plan") target = (daily_nutrition.total_daily_calories if daily_nutrition else self.calculate_target_calories( user_data.get("age"), user_data.get("weight"), user_data.get("height"))) dist = (daily_nutrition.meal_calorie_distribution or {}) if daily_nutrition else {} b_cal = dist.get("breakfast", int(target * 0.30)) l_cal = dist.get("lunch", int(target * 0.40)) d_cal = dist.get("dinner", int(target * 0.30)) def _fb(category, cal): subs = ["protein_packed", "light_fresh", "healthy_comforting"] return { sc: [ Meal( meal_name=f"{category} {sc.replace('_',' ').title()} {i+1}", category=category, sub_category=sc.replace("_"," ").title(), total_calories=cal, macronutrients=Macronutrients( carbohydrates=cal*0.5/4, protein=cal*0.3/4, fat=cal*0.2/9), description=f"A balanced {sc.replace('_',' ')} {category.lower()}.", ingredients=[ Ingredient(name="Mixed vegetables", quantity="100g", icon="🥗"), Ingredient(name="Protein source", quantity="150g", icon="🍗"), Ingredient(name="Whole grains", quantity="80g", icon="🌾"), ], ) for i in range(3) ] for sc in subs } mp = MealPlanResponse( breakfast_options=_fb("Breakfast", b_cal), lunch_options= _fb("Lunch", l_cal), dinner_options= _fb("Dinner", d_cal), user_id= user_data.get("user_id", "unknown"), blood_type=user_data.get("blood_type","O"), diet_type= user_data.get("diet_type", "standard"), meal_calorie_distribution={"breakfast": b_cal, "lunch": l_cal, "dinner": d_cal}, ) if daily_nutrition: mp.total_daily_calories = daily_nutrition.total_daily_calories mp.total_daily_macronutrients = daily_nutrition.total_daily_macronutrients return mp # ── converters ──────────────────────────────────────────────────────── def _convert_subcats(self, data: Dict[str, List[Dict]]) -> Dict[str, List[Meal]]: return {sub: self._convert_meal_list(meals) for sub, meals in data.items()} def _convert_meal_list(self, meal_list: List[Dict]) -> List[Meal]: meals = [] for md in meal_list: try: ingredients = [ Ingredient(**ing) if isinstance(ing, dict) else Ingredient(name=str(ing), quantity="as needed") for ing in md.get("ingredients", []) ] mr = md.get("macronutrients", {}) if not isinstance(mr, dict): mr = {"carbohydrates": 40, "protein": 25, "fat": 15} meals.append(Meal( meal_name= md.get("meal_name", "Unnamed Meal"), category= md.get("category", "Meal"), sub_category= md.get("sub_category","General"), total_calories=md.get("total_calories", 500), macronutrients=Macronutrients( carbohydrates=mr.get("carbohydrates", 40), protein= mr.get("protein", 25), fat= mr.get("fat", 15), ), description= md.get("description", ""), ingredients= ingredients, number_of_servings=md.get("number_of_servings", 1), )) except Exception as e: print(f"Meal conversion error: {e}") meals.append(Meal( meal_name="Default Meal", category=md.get("category","Meal") if md else "Meal", sub_category=md.get("sub_category","General") if md else "General", total_calories=500, macronutrients=Macronutrients(carbohydrates=40, protein=25, fat=15), description="A balanced meal", ingredients=[Ingredient(name="Mixed ingredients", quantity="1 serving", icon="🍽️")], )) return meals # ───────────────────────────────────────────────────────────────────────────── # FoodScanner # ───────────────────────────────────────────────────────────────────────────── class FoodScanner: def __init__(self, vector_store: Optional["ScannerVectorStore"] = None): self.llm = ChatOpenAI( model="gpt-4o", temperature=0.0, api_key=os.getenv("OPENAI_API_KEY"), ) self.vector_store = vector_store # ---- Prompt 1: Ingredient extraction ---- self.food_scan_prompt = ChatPromptTemplate.from_template(""" You are an expert food image analysis system. Your task is ONLY to identify visible food ingredients from the image. IMPORTANT RULES: 1. Identify ONLY ingredients that are visually present. 2. Do NOT use dietary logic, blood type logic, allergies, or health assumptions. 3. Do NOT guess hidden ingredients unless visually obvious. 4. Keep ingredient names simple and consistent. 5. Include sauces, oils, spices, garnishes, toppings, vegetables, meats, grains, and beverages if visible. 6. The same image should always produce the same ingredient list. 7. Do NOT classify ingredients as healthy, harmful, beneficial, or avoid. 8. Do NOT explain anything. COUNTRY CONTEXT: {country_guidance} Return ONLY valid JSON: {{ "identified_ingredients": [ "ingredient1", "ingredient2", "ingredient3" ] }} If no ingredients can be identified: {{ "identified_ingredients": [] }} """) # ---- Prompt 2: Classification ---- self.classify_prompt = ChatPromptTemplate.from_template(""" You are a precise food compatibility classifier. **USER PROFILE** - Blood Type: {blood_type} - Diet: {diet_type} - Allergies: {allergies} - Country: {country} **BLOOD GROUP RULES** (ground truth – follow these when classifying) {blood_group_rules} **REFERENCE LISTS** (same data broken out for matching) Beneficial foods for {blood_type}: {beneficial_list} Neutral foods: {neutral_list} Avoid foods: {avoid_list} **TASK** Given the list of identified ingredients below, classify each ingredient into one of three categories: - "safe" (beneficial or allowed) - "neutral" (allowed but not particularly beneficial) - "avoid" (harmful based on blood type, diet, or allergy) **RULES FOR CLASSIFICATION AND REASON** (MUST FOLLOW EXACTLY): 1. If an ingredient matches an allergy → status = "avoid", reason = "Not Suitable Due to Your Allergies: {{allergen}}" (replace {{allergen}} with the actual allergen, e.g., "Not Suitable Due to Your Allergies: peanuts") 2. If diet is "vegan" and ingredient contains any animal product (meat, dairy, egg, honey, gelatin, etc.) → status = "avoid", reason = "Not Suitable For Your Vegan Diet" 3. If diet is "vegetarian" and ingredient contains meat or seafood → status = "avoid", reason = "Not Suitable For Your Vegetarian Diet" 4. Otherwise, use BLOOD GROUP RULES / reference lists: - If ingredient is in "avoid" list → status = "avoid", reason = "Not Recommended For your Blood Type {blood_type} Diet" - If ingredient is in "beneficial" list → status = "safe", reason = "" (empty string) - If ingredient is in "neutral" list → status = "neutral", reason = "" (empty string) - If ingredient not in any list, use general nutritional knowledge for the given blood type and diet, and produce a reason accordingly (but prefer the exact strings above when possible). **IMPORTANT**: - For any "avoid", the reason MUST be exactly one of the three formats above. - For "safe" or "neutral", reason MUST be an empty string "". - Accept synonyms, plurals, and common variations (e.g., "tomatoes" = "tomato"). **IDENTIFIED INGREDIENTS** (from the image): {ingredients_json} **OUTPUT FORMAT** (valid JSON only) {{ "classifications": [ {{ "ingredient": "exact ingredient name as given", "status": "safe|neutral|avoid", "reason": "exact reason string as specified above" }} ] }} """) # ---- Prompt 3: AI-only description for the selected food type ---- self.type_description_prompt = ChatPromptTemplate.from_template(""" You are a BloodFIT nutrition coach. Generate a fresh description for this food scan. USER PROFILE - Blood Type: {blood_type} - Diet: {diet_type} - Allergies: {allergies} OVERALL RESULT TYPE: {type_title} INGREDIENTS FOUND IN THE IMAGE: - Beneficial: {beneficial_ingredients} - Neutral: {neutral_ingredients} - Avoid: {avoid_ingredients} BLOOD GROUP RULES: {blood_group_rules} RULES 1. Write ONE short description (1–2 sentences) for type "{type_title}". 2. The description MUST be generated from the ingredients and profile above. 3. Be specific — mention relevant ingredients when possible. 4. Do NOT use generic template phrases. Do NOT copy fixed marketing copy. 5. Tone: clear, supportive, BloodFIT product style. No markdown or bullets. Return ONLY valid JSON: {{ "description": "your AI-generated 1-2 sentence description" }} """) @staticmethod def get_country_guidance(country: str) -> str: g = { "china": "Traditional Chinese cuisine: steamed dishes, vegetables, rice, tofu, seafood.", "south_korea": "Korean cuisine: vegetables, rice, fermented foods, seafood.", } if not country: return "General international cuisine. Focus on whole foods appropriate for blood type and diet." k = country.strip().lower() if k in ("china","chinese"): return g["china"] if k in ("south korea","south korean","korea","korean"): return g["south_korea"] return f"General guidance for {country} cuisine." async def scan_food_image(self, image_data: bytes, user_data: Dict) -> FoodScanResponse: try: blood_type = _normalize_blood_type(user_data.get("blood_type", "O")) diet_type = _normalize_diet_type(user_data.get("diet_type", "standard")) country = user_data.get("country", "") allergies = user_data.get("allergies", "") # Optional reference data from vector store beneficial, neutral, avoid = [], [], [] if self.vector_store: beneficial = await self.vector_store.get_beneficial_foods(blood_type) neutral = await self.vector_store.get_neutral_foods(blood_type) avoid = await self.vector_store.get_avoid_foods(blood_type) blood_group_rules = format_blood_group_rules( blood_type=blood_type, beneficial=beneficial, neutral=neutral, avoid=avoid, client_rules=user_data.get("blood_group_rules"), ) def fmt(lst, max_items=50): return ", ".join(lst[:max_items]) if lst else "None provided" # ---------- STEP 1: Extract ingredients (vision) ---------- image_b64 = base64.b64encode(image_data).decode("utf-8") vision_messages = [{ "role": "user", "content": [ { "type": "text", "text": self.food_scan_prompt.format( country_guidance=self.get_country_guidance(country), ), }, { "type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}, }, ], }] response = self.llm.invoke(vision_messages) content = response.content.strip() # Parse extracted ingredients scan_data = None try: scan_data = json.loads(content) except json.JSONDecodeError: m = re.search(r'(\{.*\})', content, re.DOTALL) if m: try: scan_data = json.loads(m.group(1)) except json.JSONDecodeError: pass if not scan_data: return FoodScanResponse( identified_ingredients=[], harmful_ingredients=[], safe_ingredients=[], neutral_ingredients=[], warning_message="Unable to analyse the image. Please try a clearer photo.", safe_message="", neutral_message="", food_types=[], ) identified = scan_data.get("identified_ingredients", []) if not identified: return FoodScanResponse( identified_ingredients=[], harmful_ingredients=[], safe_ingredients=[], neutral_ingredients=[], warning_message="No ingredients detected. Please ensure the image is clear.", safe_message="", neutral_message="", food_types=[], ) # ---------- STEP 2: Classify using LLM with exact reason format ---------- classify_text = self.classify_prompt.format( blood_type=blood_type, diet_type=diet_type.capitalize(), allergies=allergies or "None", country=country or "Unknown", blood_group_rules=blood_group_rules, beneficial_list=fmt(beneficial), neutral_list=fmt(neutral), avoid_list=fmt(avoid), ingredients_json=json.dumps(identified), ) classify_messages = [{"role": "user", "content": classify_text}] classify_response = self.llm.invoke(classify_messages) classify_content = classify_response.content.strip() # Parse classification JSON try: classify_result = json.loads(classify_content) except json.JSONDecodeError: m = re.search(r'(\{.*\})', classify_content, re.DOTALL) if m: classify_result = json.loads(m.group(1)) else: raise ValueError("No valid JSON in classification response") classifications = classify_result.get("classifications", []) # Build response objects harmful_ingredients: List[HarmfulIngredient] = [] safe_ingredients: List[SafeIngredient] = [] neutral_ingredients: List[NeutralIngredient] = [] for item in classifications: ing_name = item["ingredient"] status = item["status"].lower() reason = item.get("reason", "") if status == "avoid": # Determine category from reason string if "Allergies" in reason: category = "allergy" elif "Diet" in reason: category = "diet" else: category = "blood_type" harmful_ingredients.append(HarmfulIngredient( name=ing_name, reason=reason, category=category, )) elif status == "safe": safe_ingredients.append(SafeIngredient(name=ing_name)) else: # neutral neutral_ingredients.append(NeutralIngredient(name=ing_name)) # Build messages based on classifications warning_msg = self._warning_msg(harmful_ingredients, blood_type, diet_type) safe_msg = self._safe_msg(safe_ingredients) neutral_msg = self._neutral_msg(neutral_ingredients) # Only ONE best-matching type; description is AI-generated present_types = food_types_for_scan( has_beneficial=bool(safe_ingredients), has_neutral=bool(neutral_ingredients), has_avoid=bool(harmful_ingredients), ) food_types: List[FoodTypeInfo] = [] if present_types: selected = present_types[0] ai_description = self._generate_type_description( type_title=selected.title, blood_type=blood_type, diet_type=diet_type, allergies=allergies or "None", blood_group_rules=blood_group_rules, beneficial_ingredients=[i.name for i in safe_ingredients], neutral_ingredients=[i.name for i in neutral_ingredients], avoid_ingredients=[i.name for i in harmful_ingredients], ) if not ai_description: raise HTTPException( 500, "Failed to generate AI description for food type.", ) food_types = [ FoodTypeInfo( key=selected.key, title=selected.title, description=ai_description, ) ] return FoodScanResponse( identified_ingredients=identified, harmful_ingredients=harmful_ingredients, safe_ingredients=safe_ingredients, neutral_ingredients=neutral_ingredients, warning_message=warning_msg, safe_message=safe_msg, neutral_message=neutral_msg, food_types=food_types, ) except HTTPException: raise except Exception as e: traceback.print_exc() raise HTTPException(500, f"Food scanning failed: {e}") def _generate_type_description( self, *, type_title: str, blood_type: str, diet_type: str, allergies: str, blood_group_rules: str, beneficial_ingredients: List[str], neutral_ingredients: List[str], avoid_ingredients: List[str], ) -> str: """Generate description via AI only — no hardcoded fallback text.""" prompt_text = self.type_description_prompt.format( blood_type=blood_type, diet_type=diet_type.capitalize(), allergies=allergies, type_title=type_title, beneficial_ingredients=", ".join(beneficial_ingredients) or "None", neutral_ingredients=", ".join(neutral_ingredients) or "None", avoid_ingredients=", ".join(avoid_ingredients) or "None", blood_group_rules=blood_group_rules, ) last_error = None for _ in range(2): # one retry if first parse fails try: response = self.llm.invoke([{"role": "user", "content": prompt_text}]) content = (response.content or "").strip() if not content: continue data = None try: data = json.loads(content) except json.JSONDecodeError: m = re.search(r'(\{.*\})', content, re.DOTALL) if m: try: data = json.loads(m.group(1)) except json.JSONDecodeError: data = None if isinstance(data, dict): description = (data.get("description") or "").strip() if description: return description # If model returned plain text instead of JSON, use it (still AI-generated) cleaned = re.sub(r'^```(?:json)?\s*|\s*```$', '', content).strip() if cleaned and not cleaned.startswith('{'): return cleaned.strip('"') except Exception as e: last_error = e print(f"[FoodScanner] type description AI call failed: {e}") if last_error: print(f"[FoodScanner] AI description unavailable after retries: {last_error}") return "" # -------------------- Message Builders -------------------- @staticmethod def _warning_msg(harmful, blood_type, diet_type): if not harmful: return "No Avoid ingredients found for your profile!" names = [i.name for i in harmful] if len(names) == 1: return f"Avoid {names[0]}! {harmful[0].reason}" return ( f"Avoid {', '.join(names)}! These ingredients are not suitable " f"for your Blood Type {blood_type} {diet_type.capitalize()} Diet." ) @staticmethod def _safe_msg(safe): if not safe: return "No beneficial ingredients identified." names = [i.name for i in safe] return f"Beneficial to eat: {', '.join(names)}" if len(names) <= 3 else "All other ingredients are beneficial for your profile." @staticmethod def _neutral_msg(neutral): if not neutral: return "" names = [i.name for i in neutral] if len(names) == 1: return f"{names[0]} is a neutral food – allowed but not particularly beneficial or avoid." return f"{', '.join(names)} are neutral foods – allowed but not particularly beneficial or avoid."