Arquitetura e Separação de Responsabilidades
O sistema é composto por duas camadas principais: uma interface web responsiva para criação e acompanhamento de solicitações, e um serviço de execução em segundo plano responsável pelo processamento assíncrono. A separação permite que o servidor Web foque apenas em requisições HTTP e sessão, enquanto a fila gerencia a carga de trabalho real.
Estrutura de Diretórios
videobrush_platform/
├── app.py # Ponto de entrada e configuração central
├── config.py # Variáveis de ambiente e constantes
├── extensions/
│ └── database.py # Configuração e pool de conexões MySQL
├── routes/
│ ├── __init__.py # Registro de blueprints
│ ├── auth.py # Autenticação e controle de sessão
│ └── catalog.py # Listagem e visualização de pedidos
├── services/
│ ├── worker.py # Consumidor da fila Redis
│ └── tasks.py # Lógica de execução síncrona/assíncrona
├── templates/ # Layouts Jinja2
│ ├── base.html # Herança principal
│ ├── login.html
│ └── orders.html
└── static/ # Recursos front-end (Bootstrap 5)
├── css/
└── js/
Roteiro de Autenticação e Validação de Sessão
A validação de acesso é implementada via hook de pré-requisição, garantindo que rotas protegidas só sejam acessíveis após autenticação bem-sucedida. Arquivos estáticos e a própria rota de login são excluídos desse filtro.
Criação da Aplicação Flask
from flask import Flask, redirect, session, request
from .routes import auth_bp, catalog_bp
def build_application():
app = Flask(__name__)
app.config['SECRET_KEY'] = 'chave-criptografica-segura-pode-ser-troca'
app.register_blueprint(auth_bp, url_prefix='/autenticacao')
app.register_blueprint(catalog_bp, url_prefix='/pedidos')
@app.before_request
def validar_sessao():
excecoes = ['/static', '/autenticacao/login']
if request.path in excecoes:
return
dados_usuario = session.get('credentials')
if not dados_usuario:
return redirect('/autenticacao/login')
return app
No módulo app.py, instanciamos o app retornado pela função construtora:
from routes import build_application
app = build_application()
if __name__ == '__main__':
app.run(host='127.0.0.1', port=5000, debug=True)
Gestão de Conexão com Banco de Dados
Utilizamos DBUtils para manter um pool estável de conexões MySQL, evitando overhead de handshake repetido. As funções retornam mapeamentos dicionário para facilitar manipulação em Jinja2.
Wrapper de Queries
from dbutils.pooled_db import PooledDB
import pymysql
from pymysql.cursors import DictCursor
DATABASE_POOL = PooledDB(
creator=pymysql,
mincached=2,
maxcached=3,
maxconnections=10,
blocking=True,
host='localhost',
port=3306,
user='usuario_admin',
password='senha_segura',
database='plataforma_pedidos',
charset='utf8mb4'
)
def buscar_registros(query, parametros):
conexao = DATABASE_POOL.connection()
cursor = conexao.cursor(DictCursor)
try:
cursor.execute(query, parametros)
return cursor.fetchall()
finally:
cursor.close()
conexao.release() # Devolve ao pool sem fechar fisicamente
def modificar_registros(query, parametros):
conexao = DATABASE_POOL.connection()
cursor = conexao.cursor()
try:
cursor.execute(query, parametros)
conexao.commit()
finally:
cursor.close()
conexao.release()
Autenticação de Usuários
O tratamento de credenciais verifica se os dados correspondem ao registro no banco. Em caso positivo, armazena metadados na sessão do navegador.
Rota de Login (autenticacao/auth.py)
from flask import Blueprint, render_template, request, redirect, session
from extensions.database import buscar_registros
auth_bp = Blueprint('autenticacao', __name__)
@auth_bp.route('/login', methods=['GET', 'POST'])
def exibir_login():
if request.method == 'POST':
nivel_acesso = request.form.get('nivel')
contato = request.form.get('contato')
segredo = request.form.get('segredo')
consulta = """
SELECT id, nome_completo, nivel_acesso FROM usuarios
WHERE telefone=%s AND senha=%s AND nivel=%s
"""
registro = buscar_registros(consulta, (contato, segredo, nivel_acesso))
if registro:
session['credentials'] = registro[0]
return redirect('/pedidos/lista')
return render_template('login.html', aviso='Credenciais inválidas')
return render_template('login.html')
Sistema de Templates e Padrões Visuais
Para padronizar a estrutura HTML, criamos um arquivo mestre que define cabeçalho, navegação e bloco de conteúdo variável.
Base Template (templates/base.html)
<html lang="pt-BR">
<head>
<meta charset="UTF-8">
<title>Plataforma de Pedidos</title>
<link rel="stylesheet" href="/static/css/bootstrap.min.css">
</head>
<body class="bg-light">
<nav class="navbar navbar-expand-lg navbar-dark bg-dark mb-3">
<div class="container">
<a class="navbar-brand" href="/">VideoBrush</a>
<div class="ml-auto">
<span class="text-white">Olá, {{ get_usuario_logado() }}</span>
</div>
</div>
</nav>
<main class="container px-4">
{% block conteudo %}{% endblock %}
</main>
<script src="/static/js/bootstrap.bundle.min.js"></script>
</body>
</html>
Função injetada no contexto global das views:
from flask import session
def recuperar_nome():
credenciais = session.get('credentials')
return credenciais['nome_completo'] if credenciais else 'Visitante'
# Dentro de build_application():
app.jinja_env.globals['get_usuario_logado'] = recuperar_nome
Listagem Filtrada e Join SQL
O dashboard exibe pedidos agregando informações do usuário criador. Administradores veem todas as entradas; clientes comuns veem apenas suas próprias solicitações.
Rota de Catálogo (rotas/catalog.py)
from flask import Blueprint, render_template
from extensions.database import buscar_registros
catalog_bp = Blueprint('pedidos', __name__)
STATUS_MAPEAMENTO = {
1: {'texto': 'Aguardando', 'classe': 'secondary'},
2: {'texto': 'Em Andamento', 'classe': 'warning'},
3: {'texto': 'Concluído', 'classe': 'success'},
4: {'texto': 'Falhou', 'classe': 'danger'}
}
@catalog_bp.route('/lista')
def mostrar_pedidos():
sessao = session['credentials']
if sessao['nivel_acesso'] == 'admin':
sql = "SELECT p.id, p.link, p.quantidade, p.status, u.nome_completo FROM pedidos p JOIN usuarios u ON p.usu_id=u.id"
resultado = buscar_registros(sql, ())
else:
sql = "SELECT p.id, p.link, p.quantidade, p.status, u.nome_completo FROM pedidos p JOIN usuarios u ON p.usu_id=u.id WHERE p.usu_id=%s"
resultado = buscar_registros(sql, (sessao['id'],))
return render_template('orders.html', registros=resultado, traducao_status=STATUS_MAPEAMENTO)
Tabela Renderizada (templates/orders.html)
{% extends "base.html" %}
{% block conteudo %}
<table class="table table-hover align-middle">
<thead class="table-dark">
<tr><th>ID</th><th>Link</th><th>Qtd</th><th>Status</th><th>Cliente</th></tr>
</thead>
<tbody>
{% for row in registros %}
<tr>
<td>{{ row.id }}</td>
<td>{{ row.link }}</td>
<td>{{ row.quantidade }}</td>
<td><span class="badge bg-{{ traducao_status[row.status].classe }}">{{ traducao_status[row.status].texto }}</span></td>
<td>{{ row.nome_completo }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% endblock %}
Criação de Pedidos e Sincronização Redis
A geração de novas ordens insere um registro com status inicial e empurra seu identificador para uma lista FIFO no Redis, garantindo entrega ordenada ao worker.
Lógica de Inserção
@app.route('/criar', methods=['POST'])
def criar_solicitacao():
link_url = request.form['url_destino']
qtd_alvos = int(request.form['quantidade'])
cliente_id = session['credentials']['id']
novo_id = modificate_registros(
"INSERT INTO pedidos (link, quantidade, status, usu_id) VALUES (%s, %s, 1, %s)",
(link_url, qtd_alvos, cliente_id)
)
redis_conecta.rpush('filas_video_processing', str(novo_id))
return redirect('/pedidos/lista')
Worker Assíncrono e Otimização de Memória
O processo de consumo opera independentemente, recuperando IDs da fila, atualizando estados transientes e executando tarefas paralelizadas.
Sincronização Inicial e Gerenciamento de Fila
Para evitar perda de dados caso o service reinicie, comparamos IDs pendentes no MySQL com o estado atual do Redis:
import redis
from extensions.database import buscar_registros, modificar_registros
CONEXAO_REDIS = redis.Redis(host='localhost', port=6379, db=0)
def reconciliar_fila():
pendentes_bd = [r['id'] for r in buscar_registros("SELECT id FROM pedidos WHERE status=1", ())]
pendentes_cache = CONEXAO_REDIS.lrange('filas_video_processing', 0, -1)
ids_no_redis = {int(x.decode()) for x in pendentes_cache}
faltantes = set(pendentes_bd) - ids_no_redis
if faltantes:
CONEXAO_REDIS.rpush('filas_video_processing', *[str(i) for i in faltantes])
def obter_proximo_id():
val = CONEXAO_REDIS.blpop('filas_video_processing', timeout=1)[1]
return int(val.decode()) if val else None
Alternativa eficiente para grandes volumes sem carregar tudo na RAM:
def iterador_parcial(nome_chave, tamanho_amostra=50):
total = CONEXAO_REDIS.llen(nome_chave)
indice = 0
while indice < total:
lote = CONEXAO_REDIS.lrange(nome_chave, indice, indice + tamanho_amostra - 1)
yield lote
indice += len(lote)
# Uso:
for lote in iterador_parcial('filas_video_processing'):
print(f"Processando bloco: {lote}")
Execução Paralela e Ciclo de Vida do Pedido
from concurrent.futures import ThreadPoolExecutor
import time
def simular_processamento(dados_tarefa):
time.sleep(2) # Simula trabalho externo
return True
def executar_em_lotes(dados_pendentes):
total_itens = dados_pendentes['quantidade']
with ThreadPoolExecutor(max_workers=20) as executor:
futuros = []
for _ in range(total_itens):
futuro = executor.submit(simular_processamento, dados_pendentes)
futuros.append(futuro)
for f in futuros:
f.result() # Aguarda conclusão antes de continuar
def consumir_filas():
reconciliar_fila()
while True:
id_pedido = obter_proximo_id()
if not id_pedido:
continue
registro = buscar_registros("SELECT * FROM pedidos WHERE id=%s", (id_pedido,))[0]
if not registro:
continue
# Atualiza para EM_ANDAMENTO
modificar_registros("UPDATE pedidos SET status=2 WHERE id=%s", (id_pedido,))
# Dispara executor paralelo
executar_em_lotes(registro)
# Finaliza como CONCLUIDO
modificar_registros("UPDATE pedidos SET status=3 WHERE id=%s", (id_pedido,))
Configuração de Deploy Independente
A aplicação divide-se claramente em artefatos de deployment:
videobrush_web: Responsável por servir a UI, gerenciar sessões e aceitar requisições.videobrush_worker: Script contínuo que roda separado, consomendo a fila e escalando threads para processamento pesado.