afterhours-shift-manager/src/slack-bot/app.py
Adam Moussa debac9f8a7 Add admin slash commands for shift and roster management (#39)
- /oncall admin override <date> <ext> — assign a shift
- /oncall admin open <date> — mark shift as open
- /oncall admin clear <date> — remove override, revert to weekly
- /oncall admin roster add/remove/rename — manage roster entries
- Admin access gated by admin_users list in DynamoDB CONFIG
- Help message shows admin commands for admin users
2026-05-12 15:50:24 -04:00

638 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Slack Bolt app — /oncall command handlers and interactive actions."""
import logging
import os
import re
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from slack_bolt import App
from shared.blocks import (
build_help_blocks,
build_pay_summary_blocks,
build_roster_blocks,
build_shift_change_message,
build_week_schedule,
)
from shared.ring_scheduler import update_ring_group
from shared.schedule import FALLBACK_EXTENSION, ShiftSchedule
from shared.secrets import get_secret
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 _update_3cx_routing(extension: str) -> None:
"""Update the 3CX ring group to forward calls to the given extension."""
ring_group_number = os.environ.get("RING_GROUP_NUMBER")
secret_prefix = os.environ.get("TCX_SECRET_PREFIX")
if not ring_group_number or not secret_prefix:
logger.warning("3CX env vars not set — skipping ring group update")
return
try:
update_ring_group(
extension=extension,
ring_group_number=ring_group_number,
domain=get_secret(f"{secret_prefix}domain"),
client_id=get_secret(f"{secret_prefix}client-id"),
client_secret=get_secret(f"{secret_prefix}client-secret"),
)
except Exception:
logger.exception("Failed to update 3CX ring group")
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, schedule_channel: str | None = None
) -> 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"]
is_admin = user_id in schedule.get_admin_users()
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(is_admin=is_admin))
elif text == "roster":
respond(blocks=build_roster_blocks(schedule.get_roster()))
elif text == "pay":
_show_pay(respond, schedule)
elif text.startswith("rate"):
_handle_rate(respond, schedule, text)
elif text.startswith("register"):
_handle_register(respond, schedule, user_id, text)
elif text.startswith("pick"):
_handle_pick(
respond, schedule, user_id, text, schedule_channel or channel_id, client
)
elif text.startswith("drop"):
_handle_drop(
respond, schedule, user_id, text, schedule_channel or channel_id, client
)
elif text.startswith("swap"):
_handle_swap(
respond, schedule, user_id, text, schedule_channel or channel_id, client
)
elif text.startswith("admin"):
_handle_admin(respond, schedule, user_id, text, is_admin, 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, respond):
ack()
action_id = body["actions"][0]["action_id"]
remainder = action_id.replace("pickup_", "")
if remainder.endswith("_day"):
date_str = remainder[:-4]
shift_type = "day"
else:
date_str = remainder
shift_type = "night"
user_id = body["user"]["id"]
channel_id = body["channel"]["id"]
today_str = datetime.now(EASTERN).strftime("%Y-%m-%d")
if date_str < today_str:
client.chat_postEphemeral(
channel=channel_id,
user=user_id,
text="That shift has already passed and can't be picked up.",
)
return
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
claimed = schedule.claim_open_shift(
date_str, employee["extension"], employee["name"], shift_type
)
if not claimed:
client.chat_postEphemeral(
channel=channel_id,
user=user_id,
text=f"That shift on *{date_str}* was already picked up by someone else.",
)
return
if is_today(date_str) and shift_type == "night":
_update_3cx_routing(employee["extension"])
blocks = build_shift_change_message(
user_id, date_str, "picked_up", employee["extension"], employee["name"],
shift_type=shift_type,
)
respond(
response_type="in_channel",
replace_original=False,
blocks=blocks,
text=f"Shift picked up for {date_str}",
)
_refresh_schedule_post(client)
def _refresh_schedule_post(client):
"""Update the pinned schedule message in-place after a shift change."""
channel = schedule_channel
if not channel:
return
post = schedule.get_schedule_post(channel)
if not post or not post.get("message_ts"):
return
try:
blocks = build_week_schedule(schedule)
now = datetime.now(EASTERN)
this_monday = now - timedelta(days=now.weekday())
end_date = this_monday + timedelta(days=13)
client.chat_update(
channel=channel,
ts=post["message_ts"],
blocks=blocks,
text=f"After-Hours Schedule — {now.strftime('%b %-d')} to {end_date.strftime('%b %-d')}",
)
except Exception:
logger.warning("Could not update schedule post", exc_info=True)
# ── 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 _show_pay(respond, schedule):
now = datetime.now(EASTERN)
# Show last completed week's pay (previous Monday–Sunday)
this_monday = now - timedelta(days=now.weekday())
prev_monday = this_monday - timedelta(days=7)
week_key = prev_monday.strftime("%Y-%m-%d")
pay_record = schedule.get_pay_record(week_key)
if pay_record and pay_record.get("breakdown"):
prev_sunday = prev_monday + timedelta(days=6)
week_label = (
f"{prev_monday.strftime('%b %-d')} to {prev_sunday.strftime('%b %-d')}"
)
blocks = build_pay_summary_blocks(
week_label, pay_record["breakdown"], pay_record["totals"]
)
respond(blocks=blocks)
else:
respond(
text=f"No pay record found for the week of {prev_monday.strftime('%b %-d')}."
)
def _handle_rate(respond, schedule, text):
parts = text.split()
# /oncall rate — show current rates
if len(parts) == 1:
default_rate = schedule.get_shift_rate()
roster = schedule.get_roster()
lines = [f"*Default rate:* ${default_rate:.2f}/shift\n"]
custom = [
(e["SK"], e.get("name", "Unknown"), float(e["shift_rate"]))
for e in roster
if e.get("shift_rate")
]
if custom:
lines.append("*Per-person rates:*")
for ext, name, rate in sorted(custom, key=lambda x: x[0]):
lines.append(f"• {name} (Ext {ext}) — ${rate:.2f}/shift")
else:
lines.append("_No per-person rates set — everyone uses the default._")
respond(text="\n".join(lines))
return
# /oncall rate default <amount>
if parts[1] == "default":
if len(parts) < 3:
respond(
text="Usage: `/oncall rate default <amount>` (e.g. `/oncall rate default 50`)"
)
return
try:
amount = float(parts[2].replace("$", ""))
except ValueError:
respond(text=f"Invalid amount: `{parts[2]}`")
return
schedule.set_default_shift_rate(amount)
respond(text=f"Default shift rate set to *${amount:.2f}*.")
return
# /oncall rate <extension> <amount>
if len(parts) < 3:
respond(
text="Usage: `/oncall rate <extension> <amount>` (e.g. `/oncall rate 114 75`)"
)
return
ext = parts[1]
employee = schedule.get_employee_by_extension(ext)
if not employee:
respond(text=f"Extension `{ext}` not found in the roster.")
return
try:
amount = float(parts[2].replace("$", ""))
except ValueError:
respond(text=f"Invalid amount: `{parts[2]}`")
return
schedule.set_employee_shift_rate(ext, amount)
respond(
text=f"Shift rate for *{employee['name']}* (Ext {ext}) set to *${amount:.2f}*."
)
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")
if date_str < datetime.now(EASTERN).strftime("%Y-%m-%d"):
respond(text="You can't pick up a shift in the past.")
return
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):
_update_3cx_routing(employee["extension"])
respond(text=f"You picked up the shift for *{date.strftime('%A, %b %-d')}*.")
blocks = build_shift_change_message(
user_id, date_str, "picked_up", employee["extension"], employee["name"]
)
try:
client.chat_postMessage(
channel=channel_id,
blocks=blocks,
text=f"Shift picked up for {date_str}",
)
except Exception:
logger.exception("Failed to post pickup notification to channel")
_refresh_schedule_post(client)
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")
if date_str < datetime.now(EASTERN).strftime("%Y-%m-%d"):
respond(text="You can't drop a shift in the past.")
return
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):
_update_3cx_routing(FALLBACK_EXTENSION)
respond(
text=f"You dropped the shift for *{date.strftime('%A, %b %-d')}*. It's now open for pickup."
)
blocks = build_shift_change_message(user_id, date_str, "dropped", ext, name)
try:
client.chat_postMessage(
channel=channel_id, blocks=blocks, text=f"Shift dropped for {date_str}"
)
except Exception:
logger.exception("Failed to post drop notification to channel")
_refresh_schedule_post(client)
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):
_update_3cx_routing(target["extension"])
blocks = build_shift_change_message(
target.get("slack_user_id", user_id),
date_str,
"swapped",
target["extension"],
target["name"],
)
respond(
text=f"Swapped *{date.strftime('%A, %b %-d')}* to {target['name']} (Ext {target['extension']})."
)
try:
client.chat_postMessage(
channel=channel_id, blocks=blocks, text=f"Shift swapped for {date_str}"
)
except Exception:
logger.exception("Failed to post swap notification to channel")
_refresh_schedule_post(client)
def _handle_admin(respond, schedule, user_id, text, is_admin, client):
if not is_admin:
respond(text="Admin commands are restricted. Contact an administrator.")
return
parts = text.split()
if len(parts) < 2:
respond(
text=(
"*Admin Commands:*\n"
"`admin override <date> <ext>` — Assign shift\n"
"`admin open <date>` — Mark open\n"
"`admin clear <date>` — Remove override\n"
"`admin roster add <ext> <name>` — Add employee\n"
"`admin roster remove <ext>` — Remove employee\n"
"`admin roster rename <ext> <name>` — Rename"
)
)
return
subcmd = parts[1]
if subcmd == "override":
if len(parts) < 4:
respond(text="Usage: `/oncall admin override <date> <extension>`")
return
date = parse_date(parts[2])
if not date:
respond(text=f"Couldn't parse date: `{parts[2]}`")
return
ext = parts[3]
employee = schedule.get_employee_by_extension(ext)
if not employee:
respond(text=f"Extension `{ext}` not found in the roster.")
return
date_str = date.strftime("%Y-%m-%d")
schedule.set_override(date_str, employee["extension"], employee["name"])
respond(
text=f"Override set: *{date.strftime('%A, %b %-d')}* → {employee['name']} (Ext {ext})"
)
_refresh_schedule_post(client)
elif subcmd == "open":
if len(parts) < 3:
respond(text="Usage: `/oncall admin open <date>`")
return
date = parse_date(parts[2])
if not date:
respond(text=f"Couldn't parse date: `{parts[2]}`")
return
date_str = date.strftime("%Y-%m-%d")
schedule.mark_open(date_str)
respond(text=f"*{date.strftime('%A, %b %-d')}* marked as open.")
_refresh_schedule_post(client)
elif subcmd == "clear":
if len(parts) < 3:
respond(text="Usage: `/oncall admin clear <date>`")
return
date = parse_date(parts[2])
if not date:
respond(text=f"Couldn't parse date: `{parts[2]}`")
return
date_str = date.strftime("%Y-%m-%d")
schedule.remove_override(date_str)
respond(
text=f"Override cleared for *{date.strftime('%A, %b %-d')}* — reverted to weekly schedule."
)
_refresh_schedule_post(client)
elif subcmd == "roster":
if len(parts) < 3:
respond(text="Usage: `admin roster add|remove|rename <ext> [name]`")
return
roster_cmd = parts[2]
if roster_cmd == "add":
if len(parts) < 5:
respond(text="Usage: `/oncall admin roster add <ext> <name>`")
return
ext = parts[3]
name = " ".join(parts[4:])
schedule.add_roster_entry(ext, name)
respond(text=f"Added *{name}* (Ext {ext}) to the roster.")
elif roster_cmd == "remove":
if len(parts) < 4:
respond(text="Usage: `/oncall admin roster remove <ext>`")
return
ext = parts[3]
employee = schedule.get_employee_by_extension(ext)
if not employee:
respond(text=f"Extension `{ext}` not found in the roster.")
return
schedule.remove_roster_entry(ext)
respond(
text=f"Removed *{employee.get('name', ext)}* (Ext {ext}) from the roster."
)
elif roster_cmd == "rename":
if len(parts) < 5:
respond(text="Usage: `/oncall admin roster rename <ext> <name>`")
return
ext = parts[3]
employee = schedule.get_employee_by_extension(ext)
if not employee:
respond(text=f"Extension `{ext}` not found in the roster.")
return
new_name = " ".join(parts[4:])
schedule.rename_roster_entry(ext, new_name)
respond(
text=f"Renamed Ext {ext}: {employee.get('name', '?')} → *{new_name}*"
)
else:
respond(text="Unknown roster command. Use `add`, `remove`, or `rename`.")
else:
respond(text=f"Unknown admin command: `{subcmd}`. Try `/oncall help`.")
return app