"""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 decimal import Decimal 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": Decimal(str(rate)), }) if name not in totals: totals[name] = {"shifts": 0, "total": Decimal("0"), "extension": ext, "rate": Decimal(str(rate))} totals[name]["shifts"] += 1 totals[name]["total"] += Decimal(str(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.""" totals_rows = "" for name, info in sorted(pay_record["totals"].items()): totals_rows += ( f"{name}" f"${info.get('rate', 0):.2f}${info['total']:.2f}\n" ) return f"""

Bonus Pay Summary — {week_label}

{totals_rows}
NameRateTotal

This is an automated report from Sea Haven Industries.

""" 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"Bonus 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) pay_dm_user = os.environ.get("PAY_REPORT_USER") 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"] ) # DM pay report to designated user instead of channel if pay_dm_user: slack.chat_postMessage( channel=pay_dm_user, blocks=pay_blocks, text=f"After-Hours Pay — {week_label}", ) logger.info("Sent pay summary DM to %s for week of %s", pay_dm_user, week_key) else: logger.warning("PAY_REPORT_USER not set — skipping Slack pay summary") # 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"])}