Fix ruff lint and format violations, update README (#5)

Apply ruff check --fix and ruff format across all Python files to
pass CI pipeline. Remove unused imports (os, sys), fix f-strings
without placeholders. Update README to reflect sync-roster Lambda,
corrected shared layer path, and current project structure.
This commit is contained in:
Adam Moussa 2026-05-12 19:31:27 -04:00 • committed by GitHub
parent 04297618c4
commit 440117c9a1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 364 additions and 170 deletions

View file

@ -29,6 +29,7 @@ who haven't ordered
| When | What | How |
|------|------|-----|
| Monday 6:55am ET | Sync employee roster from Slack channel membership | EventBridge → Lambda → DynamoDB |
| Monday 7am ET | Email previous week's payroll deductions to `payroll@` | EventBridge → Lambda → SES |
| Monday 8am ET | Scrape menu, generate form, upload to S3, post link to Slack | GitHub Actions cron |
| Mon–Thu | Employees visit `orders.seahavenind.com` and submit orders | S3 static form → API Gateway → Lambda → DynamoDB |
@ -43,7 +44,7 @@ Stack name: `meal-order-manager` (us-east-1)
- **CloudFront** — HTTPS distribution with custom domain `orders.seahavenind.com`
- **DynamoDB** — `meal-order-manager-orders` (orders, menu, roster, config)
- **API Gateway** — HttpApi for order submission
- **Lambda** — 5 functions: submit-order, close-form, aggregate-orders, slack-notifier, email-report
- **Lambda** — 6 functions: submit-order, close-form, aggregate-orders, slack-notifier, sync-roster, email-report
- **EventBridge** — scheduled rules (dual EST/EDT) for close, reminders, payroll email
- **Secrets Manager** — Slack bot token, form API key
- **SES** — payroll deduction emails
@ -63,7 +64,7 @@ playwright install chromium
```bash
cp samconfig.toml.example samconfig.toml
# Edit samconfig.toml with your certificate ARN, API key, etc.
# Edit samconfig.toml with your certificate ARN, etc.
sam build
sam deploy
```
@ -75,7 +76,7 @@ sam deploy
3. Update the Slack channel SSM parameter: `aws ssm put-parameter --name /meal-order-manager/slack-channel-id --value "C0XXXXXXX" --overwrite`
4. Verify SES sender identity for `adam@seahavenind.com`
5. Set up DNS: CNAME `orders.seahavenind.com` → CloudFront distribution domain
6. Populate employee roster: edit `config.json` and run `python3 scripts/seed_roster.py`
6. Roster syncs automatically from Slack channel members (runs Monday 6:55am ET), or seed manually: `python3 scripts/seed_roster.py`
## Local Workflow (no AWS)
@ -108,12 +109,13 @@ meal-order-manager/
│ ├── scraper/ # Playwright menu scraper
│ ├── server/ # Form generator + local Flask server
│ ├── aggregator/ # Order aggregation + CSV reports
│ └── shared/python/shared/ # Lambda layer (db, secrets, slack helpers)
│ └── shared/shared/ # Lambda layer (db, secrets, slack helpers)
├── functions/ # Lambda handlers
│ ├── submit_order/
│ ├── close_form/
│ ├── aggregate_orders/
│ ├── slack_notifier/
│ ├── sync_roster/
│ └── email_report/
├── scripts/ # CI/CD helper scripts
│ ├── upload_menu.py
@ -121,5 +123,6 @@ meal-order-manager/
│ └── seed_roster.py
├── template.yaml # SAM template
├── samconfig.toml.example
├── slack-app-manifest.yml # Slack app manifest (paste into api.slack.com)
└── config.json
```

View file

@ -42,8 +42,12 @@ def lambda_handler(event, context):
order_csv_key = f"reports/{week}/order-summary.csv"
payroll_csv_key = f"reports/{week}/payroll-deductions.csv"
_s3.put_object(Bucket=bucket, Key=order_csv_key, Body=order_csv, ContentType="text/csv")
_s3.put_object(Bucket=bucket, Key=payroll_csv_key, Body=payroll_csv, ContentType="text/csv")
_s3.put_object(
Bucket=bucket, Key=order_csv_key, Body=order_csv, ContentType="text/csv"
)
_s3.put_object(
Bucket=bucket, Key=payroll_csv_key, Body=payroll_csv, ContentType="text/csv"
)
summary["order_csv_s3_key"] = order_csv_key
summary["payroll_csv_s3_key"] = payroll_csv_key
@ -52,7 +56,9 @@ def lambda_handler(event, context):
_lambda.invoke(
FunctionName=os.environ["SLACK_NOTIFIER_ARN"],
InvocationType="Event",
Payload=json.dumps({"event": "orders_aggregated", "week": week}, cls=DecimalEncoder),
Payload=json.dumps(
{"event": "orders_aggregated", "week": week}, cls=DecimalEncoder
),
)
return {"status": "aggregated", "week": week, "total_employees": len(orders)}
@ -69,12 +75,14 @@ def build_summary(orders: list[dict], week: str) -> dict:
meals = []
for name, data in sorted(meal_totals.items()):
meals.append({
"meal": name,
"quantity": data["quantity"],
"unit_price": data["unit_price"],
"line_total": round(data["unit_price"] * data["quantity"], 2),
})
meals.append(
{
"meal": name,
"quantity": data["quantity"],
"unit_price": data["unit_price"],
"line_total": round(data["unit_price"] * data["quantity"], 2),
}
)
return {
"week": week,
@ -91,25 +99,38 @@ def build_order_summary_csv(summary: dict) -> str:
writer = csv.writer(buf)
writer.writerow(["Meal", "Quantity", "Unit Price", "Line Total"])
for meal in summary["meals"]:
writer.writerow([meal["meal"], meal["quantity"], f"${meal['unit_price']:.2f}", f"${meal['line_total']:.2f}"])
writer.writerow(
[
meal["meal"],
meal["quantity"],
f"${meal['unit_price']:.2f}",
f"${meal['line_total']:.2f}",
]
)
writer.writerow([])
writer.writerow(["TOTAL", summary["total_meals"], "", f"${summary['grand_total']:.2f}"])
writer.writerow(
["TOTAL", summary["total_meals"], "", f"${summary['grand_total']:.2f}"]
)
return buf.getvalue()
def build_payroll_csv(orders: list[dict]) -> str:
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(["Employee Name", "Employee Email", "Items Ordered", "Total Deduction"])
writer.writerow(
["Employee Name", "Employee Email", "Items Ordered", "Total Deduction"]
)
for order in sorted(orders, key=lambda o: o["employee_name"]):
items_str = "; ".join(
f"{item['name']} x{int(item['quantity'])} (${float(item.get('subtotal', float(item.get('price', 0)) * int(item.get('quantity', 0)))):.2f})"
for item in order.get("items", [])
)
writer.writerow([
order["employee_name"],
order["employee_email"],
items_str,
f"${float(order['total']):.2f}",
])
writer.writerow(
[
order["employee_name"],
order["employee_email"],
items_str,
f"${float(order['total']):.2f}",
]
)
return buf.getvalue()

View file

@ -44,7 +44,9 @@ def lambda_handler(event, context):
msg.attach(body)
attachment = MIMEApplication(csv_content)
attachment.add_header("Content-Disposition", "attachment", filename=f"payroll-deductions-{week}.csv")
attachment.add_header(
"Content-Disposition", "attachment", filename=f"payroll-deductions-{week}.csv"
)
msg.attach(attachment)
_ses.send_raw_email(

View file

@ -31,7 +31,7 @@ def handle_menu_published(event):
week = event.get("week", current_week())
meal_count = event.get("meal_count", "")
text = f"This week's meal order is open! Deadline: Thursday 6pm."
text = "This week's meal order is open! Deadline: Thursday 6pm."
blocks = [
{
"type": "header",
@ -121,17 +121,24 @@ def handle_reminder(event):
send_dm(
slack_id,
f"Meal orders close at 6pm today. Place your order: {form_url}",
blocks=[{
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
f"Hey {emp['name'].split()[0]}! Meal orders close at *6pm today*.\n\n"
f"*<{form_url}|Place your order>*"
),
},
}],
blocks=[
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
f"Hey {emp['name'].split()[0]}! Meal orders close at *6pm today*.\n\n"
f"*<{form_url}|Place your order>*"
),
},
}
],
)
dm_count += 1
return {"status": "reminders_sent", "week": week, "dm_count": dm_count, "missing_count": len(missing)}
return {
"status": "reminders_sent",
"week": week,
"dm_count": dm_count,
"missing_count": len(missing),
}

View file

@ -67,7 +67,12 @@ def handle_submit(event):
filtered_items = [i for i in items if i.get("quantity", 0) > 0]
total = sum((i.get("price", 0) or 0) * i.get("quantity", 0) for i in filtered_items)
slug = "".join(c if c.isalnum() or c in "- " else "" for c in name).strip().replace(" ", "-").lower()
slug = (
"".join(c if c.isalnum() or c in "- " else "" for c in name)
.strip()
.replace(" ", "-")
.lower()
)
order_data = {
"employee_name": name,
@ -79,7 +84,14 @@ def handle_submit(event):
put_order(week, slug, order_data)
return response(200, {"status": "ok", "message": f"Order saved for {name}", "total": order_data["total"]})
return response(
200,
{
"status": "ok",
"message": f"Order saved for {name}",
"total": order_data["total"],
},
)
def response(status_code: int, body: dict) -> dict:

View file

@ -5,7 +5,9 @@ from shared.slack import get_channel_members, get_user_info
def lambda_handler(event, context):
channel_id = os.environ.get("SLACK_CHANNEL_ID") or __import__("shared.secrets", fromlist=["get_parameter"]).get_parameter(os.environ["SLACK_CHANNEL_PARAM"])
channel_id = os.environ.get("SLACK_CHANNEL_ID") or __import__(
"shared.secrets", fromlist=["get_parameter"]
).get_parameter(os.environ["SLACK_CHANNEL_PARAM"])
member_ids = get_channel_members(channel_id)
if not member_ids:
@ -27,11 +29,13 @@ def lambda_handler(event, context):
skipped += 1
continue
employees.append({
"name": name,
"email": email,
"slack_user_id": uid,
})
employees.append(
{
"name": name,
"email": email,
"slack_user_id": uid,
}
)
employees.sort(key=lambda e: e["name"])

View file

@ -10,18 +10,28 @@ import os
import sys
import urllib.request
SLACK_BOT_TOKEN_SECRET = os.environ.get("SLACK_BOT_TOKEN_SECRET", "meal-order-manager/slack-bot-token")
SLACK_CHANNEL_PARAM = os.environ.get("SLACK_CHANNEL_PARAM", "/meal-order-manager/slack-channel-id")
SLACK_BOT_TOKEN_SECRET = os.environ.get(
"SLACK_BOT_TOKEN_SECRET", "meal-order-manager/slack-bot-token"
)
SLACK_CHANNEL_PARAM = os.environ.get(
"SLACK_CHANNEL_PARAM", "/meal-order-manager/slack-channel-id"
)
def get_secret(secret_id):
import boto3
return boto3.client("secretsmanager").get_secret_value(SecretId=secret_id)["SecretString"]
return boto3.client("secretsmanager").get_secret_value(SecretId=secret_id)[
"SecretString"
]
def get_parameter(name):
import boto3
return boto3.client("ssm").get_parameter(Name=name, WithDecryption=True)["Parameter"]["Value"]
return boto3.client("ssm").get_parameter(Name=name, WithDecryption=True)[
"Parameter"
]["Value"]
def main():
@ -33,7 +43,7 @@ def main():
token = get_secret(SLACK_BOT_TOKEN_SECRET)
channel = get_parameter(SLACK_CHANNEL_PARAM)
text = f"This week's meal order is open! Deadline: Thursday 6pm."
text = "This week's meal order is open! Deadline: Thursday 6pm."
blocks = [
{
"type": "header",
@ -44,8 +54,7 @@ def main():
"text": {
"type": "mrkdwn",
"text": (
f"*<{form_url}|Place your order>*\n\n"
f"*Deadline:* Thursday 6pm\n"
f"*<{form_url}|Place your order>*\n\n*Deadline:* Thursday 6pm\n"
),
},
},

View file

@ -37,12 +37,15 @@ def main():
table = boto3.resource("dynamodb").Table(TABLE_NAME)
from datetime import datetime
table.put_item(Item={
"PK": "CONFIG",
"SK": "ROSTER",
"employees": roster,
"updated_at": datetime.now().isoformat(),
})
table.put_item(
Item={
"PK": "CONFIG",
"SK": "ROSTER",
"employees": roster,
"updated_at": datetime.now().isoformat(),
}
)
print(f"Seeded {len(roster)} employees to DynamoDB")
for emp in roster:

View file

@ -40,6 +40,7 @@ def main():
table = boto3.resource("dynamodb").Table(TABLE_NAME)
import time
item = {
"PK": f"WEEK#{week}",
"SK": "MENU",

View file

@ -50,12 +50,14 @@ def generate_order_summary(orders: list[dict]) -> dict:
summary = []
for name, data in sorted(meal_totals.items()):
summary.append({
"meal": name,
"quantity": data["quantity"],
"unit_price": data["unit_price"],
"line_total": round(data["unit_price"] * data["quantity"], 2),
})
summary.append(
{
"meal": name,
"quantity": data["quantity"],
"unit_price": data["unit_price"],
"line_total": round(data["unit_price"] * data["quantity"], 2),
}
)
grand_total = sum(s["line_total"] for s in summary)
total_meals = sum(s["quantity"] for s in summary)
@ -74,18 +76,22 @@ def generate_payroll_csv(orders: list[dict], output_path: Path):
"""Write a CSV for payroll with one row per employee."""
with open(output_path, "w", newline="") as f:
writer = csv.writer(f)
writer.writerow(["Employee Name", "Employee Email", "Items Ordered", "Total Deduction"])
writer.writerow(
["Employee Name", "Employee Email", "Items Ordered", "Total Deduction"]
)
for order in sorted(orders, key=lambda o: o["employee_name"]):
items_str = "; ".join(
f"{item['name']} x{item['quantity']} (${item['subtotal']:.2f})"
for item in order.get("items", [])
)
writer.writerow([
order["employee_name"],
order["employee_email"],
items_str,
f"${order['total']:.2f}",
])
writer.writerow(
[
order["employee_name"],
order["employee_email"],
items_str,
f"${order['total']:.2f}",
]
)
def generate_order_summary_csv(summary: dict, output_path: Path):
@ -94,14 +100,18 @@ def generate_order_summary_csv(summary: dict, output_path: Path):
writer = csv.writer(f)
writer.writerow(["Meal", "Quantity", "Unit Price", "Line Total"])
for meal in summary["meals"]:
writer.writerow([
meal["meal"],
meal["quantity"],
f"${meal['unit_price']:.2f}",
f"${meal['line_total']:.2f}",
])
writer.writerow(
[
meal["meal"],
meal["quantity"],
f"${meal['unit_price']:.2f}",
f"${meal['line_total']:.2f}",
]
)
writer.writerow([])
writer.writerow(["TOTAL", summary["total_meals"], "", f"${summary['grand_total']:.2f}"])
writer.writerow(
["TOTAL", summary["total_meals"], "", f"${summary['grand_total']:.2f}"]
)
def main():
@ -131,9 +141,9 @@ def main():
generate_payroll_csv(orders, payroll_csv)
# Print summary
print(f"\n{'='*60}")
print(f"\n{'=' * 60}")
print(f"ORDER SUMMARY — Week {week}")
print(f"{'='*60}")
print(f"{'=' * 60}")
print(f"{'Meal':<45} {'Qty':>4} {'Total':>8}")
print("-" * 60)
for meal in summary["meals"]:
@ -142,7 +152,7 @@ def main():
print(f"{'TOTAL':<45} {summary['total_meals']:>4} ${summary['grand_total']:>7.2f}")
print(f"\n{summary['total_employees']} employees ordered")
print(f"\nFiles generated:")
print("\nFiles generated:")
print(f" Order summary: {summary_csv}")
print(f" Payroll report: {payroll_csv}")
print(f" Raw JSON: {summary_json}")

View file

@ -5,7 +5,6 @@ Run this once to understand the site before building the production scraper.
"""
import json
import sys
from playwright.sync_api import sync_playwright
@ -23,47 +22,71 @@ def run_recon():
def handle_response(response):
url = response.url
content_type = response.headers.get("content-type", "")
all_requests.append({
"url": url,
"status": response.status,
"content_type": content_type,
})
all_requests.append(
{
"url": url,
"status": response.status,
"content_type": content_type,
}
)
if "json" in content_type or "graphql" in url.lower():
try:
body = response.json()
api_responses.append({
"url": url,
"status": response.status,
"body_preview": json.dumps(body)[:2000],
})
api_responses.append(
{
"url": url,
"status": response.status,
"body_preview": json.dumps(body)[:2000],
}
)
except Exception:
pass
elif url.endswith(".json") or "/api/" in url:
try:
body = response.text()
api_responses.append({
"url": url,
"status": response.status,
"body_preview": body[:2000],
})
api_responses.append(
{
"url": url,
"status": response.status,
"body_preview": body[:2000],
}
)
except Exception:
pass
page.on("response", handle_response)
print("Navigating to https://redefinemeals.com/menu ...")
page.goto("https://redefinemeals.com/menu", wait_until="networkidle", timeout=60000)
page.goto(
"https://redefinemeals.com/menu", wait_until="networkidle", timeout=60000
)
page.wait_for_timeout(5000)
print(f"\n{'='*60}")
print(f"\n{'=' * 60}")
print("PAGE TITLE:", page.title())
print(f"{'='*60}")
print(f"{'=' * 60}")
print(f"\n--- All network requests ({len(all_requests)} total) ---")
for req in all_requests:
if any(ext in req["url"] for ext in [".png", ".jpg", ".jpeg", ".gif", ".svg", ".ico", ".woff", ".woff2", ".ttf", ".css"]):
if any(
ext in req["url"]
for ext in [
".png",
".jpg",
".jpeg",
".gif",
".svg",
".ico",
".woff",
".woff2",
".ttf",
".css",
]
):
continue
print(f" [{req['status']}] {req['content_type'][:30]:30s} {req['url'][:120]}")
print(
f" [{req['status']}] {req['content_type'][:30]:30s} {req['url'][:120]}"
)
print(f"\n--- JSON/API responses ({len(api_responses)} found) ---")
for resp in api_responses:
@ -71,11 +94,17 @@ def run_recon():
print(f" Status: {resp['status']}")
print(f" Body preview:\n {resp['body_preview'][:500]}")
print(f"\n--- Page structure (meal-related elements) ---")
print("\n--- Page structure (meal-related elements) ---")
for selector in [
"[class*='meal']", "[class*='menu']", "[class*='product']",
"[class*='item']", "[class*='card']", "[data-product]",
"[data-item]", ".grid > div", "article",
"[class*='meal']",
"[class*='menu']",
"[class*='product']",
"[class*='item']",
"[class*='card']",
"[data-product]",
"[data-item]",
".grid > div",
"article",
]:
elements = page.query_selector_all(selector)
if elements:
@ -88,8 +117,13 @@ def run_recon():
print(f" Text preview: {inner[:200]}")
# Also dump outer HTML of likely meal containers
print(f"\n--- Raw HTML sample (first product/card element) ---")
for selector in ["[class*='product']", "[class*='card']", "[class*='meal']", "[class*='menu-item']"]:
print("\n--- Raw HTML sample (first product/card element) ---")
for selector in [
"[class*='product']",
"[class*='card']",
"[class*='meal']",
"[class*='menu-item']",
]:
els = page.query_selector_all(selector)
if els:
html = els[0].evaluate("el => el.outerHTML")
@ -99,7 +133,7 @@ def run_recon():
break
# Check for Shopify, WooCommerce, or other known platforms
print(f"\n--- Platform detection ---")
print("\n--- Platform detection ---")
platform_checks = {
"Shopify": "Shopify" in page.content(),
"WooCommerce": "woocommerce" in page.content().lower(),

View file

@ -28,7 +28,11 @@ def run():
page.on("response", capture_response)
page.goto("https://www.redefinemeals.com/menu", wait_until="networkidle", timeout=60000)
page.goto(
"https://www.redefinemeals.com/menu",
wait_until="networkidle",
timeout=60000,
)
page.wait_for_timeout(3000)
# Extract detailed structure from first few article cards
@ -44,23 +48,44 @@ def run():
# Check for dietary tag elements
print("\n--- Dietary/tag elements ---")
for sel in ["[class*='tag']", "[class*='diet']", "[class*='label']", "[class*='badge']", "[class*='filter']"]:
for sel in [
"[class*='tag']",
"[class*='diet']",
"[class*='label']",
"[class*='badge']",
"[class*='filter']",
]:
els = page.query_selector_all(sel)
if els:
texts = [e.inner_text().strip() for e in els[:10] if e.inner_text().strip()]
texts = [
e.inner_text().strip() for e in els[:10] if e.inner_text().strip()
]
print(f" {sel}: {len(els)} elements, samples: {texts}")
# Check for filter/category buttons
print("\n--- Filter/category buttons ---")
for sel in ["button", "[class*='filter']", "[class*='category']", "[class*='tab']"]:
for sel in [
"button",
"[class*='filter']",
"[class*='category']",
"[class*='tab']",
]:
els = page.query_selector_all(sel)
if els:
texts = [e.inner_text().strip() for e in els[:20] if e.inner_text().strip()]
texts = [
e.inner_text().strip() for e in els[:20] if e.inner_text().strip()
]
print(f" {sel}: {texts}")
# Try known API patterns
print("\n--- Trying API endpoints ---")
for path in ["/api/menu", "/api/products", "/api/meals", "/api/items", "/api/categories"]:
for path in [
"/api/menu",
"/api/products",
"/api/meals",
"/api/items",
"/api/categories",
]:
try:
resp = page.evaluate(f"""
async () => {{
@ -85,7 +110,13 @@ def run():
quick_view_btns[0].click()
page.wait_for_timeout(2000)
# Look for modal content
for sel in [".modal", "[class*='modal']", "[class*='popup']", "[class*='quick-view']", "[class*='quickview']"]:
for sel in [
".modal",
"[class*='modal']",
"[class*='popup']",
"[class*='quick-view']",
"[class*='quickview']",
]:
modal = page.query_selector(sel)
if modal and modal.is_visible():
print(f" Modal selector: {sel}")
@ -94,7 +125,7 @@ def run():
except Exception as e:
print(f" Quick View click failed: {e}")
print(f"\n--- API calls captured ---")
print("\n--- API calls captured ---")
for hit in api_hits:
print(f" {hit['url']}")
print(f" {json.dumps(hit['body'])[:500]}\n")

View file

@ -9,7 +9,6 @@ replace the browser scrape in the future.
import json
import sys
import os
from datetime import datetime
from pathlib import Path
from playwright.sync_api import sync_playwright, TimeoutError as PwTimeout
@ -114,7 +113,9 @@ def _extract_card(article) -> dict | None:
calories, protein = _parse_macros(macros_text)
tag_els = article.query_selector_all(".diet_mini_tag")
dietary_tags = [t.inner_text().strip().title() for t in tag_els if t.inner_text().strip()]
dietary_tags = [
t.inner_text().strip().title() for t in tag_els if t.inner_text().strip()
]
img_el = article.query_selector("img.main_meal_img")
image_url = img_el.get_attribute("src") if img_el else None
@ -148,34 +149,51 @@ def _enrich_with_descriptions(page, articles, meals):
article.query_selector(".card_media_wrap").click()
page.wait_for_timeout(800)
modal = page.query_selector(".modal.show, [class*='modal'][class*='show'], [class*='quickview']")
modal = page.query_selector(
".modal.show, [class*='modal'][class*='show'], [class*='quickview']"
)
if not modal:
# Try broader selector
modal = page.query_selector("[class*='modal']:not([style*='display: none'])")
modal = page.query_selector(
"[class*='modal']:not([style*='display: none'])"
)
if modal and modal.is_visible():
desc_el = modal.query_selector("[class*='description'], [class*='desc'], .meal_description")
desc_el = modal.query_selector(
"[class*='description'], [class*='desc'], .meal_description"
)
if desc_el:
desc_text = desc_el.inner_text().strip()
# Filter out price strings and very short text
if desc_text and len(desc_text) > 10 and not desc_text.startswith("$"):
if (
desc_text
and len(desc_text) > 10
and not desc_text.startswith("$")
):
meals[i]["description"] = desc_text
# Grab full macro details if available
detail_tags = modal.query_selector_all(".diet_mini_tag, [class*='lifestyle'] span")
detail_tags = modal.query_selector_all(
".diet_mini_tag, [class*='lifestyle'] span"
)
for tag_el in detail_tags:
tag_text = tag_el.inner_text().strip().title()
if tag_text and tag_text not in meals[i]["dietary_tags"]:
meals[i]["dietary_tags"].append(tag_text)
# Close modal
close_btn = modal.query_selector("button[class*='close'], [aria-label='Close'], .btn-close")
close_btn = modal.query_selector(
"button[class*='close'], [aria-label='Close'], .btn-close"
)
if close_btn:
close_btn.click()
else:
page.keyboard.press("Escape")
page.wait_for_timeout(300)
except Exception as e:
print(f" Warning: Quick View failed for meal {i} ({meals[i]['name']}): {e}", file=sys.stderr)
print(
f" Warning: Quick View failed for meal {i} ({meals[i]['name']}): {e}",
file=sys.stderr,
)
try:
page.keyboard.press("Escape")
page.wait_for_timeout(300)
@ -183,12 +201,22 @@ def _enrich_with_descriptions(page, articles, meals):
pass
KNOWN_TAGS = ["Gluten Free", "Dairy Free", "Grass-Fed", "Low Carb", "Keto", "Vegan", "Vegetarian", "Nut Free"]
KNOWN_TAGS = [
"Gluten Free",
"Dairy Free",
"Grass-Fed",
"Low Carb",
"Keto",
"Vegan",
"Vegetarian",
"Nut Free",
]
def _clean_tags(raw_tags: list[str]) -> list[str]:
"""Split concatenated tags and deduplicate."""
import re
cleaned = set()
for raw in raw_tags:
# Split on known tag boundaries (e.g., "Gluten Freedairy Free" → "Gluten Free", "Dairy Free")
@ -196,7 +224,9 @@ def _clean_tags(raw_tags: list[str]) -> list[str]:
for known in KNOWN_TAGS:
if known.lower() in remaining.lower():
cleaned.add(known)
remaining = re.sub(re.escape(known), "", remaining, flags=re.IGNORECASE).strip()
remaining = re.sub(
re.escape(known), "", remaining, flags=re.IGNORECASE
).strip()
if remaining and len(remaining) > 2:
cleaned.add(remaining.strip().title())
return sorted(cleaned)
@ -213,6 +243,7 @@ def _parse_price(text: str) -> float | None:
def _parse_macros(text: str) -> tuple[int | None, str | None]:
"""Parse '570cal • 39gP' into (570, '39g')."""
import re
cal_match = re.search(r"(\d+)\s*cal", text, re.IGNORECASE)
prot_match = re.search(r"(\d+g?)\s*P", text)
calories = int(cal_match.group(1)) if cal_match else None
@ -225,7 +256,9 @@ def _parse_macros(text: str) -> tuple[int | None, str | None]:
def main():
config = load_config()
url = config.get("menu_url", "https://www.redefinemeals.com/menu")
output_dir = Path(__file__).resolve().parents[2] / config.get("output_dir", "output")
output_dir = Path(__file__).resolve().parents[2] / config.get(
"output_dir", "output"
)
output_dir.mkdir(exist_ok=True)
print(f"Scraping menu from {url} ...")
@ -238,7 +271,7 @@ def main():
print(f"\nScraped {result['meal_count']} meals")
if result["api_endpoints_found"]:
print(f"API endpoints detected (potential future shortcut):")
print("API endpoints detected (potential future shortcut):")
for ep in result["api_endpoints_found"]:
print(f" {ep}")
print(f"Output saved to {output_file}")

View file

@ -6,7 +6,6 @@ Orders are saved as JSON files in the orders directory, one per employee per wee
"""
import json
import os
from datetime import datetime
from pathlib import Path
@ -46,7 +45,9 @@ def index():
def get_menu():
menu_file = latest_menu_file()
if not menu_file:
return jsonify({"error": "No menu data available. Run scrape_menu.py first."}), 404
return jsonify(
{"error": "No menu data available. Run scrape_menu.py first."}
), 404
with open(menu_file) as f:
return jsonify(json.load(f))
@ -78,7 +79,12 @@ def submit_order():
week_dir = ORDERS_DIR / week
week_dir.mkdir(parents=True, exist_ok=True)
safe_name = "".join(c if c.isalnum() or c in "-_ " else "" for c in name).strip().replace(" ", "-").lower()
safe_name = (
"".join(c if c.isalnum() or c in "-_ " else "" for c in name)
.strip()
.replace(" ", "-")
.lower()
)
order_file = week_dir / f"{safe_name}.json"
order = {
@ -97,7 +103,9 @@ def submit_order():
with open(order_file, "w") as f:
json.dump(order, f, indent=2)
return jsonify({"status": "ok", "message": f"Order saved for {name}", "total": order["total"]})
return jsonify(
{"status": "ok", "message": f"Order saved for {name}", "total": order["total"]}
)
@app.route("/api/orders/<week>")

View file

@ -34,7 +34,9 @@ def latest_menu() -> dict:
return json.load(f)
def generate_form(menu: dict, config: dict, api_url: str = "", api_key: str = "") -> str:
def generate_form(
menu: dict, config: dict, api_url: str = "", api_key: str = ""
) -> str:
meals_json = json.dumps(menu["meals"])
roster_json = json.dumps(config.get("roster", []))
deadline = config.get("order_deadline", "Thursday 6:00 PM")
@ -295,8 +297,12 @@ init();
def main():
parser = argparse.ArgumentParser(description="Generate meal order form HTML")
parser.add_argument("--api-url", default="", help="API Gateway base URL (cloud mode)")
parser.add_argument("--api-key", default="", help="API key for order submission (cloud mode)")
parser.add_argument(
"--api-url", default="", help="API Gateway base URL (cloud mode)"
)
parser.add_argument(
"--api-key", default="", help="API key for order submission (cloud mode)"
)
args = parser.parse_args()
config = load_config()
@ -312,7 +318,7 @@ def main():
print(f"Generated order form ({mode} mode): {output_file}")
print(f" {menu['meal_count']} meals from menu scraped {menu['scraped_at'][:10]}")
if mode == "local":
print(f" Start the server with: python3 src/server/app.py")
print(" Start the server with: python3 src/server/app.py")
else:
print(f" API endpoint: {args.api_url}")

View file

@ -33,16 +33,18 @@ def ttl_days(days: int) -> int:
def put_menu(week: str, menu_data: dict):
_get_table().put_item(Item={
"PK": f"WEEK#{week}",
"SK": "MENU",
"form_status": "open",
"scraped_at": menu_data.get("scraped_at"),
"menu_url": menu_data.get("menu_url"),
"meal_count": menu_data.get("meal_count"),
"meals": menu_data.get("meals"),
"ttl": ttl_days(90),
})
_get_table().put_item(
Item={
"PK": f"WEEK#{week}",
"SK": "MENU",
"form_status": "open",
"scraped_at": menu_data.get("scraped_at"),
"menu_url": menu_data.get("menu_url"),
"meal_count": menu_data.get("meal_count"),
"meals": menu_data.get("meals"),
"ttl": ttl_days(90),
}
)
def get_menu(week: str) -> dict | None:
@ -71,32 +73,39 @@ def _to_decimal(obj):
def put_order(week: str, employee_slug: str, order_data: dict):
_get_table().put_item(Item=_to_decimal({
"PK": f"WEEK#{week}",
"SK": f"ORDER#{employee_slug}",
"employee_name": order_data["employee_name"],
"employee_email": order_data["employee_email"],
"submitted_at": order_data["submitted_at"],
"items": order_data["items"],
"total": order_data["total"],
"ttl": ttl_days(90),
}))
_get_table().put_item(
Item=_to_decimal(
{
"PK": f"WEEK#{week}",
"SK": f"ORDER#{employee_slug}",
"employee_name": order_data["employee_name"],
"employee_email": order_data["employee_email"],
"submitted_at": order_data["submitted_at"],
"items": order_data["items"],
"total": order_data["total"],
"ttl": ttl_days(90),
}
)
)
def get_orders(week: str) -> list[dict]:
resp = _get_table().query(
KeyConditionExpression=Key("PK").eq(f"WEEK#{week}") & Key("SK").begins_with("ORDER#"),
KeyConditionExpression=Key("PK").eq(f"WEEK#{week}")
& Key("SK").begins_with("ORDER#"),
)
return resp.get("Items", [])
def put_summary(week: str, summary: dict):
_get_table().put_item(Item={
"PK": f"WEEK#{week}",
"SK": "SUMMARY",
"ttl": ttl_days(90),
**summary,
})
_get_table().put_item(
Item={
"PK": f"WEEK#{week}",
"SK": "SUMMARY",
"ttl": ttl_days(90),
**summary,
}
)
def get_summary(week: str) -> dict | None:
@ -111,9 +120,11 @@ def get_roster() -> list[dict]:
def put_roster(employees: list[dict]):
_get_table().put_item(Item={
"PK": "CONFIG",
"SK": "ROSTER",
"employees": employees,
"updated_at": datetime.now(EASTERN).isoformat(),
})
_get_table().put_item(
Item={
"PK": "CONFIG",
"SK": "ROSTER",
"employees": employees,
"updated_at": datetime.now(EASTERN).isoformat(),
}
)

View file

@ -1,4 +1,3 @@
import os
import boto3
_cache = {}