Spaces:
Sleeping
Sleeping
| import sys | |
| import uuid | |
| from datetime import datetime, timedelta, timezone | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).parent / "src")) | |
| from bootstrap_secrets import ensure_secrets_file | |
| ensure_secrets_file() | |
| import streamlit as st | |
| import cookie_token_store | |
| import google_oauth | |
| from analytics import inject_umami | |
| st.set_page_config(page_title="Board Game Tracker", layout="wide") | |
| inject_umami() | |
| def _drive_redirect_uri() -> str: | |
| return st.secrets["auth"]["redirect_uri"].removesuffix("/oauth2callback") | |
| def _store_access_token(tokens: dict) -> None: | |
| st.session_state["google_access_token"] = tokens["access_token"] | |
| expires_in = tokens.get("expires_in", 3600) | |
| st.session_state["google_access_token_expires_at"] = datetime.now( | |
| timezone.utc | |
| ) + timedelta(seconds=expires_in - 60) | |
| def _access_token_valid() -> bool: | |
| expires_at = st.session_state.get("google_access_token_expires_at") | |
| return ( | |
| bool(st.session_state.get("google_access_token")) | |
| and expires_at is not None | |
| and datetime.now(timezone.utc) < expires_at | |
| ) | |
| def _handle_drive_oauth_callback() -> None: | |
| """Completes the one-time "Connect Google Drive" exchange, if we're mid-flow. | |
| st.link_button always opens the authorize URL in a new tab, so this runs | |
| in a session that never had st.session_state populated by the tab that | |
| started the flow -- the CSRF state has to be compared against a cookie | |
| (shared across tabs), not session_state (tab-scoped). | |
| """ | |
| if not st.user.is_logged_in: | |
| return | |
| code = st.query_params.get("code") | |
| state = st.query_params.get("state") | |
| if not code or not state or state != cookie_token_store.get_oauth_state(): | |
| return | |
| cookie_token_store.clear_oauth_state() | |
| try: | |
| tokens = google_oauth.exchange_code( | |
| st.secrets["auth"]["client_id"], | |
| st.secrets["auth"]["client_secret"], | |
| _drive_redirect_uri(), | |
| code, | |
| ) | |
| except Exception as e: | |
| st.query_params.clear() | |
| st.error(f"Failed to connect Google Drive: {e}") | |
| return | |
| refresh_token = tokens.get("refresh_token") | |
| if not refresh_token: | |
| st.warning( | |
| "Google didn't return a refresh_token (this happens if the app was " | |
| "already authorized without prompt=consent taking effect). " | |
| "Drive access will work for this session only -- try disconnecting " | |
| "the app at myaccount.google.com/permissions and reconnecting." | |
| ) | |
| elif "cookies" in st.secrets: | |
| cookie_token_store.save_refresh_token(st.user.email, refresh_token) | |
| _store_access_token(tokens) | |
| st.query_params.clear() | |
| def _ensure_drive_access() -> None: | |
| """Populates session_state's Drive access token, silently refreshing from a | |
| stored refresh_token when possible. Shows a one-time "Connect Google Drive" | |
| link only when neither a live token nor a stored refresh_token exists. | |
| """ | |
| if not st.user.is_logged_in or _access_token_valid(): | |
| return | |
| if "cookies" in st.secrets: | |
| refresh_token = cookie_token_store.get_refresh_token(st.user.email) | |
| if refresh_token: | |
| try: | |
| tokens = google_oauth.refresh_access_token( | |
| st.secrets["auth"]["client_id"], | |
| st.secrets["auth"]["client_secret"], | |
| refresh_token, | |
| ) | |
| _store_access_token(tokens) | |
| return | |
| except Exception as e: | |
| st.warning(f"Stored Drive access expired or was revoked ({e}); reconnecting.") | |
| cookie_token_store.delete_refresh_token(st.user.email) | |
| state = uuid.uuid4().hex | |
| cookie_token_store.save_oauth_state(state) | |
| authorize_url = google_oauth.build_authorize_url( | |
| st.secrets["auth"]["client_id"], _drive_redirect_uri(), state | |
| ) | |
| st.link_button("Connect Google Drive", authorize_url) | |
| with st.expander("Drive connection diagnostics"): | |
| st.write("Logged in as:", st.user.email) | |
| st.write("Cookie storage configured:", "cookies" in st.secrets) | |
| st.write("Raw cookies seen by the app:", list(st.context.cookies.keys())) | |
| if "auth" in st.secrets: | |
| _handle_drive_oauth_callback() | |
| with st.sidebar: | |
| if "auth" not in st.secrets: | |
| pass # no auth configured in this environment; app stays anonymous-only | |
| elif not st.user.is_logged_in: | |
| st.button("Log in with Google", on_click=st.login) | |
| else: | |
| st.caption(f"Logged in as {st.user.email}") | |
| st.button("Log out", on_click=st.logout) | |
| _ensure_drive_access() | |
| home_page = st.Page("app_pages/home.py", title="Home", url_path="home", default=True) | |
| skull_king_page = st.Page("app_pages/skull_king.py", title="Skull King", url_path="skull_king") | |
| dungeon_draft_page = st.Page("app_pages/dungeon_draft.py", title="Dungeon Draft", url_path="dungeon_draft") | |
| points_tracker_page = st.Page("app_pages/points_tracker.py", title="Points Tracker", url_path="points_tracker") | |
| game_detail_page = st.Page("app_pages/game_detail.py", title="Game Detail", url_path="game_detail") | |
| pg = st.navigation( | |
| [home_page, skull_king_page, dungeon_draft_page, points_tracker_page, game_detail_page] | |
| ) | |
| pg.run() | |