mirror of
https://github.com/Sea-Haven-Industries/afterhours-shift-manager.git
synced 2026-09-30 10:13:11 +00:00
3a51b52ffc
Anchors to the current week's Monday and shows 14 consecutive days. Updates /oncall next to jump forward by 2 weeks accordingly. Closes #1
297 lines
11 KiB
Python
297 lines
11 KiB
Python
"""Slack Bolt app — /oncall command handlers and interactive actions."""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
from datetime import datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import boto3
|
|
from slack_bolt import App
|
|
|
|
from src.blocks import (
|
|
build_help_blocks,
|
|
build_roster_blocks,
|
|
build_shift_change_message,
|
|
build_week_schedule,
|
|
)
|
|
from src.schedule import FALLBACK_EXTENSION, ShiftSchedule
|
|
|
|
logger = logging.getLogger(__name__)
|
|
EASTERN = ZoneInfo("America/New_York")
|
|
|
|
DAY_NAMES = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"]
|
|
|
|
|
|
def parse_date(text: str) -> datetime | None:
|
|
"""Parse flexible date input: today, tomorrow, day names, m/d, YYYY-MM-DD."""
|
|
now = datetime.now(EASTERN)
|
|
text = text.strip().lower()
|
|
|
|
if text == "today":
|
|
return now
|
|
if text == "tomorrow":
|
|
return now + timedelta(days=1)
|
|
|
|
if text in DAY_NAMES:
|
|
target = DAY_NAMES.index(text)
|
|
current = now.weekday()
|
|
delta = (target - current) % 7
|
|
if delta == 0:
|
|
delta = 7
|
|
return now + timedelta(days=delta)
|
|
|
|
for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%m/%d", "%m-%d"):
|
|
try:
|
|
parsed = datetime.strptime(text, fmt)
|
|
if "%Y" not in fmt:
|
|
parsed = parsed.replace(year=now.year)
|
|
if parsed.replace(tzinfo=EASTERN) < now - timedelta(days=1):
|
|
parsed = parsed.replace(year=now.year + 1)
|
|
return parsed.replace(tzinfo=EASTERN)
|
|
except ValueError:
|
|
continue
|
|
|
|
return None
|
|
|
|
|
|
def invoke_3cx_scheduler(extension: str) -> None:
|
|
"""Invoke the 3CX scheduler Lambda to update the ring group immediately."""
|
|
fn_name = os.environ.get("SCHEDULER_FUNCTION_NAME")
|
|
if not fn_name:
|
|
logger.warning("SCHEDULER_FUNCTION_NAME not set — skipping 3CX update")
|
|
return
|
|
|
|
client = boto3.client("lambda")
|
|
payload = {"force": True, "override_extension": extension}
|
|
logger.info("Invoking %s with %s", fn_name, payload)
|
|
client.invoke(
|
|
FunctionName=fn_name,
|
|
InvocationType="Event", # async — don't wait
|
|
Payload=json.dumps(payload),
|
|
)
|
|
|
|
|
|
def is_today(date_str: str) -> bool:
|
|
return date_str == datetime.now(EASTERN).strftime("%Y-%m-%d")
|
|
|
|
|
|
def create_app(bot_token: str, signing_secret: str) -> App:
|
|
app = App(
|
|
token=bot_token,
|
|
signing_secret=signing_secret,
|
|
process_before_response=True,
|
|
)
|
|
schedule = ShiftSchedule()
|
|
|
|
# ── /oncall command ─────────────────────────────────────────────────
|
|
|
|
@app.command("/oncall")
|
|
def handle_oncall(ack, command, respond, client):
|
|
ack()
|
|
text = (command.get("text") or "").strip()
|
|
user_id = command["user_id"]
|
|
channel_id = command["channel_id"]
|
|
|
|
if not text or text == "schedule":
|
|
_show_schedule(respond, schedule)
|
|
elif text == "next":
|
|
_show_next_week(respond, schedule)
|
|
elif text == "help":
|
|
respond(blocks=build_help_blocks())
|
|
elif text == "roster":
|
|
respond(blocks=build_roster_blocks(schedule.get_roster()))
|
|
elif text.startswith("register"):
|
|
_handle_register(respond, schedule, user_id, text)
|
|
elif text.startswith("pick"):
|
|
_handle_pick(respond, schedule, user_id, text, channel_id, client)
|
|
elif text.startswith("drop"):
|
|
_handle_drop(respond, schedule, user_id, text, channel_id, client)
|
|
elif text.startswith("swap"):
|
|
_handle_swap(respond, schedule, user_id, text, channel_id, client)
|
|
else:
|
|
respond(text="Unknown command. Try `/oncall help`")
|
|
|
|
# ── Interactive button: pick up open shift ──────────────────────────
|
|
|
|
@app.action(re.compile(r"^pickup_"))
|
|
def handle_pickup_button(ack, body, client):
|
|
ack()
|
|
action_id = body["actions"][0]["action_id"]
|
|
date_str = action_id.replace("pickup_", "")
|
|
user_id = body["user"]["id"]
|
|
channel_id = body["channel"]["id"]
|
|
|
|
employee = schedule.get_employee_by_slack_id(user_id)
|
|
if not employee:
|
|
client.chat_postEphemeral(
|
|
channel=channel_id,
|
|
user=user_id,
|
|
text="You're not registered. Use `/oncall register <extension>` first.",
|
|
)
|
|
return
|
|
|
|
schedule.set_override(date_str, employee["extension"], employee["name"])
|
|
|
|
if is_today(date_str):
|
|
invoke_3cx_scheduler(employee["extension"])
|
|
|
|
blocks = build_shift_change_message(
|
|
user_id, date_str, "picked_up", employee["extension"], employee["name"]
|
|
)
|
|
client.chat_postMessage(channel=channel_id, blocks=blocks, text=f"Shift picked up for {date_str}")
|
|
|
|
# ── Subcommand handlers ─────────────────────────────────────────────
|
|
|
|
def _show_schedule(respond, schedule):
|
|
blocks = build_week_schedule(schedule)
|
|
respond(blocks=blocks)
|
|
|
|
def _show_next_week(respond, schedule):
|
|
now = datetime.now(EASTERN)
|
|
# Jump 2 weeks ahead from this week's Monday
|
|
this_monday = now - timedelta(days=now.weekday())
|
|
next_start = this_monday + timedelta(days=14)
|
|
blocks = build_week_schedule(schedule, start_date=next_start)
|
|
respond(blocks=blocks)
|
|
|
|
def _handle_register(respond, schedule, user_id, text):
|
|
parts = text.split()
|
|
if len(parts) < 2:
|
|
respond(text="Usage: `/oncall register <extension>` (e.g. `/oncall register 114`)")
|
|
return
|
|
|
|
ext = parts[1].strip()
|
|
employee = schedule.register_user(user_id, ext)
|
|
if not employee:
|
|
respond(text=f"Extension {ext} not found in the roster. Check `/oncall roster`.")
|
|
return
|
|
|
|
respond(text=f"Linked your account to *{employee['name']}* (Ext {ext}).")
|
|
|
|
def _handle_pick(respond, schedule, user_id, text, channel_id, client):
|
|
parts = text.split(maxsplit=1)
|
|
if len(parts) < 2:
|
|
respond(text="Usage: `/oncall pick <date>` (e.g. `/oncall pick friday`)")
|
|
return
|
|
|
|
employee = schedule.get_employee_by_slack_id(user_id)
|
|
if not employee:
|
|
respond(text="You're not registered. Use `/oncall register <extension>` first.")
|
|
return
|
|
|
|
date = parse_date(parts[1])
|
|
if not date:
|
|
respond(text=f"Couldn't parse date: `{parts[1]}`. Try: today, tomorrow, friday, 4/5, 2026-04-05")
|
|
return
|
|
|
|
date_str = date.strftime("%Y-%m-%d")
|
|
day_name = date.strftime("%A")
|
|
ext, name, source = schedule.resolve_shift(date_str, day_name)
|
|
|
|
# Already assigned to someone else (not open)
|
|
if source in ("weekly", "override") and ext != FALLBACK_EXTENSION:
|
|
respond(text=f"That shift is already covered by {name} (Ext {ext}). They'd need to drop it first.")
|
|
return
|
|
|
|
schedule.set_override(date_str, employee["extension"], employee["name"])
|
|
|
|
if is_today(date_str):
|
|
invoke_3cx_scheduler(employee["extension"])
|
|
|
|
blocks = build_shift_change_message(
|
|
user_id, date_str, "picked_up", employee["extension"], employee["name"]
|
|
)
|
|
client.chat_postMessage(channel=channel_id, blocks=blocks, text=f"Shift picked up for {date_str}")
|
|
respond(text=f"You picked up the shift for *{date.strftime('%A, %b %-d')}*.")
|
|
|
|
def _handle_drop(respond, schedule, user_id, text, channel_id, client):
|
|
parts = text.split(maxsplit=1)
|
|
if len(parts) < 2:
|
|
respond(text="Usage: `/oncall drop <date>` (e.g. `/oncall drop friday`)")
|
|
return
|
|
|
|
employee = schedule.get_employee_by_slack_id(user_id)
|
|
if not employee:
|
|
respond(text="You're not registered. Use `/oncall register <extension>` first.")
|
|
return
|
|
|
|
date = parse_date(parts[1])
|
|
if not date:
|
|
respond(text=f"Couldn't parse date: `{parts[1]}`. Try: today, tomorrow, friday, 4/5, 2026-04-05")
|
|
return
|
|
|
|
date_str = date.strftime("%Y-%m-%d")
|
|
day_name = date.strftime("%A")
|
|
ext, name, source = schedule.resolve_shift(date_str, day_name)
|
|
|
|
if ext != employee["extension"]:
|
|
respond(text=f"That's not your shift — it belongs to {name} (Ext {ext}).")
|
|
return
|
|
|
|
schedule.mark_open(date_str)
|
|
|
|
if is_today(date_str):
|
|
invoke_3cx_scheduler(FALLBACK_EXTENSION)
|
|
|
|
blocks = build_shift_change_message(user_id, date_str, "dropped", ext, name)
|
|
client.chat_postMessage(channel=channel_id, blocks=blocks, text=f"Shift dropped for {date_str}")
|
|
respond(text=f"You dropped the shift for *{date.strftime('%A, %b %-d')}*. It's now open for pickup.")
|
|
|
|
def _handle_swap(respond, schedule, user_id, text, channel_id, client):
|
|
# Expected format: swap <date> @user OR swap <date> <extension>
|
|
parts = text.split(maxsplit=2)
|
|
if len(parts) < 3:
|
|
respond(text="Usage: `/oncall swap <date> @person` (e.g. `/oncall swap friday @sarah`)")
|
|
return
|
|
|
|
employee = schedule.get_employee_by_slack_id(user_id)
|
|
if not employee:
|
|
respond(text="You're not registered. Use `/oncall register <extension>` first.")
|
|
return
|
|
|
|
date = parse_date(parts[1])
|
|
if not date:
|
|
respond(text=f"Couldn't parse date: `{parts[1]}`.")
|
|
return
|
|
|
|
date_str = date.strftime("%Y-%m-%d")
|
|
day_name = date.strftime("%A")
|
|
ext, name, source = schedule.resolve_shift(date_str, day_name)
|
|
|
|
if ext != employee["extension"]:
|
|
respond(text=f"That's not your shift — it belongs to {name} (Ext {ext}). You can only swap your own shifts.")
|
|
return
|
|
|
|
# Resolve target user — could be <@U12345> or an extension number
|
|
target_text = parts[2].strip()
|
|
slack_id_match = re.match(r"<@(\w+)(?:\|[^>]*)?>", target_text)
|
|
if slack_id_match:
|
|
target_slack_id = slack_id_match.group(1)
|
|
target = schedule.get_employee_by_slack_id(target_slack_id)
|
|
if not target:
|
|
respond(text=f"<@{target_slack_id}> isn't registered. They need to run `/oncall register <extension>`.")
|
|
return
|
|
else:
|
|
target = schedule.get_employee_by_extension(target_text)
|
|
if not target:
|
|
respond(text=f"Extension `{target_text}` not found in the roster.")
|
|
return
|
|
|
|
schedule.set_override(date_str, target["extension"], target["name"])
|
|
|
|
if is_today(date_str):
|
|
invoke_3cx_scheduler(target["extension"])
|
|
|
|
blocks = build_shift_change_message(
|
|
target.get("slack_user_id", user_id),
|
|
date_str,
|
|
"swapped",
|
|
target["extension"],
|
|
target["name"],
|
|
)
|
|
client.chat_postMessage(channel=channel_id, blocks=blocks, text=f"Shift swapped for {date_str}")
|
|
respond(text=f"Swapped *{date.strftime('%A, %b %-d')}* to {target['name']} (Ext {target['extension']}).")
|
|
|
|
return app
|