89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import shutil
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from video_render.config import Settings
|
|
from video_render.ffmpeg import extract_audio_to_wav
|
|
from video_render.utils import ensure_workspace, remove_paths, sanitize_filename
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class VideoWorkspace:
|
|
original_filename: str
|
|
sanitized_name: str
|
|
workspace_dir: Path
|
|
output_dir: Path
|
|
source_path: Path
|
|
working_video_path: Path
|
|
audio_path: Path
|
|
|
|
|
|
class MediaPreparer:
|
|
def __init__(self, settings: Settings) -> None:
|
|
self.settings = settings
|
|
|
|
def prepare(self, filename: str) -> VideoWorkspace:
|
|
source_path = self.settings.videos_dir / filename
|
|
if not source_path.exists():
|
|
raise FileNotFoundError(f"Arquivo de vídeo não encontrado: {source_path}")
|
|
|
|
sanitized_name = sanitize_filename(Path(filename).stem)
|
|
workspace_dir = ensure_workspace(self.settings.videos_dir, sanitized_name)
|
|
|
|
transcription_json = workspace_dir / "transcription.json"
|
|
transcription_txt = workspace_dir / "transcription.txt"
|
|
temp_transcription_json = None
|
|
temp_transcription_txt = None
|
|
|
|
if transcription_json.exists():
|
|
temp_transcription_json = workspace_dir.parent / f".{sanitized_name}_transcription.json.tmp"
|
|
shutil.copy2(transcription_json, temp_transcription_json)
|
|
if transcription_txt.exists():
|
|
temp_transcription_txt = workspace_dir.parent / f".{sanitized_name}_transcription.txt.tmp"
|
|
shutil.copy2(transcription_txt, temp_transcription_txt)
|
|
|
|
existing_children = list(workspace_dir.iterdir())
|
|
if existing_children:
|
|
logger.info("Limpando workspace existente para %s", sanitized_name)
|
|
try:
|
|
remove_paths(existing_children)
|
|
except Exception as e:
|
|
logger.warning(f"Não foi possível limpar workspace (não crítico): {e}")
|
|
|
|
if temp_transcription_json and temp_transcription_json.exists():
|
|
shutil.move(str(temp_transcription_json), str(transcription_json))
|
|
logger.info("Transcrição preservada em %s", transcription_json)
|
|
if temp_transcription_txt and temp_transcription_txt.exists():
|
|
shutil.move(str(temp_transcription_txt), str(transcription_txt))
|
|
|
|
destination_name = f"{sanitized_name}{source_path.suffix.lower()}"
|
|
working_video_path = workspace_dir / destination_name
|
|
shutil.copy2(source_path, working_video_path)
|
|
logger.info("Cópia do vídeo criada em %s", working_video_path)
|
|
|
|
output_dir = ensure_workspace(self.settings.outputs_dir, sanitized_name)
|
|
existing_outputs = list(output_dir.iterdir())
|
|
if existing_outputs:
|
|
try:
|
|
remove_paths(existing_outputs)
|
|
except Exception as e:
|
|
logger.warning(f"Não foi possível limpar outputs antigos (não crítico): {e}")
|
|
|
|
audio_path = workspace_dir / "audio.wav"
|
|
extract_audio_to_wav(working_video_path, audio_path)
|
|
|
|
return VideoWorkspace(
|
|
original_filename=filename,
|
|
sanitized_name=sanitized_name,
|
|
workspace_dir=workspace_dir,
|
|
output_dir=output_dir,
|
|
source_path=source_path,
|
|
working_video_path=working_video_path,
|
|
audio_path=audio_path,
|
|
)
|