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, os from streamlit.components.v1 import html as _html from dotenv import load_dotenv load_dotenv() from gs_client import read_ws, write_ws st.set_page_config( page_title="PMJA | Login", page_icon="πŸ”", layout="wide", initial_sidebar_state="collapsed" ) # ── ESCONDE UI DO STREAMLIT ─────────────────────────────────────────── st.markdown("""""", unsafe_allow_html=True) # ── 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(): 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 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.update(logged_in=True, session_uid=c["user_id"], session_usr=c["username"], 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) # ── EMAIL ───────────────────────────────────────────────────────────── BREVO_API_KEY = os.getenv("BREVO_API_KEY", "") EMAIL_FROM_NAME = os.getenv("EMAIL_FROM_NAME", "PMJA Sistema") EMAIL_FROM_ADDRESS = os.getenv("EMAIL_FROM_ADDRESS", "") # ── HELPERS ─────────────────────────────────────────────────────────── 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", ] # ── DB ──────────────────────────────────────────────────────────────── def get_users() -> pd.DataFrame: df = read_ws("users_auth", spreadsheet="main") if df.empty: return pd.DataFrame(columns=COLS) 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 def get_user_by_id(uid): df = get_users() u = df[df["id"].astype(str).str.split(".").str[0] == str(uid).split(".")[0]] return u.iloc[0].to_dict() if not u.empty else None def login_user(username, password): df = get_users() 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 = get_users() 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 "", } write_ws("users_auth", pd.concat([df, pd.DataFrame([row])], ignore_index=True), spreadsheet="main") send_verification_email(email, code, username) return True, "Cadastro realizado!", code def resend_verification_code(username): df = get_users() 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 = get_users() idx = df2[df2["username"] == username].index[0] df2.loc[idx, "verification_code"] = str(code) df2.loc[idx, "code_expiry"] = str(expiry) write_ws("users_auth", df2, spreadsheet="main") send_verification_email(email, code, username) return True, "CΓ³digo reenviado!" def verify_email_code(username, code): df = get_users() 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 = get_users() idx = df2[df2["username"] == username].index[0] df2.loc[idx, "email_verified"] = "true" df2.loc[idx, "verification_code"] = "" df2.loc[idx, "code_expiry"] = "" write_ws("users_auth", df2, spreadsheet="main") 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), ("show_loading", False), ]: if k not in st.session_state: st.session_state[k] = v 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/rec.py") 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" # ── LOADING SCREEN MD ──────────────────────────────────────────────── LOADING_MD = """
PMJA β€” GestΓ£o de Materiais
UHE Jaguara Β· Rifaina SP
Autenticando ...
""" # ── CSS ─────────────────────────────────────────────────────────────── _CSS = """\ """ _html(_CSS, height=0) # ── 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: st.markdown( f"
" f"" "
" "
" "
PMJA Β· UHE Jaguara Β· Rifaina SP
" f"
{banner_title}
" "
", unsafe_allow_html=True, ) 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: # Mostra loading antes de autenticar loading_ph = st.empty() loading_ph.markdown(LOADING_MD, unsafe_allow_html=True) 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/rec.py") elif result == "EMAIL_NOT_VERIFIED": loading_ph.empty() 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: loading_ph.empty() 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, em, us = st.session_state.rg_n.strip(), st.session_state.rg_e.strip(), st.session_state.rg_u.strip() img, pw, pw2 = st.session_state.rg_i.strip(), st.session_state.rg_p, 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/rec.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) st.markdown( "
" "PMJA Β· UHE Jaguara Β· Rifaina SP
", unsafe_allow_html=True, ) #rec.py