import streamlit as st import pandas as pd from streamlit_cookies_controller import CookieController from datetime import datetime, timedelta import hashlib, random, string, requests, json, time, uuid from streamlit.components.v1 import html as _html from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception from config import ( BREVO_API_KEY, EMAIL_FROM_NAME, EMAIL_FROM_ADDRESS, GS_SPREADSHEET, open_spreadsheet, ) st.set_page_config( page_title="TaskFlow | Login", page_icon="πŸ”", layout="wide", initial_sidebar_state="collapsed" ) # ── COOKIE SESSION ──────────────────────────────────────────────────── _cc = CookieController() COOKIE_NAME = "pmja_session" SESSION_EXPIRY_HOURS = 8 def save_session(user_id, username, expiry_hours=SESSION_EXPIRY_HOURS): expiry = datetime.now() + timedelta(hours=expiry_hours) _cc.set(COOKIE_NAME, { "user_id": str(user_id), "username": username, "expiry": expiry.isoformat(), "token": str(uuid.uuid4()), }) st.session_state.logged_in = True st.session_state.session_uid = str(user_id) st.session_state.session_usr = username st.session_state.session_exp = expiry def load_session(): # 1) session_state ainda ativo (mesma aba sem recarregar) if st.session_state.get("logged_in") and st.session_state.get("session_exp"): if datetime.now() < st.session_state.session_exp: return { "user_id": st.session_state.session_uid, "username": st.session_state.session_usr, } clear_session() return None # 2) cookie do browser (apΓ³s F5 ou nova aba no mesmo browser) try: c = _cc.get(COOKIE_NAME) if not c: return None expiry = datetime.fromisoformat(c["expiry"]) if datetime.now() >= expiry: _cc.remove(COOKIE_NAME) return None st.session_state.logged_in = True st.session_state.session_uid = c["user_id"] st.session_state.session_usr = c["username"] st.session_state.session_exp = expiry return {"user_id": c["user_id"], "username": c["username"]} except Exception: return None def clear_session(): try: _cc.remove(COOKIE_NAME) except Exception: pass st.session_state.update( logged_in=False, user_data=None, session_uid=None, session_usr=None, session_exp=None, ) # ── BREVO β€” validado tarde (apΓ³s CSS) ──────────────────────────────── # ── DB (gspread) ────────────────────────────────────────────────────── def _is_quota_or_server_error(exc: BaseException) -> bool: msg = str(exc) return any(code in msg for code in ("429", "500", "503", "RESOURCE_EXHAUSTED", "quota")) @retry( retry=retry_if_exception(_is_quota_or_server_error), wait=wait_exponential(multiplier=2, min=5, max=60), stop=stop_after_attempt(5), reraise=True, ) def _read_worksheet(worksheet: str) -> pd.DataFrame: """Le uma worksheet com retry automatico em caso de cota esgotada.""" try: ws = open_spreadsheet().worksheet(worksheet) data = ws.get_all_records() return pd.DataFrame(data) if data else pd.DataFrame() except Exception as exc: if _is_quota_or_server_error(exc): raise # deixa o tenacity fazer retry return pd.DataFrame() def _serialize(v): """Converte valores para string sem o sufixo .0 em floats inteiros.""" if v is None or v == "" or (isinstance(v, float) and pd.isna(v)): return "" try: f = float(v) if f == int(f) and not str(v).startswith("="): return str(int(f)) except (ValueError, TypeError): pass return str(v) @retry( retry=retry_if_exception(_is_quota_or_server_error), wait=wait_exponential(multiplier=2, min=5, max=60), stop=stop_after_attempt(5), reraise=True, ) def _update_worksheet(worksheet: str, df: pd.DataFrame) -> None: """Sobrescreve a worksheet com retry automatico em caso de cota esgotada.""" if df.empty: return ws = open_spreadsheet().worksheet(worksheet) rows = [[_serialize(v) for v in row] for row in df.fillna("").values.tolist()] ws.clear() ws.update('A1', [df.columns.tolist()] + rows, value_input_option='USER_ENTERED') def hash_password(p): return hashlib.sha256(p.encode()).hexdigest() def generate_code(): return "".join(random.choices(string.digits, k=6)) def send_verification_email(email, code, username): try: body = ( "
" "

