#!/usr/bin/env python3 """Backend do Video Editor - Flask porta 5054""" import json import os import threading import time import uuid from datetime import datetime, timedelta from pathlib import Path from flask import Flask, jsonify, request, send_file, abort, Response from flask_cors import CORS from processor import processar_video # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- TEMP_DIR = Path(__file__).parent / "temp" OUTPUT_DIR = Path(__file__).parent / "output" MUSIC_DIR = Path(__file__).parent / "music" MANIFEST_PATH = MUSIC_DIR / "manifest.json" for d in [TEMP_DIR, OUTPUT_DIR, MUSIC_DIR]: d.mkdir(exist_ok=True) app = Flask(__name__) CORS(app, origins="*", supports_credentials=True) # Jobs em memória: job_id -> dict jobs: dict = {} # Cache do manifest (recarrega se mtime mudar) _manifest_cache = {"mtime": 0, "data": None} def _load_manifest() -> dict: """Carrega manifest.json (com cache por mtime).""" if not MANIFEST_PATH.exists(): return {"tracks": [], "moods": []} mtime = MANIFEST_PATH.stat().st_mtime if _manifest_cache["data"] is None or mtime != _manifest_cache["mtime"]: try: with open(MANIFEST_PATH, "r", encoding="utf-8") as f: _manifest_cache["data"] = json.load(f) _manifest_cache["mtime"] = mtime except Exception as e: print(f"[manifest] erro ao carregar: {e}") return {"tracks": [], "moods": []} return _manifest_cache["data"] def _track_by_id(track_id: str) -> dict | None: data = _load_manifest() for t in data.get("tracks", []): if t.get("id") == track_id: return t return None # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def callback_status(job_id: str, status: str, progresso: int, mensagem: str, video_path: str = None): if job_id not in jobs: return jobs[job_id].update({ "status": status, "progresso": progresso, "mensagem": mensagem, "updated_at": datetime.utcnow().isoformat() }) if video_path: jobs[job_id]["video_path"] = video_path def cleanup_old_files(): """Remove arquivos temporários e de output com mais de 24 horas.""" cutoff = datetime.now() - timedelta(hours=24) for d in [TEMP_DIR, OUTPUT_DIR]: for f in d.iterdir(): if f.is_file(): mtime = datetime.fromtimestamp(f.stat().st_mtime) if mtime < cutoff: try: f.unlink() except Exception: pass # Roda cleanup na inicialização cleanup_old_files() # --------------------------------------------------------------------------- # Endpoints # --------------------------------------------------------------------------- @app.route("/api/health", methods=["GET"]) def health(): return jsonify({"status": "ok", "timestamp": datetime.utcnow().isoformat()}) @app.route("/api/processar", methods=["POST"]) def processar(): """Inicia o processamento de um vídeo.""" if "video" not in request.files: return jsonify({"erro": "Arquivo de vídeo não enviado."}), 400 video_file = request.files["video"] if not video_file.filename: return jsonify({"erro": "Nome do arquivo inválido."}), 400 # Valida extensão ext = Path(video_file.filename).suffix.lower() if ext not in {".mp4", ".mov", ".avi", ".webm", ".mkv"}: return jsonify({"erro": f"Formato não suportado: {ext}"}), 400 modo = request.form.get("modo", "clip") usar_musica = request.form.get("musica", "true").lower() == "true" musica_id = (request.form.get("musica_id") or "").strip() or None # Se musica_id foi enviado, valida no manifest musica_path_direto = None if musica_id: t = _track_by_id(musica_id) if not t: return jsonify({"erro": f"musica_id inválido: {musica_id}"}), 400 cand = MUSIC_DIR / t["arquivo"] if not cand.exists(): return jsonify({"erro": f"arquivo da música não encontrado no servidor: {t['arquivo']}"}), 500 musica_path_direto = str(cand) usar_musica = True # garante coerência estilo_legenda = request.form.get("estilo_legenda", "branca") tamanho_legenda = request.form.get("tamanho_legenda", "medio") if tamanho_legenda not in {"pequeno", "medio", "grande"}: try: int(tamanho_legenda) except ValueError: tamanho_legenda = "medio" segmentos = [] seg_json = request.form.get("segmentos", "[]") try: segmentos = json.loads(seg_json) except Exception: pass job_id = str(uuid.uuid4()) # Salva o vídeo no temp video_path = str(TEMP_DIR / f"{job_id}_input{ext}") video_file.save(video_path) # Registra job jobs[job_id] = { "status": "fila", "progresso": 0, "mensagem": "Na fila...", "video_path": None, "created_at": datetime.utcnow().isoformat(), "updated_at": datetime.utcnow().isoformat() } # Dispara processamento em background t = threading.Thread( target=processar_video, args=(job_id, video_path, modo, segmentos, usar_musica, estilo_legenda, callback_status), kwargs={ "tamanho_legenda": tamanho_legenda, "musica_path_direto": musica_path_direto, }, daemon=True ) t.start() return jsonify({"job_id": job_id}) @app.route("/api/job/", methods=["GET"]) def status_job(job_id: str): """Retorna status atual de um job.""" if job_id not in jobs: return jsonify({"erro": "Job não encontrado."}), 404 job = jobs[job_id].copy() result = { "status": job["status"], "progresso": job["progresso"], "mensagem": job["mensagem"] } if job["status"] == "concluido" and job.get("video_path"): result["video_url"] = f"/api/download/{job_id}" return jsonify(result) @app.route("/api/download/", methods=["GET"]) def download_video(job_id: str): """Baixa o vídeo processado.""" if job_id not in jobs: return jsonify({"erro": "Job não encontrado."}), 404 job = jobs[job_id] if job["status"] != "concluido" or not job.get("video_path"): return jsonify({"erro": "Vídeo ainda não disponível."}), 404 video_path = job["video_path"] if not Path(video_path).exists(): return jsonify({"erro": "Arquivo não encontrado no servidor."}), 404 return send_file( video_path, mimetype="video/mp4", as_attachment=True, download_name=f"reel_{job_id[:8]}.mp4" ) # --------------------------------------------------------------------------- # Biblioteca de músicas # --------------------------------------------------------------------------- @app.route("/api/musicas", methods=["GET"]) def listar_musicas(): """Retorna a lista de trilhas do manifest enriquecida com preview_url.""" data = _load_manifest() tracks = [] for t in data.get("tracks", []): arquivo = MUSIC_DIR / t.get("arquivo", "") if not arquivo.exists(): continue tracks.append({ "id": t["id"], "nome": t["nome"], "mood": t["mood"], "duracao": t["duracao"], "fonte": t.get("fonte", ""), "creditos": t.get("creditos", ""), "preview_url": f"/api/musicas/{t['id']}/preview", }) return jsonify({ "tracks": tracks, "moods": data.get("moods", []), "fonte": data.get("fonte", ""), "licenca": data.get("licenca", ""), }) @app.route("/api/musicas//preview", methods=["GET"]) def preview_musica(track_id: str): """Serve o MP3 com suporte a Range (necessário pro