"""Persistent shopping list with zone-classified items. Stores a running grocery list that persists across sessions. Items are automatically classified into Costco zones using the same pipeline + ChromaDB memory that powers the route optimizer. The "Add to list" bridge: recipe ingredients → zone-classified items → shopping list. Storage: ~/.costco_route/shopping_list.json """ import json import os import re from datetime import datetime from pathlib import Path from costco_route.config import ZONES, ZONE_ORDER from costco_route.pipeline import optimize from costco_route.item_memory import lookup_items # --------------------------------------------------------------------------- # Storage # --------------------------------------------------------------------------- LIST_PATH = Path(os.environ.get( "SHOPPING_LIST_PATH", os.path.expanduser("~/.costco_route/shopping_list.json"), )) def _load_list() -> dict: """Load the shopping list from disk. Returns: Dict with items, metadata. """ if not LIST_PATH.exists(): return {"items": {}, "created_at": datetime.now().isoformat()} try: with open(LIST_PATH) as f: return json.load(f) except (json.JSONDecodeError, OSError): return {"items": {}, "created_at": datetime.now().isoformat()} def _save_list(data: dict): """Save the shopping list to disk.""" LIST_PATH.parent.mkdir(parents=True, exist_ok=True) data["updated_at"] = datetime.now().isoformat() with open(LIST_PATH, "w") as f: json.dump(data, f, indent=2, ensure_ascii=False) # --------------------------------------------------------------------------- # Core operations # --------------------------------------------------------------------------- def add_items(raw_items: str | list[str], source: str = "manual") -> dict: """Add items to the shopping list. Accepts either a raw string (stream of consciousness) or a list of pre-extracted item strings. Items are classified into Costco zones automatically. Args: raw_items: Items to add (string or list) source: Origin of items ("manual", "recipe:slug", etc.) Returns: Dict with added items, classified zones, and counts. """ data = _load_list() # If given a list, join it for the pipeline if isinstance(raw_items, list): raw_text = "\n".join(raw_items) else: raw_text = raw_items # Run through the full pipeline: extract → classify → memory overrides result = optimize(raw_text, use_memory=True, markdown=False) classified = result.get("classified", {}) # Merge classified items into the shopping list added_count = 0 added_zones = set() for zone_id, zone_items in classified.items(): for item in zone_items: # Key by normalized item name for dedup item_key = _item_key(item) if item_key not in data["items"]: data["items"][item_key] = { "name": item, "zone": zone_id, "checked": False, "added_at": datetime.now().isoformat(), "source": source, } added_count += 1 added_zones.add(zone_id) # If item already exists, keep the existing one (no overwrite) _save_list(data) return { "added": added_count, "zones": sorted(added_zones), "classified": classified, } def add_recipe_ingredients(recipe_id: str) -> dict: """Add all ingredients from a saved recipe to the shopping list. This is the "Add to list" bridge: recipe → ingredients → zones → list. Args: recipe_id: Recipe slug from the Rolodex Returns: Dict with added items, classified zones, and counts. """ from costco_route.recipe_extractor import get_recipe recipe = get_recipe(recipe_id) if not recipe: return {"error": f"Recipe not found: {recipe_id}"} ingredients = recipe.get("ingredients", []) if not ingredients: return {"error": f"Recipe has no ingredients: {recipe_id}"} title = recipe.get("title", "Unknown") source = f"recipe:{recipe_id}" # Add each ingredient individually to preserve quantities result = add_items(ingredients, source=source) # Link back to the recipe result["recipe_id"] = recipe_id result["recipe_title"] = title return result def remove_item(item_name: str) -> dict: """Remove an item from the shopping list by name (fuzzy match). Returns: Dict with removed item or not-found status. """ data = _load_list() target_key = _item_key(item_name) # Try exact key match first if target_key in data["items"]: removed = data["items"].pop(target_key) _save_list(data) return {"removed": removed["name"], "zone": removed["zone"]} # Fuzzy match for key, item_data in list(data["items"].items()): if _item_key(item_name) in key or key in _item_key(item_name): removed = data["items"].pop(key) _save_list(data) return {"removed": removed["name"], "zone": removed["zone"]} return {"error": f"Item not found: {item_name}"} def check_item(item_name: str) -> dict: """Mark an item as checked off the list. Returns: Dict with updated item or not-found status. """ data = _load_list() target_key = _item_key(item_name) # Try exact then fuzzy match_key = None if target_key in data["items"]: match_key = target_key else: for key in data["items"]: if _item_key(item_name) in key or key in _item_key(item_name): match_key = key break if not match_key: return {"error": f"Item not found: {item_name}"} data["items"][match_key]["checked"] = True _save_list(data) return {"checked": data["items"][match_key]["name"], "zone": data["items"][match_key]["zone"]} def uncheck_item(item_name: str) -> dict: """Uncheck an item on the list.""" data = _load_list() target_key = _item_key(item_name) match_key = None if target_key in data["items"]: match_key = target_key else: for key in data["items"]: if _item_key(item_name) in key or key in _item_key(item_name): match_key = key break if not match_key: return {"error": f"Item not found: {item_name}"} data["items"][match_key]["checked"] = False _save_list(data) return {"unchecked": data["items"][match_key]["name"]} def clear_list(keep_unchecked: bool = False) -> dict: """Clear the shopping list. Args: keep_unchecked: If True, only remove checked items. Returns: Dict with count of removed items. """ data = _load_list() if keep_unchecked: original_count = len(data["items"]) data["items"] = { k: v for k, v in data["items"].items() if not v.get("checked", False) } removed = original_count - len(data["items"]) else: removed = len(data["items"]) data["items"] = {} _save_list(data) return {"removed": removed, "remaining": len(data["items"])} def get_list() -> dict: """Get the full shopping list, organized by zone. Returns: Dict with zone-organized items, stats. """ data = _load_list() items = data.get("items", {}) # Organize by zone in traversal order zones = {} for zone_id in ZONE_ORDER: zone_items = [] for key, item_data in items.items(): if item_data.get("zone") == zone_id: zone_items.append(item_data) if zone_items: zones[zone_id] = zone_items # Stats total = len(items) checked = sum(1 for v in items.values() if v.get("checked", False)) unchecked = total - checked return { "zones": zones, "total": total, "checked": checked, "unchecked": unchecked, } def format_list(include_checked: bool = False) -> str: """Format the shopping list for display (Telegram/CLI). Args: include_checked: Show checked-off items with strikethrough Returns: Formatted string. """ data = get_list() if not data["zones"]: return "📋 Shopping list is empty.\n\nSend items or /add to get started." lines = ["📋 **Shopping List**"] total_items = 0 zone_count = len(data["zones"]) for zone_id in ZONE_ORDER: if zone_id not in data["zones"]: continue zone_info = ZONES.get(zone_id, {"name": f"Zone {zone_id}"}) zone_items = data["zones"][zone_id] lines.append(f"\n📦 Zone {zone_id} — {zone_info['name']}") for item in zone_items: name = item["name"] if item.get("checked"): if include_checked: lines.append(f" ✅ ~~{name}~~") else: lines.append(f" ☐ {name}") total_items += 1 # Show source tags for recipe items sources = set() for item in zone_items: src = item.get("source", "") if src.startswith("recipe:"): sources.add(src.split(":", 1)[1]) if sources: lines.append(f" 🍽 from: {', '.join(sources)}") est_minutes = max(15, zone_count * 2 + total_items) checked_str = f" · {data['checked']} checked" if data["checked"] else "" lines.append(f"\n✅ {total_items} items across {zone_count} zones — ~{est_minutes} min{checked_str}") return "\n".join(lines) def _item_key(item_name: str) -> str: """Normalize an item name to a dedup key. Removes quantities, normalizes whitespace, lowercases. "2 cups flour" → "flour" "1 lb chicken thighs" → "chicken thighs" """ # Remove common quantity patterns at the start cleaned = re.sub( r'^(\d+[\s/]?\d*)\s*(cups?|tbsp|tsp|lb|oz|pcs|pieces?|cloves?|cans?|bunches?|heads?|slices?|bags?|boxes?|packages?|pints?|quarts?|gallons?)\s+', '', item_name.lower().strip(), count=1, ) # Remove "of" after quantity removal cleaned = re.sub(r'^of\s+', '', cleaned) # Normalize whitespace cleaned = re.sub(r'\s+', ' ', cleaned).strip() return cleaned if cleaned else item_name.lower().strip()