"""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 Khalifa University Library hours:
"
"⚠️ Live hours are currently unavailable. Please check the official schedule: "
f'{LIBRARY_HOURS_URL}
'
"📍 Contacts: "
'Main Campus Library · E Building · +971 2 312 4604 '
'Habshan Library · SAN Campus · +971 2 312 3160'
)
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"See live hours for today"
hours_str = _format_libcal_day(today_entry)
is_closed = hours_str.lower().startswith("closed")
colour = "#dc2626" if is_closed else "#16a34a"
return (
f"📅 Today ({today_name}): "
f"{hours_str}"
)
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"
"
f"
{day_label}
"
f"
{hours_str}
"
f"
"
)
if not rows:
return ""
return (
"
"
+ rows + "
"
)
answer = "Here are the Khalifa University Library live hours:
"
for key, label in location_labels.items():
loc = hours_data.get(key)
if not loc:
continue
answer += f"📍 {label} "
answer += _today_line(label, loc) + " "
week_tbl = _week_table(loc)
if week_tbl:
answer += week_tbl + "
"
else:
answer += " "
answer += (
"⚠️ Hours may change during Ramadan, public holidays, and exam periods. "
f'Always confirm at: {LIBRARY_HOURS_URL}
'
''
'Main Campus Library · E Building · +971 2 312 4604 '
'Habshan Library · Building 5, SAN Campus · +971 2 312 3160'
''
)
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 (
"📅 Khalifa University Library Events
"
"Live event data is currently unavailable. Browse all upcoming events here: "
f'{EVENTS_URL}'
)
if not events:
return (
"📅 Khalifa University Library Events
"
f"No upcoming events found in the next {days} days.
"
f'Browse the full calendar: {EVENTS_URL}'
)
answer = f"📅 Upcoming Library Events, Workshops & Training (next {days} days)
"
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'