from __future__ import annotations from datetime import datetime, timezone from typing import Any from uuid import uuid4 import requests import streamlit as st TIMEOUT_SECONDS = 20 def get_database_url() -> str: url = st.secrets.get("FIREBASE_DATABASE_URL", "").strip().rstrip("/") if not url: raise RuntimeError( "Missing FIREBASE_DATABASE_URL in .streamlit/secrets.toml or Hugging Face Space secrets." ) return url def get_auth_params() -> dict[str, str]: token = st.secrets.get("FIREBASE_AUTH_TOKEN", "").strip() return {"auth": token} if token else {} def build_url(path: str) -> str: clean_path = path.strip("/") return f"{get_database_url()}/{clean_path}.json" def validate_response(response: requests.Response) -> Any: response.raise_for_status() payload = response.json() if isinstance(payload, dict) and payload.get("error"): raise RuntimeError(str(payload["error"])) return payload def load_cards() -> list[dict[str, Any]]: response = requests.get( build_url("cards"), params=get_auth_params(), timeout=TIMEOUT_SECONDS, ) payload = validate_response(response) or {} if not isinstance(payload, dict): return [] cards = [] for card_id, card in payload.items(): if not isinstance(card, dict): continue cards.append( { "id": str(card.get("id", card_id)), "title": str(card.get("title", "")), "startHour": str(card.get("startHour", "08")).zfill(2), "startMinute": str(card.get("startMinute", "00")).zfill(2), "endHour": str(card.get("endHour", "10")).zfill(2), "endMinute": str(card.get("endMinute", "00")).zfill(2), "createdAt": str(card.get("createdAt", "")), "tabId": str(card.get("tabId", "")), } ) cards.sort(key=lambda item: item.get("createdAt", "")) return cards def sanitize_card(card: dict[str, Any], card_id: str | None = None) -> dict[str, str]: now = datetime.now(timezone.utc).isoformat() final_id = card_id or str(card.get("id") or card.get("clientKey") or uuid4()) return { "id": final_id, "title": str(card.get("title", "")).strip(), "startHour": str(card.get("startHour", "08")).zfill(2), "startMinute": str(card.get("startMinute", "00")).zfill(2), "endHour": str(card.get("endHour", "10")).zfill(2), "endMinute": str(card.get("endMinute", "00")).zfill(2), "createdAt": str(card.get("createdAt") or now), "tabId": str(card.get("tabId", "")).strip(), } def load_tabs() -> list[dict[str, str]]: response = requests.get( build_url("tabs"), params=get_auth_params(), timeout=TIMEOUT_SECONDS, ) payload = validate_response(response) or {} if not isinstance(payload, dict): return [] tabs = [] for tab_id, tab in payload.items(): if not isinstance(tab, dict): continue tabs.append( { "id": str(tab.get("id", tab_id)), "title": str(tab.get("title", "Untitled Tab")), "createdAt": str(tab.get("createdAt", "")), } ) tabs.sort(key=lambda item: (item.get("createdAt", ""), item.get("title", "").lower())) return tabs def save_tab(tab: dict[str, Any]) -> dict[str, str]: now = datetime.now(timezone.utc).isoformat() tab_id = str(tab.get("id") or uuid4()) saved_tab = { "id": tab_id, "title": str(tab.get("title", "New Tab")).strip() or "New Tab", "createdAt": str(tab.get("createdAt") or now), } response = requests.put( build_url(f"tabs/{tab_id}"), params=get_auth_params(), json=saved_tab, timeout=TIMEOUT_SECONDS, ) validate_response(response) return saved_tab def save_card(card: dict[str, Any]) -> dict[str, Any]: saved_card = sanitize_card(card) response = requests.put( build_url(f"cards/{saved_card['id']}"), params=get_auth_params(), json=saved_card, timeout=TIMEOUT_SECONDS, ) validate_response(response) return saved_card def update_card(card: dict[str, Any]) -> dict[str, Any]: card_id = str(card.get("id", "")).strip() if not card_id: raise RuntimeError("Card ID is required for update.") updated_card = sanitize_card(card, card_id=card_id) response = requests.put( build_url(f"cards/{card_id}"), params=get_auth_params(), json=updated_card, timeout=TIMEOUT_SECONDS, ) validate_response(response) return updated_card def delete_card(card_id: str) -> dict[str, Any]: if not str(card_id).strip(): raise RuntimeError("Card ID is required for deletion.") response = requests.delete( build_url(f"cards/{card_id}"), params=get_auth_params(), timeout=TIMEOUT_SECONDS, ) validate_response(response) return {"success": True}