ring-scheduler-3cx/src/handler.py

import json
import logging
import os
from datetime import datetime, timezone, timedelta

import boto3
import yaml

from three_cx_client import ThreeCXClient

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# US Eastern (handles DST automatically via zoneinfo on Python 3.9+)
try:
    from zoneinfo import ZoneInfo
    EASTERN = ZoneInfo("America/New_York")
except ImportError:
    EASTERN = timezone(timedelta(hours=-5))


def load_schedule() -> dict:
    """Load schedule from config bundled in the Lambda package."""
    config_path = os.path.join(os.path.dirname(__file__), "..", "config", "schedule.yaml")
    with open(config_path) as f:
        return yaml.safe_load(f)


def get_3cx_credentials() -> dict:
    """Fetch 3CX credentials from SSM Parameter Store."""
    ssm = boto3.client("ssm")
    prefix = os.environ.get("SSM_PREFIX", "/3cx-scheduler")
    params = ssm.get_parameters_by_path(
        Path=prefix,
        WithDecryption=True,
    )
    creds = {}
    for p in params["Parameters"]:
        key = p["Name"].split("/")[-1]
        creds[key] = p["Value"]
    return creds


def resolve_extension(schedule: dict, now: datetime) -> str:
    """Determine which extension should handle after-hours calls today."""
    date_str = now.strftime("%Y-%m-%d")
    exceptions = schedule.get("exceptions") or {}
    if date_str in exceptions:
        ext = str(exceptions[date_str])
        logger.info("Using exception override for %s: Ext %s", date_str, ext)
        return ext

    day_name = now.strftime("%A")
    weekly = schedule["weekly_schedule"]
    if day_name not in weekly:
        raise ValueError(f"No schedule entry for {day_name}")
    ext = str(weekly[day_name])
    logger.info("Using weekly schedule for %s (%s): Ext %s", date_str, day_name, ext)
    return ext


def handler(event, context):
    """Lambda entry point. Triggered by two EventBridge rules (EST + EDT).
    Only one fires at the correct 8am ET depending on DST. The other fires
    at 7am or 9am ET — we skip that invocation."""
    now = datetime.now(EASTERN)
    current_hour = now.hour
    logger.info("Running 3CX ring group scheduler at %s", now.isoformat())

    if current_hour != 8:
        logger.info("Current ET hour is %d, not 8am — skipping (wrong DST rule fired)", current_hour)
        return {"skipped": True, "reason": f"ET hour is {current_hour}, not 8"}

    schedule = load_schedule()
    extension = resolve_extension(schedule, now)

    creds = get_3cx_credentials()
    client = ThreeCXClient(
        domain=creds["domain"],
        auth_mode="oauth",
        client_id=creds["client_id"],
        client_secret=creds["client_secret"],
    )

    ring_group = client.get_ring_group(schedule["ring_group_number"])
    ring_group_id = ring_group["Id"]

    client.update_ring_group_forwarding(
        ring_group_id=ring_group_id,
        closed_destination=extension,
        holiday_destination=extension,
    )

    result = {
        "date": now.strftime("%Y-%m-%d"),
        "day": now.strftime("%A"),
        "extension": extension,
        "ring_group": schedule["ring_group_number"],
    }
    logger.info("Result: %s", json.dumps(result))
    return result