afterhours-shift-manager/src/weekly_post.py

188 lines
6.4 KiB
Python
Raw Normal View History

"""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"<tr><td>{name}</td>"
f"<td>${info.get('rate', 0):.2f}</td><td><strong>${info['total']:.2f}</strong></td></tr>\n"
)
return f"""<html>
<body style="font-family: Arial, sans-serif; color: #333;">
<h2>Bonus Pay Summary &mdash; {week_label}</h2>
<table border="1" cellpadding="6" cellspacing="0" style="border-collapse: collapse;">
<tr style="background: #f0f0f0;"><th>Name</th><th>Rate</th><th>Total</th></tr>
{totals_rows}</table>
<p style="color: #888; font-size: 12px;">This is an automated report from Sea Haven Industries.</p>
</body>
</html>"""
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 (always starts on Monday of this week) ---
this_monday = now - timedelta(days=now.weekday())
blocks = build_week_schedule(schedule, start_date=this_monday)
end_date = this_monday + 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"])}