mirror of
https://github.com/Sea-Haven-Industries/afterhours-shift-manager.git
synced 2026-09-30 06:43:12 +00:00
Compare commits
3 commits
702f1fd9a2
...
cbf36b30ff
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
cbf36b30ff | ||
|
|
c3b35ed62e | ||
|
|
0d119f494b |
5 changed files with 56 additions and 90 deletions
|
|
@ -3,9 +3,9 @@
|
|||
from datetime import datetime, timedelta
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
EASTERN = ZoneInfo("America/New_York")
|
||||
from shared.schedule import WEEKEND_DAYS
|
||||
|
||||
WEEKEND_DAYS = {"Saturday", "Sunday"}
|
||||
EASTERN = ZoneInfo("America/New_York")
|
||||
|
||||
|
||||
SHIFT_LABELS = {
|
||||
|
|
@ -23,7 +23,7 @@ def _format_shift_line(
|
|||
shift_type: str = "night",
|
||||
) -> str:
|
||||
day_label = date.strftime("%a %b %-d")
|
||||
if shift_type in SHIFT_LABELS:
|
||||
if date.strftime("%A") in WEEKEND_DAYS and shift_type in SHIFT_LABELS:
|
||||
day_label += f" {SHIFT_LABELS[shift_type]}"
|
||||
if is_today:
|
||||
day_label = f"*{day_label} (today)*"
|
||||
|
|
@ -97,14 +97,15 @@ def build_week_schedule(schedule, start_date: datetime | None = None) -> list[di
|
|||
elements = []
|
||||
for date_str, shift_type in open_shifts:
|
||||
dt = datetime.strptime(date_str, "%Y-%m-%d")
|
||||
label = "Day" if shift_type == "day" else "Night"
|
||||
day_name = dt.strftime("%A")
|
||||
label = f" {SHIFT_LABELS[shift_type]}" if day_name in WEEKEND_DAYS else ""
|
||||
action_suffix = "_day" if shift_type == "day" else ""
|
||||
elements.append(
|
||||
{
|
||||
"type": "button",
|
||||
"text": {
|
||||
"type": "plain_text",
|
||||
"text": f"Pick up {dt.strftime('%a %b %-d')} {label}",
|
||||
"text": f"Pick up {dt.strftime('%a %b %-d')}{label}",
|
||||
},
|
||||
"action_id": f"pickup_{date_str}{action_suffix}",
|
||||
"style": "primary",
|
||||
|
|
|
|||
|
|
@ -24,8 +24,8 @@ def update_queue_routing(
|
|||
queue = client.get_queue(queue_number)
|
||||
client.update_queue_forwarding(
|
||||
queue_id=queue["Id"],
|
||||
closed_destination=extension,
|
||||
holiday_destination=extension,
|
||||
closed=extension,
|
||||
holiday=extension,
|
||||
)
|
||||
logger.info("Updated queue %s to forward to Ext %s", queue_number, extension)
|
||||
return {"extension": extension, "queue": queue_number}
|
||||
|
|
|
|||
|
|
@ -84,6 +84,33 @@ class ThreeCXClient:
|
|||
resp.raise_for_status()
|
||||
return resp.json().get("value", [])
|
||||
|
||||
def _update_forwarding(
|
||||
self, resource: str, resource_id: int, closed: str, holiday: str
|
||||
):
|
||||
payload = {
|
||||
"OutOfOfficeRoute": {
|
||||
"IsPromptEnabled": False,
|
||||
"Route": {"To": "Extension", "Number": closed, "External": ""},
|
||||
},
|
||||
"HolidaysRoute": {
|
||||
"IsPromptEnabled": False,
|
||||
"Route": {"To": "Extension", "Number": holiday, "External": ""},
|
||||
},
|
||||
}
|
||||
resp = self.session.patch(
|
||||
f"{self.base_url}/xapi/v1/{resource}({resource_id})",
|
||||
json=payload,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
logger.info(
|
||||
"Updated %s %s: closed->Ext %s, holiday->Ext %s",
|
||||
resource,
|
||||
resource_id,
|
||||
closed,
|
||||
holiday,
|
||||
)
|
||||
return resp.status_code
|
||||
|
||||
def get_ring_group(self, extension_number: str) -> dict:
|
||||
"""Fetch ring group config by extension number."""
|
||||
resp = self.session.get(
|
||||
|
|
@ -93,42 +120,9 @@ class ThreeCXClient:
|
|||
return resp.json()
|
||||
|
||||
def update_ring_group_forwarding(
|
||||
self,
|
||||
ring_group_id: int,
|
||||
closed_destination: str,
|
||||
holiday_destination: str,
|
||||
self, ring_group_id: int, closed: str, holiday: str
|
||||
):
|
||||
"""Update the OutOfOfficeRoute and HolidaysRoute on a ring group."""
|
||||
payload = {
|
||||
"OutOfOfficeRoute": {
|
||||
"IsPromptEnabled": False,
|
||||
"Route": {
|
||||
"To": "Extension",
|
||||
"Number": closed_destination,
|
||||
"External": "",
|
||||
},
|
||||
},
|
||||
"HolidaysRoute": {
|
||||
"IsPromptEnabled": False,
|
||||
"Route": {
|
||||
"To": "Extension",
|
||||
"Number": holiday_destination,
|
||||
"External": "",
|
||||
},
|
||||
},
|
||||
}
|
||||
resp = self.session.patch(
|
||||
f"{self.base_url}/xapi/v1/RingGroups({ring_group_id})",
|
||||
json=payload,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
logger.info(
|
||||
"Updated ring group %s: closed->Ext %s, holiday->Ext %s",
|
||||
ring_group_id,
|
||||
closed_destination,
|
||||
holiday_destination,
|
||||
)
|
||||
return resp.status_code
|
||||
return self._update_forwarding("RingGroups", ring_group_id, closed, holiday)
|
||||
|
||||
def get_queue(self, extension_number: str) -> dict:
|
||||
"""Fetch queue config by extension number."""
|
||||
|
|
@ -138,40 +132,5 @@ class ThreeCXClient:
|
|||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
def update_queue_forwarding(
|
||||
self,
|
||||
queue_id: int,
|
||||
closed_destination: str,
|
||||
holiday_destination: str,
|
||||
):
|
||||
"""Update the OutOfOfficeRoute and HolidaysRoute on a queue."""
|
||||
payload = {
|
||||
"OutOfOfficeRoute": {
|
||||
"IsPromptEnabled": False,
|
||||
"Route": {
|
||||
"To": "Extension",
|
||||
"Number": closed_destination,
|
||||
"External": "",
|
||||
},
|
||||
},
|
||||
"HolidaysRoute": {
|
||||
"IsPromptEnabled": False,
|
||||
"Route": {
|
||||
"To": "Extension",
|
||||
"Number": holiday_destination,
|
||||
"External": "",
|
||||
},
|
||||
},
|
||||
}
|
||||
resp = self.session.patch(
|
||||
f"{self.base_url}/xapi/v1/Queues({queue_id})",
|
||||
json=payload,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
logger.info(
|
||||
"Updated queue %s: closed->Ext %s, holiday->Ext %s",
|
||||
queue_id,
|
||||
closed_destination,
|
||||
holiday_destination,
|
||||
)
|
||||
return resp.status_code
|
||||
def update_queue_forwarding(self, queue_id: int, closed: str, holiday: str):
|
||||
return self._update_forwarding("Queues", queue_id, closed, holiday)
|
||||
|
|
|
|||
|
|
@ -377,9 +377,11 @@ def create_app(
|
|||
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)
|
||||
parts = text.split()
|
||||
if len(parts) < 2:
|
||||
respond(text="Usage: `/oncall pick <date>` (e.g. `/oncall pick friday`)")
|
||||
respond(
|
||||
text="Usage: `/oncall pick <date> [day|night]` (e.g. `/oncall pick friday`)"
|
||||
)
|
||||
return
|
||||
|
||||
employee = schedule.get_employee_by_slack_id(user_id)
|
||||
|
|
@ -389,10 +391,14 @@ def create_app(
|
|||
)
|
||||
return
|
||||
|
||||
date = parse_date(parts[1])
|
||||
explicit_shift = (
|
||||
parts[2] if len(parts) > 2 and parts[2] in ("day", "night") else None
|
||||
)
|
||||
date_text = parts[1]
|
||||
date = parse_date(date_text)
|
||||
if not date:
|
||||
respond(
|
||||
text=f"Couldn't parse date: `{parts[1]}`. Try: today, tomorrow, friday, 4/5, 2026-04-05"
|
||||
text=f"Couldn't parse date: `{date_text}`. Try: today, tomorrow, friday, 4/5, 2026-04-05"
|
||||
)
|
||||
return
|
||||
|
||||
|
|
@ -403,9 +409,11 @@ def create_app(
|
|||
|
||||
day_name = date.strftime("%A")
|
||||
|
||||
# On weekends, find the first available shift (day then night)
|
||||
shift_type = "night"
|
||||
if day_name in WEEKEND_DAYS:
|
||||
if explicit_shift:
|
||||
shift_type = explicit_shift
|
||||
ext, name, source = schedule.resolve_shift(date_str, day_name, shift_type)
|
||||
elif day_name in WEEKEND_DAYS:
|
||||
shift_type = "night"
|
||||
for st in ("day", "night"):
|
||||
ext, name, source = schedule.resolve_shift(date_str, day_name, st)
|
||||
if source == "available":
|
||||
|
|
@ -414,6 +422,7 @@ def create_app(
|
|||
else:
|
||||
ext, name, source = schedule.resolve_shift(date_str, day_name)
|
||||
else:
|
||||
shift_type = "night"
|
||||
ext, name, source = schedule.resolve_shift(date_str, day_name)
|
||||
|
||||
# Already assigned to someone else (not open)
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ import boto3
|
|||
from slack_sdk import WebClient
|
||||
|
||||
from shared.blocks import build_pay_summary_blocks, build_week_schedule
|
||||
from shared.schedule import FALLBACK_EXTENSION, ShiftSchedule
|
||||
from shared.schedule import FALLBACK_EXTENSION, WEEKEND_DAYS, ShiftSchedule
|
||||
from shared.secrets import get_secret
|
||||
|
||||
logger = logging.getLogger()
|
||||
|
|
@ -30,9 +30,6 @@ DAY_ORDER = [
|
|||
]
|
||||
|
||||
|
||||
WEEKEND_DAYS = {"Saturday", "Sunday"}
|
||||
|
||||
|
||||
def _add_shift_to_pay(schedule, breakdown, totals, date, day_name, shift_type="night"):
|
||||
"""Resolve a shift and add it to the pay breakdown if assigned."""
|
||||
date_str = date.strftime("%Y-%m-%d")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue