comunatec/controles
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: melhora processo de seed de dados com melhor tratamento de erros e concorrência

Browse Source
This commit is contained in:
andersonid
2025-04-04 09:24:45 -03:00
parent 9bb62c81a7
commit c7c3b95f0b
2 changed files with 148 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
seed.py Normal file
View File

@@ -0,0 +1,32 @@
from seed_data import seed_database
from functions.database import Base, engine, get_db_connection
import time
import os
def wait_for_db():
db_path = os.path.expanduser("~/.local/share/controles/database.db")
max_attempts = 30
attempt = 0
while attempt < max_attempts:
if os.path.exists(db_path):
try:
db = get_db_connection()
db.execute("SELECT 1")
return True
except:
pass
print(f"Aguardando banco de dados... tentativa {attempt + 1}/{max_attempts}")
time.sleep(1)
attempt += 1
return False
if __name__ == "__main__":
print("Aguardando banco de dados estar pronto...")
if wait_for_db():
print("Iniciando população do banco de dados...")
seed_database()
print("Banco de dados populado com sucesso!")
else:
print("Erro: Banco de dados não ficou pronto a tempo.")

220
seed_data.py
View File

@@ -2,16 +2,16 @@ from datetime import datetime, timedelta
from functions.database import ( from functions.database import (
Base, Militante, CotaMensal, TipoPagamento, Pagamento, Base, Militante, CotaMensal, TipoPagamento, Pagamento,
MaterialVendido, TipoMaterial, VendaJornalAvulso, AssinaturaAnual, MaterialVendido, TipoMaterial, VendaJornalAvulso, AssinaturaAnual,
RelatorioCotasMensais, RelatorioVendasMateriais, engine, get_db_connection, RelatorioCotasMensais, RelatorioVendasMateriais, engine, SessionLocal,
Setor, ComiteCentral, Usuario, Role, EmailMilitante, Endereco Setor, ComiteCentral, Usuario, Role, EmailMilitante, Endereco
) )
import random import random
from faker import Faker from faker import Faker
import time
fake = Faker('pt_BR') fake = Faker('pt_BR')
db_session = get_db_connection()
def criar_tipos_pagamento(): def criar_tipos_pagamento(session):
"""Cria tipos de pagamento padrão""" """Cria tipos de pagamento padrão"""
tipos = [ tipos = [
"Dinheiro", "Dinheiro",
@@ -21,11 +21,11 @@ def criar_tipos_pagamento():
"Transferência Bancária" "Transferência Bancária"
] ]
for tipo in tipos: for tipo in tipos:
if not db_session.query(TipoPagamento).filter_by(descricao=tipo).first(): if not session.query(TipoPagamento).filter_by(descricao=tipo).first():
db_session.add(TipoPagamento(descricao=tipo)) session.add(TipoPagamento(descricao=tipo))
db_session.commit() session.commit()
def criar_tipos_material(): def criar_tipos_material(session):
"""Cria tipos de material padrão""" """Cria tipos de material padrão"""
tipos = [ tipos = [
"Jornal", "Jornal",
@@ -35,98 +35,111 @@ def criar_tipos_material():
"Cartilha" "Cartilha"
] ]
for tipo in tipos: for tipo in tipos:
if not db_session.query(TipoMaterial).filter_by(descricao=tipo).first(): if not session.query(TipoMaterial).filter_by(descricao=tipo).first():
db_session.add(TipoMaterial(descricao=tipo)) session.add(TipoMaterial(descricao=tipo))
db_session.commit() session.commit()
def criar_militantes(num_militantes): def criar_militantes(session, num_militantes):
print(f"\nCriando {num_militantes} militantes...") print(f"\nCriando {num_militantes} militantes...")
militantes = [] militantes = []
emails_usados = set() # Conjunto para rastrear emails já usados emails_usados = set()
# Obter um setor existente
setor = session.query(Setor).first()
if not setor:
print("Erro: Nenhum setor encontrado!")
return []
for i in range(num_militantes): for i in range(num_militantes):
nome = fake.name() try:
cpf = fake.cpf() nome = fake.name()
cpf = fake.cpf()
# Gerar email único
while True: while True:
email = fake.email() email = fake.email()
if email not in emails_usados: if email not in emails_usados:
emails_usados.add(email) emails_usados.add(email)
break break
# Criar endereço endereco = Endereco(
endereco = Endereco( estado=fake.estado_sigla(),
estado=fake.estado_sigla(), cidade=fake.city(),
cidade=fake.city(), bairro=fake.bairro(),
bairro=fake.bairro(), rua=fake.street_name(),
rua=fake.street_name(), numero=str(random.randint(1, 999)),
numero=str(random.randint(1, 999)), complemento=f"Bloco {random.randint(1, 10)}, Apto {random.randint(1, 999)}" if random.random() < 0.3 else None,
complemento=f"Bloco {random.randint(1, 10)}, Apto {random.randint(1, 999)}" if random.random() < 0.3 else None, cep=fake.postcode()
cep=fake.postcode() )
) session.add(endereco)
db_session.add(endereco) session.flush()
db_session.flush() # Para obter o ID do endereço
print(f"Criando militante {i+1}: {nome} (CPF: {cpf})")
print(f"Criando militante {i+1}: {nome} (CPF: {cpf})")
militante = Militante(
# Criar militante nome=nome,
militante = Militante( cpf=cpf,
nome=nome, titulo_eleitoral=str(random.randint(100000000000, 999999999999)),
cpf=cpf, data_nascimento=fake.date_of_birth(minimum_age=18, maximum_age=65),
titulo_eleitoral=str(random.randint(100000000000, 999999999999)), data_entrada_oci=fake.date_between(start_date='-5y', end_date='today'),
data_nascimento=fake.date_of_birth(minimum_age=18, maximum_age=65), data_efetivacao_oci=fake.date_between(start_date='-4y', end_date='today'),
data_entrada_oci=fake.date_between(start_date='-5y', end_date='today'), telefone1=fake.phone_number(),
data_efetivacao_oci=fake.date_between(start_date='-4y', end_date='today'), telefone2=fake.phone_number() if random.random() < 0.3 else None,
telefone1=fake.phone_number(), profissao=fake.job(),
telefone2=fake.phone_number() if random.random() < 0.3 else None, regime_trabalho=random.choice(['CLT', 'PJ', 'Estatutário', 'Autônomo']),
profissao=fake.job(), empresa=fake.company(),
regime_trabalho=random.choice(['CLT', 'PJ', 'Estatutário', 'Autônomo']), contratante=fake.company() if random.random() < 0.2 else None,
empresa=fake.company(), instituicao_ensino=fake.company() if random.random() < 0.4 else None,
contratante=fake.company() if random.random() < 0.2 else None, tipo_instituicao=random.choice(['Federal', 'Estadual', 'Municipal', 'Privada']) if random.random() < 0.4 else None,
instituicao_ensino=fake.company() if random.random() < 0.4 else None, sindicato=fake.company() if random.random() < 0.6 else None,
tipo_instituicao=random.choice(['Federal', 'Estadual', 'Municipal', 'Privada']) if random.random() < 0.4 else None, cargo_sindical=random.choice(['Diretor', 'Delegado', 'Conselheiro']) if random.random() < 0.3 else None,
sindicato=fake.company() if random.random() < 0.6 else None, dirigente_sindical=random.random() < 0.2,
cargo_sindical=random.choice(['Diretor', 'Delegado', 'Conselheiro']) if random.random() < 0.3 else None, central_sindical=random.choice(['CUT', 'CSP-Conlutas', 'CTB', 'Força Sindical']) if random.random() < 0.4 else None,
dirigente_sindical=random.random() < 0.2, endereco_id=endereco.id,
central_sindical=random.choice(['CUT', 'CSP-Conlutas', 'CTB', 'Força Sindical']) if random.random() < 0.4 else None, responsabilidades=random.randint(0, 1023)
endereco_id=endereco.id, )
responsabilidades=random.randint(0, 1023) # Valor aleatório para responsabilidades session.add(militante)
) session.flush()
db_session.add(militante)
db_session.flush() # Para obter o ID do militante email_militante = EmailMilitante(
militante_id=militante.id,
# Criar email do militante endereco_email=email
email_militante = EmailMilitante( )
militante_id=militante.id, session.add(email_militante)
endereco_email=email
) militantes.append(militante)
db_session.add(email_militante)
# Commit a cada militante para evitar transações muito longas
militantes.append(militante) session.commit()
except Exception as e:
print(f"Erro ao criar militante {i+1}: {e}")
session.rollback()
continue
db_session.commit()
return militantes return militantes
def criar_cotas(militantes, quantidade_por_militante=3): def criar_cotas(session, militantes, quantidade_por_militante=3):
"""Cria cotas mensais fictícias"""
print(f"Criando {quantidade_por_militante} cotas para cada um dos {len(militantes)} militantes...") print(f"Criando {quantidade_por_militante} cotas para cada um dos {len(militantes)} militantes...")
for militante in militantes: for militante in militantes:
print(f"Criando cotas para militante {militante.nome}") try:
for i in range(quantidade_por_militante): print(f"Criando cotas para militante {militante.nome}")
data_base = datetime.now() - timedelta(days=30 * i) for i in range(quantidade_por_militante):
valor = random.uniform(50, 200) data_base = datetime.now() - timedelta(days=30 * i)
cota = CotaMensal( valor = random.uniform(50, 200)
militante_id=militante.id, cota = CotaMensal(
valor_antigo=valor, militante_id=militante.id,
valor_novo=valor * 1.1, valor_antigo=valor,
data_alteracao=data_base, valor_novo=valor * 1.1,
data_vencimento=data_base + timedelta(days=30), data_alteracao=data_base,
pago=random.choice([True, False]) data_vencimento=data_base + timedelta(days=30),
) pago=random.choice([True, False])
db_session.add(cota) )
print(f" Cota criada: valor={valor:.2f}, vencimento={data_base + timedelta(days=30)}") session.add(cota)
db_session.commit() session.commit()
except Exception as e:
print(f"Erro ao criar cotas para militante {militante.nome}: {e}")
session.rollback()
continue
print("Cotas criadas com sucesso!") print("Cotas criadas com sucesso!")
def criar_pagamentos(militantes): def criar_pagamentos(militantes):
@@ -273,21 +286,20 @@ def seed_database():
"""Função principal para popular o banco de dados com dados fictícios""" """Função principal para popular o banco de dados com dados fictícios"""
print("Populando banco de dados com dados fictícios...") print("Populando banco de dados com dados fictícios...")
criar_tipos_pagamento() session = SessionLocal()
criar_tipos_material() try:
criar_setores() criar_tipos_pagamento(session)
criar_comites() criar_tipos_material(session)
criar_roles()
militantes = criar_militantes(session, 50)
militantes = criar_militantes(50) if militantes:
criar_cotas(militantes) criar_cotas(session, militantes)
criar_pagamentos(militantes) print("Dados fictícios criados com sucesso!")
criar_materiais_vendidos(militantes) except Exception as e:
criar_vendas_jornal(militantes) print(f"Erro ao popular banco de dados: {e}")
criar_assinaturas(militantes) session.rollback()
criar_relatorios() finally:
session.close()
print("Dados fictícios criados com sucesso!")
if __name__ == "__main__": if __name__ == "__main__":
seed_database() seed_database()