Gerenciador de Arquivos com Python, Flask e PyQt5

Neste tutorial, vamos construir um sistema de gerenciamento de arquivos utilizando Python, Flask no backend e PyQt5 no frontend (com uma versão de linha de comando opcional). O sistema oferece upload, download, listagem e exclusão de arquivos através de API REST.

  1. Estrutura do Projeto

O projeto é dividido em dois módulos prinicpais:

  • Servidor (file_server.py): aplicação Flask que expõe endpoints para manipulação de arquivos.
  • Cliente GUI (file_client_gui.py): interface gráfica construída com PyQt5 que consome a API.
  • Cliente CLI (file_client_cli.py): versão por linha de comando com suporte a progresso.
  1. Servidor Flask

O servidor utiliza Flask e gevent para desempenho. As configurações principais incluem diretório de upload, limite de tamanho (1600 MB) e extensões permitidas.

# file_server.py
import os, time
from flask import Flask, request, send_from_directory, jsonify, abort
from werkzeug.utils import secure_filename
from gevent.pywsgi import WSGIServer
from gevent import monkey
monkey.patch_all(thread=False, ssl=False)

app = Flask(__name__)
app.config.update(
    UPLOAD_FOLDER=os.path.join(os.getcwd(), 'arquivos'),
    MAX_CONTENT_LENGTH=1600 * 1024 * 1024,
    ALLOWED_EXTENSIONS={'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif', 'zip', 'rar'},
    SECURE_FILENAME=True
)
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)

def validar_arquivo(nome):
    if app.config['SECURE_FILENAME']:
        nome = secure_filename(nome)
    if '.' not in nome:
        return False, "Tipo de arquivo não identificado"
    ext = nome.rsplit('.', 1)[1].lower()
    if ext not in app.config['ALLOWED_EXTENSIONS']:
        return False, "Extensão não permitida"
    return True, nome

@app.route('/api/enviar', methods=['POST'])
def handle_enviar():
    if 'arquivo' not in request.files:
        return jsonify({"erro": "Nenhum arquivo selecionado"}), 400
    arquivos = request.files.getlist('arquivo')
    resultados = []
    for arq in arquivos:
        if arq.filename == '':
            resultados.append({"status": "erro", "mensagem": "Nome vazio"})
            continue
        valido, info = validar_arquivo(arq.filename)
        if not valido:
            resultados.append({"status": "erro", "nome": arq.filename, "mensagem": info})
            continue
        nome_seguro = info
        caminho = os.path.join(app.config['UPLOAD_FOLDER'], nome_seguro)
        try:
            if os.path.exists(caminho):
                raise FileExistsError("Arquivo já existe")
            arq.save(caminho)
            stat = os.stat(caminho)
            resultados.append({
                "status": "sucesso",
                "nome": nome_seguro,
                "tamanho": stat.st_size,
                "timestamp": int(time.time())
            })
        except Exception as e:
            resultados.append({"status": "erro", "nome": arq.filename, "mensagem": str(e)})
    return jsonify({"resultados": resultados})

@app.route('/api/arquivos')
def listar_arquivos():
    try:
        lista = []
        for nome in os.listdir(app.config['UPLOAD_FOLDER']):
            caminho = os.path.join(app.config['UPLOAD_FOLDER'], nome)
            stat = os.stat(caminho)
            lista.append({
                "nome": nome,
                "tamanho": stat.st_size,
                "modificado": stat.st_mtime,
                "url_download": f"/api/baixar/{nome}"
            })
        return jsonify({"arquivos": lista})
    except Exception as e:
        return jsonify({"erro": str(e)}), 500

@app.route('/api/baixar/<path:nome>')
def baixar_arquivo(nome):
    try:
        if '../' in nome or not os.path.exists(os.path.join(app.config['UPLOAD_FOLDER'], nome)):
            abort(404)
        return send_from_directory(app.config['UPLOAD_FOLDER'], nome, as_attachment=True)
    except Exception as e:
        return jsonify({"erro": str(e)}), 404

@app.route('/api/deletar/<path:nome>', methods=['DELETE'])
def deletar_arquivo(nome):
    try:
        caminho = os.path.join(app.config['UPLOAD_FOLDER'], nome)
        if not os.path.exists(caminho):
            return jsonify({"erro": "Arquivo não encontrado"}), 404
        os.remove(caminho)
        return jsonify({"status": "sucesso"})
    except Exception as e:
        return jsonify({"erro": str(e)}), 500

if __name__ == '__main__':
    ip = '127.0.0.1'
    porta = 8885
    servidor = WSGIServer((ip, porta), app)
    print(f"Servidor rodando em http://{ip}:{porta}")
    servidor.serve_forever()

  1. Cliente com Interface Gráfica (PyQt5)

O cliente GUI utiliza PyQt5 para exibir uma janela com campo de endereço do servidor, botões para upload, download (duplo clique), listagem e exclusão.

