"""Slot Callback Handler — Execute actions when a user taps a signup slot button.
When the Clicker posts signup slots with inline buttons (format: slot|
this handler processes the button tap by:
- Looking up the original URL and slot data from the slot cache
- Creating a [REMINDER] calendar event for the selected slot
- Sending a confirmation message with the direct signup URL so the user
can complete the signup on their phone
Architecture constraint: We cannot actually sign up on behalf of the user
(browser auth, CAPTCHAs, etc.). Instead, we create a calendar placeholder
and provide the pre-filled URL for the user to complete the signup.
Slot data is cached in memory/ slot-cache/ as JSON files keyed by url_hash.
"""
import json
import os
import re
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from family_assistant.config import CHICAGO_TZ
Slot cache directory — stores URL + slot data keyed by url_hash
SLOT_CACHE_DIR = os.path.join(
os.path.dirname(file), "..", "memory", "slot-cache"
)
Ensure cache directory exists
os.makedirs(SLOT_CACHE_DIR, exist_ok=True)
def cache_slots(url, slots, url_hash=None):
"""Cache slot data so the callback handler can look it up later.
Called by Hermes when posting slot buttons — stores the URL and slot
data as a JSON file keyed by url_hash.
Args:
url: The original signup URL
slots: List of slot dicts from the Clicker
url_hash: Short hash of the URL (generated by clicker.hash_url)
Returns:
The url_hash used for the cache key
"""
if not url_hash:
from family_assistant.clicker import hash_url
url_hash = hash_url(url)
cache_path = os.path.join(SLOT_CACHE_DIR, f"{url_hash}.json")
cache_data = {
"url": url,
"url_hash": url_hash,
"slots": slots,
"cached_at": datetime.now(CHICAGO_TZ).isoformat(),
}
with open(cache_path, "w") as f:
json.dump(cache_data, f, indent=2, ensure_ascii=False)
return url_hash
def get_cached_slots(url_hash):
"""Look up cached slot data by url_hash.
Args:
url_hash: The short hash identifying the slot set
Returns:
The cache dict with url, slots, etc., or None if not found
"""
cache_path = os.path.join(SLOT_CACHE_DIR, f"{url_hash}.json")
if not os.path.exists(cache_path):
return None
try:
with open(cache_path) as f:
return json.load(f)
except (json.JSONDecodeError, OSError):
return None
def parse_slot_callback(callback_data):
"""Parse a slot callback data string.
Format: slot|<url_hash>|<slot_index>
slot_index is 0-based (matching the slot list index).
Args:
callback_data: The callback_data string from the Telegram button
Returns:
Dict with url_hash and slot_index, or None if invalid format
"""
parts = callback_data.split("|")
if len(parts) != 3 or parts[0] != "slot":
return None
try:
return {"url_hash": parts[1], "slot_index": int(parts[2])}
except (ValueError, IndexError):
return None
def handle_slot_callback(callback_data, dry_run=False):
"""Handle a slot button tap from Telegram.
This is the main entry point, called when a user taps a signup slot button.
Flow:
1. Parse the callback data to get url_hash and slot_index
2. Look up the cached slot data
3. Find the selected slot
4. Create a [REMINDER] calendar event for the slot
5. Return a confirmation message with the signup URL
Args:
callback_data: The callback_data string (format: slot|<url_hash>|<slot_index>)
dry_run: If True, don't create calendar events or send messages
Returns:
Dict with status, confirmation message, and calendar event details
"""
parsed = parse_slot_callback(callback_data)
if not parsed:
return {"status": "error", "message": f"Invalid callback format: {callback_data}"}
url_hash = parsed["url_hash"]
slot_index = parsed["slot_index"]
# Look up cached data
cached = get_cached_slots(url_hash)
if not cached:
return {
"status": "error",
"message": f"Slot data not found (hash: {url_hash}). The cache may have expired. Try fetching the signup page again.",
}
slots = cached.get("slots", [])
url = cached.get("url", "")
if slot_index >= len(slots):
return {
"status": "error",
"message": f"Invalid slot index {slot_index}. Only {len(slots)} slots available.",
}
slot = slots[slot_index]
# Extract slot details
time_str = slot.get("time", "")
date_str = slot.get("date", "")
label = slot.get("label", "Signup")
spots = slot.get("spots_remaining")
category = slot.get("category", "")
# Determine the date/time for the calendar event
event_date = None
event_time_str = None
friendly_time = None
if time_str:
try:
dt = datetime.fromisoformat(time_str)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=CHICAGO_TZ)
event_date = dt
friendly_time = dt.astimezone(CHICAGO_TZ).strftime("%a %b %d, %-I:%M %p")
event_time_str = dt.strftime("%-I:%M %p")
except (ValueError, TypeError):
friendly_time = time_str
elif date_str:
try:
dt = datetime.fromisoformat(date_str)
event_date = dt
friendly_time = dt.strftime("%a %b %d")
except (ValueError, TypeError):
friendly_time = date_str
# Build the event summary
summary = label if label else "Signup"
if not summary.startswith("[REMINDER]"):
summary = f"[REMINDER] {summary}"
# Build description with the signup URL
description_parts = [f"Selected slot: {friendly_time or 'TBD'}"]
if spots is not None:
description_parts.append(f"Spots remaining: {spots}")
description_parts.append(f"Sign up here: {url}")
description = " | ".join(description_parts)
# Create the calendar event
calendar_result = None
if event_date and not dry_run:
calendar_result = _create_slot_reminder(summary, event_date, description, time_str)
elif event_date and dry_run:
calendar_result = {"status": "DRY_RUN", "summary": summary, "date": friendly_time}
# Build confirmation message
confirmation = f"✅ **Slot selected**: {label}"
if friendly_time:
confirmation += f" — {friendly_time}"
if spots is not None:
confirmation += f" ({spots} spots left)"
confirmation += f"\n\n🔗 [Complete your signup]({url})"
if calendar_result:
confirmation += f"\n📅 Added to calendar"
return {
"status": "SLOT_SELECTED",
"slot": slot,
"url": url,
"calendar_event": calendar_result,
"confirmation": confirmation,
}
def _create_slot_reminder(summary, event_date, description, time_str):
"""Create a calendar event for the selected slot.
If the slot has a specific time, creates a timed event.
If it's date-only, creates an all-day [REMINDER] event.
Args:
summary: Event summary (already prepended with [REMINDER])
event_date: datetime object for the event
description: Event description with signup URL
time_str: ISO datetime string if timed, empty if all-day
Returns:
Dict with calendar event creation result
"""
from family_assistant.calendar_sync import create_event, get_calendar_service
import os
if time_str:
# Timed event — create a 30-min placeholder
try:
start_dt = datetime.fromisoformat(time_str)
if start_dt.tzinfo is None:
start_dt = start_dt.replace(tzinfo=CHICAGO_TZ)
except (ValueError, TypeError):
start_dt = event_date
end_dt = start_dt + timedelta(minutes=30)
event = create_event(
summary=summary,
start_dt=start_dt,
end_dt=end_dt,
description=description,
)
else:
# All-day reminder
next_day = event_date.date() + timedelta(days=1) if hasattr(event_date, 'date') else event_date + timedelta(days=1)
event = create_event(
summary=summary,
start_dt=event_date.date().isoformat() if hasattr(event_date, 'date') else event_date.isoformat(),
end_dt=next_day.isoformat(),
description=description,
)
return {
"status": "CREATED",
"summary": event.get("summary"),
"id": event["id"],
"link": event.get("htmlLink", ""),
}
def cleanup_old_cache(max_age_days=7):
"""Remove cached slot data older than max_age_days.
Called periodically by the heartbeat to prevent unbounded cache growth.
Args:
max_age_days: Maximum age in days before cache entries are removed
Returns:
Number of cache entries removed
"""
if not os.path.exists(SLOT_CACHE_DIR):
return 0
now = datetime.now(CHICAGO_TZ)
removed = 0
for filename in os.listdir(SLOT_CACHE_DIR):
if not filename.endswith(".json"):
continue
path = os.path.join(SLOT_CACHE_DIR, filename)
try:
with open(path) as f:
data = json.load(f)
cached_at = datetime.fromisoformat(data.get("cached_at", ""))
if (now - cached_at).days > max_age_days:
os.remove(path)
removed += 1
except (json.JSONDecodeError, ValueError, OSError):
# Remove corrupt files too
os.remove(path)
removed += 1
return removed