""" Lunch Notifier — pulls the Mariposa lunch menu and pushes it to a Home Assistant calendar. Visit http://localhost:2323 to manage settings. """ import calendar import json import logging import os import threading import time from datetime import datetime from pathlib import Path import requests import schedule from flask import Flask, jsonify, request from scraper import get_month_menu logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger(__name__) app = Flask(__name__) SETTINGS_FILE = Path(os.environ.get("SETTINGS_FILE", "/data/settings.json")) PORT = int(os.environ.get("PORT", "2323")) # Env-var defaults (used only when no settings file exists yet) _DEFAULTS = { "calendar_run_time": os.environ.get("CALENDAR_RUN_TIME", "18:00"), "calendar_run_day_of_month": int(os.environ.get("CALENDAR_RUN_DAY_OF_MONTH", "25")), } _status = {"last_calendar_run": None, "calendar_result": None} _lock = threading.Lock() # ── Settings ────────────────────────────────────────────────────────────────── def load_settings() -> dict: if SETTINGS_FILE.exists(): try: with open(SETTINGS_FILE) as f: # Merge over _DEFAULTS so a settings.json from before a schema # change (e.g. the pre-calendar SMS-era file, or the # weekly-schedule era before calendar_run_day_of_month # existed) still gets a valid schedule rather than silently # scheduling nothing. return {**_DEFAULTS, **json.load(f)} except Exception: pass return dict(_DEFAULTS) def save_settings(s: dict) -> None: SETTINGS_FILE.parent.mkdir(parents=True, exist_ok=True) with open(SETTINGS_FILE, "w") as f: json.dump(s, f, indent=2) # ── Calendar push ───────────────────────────────────────────────────────────── def _push_to_home_assistant(items: list[str], date_str: str) -> None: webhook_url = os.environ["HA_WEBHOOK_URL"] log.info(f"Pushing {date_str} menu → Home Assistant webhook") resp = requests.post(webhook_url, json={"date": date_str, "items": items}, timeout=10) resp.raise_for_status() log.info(f"Home Assistant push ✓ ({date_str})") def _push_calendar(items: list[str], date_str: str | None) -> str: """Push one day to Home Assistant if configured. Returns a status string.""" if not items: return "no items" if not os.environ.get("HA_WEBHOOK_URL"): return "not configured" try: _push_to_home_assistant(items, date_str) return "ok" except Exception as e: log.error(f"Home Assistant push failed ({date_str}): {e}") return f"error: {e}" def push_month_to_calendar() -> None: """Fetch whichever month is currently published and push every day that has a posted menu — days with nothing posted (weekends, breaks, holidays) are skipped automatically since there's nothing to create an event from. Runs on its own schedule (calendar_run_time/ calendar_run_day_of_month), independent of anything else.""" global _status try: month = get_month_menu() except Exception as e: log.exception("Month fetch failed") _status = {**_status, "last_calendar_run": datetime.now().isoformat(), "calendar_result": f"error: {e}"} return pushed = errors = no_items = 0 for day in month: result = _push_calendar(day["items"], day["date"]) if result == "ok": pushed += 1 elif result == "no items": no_items += 1 else: errors += 1 log.info(f"Month push: {pushed} pushed, {errors} errors, {no_items} no menu") if errors: summary = f"{pushed} ok, {errors} error" + ("s" if errors != 1 else "") elif pushed: summary = f"ok ({pushed} day{'s' if pushed != 1 else ''})" else: summary = "no items" _status = { **_status, "last_calendar_run": datetime.now().isoformat(), "calendar_result": summary, } # ── Scheduler ───────────────────────────────────────────────────────────────── def _maybe_push_month(day_of_month: int) -> None: """schedule fires this daily at run_time; only actually pushes on the configured day of the month (the `schedule` library has no native day-of-month trigger).""" if datetime.now().day == day_of_month: push_month_to_calendar() def apply_calendar_schedule(settings: dict) -> None: with _lock: schedule.clear("calendar") run_time = settings.get("calendar_run_time", "18:00") day_of_month = settings.get("calendar_run_day_of_month", 25) schedule.every().day.at(run_time).do(_maybe_push_month, day_of_month).tag("calendar") log.info(f"Scheduled calendar pull: day {day_of_month} of each month at {run_time}") def next_calendar_run(settings: dict) -> datetime: """The `schedule` job above fires daily (it has no native day-of-month trigger — see _maybe_push_month), so its own next_run is just "tomorrow", not the actual next push. Compute the real one directly from settings for display in the UI.""" run_time = settings.get("calendar_run_time", "18:00") hour, minute = (int(p) for p in run_time.split(":")) day_of_month = settings.get("calendar_run_day_of_month", 25) now = datetime.now() candidate = now.replace( day=min(day_of_month, calendar.monthrange(now.year, now.month)[1]), hour=hour, minute=minute, second=0, microsecond=0, ) if candidate <= now: y, m = (now.year, now.month + 1) if now.month < 12 else (now.year + 1, 1) candidate = candidate.replace( year=y, month=m, day=min(day_of_month, calendar.monthrange(y, m)[1]), ) return candidate def _schedule_loop() -> None: while True: with _lock: schedule.run_pending() time.sleep(30) # ── Flask API ───────────────────────────────────────────────────────────────── MASCOT_URL = "https://resources.finalsite.net/images/f_png,q_auto/v1708515016/kyreneorg/caewi1dctc6afx1d8nq1/Logo_Mariposa-Mascot.png" @app.route("/") def index(): return HTML, 200, {"Content-Type": "text/html"} @app.route("/icon.png") def icon(): r = requests.get(MASCOT_URL, timeout=10) return r.content, 200, {"Content-Type": "image/png", "Cache-Control": "public, max-age=86400"} @app.route("/api/settings", methods=["GET"]) def api_get_settings(): return jsonify(load_settings()) @app.route("/api/settings", methods=["POST"]) def api_post_settings(): data = request.get_json(force=True) s = load_settings() for key in ("calendar_run_time", "calendar_run_day_of_month"): if key in data: s[key] = data[key] save_settings(s) apply_calendar_schedule(s) return jsonify({"ok": True}) @app.route("/api/push-calendar", methods=["POST"]) def api_push_calendar(): threading.Thread(target=push_month_to_calendar, daemon=True).start() return jsonify({"ok": True}) @app.route("/api/preview") def api_preview(): """Dry-run month scrape: no Home Assistant push, no side effects.""" try: month = get_month_menu() except Exception as e: return jsonify({"error": str(e)}), 502 return jsonify({"days": month}) @app.route("/api/integrations") def api_integrations(): return jsonify({ "ha_webhook_configured": bool(os.environ.get("HA_WEBHOOK_URL")), "anthropic_api_key_configured": bool(os.environ.get("ANTHROPIC_API_KEY")), }) @app.route("/api/status") def api_status(): return jsonify({ **_status, "next_calendar_run": next_calendar_run(load_settings()).isoformat(), }) # ── Embedded UI ─────────────────────────────────────────────────────────────── HTML = """ 🍱 Lunch Notifier
Chip the Challenger

Lunch Notifier

Kyrene de la Mariposa · Menu → Home Assistant calendar

Status
Last Pull
—
Result
—
Next Scheduled
—
Manual Actions
Integrations
Home Assistant
—
Anthropic API
—
Configured via .env (HA_WEBHOOK_URL, ANTHROPIC_API_KEY) on the server, not here.
Calendar Pull Schedule
Fires once a month and pushes whichever month is currently published — e.g. day 25 catches next month's menu once the district posts it near month-end. There's no dedup, so re-pushing the same month creates duplicate calendar events.
""" # ── Entrypoint ──────────────────────────────────────────────────────────────── if __name__ == "__main__": if not SETTINGS_FILE.exists(): log.info(f"Creating default settings at {SETTINGS_FILE}") save_settings(_DEFAULTS) apply_calendar_schedule(load_settings()) threading.Thread(target=_schedule_loop, daemon=True).start() log.info(f"Web UI → http://localhost:{PORT}") app.run(host="0.0.0.0", port=PORT, debug=False, use_reloader=False)