commit 808e4a594e22817556eb798cef3e7831dc158cd1
parent 00df7cfc642828ddaf597dfd0bc3650c9e7903e8
Author: Pablo Murad <pblmrd@gmail.com>
Date: Fri, 8 May 2026 20:25:18 -0300
refatoração
Diffstat:
8 files changed, 171 insertions(+), 55 deletions(-)
diff --git a/lazier/api/routes.py b/lazier/api/routes.py
@@ -19,7 +19,7 @@ from ..core.jobs import build_job_artifact_path, get_job_store
from ..core.processing import process_source
from ..core.supported_sites import SUPPORTED_VIDEO_SITES
from ..core.playlist import is_playlist_url
-from ..utils import AUDIO_EXTENSIONS, TEXT_EXTENSIONS, VIDEO_EXTENSIONS
+from ..utils import VIDEO_EXTENSIONS, is_upload_extension_allowed
from .websocket import broadcast_progress
logger = logging.getLogger(__name__)
@@ -276,13 +276,13 @@ async def upload_files(
smart=smart,
chapters=chapters,
)
- valid_extensions = AUDIO_EXTENSIONS | VIDEO_EXTENSIONS | TEXT_EXTENSIONS | {".pdf"}
jobs = []
for file in files:
ext = Path(file.filename).suffix.lower()
- if ext not in valid_extensions:
- raise HTTPException(status_code=400, detail=f"Tipo de arquivo nao suportado: {ext}")
+ ok, err_msg = is_upload_extension_allowed(ext)
+ if not ok:
+ raise HTTPException(status_code=400, detail=err_msg or f"Tipo de arquivo nao suportado: {ext}")
file_path = UPLOAD_DIR / f"{uuid.uuid4()}_{file.filename}"
try:
diff --git a/lazier/core/cache.py b/lazier/core/cache.py
@@ -225,9 +225,16 @@ def calculate_url_hash(url: str) -> str:
_cache_manager: Optional[CacheManager] = None
+def _redis_opted_out() -> bool:
+ """Evita abrir socket ao Redis (ex.: testes ou CI sem Redis)."""
+ return os.getenv("LAZIER_SKIP_REDIS", "").strip().lower() in {"1", "true", "yes", "on"}
+
+
def get_cache_manager() -> CacheManager:
"""Retorna instância singleton do cache manager"""
global _cache_manager
+ if _redis_opted_out():
+ raise RuntimeError("Redis desabilitado (LAZIER_SKIP_REDIS)")
if _cache_manager is None:
_cache_manager = CacheManager()
return _cache_manager
diff --git a/lazier/core/file_handler.py b/lazier/core/file_handler.py
@@ -7,7 +7,12 @@ from pathlib import Path
from typing import Tuple, Optional
import mimetypes
-from ..utils import AUDIO_EXTENSIONS, VIDEO_EXTENSIONS, PDF_EXTENSIONS, TEXT_EXTENSIONS
+from ..utils import (
+ AUDIO_EXTENSIONS,
+ BLOCKED_UPLOAD_EXTENSIONS,
+ DOCUMENT_EXTENSIONS,
+ VIDEO_EXTENSIONS,
+)
# Tipos MIME aceitos
AUDIO_MIMES = {
@@ -35,9 +40,6 @@ DOCUMENT_MIMES = {
'text/html', 'application/xhtml+xml'
}
-# Combina todas as extensões permitidas
-ALLOWED_EXTENSIONS = AUDIO_EXTENSIONS | VIDEO_EXTENSIONS | PDF_EXTENSIONS | TEXT_EXTENSIONS
-
MAX_FILE_SIZE = int(os.getenv('LAZIER_MAX_UPLOAD_SIZE', '524288000')) # 500MB padrão
@@ -57,20 +59,18 @@ def validate_upload_file(file_path: str, file_size: int) -> Tuple[bool, Optional
if file_size > MAX_FILE_SIZE:
return False, None, f"Arquivo muito grande. Tamanho máximo: {MAX_FILE_SIZE / 1024 / 1024:.0f}MB"
- # Verifica extensão
ext = Path(file_path).suffix.lower()
- if ext not in ALLOWED_EXTENSIONS:
+ if ext in BLOCKED_UPLOAD_EXTENSIONS:
return False, None, f"Extensão não permitida: {ext}"
-
- # Detecta tipo usando as constantes de utils.py
+
+ if ext in DOCUMENT_EXTENSIONS:
+ return True, 'document', None
+ if ext in VIDEO_EXTENSIONS:
+ return True, 'video', None
if ext in AUDIO_EXTENSIONS:
return True, 'audio', None
- elif ext in VIDEO_EXTENSIONS:
- return True, 'video', None
- elif ext in PDF_EXTENSIONS or ext in TEXT_EXTENSIONS:
- return True, 'document', None
-
- return False, None, "Tipo de arquivo não reconhecido"
+ # Qualquer outra extensão: tentar como áudio (ffmpeg)
+ return True, 'audio', None
def save_upload_file(upload_file, upload_dir: Path) -> Path:
diff --git a/lazier/utils.py b/lazier/utils.py
@@ -18,8 +18,10 @@ AUDIO_EXTENSIONS = {
# Formatos comuns
'.mp3', '.wav', '.m4a', '.aac', '.flac', '.ogg', '.opus', '.wma',
# Formatos menos comuns
- '.3gp', '.3g2', '.amr', '.au', '.caf', '.mka', '.ra', '.rm',
- '.spx', '.tta', '.wv'
+ '.3gp', '.3g2', '.amr', '.au', '.caf', '.mka', '.ra', '.rm',
+ '.spx', '.tta', '.wv',
+ # MPEG / PCM / HiFi
+ '.mp2', '.mpa', '.aiff', '.aif', '.dsf', '.dff',
}
# Extensões de vídeo suportadas
VIDEO_EXTENSIONS = {
@@ -27,12 +29,24 @@ VIDEO_EXTENSIONS = {
'.mp4', '.avi', '.mkv', '.mov', '.wmv', '.flv', '.webm', '.m4v',
# Formatos menos comuns
'.3gp', '.3g2', '.asf', '.f4v', '.m2v', '.mts', '.m2ts', '.ogv',
- '.rmvb', '.ts', '.vob'
+ '.rmvb', '.ts', '.vob', '.mpeg', '.mpg', '.divx', '.xvid',
}
# Extensões de texto/documento suportadas
TEXT_EXTENSIONS = {'.txt', '.md', '.html', '.htm'}
PDF_EXTENSIONS = {'.pdf'}
+# Documentos aceites para sumarização (whitelist)
+DOCUMENT_EXTENSIONS = TEXT_EXTENSIONS | PDF_EXTENSIONS
+
+# Extensões que não são mídia áudio/vídeo nem documentos aceites — rejeitar no upload Web/API
+BLOCKED_UPLOAD_EXTENSIONS = {
+ '.doc', '.docx', '.xls', '.xlsx', '.ppt', '.pptx',
+ '.odt', '.ods', '.odp',
+ '.zip', '.rar', '.7z', '.tar', '.gz', '.bz2', '.xz',
+ '.exe', '.dll', '.msi', '.deb', '.rpm', '.app', '.dmg',
+ '.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp', '.tif', '.tiff', '.svg', '.ico', '.heic',
+}
+
def is_youtube_url(url: str) -> bool:
"""Verifica se a URL é do YouTube"""
@@ -53,13 +67,16 @@ def is_media_file(file_path: str) -> Tuple[bool, Optional[str]]:
path = Path(file_path)
if not path.exists():
return False, None
-
+
ext = path.suffix.lower()
+ if ext in DOCUMENT_EXTENSIONS:
+ return False, None
+ if ext in VIDEO_EXTENSIONS:
+ return True, 'video'
if ext in AUDIO_EXTENSIONS:
return True, 'audio'
- elif ext in VIDEO_EXTENSIONS:
- return True, 'video'
- return False, None
+ # Extensão desconhecida: candidata a mídia (ffmpeg decide depois)
+ return True, 'audio'
def validate_input(input_path: str) -> Tuple[bool, str, Optional[str]]:
@@ -89,39 +106,32 @@ def validate_input(input_path: str) -> Tuple[bool, str, Optional[str]]:
return False, None, f"Não é um arquivo: {input_path}"
ext = path.suffix.lower()
-
+
+ if ext in TEXT_EXTENSIONS:
+ return True, 'text', None
+ if ext in PDF_EXTENSIONS:
+ return True, 'pdf', None
+ if ext in VIDEO_EXTENSIONS:
+ return True, 'video', None
if ext in AUDIO_EXTENSIONS:
return True, 'audio', None
- elif ext in VIDEO_EXTENSIONS:
- return True, 'video', None
- elif ext in PDF_EXTENSIONS:
- return True, 'pdf', None
- elif ext in TEXT_EXTENSIONS:
- return True, 'text', None
- else:
- return False, None, f"Tipo de arquivo não suportado: {ext}"
+ # Qualquer outro ficheiro local existente: tentar como mídia áudio (ffmpeg demux)
+ return True, 'audio', None
+
+
+def is_upload_extension_allowed(ext: str) -> Tuple[bool, Optional[str]]:
"""
- Valida o input (arquivo ou URL)
- Retorna (válido, tipo, mensagem_erro)
- tipo pode ser 'youtube', 'audio', 'video' ou None
+ Valida extensão para upload Web/API.
+ Permite documentos aceites, extensão vazia ou mídia conhecida/genérica;
+ bloqueia extensões explicitamente não suportadas (office, arquivo, imagem, binário).
+ Retorna (permitido, mensagem_erro).
"""
- if not input_path:
- return False, None, "Input não fornecido"
-
- # Verifica se é URL do YouTube
- if is_youtube_url(input_path):
- return True, 'youtube', None
-
- # Verifica se é arquivo local
- is_media, media_type = is_media_file(input_path)
- if is_media:
- return True, media_type, None
-
- # Se não é nenhum dos dois
- if os.path.exists(input_path):
- return False, None, f"Tipo de arquivo não suportado: {Path(input_path).suffix}"
- else:
- return False, None, f"Arquivo ou URL não encontrado: {input_path}"
+ e = ext.lower()
+ if e in BLOCKED_UPLOAD_EXTENSIONS:
+ return False, f"Tipo de arquivo não suportado: {ext}"
+ if e in DOCUMENT_EXTENSIONS:
+ return True, None
+ return True, None
def check_ffmpeg() -> bool:
diff --git a/lazier/web/templates/index.html b/lazier/web/templates/index.html
@@ -333,7 +333,7 @@
<p class="section-lead">A saída final será em português do Brasil.</p>
<div class="stack">
- <div class="upload" id="uploadArea"><input type="file" id="fileInput" hidden multiple accept=".mp3,.wav,.m4a,.aac,.flac,.ogg,.opus,.wma,.3gp,.3g2,.amr,.au,.caf,.mka,.ra,.rm,.spx,.tta,.wv,.mp4,.avi,.mkv,.mov,.wmv,.flv,.webm,.m4v,.asf,.f4v,.m2v,.mts,.m2ts,.ogv,.rmvb,.ts,.vob,.pdf,.txt,.md,.html,.htm"></div>
+ <div class="upload" id="uploadArea"><input type="file" id="fileInput" hidden multiple accept="audio/*,video/*,application/pdf,text/plain,text/markdown,text/html,.md,.htm,.txt"></div>
<div class="field"><label for="urlInput">URL (opcional)</label><input type="text" id="urlInput" placeholder="YouTube, página, áudio em linha…"></div>
<div class="field">
<label>Modo</label>
diff --git a/pyproject.toml b/pyproject.toml
@@ -45,3 +45,7 @@ dependencies = [
[project.scripts]
lazier = "lazier.cli:cli"
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+addopts = "-p no:qt"
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -0,0 +1,20 @@
+"""
+Evita que `create_app()` (import de `lazier.api.main`) tente conectar ao Redis
+durante a coleta/execução dos testes — ligações lentas ou semi-abertas faziam
+o pytest parecer travado após `ssssss..........`.
+"""
+
+from __future__ import annotations
+
+import os
+
+os.environ.setdefault("LAZIER_SKIP_REDIS", "1")
+
+
+def pytest_configure(config) -> None: # noqa: ARG001
+ try:
+ from lazier.core import cache as lazier_cache
+
+ lazier_cache._cache_manager = None
+ except Exception:
+ pass
diff --git a/tests/test_api.py b/tests/test_api.py
@@ -183,3 +183,78 @@ class ApiTests(unittest.TestCase):
self.assertEqual(persisted_job["status"], "completed")
self.assertEqual(persisted_job["mode"], "summarize")
+
+ def test_upload_accepts_mpeg_and_generic_extension(self):
+ output_dir = Path(os.environ["LAZIER_OUTPUT_DIR"]) / "2026" / "06" / "01" / "mpeg-job"
+ output_dir.mkdir(parents=True, exist_ok=True)
+ output_path = output_dir / "transcricao.txt"
+ output_path.write_text("ok", encoding="utf-8")
+
+ fake_result = {
+ "mode": "transcribe",
+ "input_type": "audio",
+ "source_name": "clip.mpeg",
+ "metadata": {"title": "Clip"},
+ "transcription": "ok",
+ "summary": None,
+ "result_path": str(output_path),
+ "transcription_path": str(output_path),
+ "summary_path": None,
+ }
+
+ with patch("lazier.api.routes.process_source", return_value=fake_result):
+ r1 = self.client.post(
+ "/api/upload",
+ files={"files": ("clip.mpeg", b"fake", "video/mpeg")},
+ data={"format": "txt", "mode": "transcribe"},
+ )
+ r2 = self.client.post(
+ "/api/upload",
+ files={"files": ("audio.weird_ext", b"fake", "application/octet-stream")},
+ data={"format": "txt", "mode": "transcribe"},
+ )
+
+ self.assertEqual(r1.status_code, 200)
+ self.assertEqual(r2.status_code, 200)
+
+ def test_upload_rejects_blocked_extensions(self):
+ response = self.client.post(
+ "/api/upload",
+ files={"files": ("memo.docx", b"x", "application/vnd.ms-word")},
+ data={"format": "txt", "mode": "summarize"},
+ )
+ self.assertEqual(response.status_code, 400)
+
+
+class UtilsUploadExtensionTests(unittest.TestCase):
+ def test_is_upload_extension_allows_mpeg_and_unknown(self):
+ from lazier.utils import is_upload_extension_allowed
+
+ self.assertTrue(is_upload_extension_allowed(".mpeg")[0])
+ self.assertTrue(is_upload_extension_allowed(".mpg")[0])
+ self.assertTrue(is_upload_extension_allowed(".xyz")[0])
+ self.assertTrue(is_upload_extension_allowed("")[0])
+
+ def test_is_upload_extension_blocks_office_and_images(self):
+ from lazier.utils import is_upload_extension_allowed
+
+ self.assertFalse(is_upload_extension_allowed(".docx")[0])
+ self.assertFalse(is_upload_extension_allowed(".png")[0])
+
+ def test_validate_input_unknown_extension_is_audio(self):
+ from lazier.utils import validate_input
+
+ tmp = Path(os.getcwd()) / ".tmp-tests" / f"odd-{uuid.uuid4().hex[:8]}.xyz"
+ tmp.parent.mkdir(parents=True, exist_ok=True)
+ try:
+ tmp.write_bytes(b"\x00")
+ ok, typ, err = validate_input(str(tmp))
+ self.assertTrue(ok)
+ self.assertEqual(typ, "audio")
+ self.assertIsNone(err)
+ finally:
+ tmp.unlink(missing_ok=True)
+ try:
+ tmp.parent.rmdir()
+ except OSError:
+ pass
