RandomCatLover's picture
added good working stuff
69c9e26
Raw
History Blame Contribute Delete
5.3 kB
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()