mirror of
https://github.com/Sea-Haven-Industries/ring-scheduler-3cx.git
synced 2026-05-18 20:20:12 +00:00
cb5817ce95
- Fix three_cx_client import path for Lambda packaging - Expand SSM policy to include parent path and KMS decrypt for SecureString - Add force flag to bypass 8am schedule check for manual invocations - Add override_extension payload option for ad-hoc testing - Update .gitignore for output.json and .DS_Store Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
102 lines
3.2 KiB
Python
102 lines
3.2 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
import boto3
|
|
import yaml
|
|
|
|
from src.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())
|
|
|
|
force = event.get("force", False)
|
|
if not force and 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 = event.get("override_extension") or 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
|