"""Lambda handler — repoints 3CX call flow for a holiday day-shift. A holiday is a day-only shift (08:00-17:00 ET), one HOLIDAY record per date. At 08:00 this Lambda is invoked with ``{"action": "activate", "date": ...}`` and at 17:00 with ``{"action": "deactivate", ...}``. Activate: * Capture both IVR 800 routes (key-0 and no-input/timeout) into ``CONFIG.captured_ivr_routes`` — but only if they are NOT already pointed at the holiday queue (so a re-run never overwrites the real originals). * Set queue 802's agents to the holiday's assignees, or ``[FALLBACK_EXTENSION]`` ("100") when no slots are filled. Membership is set ONCE here. * Repoint BOTH IVR 800 routes to the holiday queue (802). * Mark the holiday ``activated = True``. Idempotent: no-op if the record is gone or already activated. Deactivate: * Restore both IVR routes from ``CONFIG.captured_ivr_routes`` — only the routes that currently point at the queue are reverted (defensive against manual changes); clear the captured routes afterwards. * Clear queue 802's agents. * Mark the holiday ``activated = False``. Idempotent: no-op if the record is gone or not activated. """ import json import logging import os from shared.schedule import FALLBACK_EXTENSION, ShiftSchedule from shared.secrets import get_secret from shared.three_cx_client import ThreeCXClient logger = logging.getLogger() logger.setLevel(logging.INFO) def _make_client() -> ThreeCXClient: secret_prefix = os.environ["TCX_SECRET_PREFIX"] return ThreeCXClient( domain=get_secret(f"{secret_prefix}domain"), auth_mode="oauth", client_id=get_secret(f"{secret_prefix}client-id"), client_secret=get_secret(f"{secret_prefix}client-secret"), ) def _holiday_extensions(holiday: dict) -> list[str]: """Assignee extensions for the holiday, or the fallback when none claimed.""" assignees = holiday.get("assignees", {}) or {} extensions = list(assignees.keys()) return extensions or [FALLBACK_EXTENSION] def _activate(schedule: ShiftSchedule, date: str) -> dict: holiday = schedule.get_holiday(date) if holiday is None: logger.info("No holiday record for %s — nothing to activate", date) return {"action": "activate", "date": date, "skipped": "no_record"} if holiday.get("activated"): logger.info("Holiday %s already activated — no-op", date) return {"action": "activate", "date": date, "skipped": "already_active"} queue_number = schedule.get_holiday_queue() ivr_number = schedule.get_ivr_number() extensions = _holiday_extensions(holiday) client = _make_client() # Capture the live IVR routes BEFORE repointing — but guard against storing # holiday-state routes: if both routes already target the queue, a prior # activation is in effect, so keep whatever originals we already captured. ivr = client.get_ivr(ivr_number) ivr_id = ivr["Id"] current = client.extract_ivr_routes(ivr) already_queue = str(current.get("key0")) == str(queue_number) and str( current.get("timeout") ) == str(queue_number) if already_queue: logger.info( "IVR %s already points at queue %s — not re-capturing routes", ivr_number, queue_number, ) else: schedule.set_captured_ivr_routes(current) # Set queue membership ONCE, then repoint both IVR routes to the queue. queue = client.get_queue(queue_number) client.set_queue_agents(queue["Id"], extensions) client.set_ivr_routes(ivr_id, key0_dn=queue_number, timeout_dn=queue_number) schedule.set_holiday_activated(date, True) logger.info( "Activated holiday %s: queue %s agents=%s, IVR %s -> queue", date, queue_number, extensions, ivr_number, ) return { "action": "activate", "date": date, "queue": str(queue_number), "ivr": str(ivr_number), "agents": extensions, } def _deactivate(schedule: ShiftSchedule, date: str) -> dict: holiday = schedule.get_holiday(date) if holiday is None: logger.info("No holiday record for %s — nothing to deactivate", date) return {"action": "deactivate", "date": date, "skipped": "no_record"} if not holiday.get("activated"): logger.info("Holiday %s not activated — no-op", date) return {"action": "deactivate", "date": date, "skipped": "not_active"} queue_number = schedule.get_holiday_queue() ivr_number = schedule.get_ivr_number() captured = schedule.get_captured_ivr_routes() or {} client = _make_client() ivr = client.get_ivr(ivr_number) ivr_id = ivr["Id"] live = client.extract_ivr_routes(ivr) # Only restore a route if it currently points at the queue; otherwise leave # whatever destination it has now (it was changed outside this flow). def _restore(which: str) -> str | None: live_dn = live.get(which) if str(live_dn) == str(queue_number): # Restore the captured pre-holiday DN; if it was lost, keep the live # value rather than blanking the IVR destination. return captured.get(which) or live_dn return None # not pointing at the holiday queue — leave it untouched client.set_ivr_routes( ivr_id, key0_dn=_restore("key0"), timeout_dn=_restore("timeout"), ) queue = client.get_queue(queue_number) client.set_queue_agents(queue["Id"], []) schedule.set_captured_ivr_routes(None) schedule.set_holiday_activated(date, False) logger.info( "Deactivated holiday %s: restored IVR %s, cleared queue %s", date, ivr_number, queue_number, ) return { "action": "deactivate", "date": date, "queue": str(queue_number), "ivr": str(ivr_number), } def handler(event, context): action = event.get("action") date = event.get("date") logger.info("Holiday router invoked: action=%s date=%s", action, date) if not date: return {"error": True, "reason": "missing_date"} schedule = ShiftSchedule() try: if action == "activate": result = _activate(schedule, date) elif action == "deactivate": result = _deactivate(schedule, date) else: return {"error": True, "reason": "unknown_action", "action": action} except Exception: logger.exception("Holiday router failed for action=%s date=%s", action, date) return {"error": True, "action": action, "date": date} logger.info("Result: %s", json.dumps(result, default=str)) return result