| """LibCal client + formatting: live opening hours and events via the Cloudflare Worker proxy.""" |
| import asyncio |
| import html |
| import json |
| import logging |
| import os |
| import re |
| import time |
| import uuid |
| from datetime import datetime |
| from typing import Any, Dict, List, Literal, Optional, Tuple |
| from urllib.parse import quote |
|
|
| import httpx |
| from pydantic import BaseModel, ConfigDict, Field |
|
|
| from src.config import get_settings, LIBBEE_VERSION |
|
|
| from src.agentcore.constants import LIBRARY_HOURS_URL |
| from src.agentcore.utils import _escape |
|
|
| logger = logging.getLogger(__name__) |
|
|
| _libcal_hours_cache: Dict[str, Any] = {} |
|
|
|
|
| _LIBCAL_HOURS_TTL = 60 * 60 |
|
|
|
|
| async def _libcal_fetch_hours() -> Optional[dict]: |
| now = time.time() |
| cached_at = _libcal_hours_cache.get("cached_at", 0) |
| if _libcal_hours_cache.get("data") and now - cached_at < _LIBCAL_HOURS_TTL: |
| return _libcal_hours_cache["data"] |
| settings = get_settings() |
| worker_url = getattr(settings, "cloudflare_worker_url", "").strip() |
| if not worker_url: |
| logger.warning("LibCal: CLOUDFLARE_WORKER_URL not set — cannot fetch hours") |
| return None |
| from datetime import timezone, timedelta |
| uae_now = datetime.now(timezone(timedelta(hours=4))) |
| date_from = uae_now.strftime("%Y-%m-%d") |
| date_to = (uae_now + timedelta(days=6)).strftime("%Y-%m-%d") |
| try: |
| async with httpx.AsyncClient(timeout=10) as client: |
| r = await client.get( |
| f"{worker_url.rstrip('/')}/hours", |
| params={"from": date_from, "to": date_to}, |
| ) |
| r.raise_for_status() |
| data = r.json() |
| if data.get("error"): |
| logger.warning("LibCal worker error: %s", data["error"]) |
| return None |
| _libcal_hours_cache["data"] = data |
| _libcal_hours_cache["cached_at"] = now |
| logger.info("LibCal: hours refreshed via worker") |
| return data |
| except Exception as exc: |
| logger.warning("LibCal hours fetch failed: %s", exc) |
| return None |
|
|
|
|
| def _parse_libcal_time(raw: str) -> str: |
| if not raw: |
| return raw |
| raw = raw.strip().replace("\u2013", "-").replace("–", "-") |
| raw = re.sub(r"\s+", "", raw) |
| raw = re.sub(r"(?i)(am|pm)$", lambda m: " " + m.group(1).upper(), raw) |
| return raw |
|
|
|
|
| def _format_libcal_day(day_data: dict) -> str: |
| if not day_data or not isinstance(day_data, dict): |
| return "See hours page" |
| status = str(day_data.get("status") or "").lower().strip() |
| note = str(day_data.get("note") or "").strip() |
| if status in ("closed", "0", "false"): |
| return f"Closed{f' — {note}' if note else ''}" |
| if status == "24hours": |
| return "Open 24 hours" |
| hours_list = day_data.get("hours") or [] |
| if hours_list and isinstance(hours_list, list): |
| first = hours_list[0] |
| f = first.get("from") or first.get("opens") or "" |
| t = first.get("to") or first.get("closes") or "" |
| if f and t: |
| return f"{_parse_libcal_time(f)} – {_parse_libcal_time(t)}" |
| opens = day_data.get("opens") or day_data.get("open") or "" |
| closes = day_data.get("closes") or day_data.get("close") or "" |
| if opens and closes: |
| return f"{_parse_libcal_time(opens)} – {_parse_libcal_time(closes)}" |
| times = day_data.get("times") or [] |
| if times and isinstance(times, list): |
| first = times[0] |
| f = first.get("from") or first.get("opens") or "" |
| t = first.get("to") or first.get("closes") or "" |
| if f and t: |
| return f"{_parse_libcal_time(f)} – {_parse_libcal_time(t)}" |
| if status in ("open",): |
| return "Open — see hours page for times" |
| return "See hours page" |
|
|
|
|
| async def _hours_answer() -> str: |
| from datetime import timezone, timedelta |
| uae_now = datetime.now(timezone(timedelta(hours=4))) |
| today_str = uae_now.strftime("%Y-%m-%d") |
| day_names = ["Monday","Tuesday","Wednesday","Thursday","Friday","Saturday","Sunday"] |
| today_name = day_names[uae_now.weekday()] |
| hours_data = await _libcal_fetch_hours() |
| if not hours_data: |
| return ( |
| "Here are the <strong>Khalifa University Library</strong> hours:<br><br>" |
| "⚠️ Live hours are currently unavailable. Please check the official schedule:<br>" |
| f'<a href="{LIBRARY_HOURS_URL}" target="_blank" ' |
| f'style="font-weight:700;color:#003366">{LIBRARY_HOURS_URL}</a><br><br>' |
| "<strong>📍 Contacts:</strong><br>" |
| 'Main Campus Library · E Building · <a href="tel:+97123124604">+971 2 312 4604</a><br>' |
| 'Habshan Library · SAN Campus · <a href="tel:+97123123160">+971 2 312 3160</a>' |
| ) |
| location_labels = { |
| "main_campus": "Main Campus Library", |
| "habshan": "Habshan Library (SAN Campus)", |
| } |
| def _today_line(loc_name: str, loc_data: dict) -> str: |
| dates = loc_data.get("dates", {}) |
| today_entry = dates.get(today_str, {}) |
| if not today_entry: |
| return f"<em>See <a href='{LIBRARY_HOURS_URL}' target='_blank'>live hours</a> for today</em>" |
| hours_str = _format_libcal_day(today_entry) |
| is_closed = hours_str.lower().startswith("closed") |
| colour = "#dc2626" if is_closed else "#16a34a" |
| return ( |
| f"📅 <strong>Today ({today_name}):</strong> " |
| f"<span style='color:{colour};font-weight:700'>{hours_str}</span>" |
| ) |
| def _week_table(loc_data: dict) -> str: |
| dates = loc_data.get("dates", {}) |
| if not dates: |
| return "" |
| rows = "" |
| for date_key in sorted(dates.keys())[:7]: |
| try: |
| from datetime import date as _date |
| d = _date.fromisoformat(date_key) |
| day_label = d.strftime("%a %d %b") |
| is_today = date_key == today_str |
| except ValueError: |
| day_label = date_key |
| is_today = False |
| hours_str = _format_libcal_day(dates[date_key]) |
| is_closed = hours_str.lower().startswith("closed") |
| colour = "#dc2626" if is_closed else "#16a34a" |
| weight = "700" if is_today else "400" |
| bg = "#fffbeb" if is_today else "transparent" |
| rows += ( |
| f"<tr style='background:{bg}'>" |
| f"<td style='padding:3px 12px 3px 0;color:#374151;font-weight:{weight}'>{day_label}</td>" |
| f"<td style='padding:3px 0;font-weight:600;color:{colour}'>{hours_str}</td>" |
| f"</tr>" |
| ) |
| if not rows: |
| return "" |
| return ( |
| "<table style='font-size:.84rem;border-collapse:collapse;width:100%;max-width:380px'>" |
| + rows + "</table>" |
| ) |
| answer = "Here are the <strong>Khalifa University Library</strong> live hours:<br><br>" |
| for key, label in location_labels.items(): |
| loc = hours_data.get(key) |
| if not loc: |
| continue |
| answer += f"<strong>📍 {label}</strong><br>" |
| answer += _today_line(label, loc) + "<br>" |
| week_tbl = _week_table(loc) |
| if week_tbl: |
| answer += week_tbl + "<br><br>" |
| else: |
| answer += "<br>" |
| answer += ( |
| "⚠️ <strong>Hours may change during Ramadan, public holidays, and exam periods.</strong><br>" |
| f'Always confirm at: <a href="{LIBRARY_HOURS_URL}" target="_blank" ' |
| f'style="font-weight:700;color:#003366">{LIBRARY_HOURS_URL}</a><br><br>' |
| '<span style="font-size:.82rem;color:#6b7280">' |
| 'Main Campus Library · E Building · <a href="tel:+97123124604">+971 2 312 4604</a><br>' |
| 'Habshan Library · Building 5, SAN Campus · <a href="tel:+97123123160">+971 2 312 3160</a>' |
| '</span>' |
| ) |
| return answer |
|
|
|
|
| _libcal_events_cache: Dict[str, Any] = {} |
|
|
|
|
| _LIBCAL_EVENTS_TTL = 30 * 60 |
|
|
|
|
| async def _libcal_fetch_events(days: int = 30, limit: int = 10) -> Optional[list]: |
| now = time.time() |
| cache_key = f"{days}_{limit}" |
| cached = _libcal_events_cache.get(cache_key, {}) |
| if cached.get("data") and now - cached.get("cached_at", 0) < _LIBCAL_EVENTS_TTL: |
| return cached["data"] |
| settings = get_settings() |
| worker_url = getattr(settings, "cloudflare_worker_url", "").strip() |
| if not worker_url: |
| logger.warning("LibCal events: CLOUDFLARE_WORKER_URL not set") |
| return None |
| try: |
| async with httpx.AsyncClient(timeout=10) as client: |
| r = await client.get( |
| f"{worker_url.rstrip('/')}/events", |
| params={"days": days, "limit": limit}, |
| ) |
| r.raise_for_status() |
| data = r.json() |
| if data.get("error"): |
| logger.warning("LibCal events worker error: %s", data["error"]) |
| return None |
| events_raw = data.get("events", []) |
| def _safe_str(val) -> str: |
| if val is None: |
| return "" |
| if isinstance(val, str): |
| return val |
| if isinstance(val, dict): |
| return str(val.get("name") or val.get("title") or "") |
| if isinstance(val, list): |
| return ", ".join(_safe_str(v) for v in val if v) |
| return str(val) |
| events = [] |
| for e in events_raw: |
| cats = e.get("category") or e.get("categories") or [] |
| if isinstance(cats, list): |
| cat_str = ", ".join(_safe_str(c) for c in cats) |
| else: |
| cat_str = _safe_str(cats) |
| events.append({ |
| "id": e.get("id", ""), |
| "title": _safe_str(e.get("title") or "Untitled Event"), |
| "start": _safe_str(e.get("start") or ""), |
| "end": _safe_str(e.get("end") or ""), |
| "allDay": bool(e.get("allday") or e.get("allDay")), |
| "location": _safe_str(e.get("location") or ""), |
| "description": _safe_str(e.get("description") or e.get("summary") or "")[:300], |
| "url": _safe_str(e.get("url") or ""), |
| "registration": bool(e.get("registration") or e.get("seats")), |
| "campus": _safe_str(e.get("campus") or ""), |
| "category": cat_str, |
| }) |
| _libcal_events_cache[cache_key] = {"data": events, "cached_at": now} |
| logger.info("LibCal: %d events fetched", len(events)) |
| return events |
| except Exception as exc: |
| logger.warning("LibCal events fetch failed: %s", exc) |
| return None |
|
|
|
|
| def _format_event_datetime(start: str, end: str, all_day: bool) -> str: |
| if not start: |
| return "" |
| try: |
| from datetime import datetime as _dt |
| s = _dt.fromisoformat(start.replace("Z", "+00:00")) |
| date_str = s.strftime("%a %d %b") |
| if all_day: |
| return date_str |
| time_str = s.strftime("%I:%M %p").lstrip("0") |
| if end: |
| e = _dt.fromisoformat(end.replace("Z", "+00:00")) |
| end_str = e.strftime("%I:%M %p").lstrip("0") |
| return f"{date_str} · {time_str} – {end_str}" |
| return f"{date_str} · {time_str}" |
| except Exception: |
| return start[:16].replace("T", " ") |
|
|
|
|
| async def _events_answer(days: int = 30) -> str: |
| EVENTS_URL = "https://kustar.libcal.com/calendar/events" |
| events = await _libcal_fetch_events(days=days, limit=15) |
| if not events: |
| return ( |
| "<strong>📅 Khalifa University Library Events</strong><br><br>" |
| "Live event data is currently unavailable. Browse all upcoming events here:<br>" |
| f'<a href="{EVENTS_URL}" target="_blank" ' |
| f'style="font-weight:700;color:#003366">{EVENTS_URL}</a>' |
| ) |
| if not events: |
| return ( |
| "<strong>📅 Khalifa University Library Events</strong><br><br>" |
| f"No upcoming events found in the next {days} days.<br><br>" |
| f'Browse the full calendar: <a href="{EVENTS_URL}" target="_blank"' |
| f' style="font-weight:700;color:#003366">{EVENTS_URL}</a>' |
| ) |
| answer = f"<strong>📅 Upcoming Library Events, Workshops & Training</strong> (next {days} days)<br><br>" |
| for ev in events[:10]: |
| title = _escape(str(ev.get("title") or "Untitled Event")) |
| when = _format_event_datetime( |
| str(ev.get("start") or ""), str(ev.get("end") or ""), bool(ev.get("allDay")) |
| ) |
| location = _escape(str(ev.get("location") or "")) |
| desc = _escape(str(ev.get("description") or "")) |
| url = str(ev.get("url") or EVENTS_URL) |
| reg = bool(ev.get("registration")) |
| answer += ( |
| f'<div style="margin-bottom:12px;padding:10px 12px;border-left:3px solid #003366;' |
| f'background:#f8fafc;border-radius:0 8px 8px 0">' |
| f'<div style="font-weight:700;font-size:.88rem;color:#003366">' |
| f'<a href="{url}" target="_blank" style="color:#003366;text-decoration:none">{title}</a></div>' |
| ) |
| if when: |
| answer += f'<div style="font-size:.80rem;color:#374151;margin-top:2px">🕐 {when}</div>' |
| if location: |
| answer += f'<div style="font-size:.80rem;color:#374151">📍 {location}</div>' |
| if desc: |
| answer += f'<div style="font-size:.78rem;color:#6b7280;margin-top:4px">{desc[:180]}{"…" if len(desc) > 180 else ""}</div>' |
| if reg: |
| answer += ( |
| f'<div style="margin-top:5px">' |
| f'<a href="{url}" target="_blank" ' |
| f'style="font-size:.76rem;font-weight:700;color:#003366;' |
| f'padding:3px 10px;border:1px solid #003366;border-radius:6px;text-decoration:none">' |
| f'Register →</a></div>' |
| ) |
| answer += '</div>' |
| answer += ( |
| f'<br><a href="{EVENTS_URL}" target="_blank" ' |
| f'style="font-weight:700;color:#003366">View full library events calendar →</a>' |
| ) |
| return answer |
|
|
|
|
|
|
