""" Lunch Notifier — web UI + scheduler combined. Visit http://localhost:2323 to manage settings. """ import json import logging import os import smtplib import threading import time from datetime import date, datetime from email.message import EmailMessage from pathlib import Path import requests import schedule from flask import Flask, jsonify, request from scraper import get_lunch_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 = { "phone_numbers": [p for p in [os.environ.get("PHONE_NUMBER", "")] if p], "carrier_gateway": os.environ.get("CARRIER_GATEWAY", "vtext.com"), "run_time": os.environ.get("RUN_TIME", "07:00"), "run_days": [d.strip() for d in os.environ.get("RUN_DAYS", "mon,tue,wed,thu,fri").split(",") if d.strip()], "vacations": [], } _status = { "last_run": None, "last_result": None, "last_message": None, "calendar_result": None, "last_calendar_run": None, } _lock = threading.Lock() # ── Settings ────────────────────────────────────────────────────────────────── def load_settings() -> dict: if SETTINGS_FILE.exists(): try: with open(SETTINGS_FILE) as f: return 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) def is_on_vacation(settings: dict) -> str | None: today = date.today().isoformat() for v in settings.get("vacations", []): if v["start"] <= today <= v["end"]: return v.get("label", "Vacation") return None # ── Core job ────────────────────────────────────────────────────────────────── def _send_email_sms(message: str, phone: str, gateway: str) -> None: gmail_user = os.environ["GMAIL_USER"] gmail_pass = os.environ["GMAIL_APP_PASSWORD"] to_addr = f"{phone}@{gateway}" msg = EmailMessage() msg["From"] = gmail_user msg["To"] = to_addr msg.set_content(message) log.info(f"Sending → {to_addr}") with smtplib.SMTP("smtp.gmail.com", 587, timeout=15) as smtp: smtp.starttls() smtp.login(gmail_user, gmail_pass) smtp.send_message(msg) log.info(f"Sent ✓ → {to_addr}") def _push_to_home_assistant(items: list[str], date_str: str) -> None: webhook_url = os.environ["HA_WEBHOOK_URL"] log.info(f"Pushing menu → Home Assistant webhook") resp = requests.post(webhook_url, json={"date": date_str, "items": items}, timeout=10) resp.raise_for_status() log.info("Home Assistant push ✓") def _scrape_menu() -> tuple[list[str], str | None, str]: """Fetch the menu and build the display/SMS message. No side effects.""" try: items, date_str = get_lunch_menu() except Exception as e: log.exception("Menu fetch failed") return [], None, f"Error fetching Mariposa lunch menu: {e}" if items: body = "\n".join(f"• {i}" for i in items) message = f"Mariposa Lunch {date_str}:\n{body}" else: message = f"No lunch menu posted for {date_str} (may be a break or holiday)." if len(message) > 1400: message = message[:1397] + "..." return items, date_str, message def _push_calendar(items: list[str], date_str: str | None) -> str: """Push to Home Assistant if configured. Returns a calendar_result 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: {e}") return f"error: {e}" def run_job() -> None: """Full daily job: scrape, push to HA calendar, and text everyone.""" global _status settings = load_settings() vac = is_on_vacation(settings) if vac: log.info(f"Skipping — vacation: {vac}") _status = {**_status, "last_run": datetime.now().isoformat(), "last_result": f"skipped ({vac})"} return phones = [p.strip() for p in settings.get("phone_numbers", []) if p.strip()] gateway = settings.get("carrier_gateway", "vtext.com") items, date_str, message = _scrape_menu() log.info(f"Message:\n{message}") calendar_result = _push_calendar(items, date_str) errors = [] for phone in phones: try: _send_email_sms(message, phone, gateway) except Exception as e: log.error(f"Failed to send to {phone}: {e}") errors.append(str(e)) _status = { **_status, "last_run": datetime.now().isoformat(), "last_result": "error: " + "; ".join(errors) if errors else "ok", "last_message": message, "calendar_result": calendar_result, "last_calendar_run": datetime.now().isoformat(), } def push_calendar_only() -> None: """Scrape and push to Home Assistant without texting anyone.""" global _status items, date_str, message = _scrape_menu() log.info(f"Calendar-only push — message:\n{message}") calendar_result = _push_calendar(items, date_str) _status = { **_status, "calendar_result": calendar_result, "last_calendar_run": datetime.now().isoformat(), } # ── Scheduler ───────────────────────────────────────────────────────────────── DAY_MAP_FNS = { "mon": lambda: schedule.every().monday, "tue": lambda: schedule.every().tuesday, "wed": lambda: schedule.every().wednesday, "thu": lambda: schedule.every().thursday, "fri": lambda: schedule.every().friday, "sat": lambda: schedule.every().saturday, "sun": lambda: schedule.every().sunday, } def apply_schedule(settings: dict) -> None: with _lock: schedule.clear() run_time = settings.get("run_time", "07:00") for day in settings.get("run_days", []): day = day.strip().lower() if day in DAY_MAP_FNS: DAY_MAP_FNS[day]().at(run_time).do(run_job) log.info(f"Scheduled: {day} at {run_time}") 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 ("phone_numbers", "carrier_gateway", "run_time", "run_days"): if key in data: s[key] = data[key] save_settings(s) apply_schedule(s) return jsonify({"ok": True}) @app.route("/api/vacation", methods=["POST"]) def api_add_vacation(): data = request.get_json(force=True) s = load_settings() s.setdefault("vacations", []).append({ "id": str(int(time.time() * 1000)), "start": data["start"], "end": data["end"], "label": data.get("label", "Vacation"), }) save_settings(s) return jsonify({"ok": True}) @app.route("/api/vacation/", methods=["DELETE"]) def api_delete_vacation(vid: str): s = load_settings() s["vacations"] = [v for v in s.get("vacations", []) if str(v.get("id")) != vid] save_settings(s) return jsonify({"ok": True}) @app.route("/api/send", methods=["POST"]) def api_send(): threading.Thread(target=run_job, daemon=True).start() return jsonify({"ok": True}) @app.route("/api/push-calendar", methods=["POST"]) def api_push_calendar(): threading.Thread(target=push_calendar_only, daemon=True).start() return jsonify({"ok": True}) @app.route("/api/preview") def api_preview(): """Dry-run scrape: no SMS, no Home Assistant push.""" items, date_str, message = _scrape_menu() return jsonify({"date": date_str, "items": items, "message": message}) @app.route("/api/integrations") def api_integrations(): return jsonify({ "sms_configured": bool(os.environ.get("GMAIL_USER") and os.environ.get("GMAIL_APP_PASSWORD")), "ha_webhook_configured": bool(os.environ.get("HA_WEBHOOK_URL")), }) @app.route("/api/status") def api_status(): s = load_settings() with _lock: next_runs = sorted([j.next_run for j in schedule.get_jobs() if j.next_run]) return jsonify({ **_status, "next_run": next_runs[0].isoformat() if next_runs else None, "on_vacation": is_on_vacation(s), }) # ── Embedded UI ─────────────────────────────────────────────────────────────── HTML = """ 🍱 Lunch Notifier
Chip the Challenger

Lunch Notifier

Kyrene de la Mariposa · Automated lunch menu texts

Status
Last Run
—
Result
—
Next Scheduled
—
Mode
—
Manual Actions
Integrations
SMS (Gmail)
—
Home Assistant
—
Configured via .env on the server, not here.
Schedule
Phone Numbers
    Vacation / No-Send Periods
      """ # ── Entrypoint ──────────────────────────────────────────────────────────────── if __name__ == "__main__": if not SETTINGS_FILE.exists(): log.info(f"Creating default settings at {SETTINGS_FILE}") save_settings(_DEFAULTS) apply_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)