"""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 <recipe_id> 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()