| 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} |
|
|
