JotaChina/mmpSearch
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

teste login

Browse Source
This commit is contained in:
JotaChina committed 2025-12-09 17:27:00 -03:00
1 parent 1cfce4c35f
commit 22b998796c
10 files changed
+367175 -365211
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

scripts/handler/upload_server.py
+224 -275
View File
@@ -12,11 +12,47 @@ import io
from flask import Flask, request, jsonify, send_file
from flask_cors import CORS
from werkzeug.utils import secure_filename
# --- NOVAS IMPORTAÇÕES DE AUTH ---
from flask_sqlalchemy import SQLAlchemy
from flask_bcrypt import Bcrypt
from flask_login import LoginManager, UserMixin, login_user, login_required, logout_user, current_user
from main import process_single_file, rebuild_indexes, generate_manifests, slugify
from utils import ALLOWED_EXTENSIONS, ALLOWED_SAMPLE_EXTENSIONS, MMP_FOLDER, MMPZ_FOLDER, CERT_PATH, KEY_PATH, BASE_DATA, SRC_MMPSEARCH, SAMPLE_SRC, METADATA_FOLDER, XML_IMPORTED_PATH_PREFIX, SAMPLE_MANIFEST
app = Flask(__name__)
CORS(app)
# --- CONFIGURAÇÃO DE SEGURANÇA E BANCO ---
# IMPORTANTE: Troque esta chave em produção!
app.config['SECRET_KEY'] = 'chave_secreta_super_segura_mmp_ecosystem_2025'
# O banco ficará salvo em /nethome/jotachina/projetos/mmpSearch/users.db (BASE_DATA)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(BASE_DATA, 'users.db')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# CORS precisa suportar credenciais para o cookie de login funcionar
CORS(app, supports_credentials=True)
db = SQLAlchemy(app)
bcrypt = Bcrypt(app)
login_manager = LoginManager(app)
login_manager.login_view = 'login'
# --- MODELO DE USUÁRIO ---
class User(UserMixin, db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(150), unique=True, nullable=False)
password = db.Column(db.String(150), nullable=False)
# Cria o banco na inicialização se não existir
with app.app_context():
db.create_all()
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
# --- SUAS FUNÇÕES UTILITÁRIAS MANTIDAS ---
def allowed_file(filename):
return "." in filename and filename.rsplit(".", 1)[1].lower() in ALLOWED_EXTENSIONS
@@ -45,13 +81,9 @@ def load_manifest_keys():
return []
def convert_mmpz_to_mmp(mmpz_path, mmp_target_path):
"""
Tenta descompactar .mmpz usando métodos Python (rápido) e falha para o CLI do LMMS (robusto).
Salva o resultado em mmp_target_path.
"""
"""Tenta descompactar .mmpz usando Python ou LMMS CLI."""
print(f"Tentando converter: {mmpz_path} -> {mmp_target_path}")
# --- TENTATIVA 1: Métodos Python (Memória/Rápido) ---
content = None
try:
with open(mmpz_path, "rb") as f:
@@ -79,7 +111,7 @@ def convert_mmpz_to_mmp(mmpz_path, mmp_target_path):
return True
except: pass
# 1.3 Tenta ZLIB (Qt Default)
# 1.3 Tenta ZLIB
try:
decompressed = zlib.decompress(content)
with open(mmp_target_path, "wb") as f_out:
@@ -88,35 +120,22 @@ def convert_mmpz_to_mmp(mmpz_path, mmp_target_path):
return True
except: pass
# --- TENTATIVA 2: Fallback para LMMS CLI (Lento/Robusto) ---
# Se o Python falhar, pedimos ao próprio LMMS para converter
# --- Fallback para LMMS CLI ---
print("Métodos Python falharam. Tentando fallback via LMMS CLI...")
cmd = ["lmms", "--dump", mmpz_path]
# Define ambiente sem interface gráfica para evitar erros no servidor
env_vars = os.environ.copy()
env_vars["QT_QPA_PLATFORM"] = "offscreen"
try:
# O LMMS joga o XML no stdout, então capturamos e salvamos no arquivo
with open(mmp_target_path, "w") as f_out:
subprocess.run(
cmd,
stdout=f_out,
stderr=subprocess.PIPE,
check=True,
env=env_vars
)
subprocess.run(cmd, stdout=f_out, stderr=subprocess.PIPE, check=True, env=env_vars)
# Verifica se o arquivo foi criado e tem conteúdo
if os.path.exists(mmp_target_path) and os.path.getsize(mmp_target_path) > 0:
print("Sucesso: Descompactado via LMMS CLI (--dump).")
return True
else:
print("Erro: LMMS CLI rodou, mas arquivo de saída está vazio.")
return False
except subprocess.CalledProcessError as e:
print(f"Falha crítica no LMMS CLI: {e.stderr.decode('utf-8') if e.stderr else str(e)}")
return False
@@ -125,12 +144,8 @@ def convert_mmpz_to_mmp(mmpz_path, mmp_target_path):
return False
def analyze_mmp_dependencies(mmp_filename):
"""
Analisa um arquivo .mmp (que já deve ser texto XML puro).
Retorna: (is_clean, missing_files_list)
"""
"""Analisa um arquivo .mmp XML puro."""
filepath = os.path.join(MMP_FOLDER, mmp_filename)
if not os.path.exists(filepath):
print(f"Erro: Arquivo não encontrado: {filepath}")
return False, []
@@ -143,22 +158,15 @@ def analyze_mmp_dependencies(mmp_filename):
root = tree.getroot()
factory_folders = load_manifest_keys()
missing_samples = set()
# Varre todos os audiofileprocessor
for audio_node in root.findall(".//audiofileprocessor"):
src = audio_node.get('src', '')
if not src:
continue
if not src: continue
# 1. É arquivo de fábrica?
is_factory = any(src.startswith(folder + "/") for folder in factory_folders)
# 2. Já foi importado anteriormente?
is_already_imported = src.startswith("src_mmpSearch/samples/imported/")
# Se não é nativo e não está na pasta de importados, é externo
if not is_factory and not is_already_imported:
file_name = os.path.basename(src)
missing_samples.add(file_name)
@@ -166,276 +174,40 @@ def analyze_mmp_dependencies(mmp_filename):
return len(missing_samples) == 0, list(missing_samples)
def update_xml_paths_exact(mmp_filename, replacements):
"""
Substitui caminhos no XML baseando-se no mapeamento exato:
replacements = { 'nome_original_no_xml.wav': 'novo_nome_seguro.wav' }
"""
"""Substitui caminhos no XML baseando-se no mapeamento exato."""
filepath = os.path.join(MMP_FOLDER, mmp_filename)
try:
tree = ET.parse(filepath)
root = tree.getroot()
changes_made = False
# Normaliza as chaves para minúsculo para garantir o match
# Ex: {'Kick.wav': 'kick.wav'} -> {'kick.wav': 'kick.wav'}
replacements_lower = {k.lower(): v for k, v in replacements.items()}
for audio_node in root.findall(".//audiofileprocessor"):
src = audio_node.get('src', '')
if not src: continue
# Pega o nome do arquivo que está atualmente no XML
current_basename = os.path.basename(src)
current_basename_lower = current_basename.lower()
# Verifica se esse nome está na lista de substituições
if current_basename_lower in replacements_lower:
new_filename = replacements_lower[current_basename_lower]
# Constrói o novo caminho completo
new_src = f"{XML_IMPORTED_PATH_PREFIX}/{new_filename}"
# Aplica a alteração
audio_node.set('src', new_src)
audio_node.set('vol', audio_node.get('vol', '1')) # Garante volume
audio_node.set('vol', audio_node.get('vol', '1'))
print(f"[XML FIX] Substituído: {current_basename} -> {new_src}")
changes_made = True
if changes_made:
tree.write(filepath, encoding='UTF-8', xml_declaration=True)
print(f"Projeto {mmp_filename} salvo com novos caminhos.")
return True
else:
print("Aviso: Nenhum caminho foi alterado no XML (match não encontrado).")
return True # Retorna True para não travar o processo, mas avisa no log
return True
except Exception as e:
print(f"Erro crítico ao editar XML: {e}")
return False
# --- ROTAS ---
@app.route("/api/download/<project_name>", methods=["GET"])
def download_project_package(project_name):
"""
Gera um ZIP onde:
1. O .mmp é modificado NA VOLÁTIL (sem salvar no disco) para ter caminhos curtos.
2. A estrutura do ZIP fica limpa:
- projeto.mmp
- imported/
- sample1.wav
- sample2.wav
"""
# Garante extensão .mmp
if not project_name.lower().endswith('.mmp'):
project_name += '.mmp'
clean_name = secure_filename(project_name)
mmp_path = os.path.join(MMP_FOLDER, clean_name)
if not os.path.exists(mmp_path):
return jsonify({"error": "Projeto não encontrado"}), 404
# Prepara o buffer do ZIP na memória
memory_file = io.BytesIO()
try:
# 1. Parseia o XML original do disco
tree = ET.parse(mmp_path)
root = tree.getroot()
# Conjunto para rastrear quais arquivos físicos precisamos colocar no ZIP
samples_to_pack = set()
# 2. Modifica o XML na MEMÓRIA (não afeta o arquivo original no servidor)
for audio_node in root.findall(".//audiofileprocessor"):
src = audio_node.get('src', '')
# Verifica se é um sample importado do nosso sistema
if src.startswith(XML_IMPORTED_PATH_PREFIX):
# Pega apenas o nome do arquivo (ex: 'kick_deep.wav')
filename = os.path.basename(src)
# Define o NOVO caminho curto relativo para dentro do ZIP
# Antes: src_mmpSearch/samples/imported/kick_deep.wav
# Agora: imported/kick_deep.wav
short_path = f"imported/{filename}"
# Atualiza o XML na memória
audio_node.set('src', short_path)
# Adiciona à lista de arquivos para copiar
samples_to_pack.add(filename)
# 3. Cria o ZIP
with zipfile.ZipFile(memory_file, 'w', zipfile.ZIP_DEFLATED) as zf:
# A) Escreve o .mmp MODIFICADO no ZIP
# Convertemos a árvore XML modificada para string binária
xml_str = ET.tostring(root, encoding='utf-8', method='xml')
zf.writestr(clean_name, xml_str)
# B) Adiciona os samples físicos na pasta 'imported/' do ZIP
for sample_name in samples_to_pack:
physical_path = os.path.join(XML_IMPORTED_PATH_PREFIX, sample_name)
# Caminho curto dentro do ZIP
zip_entry_name = f"imported/{sample_name}"
if os.path.exists(physical_path):
zf.write(physical_path, arcname=zip_entry_name)
print(f"[ZIP CLEAN] Adicionado: {zip_entry_name}")
else:
print(f"[ZIP CLEAN] AVISO: Arquivo faltante no disco: {sample_name}")
# Finaliza e envia
memory_file.seek(0)
return send_file(
memory_file,
mimetype='application/zip',
as_attachment=True,
download_name=f"{os.path.splitext(clean_name)[0]}.zip"
)
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({"error": f"Erro ao gerar pacote: {str(e)}"}), 500
@app.route("/api/upload", methods=["POST"])
def upload_file():
if "project_file" not in request.files:
return jsonify({"error": "Nenhum arquivo enviado"}), 400
file = request.files["project_file"]
if file.filename == "":
return jsonify({"error": "Nome do arquivo vazio"}), 400
if file and allowed_file(file.filename):
# 1. Sanitização do nome
original_name = file.filename
name_without_ext = os.path.splitext(original_name)[0]
ext = os.path.splitext(original_name)[1].lower()
clean_name = slugify(name_without_ext)
if not clean_name:
clean_name = f"upload-{int(time.time())}"
# O nome final no sistema será sempre .mmp
final_mmp_filename = f"{clean_name}.mmp"
final_mmp_path = os.path.join(MMP_FOLDER, final_mmp_filename)
# Caminho temporário para o upload original
upload_filename = f"{clean_name}{ext}"
if ext == ".mmpz":
upload_path = os.path.join(MMPZ_FOLDER, upload_filename)
else:
upload_path = os.path.join(MMP_FOLDER, upload_filename)
try:
# Salva o arquivo original
file.save(upload_path)
print(f"Upload salvo em: {upload_path}")
# 2. SE FOR MMPZ, CONVERTE IMEDIATAMENTE PARA MMP
if ext == ".mmpz":
print("Detectado arquivo compactado. Iniciando extração...")
success = convert_mmpz_to_mmp(upload_path, final_mmp_path)
if not success:
# Se falhou zlib/gzip, tenta o fallback do sistema (lmms --dump) se disponível
# Mas por segurança, retornamos erro aqui para não travar
return jsonify({"error": "Falha crítica: O arquivo .mmpz não pode ser descompactado. Tente enviar o .mmp descomprimido."}), 400
# Opcional: Remover o .mmpz original já que já extraímos
# os.remove(upload_path)
# Se já for .mmp, apenas garante que está no lugar certo
elif ext == ".mmp" and upload_path != final_mmp_path:
shutil.move(upload_path, final_mmp_path)
# A partir daqui, TRABALHAMOS APENAS COM O ARQUIVO .MMP (final_mmp_filename)
# 3. Analisa dependências no arquivo .mmp limpo
is_clean, missing_files = analyze_mmp_dependencies(final_mmp_filename)
if not is_clean:
return jsonify({
"status": "missing_samples",
"message": "Samples externos detectados.",
"project_file": final_mmp_filename, # Retornamos o nome do MMP, pois é ele que vamos editar
"missing_files": missing_files
}), 202
# 4. Se limpo, processa
return process_and_build(final_mmp_filename)
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({"error": f"Erro interno: {str(e)}"}), 500
return jsonify({"error": "Tipo de arquivo não permitido"}), 400
@app.route("/api/upload/resolve", methods=["POST"])
def resolve_samples():
"""
Recebe inputs onde:
name="nome_original.wav" (chave) -> value=Arquivo (conteúdo)
"""
project_filename = request.form.get('project_file') # Nome do .mmp
if not project_filename:
return jsonify({"error": "Nome do projeto não informado"}), 400
# Cria pasta imported
os.makedirs(XML_IMPORTED_PATH_PREFIX, exist_ok=True)
try:
os.chmod(XML_IMPORTED_PATH_PREFIX, 0o775)
except: pass
# Dicionário de substituição: { 'Original.wav': 'Novo_Seguro.wav' }
replacements = {}
# request.files é um dicionário imutável, iteramos sobre ele
for original_name, file_storage in request.files.items():
if file_storage and allowed_sample(file_storage.filename):
# 1. Salva o arquivo fisicamente
# Usamos secure_filename no arquivo NOVO para evitar problemas no disco
safe_new_name = secure_filename(file_storage.filename)
# Se secure_filename deixou vazio (ex: "???"), gera um nome
if not safe_new_name:
safe_new_name = f"sample_{int(time.time())}.wav"
save_path = os.path.join(XML_IMPORTED_PATH_PREFIX, safe_new_name)
file_storage.save(save_path)
try:
os.chmod(save_path, 0o664)
except: pass
# 2. Mapeia para substituição no XML
# A chave 'original_name' vem do `name` do input HTML, que definimos como o nome original
replacements[original_name] = safe_new_name
if not replacements:
return jsonify({"error": "Nenhum arquivo válido enviado."}), 400
# 3. Atualiza o XML
if update_xml_paths_exact(project_filename, replacements):
# 4. Processa (Gera WAV na pasta correta e JSON)
return process_and_build(project_filename)
else:
return jsonify({"error": "Falha ao atualizar o arquivo de projeto."}), 500
def process_and_build(filename):
"""Encapsula a chamada do main.py"""
result = process_single_file(filename)
if result["success"]:
rebuild_indexes()
run_jekyll_build()
@@ -447,8 +219,184 @@ def process_and_build(filename):
else:
return jsonify({"error": f"Erro no processamento: {result['error']}"}), 500
# Rota avulsa de sample mantida igual...
# ==============================================================================
# ROTAS DE AUTENTICAÇÃO
# ==============================================================================
@app.route('/api/register', methods=['POST'])
def register():
data = request.json
if not data or not data.get('username') or not data.get('password'):
return jsonify({"message": "Dados incompletos"}), 400
if User.query.filter_by(username=data['username']).first():
return jsonify({"message": "Usuário já existe"}), 400
hashed_password = bcrypt.generate_password_hash(data['password']).decode('utf-8')
new_user = User(username=data['username'], password=hashed_password)
try:
db.session.add(new_user)
db.session.commit()
return jsonify({"message": "Usuário criado com sucesso!"}), 201
except Exception as e:
return jsonify({"message": f"Erro: {str(e)}"}), 500
@app.route('/api/login', methods=['POST'])
def login():
data = request.json
user = User.query.filter_by(username=data['username']).first()
if user and bcrypt.check_password_hash(user.password, data['password']):
login_user(user)
return jsonify({"message": "Login realizado", "user": user.username}), 200
return jsonify({"message": "Credenciais inválidas"}), 401
@app.route('/api/logout', methods=['POST'])
@login_required
def logout():
logout_user()
return jsonify({"message": "Logout realizado"}), 200
@app.route('/api/check_auth', methods=['GET'])
def check_auth():
if current_user.is_authenticated:
return jsonify({"logged_in": True, "user": current_user.username})
return jsonify({"logged_in": False})
# ==============================================================================
# ROTAS PRINCIPAIS
# ==============================================================================
@app.route("/api/download/<project_name>", methods=["GET"])
def download_project_package(project_name):
"""Gera um ZIP com caminhos limpos (Não exige login para baixar)."""
if not project_name.lower().endswith('.mmp'):
project_name += '.mmp'
clean_name = secure_filename(project_name)
mmp_path = os.path.join(MMP_FOLDER, clean_name)
if not os.path.exists(mmp_path):
return jsonify({"error": "Projeto não encontrado"}), 404
memory_file = io.BytesIO()
try:
tree = ET.parse(mmp_path)
root = tree.getroot()
samples_to_pack = set()
for audio_node in root.findall(".//audiofileprocessor"):
src = audio_node.get('src', '')
if src.startswith(XML_IMPORTED_PATH_PREFIX):
filename = os.path.basename(src)
short_path = f"imported/{filename}"
audio_node.set('src', short_path)
samples_to_pack.add(filename)
with zipfile.ZipFile(memory_file, 'w', zipfile.ZIP_DEFLATED) as zf:
xml_str = ET.tostring(root, encoding='utf-8', method='xml')
zf.writestr(clean_name, xml_str)
for sample_name in samples_to_pack:
physical_path = os.path.join(XML_IMPORTED_PATH_PREFIX, sample_name)
zip_entry_name = f"imported/{sample_name}"
if os.path.exists(physical_path):
zf.write(physical_path, arcname=zip_entry_name)
memory_file.seek(0)
return send_file(
memory_file,
mimetype='application/zip',
as_attachment=True,
download_name=f"{os.path.splitext(clean_name)[0]}.zip"
)
except Exception as e:
return jsonify({"error": f"Erro ao gerar pacote: {str(e)}"}), 500
@app.route("/api/upload", methods=["POST"])
@login_required # <--- PROTEGIDO
def upload_file():
if "project_file" not in request.files:
return jsonify({"error": "Nenhum arquivo enviado"}), 400
file = request.files["project_file"]
if file.filename == "":
return jsonify({"error": "Nome do arquivo vazio"}), 400
if file and allowed_file(file.filename):
original_name = file.filename
name_without_ext = os.path.splitext(original_name)[0]
ext = os.path.splitext(original_name)[1].lower()
clean_name = slugify(name_without_ext)
if not clean_name:
clean_name = f"upload-{int(time.time())}"
final_mmp_filename = f"{clean_name}.mmp"
final_mmp_path = os.path.join(MMP_FOLDER, final_mmp_filename)
upload_filename = f"{clean_name}{ext}"
if ext == ".mmpz":
upload_path = os.path.join(MMPZ_FOLDER, upload_filename)
else:
upload_path = os.path.join(MMP_FOLDER, upload_filename)
try:
file.save(upload_path)
print(f"Upload salvo em: {upload_path} (User: {current_user.username})")
if ext == ".mmpz":
success = convert_mmpz_to_mmp(upload_path, final_mmp_path)
if not success:
return jsonify({"error": "Falha crítica na descompactação."}), 400
elif ext == ".mmp" and upload_path != final_mmp_path:
shutil.move(upload_path, final_mmp_path)
is_clean, missing_files = analyze_mmp_dependencies(final_mmp_filename)
if not is_clean:
return jsonify({
"status": "missing_samples",
"message": "Samples externos detectados.",
"project_file": final_mmp_filename,
"missing_files": missing_files
}), 202
return process_and_build(final_mmp_filename)
except Exception as e:
return jsonify({"error": f"Erro interno: {str(e)}"}), 500
return jsonify({"error": "Tipo de arquivo não permitido"}), 400
@app.route("/api/upload/resolve", methods=["POST"])
@login_required # <--- PROTEGIDO
def resolve_samples():
project_filename = request.form.get('project_file')
if not project_filename:
return jsonify({"error": "Nome do projeto não informado"}), 400
os.makedirs(XML_IMPORTED_PATH_PREFIX, exist_ok=True)
replacements = {}
for original_name, file_storage in request.files.items():
if file_storage and allowed_sample(file_storage.filename):
safe_new_name = secure_filename(file_storage.filename)
if not safe_new_name: safe_new_name = f"sample_{int(time.time())}.wav"
save_path = os.path.join(XML_IMPORTED_PATH_PREFIX, safe_new_name)
file_storage.save(save_path)
replacements[original_name] = safe_new_name
if not replacements:
return jsonify({"error": "Nenhum arquivo válido enviado."}), 400
if update_xml_paths_exact(project_filename, replacements):
return process_and_build(project_filename)
else:
return jsonify({"error": "Falha ao atualizar o arquivo de projeto."}), 500
@app.route('/api/upload/sample', methods=['POST'])
@login_required # <--- PROTEGIDO
def upload_sample_standalone():
if 'sample_file' not in request.files: return jsonify({'error': 'Nenhum arquivo'}), 400
file = request.files['sample_file']
@@ -466,4 +414,5 @@ def upload_sample_standalone():
if __name__ == "__main__":
context = (CERT_PATH, KEY_PATH) if os.path.exists(CERT_PATH) else "adhoc"
# Escuta em 0.0.0.0
app.run(host="0.0.0.0", port=33002, ssl_context=context, debug=True)