File size: 5,121 Bytes
21fe460 45cf072 21fe460 45cf072 21fe460 45cf072 21fe460 45cf072 21fe460 45cf072 21fe460 45cf072 21fe460 45cf072 21fe460 45cf072 fd42852 45cf072 fd42852 45cf072 21fe460 fd42852 21fe460 45cf072 21fe460 45cf072 21fe460 45cf072 21fe460 45cf072 21fe460 45cf072 21fe460 45cf072 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | 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}
|