# -*- coding: utf-8 -*- import os import sys import platform import importlib from uuid import uuid4 from datetime import date, datetime, time import streamlit as st from dotenv import load_dotenv from sqlalchemy import text, func # ✅ Usa toda a largura da página (chamar antes de qualquer outro st.*) st.set_page_config(layout="wide") # Carrega variáveis de ambiente load_dotenv() # -------------------------------------------------------------------------------------- # Utilitários de robustez / compatibilidade # -------------------------------------------------------------------------------------- def _module_stub(name: str, hint: str = ""): """Stub com .main() e .pagina() que exibe aviso amigável quando o módulo não puder ser importado/usado no ambiente atual.""" class _Stub: def __init__(self, _name, _hint): self.__name = _name self.__hint = _hint or "Módulo indisponível nesta hospedagem." def main(self, *args, **kwargs): st.warning(f"🔒 Módulo **{self.__name}** indisponível.\n\n{self.__hint}") def pagina(self, *args, **kwargs): self.main(*args, **kwargs) return _Stub(name, hint) def _try_import(module_name: str, on_fail_hint: str = ""): """Import 'seguro': caso falhe, retorna stub que não derruba o app.""" try: return importlib.import_module(module_name) except Exception as e: return _module_stub(module_name, f"{on_fail_hint}\n\n**Detalhe técnico:** {e}") def _ensure_db_case_alias(expected_names=("load.db", "Load.db", "LOAD.DB")): """Se existir um .db com caixa diferente, cria CÓPIA 'load.db' (Linux é case-sensitive).""" base = os.path.abspath(os.getcwd()) candidates = [os.path.join(base, n) for n in expected_names] lower = os.path.join(base, "load.db") if os.path.exists(lower): return lower for path in candidates: if os.path.exists(path): try: import shutil shutil.copy(path, lower) return lower except Exception: pass return lower # Ajuste preventivo do case do banco (caso módulos internos usem 'load.db') _DB_ALIAS = _ensure_db_case_alias() # -------------------------------------------------------------------------------------- # Imports internos essenciais # -------------------------------------------------------------------------------------- from utils_operacao import obter_grupos_disponiveis, obter_modulos_para_grupo from utils_info import INFO_CONTEUDO, INFO_MODULOS, INFO_MAP_PAGINA_ID from utils_permissoes import verificar_permissao from modules_map import MODULES from banco import engine, Base, SessionLocal from models import QuizPontuacao, IOIRunSugestao, AvisoGlobal # ----- login / logo: import e “safe wrappers” ----------------------------------------- # login() try: from login import login as _login_orig except Exception as _e_login_import: _login_orig = None _login_import_err = _e_login_import def login_safe(): """Tenta usar o login normal; se falhar, permite autologin (DISABLE_AUTH=1) ou opção emergencial (ALLOW_EMERGENCY_LOGIN=1).""" # Autologin (debug) if os.getenv("DISABLE_AUTH", "0") == "1": st.session_state.logado = True st.session_state.usuario = os.getenv("DEMO_USER", "demo") st.session_state.perfil = os.getenv("DEMO_PERFIL", "admin") st.session_state.email = os.getenv("DEMO_EMAIL", "demo@example.com") st.info("🔓 Autologin ativado (DISABLE_AUTH=1). **Não use em produção.**") return # Fluxo normal if _login_orig: try: _login_orig() return except Exception as e: st.error(f"Falha ao carregar tela de login (login.py): {e}") # Emergencial (opcional) if os.getenv("ALLOW_EMERGENCY_LOGIN", "0") == "1": with st.form("emergency_login"): u = st.text_input("Usuário") p = st.text_input("Senha", type="password") ok = st.form_submit_button("Entrar (emergencial)") if ok and u: st.session_state.logado = True st.session_state.usuario = u st.session_state.perfil = "admin" st.session_state.email = f"{u}@local" st.warning("⚠️ Login emergencial ativo. Desative após os testes (ALLOW_EMERGENCY_LOGIN=0).") st.rerun() else: st.info( "Login temporariamente indisponível. " "Para testar no Spaces, defina **DISABLE_AUTH=1** em *Settings → Secrets*." ) # exibir_logo() try: from utils_layout import exibir_logo as _exibir_logo_orig except Exception as _e_layout_import: _exibir_logo_orig = None _exibir_logo_import_err = _e_layout_import def _resolve_logo_path() -> str | None: """Procura a logo localmente; aceita LOGO_PATH absoluto ou relativo.""" cand = [] env_path = os.getenv("LOGO_PATH") if env_path: cand.append(env_path) cand += ["logo.png", "assets/logo.png", "images/logo.png", "static/logo.png"] base = os.path.abspath(os.getcwd()) for p in cand: if not p: continue full = p if os.path.isabs(p) else os.path.join(base, p) if os.path.exists(full): return full return None def exibir_logo_safe(top: bool = False, sidebar: bool = False): """Usa sua função original; se falhar, exibe imagem local ou fallback de texto.""" try: if _exibir_logo_orig: return _exibir_logo_orig(top=top, sidebar=sidebar) except Exception as e: st.sidebar.warning(f"Logo padrão indisponível ({e}). Usando fallback.") path = _resolve_logo_path() if path: if top: st.image(path, use_column_width=False) if sidebar: st.sidebar.image(path, use_column_width=True) else: if top: st.markdown("### IOI‑RUN") if sidebar: st.sidebar.markdown("### IOI‑RUN") # -------------------------------------------------------------------------------------- # Imports de páginas/módulos com fallback # -------------------------------------------------------------------------------------- formulario = _try_import("formulario") consulta = _try_import("consulta") relatorio = _try_import("relatorio") administracao = _try_import("administracao") quiz = _try_import("quiz") ranking = _try_import("ranking") quiz_admin = _try_import("quiz_admin") usuarios_admin = _try_import("usuarios_admin", "Adicione **bcrypt** ao requirements.txt.") videos = _try_import("videos") auditoria = _try_import("auditoria") importar_excel = _try_import("importar_excel") calendario = _try_import("calendario", "Instale **streamlit-calendar** no requirements.txt.") auditoria_cleanup = _try_import("auditoria_cleanup") jogos = _try_import("jogos") db_tools = _try_import("db_tools") db_admin = _try_import("db_admin") db_monitor = _try_import("db_monitor") operacao = _try_import("operacao") db_export_import = _try_import("db_export_import") resposta = _try_import("resposta") # 📬 Admin repositorio_load = _try_import("repositorio_load") produtividade_especialista = _try_import("Produtividade_Especialista") rnc = _try_import("rnc") rnc_listagem = _try_import("rnc_listagem") rnc_relatorio = _try_import("rnc_relatorio") sugestoes_usuario = _try_import("sugestoes_usuario") repo_rnc = _try_import("repo_rnc") recebimento = _try_import("recebimento") # Outlook/COM (somente Windows). Em Linux/Spaces: desativado ou stub. _DISABLE_OUTLOOK = os.getenv("DISABLE_OUTLOOK", "0") == "1" if platform.system().lower() != "windows" or _DISABLE_OUTLOOK: outlook_relatorio = _module_stub( "outlook_relatorio", "Este módulo usa automação COM do Outlook (pywin32), disponível **apenas no Windows**.\n" "No servidor Linux, ele foi desativado. Para manter essa função em nuvem, use **Microsoft Graph API**." ) else: outlook_relatorio = _try_import("outlook_relatorio") # -------------------------------------------------------------------------------------- # Roteamento condicional de banco # -------------------------------------------------------------------------------------- try: from db_router import current_db_choice, bank_label _HAS_ROUTER = True except Exception: _HAS_ROUTER = False def current_db_choice() -> str: return "prod" def bank_label(choice: str) -> str: return "🟢 Produção" if choice == "prod" else "🔴 Teste" # -------------------------------------------------------------------------------------- # RERUN por querystring (?rr=1) # -------------------------------------------------------------------------------------- def _get_query_params(): try: return dict(st.query_params) # Streamlit >= 1.32 except Exception: try: return dict(st.experimental_get_query_params()) except Exception: return {} def _set_query_params(new_params: dict): try: st.query_params = new_params except Exception: try: st.experimental_set_query_params(**new_params) except Exception: pass def _check_rerun_qs(pagina_atual: str = ""): try: if st.session_state.get("__qs_rr_consumed__", False): return # Não aplica RR dentro de módulos sensíveis if pagina_atual in ("resposta", "outlook_relatorio", "formulario"): return params = _get_query_params() rr_raw = params.get("rr", ["0"]) rr = rr_raw[0] if isinstance(rr_raw, (list, tuple)) else str(rr_raw) if str(rr).lower() in ("1", "true"): new_params = {k: v for k, v in params.items() if k != "rr"} _set_query_params(new_params) st.session_state["__qs_rr_consumed__"] = True st.rerun() except Exception: pass # -------------------------------------------------------------------------------------- # Sessões / DB helpers # -------------------------------------------------------------------------------------- def _get_db_session(): try: from db_router import get_session_for_current_db return get_session_for_current_db() except Exception: pass try: from db_router import get_engine_for_current_db from sqlalchemy.orm import sessionmaker Eng = get_engine_for_current_db() return sessionmaker(bind=Eng)() except Exception: pass from sqlalchemy.orm import sessionmaker return sessionmaker(bind=engine)() Base.metadata.create_all(bind=engine) def quiz_respondido_hoje(usuario: str) -> bool: db = _get_db_session() try: inicio_dia = datetime.combine(date.today(), time.min) return ( db.query(QuizPontuacao) .filter( QuizPontuacao.usuario == usuario, QuizPontuacao.data >= inicio_dia ) .first() is not None ) finally: try: db.close() except Exception: pass _SESS_TTL_MIN = 5 # janela para considerar "online" def _get_session_id() -> str: if "_sid" not in st.session_state: st.session_state["_sid"] = f"{uuid4()}" return st.session_state["_sid"] def _ensure_sessao_table(db) -> None: dialect = db.bind.dialect.name if dialect == "sqlite": db.execute(text(""" CREATE TABLE IF NOT EXISTS sessao_web ( id INTEGER PRIMARY KEY AUTOINCREMENT, usuario TEXT NOT NULL, session_id TEXT NOT NULL UNIQUE, last_seen TIMESTAMP NOT NULL, ativo INTEGER NOT NULL DEFAULT 1 ) """)) elif dialect in ("postgresql", "postgres"): db.execute(text(""" CREATE TABLE IF NOT EXISTS sessao_web ( id SERIAL PRIMARY KEY, usuario TEXT NOT NULL, session_id TEXT NOT NULL UNIQUE, last_seen TIMESTAMPTZ NOT NULL, ativo BOOLEAN NOT NULL DEFAULT TRUE ) """)) else: # mysql / mariadb db.execute(text(""" CREATE TABLE IF NOT EXISTS sessao_web ( id INT AUTO_INCREMENT PRIMARY KEY, usuario VARCHAR(255) NOT NULL, session_id VARCHAR(255) NOT NULL UNIQUE, last_seen TIMESTAMP NOT NULL, ativo TINYINT(1) NOT NULL DEFAULT 1 ) """)) db.commit() def _session_heartbeat(usuario: str) -> None: if not usuario: return db = _get_db_session() try: _ensure_sessao_table(db) sid = _get_session_id() now_sql = "CURRENT_TIMESTAMP" upd = db.execute( text(f"UPDATE sessao_web SET last_seen = {now_sql}, ativo = 1 WHERE session_id = :sid"), {"sid": sid} ) if upd.rowcount == 0: db.execute( text(f"INSERT INTO sessao_web (usuario, session_id, last_seen, ativo) " f"VALUES (:usuario, :sid, {now_sql}, 1)"), {"usuario": usuario, "sid": sid} ) dialect = db.bind.dialect.name if dialect in ("postgresql", "postgres"): cleanup_sql = f"UPDATE sessao_web SET ativo = FALSE WHERE last_seen < (NOW() - INTERVAL '{_SESS_TTL_MIN * 2} minutes')" elif dialect == "sqlite": cleanup_sql = f"UPDATE sessao_web SET ativo = 0 WHERE last_seen < datetime(CURRENT_TIMESTAMP, '-{_SESS_TTL_MIN * 2} minutes')" else: cleanup_sql = f"UPDATE sessao_web SET ativo = 0 WHERE last_seen < DATE_SUB(CURRENT_TIMESTAMP, INTERVAL {_SESS_TTL_MIN * 2} MINUTE)" db.execute(text(cleanup_sql)) db.commit() except Exception: db.rollback() finally: try: db.close() except Exception: pass def _get_active_users_count() -> int: db = _get_db_session() try: _ensure_sessao_table(db) dialect = db.bind.dialect.name if dialect in ("postgresql", "postgres"): threshold = f"(NOW() - INTERVAL '{_SESS_TTL_MIN} minutes')" elif dialect == "sqlite": threshold = f"datetime(CURRENT_TIMESTAMP, '-{_SESS_TTL_MIN} minutes')" else: threshold = f"DATE_SUB(CURRENT_TIMESTAMP, INTERVAL {_SESS_TTL_MIN} MINUTE)" res = db.execute( text(f"SELECT COUNT(DISTINCT usuario) AS c FROM sessao_web WHERE ativo = 1 AND last_seen >= {threshold}") ).fetchone() return int(res[0] if res and res[0] is not None else 0) except Exception: return 0 finally: try: db.close() except Exception: pass def _mark_session_inactive() -> None: sid = st.session_state.get("_sid") if not sid: return db = _get_db_session() try: _ensure_sessao_table(db) db.execute(text("UPDATE sessao_web SET ativo = 0 WHERE session_id = :sid"), {"sid": sid}) db.commit() except Exception: db.rollback() finally: try: db.close() except Exception: pass # -------------------------------------------------------------------------------------- # Aviso Global (banner superior) # -------------------------------------------------------------------------------------- def _sanitize_largura(largura_raw: str) -> str: val = (largura_raw or "").strip() if not val: return "100%" if val.endswith("%") or val.endswith("px"): return val if val.isdigit(): return f"{val}px" return "100%" def obter_aviso_ativo(db): try: aviso = ( db.query(AvisoGlobal) .filter(AvisoGlobal.ativo == True) .order_by(AvisoGlobal.updated_at.desc(), AvisoGlobal.created_at.desc()) .first() ) return aviso except Exception: return None def _render_aviso_global_topbar(): try: db = _get_db_session() except Exception as e: st.sidebar.warning(f"Aviso desativado: sessão indisponível ({e})") return aviso = None try: aviso = obter_aviso_ativo(db) except Exception as e: st.sidebar.warning(f"Aviso desativado: falha ao consultar ({e})") aviso = None finally: try: db.close() except Exception: pass if not aviso: return try: largura = _sanitize_largura(getattr(aviso, "largura", "100%")) bg = getattr(aviso, "bg_color", "#FFF3CD") or "#FFF3CD" fg = getattr(aviso, "text_color", "#664D03") or "#664D03" efeito = (getattr(aviso, "efeito", "marquee") or "marquee").lower() velocidade = int(getattr(aviso, "velocidade", 20) or 20) try: font_size = max(10, min(int(getattr(aviso, "font_size", 14)), 48)) except Exception: font_size = 14 altura = 52 # px st.markdown( f"""
""", unsafe_allow_html=True ) except Exception as e: st.sidebar.warning(f"Aviso desativado: erro de render ({e})") return # -------------------------------------------------------------------------------------- # Logout # -------------------------------------------------------------------------------------- def logout(): _mark_session_inactive() st.session_state.logado = False st.session_state.usuario = None st.session_state.perfil = None st.session_state.nome = None st.session_state.email = None st.session_state.quiz_verificado = False st.rerun() # -------------------------------------------------------------------------------------- # 🎂 Banner/efeito de aniversário # -------------------------------------------------------------------------------------- def _show_birthday_banner_if_needed(): if st.session_state.get("__show_birthday__"): st.session_state["__show_birthday__"] = False st.markdown( """