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
This commit is contained in:
Adam Moussa 2026-05-12 15:50:24 -04:00
parent e9b55d357b
commit debac9f8a7
3 changed files with 162 additions and 2 deletions

View file

@ -144,7 +144,7 @@ def build_shift_change_message(
return [{"type": "section", "text": {"type": "mrkdwn", "text": text}}]
def build_help_blocks() -> list[dict]:
def build_help_blocks(is_admin: bool = False) -> list[dict]:
"""Build help message blocks."""
text = (
"*After-Hours Shift Manager*\n\n"
@ -162,6 +162,16 @@ def build_help_blocks() -> list[dict]:
"`/oncall help` — Show this help message\n\n"
"_Dates can be: today, tomorrow, monday, friday, 4/5, 2026-04-05_"
)
if is_admin:
text += (
"\n\n*Admin Commands*\n"
"`/oncall admin override <date> <ext>` — Assign a shift to an extension\n"
"`/oncall admin open <date>` — Mark a shift as open\n"
"`/oncall admin clear <date>` — Remove override (revert to weekly)\n"
"`/oncall admin roster add <ext> <name>` — Add a roster entry\n"
"`/oncall admin roster remove <ext>` — Remove a roster entry\n"
"`/oncall admin roster rename <ext> <name>` — Rename an employee"
)
return [{"type": "section", "text": {"type": "mrkdwn", "text": text}}]

View file

@ -179,6 +179,32 @@ class ShiftSchedule:
resp = self.table.get_item(Key={"PK": "CONFIG", "SK": "CONFIG"})
return resp.get("Item", {})
def get_admin_users(self) -> list[str]:
config = self.get_config()
return config.get("admin_users", [])
def add_roster_entry(self, extension: str, name: str) -> None:
self.table.put_item(
Item={
"PK": "ROSTER",
"SK": extension,
"name": name,
"extension": extension,
"slack_user_id": "",
}
)
def remove_roster_entry(self, extension: str) -> None:
self.table.delete_item(Key={"PK": "ROSTER", "SK": extension})
def rename_roster_entry(self, extension: str, new_name: str) -> None:
self.table.update_item(
Key={"PK": "ROSTER", "SK": extension},
UpdateExpression="SET #n = :name",
ExpressionAttributeNames={"#n": "name"},
ExpressionAttributeValues={":name": new_name},
)
def set_default_shift_rate(self, rate: float) -> None:
"""Set the global default shift rate in config."""
self.table.update_item(

View file

@ -107,12 +107,14 @@ def create_app(
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())
respond(blocks=build_help_blocks(is_admin=is_admin))
elif text == "roster":
respond(blocks=build_roster_blocks(schedule.get_roster()))
elif text == "pay":
@ -133,6 +135,8 @@ def create_app(
_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`")
@ -511,4 +515,124 @@ def create_app(
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