"""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_pay_summary_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, 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"] 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 == "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) 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"] 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"] 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 ` 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": 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 _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 if parts[1] == "default": if len(parts) < 3: respond(text="Usage: `/oncall rate default ` (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 if len(parts) < 3: respond(text="Usage: `/oncall rate ` (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 ` (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 ` (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 ` 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 ` (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 ` 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 @user OR swap parts = text.split(maxsplit=2) if len(parts) < 3: respond(text="Usage: `/oncall swap @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 ` 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 `.") 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