from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy import func
from app.database import get_db
from app.models import (
MealPlan, MealPlanItem, Recipe, HomePantry,
FamilyProfile, Ingredient, GroceryItem
)
from app.schemas import ShoppingListResponse, ShoppingListItem
from datetime import date
from typing import List, Optional
from collections import defaultdict
from uuid import UUID
router = APIRouter()
def _parse_uuid(value):
if not value:
return None
try:
return UUID(value) if isinstance(value, str) else value
except ValueError:
return None
@router.get("", response_model=ShoppingListResponse)
def get_shopping_list(
week_start: Optional[date] = Query(
None,
description="ISO date of the week's Monday (YYYY-MM-DD). Omit for the latest approved/locked plan.",
),
db: Session = Depends(get_db),
):
profile = db.query(FamilyProfile).first()
if not profile:
raise HTTPException(status_code=404, detail="Family profile not found")
if week_start is not None:
current_plan = db.query(MealPlan).filter(
MealPlan.family_profile_id == profile.id,
MealPlan.week_start_date == week_start,
).first()
else:
current_plan = db.query(MealPlan).filter(
MealPlan.family_profile_id == profile.id,
MealPlan.status.in_(['approved', 'locked'])
).order_by(MealPlan.week_start_date.desc()).first()
if not current_plan:
current_plan = db.query(MealPlan).filter(
MealPlan.family_profile_id == profile.id
).order_by(MealPlan.week_start_date.desc()).first()
if not current_plan:
return ShoppingListResponse(
week_start_date=date.today(),
items=[],
total_estimated_cost=0.0,
sale_items_count=0,
by_aisle={}
)
pantry_items = {
p.ingredient_id: p for p in db.query(HomePantry).filter(
HomePantry.family_profile_id == profile.id
).all()
}
ingredient_map = {}
all_ingredient_ids = set()
for item in current_plan.items:
recipe = db.query(Recipe).filter(Recipe.id == item.recipe_id).first()
if recipe and recipe.ingredients:
for ing in recipe.ingredients:
ing_id = _parse_uuid(ing.get('ingredient_id'))
if ing_id:
all_ingredient_ids.add(ing_id)
if all_ingredient_ids:
ingredients = db.query(Ingredient).filter(
Ingredient.id.in_(all_ingredient_ids)
).all()
ingredient_map = {i.id: i for i in ingredients}
grocery_items = {}
if all_ingredient_ids:
groceries = db.query(GroceryItem).filter(
GroceryItem.ingredient_id.in_(all_ingredient_ids),
GroceryItem.is_on_sale == True
).all()
for g in groceries:
grocery_items[g.ingredient_id] = g
aggregated = defaultdict(lambda: {"quantity": 0.0, "unit": None, "name": ""})
for plan_item in current_plan.items:
recipe = db.query(Recipe).filter(Recipe.id == plan_item.recipe_id).first()
if recipe and recipe.ingredients:
for ing in recipe.ingredients:
ing_id = _parse_uuid(ing.get('ingredient_id'))
name = ing.get('name')
if not name and ing_id and ing_id in ingredient_map:
name = ingredient_map[ing_id].name
if not name:
name = 'Unknown'
quantity = ing.get('quantity', 1.0) or 1.0
unit = ing.get('unit')
key = name.lower()
aggregated[key]["quantity"] += quantity
aggregated[key]["unit"] = unit
aggregated[key]["name"] = name
aggregated[key]["ingredient_id"] = ing_id
shopping_items = []
total_cost = 0.0
sale_count = 0
for name_key, data in aggregated.items():
ingredient_id = data.get("ingredient_id")
in_pantry = ingredient_id and ingredient_id in pantry_items
ing_obj = ingredient_map.get(ingredient_id) if ingredient_id else None
price = float(ing_obj.typical_price) if ing_obj and ing_obj.typical_price is not None else None
sale_price = None
is_on_sale = False
if ingredient_id and ingredient_id in grocery_items:
g = grocery_items[ingredient_id]
is_on_sale = True
sale_price = float(g.current_price) if g.current_price else None
price = sale_price
sale_count += 1
if price:
total_cost += price * data["quantity"]
shopping_items.append(ShoppingListItem(
ingredient_id=ingredient_id,
name=data["name"],
quantity=data["quantity"],
unit=data["unit"],
aisle=ing_obj.aisle if ing_obj else None,
estimated_price=price,
is_on_sale=is_on_sale,
sale_price=sale_price,
in_season=grocery_items.get(ingredient_id).in_season if ingredient_id and ingredient_id in grocery_items else False,
in_pantry=in_pantry
))
by_aisle = defaultdict(list)
for item in shopping_items:
aisle = item.aisle or "Other"
by_aisle[aisle].append(item)
return ShoppingListResponse(
week_start_date=current_plan.week_start_date,
items=shopping_items,
total_estimated_cost=round(total_cost, 2),
sale_items_count=sale_count,
by_aisle=dict(by_aisle)
)
@router.get("/print")
def print_shopping_list(db: Session = Depends(get_db)):
shopping_list = get_shopping_list.__wrapped__(None, db)
html = f"""
Shopping List - Week of {shopping_list.week_start_date}
Shopping List - Week of {shopping_list.week_start_date}
Estimated Total: ${shopping_list.total_estimated_cost:.2f}
Sale Items: {shopping_list.sale_items_count}
"""
for aisle, items in shopping_list.by_aisle.items():
html += f'{aisle}
'
for item in items:
sale_class = 'sale' if item.is_on_sale else ''
price = f'${item.sale_price:.2f}' if item.sale_price else (f'${item.estimated_price:.2f}' if item.estimated_price else '')
html += f'
{item.name}{item.quantity} {item.unit or ""} {price}
'
html += '