Public Access
R1 stabilization: pytest harness with transactional db fixture, smoke + alembic + auth + scrape + approval + swiftly tests, github actions ci yaml. Bearer-token admin auth + signed-cookie session for family ui mutations. Async POST /api/admin/scrape (BackgroundTasks, returns 202). Path canonicalization (no /list, /planned suffixes). DATABASE_URL fail-fast on empty. R2 deferred-risk spikes: live lucky california fetch (R2-A), full email+per-voter approval click round trip with single-use enforcement (R2-B, console email backend, sendgrid stub). R3-0 phase 3 redesign: replaced playwright html scraper with requests based swiftly json api client. 17 categories, ~10k products per scrape, upsert by (source, external_id). 401 surfaces actionable token-refresh message via ScrapeLog.error_message. Pre-existing defects fixed: shopping_list.py syntax error blocking app import, MealPlan.votes orphan relationship, JSONB(astext=True) invalid kwarg, missing requests dep, calorie_target schema drift, every SQLEnum needed values_callable, 0001 had empty downgrade(), seed had duplicate ingredient rows. Migrations added: 0003 grocery_item.description, 0004 family_profile. calorie_target, 0005 grocery_item.external_id + source + composite index. Verified: 31/31 pytest green, alembic upgrade->downgrade->upgrade clean, frontend npm run build clean, live scrape 9,960 grocery_item rows in 36s. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
175 lines
6.3 KiB
Python
175 lines
6.3 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
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
|
|
from collections import defaultdict
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("", response_model=ShoppingListResponse)
|
|
def get_shopping_list(db: Session = Depends(get_db)):
|
|
profile = db.query(FamilyProfile).first()
|
|
if not profile:
|
|
raise HTTPException(status_code=404, detail="Family profile not found")
|
|
|
|
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:
|
|
if ing.get('ingredient_id'):
|
|
all_ingredient_ids.add(ing['ingredient_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:
|
|
name = ing.get('name', 'Unknown')
|
|
quantity = ing.get('quantity', 1.0) or 1.0
|
|
unit = ing.get('unit')
|
|
ing_id = ing.get('ingredient_id')
|
|
|
|
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 = ing_obj.typical_price if ing_obj 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"""
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<title>Shopping List - Week of {shopping_list.week_start_date}</title>
|
|
<style>
|
|
body {{ font-family: Arial, sans-serif; margin: 40px; }}
|
|
h1 {{ border-bottom: 2px solid #333; padding-bottom: 10px; }}
|
|
.aisle {{ margin: 20px 0; }}
|
|
.aisle h2 {{ background: #f5f5f5; padding: 10px; margin: 0; }}
|
|
.item {{ padding: 8px 0; border-bottom: 1px solid #eee; display: flex; justify-content: space-between; }}
|
|
.item .name {{ flex: 1; }}
|
|
.sale {{ color: red; font-weight: bold; }}
|
|
.total {{ margin-top: 30px; font-size: 1.2em; font-weight: bold; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>Shopping List - Week of {shopping_list.week_start_date}</h1>
|
|
<div class="total">Estimated Total: ${shopping_list.total_estimated_cost:.2f}</div>
|
|
<div class="total">Sale Items: {shopping_list.sale_items_count}</div>
|
|
"""
|
|
|
|
for aisle, items in shopping_list.by_aisle.items():
|
|
html += f'<div class="aisle"><h2>{aisle}</h2>'
|
|
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'<div class="item {sale_class}"><span class="name">{item.name}</span><span>{item.quantity} {item.unit or ""} {price}</span></div>'
|
|
html += '</div>'
|
|
|
|
html += '</body></html>'
|
|
|
|
return {"html": html} |