import hashlib import sqlite3 import re from pathlib import Path from config import Config BASE_DIR = Path(__file__).resolve().parent DB_PATH = Path(getattr(Config, "DB_PATH", BASE_DIR / "database" / "cofihst.sqlite3")) class Error(Exception): pass def gerar_hash(password): return hashlib.sha256(password.encode("utf-8")).hexdigest() class SQLiteCursorCompat: def __init__(self, conn, dictionary=False): self.conn = conn self.dictionary = dictionary self.cur = conn.cursor() self._last_result = None @property def lastrowid(self): return self.cur.lastrowid def _convert_sql(self, sql): sql = sql.strip() sql = sql.replace("%s", "?") sql = re.sub(r"DATE_ADD\(CURDATE\(\),\s*INTERVAL\s*\?\s*DAY\)", "date('now', '+' || ? || ' days')", sql, flags=re.I) sql = re.sub(r"CURDATE\(\)", "date('now')", sql, flags=re.I) sql = re.sub(r"NOW\(\)", "CURRENT_TIMESTAMP", sql, flags=re.I) sql = re.sub(r"DATE_FORMAT\(([^,]+),\s*'([^']+)'\)", lambda m: f"strftime('{m.group(2).replace('%Y','%Y').replace('%m','%m').replace('%d','%d')}', {m.group(1)})", sql, flags=re.I) sql = re.sub(r"YEAR\(([^)]+)\)", r"CAST(strftime('%Y', \1) AS INTEGER)", sql, flags=re.I) sql = re.sub(r"MONTH\(([^)]+)\)", r"CAST(strftime('%m', \1) AS INTEGER)", sql, flags=re.I) sql = re.sub(r"DAY\(([^)]+)\)", r"CAST(strftime('%d', \1) AS INTEGER)", sql, flags=re.I) return sql def execute(self, sql, params=None): params = tuple(params or ()) raw_sql = sql.strip() upper = raw_sql.upper() if "INFORMATION_SCHEMA.COLUMNS" in upper: table = "consumiveis" m = re.search(r"TABLE_NAME\s*=\s*'([^']+)'", raw_sql, flags=re.I) if m: table = m.group(1) info = self.conn.execute(f"PRAGMA table_info({table})").fetchall() self._last_result = [{"COLUMN_NAME": row[1]} for row in info] return self if "SELECT VERSION" in upper: self.cur.execute("SELECT sqlite_version();") return self if upper.startswith("CREATE DATABASE") or upper.startswith("USE "): self._last_result = [] return self if upper.startswith("ALTER TABLE") and ", ADD COLUMN" in upper: prefix, rest = raw_sql.split("ADD COLUMN", 1) parts = ["ADD COLUMN " + p.strip().rstrip(";") for p in rest.split(", ADD COLUMN")] for part in parts: stmt = self._convert_sql(prefix + part) try: self.cur.execute(stmt) except sqlite3.OperationalError as e: if "duplicate column name" not in str(e).lower(): raise return self sql = self._convert_sql(raw_sql) self.cur.execute(sql, params) return self def fetchone(self): if self._last_result is not None: if not self._last_result: return None return self._last_result.pop(0) row = self.cur.fetchone() if row is None: return None return dict(row) if self.dictionary else tuple(row) def fetchall(self): if self._last_result is not None: dados = self._last_result self._last_result = None return dados rows = self.cur.fetchall() return [dict(r) for r in rows] if self.dictionary else [tuple(r) for r in rows] def close(self): self.cur.close() class SQLiteConnectionCompat: def __init__(self, path): path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) self.conn = sqlite3.connect(path) self.conn.row_factory = sqlite3.Row self.conn.execute("PRAGMA foreign_keys = ON;") def cursor(self, dictionary=False): return SQLiteCursorCompat(self.conn, dictionary=dictionary) def commit(self): self.conn.commit() def rollback(self): self.conn.rollback() def close(self): self.conn.close() def is_connected(self): return True SQLITE_SCHEMA = "\nCREATE TABLE IF NOT EXISTS utilizadores (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n nome TEXT NOT NULL,\n username TEXT NOT NULL UNIQUE,\n password_hash TEXT NOT NULL,\n cargo TEXT NOT NULL CHECK (cargo IN ('CEO', 'Admin', 'Gestor', 'Medico', 'Enfermeiro')),\n ativo INTEGER NOT NULL DEFAULT 1,\n criado_em TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP\n);\nCREATE TABLE IF NOT EXISTS empresas (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n nome TEXT NOT NULL UNIQUE,\n nif TEXT, telefone TEXT, email TEXT, morada TEXT,\n ativo INTEGER NOT NULL DEFAULT 1,\n criado_em TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP\n);\nCREATE TABLE IF NOT EXISTS utentes (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n numero_utente TEXT, nome TEXT NOT NULL, data_nascimento TEXT, sexo TEXT,\n numero_identificacao TEXT, nif TEXT, telefone TEXT, email TEXT, morada TEXT,\n codigo_postal TEXT, localidade TEXT, empresa_id INTEGER, departamento TEXT,\n funcao TEXT, data_admissao TEXT, tipo_exame TEXT, data_ultimo_exame TEXT,\n data_proximo_exame TEXT, observacoes TEXT,\n ativo INTEGER NOT NULL DEFAULT 1,\n criado_em TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,\n FOREIGN KEY (empresa_id) REFERENCES empresas(id)\n);\nCREATE TABLE IF NOT EXISTS marcacoes (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n utente_id INTEGER NOT NULL,\n medico_id INTEGER,\n data_marcacao TEXT NOT NULL,\n hora_marcacao TEXT NOT NULL,\n tipo_exame TEXT,\n estado TEXT NOT NULL DEFAULT 'Marcado',\n observacoes TEXT,\n criado_em TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,\n FOREIGN KEY (utente_id) REFERENCES utentes(id),\n FOREIGN KEY (medico_id) REFERENCES utilizadores(id)\n);\nCREATE TABLE IF NOT EXISTS fornecedores (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n nome TEXT NOT NULL,\n email TEXT, telefone TEXT, observacoes TEXT,\n ativo INTEGER NOT NULL DEFAULT 1,\n criado_em TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP\n);\nCREATE TABLE IF NOT EXISTS consumiveis (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n codigo TEXT, categoria TEXT, nome_produto TEXT NOT NULL,\n unidade TEXT NOT NULL DEFAULT 'unidade',\n quantidade_stock INTEGER NOT NULL DEFAULT 0,\n stock_minimo INTEGER NOT NULL DEFAULT 0,\n preco_unitario REAL NOT NULL DEFAULT 0.00,\n tipo_embalagem TEXT DEFAULT 'unidade',\n unidades_por_embalagem INTEGER NOT NULL DEFAULT 1,\n fornecedor_id INTEGER,\n data_validade TEXT,\n observacoes TEXT,\n referencia_fornecedor TEXT,\n ativo INTEGER NOT NULL DEFAULT 1,\n criado_em TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,\n FOREIGN KEY (fornecedor_id) REFERENCES fornecedores(id)\n);\nCREATE TABLE IF NOT EXISTS consumiveis_usados_consulta (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n marcacao_id INTEGER, utente_id INTEGER NOT NULL, medico_id INTEGER,\n consumivel_id INTEGER NOT NULL, quantidade_usada INTEGER NOT NULL,\n unidade TEXT, preco_unitario REAL NOT NULL DEFAULT 0.00,\n data_registo TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,\n FOREIGN KEY (marcacao_id) REFERENCES marcacoes(id),\n FOREIGN KEY (utente_id) REFERENCES utentes(id),\n FOREIGN KEY (medico_id) REFERENCES utilizadores(id),\n FOREIGN KEY (consumivel_id) REFERENCES consumiveis(id)\n);\nCREATE TABLE IF NOT EXISTS movimentos_stock (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n consumivel_id INTEGER NOT NULL,\n tipo_movimento TEXT NOT NULL,\n quantidade INTEGER NOT NULL,\n stock_antes INTEGER NOT NULL,\n stock_depois INTEGER NOT NULL,\n utilizador_id INTEGER,\n origem TEXT,\n data_movimento TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,\n FOREIGN KEY (consumivel_id) REFERENCES consumiveis(id),\n FOREIGN KEY (utilizador_id) REFERENCES utilizadores(id)\n);\nCREATE TABLE IF NOT EXISTS historico_acoes (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n utilizador_id INTEGER,\n utilizador_nome TEXT,\n cargo TEXT,\n acao TEXT NOT NULL,\n modulo TEXT,\n entidade TEXT,\n id_entidade INTEGER,\n valor_antigo TEXT,\n valor_novo TEXT,\n data_acao TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,\n FOREIGN KEY (utilizador_id) REFERENCES utilizadores(id)\n);\nCREATE TABLE IF NOT EXISTS notificacoes (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n titulo TEXT NOT NULL,\n mensagem TEXT NOT NULL,\n tipo TEXT,\n cargo_destino TEXT,\n lida INTEGER NOT NULL DEFAULT 0,\n criada_em TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP\n);\n" def criar_base_dados_se_nao_existir(): DB_PATH.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(DB_PATH) conn.execute("PRAGMA foreign_keys = ON;") conn.executescript(SQLITE_SCHEMA) conn.execute("CREATE VIEW IF NOT EXISTS consumiveis_consulta AS SELECT * FROM consumiveis_usados_consulta;") conn.commit() conn.close() def ligar_mysql(usar_base=True): # Mantém o nome antigo para não ser preciso alterar o resto do código. # Internamente, agora usa SQLite local. criar_base_dados_se_nao_existir() return SQLiteConnectionCompat(DB_PATH) def testar_ligacao_mysql(): try: criar_base_dados_se_nao_existir() ligacao = SQLiteConnectionCompat(DB_PATH) cursor = ligacao.cursor() cursor.execute("SELECT sqlite_version();") versao = cursor.fetchone()[0] cursor.close() ligacao.close() return True, f"SQLite ligado com sucesso. Versão: {versao}" except Exception as erro: return False, str(erro) def executar_schema_mysql(caminho_schema=None): try: criar_base_dados_se_nao_existir() criar_utilizador_ceo_inicial() garantir_consumiveis_iniciais() return True, "Base de dados SQLite inicializada com sucesso. CEO inicial confirmado." except Exception as erro: return False, str(erro) def criar_utilizador_ceo_inicial(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id FROM utilizadores WHERE username = %s LIMIT 1; """, ("ceo",)) existe = cursor.fetchone() if not existe: cursor.execute(""" INSERT INTO utilizadores (nome, username, password_hash, cargo, ativo) VALUES (%s, %s, %s, %s, %s); """, ("CEO", "ceo", gerar_hash("1234"), "CEO", 1)) ligacao.commit() cursor.close() ligacao.close() def autenticar_utilizador(username, password): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, nome, username, password_hash, cargo, ativo FROM utilizadores WHERE username = %s LIMIT 1; """, (username,)) utilizador = cursor.fetchone() cursor.close() ligacao.close() if not utilizador: return None if not utilizador["ativo"]: return None if utilizador["password_hash"] != gerar_hash(password): return None return { "id": utilizador["id"], "nome": utilizador["nome"], "username": utilizador["username"], "cargo": utilizador["cargo"], } # ---------------- GESTÃO DE UTILIZADORES ---------------- def listar_utilizadores(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, nome, username, cargo, ativo, criado_em FROM utilizadores ORDER BY cargo, nome; """) utilizadores = cursor.fetchall() cursor.close() ligacao.close() return utilizadores def username_existe(username): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id FROM utilizadores WHERE username = %s LIMIT 1; """, (username,)) resultado = cursor.fetchone() cursor.close() ligacao.close() return resultado is not None def criar_utilizador(nome, username, password, cargo): if username_existe(username): return False, "Esse username já está em uso." ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" INSERT INTO utilizadores (nome, username, password_hash, cargo, ativo) VALUES (%s, %s, %s, %s, 1); """, (nome, username, gerar_hash(password), cargo)) ligacao.commit() cursor.close() ligacao.close() return True, "Utilizador criado com sucesso." def obter_utilizador_por_id(user_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, nome, username, cargo, ativo FROM utilizadores WHERE id = %s LIMIT 1; """, (user_id,)) utilizador = cursor.fetchone() cursor.close() ligacao.close() return utilizador def alterar_estado_utilizador(user_id, ativo): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" UPDATE utilizadores SET ativo = %s WHERE id = %s; """, (ativo, user_id)) ligacao.commit() cursor.close() ligacao.close() return True, "Estado do utilizador atualizado." def atualizar_password_utilizador(user_id, nova_password): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" UPDATE utilizadores SET password_hash = %s WHERE id = %s; """, (gerar_hash(nova_password), user_id)) ligacao.commit() cursor.close() ligacao.close() return True, "Password atualizada com sucesso." # ---------------- GESTÃO DE EMPRESAS E UTENTES ---------------- def listar_empresas(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, nome, nif, telefone, email, ativo FROM empresas WHERE ativo = 1 ORDER BY nome; """) empresas = cursor.fetchall() cursor.close() ligacao.close() return empresas def criar_empresa(nome, nif="", telefone="", email="", morada=""): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" SELECT id FROM empresas WHERE nome = %s LIMIT 1; """, (nome,)) existe = cursor.fetchone() if existe: cursor.close() ligacao.close() return False, "Essa empresa já existe." cursor.execute(""" INSERT INTO empresas (nome, nif, telefone, email, morada, ativo) VALUES (%s, %s, %s, %s, %s, 1); """, (nome, nif, telefone, email, morada)) ligacao.commit() cursor.close() ligacao.close() return True, "Empresa criada com sucesso." def listar_utentes(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT u.id, u.numero_utente, u.nome, u.data_nascimento, u.sexo, u.telefone, u.email, u.departamento, u.funcao, u.tipo_exame, u.ativo, e.nome AS empresa FROM utentes u LEFT JOIN empresas e ON e.id = u.empresa_id WHERE u.ativo = 1 ORDER BY e.nome, u.nome; """) utentes = cursor.fetchall() cursor.close() ligacao.close() return utentes def criar_utente(dados): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" INSERT INTO utentes ( numero_utente, nome, data_nascimento, sexo, numero_identificacao, nif, telefone, email, morada, codigo_postal, localidade, empresa_id, departamento, funcao, data_admissao, tipo_exame, data_ultimo_exame, data_proximo_exame, observacoes, ativo ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 1); """, ( dados.get("numero_utente"), dados.get("nome"), dados.get("data_nascimento") or None, dados.get("sexo"), dados.get("numero_identificacao"), dados.get("nif"), dados.get("telefone"), dados.get("email"), dados.get("morada"), dados.get("codigo_postal"), dados.get("localidade"), dados.get("empresa_id") or None, dados.get("departamento"), dados.get("funcao"), dados.get("data_admissao") or None, dados.get("tipo_exame"), dados.get("data_ultimo_exame") or None, dados.get("data_proximo_exame") or None, dados.get("observacoes"), )) ligacao.commit() cursor.close() ligacao.close() return True, "Utente criado com sucesso." def desativar_utente(utente_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" UPDATE utentes SET ativo = 0 WHERE id = %s; """, (utente_id,)) ligacao.commit() cursor.close() ligacao.close() return True, "Utente removido da lista ativa." # ---------------- GESTÃO DE MARCAÇÕES ---------------- def listar_medicos_ativos(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, nome, username FROM utilizadores WHERE ativo = 1 AND cargo = 'Medico' ORDER BY nome; """) medicos = cursor.fetchall() cursor.close() ligacao.close() return medicos def listar_enfermeiros_ativos(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, nome, username FROM utilizadores WHERE ativo = 1 AND cargo = 'Enfermeiro' ORDER BY nome; """) enfermeiros = cursor.fetchall() cursor.close() ligacao.close() return enfermeiros def listar_utentes_para_select(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT u.id, u.nome, u.numero_utente, e.nome AS empresa FROM utentes u LEFT JOIN empresas e ON e.id = u.empresa_id WHERE u.ativo = 1 ORDER BY e.nome, u.nome; """) utentes = cursor.fetchall() cursor.close() ligacao.close() return utentes def criar_marcacao(utente_id, medico_id, data_marcacao, hora_marcacao, tipo_exame, observacoes, enfermeiro_id=None, local_consulta=None, convocatoria=None, clinica_parceira=None): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" INSERT INTO marcacoes ( utente_id, medico_id, data_marcacao, hora_marcacao, tipo_exame, estado, observacoes, enfermeiro_id, local_consulta, convocatoria, clinica_parceira ) VALUES (%s, %s, %s, %s, %s, 'Marcado', %s, %s, %s, %s, %s); """, ( utente_id, medico_id or None, data_marcacao, hora_marcacao, tipo_exame, observacoes, enfermeiro_id or None, local_consulta or None, convocatoria or None, clinica_parceira or None )) ligacao.commit() cursor.close() ligacao.close() return True, "Marcação criada com sucesso." def listar_marcacoes(data=None, medico_id=None, estado=None): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) filtros = [] params = [] if data: filtros.append("m.data_marcacao = %s") params.append(data) if medico_id: filtros.append("m.medico_id = %s") params.append(medico_id) if estado: filtros.append("m.estado = %s") params.append(estado) where = ("WHERE " + " AND ".join(filtros)) if filtros else "" cursor.execute(f""" SELECT m.id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, m.aptidao, m.observacoes, m.local_consulta, m.convocatoria, m.clinica_parceira, m.enfermeiro_id, u.nome AS utente, u.numero_utente, e.nome AS empresa, med.nome AS medico, enf.nome AS enfermeiro FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id LEFT JOIN utilizadores med ON med.id = m.medico_id LEFT JOIN utilizadores enf ON enf.id = m.enfermeiro_id {where} ORDER BY m.data_marcacao DESC, m.hora_marcacao DESC; """, params) marcacoes = cursor.fetchall() cursor.close() ligacao.close() return marcacoes def cancelar_marcacao(marcacao_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" UPDATE marcacoes SET estado = 'Cancelado' WHERE id = %s; """, (marcacao_id,)) ligacao.commit() cursor.close() ligacao.close() return True, "Marcação cancelada." def concluir_marcacao(marcacao_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" UPDATE marcacoes SET estado = 'Concluido' WHERE id = %s; """, (marcacao_id,)) ligacao.commit() cursor.close() ligacao.close() return True, "Marcação concluída." def obter_marcacao_completa(marcacao_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT m.id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, m.aptidao, m.observacoes, u.nome AS utente, u.numero_utente, u.sexo, u.data_nascimento, u.funcao, u.departamento, u.telefone, u.email, u.nif, e.nome AS empresa, med.nome AS medico FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id LEFT JOIN utilizadores med ON med.id = m.medico_id WHERE m.id = %s LIMIT 1; """, (marcacao_id,)) row = cursor.fetchone() cursor.close() ligacao.close() return row def gravar_nota_clinica(marcacao_id, observacoes): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" UPDATE marcacoes SET observacoes = %s WHERE id = %s; """, (observacoes, marcacao_id)) ligacao.commit() cursor.close() ligacao.close() return True def registar_aptidao(marcacao_id, aptidao, observacoes=None): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" UPDATE marcacoes SET aptidao = %s, estado = 'Concluido', observacoes = COALESCE(%s, observacoes) WHERE id = %s; """, (aptidao, observacoes, marcacao_id)) ligacao.commit() cursor.close() ligacao.close() return True, "Aptidão registada." def listar_agenda_dia(data, medico_id=None): """Devolve todas as marcações de um dia, opcionalmente filtradas por médico.""" ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) if medico_id: cursor.execute(""" SELECT m.id, m.hora_marcacao, m.tipo_exame, m.estado, m.aptidao, m.observacoes, m.medico_id, u.nome AS utente, u.numero_utente, e.nome AS empresa, med.nome AS medico FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id LEFT JOIN utilizadores med ON med.id = m.medico_id WHERE m.data_marcacao = %s AND m.medico_id = %s ORDER BY m.hora_marcacao ASC; """, (data, medico_id)) else: cursor.execute(""" SELECT m.id, m.hora_marcacao, m.tipo_exame, m.estado, m.aptidao, m.observacoes, m.medico_id, u.nome AS utente, u.numero_utente, e.nome AS empresa, med.nome AS medico FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id LEFT JOIN utilizadores med ON med.id = m.medico_id WHERE m.data_marcacao = %s ORDER BY med.nome ASC, m.hora_marcacao ASC; """, (data,)) rows = cursor.fetchall() cursor.close() ligacao.close() return rows def listar_notas_pendentes(): """Devolve marcações canceladas ou com observações de reagendamento.""" ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT m.id, m.data_marcacao, m.estado, m.observacoes, u.nome AS utente, e.nome AS empresa, med.nome AS medico FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id LEFT JOIN utilizadores med ON med.id = m.medico_id WHERE (m.estado IN ('Cancelado','Faltou') OR (m.observacoes IS NOT NULL AND m.observacoes != '')) ORDER BY m.data_marcacao DESC LIMIT 50; """) rows = cursor.fetchall() cursor.close() ligacao.close() return rows # ---------------- GESTÃO DE CONSUMÍVEIS / STOCK ---------------- def listar_fornecedores(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, nome, email, telefone, ativo FROM fornecedores WHERE ativo = 1 ORDER BY nome; """) fornecedores = cursor.fetchall() cursor.close() ligacao.close() return fornecedores def criar_fornecedor(nome, email="", telefone="", observacoes=""): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" INSERT INTO fornecedores (nome, email, telefone, observacoes, ativo) VALUES (%s, %s, %s, %s, 1); """, (nome, email, telefone, observacoes)) ligacao.commit() cursor.close() ligacao.close() return True, "Fornecedor criado com sucesso." def criar_consumivel(dados): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" INSERT INTO consumiveis ( codigo, categoria, nome_produto, unidade, quantidade_stock, stock_minimo, preco_unitario, tipo_embalagem, unidades_por_embalagem, fornecedor_id, ativo ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 1); """, ( dados.get("codigo"), dados.get("categoria"), dados.get("nome_produto"), dados.get("unidade") or "unidade", int(dados.get("quantidade_stock") or 0), int(dados.get("stock_minimo") or 0), float(dados.get("preco_unitario") or 0), dados.get("tipo_embalagem") or "unidade", int(dados.get("unidades_por_embalagem") or 1), dados.get("fornecedor_id") or None, )) ligacao.commit() cursor.close() ligacao.close() return True, "Consumível criado com sucesso." def listar_consumiveis(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT c.id, c.codigo, c.categoria, c.nome_produto, c.unidade, c.quantidade_stock, c.stock_minimo, c.preco_unitario, c.tipo_embalagem, c.unidades_por_embalagem, c.ativo, f.nome AS fornecedor FROM consumiveis c LEFT JOIN fornecedores f ON f.id = c.fornecedor_id WHERE c.ativo = 1 ORDER BY c.categoria, c.nome_produto; """) consumiveis = cursor.fetchall() cursor.close() ligacao.close() return consumiveis def listar_stock_baixo(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT c.id, c.codigo, c.categoria, c.nome_produto, c.unidade, c.quantidade_stock, c.stock_minimo, c.preco_unitario, c.tipo_embalagem, c.unidades_por_embalagem, f.nome AS fornecedor FROM consumiveis c LEFT JOIN fornecedores f ON f.id = c.fornecedor_id WHERE c.ativo = 1 AND c.quantidade_stock <= c.stock_minimo ORDER BY c.categoria, c.nome_produto; """) consumiveis = cursor.fetchall() cursor.close() ligacao.close() return consumiveis def obter_consumivel_por_id(consumivel_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, quantidade_stock FROM consumiveis WHERE id = %s LIMIT 1; """, (consumivel_id,)) consumivel = cursor.fetchone() cursor.close() ligacao.close() return consumivel def atualizar_stock_consumivel(consumivel_id, novo_stock, utilizador_id=None): consumivel = obter_consumivel_por_id(consumivel_id) if not consumivel: return False, "Consumível não encontrado." stock_antes = int(consumivel["quantidade_stock"]) novo_stock = int(novo_stock) diferenca = novo_stock - stock_antes ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" UPDATE consumiveis SET quantidade_stock = %s WHERE id = %s; """, (novo_stock, consumivel_id)) cursor.execute(""" INSERT INTO movimentos_stock ( consumivel_id, tipo_movimento, quantidade, stock_antes, stock_depois, utilizador_id, origem ) VALUES (%s, %s, %s, %s, %s, %s, %s); """, ( consumivel_id, "Atualização manual", diferenca, stock_antes, novo_stock, utilizador_id, "Gestão de Consumíveis Web" )) ligacao.commit() cursor.close() ligacao.close() return True, "Stock atualizado com sucesso." def desativar_consumivel(consumivel_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" UPDATE consumiveis SET ativo = 0 WHERE id = %s; """, (consumivel_id,)) ligacao.commit() cursor.close() ligacao.close() return True, "Consumível removido da lista ativa." # ---------------- CONSUMÍVEIS POR CONSULTA ---------------- def listar_consultas_do_medico(medico_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT m.id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, u.nome AS utente, u.numero_utente, u.sexo, u.data_nascimento, e.nome AS empresa FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id WHERE m.medico_id = %s AND m.estado IN ('Marcado','Marcada','Confirmada') ORDER BY m.data_marcacao ASC, m.hora_marcacao ASC; """, (medico_id,)) consultas = cursor.fetchall() cursor.close() ligacao.close() return consultas def listar_consultas_para_consumiveis(cargo, user_id): if cargo == "Medico": return listar_consultas_do_medico(user_id) ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT m.id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, u.nome AS utente, u.numero_utente, u.sexo, u.data_nascimento, e.nome AS empresa, med.nome AS medico FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id LEFT JOIN utilizadores med ON med.id = m.medico_id WHERE m.estado IN ('Marcado','Marcada','Confirmada') ORDER BY m.data_marcacao ASC, m.hora_marcacao ASC; """) consultas = cursor.fetchall() cursor.close() ligacao.close() return consultas def obter_marcacao_para_consumiveis(marcacao_id, cargo, user_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) if cargo == "Medico": cursor.execute(""" SELECT m.id, m.utente_id, m.medico_id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, u.nome AS utente, u.numero_utente, u.sexo, u.data_nascimento, e.nome AS empresa FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id WHERE m.id = %s AND m.medico_id = %s AND m.estado = 'Marcado' LIMIT 1; """, (marcacao_id, user_id)) else: cursor.execute(""" SELECT m.id, m.utente_id, m.medico_id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, u.nome AS utente, u.numero_utente, u.sexo, u.data_nascimento, e.nome AS empresa FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id WHERE m.id = %s AND m.estado = 'Marcado' LIMIT 1; """, (marcacao_id,)) marcacao = cursor.fetchone() cursor.close() ligacao.close() return marcacao def listar_consumiveis_para_consulta(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, categoria, nome_produto, unidade, quantidade_stock, stock_minimo, preco_unitario FROM consumiveis WHERE ativo = 1 ORDER BY categoria, nome_produto; """) consumiveis = cursor.fetchall() cursor.close() ligacao.close() return consumiveis def gravar_consumiveis_consulta(marcacao_id, itens, user_id, cargo): marcacao = obter_marcacao_para_consumiveis(marcacao_id, cargo, user_id) if not marcacao: return False, "Marcação não encontrada, sem permissão ou já concluída." if not itens: return False, "Adicione pelo menos um consumível." ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) try: for item in itens: consumivel_id = int(item["consumivel_id"]) quantidade = int(item["quantidade"]) if quantidade <= 0: raise Exception("A quantidade deve ser maior que zero.") cursor.execute(""" SELECT id, nome_produto, unidade, quantidade_stock, preco_unitario FROM consumiveis WHERE id = %s AND ativo = 1 LIMIT 1; """, (consumivel_id,)) consumivel = cursor.fetchone() if not consumivel: raise Exception("Consumível não encontrado.") stock_antes = int(consumivel["quantidade_stock"]) if quantidade > stock_antes: raise Exception(f"Stock insuficiente para {consumivel['nome_produto']}. Stock atual: {stock_antes}.") stock_depois = stock_antes - quantidade cursor.execute(""" INSERT INTO consumiveis_usados_consulta ( marcacao_id, utente_id, medico_id, consumivel_id, quantidade_usada, unidade, preco_unitario ) VALUES (%s, %s, %s, %s, %s, %s, %s); """, ( marcacao["id"], marcacao["utente_id"], marcacao["medico_id"], consumivel_id, quantidade, consumivel["unidade"], consumivel["preco_unitario"] )) cursor.execute(""" UPDATE consumiveis SET quantidade_stock = %s WHERE id = %s; """, (stock_depois, consumivel_id)) cursor.execute(""" INSERT INTO movimentos_stock ( consumivel_id, tipo_movimento, quantidade, stock_antes, stock_depois, utilizador_id, origem ) VALUES (%s, %s, %s, %s, %s, %s, %s); """, ( consumivel_id, "Saída por consulta", -quantidade, stock_antes, stock_depois, user_id, f"Consulta #{marcacao['id']}" )) cursor.execute(""" UPDATE marcacoes SET estado = 'Concluido' WHERE id = %s; """, (marcacao["id"],)) ligacao.commit() except Exception as erro: ligacao.rollback() cursor.close() ligacao.close() return False, str(erro) cursor.close() ligacao.close() return True, "Consumíveis registados e consulta concluída com sucesso." def listar_historico_consumiveis_consulta(cargo, user_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) filtro = "" params = [] if cargo == "Medico": filtro = "WHERE m.medico_id = %s" params.append(user_id) cursor.execute(f""" SELECT cuc.id, cuc.data_registo, cuc.quantidade_usada, cuc.unidade, cuc.preco_unitario, con.nome_produto, m.id AS marcacao_id, u.nome AS utente, e.nome AS empresa, med.nome AS medico FROM consumiveis_usados_consulta cuc INNER JOIN consumiveis con ON con.id = cuc.consumivel_id INNER JOIN marcacoes m ON m.id = cuc.marcacao_id INNER JOIN utentes u ON u.id = cuc.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id LEFT JOIN utilizadores med ON med.id = cuc.medico_id {filtro} ORDER BY cuc.data_registo DESC; """, tuple(params)) historico = cursor.fetchall() cursor.close() ligacao.close() return historico def listar_consumos_para_exportar(cargo, user_id, ano, mes): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) filtro_cargo = "" # strftime returns zero-padded strings, so format mes accordingly params = [str(ano), str(mes).zfill(2)] if cargo == "Medico": filtro_cargo = "AND cuc.medico_id = %s" params.append(user_id) cursor.execute(f""" SELECT strftime('%d/%m/%Y', cuc.data_registo) AS data, u.nome AS utente, e.nome AS empresa, med.nome AS medico, con.nome_produto AS produto, cuc.quantidade_usada AS quantidade, cuc.unidade, cuc.preco_unitario AS preco_unitario FROM consumiveis_usados_consulta cuc INNER JOIN consumiveis con ON con.id = cuc.consumivel_id INNER JOIN marcacoes m ON m.id = cuc.marcacao_id INNER JOIN utentes u ON u.id = cuc.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id LEFT JOIN utilizadores med ON med.id = cuc.medico_id WHERE strftime('%Y', cuc.data_registo) = %s AND strftime('%m', cuc.data_registo) = %s {filtro_cargo} ORDER BY cuc.data_registo ASC; """, tuple(params)) rows = cursor.fetchall() cursor.close() ligacao.close() return rows def pesquisar_empresas_utentes(termo=""): termo_like = f"%{termo}%" ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT e.id, e.nome, e.nif, e.telefone, e.email, COUNT(u.id) AS total_utentes FROM empresas e LEFT JOIN utentes u ON u.empresa_id = e.id AND u.ativo = 1 WHERE e.ativo = 1 AND (%s = '' OR e.nome LIKE %s OR e.nif LIKE %s) GROUP BY e.id, e.nome, e.nif, e.telefone, e.email ORDER BY e.nome; """, (termo, termo_like, termo_like)) empresas = cursor.fetchall() cursor.close() ligacao.close() return empresas def listar_utentes_por_empresa(empresa_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT u.id, u.numero_utente, u.nome, u.data_nascimento, u.sexo, u.telefone, u.email, u.departamento, u.funcao, u.tipo_exame FROM utentes u WHERE u.empresa_id = %s AND u.ativo = 1 ORDER BY u.nome; """, (empresa_id,)) utentes = cursor.fetchall() cursor.close() ligacao.close() return utentes def pesquisar_utentes_global(termo=""): termo_like = f"%{termo}%" ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT u.id, u.numero_utente, u.nome, u.data_nascimento, u.sexo, u.telefone, u.email, u.departamento, u.funcao, u.tipo_exame, e.nome AS empresa FROM utentes u LEFT JOIN empresas e ON e.id = u.empresa_id WHERE u.ativo = 1 AND ( %s = '' OR u.nome LIKE %s OR u.numero_utente LIKE %s OR u.nif LIKE %s OR e.nome LIKE %s ) ORDER BY e.nome, u.nome; """, (termo, termo_like, termo_like, termo_like, termo_like)) utentes = cursor.fetchall() cursor.close() ligacao.close() return utentes def obter_resumo_empresa(empresa_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT e.id, e.nome, COUNT(DISTINCT u.id) AS total_utentes, COUNT(DISTINCT m.id) AS total_consultas, COALESCE(SUM(cuc.quantidade_usada), 0) AS total_unidades, COALESCE(SUM(cuc.quantidade_usada * cuc.preco_unitario), 0) AS custo_total FROM empresas e LEFT JOIN utentes u ON u.empresa_id = e.id AND u.ativo = 1 LEFT JOIN marcacoes m ON m.utente_id = u.id LEFT JOIN consumiveis_usados_consulta cuc ON cuc.marcacao_id = m.id WHERE e.id = %s GROUP BY e.id, e.nome; """, (empresa_id,)) resumo = cursor.fetchone() cursor.close() ligacao.close() return resumo def obter_detalhe_utente(utente_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT u.*, e.nome AS empresa FROM utentes u LEFT JOIN empresas e ON e.id = u.empresa_id WHERE u.id = %s LIMIT 1; """, (utente_id,)) utente = cursor.fetchone() cursor.close() ligacao.close() return utente def listar_consultas_utente(utente_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT m.id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, med.nome AS medico, COALESCE(SUM(cuc.quantidade_usada), 0) AS total_unidades, COALESCE(SUM(cuc.quantidade_usada * cuc.preco_unitario), 0) AS custo_total FROM marcacoes m LEFT JOIN utilizadores med ON med.id = m.medico_id LEFT JOIN consumiveis_usados_consulta cuc ON cuc.marcacao_id = m.id WHERE m.utente_id = %s GROUP BY m.id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, med.nome ORDER BY m.data_marcacao DESC, m.hora_marcacao DESC; """, (utente_id,)) consultas = cursor.fetchall() cursor.close() ligacao.close() return consultas def listar_consumos_utente(utente_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT cuc.data_registo, cuc.quantidade_usada, cuc.unidade, cuc.preco_unitario, c.nome_produto, c.categoria, m.id AS marcacao_id, med.nome AS medico FROM consumiveis_usados_consulta cuc INNER JOIN consumiveis c ON c.id = cuc.consumivel_id INNER JOIN marcacoes m ON m.id = cuc.marcacao_id LEFT JOIN utilizadores med ON med.id = cuc.medico_id WHERE cuc.utente_id = %s ORDER BY cuc.data_registo DESC; """, (utente_id,)) consumos = cursor.fetchall() cursor.close() ligacao.close() return consumos # ---------------- HISTÓRICO GERAL ---------------- def listar_historico_movimentos_stock(limite=200): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT ms.id, ms.data_movimento, ms.tipo_movimento, ms.quantidade, ms.stock_antes, ms.stock_depois, ms.origem, c.nome_produto, c.categoria, u.nome AS utilizador, u.cargo FROM movimentos_stock ms INNER JOIN consumiveis c ON c.id = ms.consumivel_id LEFT JOIN utilizadores u ON u.id = ms.utilizador_id ORDER BY ms.data_movimento DESC LIMIT %s; """, (limite,)) movimentos = cursor.fetchall() cursor.close() ligacao.close() return movimentos def listar_historico_consumos_geral(limite=200): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT cuc.id, cuc.data_registo, cuc.quantidade_usada, cuc.unidade, cuc.preco_unitario, c.nome_produto, c.categoria, m.id AS marcacao_id, u.nome AS utente, e.nome AS empresa, med.nome AS medico FROM consumiveis_usados_consulta cuc INNER JOIN consumiveis c ON c.id = cuc.consumivel_id INNER JOIN marcacoes m ON m.id = cuc.marcacao_id INNER JOIN utentes u ON u.id = cuc.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id LEFT JOIN utilizadores med ON med.id = cuc.medico_id ORDER BY cuc.data_registo DESC LIMIT %s; """, (limite,)) consumos = cursor.fetchall() cursor.close() ligacao.close() return consumos def listar_historico_marcacoes(limite=200): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT m.id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, m.criado_em, u.nome AS utente, e.nome AS empresa, med.nome AS medico FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id LEFT JOIN utilizadores med ON med.id = m.medico_id ORDER BY m.criado_em DESC, m.data_marcacao DESC LIMIT %s; """, (limite,)) marcacoes = cursor.fetchall() cursor.close() ligacao.close() return marcacoes def obter_resumo_historico(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) resumo = {} cursor.execute("SELECT COUNT(*) AS total FROM utilizadores WHERE ativo = 1;") resumo["utilizadores"] = cursor.fetchone()["total"] cursor.execute("SELECT COUNT(*) AS total FROM utentes WHERE ativo = 1;") resumo["utentes"] = cursor.fetchone()["total"] cursor.execute("SELECT COUNT(*) AS total FROM empresas WHERE ativo = 1;") resumo["empresas"] = cursor.fetchone()["total"] cursor.execute("SELECT COUNT(*) AS total FROM consumiveis WHERE ativo = 1;") resumo["consumiveis"] = cursor.fetchone()["total"] cursor.execute("SELECT COUNT(*) AS total FROM marcacoes;") resumo["marcacoes"] = cursor.fetchone()["total"] cursor.execute("SELECT COUNT(*) AS total FROM consumiveis_usados_consulta;") resumo["consumos"] = cursor.fetchone()["total"] cursor.execute("SELECT COUNT(*) AS total FROM movimentos_stock;") resumo["movimentos_stock"] = cursor.fetchone()["total"] cursor.close() ligacao.close() return resumo # ---------------- PERFIL DO CEO ---------------- def obter_utilizador_por_username(username): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, nome, username, password_hash, cargo, ativo FROM utilizadores WHERE username = %s LIMIT 1; """, (username,)) utilizador = cursor.fetchone() cursor.close() ligacao.close() return utilizador def username_em_uso_por_outro(username, user_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id FROM utilizadores WHERE username = %s AND id <> %s LIMIT 1; """, (username, user_id)) existe = cursor.fetchone() cursor.close() ligacao.close() return existe is not None def atualizar_perfil_ceo(user_id, novo_nome, novo_username, password_atual="", nova_password="", confirmar_password=""): utilizador = obter_utilizador_por_id(user_id) if not utilizador: return False, "Utilizador não encontrado." if utilizador["cargo"] != "CEO": return False, "Esta área é exclusiva do CEO." if not novo_nome: return False, "O nome não pode ficar vazio." if not novo_username: return False, "O username não pode ficar vazio." if " " in novo_username: return False, "O username não deve ter espaços." if username_em_uso_por_outro(novo_username, user_id): return False, "Esse username já está em uso." quer_mudar_password = bool(password_atual or nova_password or confirmar_password) ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) if quer_mudar_password: cursor.execute(""" SELECT password_hash FROM utilizadores WHERE id = %s LIMIT 1; """, (user_id,)) row = cursor.fetchone() if not row: cursor.close() ligacao.close() return False, "Utilizador não encontrado." if not password_atual or not nova_password or not confirmar_password: cursor.close() ligacao.close() return False, "Para alterar a password, preencha todos os campos de password." if gerar_hash(password_atual) != row["password_hash"]: cursor.close() ligacao.close() return False, "A password atual está incorreta." if nova_password != confirmar_password: cursor.close() ligacao.close() return False, "A nova password e a confirmação não coincidem." if len(nova_password) < 4: cursor.close() ligacao.close() return False, "A nova password deve ter pelo menos 4 caracteres." cursor.execute(""" UPDATE utilizadores SET nome = %s, username = %s, password_hash = %s WHERE id = %s AND cargo = 'CEO'; """, (novo_nome, novo_username, gerar_hash(nova_password), user_id)) else: cursor.execute(""" UPDATE utilizadores SET nome = %s, username = %s WHERE id = %s AND cargo = 'CEO'; """, (novo_nome, novo_username, user_id)) ligacao.commit() cursor.close() ligacao.close() return True, "Perfil do CEO atualizado com sucesso." # ---------------- FORNECEDORES ---------------- def listar_fornecedores_com_totais(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT f.id, f.nome, f.email, f.telefone, f.observacoes, f.ativo, f.criado_em, COUNT(c.id) AS total_consumiveis FROM fornecedores f LEFT JOIN consumiveis c ON c.fornecedor_id = f.id AND c.ativo = 1 WHERE f.ativo = 1 GROUP BY f.id, f.nome, f.email, f.telefone, f.observacoes, f.ativo, f.criado_em ORDER BY f.nome; """) fornecedores = cursor.fetchall() cursor.close() ligacao.close() return fornecedores def atualizar_fornecedor(fornecedor_id, nome, email, telefone, observacoes): if not nome: return False, "O nome do fornecedor é obrigatório." ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" UPDATE fornecedores SET nome = %s, email = %s, telefone = %s, observacoes = %s WHERE id = %s; """, (nome, email, telefone, observacoes, fornecedor_id)) ligacao.commit() cursor.close() ligacao.close() return True, "Fornecedor atualizado com sucesso." def obter_fornecedor_por_id(fornecedor_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, nome, email, telefone, observacoes, ativo FROM fornecedores WHERE id = %s LIMIT 1; """, (fornecedor_id,)) fornecedor = cursor.fetchone() cursor.close() ligacao.close() return fornecedor def desativar_fornecedor(fornecedor_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" UPDATE fornecedores SET ativo = 0 WHERE id = %s; """, (fornecedor_id,)) ligacao.commit() cursor.close() ligacao.close() return True, "Fornecedor removido da lista ativa." # ---------------- AS MINHAS CONSULTAS / ÁREA MÉDICA ---------------- def listar_minhas_consultas(medico_id, estado=""): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) if estado: cursor.execute(""" SELECT m.id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, m.aptidao, m.observacoes, u.nome AS utente, u.numero_utente, u.sexo, u.data_nascimento, u.telefone, e.nome AS empresa FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id WHERE m.medico_id = %s AND m.estado = %s ORDER BY m.data_marcacao ASC, m.hora_marcacao ASC; """, (medico_id, estado)) else: cursor.execute(""" SELECT m.id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, m.aptidao, m.observacoes, u.nome AS utente, u.numero_utente, u.sexo, u.data_nascimento, u.telefone, e.nome AS empresa FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id WHERE m.medico_id = %s ORDER BY m.data_marcacao DESC, m.hora_marcacao DESC; """, (medico_id,)) consultas = cursor.fetchall() cursor.close() ligacao.close() return consultas def obter_resumo_minhas_consultas(medico_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT SUM(CASE WHEN estado IN ('Marcado','Marcada','Confirmada') THEN 1 ELSE 0 END) AS marcadas, SUM(CASE WHEN estado IN ('Concluido','Concluído','Realizada') THEN 1 ELSE 0 END) AS concluidas, SUM(CASE WHEN estado IN ('Cancelado','Cancelada','Faltou') THEN 1 ELSE 0 END) AS canceladas, COUNT(*) AS total FROM marcacoes WHERE medico_id = %s; """, (medico_id,)) resumo = cursor.fetchone() cursor.close() ligacao.close() if not resumo: return {"marcadas": 0, "concluidas": 0, "canceladas": 0, "total": 0} return { "marcadas": resumo["marcadas"] or 0, "concluidas": resumo["concluidas"] or 0, "canceladas": resumo["canceladas"] or 0, "total": resumo["total"] or 0, } def obter_detalhe_consulta_medico(marcacao_id, medico_id): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT m.id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, m.aptidao, m.observacoes, u.nome AS utente, u.numero_utente, u.sexo, u.data_nascimento, u.telefone, u.email, u.departamento, u.funcao, e.nome AS empresa FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id WHERE m.id = %s AND m.medico_id = %s LIMIT 1; """, (marcacao_id, medico_id)) consulta = cursor.fetchone() cursor.close() ligacao.close() return consulta # ---------------- DADOS INICIAIS E LOGIN MÉDICO SIMPLES ---------------- CONSUMIVEIS_INICIAIS = [ ("1. Materiais de Proteção Individual (EPIs)", "Máscaras N95", "unidade", 0, 10), ("1. Materiais de Proteção Individual (EPIs)", "Máscaras cirúrgicas", "unidade", 0, 20), ("1. Materiais de Proteção Individual (EPIs)", "Batas descartáveis", "unidade", 0, 10), ("1. Materiais de Proteção Individual (EPIs)", "Luvas descartáveis", "unidade", 0, 50), ("1. Materiais de Proteção Individual (EPIs)", "Óculos de proteção", "unidade", 0, 5), ("1. Materiais de Proteção Individual (EPIs)", "Pezinhos descartáveis", "unidade", 0, 20), ("2. Materiais de Emergência", "Kit de primeiros socorros", "unidade", 0, 2), ("2. Materiais de Emergência", "Ligaduras", "unidade", 0, 10), ("2. Materiais de Emergência", "Compressas", "unidade", 0, 20), ("2. Materiais de Emergência", "Antissépticos - álcool", "unidade", 0, 5), ("2. Materiais de Emergência", "Antissépticos - soluções iodadas", "unidade", 0, 5), ("2. Materiais de Emergência", "Paracetamol", "unidade", 0, 10), ("2. Materiais de Emergência", "Anti-inflamatório", "unidade", 0, 10), ("3. Materiais de Avaliação", "Pilhas AA para esfigmomanómetro", "unidade", 0, 8), ("3. Materiais de Avaliação", "Pilhas AAA para esfigmomanómetro", "unidade", 0, 8), ("4. Materiais para Exames", "Algodão", "unidade", 0, 10), ("4. Materiais para Exames", "Lancetas", "unidade", 0, 20), ("4. Materiais para Exames", "Copos de urina", "unidade", 0, 20), ("4. Materiais para Exames", "Tiras de colesterol", "unidade", 0, 20), ("4. Materiais para Exames", "Tiras de glicémia", "unidade", 0, 20), ("4. Materiais para Exames", "Tiras de urina tipo II", "unidade", 0, 20), ("4. Materiais para Exames", "Pensos rápidos", "unidade", 0, 20), ("4. Materiais para Exames", "Rolo de eletrocardiograma Cardioline", "unidade", 0, 3), ("4. Materiais para Exames", "Rolo de eletrocardiograma BTL", "unidade", 0, 3), ("4. Materiais para Exames", "Rolo de marquesa", "unidade", 0, 5), ("4.1 Outros Exames", "Testes de álcool", "unidade", 0, 10), ("4.1 Outros Exames", "Testes de drogas", "unidade", 0, 10), ("4.1 Outros Exames", "Testes de espirometria", "unidade", 0, 10), ("5. Desinfetantes", "Álcool", "unidade", 0, 5), ("5. Desinfetantes", "Gel para ECG", "unidade", 0, 5), ("7. Suprimentos Gerais", "Contentor perfurocortantes", "unidade", 0, 2), ("7. Suprimentos Gerais", "Sacos de lixo hospitalar", "unidade", 0, 20), ("7. Suprimentos Gerais", "Etiquetas e adesivos", "unidade", 0, 20), ("Produtos Gerais", "Detergente multiusos", "unidade", 0, 5), ("Produtos Gerais", "Detergente desinfetante", "unidade", 0, 5), ("Produtos Gerais", "Detergente WC", "unidade", 0, 5), ("Produtos Gerais", "Desinfetante", "unidade", 0, 5), ("Produtos Gerais", "Lixívia", "unidade", 0, 5), ("Produtos Gerais", "Limpa vidros", "unidade", 0, 5), ("Produtos Gerais", "Luvas", "unidade", 0, 20), ("Higiene Sanitária", "Papel Zig Zag", "unidade", 0, 10), ("Higiene Sanitária", "Papel higiénico", "unidade", 0, 10), ("Higiene Sanitária", "Sabonete líquido", "unidade", 0, 5), ("Higiene Sanitária", "Desinfetante sanitário", "unidade", 0, 5), ("Higiene Sanitária", "Ambientador", "unidade", 0, 5), ("Panos e Esponjas", "Panos variados por área", "unidade", 0, 10), ("Panos e Esponjas", "Panos reutilizáveis para limpeza geral", "unidade", 0, 10), ("Panos e Esponjas", "Panos ou esponjas descartáveis", "unidade", 0, 10), ("Coleta de Resíduos e Descartáveis", "Sacos de lixo 30L", "unidade", 0, 20), ("Coleta de Resíduos e Descartáveis", "Sacos de lixo 50L", "unidade", 0, 20), ("Limpeza de Pisos", "MOPA profissional", "unidade", 0, 2), ("Limpeza de Pisos", "Balde com ou sem espremedor", "unidade", 0, 2), ("Limpeza de Pisos", "Esfregona", "unidade", 0, 3), ("Limpeza de Pisos", "Vassoura", "unidade", 0, 3), ("Limpeza de Pisos", "Pá", "unidade", 0, 3), ("Lavagem de Loiça", "Esfregão", "unidade", 0, 10), ("Lavagem de Loiça", "Detergente desengordurante", "unidade", 0, 5), ("Lavagem de Loiça", "Rolo de papel", "unidade", 0, 10), ("Limpeza de Superfícies", "Fibras abrasivas e esponjas", "unidade", 0, 10), ("Limpeza de Superfícies", "Borrifadores e pulverizadores", "unidade", 0, 5), ] def garantir_consumiveis_iniciais(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute("SELECT COUNT(*) AS total FROM consumiveis;") total = cursor.fetchone()["total"] if total == 0: for categoria, nome, unidade, stock, minimo in CONSUMIVEIS_INICIAIS: cursor.execute(""" INSERT INTO consumiveis ( categoria, nome_produto, unidade, quantidade_stock, stock_minimo, preco_unitario, tipo_embalagem, unidades_por_embalagem, ativo ) VALUES (%s, %s, %s, %s, %s, 0.00, 'unidade', 1, 1); """, (categoria, nome, unidade, stock, minimo)) ligacao.commit() cursor.close() ligacao.close() def garantir_dados_iniciais(): criar_utilizador_ceo_inicial() garantir_consumiveis_iniciais() def autenticar_medico_por_nome(nome_ou_username): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, nome, username, cargo, ativo FROM utilizadores WHERE ativo = 1 AND cargo = 'Medico' AND (LOWER(nome) = LOWER(%s) OR LOWER(username) = LOWER(%s)) LIMIT 1; """, (nome_ou_username, nome_ou_username)) medico = cursor.fetchone() cursor.close() ligacao.close() if not medico: return None return { "id": medico["id"], "nome": medico["nome"], "username": medico["username"], "cargo": medico["cargo"], } def autenticar_enfermeiro_por_nome(nome_ou_username): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, nome, username, cargo, ativo FROM utilizadores WHERE ativo = 1 AND cargo = 'Enfermeiro' AND (LOWER(nome) = LOWER(%s) OR LOWER(username) = LOWER(%s)) LIMIT 1; """, (nome_ou_username, nome_ou_username)) enfermeiro = cursor.fetchone() cursor.close() ligacao.close() if not enfermeiro: return None return { "id": enfermeiro["id"], "nome": enfermeiro["nome"], "username": enfermeiro["username"], "cargo": enfermeiro["cargo"], } # ---------------- STOCK ORGANIZADO E EDITÁVEL ---------------- def garantir_colunas_stock_extra(): """Garante campos extra necessários para gestão empresarial de stock.""" ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'consumiveis'; """) colunas = {row["COLUMN_NAME"] for row in cursor.fetchall()} alteracoes = [] if "data_validade" not in colunas: alteracoes.append("ADD COLUMN data_validade DATE NULL") if "observacoes" not in colunas: alteracoes.append("ADD COLUMN observacoes TEXT NULL") if "referencia_fornecedor" not in colunas: alteracoes.append("ADD COLUMN referencia_fornecedor VARCHAR(120) NULL") if alteracoes: cursor.execute("ALTER TABLE consumiveis " + ", ".join(alteracoes) + ";") ligacao.commit() cursor.close() ligacao.close() def categorias_stock_disponiveis(): garantir_colunas_stock_extra() ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT categoria, COUNT(*) AS total FROM consumiveis WHERE ativo = 1 GROUP BY categoria ORDER BY categoria; """) categorias = cursor.fetchall() cursor.close() ligacao.close() return categorias def listar_consumiveis_por_categoria(): garantir_colunas_stock_extra() ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT c.id, c.codigo, c.categoria, c.nome_produto, c.unidade, c.quantidade_stock, c.stock_minimo, c.preco_unitario, c.tipo_embalagem, c.unidades_por_embalagem, c.data_validade, c.referencia_fornecedor, c.observacoes, c.ativo, f.nome AS fornecedor, f.id AS fornecedor_id FROM consumiveis c LEFT JOIN fornecedores f ON f.id = c.fornecedor_id WHERE c.ativo = 1 ORDER BY c.categoria, c.nome_produto; """) produtos = cursor.fetchall() cursor.close() ligacao.close() grupos = {} for produto in produtos: categoria = produto["categoria"] or "Sem categoria" grupos.setdefault(categoria, []).append(produto) return grupos def obter_consumivel_completo(consumivel_id): garantir_colunas_stock_extra() ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, codigo, categoria, nome_produto, unidade, quantidade_stock, stock_minimo, preco_unitario, tipo_embalagem, unidades_por_embalagem, data_validade, referencia_fornecedor, fornecedor_id, observacoes, ativo FROM consumiveis WHERE id = %s LIMIT 1; """, (consumivel_id,)) produto = cursor.fetchone() cursor.close() ligacao.close() return produto def atualizar_consumivel_completo(consumivel_id, dados, utilizador_id=None): garantir_colunas_stock_extra() produto_atual = obter_consumivel_completo(consumivel_id) if not produto_atual: return False, "Produto não encontrado." nome_produto = dados.get("nome_produto", "").strip() if not nome_produto: return False, "O nome do produto é obrigatório." try: novo_stock = int(dados.get("quantidade_stock") or 0) stock_minimo = int(dados.get("stock_minimo") or 0) unidades_por_embalagem = int(dados.get("unidades_por_embalagem") or 1) preco_unitario = float(str(dados.get("preco_unitario") or 0).replace(",", ".")) except ValueError: return False, "Stock, mínimo, unidades por caixa e preço têm de ser valores válidos." if unidades_por_embalagem <= 0: return False, "As unidades por caixa têm de ser maiores que zero." stock_antes = int(produto_atual["quantidade_stock"]) diferenca = novo_stock - stock_antes ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" UPDATE consumiveis SET codigo = %s, categoria = %s, nome_produto = %s, unidade = %s, quantidade_stock = %s, stock_minimo = %s, preco_unitario = %s, tipo_embalagem = %s, unidades_por_embalagem = %s, data_validade = %s, referencia_fornecedor = %s, fornecedor_id = %s, observacoes = %s WHERE id = %s; """, ( dados.get("codigo", "").strip(), dados.get("categoria", "").strip(), nome_produto, dados.get("unidade", "unidade").strip() or "unidade", novo_stock, stock_minimo, preco_unitario, dados.get("tipo_embalagem", "caixa").strip() or "caixa", unidades_por_embalagem, dados.get("data_validade") or None, dados.get("referencia_fornecedor", "").strip(), dados.get("fornecedor_id") or None, dados.get("observacoes", "").strip(), consumivel_id )) if diferenca != 0: cursor.execute(""" INSERT INTO movimentos_stock ( consumivel_id, tipo_movimento, quantidade, stock_antes, stock_depois, utilizador_id, origem ) VALUES (%s, %s, %s, %s, %s, %s, %s); """, ( consumivel_id, "Edição de produto", diferenca, stock_antes, novo_stock, utilizador_id, "Gestão de Stock Web" )) ligacao.commit() cursor.close() ligacao.close() return True, "Produto atualizado com sucesso." def calcular_caixas_para_repor(stock_atual, stock_minimo, unidades_por_caixa): stock_atual = int(stock_atual or 0) stock_minimo = int(stock_minimo or 0) unidades_por_caixa = int(unidades_por_caixa or 1) if stock_atual > stock_minimo: return 0 falta = max(stock_minimo - stock_atual, 0) if falta == 0: falta = unidades_por_caixa caixas = (falta + unidades_por_caixa - 1) // unidades_por_caixa return caixas # Override da criação de consumível com campos extra. def criar_consumivel(dados): garantir_colunas_stock_extra() ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor() cursor.execute(""" INSERT INTO consumiveis ( codigo, categoria, nome_produto, unidade, quantidade_stock, stock_minimo, preco_unitario, tipo_embalagem, unidades_por_embalagem, data_validade, referencia_fornecedor, fornecedor_id, observacoes, ativo ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 1); """, ( dados.get("codigo"), dados.get("categoria"), dados.get("nome_produto"), dados.get("unidade") or "unidade", int(dados.get("quantidade_stock") or 0), int(dados.get("stock_minimo") or 0), float(str(dados.get("preco_unitario") or 0).replace(",", ".")), dados.get("tipo_embalagem") or "caixa", int(dados.get("unidades_por_embalagem") or 1), dados.get("data_validade") or None, dados.get("referencia_fornecedor") or "", dados.get("fornecedor_id") or None, dados.get("observacoes") or "", )) ligacao.commit() cursor.close() ligacao.close() return True, "Consumível criado com sucesso." # ---------------- STOCK BAIXO ORGANIZADO + ALERTA ---------------- def contar_stock_baixo(): garantir_colunas_stock_extra() ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT COUNT(*) AS total FROM consumiveis WHERE ativo = 1 AND quantidade_stock <= stock_minimo; """) total = cursor.fetchone()["total"] or 0 cursor.close() ligacao.close() return total # Override da listagem de stock baixo com cálculo de caixas a pedir. def listar_stock_baixo(): garantir_colunas_stock_extra() ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT c.id, c.codigo, c.categoria, c.nome_produto, c.unidade, c.quantidade_stock, c.stock_minimo, c.preco_unitario, c.tipo_embalagem, c.unidades_por_embalagem, c.data_validade, c.referencia_fornecedor, f.nome AS fornecedor, f.email AS fornecedor_email FROM consumiveis c LEFT JOIN fornecedores f ON f.id = c.fornecedor_id WHERE c.ativo = 1 AND c.quantidade_stock <= c.stock_minimo ORDER BY c.categoria, c.nome_produto; """) produtos = cursor.fetchall() cursor.close() ligacao.close() for produto in produtos: stock_atual = int(produto["quantidade_stock"] or 0) stock_minimo = int(produto["stock_minimo"] or 0) unidades_por_caixa = int(produto["unidades_por_embalagem"] or 1) falta_unidades = max(stock_minimo - stock_atual, 0) # Se está exatamente no mínimo, sugere pelo menos 1 caixa. if falta_unidades == 0: caixas_a_pedir = 1 else: caixas_a_pedir = (falta_unidades + unidades_por_caixa - 1) // unidades_por_caixa produto["falta_unidades"] = falta_unidades produto["caixas_a_pedir"] = caixas_a_pedir return produtos # ---------------- EXCEL E EMAIL DE REPOSIÇÃO ---------------- def listar_stock_para_excel(): garantir_colunas_stock_extra() ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT c.codigo, c.categoria, c.nome_produto, c.unidade, c.quantidade_stock, c.stock_minimo, c.preco_unitario, c.tipo_embalagem, c.unidades_por_embalagem, c.data_validade, c.referencia_fornecedor, f.nome AS fornecedor, f.email AS fornecedor_email, c.observacoes FROM consumiveis c LEFT JOIN fornecedores f ON f.id = c.fornecedor_id WHERE c.ativo = 1 ORDER BY c.categoria, c.nome_produto; """) dados = cursor.fetchall() cursor.close() ligacao.close() return dados def listar_reposicao_por_fornecedor(): produtos = listar_stock_baixo() grupos = {} for p in produtos: fornecedor = p.get("fornecedor") or "Sem fornecedor" email = p.get("fornecedor_email") or "" chave = (fornecedor, email) if chave not in grupos: grupos[chave] = [] grupos[chave].append(p) resultado = [] for (fornecedor, email), itens in grupos.items(): linhas = [] for item in itens: linhas.append({ "produto": item["nome_produto"], "categoria": item.get("categoria") or "", "stock_atual": item["quantidade_stock"], "stock_minimo": item["stock_minimo"], "unidade": item["unidade"], "unidades_por_caixa": item.get("unidades_por_embalagem") or 1, "caixas_a_pedir": item.get("caixas_a_pedir") or 1, "referencia": item.get("referencia_fornecedor") or "", }) resultado.append({ "fornecedor": fornecedor, "email": email, "itens": linhas, }) return resultado def _valor_vazio_excel(valor): """Devolve True quando o valor vindo do Excel deve ser tratado como vazio.""" if valor is None: return True if isinstance(valor, str): v = valor.strip() return v == "" or v in {"-", "—", "--", "N/A", "n/a", "NA", "na"} return False def _texto_excel(valor, padrao=""): if _valor_vazio_excel(valor): return padrao return str(valor).strip() def _numero_excel(valor, padrao=0.0): """Converte números do Excel de forma segura. Aceita células numéricas, texto com vírgula decimal, símbolo €, espaços, e ignora células vazias em vez de rebentar com: could not convert string to float: ' ' """ if _valor_vazio_excel(valor): return padrao if isinstance(valor, (int, float)): return float(valor) texto = str(valor).strip() texto = texto.replace("€", "").replace(" ", "").replace("\u00a0", "") # Formato português: 1.234,56 -> 1234.56 if "," in texto and "." in texto: texto = texto.replace(".", "").replace(",", ".") else: texto = texto.replace(",", ".") try: return float(texto) except ValueError: return padrao def _inteiro_excel(valor, padrao=0): return int(round(_numero_excel(valor, padrao))) def _data_excel(valor): if _valor_vazio_excel(valor): return None try: # datetime/date do openpyxl if hasattr(valor, "strftime"): return valor.strftime("%Y-%m-%d") except Exception: pass return str(valor).strip() def _obter_linha(linha, *nomes, padrao=None): """Permite importar ficheiros com cabeçalhos ligeiramente diferentes.""" normalizados = { str(k).strip().lower().replace(" ", "_"): v for k, v in linha.items() if k is not None } for nome in nomes: chave = str(nome).strip().lower().replace(" ", "_") if chave in normalizados: return normalizados[chave] return padrao def importar_stock_linhas(linhas, utilizador_id=None): garantir_colunas_stock_extra() criados = 0 atualizados = 0 ignorados = 0 ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) for numero_linha, linha in enumerate(linhas, start=2): # Aceita o ficheiro exportado pelo sistema e também cabeçalhos mais "humanos". nome = _texto_excel(_obter_linha( linha, "nome_produto", "nome produto", "produto", "nome", "consumivel", "consumível" )) if not nome: ignorados += 1 continue codigo = _texto_excel(_obter_linha(linha, "codigo", "código", "cod", "ref", "referencia", "referência")) categoria = _texto_excel(_obter_linha(linha, "categoria", "família", "familia")) unidade = _texto_excel(_obter_linha(linha, "unidade", "un"), "unidade") or "unidade" quantidade_stock = _inteiro_excel(_obter_linha( linha, "quantidade_stock", "quantidade stock", "stock", "quantidade", "qtd", "qt" ), 0) stock_minimo = _inteiro_excel(_obter_linha( linha, "stock_minimo", "stock mínimo", "stock minimo", "minimo", "mínimo" ), 0) preco_unitario = _numero_excel(_obter_linha( linha, "preco_unitario", "preço_unitário", "preco unitario", "preço unitário", "preco", "preço" ), 0.0) tipo_embalagem = _texto_excel(_obter_linha( linha, "tipo_embalagem", "tipo embalagem", "embalagem" ), "caixa") or "caixa" unidades_por_embalagem = _inteiro_excel(_obter_linha( linha, "unidades_por_embalagem", "unidades por embalagem", "unidades_caixa", "unidades por caixa" ), 1) or 1 data_validade = _data_excel(_obter_linha(linha, "data_validade", "data validade", "validade")) referencia_fornecedor = _texto_excel(_obter_linha( linha, "referencia_fornecedor", "referência_fornecedor", "referencia fornecedor", "referência fornecedor" )) observacoes = _texto_excel(_obter_linha(linha, "observacoes", "observações", "obs")) cursor.execute(""" SELECT id, quantidade_stock FROM consumiveis WHERE ativo = 1 AND ( (%s <> '' AND codigo = %s) OR nome_produto = %s ) LIMIT 1; """, (codigo, codigo, nome)) existente = cursor.fetchone() if existente: stock_antes = int(existente["quantidade_stock"] or 0) cursor.execute(""" UPDATE consumiveis SET codigo = %s, categoria = %s, nome_produto = %s, unidade = %s, quantidade_stock = %s, stock_minimo = %s, preco_unitario = %s, tipo_embalagem = %s, unidades_por_embalagem = %s, data_validade = %s, referencia_fornecedor = %s, observacoes = %s WHERE id = %s; """, ( codigo, categoria, nome, unidade, quantidade_stock, stock_minimo, preco_unitario, tipo_embalagem, unidades_por_embalagem, data_validade, referencia_fornecedor, observacoes, existente["id"], )) if quantidade_stock != stock_antes: cursor.execute(""" INSERT INTO movimentos_stock ( consumivel_id, tipo_movimento, quantidade, stock_antes, stock_depois, utilizador_id, origem ) VALUES (%s, %s, %s, %s, %s, %s, %s); """, ( existente["id"], "Importação Excel", quantidade_stock - stock_antes, stock_antes, quantidade_stock, utilizador_id, "Importação de ficheiro Excel" )) atualizados += 1 else: cursor.execute(""" INSERT INTO consumiveis ( codigo, categoria, nome_produto, unidade, quantidade_stock, stock_minimo, preco_unitario, tipo_embalagem, unidades_por_embalagem, data_validade, referencia_fornecedor, observacoes, ativo ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 1); """, ( codigo, categoria, nome, unidade, quantidade_stock, stock_minimo, preco_unitario, tipo_embalagem, unidades_por_embalagem, data_validade, referencia_fornecedor, observacoes, )) novo_id = cursor.lastrowid cursor.execute(""" INSERT INTO movimentos_stock ( consumivel_id, tipo_movimento, quantidade, stock_antes, stock_depois, utilizador_id, origem ) VALUES (%s, %s, %s, %s, %s, %s, %s); """, ( novo_id, "Criação por Excel", quantidade_stock, 0, quantidade_stock, utilizador_id, "Importação de ficheiro Excel" )) criados += 1 ligacao.commit() cursor.close() ligacao.close() return { "criados": criados, "atualizados": atualizados, "ignorados": ignorados, } # ---------------- MELHORIAS GERAIS PRO ---------------- def listar_produtos_validade_proxima(dias=60): garantir_colunas_stock_extra() ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT id, codigo, categoria, nome_produto, quantidade_stock, stock_minimo, unidade, data_validade FROM consumiveis WHERE ativo = 1 AND data_validade IS NOT NULL AND data_validade <= DATE_ADD(CURDATE(), INTERVAL %s DAY) ORDER BY data_validade ASC, nome_produto ASC LIMIT 50; """, (dias,)) dados = cursor.fetchall() cursor.close() ligacao.close() return dados def listar_produtos_mais_consumidos(limit=10): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT c.nome_produto, c.categoria, c.unidade, SUM(cc.quantidade_usada) AS total_usado FROM consumiveis_consulta cc INNER JOIN consumiveis c ON c.id = cc.consumivel_id GROUP BY c.id, c.nome_produto, c.categoria, c.unidade ORDER BY total_usado DESC LIMIT %s; """, (limit,)) dados = cursor.fetchall() cursor.close() ligacao.close() return dados def listar_ultimos_consumos_dashboard(limit=8): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT cc.data_registo, cc.quantidade_usada, c.nome_produto, c.unidade, u.nome AS utente, e.nome AS empresa, med.nome AS medico FROM consumiveis_consulta cc INNER JOIN consumiveis c ON c.id = cc.consumivel_id INNER JOIN marcacoes m ON m.id = cc.marcacao_id INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id LEFT JOIN utilizadores med ON med.id = m.medico_id ORDER BY cc.data_registo DESC LIMIT %s; """, (limit,)) dados = cursor.fetchall() cursor.close() ligacao.close() return dados def obter_resumo_dashboard_ceo(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) resumo = {} consultas = [ ("utentes", "SELECT COUNT(*) AS total FROM utentes WHERE ativo = 1"), ("consultas_hoje", "SELECT COUNT(*) AS total FROM marcacoes WHERE data_marcacao = CURDATE()"), ("stock_baixo", "SELECT COUNT(*) AS total FROM consumiveis WHERE ativo = 1 AND quantidade_stock <= stock_minimo"), ("produtos", "SELECT COUNT(*) AS total FROM consumiveis WHERE ativo = 1"), ("fornecedores", "SELECT COUNT(*) AS total FROM fornecedores WHERE ativo = 1"), ("movimentos", "SELECT COUNT(*) AS total FROM movimentos_stock"), ] for chave, sql in consultas: cursor.execute(sql) resumo[chave] = cursor.fetchone()["total"] or 0 cursor.close() ligacao.close() return resumo def listar_movimentos_stock_completo(limit=250, termo="", tipo=""): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) filtros = [] params = [] if termo: filtros.append("(c.nome_produto LIKE %s OR c.codigo LIKE %s OR m.origem LIKE %s)") like = f"%{termo}%" params.extend([like, like, like]) if tipo: filtros.append("m.tipo_movimento = %s") params.append(tipo) where_sql = "" if filtros: where_sql = "WHERE " + " AND ".join(filtros) params.append(limit) cursor.execute(f""" SELECT m.id, m.data_movimento, m.tipo_movimento, m.quantidade, m.stock_antes, m.stock_depois, m.origem, c.codigo, c.nome_produto, c.categoria, c.unidade, u.nome AS utilizador FROM movimentos_stock m INNER JOIN consumiveis c ON c.id = m.consumivel_id LEFT JOIN utilizadores u ON u.id = m.utilizador_id {where_sql} ORDER BY m.data_movimento DESC LIMIT %s; """, tuple(params)) dados = cursor.fetchall() cursor.close() ligacao.close() return dados def listar_tipos_movimento_stock(): ligacao = ligar_mysql(usar_base=True) cursor = ligacao.cursor(dictionary=True) cursor.execute(""" SELECT DISTINCT tipo_movimento FROM movimentos_stock WHERE tipo_movimento IS NOT NULL AND tipo_movimento <> '' ORDER BY tipo_movimento; """) dados = cursor.fetchall() cursor.close() ligacao.close() return [d["tipo_movimento"] for d in dados] # Compatibilidade com os nomes usados no app.py após a conversão para SQLite testar_ligacao_sqlite = testar_ligacao_mysql executar_schema_sqlite = executar_schema_mysql # ---------------- GESTÃO DE UTILIZADORES / CARGOS ---------------- def ligar_db(): import os os.makedirs("database", exist_ok=True) conn = sqlite3.connect("database/cofihst.sqlite3") conn.row_factory = sqlite3.Row return conn # ============================================================ # GESTÃO DE UTILIZADORES, CARGOS E GESTOR DE EQUIPA - FINAL # ============================================================ def garantir_tabela_utilizadores_base(): conn = ligar_db() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS utilizadores ( id INTEGER PRIMARY KEY AUTOINCREMENT, nome TEXT NOT NULL, username TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, cargo TEXT NOT NULL DEFAULT 'Medico', ativo INTEGER DEFAULT 1, criado_em TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) cur.execute("SELECT id FROM utilizadores WHERE username = 'ceo'") if not cur.fetchone(): cur.execute(""" INSERT INTO utilizadores (nome, username, password_hash, cargo, ativo) VALUES (?, ?, ?, ?, 1); """, ("CEO", "ceo", gerar_hash("1234"), "CEO")) conn.commit() conn.close() def garantir_tabelas_gestao_equipa(): garantir_tabela_utilizadores_base() conn = ligar_db() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS equipas ( id INTEGER PRIMARY KEY AUTOINCREMENT, nome TEXT NOT NULL UNIQUE, descricao TEXT, gestor_id INTEGER, ativo INTEGER DEFAULT 1, criado_em TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (gestor_id) REFERENCES utilizadores(id) ); """) cur.execute(""" CREATE TABLE IF NOT EXISTS equipa_membros ( id INTEGER PRIMARY KEY AUTOINCREMENT, equipa_id INTEGER NOT NULL, utilizador_id INTEGER NOT NULL, ativo INTEGER DEFAULT 1, criado_em TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (equipa_id) REFERENCES equipas(id), FOREIGN KEY (utilizador_id) REFERENCES utilizadores(id), UNIQUE(equipa_id, utilizador_id) ); """) conn.commit() conn.close() def listar_utilizadores_admin(): garantir_tabelas_gestao_equipa() conn = ligar_db() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute(""" SELECT id, nome, username, cargo, ativo FROM utilizadores ORDER BY CASE cargo WHEN 'CEO' THEN 1 WHEN 'Admin' THEN 2 WHEN 'Gestor' THEN 3 WHEN 'GestorEquipa' THEN 4 WHEN 'Medico' THEN 5 ELSE 6 END, nome ASC; """) dados = [dict(r) for r in cur.fetchall()] conn.close() return dados def criar_utilizador_admin(nome, username, password, cargo): garantir_tabelas_gestao_equipa() nome = (nome or "").strip() username = (username or "").strip().lower() password = (password or "").strip() cargo = (cargo or "").strip() if not nome or not username or not cargo: return False, "Nome, username e cargo são obrigatórios." cargos_validos = ["Admin", "Gestor", "GestorEquipa", "Medico", "Enfermeiro"] if cargo not in cargos_validos: return False, "Cargo inválido." if cargo in ["Admin", "Gestor", "GestorEquipa"] and not password: return False, "A password é obrigatória para Admin, Gestor e Gestor de Equipa." if cargo in ["Medico", "Enfermeiro"] and not password: password = "sem_password" password_hash = gerar_hash(password) conn = ligar_db() cur = conn.cursor() try: cur.execute(""" INSERT INTO utilizadores (nome, username, password_hash, cargo, ativo) VALUES (?, ?, ?, ?, 1); """, (nome, username, password_hash, cargo)) conn.commit() return True, f"Utilizador {nome} criado com sucesso." except sqlite3.IntegrityError: return False, "Já existe um utilizador com esse username." except Exception as e: return False, f"Erro ao criar utilizador: {e}" finally: conn.close() def alterar_estado_utilizador_admin(user_id, ativo): garantir_tabelas_gestao_equipa() conn = ligar_db() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute("SELECT cargo FROM utilizadores WHERE id = ?", (user_id,)) row = cur.fetchone() if not row: conn.close() return False, "Utilizador não encontrado." if row["cargo"] == "CEO": conn.close() return False, "Não é possível desativar o CEO." cur.execute("UPDATE utilizadores SET ativo = ? WHERE id = ?", (ativo, user_id)) conn.commit() conn.close() return True, "Estado atualizado." def remover_utilizador_admin(user_id): garantir_tabelas_gestao_equipa() conn = ligar_db() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute("SELECT cargo FROM utilizadores WHERE id = ?", (user_id,)) row = cur.fetchone() if not row: conn.close() return False, "Utilizador não encontrado." if row["cargo"] == "CEO": conn.close() return False, "Não é possível remover o CEO." cur.execute("DELETE FROM equipa_membros WHERE utilizador_id = ?", (user_id,)) cur.execute("UPDATE equipas SET gestor_id = NULL WHERE gestor_id = ?", (user_id,)) cur.execute("DELETE FROM utilizadores WHERE id = ?", (user_id,)) conn.commit() conn.close() return True, "Utilizador removido." def atualizar_password_utilizador_admin(user_id, nova_password): garantir_tabelas_gestao_equipa() nova_password = (nova_password or "").strip() if not nova_password: return False, "A nova password não pode estar vazia." conn = ligar_db() cur = conn.cursor() password_hash = gerar_hash(nova_password) cur.execute("UPDATE utilizadores SET password_hash = ? WHERE id = ?", (password_hash, user_id)) conn.commit() conn.close() return True, "Password atualizada." def listar_gestores_equipa(): garantir_tabelas_gestao_equipa() conn = ligar_db() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute(""" SELECT id, nome, username, cargo FROM utilizadores WHERE ativo = 1 AND cargo IN ('GestorEquipa', 'Gestor', 'Admin', 'CEO') ORDER BY nome ASC; """) dados = [dict(r) for r in cur.fetchall()] conn.close() return dados def listar_membros_para_equipa(): garantir_tabelas_gestao_equipa() conn = ligar_db() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute(""" SELECT id, nome, username, cargo FROM utilizadores WHERE ativo = 1 AND cargo IN ('Medico', 'Gestor', 'GestorEquipa', 'Admin') ORDER BY cargo ASC, nome ASC; """) dados = [dict(r) for r in cur.fetchall()] conn.close() return dados def criar_equipa(nome, descricao, gestor_id): garantir_tabelas_gestao_equipa() nome = (nome or "").strip() descricao = (descricao or "").strip() gestor_id = gestor_id or None if not nome: return False, "O nome da equipa é obrigatório." conn = ligar_db() cur = conn.cursor() try: cur.execute(""" INSERT INTO equipas (nome, descricao, gestor_id, ativo) VALUES (?, ?, ?, 1); """, (nome, descricao, gestor_id)) conn.commit() return True, "Equipa criada com sucesso." except sqlite3.IntegrityError: return False, "Já existe uma equipa com esse nome." except Exception as e: return False, f"Erro ao criar equipa: {e}" finally: conn.close() def listar_equipas(): garantir_tabelas_gestao_equipa() conn = ligar_db() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute(""" SELECT e.id, e.nome, e.descricao, e.ativo, e.gestor_id, u.nome AS gestor_nome, COUNT(em.id) AS total_membros FROM equipas e LEFT JOIN utilizadores u ON u.id = e.gestor_id LEFT JOIN equipa_membros em ON em.equipa_id = e.id AND em.ativo = 1 GROUP BY e.id, e.nome, e.descricao, e.ativo, e.gestor_id, u.nome ORDER BY e.nome ASC; """) dados = [dict(r) for r in cur.fetchall()] conn.close() return dados def obter_equipa(equipa_id): garantir_tabelas_gestao_equipa() conn = ligar_db() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute(""" SELECT e.id, e.nome, e.descricao, e.ativo, e.gestor_id, u.nome AS gestor_nome FROM equipas e LEFT JOIN utilizadores u ON u.id = e.gestor_id WHERE e.id = ?; """, (equipa_id,)) row = cur.fetchone() conn.close() return dict(row) if row else None def listar_membros_equipa(equipa_id): garantir_tabelas_gestao_equipa() conn = ligar_db() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute(""" SELECT em.id, em.equipa_id, em.utilizador_id, u.nome, u.username, u.cargo FROM equipa_membros em INNER JOIN utilizadores u ON u.id = em.utilizador_id WHERE em.equipa_id = ? AND em.ativo = 1 ORDER BY u.cargo ASC, u.nome ASC; """, (equipa_id,)) dados = [dict(r) for r in cur.fetchall()] conn.close() return dados def adicionar_membro_equipa(equipa_id, utilizador_id): garantir_tabelas_gestao_equipa() conn = ligar_db() cur = conn.cursor() try: cur.execute(""" INSERT INTO equipa_membros (equipa_id, utilizador_id, ativo) VALUES (?, ?, 1) ON CONFLICT(equipa_id, utilizador_id) DO UPDATE SET ativo = 1; """, (equipa_id, utilizador_id)) conn.commit() return True, "Membro adicionado à equipa." except Exception as e: return False, f"Erro ao adicionar membro: {e}" finally: conn.close() def remover_membro_equipa(equipa_id, utilizador_id): garantir_tabelas_gestao_equipa() conn = ligar_db() cur = conn.cursor() cur.execute(""" UPDATE equipa_membros SET ativo = 0 WHERE equipa_id = ? AND utilizador_id = ?; """, (equipa_id, utilizador_id)) conn.commit() conn.close() return True, "Membro removido da equipa." def alterar_estado_equipa(equipa_id, ativo): garantir_tabelas_gestao_equipa() conn = ligar_db() cur = conn.cursor() cur.execute("UPDATE equipas SET ativo = ? WHERE id = ?", (ativo, equipa_id)) conn.commit() conn.close() return True, "Estado da equipa atualizado." def dashboard_gestor_equipa_resumo(gestor_id): garantir_tabelas_gestao_equipa() conn = ligar_db() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute("SELECT COUNT(*) AS total FROM equipas WHERE gestor_id = ? AND ativo = 1", (gestor_id,)) total_equipas = cur.fetchone()["total"] cur.execute(""" SELECT COUNT(em.id) AS total FROM equipa_membros em INNER JOIN equipas e ON e.id = em.equipa_id WHERE e.gestor_id = ? AND e.ativo = 1 AND em.ativo = 1; """, (gestor_id,)) total_membros = cur.fetchone()["total"] conn.close() return { "total_equipas": total_equipas or 0, "total_membros": total_membros or 0 } # ============================================================ # MARCAÇÕES - EXPORTAR / IMPORTAR EXCEL # ============================================================ def listar_marcacoes_para_excel(data_inicio=None, data_fim=None, estado=None): """Devolve todas as marcações formatadas para exportação Excel.""" conn = ligar_db() conn.row_factory = sqlite3.Row cur = conn.cursor() filtros = [] params = [] if data_inicio: filtros.append("m.data_marcacao >= ?") params.append(data_inicio) if data_fim: filtros.append("m.data_marcacao <= ?") params.append(data_fim) if estado: filtros.append("m.estado = ?") params.append(estado) where = ("WHERE " + " AND ".join(filtros)) if filtros else "" cur.execute(f""" SELECT m.id, m.data_marcacao, m.hora_marcacao, m.tipo_exame, m.estado, m.aptidao, m.observacoes, m.local_consulta, m.convocatoria, m.clinica_parceira, u.nome AS utente, u.numero_utente, e.nome AS empresa, med.nome AS medico, enf.nome AS enfermeiro FROM marcacoes m INNER JOIN utentes u ON u.id = m.utente_id LEFT JOIN empresas e ON e.id = u.empresa_id LEFT JOIN utilizadores med ON med.id = m.medico_id LEFT JOIN utilizadores enf ON enf.id = m.enfermeiro_id {where} ORDER BY m.data_marcacao DESC, m.hora_marcacao DESC; """, params) rows = cur.fetchall() conn.close() return [dict(r) for r in rows]