""" CRUD de FUNCIONARIO_PAINEL — contas individuais de funcionário para login no painel (substitui/complementa o ADMIN_USER/ADMIN_PASSWORD único). A tabela já existe no banco de dev. Em qualquer outro ambiente, crie-a primeiro com: CREATE TABLE FUNCIONARIO_PAINEL ( ID INT UNSIGNED NOT NULL AUTO_INCREMENT, NOME VARCHAR(80) NOT NULL, LOGIN VARCHAR(40) NOT NULL, SENHA_HASH VARCHAR(255) NOT NULL, PERFIL ENUM('ADMIN','PADRAO') NOT NULL DEFAULT 'ADMIN', ATIVO TINYINT(1) NOT NULL DEFAULT 1, CRIADO_EM DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (ID), UNIQUE KEY UQ_FUNCIONARIO_PAINEL_LOGIN (LOGIN) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci; Se a tabela já existir sem a coluna PERFIL (ambientes criados antes dessa funcionalidade), rode: ALTER TABLE FUNCIONARIO_PAINEL ADD COLUMN PERFIL ENUM('ADMIN','PADRAO') NOT NULL DEFAULT 'ADMIN' AFTER SENHA_HASH; O DEFAULT 'ADMIN' do ALTER TABLE é intencional: todo funcionário cadastrado antes dessa migração vira admin automaticamente (decisão do projeto — ver docs/superpowers/specs/2026-08-13-permissoes-login-sessao-design.md). Cadastros novos sempre informam PERFIL explicitamente (vindo do formulário), nunca dependem desse default. Nunca excluído de verdade — "excluir" aqui é alternar ATIVO para 0 (mesma lógica de nunca apagar histórico usada no resto do projeto). """ from .connection import get_connection from ..senha import gerar_hash def listar_funcionarios() -> list[dict]: conn = get_connection() try: with conn.cursor() as cur: cur.execute( "SELECT ID, NOME, LOGIN, PERFIL, ATIVO, CRIADO_EM FROM FUNCIONARIO_PAINEL ORDER BY NOME" ) return cur.fetchall() finally: conn.close() def obter_funcionario(funcionario_id: int) -> dict | None: conn = get_connection() try: with conn.cursor() as cur: cur.execute( "SELECT ID, NOME, LOGIN, PERFIL, ATIVO, CRIADO_EM FROM FUNCIONARIO_PAINEL WHERE ID = %s", (funcionario_id,), ) return cur.fetchone() finally: conn.close() def obter_por_login(login: str) -> dict | None: """Usado pelo login (app/auth.py) — inclui SENHA_HASH, ao contrário das outras consultas.""" conn = get_connection() try: with conn.cursor() as cur: cur.execute( "SELECT ID, NOME, LOGIN, SENHA_HASH, PERFIL, ATIVO FROM FUNCIONARIO_PAINEL WHERE LOGIN = %s", (login,), ) return cur.fetchone() finally: conn.close() def login_em_uso(login: str, ignorar_id: int | None = None) -> bool: conn = get_connection() try: with conn.cursor() as cur: sql = "SELECT ID FROM FUNCIONARIO_PAINEL WHERE LOGIN = %s" params = [login] if ignorar_id is not None: sql += " AND ID != %s" params.append(ignorar_id) cur.execute(sql, params) return cur.fetchone() is not None finally: conn.close() def criar_funcionario(nome: str, login: str, senha: str, perfil: str) -> int: conn = get_connection() try: with conn.cursor() as cur: cur.execute( "INSERT INTO FUNCIONARIO_PAINEL (NOME, LOGIN, SENHA_HASH, PERFIL, ATIVO) VALUES (%s, %s, %s, %s, 1)", (nome, login, gerar_hash(senha), perfil), ) return cur.lastrowid finally: conn.close() def atualizar_funcionario(funcionario_id: int, nome: str, login: str, perfil: str) -> None: """Nome/login/perfil — troca de senha é redefinir_senha().""" conn = get_connection() try: with conn.cursor() as cur: cur.execute( "UPDATE FUNCIONARIO_PAINEL SET NOME = %s, LOGIN = %s, PERFIL = %s WHERE ID = %s", (nome, login, perfil, funcionario_id), ) finally: conn.close() def redefinir_senha(funcionario_id: int, nova_senha: str) -> bool: """True se o ID realmente existia (cur.rowcount > 0) — sem essa checagem um ID inexistente fazia um UPDATE que não afeta nenhuma linha, sem erro nenhum do MySQL, e o chamador registrava "sucesso" no log mesmo sem ter mudado a senha de ninguém.""" conn = get_connection() try: with conn.cursor() as cur: cur.execute( "UPDATE FUNCIONARIO_PAINEL SET SENHA_HASH = %s WHERE ID = %s", (gerar_hash(nova_senha), funcionario_id), ) return cur.rowcount > 0 finally: conn.close() def alternar_ativo(funcionario_id: int) -> None: conn = get_connection() try: with conn.cursor() as cur: cur.execute("UPDATE FUNCIONARIO_PAINEL SET ATIVO = 1 - ATIVO WHERE ID = %s", (funcionario_id,)) finally: conn.close() def contar_admins_ativos() -> int: conn = get_connection() try: with conn.cursor() as cur: cur.execute("SELECT COUNT(*) AS total FROM FUNCIONARIO_PAINEL WHERE PERFIL = 'ADMIN' AND ATIVO = 1") return cur.fetchone()["total"] finally: conn.close()