PMJA Β· Scrum

" "

" "Código de verificação

" "
" f"{code}" "
" "

VΓ‘lido por 5 minutos.

" ) r = requests.post( "https://api.brevo.com/v3/smtp/email", headers={"api-key": BREVO_API_KEY, "content-type": "application/json"}, data=json.dumps({ "sender": {"name": EMAIL_FROM_NAME, "email": EMAIL_FROM_ADDRESS}, "to": [{"email": email}], "subject": "CΓ³digo – PMJA", "htmlContent": body, }), ) return r.status_code in [200, 201] except Exception: return False COLS = [ "id", "username", "email", "password", "full_name", "created_at", "last_login", "email_verified", "verification_code", "code_expiry", "image_url", ] def init_users_sheet(): try: df = _read_worksheet("users_auth") if not df.empty: for c in COLS: if c not in df.columns: df[c] = "" df["email_verified"] = ( df["email_verified"].astype(str).str.strip().str.lower().replace("nan", "false") ) for c in ["verification_code", "code_expiry", "image_url"]: df[c] = df[c].fillna("").astype(str).replace("nan", "").replace("None", "") return df except Exception: return pd.DataFrame(columns=COLS) def get_user_by_id(uid): df = init_users_sheet() uid = str(uid) u = df[df["id"].astype(str) == uid] return u.iloc[0].to_dict() if not u.empty else None def login_user(username, password): df = init_users_sheet() if df.empty: return False, "Nenhum usuΓ‘rio cadastrado" u = df[df["username"] == username] if u.empty: return False, "UsuΓ‘rio nΓ£o encontrado" if u.iloc[0]["password"] != hash_password(password): return False, "Senha incorreta" if str(u.iloc[0]["email_verified"]).strip().lower() not in ["true", "1", "yes", "1.0"]: return False, "EMAIL_NOT_VERIFIED" return True, u.iloc[0].to_dict() def register_user(username, email, password, full_name, image_url=""): df = init_users_sheet() if not df.empty and username in df["username"].values: return False, "UsuΓ‘rio jΓ‘ existe", None if not df.empty and email in df["email"].values: return False, "Email jΓ‘ cadastrado", None code = generate_code() expiry = (datetime.now() + timedelta(minutes=5)).strftime("%d/%m/%Y %H:%M:%S") new_id = 1 if df.empty else int(df["id"].max()) + 1 row = { "id": str(new_id), "username": str(username), "email": str(email), "password": hash_password(password), "full_name": str(full_name), "created_at": datetime.now().strftime("%d/%m/%Y %H:%M:%S"), "last_login": "", "email_verified": "false", "verification_code": str(code), "code_expiry": str(expiry), "image_url": str(image_url).strip() if image_url else "", } _update_worksheet( "users_auth", pd.concat([df, pd.DataFrame([row])], ignore_index=True), ) send_verification_email(email, code, username) return True, "Cadastro realizado!", code def resend_verification_code(username): df = init_users_sheet() u = df[df["username"] == username] if u.empty: return False, "UsuΓ‘rio nΓ£o encontrado" code = generate_code() expiry = (datetime.now() + timedelta(minutes=5)).strftime("%d/%m/%Y %H:%M:%S") email = u.iloc[0]["email"] df2 = _read_worksheet("users_auth") idx = df2[df2["username"] == username].index[0] df2.loc[idx, "verification_code"] = str(code) df2.loc[idx, "code_expiry"] = str(expiry) _update_worksheet("users_auth", df2) send_verification_email(email, code, username) return True, "CΓ³digo reenviado!" def verify_email_code(username, code): df = init_users_sheet() u = df[df["username"] == username] if u.empty: return False, "UsuΓ‘rio nΓ£o encontrado" stored = str(u.iloc[0]["verification_code"]).strip().replace(" ", "").replace(".0", "") inp = str(code).strip().replace(" ", "") if stored != inp and stored.lstrip("0") != inp.lstrip("0"): return False, "CΓ³digo incorreto" df2 = _read_worksheet("users_auth") idx = df2[df2["username"] == username].index[0] df2.loc[idx, "email_verified"] = "true" df2.loc[idx, "verification_code"] = "" df2.loc[idx, "code_expiry"] = "" _update_worksheet("users_auth", df2) time.sleep(1) return True, "Email verificado!" # ── session state ───────────────────────────────────────────────────── for k, v in [ ("logged_in", False), ("user_data", None), ("page", "login"), ("temp_username", None), ("msg", ""), ("msg_type", ""), ("_action", None), ("session_uid", None), ("session_usr", None), ("session_exp", None), ]: if k not in st.session_state: st.session_state[k] = v # ── verifica sessΓ£o ativa ───────────────────────────────────────────── # O CookieController precisa de um rerun para ler os cookies do browser. # Na primeira renderizacao ele ainda nao os carregou, entao bloqueamos # o render do formulario ate o cookie estar disponivel (ou confirmado ausente). _cookie_raw = _cc.get(COOKIE_NAME) # None = ainda nao lido OR nao existe if not st.session_state.get("_cookies_ready"): # Se nao ha cookie, nao precisa esperar β€” marca pronto e continua if _cookie_raw is None: st.session_state["_cookies_ready"] = True else: # Ha cookie mas o estado interno ainda nao foi populado β€” carrega direto st.session_state["_cookies_ready"] = True # Carrega sessao direto aqui antes do st.stop() try: expiry = datetime.fromisoformat(_cookie_raw["expiry"]) if datetime.now() < expiry: st.session_state.logged_in = True st.session_state.session_uid = _cookie_raw["user_id"] st.session_state.session_usr = _cookie_raw["username"] st.session_state.session_exp = expiry ud = get_user_by_id(_cookie_raw["user_id"]) if ud: st.session_state.user_data = ud st.switch_page("pages/tasks.py") st.stop() else: _cc.remove(COOKIE_NAME) except Exception: pass if not st.session_state.logged_in: s = load_session() if s: ud = get_user_by_id(s["user_id"]) if ud: st.session_state.logged_in = True st.session_state.user_data = ud st.switch_page("pages/tasks.py") st.stop() # nao renderiza nada enquanto o redirect acontece # ── process pending action ──────────────────────────────────────────── action = st.session_state.pop("_action", None) if "_action" in st.session_state else None if action == "go_register": st.session_state.page = "register" st.session_state.msg = "" elif action == "go_login": st.session_state.page = "login" st.session_state.msg = "" elif action == "resend": ok, msg = resend_verification_code(st.session_state.temp_username) st.session_state.msg = msg st.session_state.msg_type = "success" if ok else "error" elif action == "logout": clear_session() st.session_state.page = "login" elif action == "extend": user = st.session_state.user_data if user: save_session(user["id"], user["username"], SESSION_EXPIRY_HOURS) st.session_state.msg = "SessΓ£o estendida por 2 horas" st.session_state.msg_type = "success" # ── CSS ─────────────────────────────────────────────────────────────── _CSS = """\ """ _html(_CSS, height=0) # ── Brevo: validado apΓ³s CSS para erro ficar estilizado ─────────────── if not BREVO_API_KEY or not EMAIL_FROM_ADDRESS: st.error("Configure BREVO_API_KEY e EMAIL_FROM_ADDRESS no .env") st.stop() # ── page state ──────────────────────────────────────────────────────── pg = "dashboard" if st.session_state.logged_in else st.session_state.page user = st.session_state.user_data PHOTOS = { "login": "https://drudu6g9smo13.cloudfront.net/wp-content/uploads/2023/08/UHE-Jaguara.jpg", "register": "https://s2.glbimg.com/e844-OclDbLw-PWuboUy_wtbhiQ=/512x320/smart/e.glbimg.com/og/ed/f/original/2021/12/01/hidreletrica_jaguara_-_rifaina_sp_SGhwNiF.jpg", "verify": "https://upload.wikimedia.org/wikipedia/commons/9/99/Usina_Hidrel%C3%A9trica_de_Jaguara_%28Rifaina-SP_%284478616135%29.jpg", "dashboard": "https://drudu6g9smo13.cloudfront.net/wp-content/uploads/2023/08/UHE-Jaguara.jpg", } TITLES = { "login": ("Bem-vindo de volta", "ACESSAR CONTA"), "register": ("Criar conta", "DADOS DE CADASTRO"), "verify": ("Verificar email", "CΓ“DIGO DE VERIFICAÇÃO"), "dashboard": ( ("OlΓ‘, " + user["full_name"].split()[0]) if user else "Painel", "PAINEL", ), } banner_title, form_label = TITLES.get(pg, TITLES["login"]) photo = PHOTOS.get(pg, PHOTOS["login"]) def flush(): m, t = st.session_state.msg, st.session_state.msg_type st.session_state.msg = "" st.session_state.msg_type = "" if not m: return {"error": st.error, "success": st.success}.get(t, st.info)(m) def sp(px): st.markdown(f"
", unsafe_allow_html=True) def divider(): st.markdown( "
" "
" "ou" "
", unsafe_allow_html=True, ) # ═════════════════════════════════════════════════════════════════════ # LAYOUT # ═════════════════════════════════════════════════════════════════════ _, card, _ = st.columns([1, 1.4, 1]) with card: # BANNER st.markdown( f"
" f"" "
" "
" "
PMJA Β· UHE Jaguara Β· Rifaina SP
" f"
{banner_title}
" "
", unsafe_allow_html=True, ) # SECTION LABEL st.markdown( f"
{form_label}
", unsafe_allow_html=True, ) # ══════════════════════════════════════════════════════════════ # LOGIN # ══════════════════════════════════════════════════════════════ if pg == "login": flush() sp(14) with st.form("f_login", border=False): st.text_input("UsuΓ‘rio", placeholder="nome_usuario", key="li_u") st.text_input("Senha", placeholder="β€’β€’β€’β€’β€’β€’β€’β€’", type="password", key="li_p") sub = st.form_submit_button("Entrar β†’", type="primary", use_container_width=True) if sub: u = st.session_state.li_u.strip() p = st.session_state.li_p if not (u and p): st.session_state.msg = "Preencha todos os campos" st.session_state.msg_type = "error" st.rerun() else: ok, result = login_user(u, p) if ok: st.session_state.logged_in = True st.session_state.user_data = result save_session(result["id"], u, SESSION_EXPIRY_HOURS) st.switch_page("pages/tasks.py") elif result == "EMAIL_NOT_VERIFIED": resend_verification_code(u) st.session_state.temp_username = u st.session_state.page = "verify" st.session_state.msg = "πŸ“§ CΓ³digo reenviado para seu email" st.session_state.msg_type = "info" st.rerun() else: st.session_state.msg = result st.session_state.msg_type = "error" st.rerun() divider() if st.button("Criar nova conta β†’", use_container_width=True, key="go_reg"): st.session_state.page = "register" st.session_state.msg = "" st.rerun() sp(4) # ══════════════════════════════════════════════════════════════ # REGISTER # ══════════════════════════════════════════════════════════════ elif pg == "register": flush() sp(14) with st.form("f_register", border=False): st.text_input("Nome completo", placeholder="Seu nome completo", key="rg_n") st.text_input("Email", placeholder="seu@email.com", key="rg_e") c1, c2 = st.columns(2) with c1: st.text_input("UsuΓ‘rio", placeholder="nome_usuario", key="rg_u") with c2: st.text_input("URL da foto", placeholder="https://...", key="rg_i") c3, c4 = st.columns(2) with c3: st.text_input("Senha", placeholder="MΓ­n. 6 caracteres", type="password", key="rg_p") with c4: st.text_input("Confirmar senha", placeholder="Repita a senha", type="password", key="rg_p2") sub = st.form_submit_button("Cadastrar β†’", type="primary", use_container_width=True) if sub: fn = st.session_state.rg_n.strip() em = st.session_state.rg_e.strip() us = st.session_state.rg_u.strip() img = st.session_state.rg_i.strip() pw = st.session_state.rg_p pw2 = st.session_state.rg_p2 if not all([fn, em, us, pw, pw2]): st.session_state.msg = "Preencha todos os campos" st.session_state.msg_type = "error" elif len(pw) < 6: st.session_state.msg = "Senha mΓ­nimo 6 caracteres" st.session_state.msg_type = "error" elif pw != pw2: st.session_state.msg = "Senhas nΓ£o coincidem" st.session_state.msg_type = "error" else: ok, msg, _ = register_user(us, em, pw, fn, img) if ok: st.session_state.temp_username = us st.session_state.page = "verify" st.session_state.msg = "Conta criada! Verifique seu email." st.session_state.msg_type = "success" else: st.session_state.msg = msg st.session_state.msg_type = "error" st.rerun() divider() if st.button("← JΓ‘ tenho conta", use_container_width=True, key="go_log"): st.session_state.page = "login" st.session_state.msg = "" st.rerun() sp(4) # ══════════════════════════════════════════════════════════════ # VERIFY # ══════════════════════════════════════════════════════════════ elif pg == "verify": flush() sp(14) st.markdown( "
" "πŸ“§" "" "CΓ³digo enviado por email. VΓ‘lido por 5 minutos.
", unsafe_allow_html=True, ) with st.form("f_verify", border=False): st.text_input("CΓ³digo de 6 dΓ­gitos", placeholder="000000", max_chars=6, key="vf_c") sp(8) cv1, cv2 = st.columns(2) with cv1: confirm = st.form_submit_button("Confirmar", type="primary", use_container_width=True) with cv2: go_back = st.form_submit_button("← Voltar", use_container_width=True) if confirm: ok, msg = verify_email_code(st.session_state.temp_username, st.session_state.vf_c) st.session_state.page = "login" if ok else "verify" st.session_state.msg = msg st.session_state.msg_type = "success" if ok else "error" st.rerun() if go_back: st.session_state.page = "login" st.rerun() sp(8) if st.button("Reenviar cΓ³digo por email", use_container_width=True, key="resend"): ok, msg = resend_verification_code(st.session_state.temp_username) st.session_state.msg = msg st.session_state.msg_type = "success" if ok else "error" st.rerun() sp(4) # ══════════════════════════════════════════════════════════════ # DASHBOARD # ══════════════════════════════════════════════════════════════ else: flush() uname = user["full_name"] if user else "" uemail = user["email"] if user else "" initial = uname[0].upper() if uname else "?" sp(14) st.markdown( "
" "
{initial}
" f"
{uname}
" f"
{uemail}
", unsafe_allow_html=True, ) if st.button("πŸ“‹ Acessar Tarefas β†’", type="primary", use_container_width=True, key="tasks"): st.switch_page("pages/tasks.py") sp(10) d1, d2 = st.columns(2) with d1: if st.button("πŸ”„ Estender SessΓ£o", use_container_width=True, key="extend"): save_session(user["id"], user["username"], SESSION_EXPIRY_HOURS) st.session_state.msg = "SessΓ£o estendida por 2 horas" st.session_state.msg_type = "success" st.rerun() with d2: if st.button("Sair", use_container_width=True, key="logout"): clear_session() st.session_state.page = "login" st.rerun() sp(4) # FOOTER st.markdown( "
" "PMJA Β· UHE Jaguara Β· Rifaina SP
", unsafe_allow_html=True, )