import ollama import json import asyncio from config import settings def _get_client(): return ollama.Client(host=settings.ollama_host) def _chat_sync(messages: list, json_format: bool = True) -> str: client = _get_client() kwargs = {"temperature": 0.7, "num_predict": 4096 if json_format else 1024} response = client.chat( model=settings.model_name, messages=messages, **({"format": "json"} if json_format else {}), options=kwargs, ) return response["message"]["content"] async def _chat(messages: list) -> dict: loop = asyncio.get_running_loop() raw = await loop.run_in_executor(None, lambda: _chat_sync(messages)) try: return json.loads(raw) except json.JSONDecodeError as e: raise ValueError(f"Ollama returned invalid JSON: {e}. Raw: {raw[:200]}") async def generate_weekly_menu(pantry_context: dict, user_notes: str | None = None) -> dict: """Suggest 8-12 diverse recipes for the week, prioritizing expiring pantry items.""" user_message = f"""Given my current pantry and recent meal history, suggest 8-12 diverse recipes I could make this week. PANTRY STATE: {json.dumps(pantry_context, indent=2)} GUIDELINES: - Suggest genuinely good, practical dishes — do NOT limit yourself to only pantry ingredients - Use pantry ingredients where it makes sense, but freely include recipes that require buying additional items - Avoid meals eaten in the last 7 days (see recent_meals) - Include a mix of breakfast, lunch, and dinner options - Keep each recipe under 60 minutes - Include full ingredient lists with quantities and units Respond ONLY with this JSON structure: {{ "recipes": [ {{ "name": "...", "meal_type": "breakfast", "ingredients": [{{"name": "...", "quantity": 1.0, "unit": "cup"}}], "description": "1-2 sentence description of the dish's flavor and appeal", "instructions": "1. ... 2. ... (4-8 numbered steps, 1-2 sentences each)", "time_minutes": 20, "serves": 2 }} ], "notes": "brief explanation of choices" }}""" if user_notes: user_message += f"\n\nUSER PREFERENCES:\n{user_notes}" messages = [ {"role": "system", "content": settings.system_prompt}, {"role": "user", "content": user_message}, ] return await _chat(messages) async def generate_grocery_list(recipes: list, pantry_context: dict, user_notes: str | None = None) -> dict: """Generate a minimal grocery list based on a list of recipes and current pantry.""" user_message = f"""Given these recipes and current pantry, generate a minimal grocery list. RECIPES: {json.dumps(recipes, indent=2)} CURRENT PANTRY: {json.dumps(pantry_context["available_ingredients"], indent=2)} RULES: - Only list ingredients NOT already sufficiently in the pantry - Consolidate duplicate ingredients across recipes (buy once) - Group items by store section (produce, dairy, protein, pantry, frozen, bakery) - Estimate realistic retail costs in USD - Prefer bulk when ingredient appears in 3+ recipes Respond ONLY with this JSON: {{ "items": [ {{ "name": "...", "quantity": 2.0, "unit": "lbs", "store_section": "produce", "estimated_cost": 3.50, "used_in_meals": ["Recipe A", "Recipe B"], "reason": "not in pantry" }} ], "total_estimate": 45.00, "shopping_notes": "..." }}""" if user_notes: user_message += f"\n\nUSER PREFERENCES:\n{user_notes}" messages = [ {"role": "system", "content": settings.system_prompt}, {"role": "user", "content": user_message}, ] return await _chat(messages) async def generate_grocery_list_from_pantry(pantry_context: dict, user_notes: str | None = None) -> dict: """Generate a grocery list based on pantry state alone, without a menu plan.""" user_message = f"""Given my current pantry, generate a practical grocery list to stock up for the week. CURRENT PANTRY: {json.dumps(pantry_context["available_ingredients"], indent=2)} RULES: - Suggest ingredients that would complement what's already in the pantry - Group items by store section (produce, dairy, protein, pantry, frozen, bakery) - Estimate realistic retail costs in USD - Focus on versatile staples that enable multiple meals Respond ONLY with this JSON: {{ "items": [ {{ "name": "...", "quantity": 2.0, "unit": "lbs", "store_section": "produce", "estimated_cost": 3.50, "used_in_meals": [], "reason": "versatile staple" }} ], "total_estimate": 45.00, "shopping_notes": "..." }}""" if user_notes: user_message += f"\n\nUSER PREFERENCES:\n{user_notes}" messages = [ {"role": "system", "content": settings.system_prompt}, {"role": "user", "content": user_message}, ] return await _chat(messages) async def suggest_recipe(pantry_context: dict) -> dict: """Suggest one recipe using available pantry ingredients.""" user_message = f"""Suggest ONE recipe I can make right now using only these pantry ingredients (plus common staples). PANTRY: {json.dumps(pantry_context["available_ingredients"], indent=2)} Respond ONLY with this JSON: {{ "recipe": {{ "name": "...", "meal_type": "dinner", "ingredients": [{{"name": "...", "quantity": 1.0, "unit": "cup"}}], "instructions": "Step 1... Step 2...", "time_minutes": 30, "serves": 2 }} }}""" messages = [ {"role": "system", "content": settings.system_prompt}, {"role": "user", "content": user_message}, ] return await _chat(messages) async def generate_replacement_recipe(meal_type: str, existing_names: list, pantry_context: dict, user_notes: str | None = None) -> dict: """Generate a single replacement recipe of the given meal type.""" user_message = f"""Suggest ONE {meal_type} recipe to replace a dish I don't want. CURRENT PANTRY: {json.dumps(pantry_context["available_ingredients"], indent=2)} ALREADY IN PLAN (do not suggest these): {json.dumps(existing_names, indent=2)} GUIDELINES: - Suggest a genuinely good, practical {meal_type} dish - Do NOT limit yourself to only pantry ingredients - Under 60 minutes - Must not be any of the dishes already in the plan Respond ONLY with this JSON: {{ "recipe": {{ "name": "...", "meal_type": "{meal_type}", "ingredients": [{{"name": "...", "quantity": 1.0, "unit": "cup"}}], "description": "1-2 sentence description of the dish's flavor and appeal", "instructions": "1. ... 2. ... (4-8 numbered steps, 1-2 sentences each)", "time_minutes": 30, "serves": 2 }} }}""" if user_notes: user_message += f"\n\nUSER PREFERENCES:\n{user_notes}" messages = [ {"role": "system", "content": settings.system_prompt}, {"role": "user", "content": user_message}, ] return await _chat(messages) async def get_available_models() -> list: """Get list of available Ollama model names.""" def _list_models_sync(): client = _get_client() try: models_resp = client.list() return [m["name"] for m in models_resp.get("models", [])] except Exception: return [] loop = asyncio.get_running_loop() return await loop.run_in_executor(None, _list_models_sync) async def chat_with_commis(message: str, history: list, pantry_context: dict, menu_context: list, grocery_context: list) -> str: """Have a conversation with Commis using full kitchen context.""" system_prompt = f"""You are Commis, a friendly and knowledgeable personal chef assistant. You have full visibility into the user's kitchen — their pantry, this week's recipe suggestions, and their grocery list. Use this context to give personalized, practical cooking advice. PANTRY (ingredients on hand): {json.dumps(pantry_context.get("available_ingredients", []), indent=2)} EXPIRING SOON: {json.dumps(pantry_context.get("expiring_soon", []), indent=2)} RECENT MEALS (last 14 days): {json.dumps(pantry_context.get("recent_meals", []), indent=2)} THIS WEEK'S RECIPE SUGGESTIONS: {json.dumps(menu_context, indent=2)} GROCERY LIST: {json.dumps(grocery_context, indent=2)} Be conversational, helpful, and concise. Reference specific ingredients and recipes by name when relevant. You do not need to respond with JSON.""" messages = [{"role": "system", "content": system_prompt}] messages.extend(history) messages.append({"role": "user", "content": message}) loop = asyncio.get_running_loop() return await loop.run_in_executor(None, lambda: _chat_sync(messages, json_format=False)) async def parse_recipe_from_text(text: str) -> dict: """Ask Ollama to parse unstructured recipe text into our standard recipe schema.""" user_message = f"""Parse this recipe text and extract structured data. RECIPE TEXT: {text[:4000]} Respond ONLY with this JSON: {{ "recipe": {{ "name": "...", "meal_type": "dinner", "ingredients": [{{"name": "...", "quantity": 1.0, "unit": "cup"}}], "description": "1-2 sentence description of the dish's flavor and appeal", "instructions": "1. ... 2. ... (4-8 numbered steps, 1-2 sentences each)", "time_minutes": 30, "serves": 2 }} }} meal_type must be one of: breakfast, lunch, dinner, snack""" messages = [ {"role": "system", "content": settings.system_prompt}, {"role": "user", "content": user_message}, ] return await _chat(messages) async def generate_single_recipe(description: str, existing_names: list, pantry_context: dict) -> dict: """Generate one recipe matching a user description.""" user_message = f"""Suggest ONE recipe based on this description: "{description}" CURRENT PANTRY: {json.dumps(pantry_context["available_ingredients"], indent=2)} ALREADY IN PLAN (do not suggest these): {json.dumps(existing_names, indent=2)} GUIDELINES: - Match the description as closely as possible - Do NOT limit yourself to only pantry ingredients - Under 60 minutes - Must not be any dish already in the plan Respond ONLY with this JSON: {{ "recipe": {{ "name": "...", "description": "1-2 sentence description of the dish's flavor and appeal", "meal_type": "dinner", "ingredients": [{{"name": "...", "quantity": 1.0, "unit": "cup"}}], "instructions": "1. ... 2. ... (4-8 numbered steps, 1-2 sentences each)", "time_minutes": 30, "serves": 2 }} }} meal_type must be one of: breakfast, lunch, dinner, snack""" messages = [ {"role": "system", "content": settings.system_prompt}, {"role": "user", "content": user_message}, ] return await _chat(messages)