# file_client_gui.py
import sys, os
import requests
from PyQt5.QtWidgets import (
    QApplication, QMainWindow, QVBoxLayout, QWidget, QPushButton,
    QFileDialog, QListWidget, QLabel, QLineEdit, QMessageBox, QHBoxLayout
)
from PyQt5.QtCore import Qt

class GerenciadorArquivos(QMainWindow):
    def __init__(self, url_servidor="http://127.0.0.1:8885"):
        super().__init__()
        self.url_servidor = url_servidor
        self.iniciar_ui()

    def iniciar_ui(self):
        self.setWindowTitle("Gerenciador de Arquivos")
        self.setGeometry(100, 100, 600, 400)
        central = QWidget()
        self.setCentralWidget(central)
        layout = QVBoxLayout()

        # Endereço do servidor
        layout_endereco = QHBoxLayout()
        lbl_endereco = QLabel("Servidor:")
        self.input_endereco = QLineEdit(self.url_servidor)
        layout_endereco.addWidget(lbl_endereco)
        layout_endereco.addWidget(self.input_endereco)
        layout.addLayout(layout_endereco)

        # Upload
        self.btn_upload = QPushButton("Enviar arquivo")
        self.btn_upload.clicked.connect(self.enviar_arquivo)
        layout.addWidget(self.btn_upload)

        # Lista de arquivos
        self.lista_arquivos = QListWidget()
        self.lista_arquivos.itemDoubleClicked.connect(self.baixar_arquivo)
        layout.addWidget(self.lista_arquivos)

        # Botões de ação
        layout_botoes = QHBoxLayout()
        self.btn_atualizar = QPushButton("Atualizar lista")
        self.btn_atualizar.clicked.connect(self.atualizar_lista)
        layout_botoes.addWidget(self.btn_atualizar)
        self.btn_deletar = QPushButton("Deletar selecionado")
        self.btn_deletar.clicked.connect(self.deletar_arquivo)
        layout_botoes.addWidget(self.btn_deletar)
        layout.addLayout(layout_botoes)

        central.setLayout(layout)
        self.atualizar_lista()

    def enviar_arquivo(self):
        caminhos, _ = QFileDialog.getOpenFileNames(self, "Selecionar arquivos")
        if not caminhos:
            return
        for caminho in caminhos:
            try:
                with open(caminho, 'rb') as f:
                    arqs = {'arquivo': (os.path.basename(caminho), f)}
                    resp = requests.post(f"{self.url_servidor}/api/enviar", files=arqs)
                    dados = resp.json()
                    ok = any(r['status'] == 'sucesso' for r in dados.get('resultados', []))
                    if ok:
                        QMessageBox.information(self, "Sucesso", f"{os.path.basename(caminho)} enviado com sucesso!")
                    else:
                        QMessageBox.warning(self, "Erro", "Falha no envio")
            except Exception as e:
                QMessageBox.critical(self, "Erro", f"Exceção: {str(e)}")
        self.atualizar_lista()

    def atualizar_lista(self):
        try:
            resp = requests.get(f"{self.url_servidor}/api/arquivos")
            dados = resp.json()
            if 'arquivos' in dados:
                self.lista_arquivos.clear()
                for arq in dados['arquivos']:
                    item = f"{arq['nome']} ({arq['tamanho'] / 1024:.2f} KB)"
                    self.lista_arquivos.addItem(item)
        except Exception as e:
            QMessageBox.critical(self, "Erro", f"Falha ao listar: {str(e)}")

    def baixar_arquivo(self, item):
        nome = item.text().split(" ")[0]
        destino, _ = QFileDialog.getSaveFileName(self, "Salvar como", nome)
        if not destino:
            return
        try:
            resp = requests.get(f"{self.url_servidor}/api/baixar/{nome}", stream=True)
            if resp.status_code == 404:
                QMessageBox.warning(self, "Erro", "Arquivo não encontrado")
                return
            total = int(resp.headers.get('content-length', 0))
            with open(destino, 'wb') as f:
                baixado = 0
                for chunk in resp.iter_content(chunk_size=1024):
                    if chunk:
                        f.write(chunk)
                        baixado += len(chunk)
                        pct = int(baixado / total * 100) if total else 100
                        self.statusBar().showMessage(f"Baixando... {pct}%")
                self.statusBar().clearMessage()
            QMessageBox.information(self, "Sucesso", f"Salvo em {destino}")
        except Exception as e:
            QMessageBox.critical(self, "Erro", f"Falha no download: {str(e)}")

    def deletar_arquivo(self):
        item = self.lista_arquivos.currentItem()
        if not item:
            QMessageBox.warning(self, "Aviso", "Selecione um arquivo")
            return
        nome = item.text().split(" ")[0]
        try:
            resp = requests.delete(f"{self.url_servidor}/api/deletar/{nome}")
            dados = resp.json()
            if dados.get('status') == 'sucesso':
                QMessageBox.information(self, "Sucesso", f"{nome} deletado")
                self.atualizar_lista()
            else:
                QMessageBox.warning(self, "Erro", "Falha ao deletar")
        except Exception as e:
            QMessageBox.critical(self, "Erro", f"Exceção: {str(e)}")

