273 lines
10 KiB
Python
273 lines
10 KiB
Python
from sqlalchemy import create_engine, Column, Integer, String, Boolean, Numeric, Date, ForeignKey
|
|
from sqlalchemy.orm import relationship, sessionmaker
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
import pyotp
|
|
import os
|
|
|
|
Base = declarative_base()
|
|
engine = create_engine('sqlite:///database.db', echo=True)
|
|
SessionLocal = sessionmaker(bind=engine)
|
|
|
|
def get_db_connection():
|
|
"""
|
|
Retorna uma nova sessão do banco de dados
|
|
"""
|
|
return SessionLocal()
|
|
|
|
def execute_query(query, params=None):
|
|
"""
|
|
Executa uma query usando SQLAlchemy
|
|
"""
|
|
session = get_db_connection()
|
|
try:
|
|
result = session.execute(query, params)
|
|
session.commit()
|
|
return result
|
|
except Exception as e:
|
|
session.rollback()
|
|
raise e
|
|
finally:
|
|
session.close()
|
|
|
|
class Militante(Base):
|
|
__tablename__ = 'militantes'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
nome = Column(String(100), nullable=False)
|
|
cpf = Column(String(14), unique=True)
|
|
email = Column(String(100), unique=True)
|
|
telefone = Column(String(15))
|
|
endereco = Column(String(255))
|
|
filiado = Column(Boolean, default=False)
|
|
|
|
cotas_mensais = relationship("CotaMensal", back_populates="militante")
|
|
pagamentos = relationship("Pagamento", back_populates="militante")
|
|
materiais_vendidos = relationship("MaterialVendido", back_populates="militante")
|
|
vendas_jornais = relationship("VendaJornalAvulso", back_populates="militante")
|
|
assinaturas = relationship("AssinaturaAnual", back_populates="militante")
|
|
|
|
class CotaMensal(Base):
|
|
__tablename__ = 'cotas_mensais'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
militante_id = Column(Integer, ForeignKey('militantes.id'))
|
|
valor_antigo = Column(Numeric(10, 2), nullable=False)
|
|
valor_novo = Column(Numeric(10, 2), nullable=False)
|
|
data_alteracao = Column(Date, nullable=False)
|
|
|
|
militante = relationship("Militante", back_populates="cotas_mensais")
|
|
|
|
class TipoPagamento(Base):
|
|
__tablename__ = 'tipos_pagamento'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
descricao = Column(String(100), nullable=False)
|
|
|
|
pagamentos = relationship("Pagamento", back_populates="tipo_pagamento")
|
|
|
|
class Pagamento(Base):
|
|
__tablename__ = 'pagamentos'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
militante_id = Column(Integer, ForeignKey('militantes.id'))
|
|
tipo_pagamento_id = Column(Integer, ForeignKey('tipos_pagamento.id'))
|
|
valor = Column(Numeric(10, 2), nullable=False)
|
|
data_pagamento = Column(Date, nullable=False)
|
|
|
|
militante = relationship("Militante", back_populates="pagamentos")
|
|
tipo_pagamento = relationship("TipoPagamento", back_populates="pagamentos")
|
|
|
|
class TipoMaterial(Base):
|
|
__tablename__ = 'tipos_materiais'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
descricao = Column(String(100), nullable=False)
|
|
|
|
materiais_vendidos = relationship("MaterialVendido", back_populates="tipo_material")
|
|
assinaturas = relationship("AssinaturaAnual", back_populates="tipo_material")
|
|
|
|
class MaterialVendido(Base):
|
|
__tablename__ = 'materiais_vendidos'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
militante_id = Column(Integer, ForeignKey('militantes.id'))
|
|
tipo_material_id = Column(Integer, ForeignKey('tipos_materiais.id'))
|
|
descricao = Column(String(255), nullable=False)
|
|
valor = Column(Numeric(10, 2), nullable=False)
|
|
data_venda = Column(Date, nullable=False)
|
|
|
|
militante = relationship("Militante", back_populates="materiais_vendidos")
|
|
tipo_material = relationship("TipoMaterial", back_populates="materiais_vendidos")
|
|
|
|
class VendaJornalAvulso(Base):
|
|
__tablename__ = 'vendas_jornais_avulsos'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
militante_id = Column(Integer, ForeignKey('militantes.id'))
|
|
quantidade = Column(Integer, nullable=False)
|
|
valor_total = Column(Numeric(10, 2), nullable=False)
|
|
data_venda = Column(Date, nullable=False)
|
|
|
|
militante = relationship("Militante", back_populates="vendas_jornais")
|
|
|
|
class AssinaturaAnual(Base):
|
|
__tablename__ = 'assinaturas_anuais'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
militante_id = Column(Integer, ForeignKey('militantes.id'))
|
|
tipo_material_id = Column(Integer, ForeignKey('tipos_materiais.id'))
|
|
quantidade = Column(Integer, nullable=False)
|
|
valor_total = Column(Numeric(10, 2), nullable=False)
|
|
data_inicio = Column(Date, nullable=False)
|
|
data_fim = Column(Date, nullable=False)
|
|
|
|
militante = relationship("Militante", back_populates="assinaturas")
|
|
tipo_material = relationship("TipoMaterial", back_populates="assinaturas")
|
|
|
|
class Setor(Base):
|
|
__tablename__ = 'setores'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
nome = Column(String(100), nullable=False)
|
|
|
|
relatorios_cotas = relationship("RelatorioCotasMensais", back_populates="setor")
|
|
relatorios_vendas = relationship("RelatorioVendasMateriais", back_populates="setor")
|
|
usuarios = relationship("Usuario", back_populates="setor")
|
|
|
|
class ComiteCentral(Base):
|
|
__tablename__ = 'comites_centrais'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
nome = Column(String(100), nullable=False)
|
|
|
|
relatorios_cotas = relationship("RelatorioCotasMensais", back_populates="comite")
|
|
relatorios_vendas = relationship("RelatorioVendasMateriais", back_populates="comite")
|
|
|
|
class RelatorioCotasMensais(Base):
|
|
__tablename__ = 'relatorio_cotas_mensais'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
setor_id = Column(Integer, ForeignKey('setores.id'))
|
|
comite_id = Column(Integer, ForeignKey('comites_centrais.id'))
|
|
total_cotas = Column(Numeric(10, 2), nullable=False)
|
|
data_relatorio = Column(Date, nullable=False)
|
|
|
|
setor = relationship("Setor", back_populates="relatorios_cotas")
|
|
comite = relationship("ComiteCentral", back_populates="relatorios_cotas")
|
|
|
|
class RelatorioVendasMateriais(Base):
|
|
__tablename__ = 'relatorio_vendas_materiais'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
setor_id = Column(Integer, ForeignKey('setores.id'))
|
|
comite_id = Column(Integer, ForeignKey('comites_centrais.id'))
|
|
total_vendas = Column(Numeric(10, 2), nullable=False)
|
|
data_relatorio = Column(Date, nullable=False)
|
|
|
|
setor = relationship("Setor", back_populates="relatorios_vendas")
|
|
comite = relationship("ComiteCentral", back_populates="relatorios_vendas")
|
|
|
|
class Usuario(Base):
|
|
__tablename__ = 'usuarios'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
username = Column(String(50), unique=True, nullable=False)
|
|
password_hash = Column(String(255), nullable=False)
|
|
email = Column(String(100), unique=True, nullable=True)
|
|
otp_secret = Column(String(32), nullable=True)
|
|
role_id = Column(Integer, ForeignKey('roles.id'), nullable=True)
|
|
setor_id = Column(Integer, ForeignKey('setores.id'), nullable=True)
|
|
ativo = Column(Boolean, default=True)
|
|
is_admin = Column(Boolean, default=False)
|
|
|
|
role = relationship("Role", back_populates="usuarios")
|
|
setor = relationship("Setor", back_populates="usuarios")
|
|
|
|
def __init__(self, username, password, is_admin=False):
|
|
self.username = username
|
|
self.set_password(password)
|
|
self.otp_secret = pyotp.random_base32()
|
|
self.is_admin = is_admin
|
|
self.ativo = True
|
|
|
|
def set_password(self, password):
|
|
self.password_hash = generate_password_hash(password)
|
|
|
|
def check_password(self, password):
|
|
return check_password_hash(self.password_hash, password)
|
|
|
|
def verify_otp(self, otp_code):
|
|
totp = pyotp.TOTP(self.otp_secret)
|
|
return totp.verify(otp_code)
|
|
|
|
def get_otp_uri(self):
|
|
totp = pyotp.TOTP(self.otp_secret)
|
|
return totp.provisioning_uri(self.username, issuer_name="Sistema de Gestão")
|
|
|
|
class Role(Base):
|
|
__tablename__ = 'roles'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
nome = Column(String(50), unique=True, nullable=False)
|
|
nivel = Column(Integer, nullable=False) # Nível hierárquico (1: admin, 2: coordenador, 3: militante)
|
|
|
|
usuarios = relationship("Usuario", back_populates="role")
|
|
permissoes = relationship("RolePermissao", back_populates="role")
|
|
|
|
class Permissao(Base):
|
|
__tablename__ = 'permissoes'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
nome = Column(String(50), unique=True, nullable=False)
|
|
descricao = Column(String(255))
|
|
|
|
roles = relationship("RolePermissao", back_populates="permissao")
|
|
|
|
class RolePermissao(Base):
|
|
__tablename__ = 'roles_permissoes'
|
|
|
|
role_id = Column(Integer, ForeignKey('roles.id'), primary_key=True)
|
|
permissao_id = Column(Integer, ForeignKey('permissoes.id'), primary_key=True)
|
|
|
|
role = relationship("Role", back_populates="permissoes")
|
|
permissao = relationship("Permissao", back_populates="roles")
|
|
|
|
# Remover o banco de dados existente (se existir)
|
|
if os.path.exists('database.db'):
|
|
os.remove('database.db')
|
|
|
|
# Criar todas as tabelas novamente
|
|
Base.metadata.create_all(engine)
|
|
|
|
# Criar roles iniciais
|
|
def create_initial_data():
|
|
session = get_db_connection()
|
|
try:
|
|
# Criar role de admin
|
|
admin_role = Role(nome="Administrador", nivel=1)
|
|
session.add(admin_role)
|
|
session.flush() # Para obter o ID da role
|
|
|
|
# Criar usuário admin
|
|
admin = Usuario(
|
|
username="admin",
|
|
password="admin123",
|
|
is_admin=True
|
|
)
|
|
admin.role_id = admin_role.id
|
|
session.add(admin)
|
|
|
|
session.commit()
|
|
print(f"Segredo OTP do admin: {admin.otp_secret}")
|
|
print("Usuário admin criado com sucesso!")
|
|
|
|
except Exception as e:
|
|
print(f"Erro ao criar dados iniciais: {e}")
|
|
session.rollback()
|
|
finally:
|
|
session.close()
|
|
|
|
# Executar a criação dos dados iniciais
|
|
if __name__ == "__main__":
|
|
create_initial_data() |