"""Lambda handler — posts the weekly on-call schedule and previous week's pay summary
to Slack every Monday at 7am ET, and emails the pay summary to payroll."""
import logging
import os
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
import boto3
from slack_sdk import WebClient
from src.blocks import build_pay_summary_blocks, build_week_schedule
from src.schedule import FALLBACK_EXTENSION, ShiftSchedule
logger = logging.getLogger()
logger.setLevel(logging.INFO)
EASTERN = ZoneInfo("America/New_York")
DAY_ORDER = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
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")
ext, name, source = schedule.resolve_shift(date_str, day_name, shift_type)
if source == "available" or ext == FALLBACK_EXTENSION:
return
rate = schedule.get_shift_rate(ext)
label_suffix = " (Day)" if shift_type == "day" else ""
breakdown.append({
"date": date_str,
"date_label": date.strftime("%b %-d"),
"day": f"{day_name[:3]}{label_suffix}",
"name": name,
"extension": ext,
"rate": rate,
})
if name not in totals:
totals[name] = {"shifts": 0, "total": 0.0, "extension": ext, "rate": rate}
totals[name]["shifts"] += 1
totals[name]["total"] += rate
def _calculate_weekly_pay(schedule: ShiftSchedule, week_start: datetime) -> dict:
"""Calculate pay for a Mon–Sun week. Returns pay record dict."""
default_rate = schedule.get_shift_rate()
breakdown = []
totals = {}
for i in range(7):
date = week_start + timedelta(days=i)
day_name = DAY_ORDER[i]
# Weekend day shifts (8am–5pm)
if day_name in WEEKEND_DAYS:
_add_shift_to_pay(schedule, breakdown, totals, date, day_name, "day")
# After-hours (night) shift
_add_shift_to_pay(schedule, breakdown, totals, date, day_name, "night")
return {
"week_start": week_start.strftime("%Y-%m-%d"),
"default_rate": str(default_rate),
"breakdown": breakdown,
"totals": totals,
}
def _build_pay_email_html(week_label: str, pay_record: dict) -> str:
"""Build an HTML email body for the weekly pay summary."""
rows = ""
for entry in pay_record["breakdown"]:
rows += (
f"| {entry['day']} | {entry['date_label']} | "
f"{entry['name']} | Ext {entry['extension']} | "
f"${entry['rate']:.2f} |
\n"
)
totals_rows = ""
for name, info in sorted(pay_record["totals"].items()):
totals_rows += (
f"| {name} | {info['shifts']} | "
f"${info.get('rate', 0):.2f} | ${info['total']:.2f} |
\n"
)
return f"""
After-Hours Pay Summary — {week_label}
Shift Breakdown
| Day | Date | Name | Extension | Rate |
|---|
{rows}
Totals by Person
| Name | Shifts | Rate | Total |
|---|
{totals_rows}
This is an automated report from the After-Hours Shift Manager.
"""
def _send_pay_email(week_label: str, pay_record: dict) -> None:
"""Send the weekly pay summary email via SES."""
sender = os.environ.get("SES_SENDER", "noreply@seahaven.com")
recipients = os.environ.get("PAYROLL_RECIPIENTS", "").split(",")
recipients = [r.strip() for r in recipients if r.strip()]
if not recipients:
logger.warning("No PAYROLL_RECIPIENTS configured — skipping email")
return
ses = boto3.client("ses")
html_body = _build_pay_email_html(week_label, pay_record)
ses.send_email(
Source=sender,
Destination={"ToAddresses": recipients},
Message={
"Subject": {"Data": f"After-Hours Pay Summary — {week_label}"},
"Body": {"Html": {"Data": html_body}},
},
)
logger.info("Sent pay email to %s", recipients)
def handler(event, context):
now = datetime.now(EASTERN)
# DST guard — same pattern as the 3CX scheduler
force = event.get("force", False)
if not force and now.hour != 7:
logger.info("ET hour is %d, not 7am — skipping (wrong DST rule fired)", now.hour)
return {"skipped": True}
ssm = boto3.client("ssm")
bot_token = ssm.get_parameter(
Name=os.environ["SLACK_BOT_TOKEN_PARAM"], WithDecryption=True
)["Parameter"]["Value"]
channel_id = ssm.get_parameter(
Name=os.environ["SHIFT_CHANNEL_PARAM"], WithDecryption=True
)["Parameter"]["Value"]
schedule = ShiftSchedule()
slack = WebClient(token=bot_token)
# --- Previous week's pay summary ---
prev_monday = now - timedelta(days=7)
prev_monday = prev_monday.replace(hour=0, minute=0, second=0, microsecond=0)
pay_record = _calculate_weekly_pay(schedule, prev_monday)
if pay_record["breakdown"]:
week_key = prev_monday.strftime("%Y-%m-%d")
schedule.save_pay_record(week_key, pay_record)
prev_sunday = prev_monday + timedelta(days=6)
week_label = f"{prev_monday.strftime('%b %-d')} to {prev_sunday.strftime('%b %-d')}"
pay_blocks = build_pay_summary_blocks(
week_label, pay_record["breakdown"], pay_record["totals"]
)
slack.chat_postMessage(
channel=channel_id,
blocks=pay_blocks,
text=f"After-Hours Pay — {week_label}",
)
logger.info("Posted pay summary for week of %s", week_key)
# Email pay summary to payroll
_send_pay_email(week_label, pay_record)
# --- Two-week schedule ---
blocks = build_week_schedule(schedule, start_date=now)
end_date = now + timedelta(days=13)
slack.chat_postMessage(
channel=channel_id,
blocks=blocks,
text=f"After-Hours Schedule — {now.strftime('%b %-d')} to {end_date.strftime('%b %-d')}",
)
logger.info("Posted weekly schedule to channel %s", channel_id)
return {"posted": True, "channel": channel_id, "pay_calculated": bool(pay_record["breakdown"])}