if __name__ == "__main__":
    app = QApplication(sys.argv)
    janela = GerenciadorArquivos()
    janela.show()
    sys.exit(app.exec_())

  1. Cliente de Linha de Comando (CLI)

Para automação, oferecemos uma versão CLI com comandos enviar, baixar, listar e deletar, com progresso via tqdm.

# file_client_cli.py
import requests, argparse, os, time
from tqdm import tqdm
from prettytable import PrettyTable

class ClienteArquivos:
    def __init__(self, base="http://127.0.0.1:8885"):
        self.base = base.rstrip('/')
        self.sessao = requests.Session()

    def tratar_resposta(self, resp):
        try:
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            print(f"Erro HTTP: {e}")
            return None

    def enviar(self, caminho):
        if not os.path.exists(caminho):
            print(f"Arquivo não encontrado: {caminho}")
            return False
        url = f"{self.base}/api/enviar"
        nome = os.path.basename(caminho)
        tam = os.path.getsize(caminho)
        try:
            with open(caminho, 'rb') as f, tqdm(total=tam, unit='B', unit_scale=True, desc=nome) as pbar:
                arqs = {'arquivo': (nome, f, 'application/octet-stream')}
                resp = self.sessao.post(url, files=arqs)
                dados = self.tratar_resposta(resp)
                if dados:
                    for r in dados.get('resultados', []):
                        if r['status'] == 'sucesso':
                            print(f"Enviado: {r['nome']} ({r['tamanho']} bytes)")
                        else:
                            print(f"Falha: {r.get('mensagem', 'Desconhecido')}")
                    return True
            return False
        except Exception as e:
            print(f"Erro no upload: {e}")
            return False

    def baixar(self, nome, destino=None):
        url = f"{self.base}/api/baixar/{nome}"
        destino = destino or nome
        try:
            with self.sessao.get(url, stream=True) as resp:
                if resp.status_code == 404:
                    print("Arquivo não encontrado")
                    return False
                total = int(resp.headers.get('content-length', 0))
                with open(destino, 'wb') as f, tqdm(total=total, unit='B', unit_scale=True, desc=nome) as pbar:
                    for chunk in resp.iter_content(chunk_size=1024):
                        if chunk:
                            f.write(chunk)
                            pbar.update(len(chunk))
            print(f"Salvo em: {os.path.abspath(destino)}")
            return True
        except Exception as e:
            print(f"Erro no download: {e}")
            return False

    def listar(self, tabela=True):
        url = f"{self.base}/api/arquivos"
        try:
            resp = self.sessao.get(url)
            dados = self.tratar_resposta(resp)
            if not dados or 'arquivos' not in dados:
                return False
            if tabela:
                t = PrettyTable()
                t.field_names = ["Nome", "Tamanho", "Modificado", "URL"]
                t.align["Nome"] = "l"
                t.align["Tamanho"] = "r"
                for arq in dados['arquivos']:
                    tam = f"{arq['tamanho']/1024:.2f} KB" if arq['tamanho'] < 1024*1024 else f"{arq['tamanho']/(1024*1024):.2f} MB"
                    mod = time.strftime('%Y-%m-%d %H:%M', time.localtime(arq['modificado']))
                    t.add_row([arq['nome'], tam, mod, arq['url_download']])
                print(t)
            return dados['arquivos']
        except Exception as e:
            print(f"Erro ao listar: {e}")
            return False

    def deletar(self, nome):
        url = f"{self.base}/api/deletar/{nome}"
        try:
            resp = self.sessao.delete(url)
            dados = self.tratar_resposta(resp)
            if dados and dados.get('status') == 'sucesso':
                print(f"Deletado: {nome}")
                return True
            return False
        except Exception as e:
            print(f"Erro ao deletar: {e}")
            return False

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Cliente de Arquivos CLI")
    parser.add_argument('--servidor', default='http://127.0.0.1:8885')
    sub = parser.add_subparsers(dest='comando', required=True)

    p_enviar = sub.add_parser('enviar')
    p_enviar.add_argument('caminho')

    p_baixar = sub.add_parser('baixar')
    p_baixar.add_argument('nome')
    p_baixar.add_argument('--destino')

    p_listar = sub.add_parser('listar')

    p_deletar = sub.add_parser('deletar')
    p_deletar.add_argument('nome')

    args = parser.parse_args()
    cliente = ClienteArquivos(base=args.servidor)

    if args.comando == 'enviar':
        cliente.enviar(args.caminho)
    elif args.comando == 'baixar':
        cliente.baixar(args.nome, args.destino)
    elif args.comando == 'listar':
        cliente.listar()
    elif args.comando == 'deletar':
        cliente.deletar(args.nome)

  1. Execução

Primeiro inicie o servidor:

python file_server.py

Depois execute o cliente GUI ou CLI conforme prfeerir.

Tags: Python Flask pyqt5 REST API gerenciamento de arquivos

Publicado em 7-8 21:05