from __future__ import annotations import argparse import json import math import os import random import re import shutil import subprocess import sys import tempfile import threading from dataclasses import dataclass from pathlib import Path from typing import Callable, Optional try: import numpy as np except ImportError: # pragma: no cover - reported with a friendly message later np = None import tkinter as tk from tkinter import filedialog, messagebox, simpledialog, ttk try: from PIL import Image, ImageDraw, ImageFont, ImageTk except ImportError: # Preview falls back to Tk's native PNG support. Image = None ImageDraw = None ImageFont = None ImageTk = None try: from tkinterdnd2 import DND_FILES, TkinterDnD except ImportError: # Drag and drop is optional until requirements are installed. DND_FILES = None TkinterDnD = None FRAME_SIZE = 64 MAX_ANALYSIS_FRAMES = 2400 DEFAULT_TARGET_SECONDS = 20.0 DEFAULT_MIN_LOOP_SECONDS = 4.0 VIDEO_EXTENSIONS = {".mp4", ".mov", ".mkv", ".avi", ".webm", ".m4v", ".wmv", ".flv"} GIF_MODE_LABELS = { "nao criar GIF": "none", "todos": "all", "selecionados": "manual", "aleatorios": "random", } GIF_VARIANT_LABELS = { "sem marca": "plain", "com marca": "watermarked", "ambas": "both", } SCOPE_LABELS = {"todos": "all", "selecionados": "selected"} POSITION_LABELS = { "superior esquerdo": "top-left", "superior direito": "top-right", "inferior esquerdo": "bottom-left", "inferior direito": "bottom-right", "centro": "center", } StatusCallback = Optional[Callable[[str], None]] class VideoLoopError(RuntimeError): """An expected error while analyzing or rendering a video.""" @dataclass(frozen=True) class VideoInfo: duration: float width: int height: int @dataclass(frozen=True) class LoopCandidate: start: float end: float duration: float score: float @dataclass(frozen=True) class ProcessResult: input_path: Path output_path: Path source_duration: float output_duration: float candidate: LoopCandidate repeats: int @dataclass(frozen=True) class WatermarkConfig: mode: str text: str = "" image_path: Optional[Path] = None opacity: float = 0.80 font_size: int = 36 image_scale: int = 25 position: str = "bottom-right" x_percent: Optional[float] = None y_percent: Optional[float] = None @dataclass(frozen=True) class PipelineOptions: target_seconds: float min_loop_seconds: float prefix: str = "looping" create_plain: bool = True create_watermarked: bool = False watermark: Optional[WatermarkConfig] = None clean_metadata: bool = True gif_mode: str = "none" gif_count: int = 0 gif_variant: str = "plain" gif_fps: int = 24 gif_width: Optional[int] = None skip_existing: bool = True watermarks: tuple[WatermarkConfig, ...] = () @dataclass(frozen=True) class PipelineReport: output_folder: Path loop_folder: Path gif_folder: Path outputs: tuple[Path, ...] gifs: tuple[Path, ...] failures: tuple[tuple[Path, str], ...] skipped: tuple[Path, ...] cancelled: bool created_videos: tuple[Path, ...] = () created_gifs: tuple[Path, ...] = () def _require_numpy() -> None: if np is None: raise VideoLoopError( "A biblioteca NumPy nao esta instalada. Execute: python -m pip install numpy" ) def _find_executable(name: str) -> str: executable = shutil.which(name) if executable is not None: return executable local = Path(os.environ.get("LOCALAPPDATA", "")) candidates = [] links = local / "Microsoft" / "WinGet" / "Links" if links.is_dir(): candidates.extend(links.glob(f"{name}*.exe")) packages = local / "Microsoft" / "WinGet" / "Packages" if packages.is_dir(): candidates.extend(packages.glob(f"*{name}*/**/{name}.exe")) for candidate in candidates: if candidate.is_file(): return str(candidate) raise VideoLoopError( f"Nao encontrei {name}. Instale o FFmpeg e adicione a pasta bin ao PATH." ) def _run_checked(command: list[str]) -> None: result = subprocess.run( command, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", ) if result.returncode != 0: details = result.stderr.strip() if len(details) > 1800: details = details[-1800:] raise VideoLoopError(details or "O FFmpeg nao conseguiu processar o video.") def _as_float(value: object) -> Optional[float]: if value is None or value == "N/A": return None try: return float(value) except (TypeError, ValueError): return None def probe_video(path: Path) -> VideoInfo: """Read the basic video metadata using ffprobe.""" if not path.is_file(): raise VideoLoopError(f"Arquivo de entrada nao encontrado: {path}") ffprobe = _find_executable("ffprobe") command = [ ffprobe, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height,duration:format=duration", "-of", "json", str(path), ] result = subprocess.run( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", ) if result.returncode != 0: details = result.stderr.strip() raise VideoLoopError(details or "Nao foi possivel ler os dados do video.") try: metadata = json.loads(result.stdout) stream = metadata["streams"][0] except (KeyError, IndexError, json.JSONDecodeError) as exc: raise VideoLoopError("O arquivo nao contem um fluxo de video valido.") from exc duration = _as_float(stream.get("duration")) if duration is None: duration = _as_float(metadata.get("format", {}).get("duration")) if duration is None or duration <= 0: raise VideoLoopError("Nao foi possivel descobrir a duracao do video.") try: width = int(stream.get("width") or 0) height = int(stream.get("height") or 0) except (TypeError, ValueError): width, height = 0, 0 return VideoInfo(duration=duration, width=width, height=height) def analysis_fps(duration: float) -> float: """Choose enough samples for short videos without exploding on long ones.""" return max(1.0, min(10.0, MAX_ANALYSIS_FRAMES / max(duration, 0.1))) def sample_frames(path: Path, fps: float) -> np.ndarray: """Decode small, fixed-size RGB frames for the loop search.""" _require_numpy() ffmpeg = _find_executable("ffmpeg") video_filter = ( f"fps={fps:.6f}," f"scale={FRAME_SIZE}:{FRAME_SIZE}:force_original_aspect_ratio=decrease," f"pad={FRAME_SIZE}:{FRAME_SIZE}:(ow-iw)/2:(oh-ih)/2:color=black," "format=rgb24" ) command = [ ffmpeg, "-hide_banner", "-loglevel", "error", "-nostdin", "-i", str(path), "-vf", video_filter, "-frames:v", str(MAX_ANALYSIS_FRAMES), "-f", "rawvideo", "-pix_fmt", "rgb24", "pipe:1", ] result = subprocess.run( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) if result.returncode != 0: details = result.stderr.decode("utf-8", errors="replace").strip() raise VideoLoopError(details or "Nao foi possivel extrair quadros do video.") bytes_per_frame = FRAME_SIZE * FRAME_SIZE * 3 frame_count = len(result.stdout) // bytes_per_frame if frame_count < 2: raise VideoLoopError("O video tem poucos quadros para encontrar um loop.") usable = result.stdout[: frame_count * bytes_per_frame] return np.frombuffer(usable, dtype=np.uint8).reshape( (frame_count, FRAME_SIZE, FRAME_SIZE, 3) ).copy() def _frame_features(frames: np.ndarray) -> np.ndarray: """Create compact features that tolerate small brightness changes.""" rgb = frames.astype(np.float32) / 255.0 gray = rgb @ np.array([0.299, 0.587, 0.114], dtype=np.float32) # Average 64x64 frames into an 8x8 image, then normalize each frame. pooled = gray.reshape((-1, 8, 8, 8, 8)).mean(axis=(2, 4)) mean = pooled.mean(axis=(1, 2), keepdims=True) spread = pooled.std(axis=(1, 2), keepdims=True) normalized = np.clip((pooled - mean) / (spread + 0.05), -3.0, 3.0) / 3.0 # A little color information prevents different colored scenes from matching. color_mean = rgb.mean(axis=(1, 2)) * 0.25 return np.concatenate((normalized.reshape((len(frames), -1)), color_mean), axis=1) def _fallback_candidate(source_duration: float) -> LoopCandidate: return LoopCandidate( start=0.0, end=source_duration, duration=source_duration, score=1.0, ) def find_best_loop( frames: np.ndarray, source_duration: float, min_loop_seconds: float, sample_fps: float, ) -> LoopCandidate: """Find the lowest-mismatch repeated section in sampled frames. A candidate compares a short sequence at its start with the sequence immediately after its end. This checks both the picture and the direction of movement, rather than trusting only one pair of nearly identical frames. """ _require_numpy() if min_loop_seconds <= 0: raise VideoLoopError("A duracao minima do loop deve ser maior que zero.") if min_loop_seconds > source_duration: min_loop_seconds = source_duration frame_count = len(frames) minimum_frames = max(1, math.ceil(min_loop_seconds * sample_fps - 1e-9)) if frame_count <= minimum_frames + 1: return _fallback_candidate(source_duration) features = _frame_features(frames) frame_activity = np.mean(np.abs(np.diff(features, axis=0)), axis=1) global_activity = float(np.mean(frame_activity)) if len(frame_activity) else 0.0 def static_penalty(start_index: int, end_index: int) -> float: if global_activity <= 1e-4: return 0.0 values = frame_activity[start_index : min(end_index, len(frame_activity))] if len(values) == 0: return 0.0 ratio = float(np.mean(values)) / global_activity if ratio < 0.15: return 0.75 if ratio < 0.35: return 0.20 return 0.0 window_size = max(2, min(6, round(sample_fps * 0.4))) window_size = min(window_size, frame_count - minimum_frames) if window_size < 2: return _fallback_candidate(source_duration) window_count = frame_count - window_size + 1 windows = np.stack( [features[index : index + window_size].reshape(-1) for index in range(window_count)] ) motion_windows = np.diff( windows.reshape((window_count, window_size, -1)), axis=1 ).reshape((window_count, -1)) max_start = window_count - 1 - minimum_frames max_end = window_count - 1 if max_start < 0: return _fallback_candidate(source_duration) best_score = float("inf") best_start = 0 best_end = minimum_frames # A source may contain exactly one cycle, so there is no frame after the # end to compare with the beginning. In that case compare the closing # frame and its motion with the opening boundary. last_frame = features[-1] last_motion = features[-1] - features[-2] for start_index in range(frame_count - minimum_frames + 1): if start_index + 1 >= frame_count: continue boundary_distance = float( np.mean(np.abs(last_frame - features[start_index])) ) motion_distance = float( np.mean( np.abs(last_motion - (features[start_index + 1] - features[start_index])) ) ) duration = (frame_count - start_index) / sample_fps duration_penalty = 0.02 * max(0.0, duration - min_loop_seconds) / max( min_loop_seconds, 1.0 ) score = ( boundary_distance * 0.70 + motion_distance * 0.30 + duration_penalty + static_penalty(start_index, frame_count) ) if score < best_score: best_score = score best_start = start_index best_end = frame_count for start_index in range(max_start + 1): end_indices = np.arange(start_index + minimum_frames, max_end + 1) sequence_distance = np.mean( np.abs(windows[end_indices] - windows[start_index]), axis=1 ) boundary_distance = np.mean( np.abs(features[end_indices] - features[start_index]), axis=1 ) motion_distance = np.mean( np.abs(motion_windows[end_indices] - motion_windows[start_index]), axis=1 ) durations = (end_indices - start_index) / sample_fps # Prefer the shortest valid loop only when visual quality is comparable. duration_penalty = 0.02 * np.maximum( 0.0, durations - min_loop_seconds ) / max(min_loop_seconds, 1.0) activity_penalty = np.array( [static_penalty(start_index, int(end)) for end in end_indices], dtype=np.float32, ) scores = ( sequence_distance * 0.60 + boundary_distance * 0.20 + motion_distance * 0.20 + duration_penalty + activity_penalty ) local_index = int(np.argmin(scores)) local_score = float(scores[local_index]) if local_score < best_score: best_score = local_score best_start = start_index best_end = int(end_indices[local_index]) start = best_start / sample_fps end = min(best_end / sample_fps, source_duration) duration = end - start if duration < min_loop_seconds - 0.05: return _fallback_candidate(source_duration) return LoopCandidate( start=start, end=end, duration=duration, score=best_score, ) def analyze_video( input_path: Path, min_loop_seconds: float, status: StatusCallback = None, ) -> tuple[VideoInfo, LoopCandidate, float]: info = probe_video(input_path) if min_loop_seconds <= 0: raise VideoLoopError("A duracao minima do loop deve ser maior que zero.") if min_loop_seconds > info.duration + 0.05: if status: status( f"Video ({info.duration:.2f}s) mais curto que o loop minimo " f"({min_loop_seconds:.2f}s); usarei o video inteiro como loop." ) min_loop_seconds = info.duration fps = analysis_fps(info.duration) if status: status(f"Analisando quadros em {fps:.1f} fps...") frames = sample_frames(input_path, fps) candidate = find_best_loop(frames, info.duration, min_loop_seconds, fps) return info, candidate, fps def repetition_count(loop_duration: float, target_seconds: float) -> int: if not math.isfinite(loop_duration) or loop_duration <= 0: raise VideoLoopError("O loop encontrado tem duracao invalida.") if not math.isfinite(target_seconds) or target_seconds <= 0: raise VideoLoopError("A duracao desejada deve ser maior que zero.") return max(1, math.ceil(target_seconds / loop_duration - 1e-9)) def _is_mp4_family(path: Path) -> bool: return path.suffix.lower() in {".mp4", ".m4v", ".mov"} def render_loop( input_path: Path, output_path: Path, candidate: LoopCandidate, target_seconds: float, status: StatusCallback = None, ) -> tuple[int, float]: """Render a measured segment repeatedly with normalized timestamps. The old implementation used stream copying with ``-stream_loop``. That can leave audio/timestamps longer than the video and make players freeze the last frame. Repeated entries are now concatenated and reencoded, so every repetition has fresh, monotonic timestamps. """ if output_path.resolve() == input_path.resolve(): raise VideoLoopError("Escolha um arquivo de saida diferente do arquivo original.") if not output_path.parent.exists(): raise VideoLoopError(f"A pasta de saida nao existe: {output_path.parent}") ffmpeg = _find_executable("ffmpeg") output_path.parent.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix="video_loop_") as temporary_directory: temporary_root = Path(temporary_directory) segment_path = temporary_root / "segment.mp4" extract_command = [ ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-i", str(input_path), "-ss", f"{candidate.start:.6f}", "-t", f"{candidate.duration:.6f}", "-map", "0:v:0", "-map", "0:a:0?", "-vf", "setpts=PTS-STARTPTS", "-af", "asetpts=PTS-STARTPTS", "-c:v", "libx264", "-preset", "medium", "-crf", "18", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "192k", "-avoid_negative_ts", "make_zero", str(segment_path), ] _run_checked(extract_command) segment_duration = probe_video(segment_path).duration repeats = repetition_count(segment_duration, target_seconds) if status: status( f"Trecho real de {segment_duration:.3f}s; " f"montando {repeats} repeticoes completas..." ) concat_list = temporary_root / "concat.txt" concat_line = str(segment_path.resolve()).replace("\\", "/").replace("'", "'\\''") concat_list.write_text( "".join(f"file '{concat_line}'\n" for _ in range(repeats)), encoding="utf-8", ) loop_command = [ ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-f", "concat", "-safe", "0", "-i", str(concat_list), "-map", "0:v:0", "-map", "0:a:0?", "-vf", "setpts=PTS-STARTPTS", "-af", "asetpts=PTS-STARTPTS", "-shortest", "-avoid_negative_ts", "make_zero", "-c:v", "libx264", "-preset", "medium", "-crf", "18", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "192k", ] if _is_mp4_family(output_path): loop_command.extend(["-movflags", "+faststart"]) loop_command.append(str(output_path)) _run_checked(loop_command) output_duration = probe_video(output_path).duration if output_duration + max(0.05, segment_duration / 20) < target_seconds: raise VideoLoopError( f"A saida ficou menor que o alvo ({output_duration:.2f}s < {target_seconds:.2f}s)." ) return repeats, output_duration def process_video( input_path: Path, output_path: Path, target_seconds: float, min_loop_seconds: float, status: StatusCallback = None, ) -> ProcessResult: info, candidate, _ = analyze_video(input_path, min_loop_seconds, status) if status: status( f"Loop escolhido: {candidate.start:.2f}s ate {candidate.end:.2f}s " f"({candidate.duration:.2f}s)." ) repeats, output_duration = render_loop( input_path, output_path, candidate, target_seconds, status, ) return ProcessResult( input_path=input_path, output_path=output_path, source_duration=info.duration, output_duration=output_duration, candidate=candidate, repeats=repeats, ) def default_output_path(input_path: Path) -> Path: return input_path.with_name(f"{input_path.stem}_looped.mp4") def is_video_path(path: Path) -> bool: return path.is_file() and path.suffix.lower() in VIDEO_EXTENSIONS def _valid_generated_output(path: Path) -> bool: try: if not path.is_file() or path.stat().st_size < 1024: return False if path.suffix.lower() == ".gif": return True return probe_video(path).duration > 0.05 except (OSError, VideoLoopError): return False def video_files_in_folder(folder: Path) -> list[Path]: """Collect videos recursively while ignoring generated looping folders.""" if not folder.is_dir(): return [] files = [] for path in folder.rglob("*"): if not is_video_path(path): continue if any(part.casefold() == "looping" for part in path.parts): continue files.append(path) return sorted(files, key=lambda path: str(path).casefold()) def unique_video_paths(paths: list[Path]) -> list[Path]: result = [] seen = set() for path in paths: if not is_video_path(path): continue resolved = str(path.resolve()).casefold() if resolved in seen: continue seen.add(resolved) result.append(path) return result def sanitize_prefix(prefix: str) -> str: cleaned = re.sub(r"[^\w\-]+", "_", prefix.strip(), flags=re.UNICODE) return cleaned.strip("_") or "looping" def _filter_path(path: Path) -> str: value = path.resolve().as_posix() for character in (":", "'", ",", "[", "]"): value = value.replace(character, "\\" + character) return value def _position_expressions(position: str, overlay: bool = False) -> tuple[str, str]: margin = "24" if overlay: positions = { "top-left": (margin, margin), "top-right": (f"main_w-overlay_w-{margin}", margin), "bottom-left": (margin, f"main_h-overlay_h-{margin}"), "bottom-right": ( f"main_w-overlay_w-{margin}", f"main_h-overlay_h-{margin}", ), "center": ("(main_w-overlay_w)/2", "(main_h-overlay_h)/2"), } else: positions = { "top-left": (margin, margin), "top-right": (f"w-tw-{margin}", margin), "bottom-left": (margin, f"h-th-{margin}"), "bottom-right": (f"w-tw-{margin}", f"h-th-{margin}"), "center": ("(w-tw)/2", "(h-th)/2"), } return positions.get(position, positions["bottom-right"]) def _default_font_file() -> Optional[Path]: windows_fonts = Path(os.environ.get("WINDIR", "C:/Windows")) / "Fonts" candidates = ( windows_fonts / "arial.ttf", windows_fonts / "segoeui.ttf", Path("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"), Path("/usr/share/fonts/dejavu/DejaVuSans.ttf"), ) return next((path for path in candidates if path.is_file()), None) def _video_output_args(output_path: Path) -> list[str]: args = [ "-map_metadata", "-1", "-map_metadata:s", "-1", "-map_chapters", "-1", "-c:v", "libx264", "-preset", "medium", "-crf", "18", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "192k", ] if _is_mp4_family(output_path): args.extend(["-movflags", "+faststart"]) return args def strip_metadata(path: Path) -> None: """Remove container metadata without changing the encoded streams.""" ffmpeg = _find_executable("ffmpeg") temporary = path.with_name(f".{path.stem}.metadata_tmp{path.suffix}") temporary.unlink(missing_ok=True) command = [ ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-i", str(path), "-map", "0", "-map_metadata", "-1", "-map_metadata:s", "-1", "-map_chapters", "-1", "-c", "copy", "-fflags", "+bitexact", str(temporary), ] try: _run_checked(command) os.replace(temporary, path) finally: temporary.unlink(missing_ok=True) def _validate_watermark(config: Optional[WatermarkConfig]) -> WatermarkConfig: if config is None: raise VideoLoopError("Configure uma marca d'agua ou desative a versao com marca.") if config.mode == "text" and not config.text.strip(): raise VideoLoopError("Digite o texto da marca d'agua.") if config.mode == "image": if config.image_path is None or not config.image_path.is_file(): raise VideoLoopError("Escolha uma imagem valida para a marca d'agua.") if not 0.05 <= config.opacity <= 1.0: raise VideoLoopError("A opacidade deve ficar entre 0.05 e 1.0.") if config.font_size < 8 or config.font_size > 500: raise VideoLoopError("O tamanho do texto deve ficar entre 8 e 500.") if config.image_scale < 1 or config.image_scale > 100: raise VideoLoopError("A escala da imagem deve ficar entre 1 e 100%.") for name, value in (("X", config.x_percent), ("Y", config.y_percent)): if value is not None and not 0.0 <= value <= 100.0: raise VideoLoopError(f"A posicao {name} da marca d'agua deve ficar entre 0 e 100%.") return config def _clamped_percent(value: Optional[float]) -> Optional[float]: if value is None: return None return min(100.0, max(0.0, value)) def _percent_x_expression(percent: float, anchor: str) -> str: return f"(w*{percent / 100:.4f}-{anchor}w/2)" def _percent_y_expression(percent: float, anchor: str) -> str: return f"(h*{percent / 100:.4f}-{anchor}h/2)" def _text_watermark_filter(config: WatermarkConfig, text_file: Path) -> str: x, y = _position_expressions(config.position) if (percent := _clamped_percent(config.x_percent)) is not None: x = _percent_x_expression(percent, "t") if (percent := _clamped_percent(config.y_percent)) is not None: y = _percent_y_expression(percent, "t") font_file = _default_font_file() font_part = f"fontfile='{_filter_path(font_file)}':" if font_file else "" return ( f"drawtext={font_part}textfile='{_filter_path(text_file)}':" f"fontcolor=white@{config.opacity:.3f}:fontsize={config.font_size}:" f"borderw=2:bordercolor=black@0.55:shadowx=1:shadowy=1:x={x}:y={y}" ) def _image_overlay_expressions(config: WatermarkConfig) -> tuple[str, str]: x, y = _position_expressions(config.position, overlay=True) if (percent := _clamped_percent(config.x_percent)) is not None: x = f"(main_w*{percent / 100:.4f}-overlay_w/2)" if (percent := _clamped_percent(config.y_percent)) is not None: y = f"(main_h*{percent / 100:.4f}-overlay_h/2)" return x, y def _image_watermark_filter(config: WatermarkConfig) -> str: if config.image_path is None: raise VideoLoopError("Imagem de marca d'agua ausente.") x, y = _image_overlay_expressions(config) scale = config.image_scale / 100.0 return ( f"[1:v]format=rgba,scale=trunc(iw*{scale:.4f}/2)*2:-1," f"colorchannelmixer=aa={config.opacity:.3f}[watermark];" f"[0:v][watermark]overlay=x={x}:y={y}:eof_action=repeat:shortest=1[out]" ) def _watermark_chain( configs: tuple[WatermarkConfig, ...], text_directory: Path, ) -> tuple[str, tuple[Path, ...], str]: """Build one filter chain that applies every watermark layer in a single pass.""" filters: list[str] = [] image_inputs: list[Path] = [] current = "[0:v]" for index, config in enumerate(configs): output_label = f"[wm_{index}]" if config.mode == "text": text_file = text_directory / f"text_{index}.txt" text_file.write_text(config.text, encoding="utf-8") filters.append( f"{current}{_text_watermark_filter(config, text_file)}{output_label}" ) else: if config.image_path is None: raise VideoLoopError("Imagem de marca d'agua ausente.") input_index = len(image_inputs) + 1 image_inputs.append(config.image_path) x, y = _image_overlay_expressions(config) scale = config.image_scale / 100.0 image_label = f"[watermark_{index}]" filters.append( f"[{input_index}:v]format=rgba," f"scale=trunc(iw*{scale:.4f}/2)*2:-1," f"colorchannelmixer=aa={config.opacity:.3f}{image_label}" ) # shortest=1 ends the overlay exactly with the video; without it the # infinite image input keeps the encoder alive and freezes the tail. filters.append( f"{current}{image_label}overlay=x={x}:y={y}:" f"eof_action=repeat:shortest=1{output_label}" ) current = output_label return ";".join(filters), tuple(image_inputs), current def apply_watermarks( input_path: Path, output_path: Path, configs: tuple[WatermarkConfig, ...], ) -> None: """Apply every watermark layer in a single encoded pass. The previous version rendered one layer per pass and relied on ``-shortest`` with an infinite image input. FFmpeg could then keep encoding after the video ended, producing a frozen tail (video longer than audio). The chain here terminates with the main video and the duration is validated. """ if not configs: raise VideoLoopError("Nenhuma marca d'agua foi configurada.") configs = tuple(_validate_watermark(config) for config in configs) info = probe_video(input_path) if info.duration <= 0: raise VideoLoopError("Nao foi possivel medir a duracao do video.") ffmpeg = _find_executable("ffmpeg") output_path.parent.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix="watermark_") as temporary_directory: filters, image_inputs, final_label = _watermark_chain( configs, Path(temporary_directory) ) command = [ ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-i", str(input_path), ] for image_path in image_inputs: command.extend(["-loop", "1", "-i", str(image_path)]) command.extend( [ "-filter_complex", filters, "-map", final_label, "-map", "0:a:0?", "-t", f"{info.duration:.6f}", ] ) command.extend(_video_output_args(output_path)) command.append(str(output_path)) _run_checked(command) rendered = probe_video(output_path).duration tolerance = max(0.25, info.duration * 0.02) if abs(rendered - info.duration) > tolerance: raise VideoLoopError( "A marca d'agua alterou a duracao do video " f"({rendered:.2f}s em vez de {info.duration:.2f}s)." ) def apply_watermark( input_path: Path, output_path: Path, config: WatermarkConfig, ) -> None: """Create a watermarked MP4, supporting either text or an image.""" apply_watermarks(input_path, output_path, (config,)) def extract_frame( input_path: Path, output_png: Path, at_seconds: float = 0.5, max_side: int = 480, ) -> None: """Grab one clean frame, scaled to fit max_side, for the visual editor.""" ffmpeg = _find_executable("ffmpeg") output_png.parent.mkdir(parents=True, exist_ok=True) command = [ ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-ss", f"{max(0.0, at_seconds):.3f}", "-i", str(input_path), "-vf", f"scale={max_side}:{max_side}:force_original_aspect_ratio=decrease", "-frames:v", "1", "-an", str(output_png), ] _run_checked(command) def render_watermark_preview( input_path: Path, output_png: Path, configs: tuple[WatermarkConfig, ...], at_seconds: float = 0.5, ) -> None: """Render an exact one-frame preview using the same FFmpeg filters as output.""" if not configs: extract_frame(input_path, output_png, at_seconds=at_seconds, max_side=560) return configs = tuple(_validate_watermark(config) for config in configs) ffmpeg = _find_executable("ffmpeg") output_png.parent.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix="exact_watermark_preview_") as temporary_directory: filters, image_inputs, final_label = _watermark_chain( configs, Path(temporary_directory) ) command = [ ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-ss", f"{max(0.0, at_seconds):.3f}", "-i", str(input_path), ] for image_path in image_inputs: command.extend(["-loop", "1", "-i", str(image_path)]) command.extend( [ "-filter_complex", f"{filters};{final_label}format=rgb24[out]", "-map", "[out]", "-frames:v", "1", "-an", str(output_png), ] ) _run_checked(command) def convert_to_gif( input_path: Path, output_path: Path, fps: int = 24, width: Optional[int] = None, ) -> None: """Convert a video to a looping GIF using a two-pass palette.""" if fps < 1: raise VideoLoopError("O FPS do GIF deve ser maior que zero.") if width is not None and width < 1: raise VideoLoopError("A largura do GIF deve ser maior que zero.") ffmpeg = _find_executable("ffmpeg") output_path.parent.mkdir(parents=True, exist_ok=True) video_filter = f"fps={fps}" if width is not None: video_filter += f",scale={width}:-1:flags=lanczos" with tempfile.TemporaryDirectory(prefix="gif_palette_") as temporary_directory: palette = Path(temporary_directory) / "palette.png" palette_command = [ ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-i", str(input_path), "-vf", f"{video_filter},palettegen=stats_mode=diff", str(palette), ] _run_checked(palette_command) gif_command = [ ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-nostdin", "-i", str(input_path), "-i", str(palette), "-filter_complex", f"[0:v]{video_filter}[video];" f"[video][1:v]paletteuse=dither=sierra2_4a[out]", "-map", "[out]", "-an", "-loop", "0", "-map_metadata", "-1", "-map_chapters", "-1", str(output_path), ] _run_checked(gif_command) def _pipeline_gif_indices( options: PipelineOptions, total: int, selected_indices: Optional[set[int]], ) -> set[int]: if options.gif_mode == "none": return set() if options.gif_mode == "all": return set(range(total)) if options.gif_mode == "manual": return set(selected_indices or set()) if options.gif_mode == "random": count = max(0, min(options.gif_count, total)) return set(random.sample(range(total), count)) raise VideoLoopError("Modo de selecao de GIF desconhecido.") def process_pipeline( input_paths: list[Path], output_parent: Path, options: PipelineOptions, gif_selected_indices: Optional[set[int]] = None, watermark_indices: Optional[set[int]] = None, cancel_event: Optional[threading.Event] = None, status: StatusCallback = None, ) -> PipelineReport: """Run looping, variants, GIF conversion, renaming and metadata cleanup.""" paths = unique_video_paths(input_paths) if not paths: raise VideoLoopError("Adicione pelo menos um video para processar.") if not options.create_plain and not options.create_watermarked: raise VideoLoopError("Escolha pelo menos uma versao de video para criar.") if options.create_watermarked: watermark_configs = options.watermarks or ( (options.watermark,) if options.watermark is not None else () ) if not watermark_configs: raise VideoLoopError("Configure pelo menos uma marca d'agua.") watermark_configs = tuple(_validate_watermark(config) for config in watermark_configs) else: watermark_configs = () if options.gif_mode != "none" and options.gif_variant not in { "plain", "watermarked", "both", }: raise VideoLoopError("Versao de GIF desconhecida.") gif_indices = _pipeline_gif_indices(options, len(paths), gif_selected_indices) if options.gif_variant in {"watermarked", "both"} and options.gif_mode != "none" and not options.create_watermarked: raise VideoLoopError("GIF com marca exige a versao de video com marca.") prefix = sanitize_prefix(options.prefix) output_parent.mkdir(parents=True, exist_ok=True) loop_folder = output_parent / "looping" gif_folder = output_parent / "gif" loop_folder.mkdir(parents=True, exist_ok=True) gif_folder.mkdir(parents=True, exist_ok=True) watermarked_folder = loop_folder / "com_marca" if options.create_watermarked: watermarked_folder.mkdir(parents=True, exist_ok=True) outputs: list[Path] = [] gifs: list[Path] = [] failures: list[tuple[Path, str]] = [] skipped: list[Path] = [] created_videos: list[Path] = [] created_gifs: list[Path] = [] cancelled = False with tempfile.TemporaryDirectory(prefix="pipeline_work_") as temporary_directory: temporary_root = Path(temporary_directory) for index, input_path in enumerate(paths): if cancel_event is not None and cancel_event.is_set(): cancelled = True break filename = f"{prefix}_{index + 1:03d}.mp4" plain_path = loop_folder / filename if options.create_plain else None should_watermark = options.create_watermarked and ( watermark_indices is None or index in watermark_indices ) watermarked_path = ( watermarked_folder / filename if should_watermark else None ) gif_paths: list[tuple[str, Path]] = [] if index in gif_indices: if options.gif_variant in {"plain", "both"}: gif_paths.append(("plain", gif_folder / "sem_marca" / f"{Path(filename).stem}.gif")) if options.gif_variant in {"watermarked", "both"}: gif_paths.append(("watermarked", gif_folder / "com_marca" / f"{Path(filename).stem}.gif")) expected = [path for path in (plain_path, watermarked_path) if path is not None] expected.extend(path for _, path in gif_paths) if options.skip_existing and expected and all( _valid_generated_output(path) for path in expected ): skipped.append(input_path) outputs.extend(path for path in (plain_path, watermarked_path) if path is not None) gifs.extend(path for _, path in gif_paths) if status: status(f"[{index + 1}/{len(paths)}] Pulando existente: {input_path.name}") continue if status: status(f"[{index + 1}/{len(paths)}] Processando {input_path.name}...") try: working_path = temporary_root / f"working_{index + 1:03d}.mp4" if ( options.skip_existing and plain_path is not None and _valid_generated_output(plain_path) ): base_path = plain_path elif ( options.skip_existing and not options.create_plain and watermarked_path is not None and _valid_generated_output(watermarked_path) ): base_path = watermarked_path else: base_path = plain_path or working_path process_video( input_path, base_path, options.target_seconds, options.min_loop_seconds, status=status, ) if plain_path is not None: created_videos.append(plain_path) if options.clean_metadata: strip_metadata(plain_path) if ( should_watermark and watermarked_path is not None and (not _valid_generated_output(watermarked_path) or not options.skip_existing) ): apply_watermarks(base_path, watermarked_path, watermark_configs) created_videos.append(watermarked_path) if options.clean_metadata: strip_metadata(watermarked_path) if plain_path is not None: outputs.append(plain_path) if watermarked_path is not None: outputs.append(watermarked_path) original_watermarked_path: Optional[Path] = None if any(variant == "watermarked" for variant, _ in gif_paths): if status: status(f"[{index + 1}/{len(paths)}] Preparando marca do GIF original...") original_watermarked_path = ( temporary_root / f"gif_watermarked_{index + 1:03d}.mp4" ) apply_watermarks(input_path, original_watermarked_path, watermark_configs) if options.clean_metadata: strip_metadata(original_watermarked_path) for variant, gif_path in gif_paths: if options.skip_existing and _valid_generated_output(gif_path): gifs.append(gif_path) continue if variant == "plain": # GIFs intentionally use the original clip, not the extended loop. gif_source = input_path else: if original_watermarked_path is None: raise VideoLoopError( "Nao foi possivel preparar a marca d'agua do GIF original." ) gif_source = original_watermarked_path convert_to_gif( gif_source, gif_path, fps=options.gif_fps, width=options.gif_width, ) gifs.append(gif_path) created_gifs.append(gif_path) except Exception as exc: failures.append((input_path, str(exc))) if status: status(f"[{index + 1}/{len(paths)}] Falha: {input_path.name}: {exc}") return PipelineReport( output_folder=output_parent, loop_folder=loop_folder, gif_folder=gif_folder, outputs=tuple(outputs), gifs=tuple(gifs), failures=tuple(failures), skipped=tuple(skipped), cancelled=cancelled, created_videos=tuple(created_videos), created_gifs=tuple(created_gifs), ) def _format_seconds(seconds: float) -> str: minutes, remaining = divmod(max(0.0, seconds), 60.0) if minutes >= 60: hours, minutes = divmod(int(minutes), 60) return f"{hours}h {int(minutes):02d}m {remaining:04.1f}s" if minutes >= 1: return f"{int(minutes)}m {remaining:04.1f}s" return f"{remaining:.2f}s" TkBase = TkinterDnD.Tk if TkinterDnD is not None else tk.Tk class LegacyLoopApp(TkBase): def __init__(self) -> None: super().__init__() self.title("Detector de Loop de Video") self.geometry("1080x720") self.minsize(920, 640) self.protocol("WM_DELETE_WINDOW", self._on_close) self.input_paths: list[Path] = [] self.output_dir_var = tk.StringVar() self.target_var = tk.StringVar(value=str(int(DEFAULT_TARGET_SECONDS))) self.minimum_var = tk.StringVar(value=str(int(DEFAULT_MIN_LOOP_SECONDS))) self.skip_existing_var = tk.BooleanVar(value=True) self.prefix_var = tk.StringVar(value="looping") self.create_plain_var = tk.BooleanVar(value=True) self.create_watermarked_var = tk.BooleanVar(value=False) self.clean_metadata_var = tk.BooleanVar(value=True) self.watermark_mode_var = tk.StringVar(value="text") self.watermark_text_var = tk.StringVar(value="Python/Tabu") self.watermark_image_var = tk.StringVar() self.watermark_opacity_var = tk.StringVar(value="0.80") self.watermark_size_var = tk.StringVar(value="36") self.watermark_scale_var = tk.StringVar(value="25") self.watermark_position_var = tk.StringVar(value="inferior direito") self.watermark_scope_var = tk.StringVar(value="todos") self.watermark_x_var = tk.StringVar() self.watermark_y_var = tk.StringVar() self.preset_var = tk.StringVar() self.gif_mode_var = tk.StringVar(value="nao criar GIF") self.gif_random_count_var = tk.StringVar(value="1") self.gif_variant_var = tk.StringVar(value="sem marca") self.gif_width_var = tk.StringVar() self.status_var = tk.StringVar(value="Adicione um ou mais videos para comecar.") self.details_var = tk.StringVar() self.preview_status_var = tk.StringVar(value="O preview aparecera depois do processamento.") self.cancel_event = threading.Event() self.processing = False self.closing = False self.last_output_folder: Optional[Path] = None self.preview_tempdir: Optional[tempfile.TemporaryDirectory] = None self.preview_frames: list[Path] = [] self.preview_index = 0 self.preview_playing = False self.preview_after_id: Optional[str] = None self.preview_photo = None self.editor_mode = False self.editor_bg_photo = None self.editor_overlay_photo = None self.editor_canvas = None self.editor_items: list[int] = [] self.editor_frame_size = (0, 0) self.editor_source_size = (0, 0) self.presets_file = Path(__file__).with_name("watermark_presets.json") self._build_widgets() self._enable_drag_and_drop() self._refresh_preset_combo() def _build_widgets(self) -> None: root = ttk.Frame(self, padding=18) root.pack(fill="both", expand=True) root.columnconfigure(0, weight=3) root.columnconfigure(1, weight=2) root.rowconfigure(2, weight=1) ttk.Label( root, text="Detector e Amplificador de Loop", font=("Segoe UI", 17, "bold"), ).grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 3)) ttk.Label( root, text="Selecione varios videos, processe em lote e confira o resultado no preview.", ).grid(row=1, column=0, columnspan=2, sticky="w", pady=(0, 0)) left = ttk.LabelFrame(root, text="Videos para processar", padding=12) left.grid(row=2, column=0, sticky="nsew", padx=(0, 10), pady=(18, 0)) left.columnconfigure(0, weight=1) left.rowconfigure(1, weight=1) drop_text = ( "Arraste arquivos ou pastas para esta lista" if DND_FILES is not None else "Use os botoes abaixo para adicionar arquivos ou pastas" ) self.drop_hint = ttk.Label(left, text=drop_text) self.drop_hint.grid(row=0, column=0, sticky="w", pady=(0, 7)) list_frame = ttk.Frame(left) list_frame.grid(row=1, column=0, sticky="nsew") list_frame.columnconfigure(0, weight=1) list_frame.rowconfigure(0, weight=1) self.file_list = tk.Listbox( list_frame, selectmode=tk.EXTENDED, exportselection=False, activestyle="none", height=12, ) self.file_list.grid(row=0, column=0, sticky="nsew") file_scroll = ttk.Scrollbar(list_frame, orient="vertical", command=self.file_list.yview) file_scroll.grid(row=0, column=1, sticky="ns") self.file_list.configure(yscrollcommand=file_scroll.set) file_buttons = ttk.Frame(left) file_buttons.grid(row=2, column=0, sticky="ew", pady=(9, 0)) for column in range(4): file_buttons.columnconfigure(column, weight=1) self.add_files_button = ttk.Button( file_buttons, text="Adicionar arquivos", command=self._choose_files ) self.add_files_button.grid(row=0, column=0, sticky="ew", padx=(0, 4)) self.add_folder_button = ttk.Button( file_buttons, text="Adicionar pasta", command=self._choose_input_folder ) self.add_folder_button.grid(row=0, column=1, sticky="ew", padx=4) self.remove_button = ttk.Button( file_buttons, text="Remover", command=self._remove_selected ) self.remove_button.grid(row=0, column=2, sticky="ew", padx=4) self.clear_button = ttk.Button( file_buttons, text="Limpar", command=self._clear_files ) self.clear_button.grid(row=0, column=3, sticky="ew", padx=(4, 0)) self.file_buttons = [ self.add_files_button, self.add_folder_button, self.remove_button, self.clear_button, ] output_frame = ttk.LabelFrame(left, text="Destino", padding=10) output_frame.grid(row=3, column=0, sticky="ew", pady=(14, 0)) output_frame.columnconfigure(1, weight=1) ttk.Label(output_frame, text="Pasta escolhida:").grid( row=0, column=0, sticky="w", padx=(0, 8) ) self.output_entry = ttk.Entry(output_frame, textvariable=self.output_dir_var) self.output_entry.grid(row=0, column=1, sticky="ew") self.output_button = ttk.Button( output_frame, text="Escolher...", command=self._choose_output_folder ) self.output_button.grid(row=0, column=2, padx=(8, 0)) ttk.Label( output_frame, text="Os arquivos serao criados automaticamente em: pasta escolhida\\looping", ).grid(row=1, column=0, columnspan=3, sticky="w", pady=(7, 0)) options = ttk.Notebook(left) options.grid(row=4, column=0, sticky="ew", pady=(12, 0)) process_tab = ttk.Frame(options, padding=9) process_tab.columnconfigure(1, weight=1) options.add(process_tab, text="Processamento") ttk.Label(process_tab, text="Duracao desejada (s):").grid( row=0, column=0, sticky="w", padx=(0, 8), pady=2 ) ttk.Entry(process_tab, textvariable=self.target_var, width=10).grid( row=0, column=1, sticky="w", pady=2 ) ttk.Label(process_tab, text="Loop minimo (s):").grid( row=1, column=0, sticky="w", padx=(0, 8), pady=2 ) ttk.Entry(process_tab, textvariable=self.minimum_var, width=10).grid( row=1, column=1, sticky="w", pady=2 ) ttk.Label(process_tab, text="Prefixo dos arquivos:").grid( row=2, column=0, sticky="w", padx=(0, 8), pady=2 ) ttk.Entry(process_tab, textvariable=self.prefix_var, width=18).grid( row=2, column=1, sticky="w", pady=2 ) self.plain_check = ttk.Checkbutton( process_tab, text="Criar versao sem marca d'agua", variable=self.create_plain_var, ) self.plain_check.grid(row=3, column=0, columnspan=2, sticky="w", pady=(4, 0)) self.watermarked_check = ttk.Checkbutton( process_tab, text="Criar versao com marca d'agua", variable=self.create_watermarked_var, ) self.watermarked_check.grid(row=4, column=0, columnspan=2, sticky="w") self.metadata_check = ttk.Checkbutton( process_tab, text="Limpar metadados das saidas", variable=self.clean_metadata_var, ) self.metadata_check.grid(row=5, column=0, columnspan=2, sticky="w") self.skip_check = ttk.Checkbutton( process_tab, text="Pular resultados que ja existem", variable=self.skip_existing_var, ) self.skip_check.grid(row=6, column=0, columnspan=2, sticky="w") watermark_tab = ttk.Frame(options, padding=9) watermark_tab.columnconfigure(1, weight=1) options.add(watermark_tab, text="Marca d'agua") ttk.Label(watermark_tab, text="Marca salva:").grid( row=0, column=0, sticky="w", pady=2 ) self.preset_combo = ttk.Combobox( watermark_tab, textvariable=self.preset_var, state="readonly", width=22, ) self.preset_combo.grid(row=0, column=1, sticky="w", pady=2) self.preset_combo.bind("<>", self._load_selected_preset) self.preset_save_button = ttk.Button( watermark_tab, text="Salvar", command=self._save_preset ) self.preset_save_button.grid(row=0, column=2, padx=(6, 0), pady=2) self.preset_delete_button = ttk.Button( watermark_tab, text="Excluir", command=self._delete_selected_preset ) self.preset_delete_button.grid(row=0, column=3, padx=(4, 0), pady=2) ttk.Label(watermark_tab, text="Tipo:").grid(row=1, column=0, sticky="w") ttk.Radiobutton( watermark_tab, text="Texto", value="text", variable=self.watermark_mode_var, command=self._toggle_watermark_fields, ).grid(row=1, column=1, sticky="w") ttk.Radiobutton( watermark_tab, text="Imagem", value="image", variable=self.watermark_mode_var, command=self._toggle_watermark_fields, ).grid(row=1, column=2, sticky="w") ttk.Label(watermark_tab, text="Texto:").grid(row=2, column=0, sticky="w", pady=2) self.watermark_text_entry = ttk.Entry( watermark_tab, textvariable=self.watermark_text_var, width=24 ) self.watermark_text_entry.grid(row=2, column=1, columnspan=2, sticky="ew", pady=2) ttk.Label(watermark_tab, text="Imagem:").grid(row=3, column=0, sticky="w", pady=2) self.watermark_image_entry = ttk.Entry( watermark_tab, textvariable=self.watermark_image_var, width=24 ) self.watermark_image_entry.grid(row=3, column=1, sticky="ew", pady=2) self.watermark_image_button = ttk.Button( watermark_tab, text="Escolher", command=self._choose_watermark_image ) self.watermark_image_button.grid(row=3, column=2, padx=(6, 0), pady=2) ttk.Label(watermark_tab, text="Opacidade:").grid(row=4, column=0, sticky="w", pady=2) self.watermark_opacity_entry = ttk.Entry( watermark_tab, textvariable=self.watermark_opacity_var, width=8 ) self.watermark_opacity_entry.grid(row=4, column=1, sticky="w", pady=2) ttk.Label(watermark_tab, text="Tamanho texto:").grid(row=5, column=0, sticky="w", pady=2) self.watermark_size_entry = ttk.Entry( watermark_tab, textvariable=self.watermark_size_var, width=8 ) self.watermark_size_entry.grid(row=5, column=1, sticky="w", pady=2) ttk.Label(watermark_tab, text="Escala imagem (%):").grid(row=6, column=0, sticky="w", pady=2) self.watermark_scale_entry = ttk.Entry( watermark_tab, textvariable=self.watermark_scale_var, width=8 ) self.watermark_scale_entry.grid(row=6, column=1, sticky="w", pady=2) ttk.Label(watermark_tab, text="X (%):").grid(row=7, column=0, sticky="w", pady=2) self.watermark_x_entry = ttk.Entry( watermark_tab, textvariable=self.watermark_x_var, width=8 ) self.watermark_x_entry.grid(row=7, column=1, sticky="w", pady=2) ttk.Label(watermark_tab, text="Y (%):").grid(row=8, column=0, sticky="w", pady=2) self.watermark_y_entry = ttk.Entry( watermark_tab, textvariable=self.watermark_y_var, width=8 ) self.watermark_y_entry.grid(row=8, column=1, sticky="w", pady=2) ttk.Label(watermark_tab, text="0 = borda esquerda/topo, 100 = direita/base").grid( row=9, column=0, columnspan=3, sticky="w", pady=(3, 0) ) ttk.Label(watermark_tab, text="Posicao padrao:").grid(row=10, column=0, sticky="w", pady=2) self.watermark_position_combo = ttk.Combobox( watermark_tab, textvariable=self.watermark_position_var, values=tuple(POSITION_LABELS), state="readonly", width=16, ) self.watermark_position_combo.grid(row=10, column=1, sticky="w", pady=2) ttk.Label(watermark_tab, text="Aplicar em:").grid(row=11, column=0, sticky="w", pady=2) self.watermark_scope_combo = ttk.Combobox( watermark_tab, textvariable=self.watermark_scope_var, values=tuple(SCOPE_LABELS), state="readonly", width=16, ) self.watermark_scope_combo.grid(row=11, column=1, sticky="w", pady=2) ttk.Label( watermark_tab, text="todos = todos; selecionados = itens marcados na lista" ).grid(row=12, column=0, columnspan=3, sticky="w", pady=(3, 0)) self.watermark_preview_button = ttk.Button( watermark_tab, text="Editor visual da marca d'agua", command=self._open_watermark_editor, ) self.watermark_preview_button.grid( row=13, column=0, columnspan=3, sticky="ew", pady=(6, 0) ) for entry in ( self.watermark_text_entry, self.watermark_image_entry, self.watermark_opacity_entry, self.watermark_size_entry, self.watermark_scale_entry, self.watermark_x_entry, self.watermark_y_entry, ): entry.bind("", self._update_editor_overlay) self.watermark_position_combo.bind( "<>", self._update_editor_overlay ) gif_tab = ttk.Frame(options, padding=9) gif_tab.columnconfigure(1, weight=1) options.add(gif_tab, text="GIF") ttk.Label(gif_tab, text="Converter para GIF:").grid(row=0, column=0, sticky="w", pady=2) self.gif_mode_combo = ttk.Combobox( gif_tab, textvariable=self.gif_mode_var, values=tuple(GIF_MODE_LABELS), state="readonly", width=16, ) self.gif_mode_combo.grid(row=0, column=1, sticky="w", pady=2) self.gif_mode_combo.bind("<>", self._toggle_gif_fields) ttk.Label(gif_tab, text="aleatorios: quantidade").grid(row=1, column=0, sticky="w", pady=2) self.gif_random_entry = ttk.Entry(gif_tab, textvariable=self.gif_random_count_var, width=8) self.gif_random_entry.grid(row=1, column=1, sticky="w", pady=2) ttk.Label(gif_tab, text="Fonte do GIF:").grid(row=2, column=0, sticky="w", pady=2) self.gif_variant_combo = ttk.Combobox( gif_tab, textvariable=self.gif_variant_var, values=tuple(GIF_VARIANT_LABELS), state="readonly", width=16, ) self.gif_variant_combo.grid(row=2, column=1, sticky="w", pady=2) ttk.Label(gif_tab, text="FPS:").grid(row=3, column=0, sticky="w", pady=2) ttk.Label(gif_tab, text="24 (fixo)").grid(row=3, column=1, sticky="w", pady=2) ttk.Label(gif_tab, text="Largura:").grid(row=4, column=0, sticky="w", pady=2) self.gif_width_entry = ttk.Entry(gif_tab, textvariable=self.gif_width_var, width=10) self.gif_width_entry.grid(row=4, column=1, sticky="w", pady=2) ttk.Label(gif_tab, text="vazio = mesma resolucao do video").grid( row=5, column=0, columnspan=2, sticky="w", pady=(3, 0) ) self._toggle_watermark_fields() self._toggle_gif_fields() actions = ttk.Frame(left) actions.grid(row=5, column=0, sticky="ew", pady=(14, 0)) actions.columnconfigure(0, weight=1) actions.columnconfigure(1, weight=1) self.process_button = ttk.Button( actions, text="Encontrar e criar loopings", command=self._start ) self.process_button.grid(row=0, column=0, sticky="ew", padx=(0, 5)) self.cancel_button = ttk.Button( actions, text="Parar depois do atual", command=self._cancel, state="disabled" ) self.cancel_button.grid(row=0, column=1, sticky="ew", padx=(5, 0)) self.progress = ttk.Progressbar(actions, mode="indeterminate") self.progress.grid(row=1, column=0, columnspan=2, sticky="ew", pady=(10, 0)) preview = ttk.LabelFrame(root, text="Preview do resultado", padding=12) preview.grid(row=2, column=1, sticky="nsew", pady=(18, 0)) preview.columnconfigure(0, weight=1) preview.rowconfigure(0, weight=1) self.preview_frame_label = ttk.Label( preview, text="Nenhum resultado carregado.", anchor="center", justify="center", ) self.preview_frame_label.grid(row=0, column=0, sticky="nsew", pady=(0, 10)) self.preview_editor_canvas = tk.Canvas( preview, highlightthickness=0, background="#1e1e1e" ) self.preview_editor_canvas.bind("", self._editor_press) self.preview_editor_canvas.bind("", self._editor_drag) self.editor_canvas = self.preview_editor_canvas ttk.Label(preview, textvariable=self.preview_status_var).grid( row=1, column=0, sticky="w", pady=(0, 8) ) preview_buttons = ttk.Frame(preview) preview_buttons.grid(row=2, column=0, sticky="ew") preview_buttons.columnconfigure(1, weight=1) self.previous_button = ttk.Button( preview_buttons, text="Anterior", command=self._previous_preview, state="disabled" ) self.previous_button.grid(row=0, column=0, sticky="ew", padx=(0, 3)) self.play_button = ttk.Button( preview_buttons, text="Reproduzir", command=self._toggle_preview, state="disabled" ) self.play_button.grid(row=0, column=1, sticky="ew", padx=3) self.next_button = ttk.Button( preview_buttons, text="Proximo", command=self._next_preview, state="disabled" ) self.next_button.grid(row=0, column=2, sticky="ew", padx=(3, 0)) self.open_folder_button = ttk.Button( preview, text="Abrir pasta looping", command=self._open_output_folder, state="disabled", ) self.open_folder_button.grid(row=3, column=0, sticky="ew", pady=(10, 0)) ttk.Separator(root).grid(row=3, column=0, columnspan=2, sticky="ew", pady=(14, 8)) ttk.Label(root, textvariable=self.status_var, wraplength=1020).grid( row=4, column=0, columnspan=2, sticky="w" ) ttk.Label(root, textvariable=self.details_var, wraplength=1020).grid( row=5, column=0, columnspan=2, sticky="w", pady=(4, 0) ) log_frame = ttk.LabelFrame(root, text="Log do processamento", padding=5) log_frame.grid(row=6, column=0, columnspan=2, sticky="ew", pady=(6, 0)) log_frame.columnconfigure(0, weight=1) self.log_text = tk.Text(log_frame, height=4, state="disabled", wrap="word") self.log_text.grid(row=0, column=0, sticky="ew") log_scroll = ttk.Scrollbar(log_frame, orient="vertical", command=self.log_text.yview) log_scroll.grid(row=0, column=1, sticky="ns") self.log_text.configure(yscrollcommand=log_scroll.set) def _toggle_watermark_fields(self) -> None: text_state = "normal" if self.watermark_mode_var.get() == "text" else "disabled" image_state = "normal" if self.watermark_mode_var.get() == "image" else "disabled" self.watermark_text_entry.configure(state=text_state) self.watermark_image_entry.configure(state=image_state) self.watermark_image_button.configure(state=image_state) self._update_editor_overlay() def _toggle_gif_fields(self, _event: object = None) -> None: mode = GIF_MODE_LABELS.get(self.gif_mode_var.get(), "none") random_state = "normal" if mode == "random" else "disabled" variant_state = "normal" if mode != "none" else "disabled" self.gif_random_entry.configure(state=random_state) self.gif_variant_combo.configure( state="readonly" if variant_state == "normal" else "disabled" ) self.gif_width_entry.configure(state=variant_state) def _choose_watermark_image(self) -> None: path = filedialog.askopenfilename( title="Escolha a imagem da marca d'agua", filetypes=[ ("Imagens", "*.png *.jpg *.jpeg *.webp *.bmp"), ("Todos os arquivos", "*.*"), ], ) if path: self.watermark_image_var.set(path) def _build_watermark_config(self) -> WatermarkConfig: try: opacity = float(self.watermark_opacity_var.get().replace(",", ".")) font_size = int(self.watermark_size_var.get()) image_scale = int(self.watermark_scale_var.get()) except ValueError as exc: raise VideoLoopError( "Opacidade, tamanho do texto e escala precisam ser numericos." ) from exc x_text = self.watermark_x_var.get().strip() y_text = self.watermark_y_var.get().strip() try: x_percent = float(x_text.replace(",", ".")) if x_text else None y_percent = float(y_text.replace(",", ".")) if y_text else None except ValueError as exc: raise VideoLoopError( "As posicoes X e Y da marca d'agua devem ser numeros de 0 a 100." ) from exc image_text = self.watermark_image_var.get().strip() return WatermarkConfig( mode=self.watermark_mode_var.get(), text=self.watermark_text_var.get(), image_path=Path(image_text) if image_text else None, opacity=opacity, font_size=font_size, image_scale=image_scale, position=POSITION_LABELS.get(self.watermark_position_var.get(), "bottom-right"), x_percent=x_percent, y_percent=y_percent, ) def _open_watermark_editor(self) -> None: if not self.input_paths: messagebox.showwarning("Editor visual", "Adicione um video primeiro.") return selected = self.file_list.curselection() input_path = self.input_paths[selected[0]] if selected else self.input_paths[0] try: source_info = probe_video(input_path) except VideoLoopError as exc: messagebox.showerror("Editor visual", str(exc)) return self._stop_preview_playback() if self.preview_tempdir is not None: self.preview_tempdir.cleanup() self.preview_tempdir = tempfile.TemporaryDirectory(prefix="watermark_editor_") frame_path = Path(self.preview_tempdir.name) / "frame.png" self._set_preview_mode("editor") self.editor_canvas.delete("all") self.editor_items = [] self.editor_clean_image = None self.editor_exact_image = None self.editor_bg_image = None self.editor_bg_photo = None self.editor_overlay_photo = None self.preview_status_var.set("Carregando quadro do video...") threading.Thread( target=self._editor_worker, args=(input_path, frame_path, source_info), daemon=True, ).start() def _editor_worker( self, input_path: Path, frame_path: Path, source_info: VideoInfo, ) -> None: try: extract_frame(input_path, frame_path, at_seconds=0.5, max_side=480) except Exception as exc: self._safe_after(self._preview_failed, str(exc)) return self._safe_after(self._editor_frame_ready, frame_path, source_info) def _editor_frame_ready(self, frame_path: Path, source_info: VideoInfo) -> None: if Image is None or ImageTk is None: self.preview_status_var.set( "Editor visual precisa do Pillow: python -m pip install Pillow" ) return try: with Image.open(frame_path) as image: background = ImageTk.PhotoImage(image.convert("RGB")) except (OSError, tk.TclError) as exc: self.preview_status_var.set(f"Editor visual: {exc}") return canvas = self.preview_editor_canvas self.editor_bg_photo = background canvas.configure(width=background.width(), height=background.height()) canvas.delete("all") self.editor_items = [canvas.create_image(0, 0, image=background, anchor="nw")] self.editor_frame_size = (background.width(), background.height()) self.editor_source_size = (source_info.width or 1, source_info.height or 1) if not self.watermark_x_var.get().strip(): self.watermark_x_var.set("50.0") if not self.watermark_y_var.get().strip(): self.watermark_y_var.set("80.0") self._update_editor_overlay() self.preview_status_var.set( "Arraste a marca d'agua ate o lugar certo; o X/Y sao atualizados ao vivo." ) def _set_preview_mode(self, mode: str) -> None: if mode == "editor": self.preview_frame_label.grid_remove() self.preview_editor_canvas.grid(row=0, column=0, sticky="nsew", pady=(0, 10)) self.editor_mode = True else: self.preview_editor_canvas.grid_remove() self.preview_frame_label.grid(row=0, column=0, sticky="nsew", pady=(0, 10)) self.editor_mode = False def _editor_opacity(self) -> int: try: opacity = float(self.watermark_opacity_var.get().replace(",", ".")) except ValueError: opacity = 0.8 return int(min(1.0, max(0.05, opacity)) * 255) def _editor_image_scale(self) -> int: try: return min(100, max(1, int(self.watermark_scale_var.get()))) except ValueError: return 25 def _update_editor_overlay(self, _event: object = None) -> None: if ( not self.editor_mode or self.editor_canvas is None or self.editor_bg_photo is None or Image is None or ImageTk is None ): return frame_w, frame_h = self.editor_frame_size if frame_w <= 0 or frame_h <= 0: return try: fx = min(1.0, max(0.0, float(self.watermark_x_var.get().replace(",", ".")) / 100.0)) fy = min(1.0, max(0.0, float(self.watermark_y_var.get().replace(",", ".")) / 100.0)) except ValueError: return overlay = Image.new("RGBA", (frame_w, frame_h), (0, 0, 0, 0)) alpha = self._editor_opacity() if self.watermark_mode_var.get() == "text": self._draw_editor_text(overlay, fx, fy, alpha) else: self._draw_editor_image(overlay, fx, fy, alpha) self.editor_overlay_photo = ImageTk.PhotoImage(overlay) canvas = self.editor_canvas for item in self.editor_items[1:]: canvas.delete(item) self.editor_items = self.editor_items[:1] self.editor_items.append( canvas.create_image(0, 0, image=self.editor_overlay_photo, anchor="nw") ) def _draw_editor_text( self, overlay: "Image.Image", fx: float, fy: float, alpha: int, ) -> None: draw = ImageDraw.Draw(overlay) frame_w, frame_h = self.editor_frame_size source_w = self.editor_source_size[0] or frame_w factor = frame_w / max(1, source_w) try: size = max(6, int(float(self.watermark_size_var.get()) * factor)) except ValueError: size = max(6, int(36 * factor)) font = None font_path = _default_font_file() if ImageFont is not None and font_path: try: font = ImageFont.truetype(str(font_path), size) except OSError: font = None if font is None: font = ImageFont.load_default() draw.multiline_text( (fx * frame_w, fy * frame_h), self.watermark_text_var.get() or " ", font=font, fill=(255, 255, 255, alpha), anchor="mm", align="center", spacing=4, stroke_width=2, stroke_fill=(0, 0, 0, int(alpha * 0.55)), ) def _draw_editor_image( self, overlay: "Image.Image", fx: float, fy: float, alpha: int, ) -> None: path_text = self.watermark_image_var.get().strip() if not path_text: return try: stamp = Image.open(path_text).convert("RGBA") except OSError: return frame_w, frame_h = self.editor_frame_size target_w = max(8, int(frame_w * self._editor_image_scale() / 100.0)) target_h = max(1, int(stamp.height * target_w / max(1, stamp.width))) stamp = stamp.resize((target_w, target_h)) stamp.putalpha(stamp.split()[3].point(lambda value: value * alpha // 255)) x = int(min(max(0, frame_w - stamp.width), fx * frame_w - stamp.width / 2)) y = int(min(max(0, frame_h - stamp.height), fy * frame_h - stamp.height / 2)) overlay.alpha_composite(stamp, (x, y)) def _editor_press(self, event: object) -> None: self._editor_drag(event) def _editor_drag(self, event: object) -> None: if not self.editor_mode or self.editor_frame_size[0] <= 0: return frame_w, frame_h = self.editor_frame_size fx = min(1.0, max(0.0, event.x / frame_w)) fy = min(1.0, max(0.0, event.y / frame_h)) self.watermark_x_var.set(f"{fx * 100:.1f}") self.watermark_y_var.set(f"{fy * 100:.1f}") self._update_editor_overlay() def _read_presets(self) -> dict: if not self.presets_file.is_file(): return {} try: data = json.loads(self.presets_file.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): return {} return data if isinstance(data, dict) else {} def _write_presets(self, presets: dict) -> None: self.presets_file.write_text( json.dumps(presets, ensure_ascii=False, indent=2), encoding="utf-8", ) def _refresh_preset_combo(self) -> None: presets = self._read_presets() self.preset_combo.configure(values=tuple(sorted(presets))) if self.preset_var.get() not in presets: self.preset_var.set("") def _save_preset(self) -> None: name = (simpledialog.askstring( "Salvar marca d'agua", "Nome da marca d'agua:", parent=self ) or "").strip() if not name: return presets = self._read_presets() presets[name] = { "mode": self.watermark_mode_var.get(), "text": self.watermark_text_var.get(), "image_path": self.watermark_image_var.get(), "opacity": self.watermark_opacity_var.get(), "font_size": self.watermark_size_var.get(), "image_scale": self.watermark_scale_var.get(), "position": self.watermark_position_var.get(), "x_percent": self.watermark_x_var.get(), "y_percent": self.watermark_y_var.get(), } try: self._write_presets(presets) except OSError as exc: messagebox.showerror("Marca d'agua", f"Nao foi possivel salvar: {exc}") return self._refresh_preset_combo() self.preset_var.set(name) self.status_var.set(f"Marca d'agua '{name}' salva.") def _load_selected_preset(self, _event: object = None) -> None: name = self.preset_var.get() preset = self._read_presets().get(name) if not preset: return self.watermark_mode_var.set(preset.get("mode", "text")) self.watermark_text_var.set(preset.get("text", "")) self.watermark_image_var.set(preset.get("image_path", "")) self.watermark_opacity_var.set(str(preset.get("opacity", "0.80"))) self.watermark_size_var.set(str(preset.get("font_size", "36"))) self.watermark_scale_var.set(str(preset.get("image_scale", "25"))) self.watermark_position_var.set(preset.get("position", "inferior direito")) self.watermark_x_var.set(str(preset.get("x_percent", ""))) self.watermark_y_var.set(str(preset.get("y_percent", ""))) self._toggle_watermark_fields() self._update_editor_overlay() self.status_var.set(f"Marca d'agua '{name}' carregada.") def _delete_selected_preset(self) -> None: name = self.preset_var.get() if not name: messagebox.showwarning("Marca d'agua", "Escolha uma marca salva para excluir.") return presets = self._read_presets() presets.pop(name, None) try: self._write_presets(presets) except OSError as exc: messagebox.showerror("Marca d'agua", f"Nao foi possivel excluir: {exc}") return self._refresh_preset_combo() self.status_var.set(f"Marca d'agua '{name}' excluida.") def _enable_drag_and_drop(self) -> None: if DND_FILES is None: return for widget in (self.file_list, self.drop_hint): try: widget.drop_target_register(DND_FILES) widget.dnd_bind("<>", self._on_drop) except tk.TclError: pass def _on_drop(self, event: object) -> str: dropped = getattr(event, "data", "") paths = [Path(item) for item in self.tk.splitlist(dropped)] self._add_paths(paths) return "break" def _add_paths(self, paths: list[Path]) -> None: found: list[Path] = [] for path in paths: if path.is_dir(): found.extend(video_files_in_folder(path)) elif is_video_path(path): found.append(path) if not found: messagebox.showwarning( "Nenhum video encontrado", "Escolha arquivos de video ou uma pasta que contenha videos.", ) return self.input_paths = unique_video_paths(self.input_paths + found) self._refresh_file_list() if not self.output_dir_var.get().strip(): self.output_dir_var.set(str(self.input_paths[0].parent)) self.status_var.set(f"{len(self.input_paths)} video(s) na fila.") def _refresh_file_list(self) -> None: self.file_list.delete(0, tk.END) for path in self.input_paths: self.file_list.insert(tk.END, str(path)) def _choose_files(self) -> None: paths = filedialog.askopenfilenames( title="Escolha um ou mais videos", filetypes=[ ("Videos", "*.mp4 *.mov *.mkv *.avi *.webm *.m4v *.wmv *.flv"), ("Todos os arquivos", "*.*"), ], ) self._add_paths([Path(path) for path in paths]) def _choose_input_folder(self) -> None: path = filedialog.askdirectory(title="Escolha uma pasta com videos") if path: self._add_paths([Path(path)]) def _remove_selected(self) -> None: selected = set(self.file_list.curselection()) if not selected: return self.input_paths = [ path for index, path in enumerate(self.input_paths) if index not in selected ] self._refresh_file_list() self.status_var.set(f"{len(self.input_paths)} video(s) na fila.") def _clear_files(self) -> None: self.input_paths.clear() self._refresh_file_list() self.status_var.set("Fila limpa. Adicione um ou mais videos.") def _choose_output_folder(self) -> None: initial = self.output_dir_var.get().strip() or None path = filedialog.askdirectory(title="Escolha a pasta onde salvar os loopings", initialdir=initial) if path: self.output_dir_var.set(path) def _start(self) -> None: try: paths = list(self.input_paths) output_text = self.output_dir_var.get().strip() output_parent = Path(output_text).expanduser() if output_text else None target = float(self.target_var.get().replace(",", ".")) minimum = float(self.minimum_var.get().replace(",", ".")) if not paths: raise VideoLoopError("Adicione pelo menos um video.") if output_parent is None or not output_parent.is_dir(): raise VideoLoopError("Escolha uma pasta de saida valida.") if not math.isfinite(target) or not math.isfinite(minimum): raise VideoLoopError("As duracoes precisam ser numeros validos.") if not math.isfinite(target) or not math.isfinite(minimum): raise VideoLoopError("As duracoes precisam ser numeros validos.") if target <= 0 or minimum <= 0: raise VideoLoopError("As duracoes devem ser maiores que zero.") try: gif_count = int(self.gif_random_count_var.get()) except ValueError as exc: raise VideoLoopError("A quantidade aleatoria de GIFs deve ser inteira.") from exc width_text = self.gif_width_var.get().strip() try: gif_width = int(width_text) if width_text else None except ValueError as exc: raise VideoLoopError("A largura do GIF deve ser um numero inteiro.") from exc if gif_width is not None and gif_width < 1: raise VideoLoopError("A largura do GIF deve ser maior que zero.") gif_mode = GIF_MODE_LABELS.get(self.gif_mode_var.get(), "none") gif_variant = GIF_VARIANT_LABELS.get(self.gif_variant_var.get(), "plain") gif_selected = set(self.file_list.curselection()) if gif_mode == "manual" and not gif_selected: raise VideoLoopError("Selecione na lista os videos que serao convertidos em GIF.") if gif_mode == "random" and gif_count < 1: raise VideoLoopError("Informe quantos GIFs aleatorios deseja criar.") watermark_selected = set(self.file_list.curselection()) watermark_scope = SCOPE_LABELS.get(self.watermark_scope_var.get(), "all") if self.create_watermarked_var.get() and watermark_scope == "selected": if not watermark_selected: raise VideoLoopError("Selecione os videos que receberao marca d'agua.") watermark_indices = watermark_selected else: watermark_indices = None watermark = ( _validate_watermark(self._build_watermark_config()) if self.create_watermarked_var.get() else None ) if not self.create_plain_var.get() and not self.create_watermarked_var.get(): raise VideoLoopError("Escolha pelo menos uma versao de video para criar.") if gif_variant in {"plain", "both"} and gif_mode != "none" and not self.create_plain_var.get(): raise VideoLoopError("GIF sem marca exige a versao sem marca ativada.") if gif_variant in {"watermarked", "both"} and gif_mode != "none" and not self.create_watermarked_var.get(): raise VideoLoopError("GIF com marca exige a versao com marca ativada.") options = PipelineOptions( target_seconds=target, min_loop_seconds=minimum, prefix=sanitize_prefix(self.prefix_var.get()), create_plain=self.create_plain_var.get(), create_watermarked=self.create_watermarked_var.get(), watermark=watermark, clean_metadata=self.clean_metadata_var.get(), gif_mode=gif_mode, gif_count=gif_count, gif_variant=gif_variant, gif_fps=24, gif_width=gif_width, skip_existing=self.skip_existing_var.get(), ) except ValueError: messagebox.showerror("Valor invalido", "Digite duracoes como 20 ou 4.5.") return except VideoLoopError as exc: messagebox.showerror("Nao foi possivel iniciar", str(exc)) return self.cancel_event.clear() self.processing = True self._set_processing_state(True) self._clear_log() self.details_var.set("") self.status_var.set("Iniciando processamento em lote...") worker = threading.Thread( target=self._worker, args=(paths, output_parent, options, gif_selected, watermark_indices), daemon=True, ) worker.start() def _set_processing_state(self, running: bool) -> None: state = "disabled" if running else "normal" for button in self.file_buttons: button.configure(state=state) self.output_button.configure(state=state) self.watermark_preview_button.configure(state=state) self.process_button.configure(state=state) self.cancel_button.configure(state="normal" if running else "disabled") if running: self.progress.start(10) else: self.progress.stop() def _cancel(self) -> None: if not self.processing: return self.cancel_event.set() self.cancel_button.configure(state="disabled") self.status_var.set("Parada solicitada; terminando o video atual...") def _safe_after(self, func, *args) -> None: if self.closing: return try: if not self.winfo_exists(): return self.after(0, func, *args) except (tk.TclError, RuntimeError): pass def _post_status(self, text: str) -> None: self._safe_after(self._set_status_and_log, text) def _set_status_and_log(self, text: str) -> None: self.status_var.set(text) self._append_log(text) def _append_log(self, text: str) -> None: self.log_text.configure(state="normal") self.log_text.insert("end", text + "\n") self.log_text.see("end") self.log_text.configure(state="disabled") def _clear_log(self) -> None: self.log_text.configure(state="normal") self.log_text.delete("1.0", "end") self.log_text.configure(state="disabled") def _worker( self, paths: list[Path], output_parent: Path, options: PipelineOptions, gif_selected: set[int], watermark_indices: Optional[set[int]], ) -> None: try: report = process_pipeline( paths, output_parent, options, gif_selected_indices=gif_selected, watermark_indices=watermark_indices, status=self._post_status, cancel_event=self.cancel_event, ) except Exception as exc: self._safe_after(self._show_batch_error, str(exc)) return self._safe_after(self._show_batch_result, report) def _show_batch_error(self, message: str) -> None: self.processing = False self._set_processing_state(False) self.status_var.set("Nao foi possivel concluir o processamento.") self._append_log("ERRO: " + message) messagebox.showerror("Erro", message) def _show_batch_result(self, report: PipelineReport) -> None: self.processing = False self._set_processing_state(False) self.last_output_folder = report.output_folder self.open_folder_button.configure(state="normal") summary = [ f"{len(report.created_videos)} video(s) criado(s)", f"{len(report.created_gifs)} GIF(s) criado(s)", f"{len(report.skipped)} pulado(s)", f"{len(report.failures)} falha(s)", ] if report.cancelled: summary.append("cancelado") self.status_var.set( f"Processamento concluido. Pasta: {report.output_folder}" ) self.details_var.set(" | ".join(summary)) self._append_log("Resumo: " + " | ".join(summary)) preview_source = ( report.created_videos[-1] if report.created_videos else (report.outputs[-1] if report.outputs else None) ) if preview_source is not None: self._prepare_preview(preview_source) if report.failures: failures = "\n".join( f"- {path.name}: {message}" for path, message in report.failures[:6] ) if len(report.failures) > 6: failures += "\n- ..." messagebox.showwarning( "Processamento parcial", f"Alguns videos nao foram processados:\n\n{failures}", ) elif not report.created_videos and not report.created_gifs: if report.skipped: messagebox.showinfo( "Nada foi recriado", "Todos os arquivos ja existiam e foram pulados.\n\n" f"Pasta: {report.loop_folder}\n\n" "Desmarque 'Pular resultados que ja existem' para recriar.", ) else: messagebox.showwarning( "Nada foi processado", "Nenhum arquivo foi criado. Confira a fila de videos e a " "pasta de saida.", ) else: messagebox.showinfo( "Pipeline concluido", f"Videos criados: {len(report.created_videos)}\n" f"GIFs criados: {len(report.created_gifs)}\n" f"Ja existiam (pulados): {len(report.skipped)}\n\n" f"Loopings: {report.loop_folder}\n" f"GIFs: {report.gif_folder}", ) def _clear_preview(self) -> None: self._stop_preview_playback() self._set_preview_mode("result") self.preview_frames = [] self.preview_index = 0 self.preview_photo = None if self.preview_tempdir is not None: self.preview_tempdir.cleanup() self.preview_tempdir = None self.preview_frame_label.configure(image="", text="Nenhum resultado carregado.") self.preview_status_var.set("O preview aparecera depois do processamento.") for button in (self.previous_button, self.play_button, self.next_button): button.configure(state="disabled") def _prepare_preview(self, output_path: Path) -> None: self._clear_preview() self.preview_status_var.set("Gerando preview...") self.preview_tempdir = tempfile.TemporaryDirectory(prefix="loop_preview_") preview_folder = Path(self.preview_tempdir.name) threading.Thread( target=self._preview_worker, args=(output_path, preview_folder), daemon=True, ).start() def _preview_worker(self, output_path: Path, preview_folder: Path) -> None: try: ffmpeg = _find_executable("ffmpeg") command = [ ffmpeg, "-hide_banner", "-loglevel", "error", "-nostdin", "-i", str(output_path), "-vf", "fps=5,scale=640:640:force_original_aspect_ratio=decrease", "-frames:v", "300", "-y", str(preview_folder / "frame_%04d.png"), ] result = subprocess.run( command, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, ) if result.returncode != 0: details = result.stderr.decode("utf-8", errors="replace").strip() raise VideoLoopError(details or "Nao foi possivel criar o preview.") frames = sorted(preview_folder.glob("frame_*.png")) if not frames: raise VideoLoopError("Nenhum quadro foi gerado para o preview.") except Exception as exc: self._safe_after(self._preview_failed, str(exc)) return self._safe_after(self._preview_ready, frames) def _preview_failed(self, message: str) -> None: self.preview_status_var.set(f"Preview indisponivel: {message}") def _preview_ready(self, frames: list[Path]) -> None: self.preview_frames = frames self.preview_index = 0 for button in (self.previous_button, self.play_button, self.next_button): button.configure(state="normal") self._display_preview_frame() def _display_preview_frame(self) -> None: if not self.preview_frames: return frame_path = self.preview_frames[self.preview_index] try: if Image is not None and ImageTk is not None: with Image.open(frame_path) as image: image = image.convert("RGB") image.thumbnail((640, 380)) self.preview_photo = ImageTk.PhotoImage(image) else: self.preview_photo = tk.PhotoImage(file=str(frame_path)) self.preview_frame_label.configure(image=self.preview_photo, text="") self.preview_status_var.set( f"Quadro {self.preview_index + 1} de {len(self.preview_frames)}" ) except (OSError, tk.TclError) as exc: self.preview_status_var.set(f"Nao foi possivel mostrar o preview: {exc}") def _previous_preview(self) -> None: if not self.preview_frames: return self.preview_index = (self.preview_index - 1) % len(self.preview_frames) self._display_preview_frame() def _next_preview(self) -> None: if not self.preview_frames: return self.preview_index = (self.preview_index + 1) % len(self.preview_frames) self._display_preview_frame() def _toggle_preview(self) -> None: if not self.preview_frames: return if self.preview_playing: self._stop_preview_playback() return self.preview_playing = True self.play_button.configure(text="Pausar") self._play_preview_frame() def _play_preview_frame(self) -> None: if not self.preview_playing or not self.preview_frames: return self._display_preview_frame() self.preview_index = (self.preview_index + 1) % len(self.preview_frames) self.preview_after_id = self.after(200, self._play_preview_frame) def _stop_preview_playback(self) -> None: self.preview_playing = False if self.preview_after_id is not None: try: self.after_cancel(self.preview_after_id) except tk.TclError: pass self.preview_after_id = None if hasattr(self, "play_button"): self.play_button.configure(text="Reproduzir") def _open_output_folder(self) -> None: if self.last_output_folder is None or not self.last_output_folder.exists(): return try: if os.name == "nt": os.startfile(str(self.last_output_folder)) elif sys.platform == "darwin": subprocess.Popen(["open", str(self.last_output_folder)]) else: subprocess.Popen(["xdg-open", str(self.last_output_folder)]) except OSError as exc: messagebox.showerror("Erro", f"Nao foi possivel abrir a pasta:\n{exc}") def _on_close(self) -> None: self.closing = True self.cancel_event.set() self._stop_preview_playback() if self.preview_tempdir is not None: self.preview_tempdir.cleanup() self.preview_tempdir = None self.destroy() class LoopApp(TkBase): """Simplified, batch-oriented interface for the complete video pipeline.""" def __init__(self) -> None: super().__init__() self.title("Loop Studio") self.geometry("1180x800") self.minsize(980, 700) self.protocol("WM_DELETE_WINDOW", self._on_close) self.input_paths: list[Path] = [] self.output_dir_var = tk.StringVar() self.target_var = tk.StringVar(value="20") self.minimum_var = tk.StringVar(value="4") self.prefix_var = tk.StringVar(value="looping") self.skip_existing_var = tk.BooleanVar(value=True) self.create_plain_var = tk.BooleanVar(value=True) self.create_watermarked_var = tk.BooleanVar(value=False) self.clean_metadata_var = tk.BooleanVar(value=True) self.gif_mode_var = tk.StringVar(value="nao criar GIF") self.gif_variant_var = tk.StringVar(value="sem marca") self.gif_random_count_var = tk.StringVar(value="1") self.gif_width_var = tk.StringVar() self.watermark_mode_var = tk.StringVar(value="text") self.watermark_text_var = tk.StringVar(value="Python/Tabu") self.watermark_image_var = tk.StringVar() self.watermark_opacity_var = tk.StringVar(value="0.80") self.watermark_size_var = tk.StringVar(value="36") self.watermark_scale_var = tk.StringVar(value="25") self.watermark_position_var = tk.StringVar(value="inferior direito") self.watermark_x_var = tk.StringVar(value="50.0") self.watermark_y_var = tk.StringVar(value="80.0") self.watermark_layer_var = tk.StringVar(value="Marca 1") self.watermark2_enabled_var = tk.BooleanVar(value=False) self.layer_states = { "Marca 1": { "mode": "text", "text": "Python/Tabu", "image_path": "", "opacity": "0.80", "font_size": "36", "image_scale": "25", "position": "inferior direito", "x_percent": "50.0", "y_percent": "80.0", }, "Marca 2": { "mode": "text", "text": "", "image_path": "", "opacity": "0.80", "font_size": "28", "image_scale": "20", "position": "superior direito", "x_percent": "75.0", "y_percent": "20.0", }, } self.active_layer_name = "Marca 1" self.preset_var = tk.StringVar() self.settings_file = Path(__file__).with_name("loop_studio_settings.json") self.status_var = tk.StringVar(value="Adicione videos para comecar.") self.details_var = tk.StringVar() self.preview_status_var = tk.StringVar(value="O primeiro video aparecera aqui.") self.cancel_event = threading.Event() self.processing = False self.closing = False self.last_output_folder: Optional[Path] = None self.preview_tempdir: Optional[tempfile.TemporaryDirectory] = None self.preview_frames: list[Path] = [] self.preview_index = 0 self.preview_playing = False self.preview_after_id: Optional[str] = None self.preview_photo = None self.editor_mode = True self.editor_source_path: Optional[Path] = None self.editor_clean_image = None self.editor_exact_image = None self.editor_render_after_id: Optional[str] = None self.editor_render_generation = 0 self.editor_bg_image = None self.editor_bg_photo = None self.editor_overlay_photo = None self.editor_items: list[int] = [] self.editor_frame_size = (0, 0) self.editor_display_origin = (0, 0) self.editor_source_size = (0, 0) self.presets_file = Path(__file__).with_name("watermark_presets.json") self._load_default_settings() self._build_widgets() self._enable_drag_and_drop() self._refresh_preset_combo() def _build_widgets(self) -> None: shell = ttk.Frame(self, padding=16) shell.pack(fill="both", expand=True) header = ttk.Frame(shell) header.pack(fill="x", pady=(0, 10)) ttk.Label(header, text="LOOP STUDIO", font=("Segoe UI", 22, "bold")).pack( side="left" ) ttk.Label( header, text="loop + marca d'agua + GIF em um so lugar", foreground="#666666", ).pack(side="left", padx=(14, 0), pady=(7, 0)) bottom = ttk.Frame(shell) bottom.pack(side="bottom", fill="x", pady=(10, 0)) ttk.Label(bottom, textvariable=self.status_var, anchor="w").pack(fill="x") ttk.Label(bottom, textvariable=self.details_var, anchor="w").pack(fill="x") action_bar = ttk.Frame(bottom) action_bar.pack(fill="x", pady=(7, 0)) self.progress = ttk.Progressbar(action_bar, mode="indeterminate") self.progress.pack(side="left", fill="x", expand=True, padx=(0, 10)) self.cancel_button = ttk.Button( action_bar, text="Parar", command=self._cancel, state="disabled" ) self.cancel_button.pack(side="right", padx=(8, 0)) self.start_button = tk.Button( action_bar, text="COMEÇAR", command=self._start, font=("Segoe UI", 14, "bold"), bg="#2e7d32", fg="white", activebackground="#1b5e20", activeforeground="white", relief="flat", padx=34, pady=9, cursor="hand2", ) self.start_button.pack(side="right") content = ttk.Frame(shell) content.pack(fill="both", expand=True) content.columnconfigure(0, weight=2) content.columnconfigure(1, weight=3) content.rowconfigure(0, weight=1) left = ttk.Frame(content) left.grid(row=0, column=0, sticky="nsew", padx=(0, 10)) left.columnconfigure(0, weight=1) left.rowconfigure(0, weight=1) files = ttk.LabelFrame(left, text="1. Videos", padding=10) files.grid(row=0, column=0, sticky="nsew") files.columnconfigure(0, weight=1) files.rowconfigure(1, weight=1) self.drop_hint = ttk.Label( files, text=( "Arraste arquivos/pastas para a lista" if DND_FILES is not None else "Adicione arquivos ou uma pasta pelos botoes" ), foreground="#666666", ) self.drop_hint.grid(row=0, column=0, sticky="w", pady=(0, 6)) list_frame = ttk.Frame(files) list_frame.grid(row=1, column=0, sticky="nsew") list_frame.columnconfigure(0, weight=1) list_frame.rowconfigure(0, weight=1) self.file_list = tk.Listbox( list_frame, selectmode=tk.EXTENDED, exportselection=False, activestyle="none", font=("Segoe UI", 9), ) self.file_list.grid(row=0, column=0, sticky="nsew") self.file_list.bind("<>", self._on_list_select) scroll = ttk.Scrollbar(list_frame, orient="vertical", command=self.file_list.yview) scroll.grid(row=0, column=1, sticky="ns") self.file_list.configure(yscrollcommand=scroll.set) file_buttons = ttk.Frame(files) file_buttons.grid(row=2, column=0, sticky="ew", pady=(8, 0)) for column in range(4): file_buttons.columnconfigure(column, weight=1) self.add_files_button = ttk.Button( file_buttons, text="+ Arquivos", command=self._choose_files ) self.add_folder_button = ttk.Button( file_buttons, text="+ Pasta", command=self._choose_input_folder ) self.remove_button = ttk.Button( file_buttons, text="Remover", command=self._remove_selected ) self.clear_button = ttk.Button(file_buttons, text="Limpar", command=self._clear_files) for column, button in enumerate( (self.add_files_button, self.add_folder_button, self.remove_button, self.clear_button) ): button.grid(row=0, column=column, sticky="ew", padx=2) self.file_buttons = [ self.add_files_button, self.add_folder_button, self.remove_button, self.clear_button, ] destination = ttk.LabelFrame(left, text="Destino", padding=8) destination.grid(row=1, column=0, sticky="ew", pady=(10, 0)) destination.columnconfigure(1, weight=1) ttk.Label(destination, text="Pasta:").grid(row=0, column=0, padx=(0, 6)) self.output_entry = ttk.Entry(destination, textvariable=self.output_dir_var) self.output_entry.grid(row=0, column=1, sticky="ew") self.output_button = ttk.Button( destination, text="Escolher", command=self._choose_output_folder ) self.output_button.grid(row=0, column=2, padx=(6, 0)) ttk.Label(destination, text="cria automaticamente as pastas looping e gif").grid( row=1, column=0, columnspan=3, sticky="w", pady=(4, 0) ) options = ttk.LabelFrame(left, text="2. Opcoes", padding=8) options.grid(row=2, column=0, sticky="ew", pady=(10, 0)) options.columnconfigure(1, weight=1) options.columnconfigure(3, weight=1) ttk.Label(options, text="Duracao (s):").grid(row=0, column=0, sticky="w") ttk.Entry(options, textvariable=self.target_var, width=8).grid( row=0, column=1, sticky="w", padx=(4, 10) ) ttk.Label(options, text="Min. loop:").grid(row=0, column=2, sticky="w") ttk.Entry(options, textvariable=self.minimum_var, width=8).grid( row=0, column=3, sticky="w", padx=4 ) ttk.Label(options, text="Prefixo:").grid(row=1, column=0, sticky="w", pady=(6, 0)) ttk.Entry(options, textvariable=self.prefix_var).grid( row=1, column=1, columnspan=3, sticky="ew", padx=(4, 0), pady=(6, 0) ) ttk.Label(options, text="GIF:").grid(row=2, column=0, sticky="w", pady=(6, 0)) self.gif_mode_combo = ttk.Combobox( options, textvariable=self.gif_mode_var, values=tuple(GIF_MODE_LABELS), state="readonly", width=15 ) self.gif_mode_combo.grid(row=2, column=1, sticky="w", padx=(4, 6), pady=(6, 0)) self.gif_mode_combo.bind("<>", self._toggle_gif_fields) self.gif_variant_combo = ttk.Combobox( options, textvariable=self.gif_variant_var, values=tuple(GIF_VARIANT_LABELS), state="readonly", width=13 ) self.gif_variant_combo.grid(row=2, column=2, sticky="w", pady=(6, 0)) ttk.Label(options, text="24 FPS / original").grid(row=2, column=3, sticky="e", pady=(6, 0)) ttk.Label(options, text="GIF largura:").grid(row=3, column=0, sticky="w", pady=(6, 0)) self.gif_width_entry = ttk.Entry(options, textvariable=self.gif_width_var, width=8) self.gif_width_entry.grid(row=3, column=1, sticky="w", padx=(4, 0), pady=(6, 0)) ttk.Label(options, text="aleatorios:").grid(row=3, column=2, sticky="e", pady=(6, 0)) self.gif_random_entry = ttk.Entry( options, textvariable=self.gif_random_count_var, width=6 ) self.gif_random_entry.grid(row=3, column=3, sticky="w", padx=(4, 0), pady=(6, 0)) self.plain_check = ttk.Checkbutton( options, text="Sem marca", variable=self.create_plain_var ) self.plain_check.grid(row=4, column=0, columnspan=2, sticky="w", pady=(8, 0)) self.watermarked_check = ttk.Checkbutton( options, text="Com marca", variable=self.create_watermarked_var ) self.watermarked_check.grid(row=4, column=2, columnspan=2, sticky="w", pady=(8, 0)) self.metadata_check = ttk.Checkbutton( options, text="Limpar metadados", variable=self.clean_metadata_var ) self.metadata_check.grid(row=5, column=0, columnspan=2, sticky="w") self.skip_check = ttk.Checkbutton( options, text="Pular existentes", variable=self.skip_existing_var ) self.skip_check.grid(row=5, column=2, columnspan=2, sticky="w") ttk.Label( options, text="A marca ajustada no preview sera aplicada a todos os videos.", foreground="#666666", ).grid(row=6, column=0, columnspan=4, sticky="w", pady=(5, 0)) right = ttk.LabelFrame(content, text="3. Marca d'agua e preview", padding=10) right.grid(row=0, column=1, sticky="nsew") right.columnconfigure(0, weight=1) right.rowconfigure(1, weight=1) mode_bar = ttk.Frame(right) mode_bar.grid(row=0, column=0, sticky="ew", pady=(0, 5)) self.editor_mode_button = ttk.Button( mode_bar, text="Editar marca d'agua", command=self._show_editor_mode ) self.editor_mode_button.pack(side="left") self.result_mode_button = ttk.Button( mode_bar, text="Ver resultado", command=self._show_result_mode ) self.result_mode_button.pack(side="left", padx=(6, 0)) self.defaults_button = ttk.Button( mode_bar, text="Salvar padrao", command=self._save_default_settings ) self.defaults_button.pack(side="right") ttk.Label(mode_bar, text="Camada:").pack(side="right", padx=(8, 4)) self.layer_combo = ttk.Combobox( mode_bar, textvariable=self.watermark_layer_var, values=("Marca 1", "Marca 2"), state="readonly", width=10, ) self.layer_combo.pack(side="right") self.layer_combo.bind("<>", self._switch_layer) self.layer2_check = ttk.Checkbutton( mode_bar, text="Ativar 2", variable=self.watermark2_enabled_var, command=self._update_editor_overlay, ) self.layer2_check.pack(side="right", padx=(8, 0)) preview_area = ttk.Frame(right) preview_area.grid(row=1, column=0, sticky="nsew") preview_area.columnconfigure(0, weight=1) preview_area.rowconfigure(0, weight=1) self.editor_canvas = tk.Canvas( preview_area, width=480, height=420, background="#202124", highlightthickness=0 ) self.editor_canvas.grid(row=0, column=0, sticky="nsew") self.editor_canvas.bind("", self._editor_press) self.editor_canvas.bind("", self._editor_drag) self.editor_canvas.bind("", self._editor_wheel) self.editor_canvas.bind("", self._on_editor_resize) self.preview_frame_label = ttk.Label( preview_area, text="Nenhum resultado carregado.", anchor="center", justify="center" ) self.preview_status_var.set("Adicione videos; o primeiro aparece aqui automaticamente.") self.preview_frame_label.grid_remove() watermark_controls = ttk.Frame(right) watermark_controls.grid(row=2, column=0, sticky="ew", pady=(8, 0)) watermark_controls.columnconfigure(1, weight=1) ttk.Label(watermark_controls, text="Texto:").grid(row=0, column=0, sticky="w") self.watermark_text_entry = ttk.Entry( watermark_controls, textvariable=self.watermark_text_var ) self.watermark_text_entry.grid(row=0, column=1, columnspan=3, sticky="ew", padx=(5, 0)) ttk.Label(watermark_controls, text="Imagem:").grid(row=1, column=0, sticky="w", pady=(4, 0)) self.watermark_image_entry = ttk.Entry( watermark_controls, textvariable=self.watermark_image_var ) self.watermark_image_entry.grid(row=1, column=1, columnspan=2, sticky="ew", padx=(5, 0), pady=(4, 0)) self.watermark_image_button = ttk.Button( watermark_controls, text="Escolher", command=self._choose_watermark_image ) self.watermark_image_button.grid(row=1, column=3, padx=(5, 0), pady=(4, 0)) ttk.Radiobutton( watermark_controls, text="Texto", value="text", variable=self.watermark_mode_var, command=self._toggle_watermark_fields ).grid(row=2, column=0, sticky="w", pady=(5, 0)) ttk.Radiobutton( watermark_controls, text="Imagem", value="image", variable=self.watermark_mode_var, command=self._toggle_watermark_fields ).grid(row=2, column=1, sticky="w", pady=(5, 0)) ttk.Label(watermark_controls, text="Tamanho:").grid(row=2, column=2, sticky="e", pady=(5, 0)) self.size_scale = ttk.Scale( watermark_controls, from_=10, to=150, orient="horizontal", command=self._on_size_scale ) self.size_scale.grid(row=2, column=3, sticky="ew", padx=(6, 0), pady=(5, 0)) self.size_value_label = ttk.Label(watermark_controls, text="36", width=5) self.size_value_label.grid(row=2, column=4, sticky="e", padx=(5, 0), pady=(5, 0)) ttk.Label(watermark_controls, text="Opacidade:").grid(row=3, column=2, sticky="e") self.opacity_scale = ttk.Scale( watermark_controls, from_=5, to=100, orient="horizontal", command=self._on_opacity_scale ) self.opacity_scale.grid(row=3, column=3, sticky="ew", padx=(6, 0)) self.opacity_value_label = ttk.Label(watermark_controls, text="80%", width=5) self.opacity_value_label.grid(row=3, column=4, sticky="e", padx=(5, 0)) ttk.Label(watermark_controls, text="Marca salva:").grid(row=4, column=0, sticky="w", pady=(5, 0)) self.preset_combo = ttk.Combobox( watermark_controls, textvariable=self.preset_var, state="readonly", width=20 ) self.preset_combo.grid(row=4, column=1, sticky="ew", padx=(5, 0), pady=(5, 0)) self.preset_combo.bind("<>", self._load_selected_preset) self.preset_save_button = ttk.Button( watermark_controls, text="Salvar", command=self._save_preset ) self.preset_save_button.grid(row=4, column=2, padx=(6, 0), pady=(5, 0)) self.preset_delete_button = ttk.Button( watermark_controls, text="Excluir", command=self._delete_selected_preset ) self.preset_delete_button.grid(row=4, column=3, padx=(6, 0), pady=(5, 0)) for entry in (self.watermark_text_entry, self.watermark_image_entry): entry.bind("", self._update_editor_overlay) result_controls = ttk.Frame(right) result_controls.grid(row=3, column=0, sticky="ew", pady=(8, 0)) self.previous_button = ttk.Button( result_controls, text="Anterior", command=self._previous_preview, state="disabled" ) self.previous_button.pack(side="left", fill="x", expand=True) self.play_button = ttk.Button( result_controls, text="Reproduzir", command=self._toggle_preview, state="disabled" ) self.play_button.pack(side="left", fill="x", expand=True, padx=5) self.next_button = ttk.Button( result_controls, text="Proximo", command=self._next_preview, state="disabled" ) self.next_button.pack(side="left", fill="x", expand=True) self.open_folder_button = ttk.Button( right, text="Abrir pasta de saida", command=self._open_output_folder, state="disabled" ) self.open_folder_button.grid(row=4, column=0, sticky="ew", pady=(6, 0)) self.editor_controls = [ self.watermark_text_entry, self.watermark_image_entry, self.watermark_image_button, self.preset_combo, self.preset_save_button, self.preset_delete_button, self.size_scale, self.opacity_scale, self.layer_combo, self.layer2_check, ] self._toggle_watermark_fields() self._toggle_gif_fields() self._sync_editor_controls() def _save_active_layer(self, name: Optional[str] = None) -> None: layer_name = name or getattr(self, "active_layer_name", "Marca 1") self.layer_states[layer_name] = { "mode": self.watermark_mode_var.get(), "text": self.watermark_text_var.get(), "image_path": self.watermark_image_var.get(), "opacity": self.watermark_opacity_var.get(), "font_size": self.watermark_size_var.get(), "image_scale": self.watermark_scale_var.get(), "position": self.watermark_position_var.get(), "x_percent": self.watermark_x_var.get(), "y_percent": self.watermark_y_var.get(), } def _load_active_layer(self) -> None: state = self.layer_states[self.active_layer_name] self.watermark_mode_var.set(state["mode"]) self.watermark_text_var.set(state["text"]) self.watermark_image_var.set(state["image_path"]) self.watermark_opacity_var.set(state["opacity"]) self.watermark_size_var.set(state["font_size"]) self.watermark_scale_var.set(state["image_scale"]) self.watermark_position_var.set(state["position"]) self.watermark_x_var.set(state["x_percent"]) self.watermark_y_var.set(state["y_percent"]) self._toggle_watermark_fields() def _switch_layer(self, _event: object = None) -> None: old_layer = getattr(self, "active_layer_name", "Marca 1") self._save_active_layer(old_layer) self.active_layer_name = self.watermark_layer_var.get() self._load_active_layer() def _state_to_config(self, state: dict) -> WatermarkConfig: try: x_percent = float(state["x_percent"].replace(",", ".")) y_percent = float(state["y_percent"].replace(",", ".")) opacity = float(state["opacity"].replace(",", ".")) font_size = int(state["font_size"]) image_scale = int(state["image_scale"]) except (KeyError, TypeError, ValueError) as exc: raise VideoLoopError("A configuracao de uma marca d'agua esta invalida.") from exc image_text = state.get("image_path", "").strip() return WatermarkConfig( mode=state.get("mode", "text"), text=state.get("text", ""), image_path=Path(image_text) if image_text else None, opacity=opacity, font_size=font_size, image_scale=image_scale, position=POSITION_LABELS.get(state.get("position", "inferior direito"), "bottom-right"), x_percent=x_percent, y_percent=y_percent, ) def _build_watermark_configs(self) -> tuple[WatermarkConfig, ...]: self._save_active_layer() configs = [self._state_to_config(self.layer_states["Marca 1"])] if self.watermark2_enabled_var.get(): configs.append(self._state_to_config(self.layer_states["Marca 2"])) return tuple(_validate_watermark(config) for config in configs) def _toggle_watermark_fields(self) -> None: text_state = "normal" if self.watermark_mode_var.get() == "text" else "disabled" image_state = "normal" if self.watermark_mode_var.get() == "image" else "disabled" self.watermark_text_entry.configure(state=text_state) self.watermark_image_entry.configure(state=image_state) self.watermark_image_button.configure(state=image_state) self._sync_editor_controls() self._update_editor_overlay() def _toggle_gif_fields(self, _event: object = None) -> None: mode = GIF_MODE_LABELS.get(self.gif_mode_var.get(), "none") self.gif_random_entry.configure(state="normal" if mode == "random" else "disabled") enabled = mode != "none" self.gif_variant_combo.configure(state="readonly" if enabled else "disabled") self.gif_width_entry.configure(state="normal" if enabled else "disabled") def _sync_editor_controls(self) -> None: image_mode = self.watermark_mode_var.get() == "image" if image_mode: self.size_scale.configure(from_=5, to=80) try: value = float(self.watermark_scale_var.get()) except ValueError: value = 25 self.size_scale.set(min(80, max(5, value))) self.size_value_label.configure(text=f"{int(value)}%") else: self.size_scale.configure(from_=10, to=150) try: value = float(self.watermark_size_var.get()) except ValueError: value = 36 self.size_scale.set(min(150, max(10, value))) self.size_value_label.configure(text=str(int(value))) try: opacity = float(self.watermark_opacity_var.get().replace(",", ".")) * 100 except ValueError: opacity = 80 opacity = min(100, max(5, opacity)) self.opacity_scale.set(opacity) self.opacity_value_label.configure(text=f"{int(opacity)}%") def _on_size_scale(self, value: str) -> None: numeric = int(float(value)) if self.watermark_mode_var.get() == "image": self.watermark_scale_var.set(str(numeric)) self.size_value_label.configure(text=f"{numeric}%") else: self.watermark_size_var.set(str(numeric)) self.size_value_label.configure(text=str(numeric)) self._update_editor_overlay() def _on_opacity_scale(self, value: str) -> None: numeric = min(100, max(5, int(float(value)))) self.watermark_opacity_var.set(f"{numeric / 100:.2f}") self.opacity_value_label.configure(text=f"{numeric}%") self._update_editor_overlay() def _choose_watermark_image(self) -> None: path = filedialog.askopenfilename( title="Escolha a imagem da marca d'agua", filetypes=[("Imagens", "*.png *.jpg *.jpeg *.webp *.bmp"), ("Todos", "*.*")], ) if path: self.watermark_image_var.set(path) self.watermark_mode_var.set("image") self._toggle_watermark_fields() def _editor_wheel(self, event: object) -> None: step = 4 if self.watermark_mode_var.get() == "text" else 2 direction = 1 if getattr(event, "delta", 0) > 0 else -1 if self.watermark_mode_var.get() == "image": current = self._editor_image_scale() self.watermark_scale_var.set(str(min(80, max(5, current + direction * step // 2)))) else: try: current = int(self.watermark_size_var.get()) except ValueError: current = 36 self.watermark_size_var.set(str(min(150, max(10, current + direction * step)))) self._sync_editor_controls() self._update_editor_overlay() def _show_editor_mode(self) -> None: self._stop_preview_playback() self._set_preview_mode("editor") if self.editor_source_path is None and self.input_paths: self._load_editor_frame(self.input_paths[0]) elif not self.input_paths: self.preview_status_var.set("Adicione videos para editar a marca d'agua.") def _show_result_mode(self) -> None: if not self.preview_frames: self.preview_status_var.set("Nenhum resultado ainda. Clique em COMEÇAR primeiro.") return self._set_preview_mode("result") self._display_preview_frame() def _build_watermark_config(self) -> WatermarkConfig: try: opacity = float(self.watermark_opacity_var.get().replace(",", ".")) font_size = int(self.watermark_size_var.get()) image_scale = int(self.watermark_scale_var.get()) x_percent = float(self.watermark_x_var.get().replace(",", ".")) y_percent = float(self.watermark_y_var.get().replace(",", ".")) except ValueError as exc: raise VideoLoopError( "Tamanho, opacidade e posicao da marca precisam ser numericos." ) from exc return WatermarkConfig( mode=self.watermark_mode_var.get(), text=self.watermark_text_var.get(), image_path=Path(self.watermark_image_var.get().strip()) if self.watermark_image_var.get().strip() else None, opacity=opacity, font_size=font_size, image_scale=image_scale, position=POSITION_LABELS.get(self.watermark_position_var.get(), "bottom-right"), x_percent=x_percent, y_percent=y_percent, ) def _load_editor_frame(self, input_path: Path) -> None: if self.processing or ( self.editor_source_path == input_path and self.editor_bg_photo is not None ): return try: source_info = probe_video(input_path) except VideoLoopError as exc: self.preview_status_var.set(str(exc)) return self.editor_render_generation += 1 if self.editor_render_after_id is not None: try: self.after_cancel(self.editor_render_after_id) except tk.TclError: pass self.editor_render_after_id = None self.editor_source_path = input_path self._set_preview_mode("editor") self.editor_canvas.delete("all") self.editor_items = [] self.editor_clean_image = None self.editor_exact_image = None self.editor_bg_photo = None self.editor_overlay_photo = None if self.preview_tempdir is not None: self.preview_tempdir.cleanup() self.preview_tempdir = tempfile.TemporaryDirectory(prefix="watermark_editor_") frame_path = Path(self.preview_tempdir.name) / "frame.png" self.preview_status_var.set(f"Carregando preview de {input_path.name}...") threading.Thread( target=self._editor_worker, args=(input_path, frame_path, source_info), daemon=True, ).start() def _editor_worker( self, input_path: Path, frame_path: Path, source_info: VideoInfo ) -> None: try: extract_frame(input_path, frame_path, at_seconds=0.5, max_side=560) except Exception as exc: self._safe_after(self._preview_failed, str(exc)) return self._safe_after(self._editor_frame_ready, frame_path, source_info) def _editor_frame_ready(self, frame_path: Path, source_info: VideoInfo) -> None: if Image is None or ImageTk is None: self.preview_status_var.set("Instale Pillow para usar o editor visual.") return try: with Image.open(frame_path) as image: background_image = image.convert("RGB").copy() except (OSError, tk.TclError) as exc: self.preview_status_var.set(f"Nao foi possivel carregar o preview: {exc}") return self.editor_clean_image = background_image self.editor_exact_image = None self.editor_bg_image = background_image self.editor_source_size = (source_info.width or 1, source_info.height or 1) self._sync_editor_controls() self._on_editor_resize() self._schedule_exact_editor_preview() self.preview_status_var.set( "Arraste a marca; rode a rodinha do mouse para aumentar ou diminuir." ) def _on_editor_resize(self, _event: object = None) -> None: source_image = self.editor_exact_image or self.editor_clean_image or self.editor_bg_image if not self.editor_mode or source_image is None or ImageTk is None: return canvas_width = max(1, self.editor_canvas.winfo_width()) canvas_height = max(1, self.editor_canvas.winfo_height()) image = source_image.copy() resampling = getattr(getattr(Image, "Resampling", Image), "LANCZOS", 1) image.thumbnail((canvas_width, canvas_height), resampling) origin_x = max(0, (canvas_width - image.width) // 2) origin_y = max(0, (canvas_height - image.height) // 2) self.editor_display_origin = (origin_x, origin_y) self.editor_frame_size = (image.width, image.height) self.editor_bg_photo = ImageTk.PhotoImage(image) self.editor_canvas.delete("all") self.editor_items = [ self.editor_canvas.create_image(origin_x, origin_y, image=self.editor_bg_photo, anchor="nw") ] if self.editor_exact_image is None: self._draw_editor_overlay_pil() def _set_preview_mode(self, mode: str) -> None: self.editor_mode = mode == "editor" if self.editor_mode: self.preview_frame_label.grid_remove() self.editor_canvas.grid(row=0, column=0, sticky="nsew") else: self.editor_canvas.grid_remove() self.preview_frame_label.grid(row=0, column=0, sticky="nsew") def _editor_opacity(self) -> int: try: value = float(self.watermark_opacity_var.get().replace(",", ".")) except ValueError: value = 0.8 return int(min(1.0, max(0.05, value)) * 255) def _editor_image_scale(self) -> int: try: return min(80, max(5, int(self.watermark_scale_var.get()))) except ValueError: return 25 def _update_editor_overlay(self, _event: object = None) -> None: if not self.editor_mode or self.editor_clean_image is None or Image is None or ImageTk is None: return self.editor_exact_image = None self._on_editor_resize() self._schedule_exact_editor_preview() def _draw_editor_overlay_pil(self) -> None: if not self.editor_mode or self.editor_clean_image is None or Image is None or ImageTk is None: return self._save_active_layer() frame_w, frame_h = self.editor_frame_size overlay = Image.new("RGBA", (frame_w, frame_h), (0, 0, 0, 0)) for layer_name in ("Marca 1", "Marca 2"): if layer_name == "Marca 2" and not self.watermark2_enabled_var.get(): continue state = self.layer_states[layer_name] try: fx = min(1.0, max(0.0, float(state["x_percent"]) / 100.0)) fy = min(1.0, max(0.0, float(state["y_percent"]) / 100.0)) opacity = int(min(1.0, max(0.05, float(state["opacity"]))) * 255) except (KeyError, TypeError, ValueError): continue if state["mode"] == "text": self._draw_editor_text(overlay, state, fx, fy, opacity) else: self._draw_editor_image(overlay, state, fx, fy, opacity) self.editor_overlay_photo = ImageTk.PhotoImage(overlay) for item in self.editor_items[1:]: self.editor_canvas.delete(item) self.editor_items = self.editor_items[:1] self.editor_items.append( self.editor_canvas.create_image( self.editor_display_origin[0], self.editor_display_origin[1], image=self.editor_overlay_photo, anchor="nw", ) ) def _draw_editor_text(self, overlay, state: dict, fx: float, fy: float, alpha: int) -> None: if ImageDraw is None or ImageFont is None: return draw = ImageDraw.Draw(overlay) frame_w, frame_h = self.editor_frame_size factor = frame_w / max(1, self.editor_source_size[0]) try: size = max(6, int(float(state["font_size"]) * factor)) except ValueError: size = max(6, int(36 * factor)) font = ImageFont.load_default() font_path = _default_font_file() if font_path: try: font = ImageFont.truetype(str(font_path), size) except OSError: pass draw.multiline_text( (fx * frame_w, fy * frame_h), state.get("text", "") or " ", font=font, fill=(255, 255, 255, alpha), anchor="mm", align="center", stroke_width=max(1, int(size / 18)), stroke_fill=(0, 0, 0, int(alpha * 0.55)), ) def _draw_editor_image(self, overlay, state: dict, fx: float, fy: float, alpha: int) -> None: path_text = state.get("image_path", "").strip() if not path_text or Image is None: return try: stamp = Image.open(path_text).convert("RGBA") except OSError: return frame_w, frame_h = self.editor_frame_size target_w = max(8, int(frame_w * min(80, max(5, int(state["image_scale"]))) / 100)) target_h = max(1, int(stamp.height * target_w / max(1, stamp.width))) stamp = stamp.resize((target_w, target_h)) stamp.putalpha(stamp.getchannel("A").point(lambda value: value * alpha // 255)) x = int(min(max(0, frame_w - stamp.width), fx * frame_w - stamp.width / 2)) y = int(min(max(0, frame_h - stamp.height), fy * frame_h - stamp.height / 2)) overlay.alpha_composite(stamp, (x, y)) def _schedule_exact_editor_preview(self) -> None: if not self.editor_source_path or not self.editor_mode: return self.editor_render_generation += 1 generation = self.editor_render_generation if self.editor_render_after_id is not None: try: self.after_cancel(self.editor_render_after_id) except tk.TclError: pass self.editor_render_after_id = self.after( 180, self._start_exact_editor_preview, generation ) def _start_exact_editor_preview(self, generation: int) -> None: if generation != self.editor_render_generation or self.editor_source_path is None: return try: configs = self._build_watermark_configs() except VideoLoopError as exc: self.preview_status_var.set(str(exc)) return if self.preview_tempdir is None: self.preview_tempdir = tempfile.TemporaryDirectory(prefix="watermark_editor_") output_path = Path(self.preview_tempdir.name) / f"exact_{generation}.png" threading.Thread( target=self._exact_editor_preview_worker, args=(self.editor_source_path, output_path, configs, generation), daemon=True, ).start() def _exact_editor_preview_worker( self, input_path: Path, output_path: Path, configs: tuple[WatermarkConfig, ...], generation: int, ) -> None: try: render_watermark_preview(input_path, output_path, configs) except Exception as exc: self._safe_after(self._preview_failed, str(exc)) return self._safe_after(self._exact_editor_preview_ready, output_path, generation) def _exact_editor_preview_ready(self, output_path: Path, generation: int) -> None: if generation != self.editor_render_generation or Image is None: return try: with Image.open(output_path) as image: self.editor_exact_image = image.convert("RGB").copy() except OSError as exc: self.preview_status_var.set(f"Preview da marca: {exc}") return self._on_editor_resize() self.preview_status_var.set( "Preview exato do FFmpeg: arraste ou use a rodinha para ajustar." ) def _editor_press(self, event: object) -> None: self._editor_drag(event) def _editor_drag(self, event: object) -> None: if not self.editor_mode or not self.editor_frame_size[0]: return width, height = self.editor_frame_size local_x = event.x - self.editor_display_origin[0] local_y = event.y - self.editor_display_origin[1] self.watermark_x_var.set(f"{min(100, max(0, local_x / width * 100)):.1f}") self.watermark_y_var.set(f"{min(100, max(0, local_y / height * 100)):.1f}") self._update_editor_overlay() def _load_default_settings(self) -> None: if not self.settings_file.is_file(): return try: settings = json.loads(self.settings_file.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return if not isinstance(settings, dict): return simple_values = { "target": self.target_var, "minimum": self.minimum_var, "prefix": self.prefix_var, "gif_mode": self.gif_mode_var, "gif_variant": self.gif_variant_var, "gif_random_count": self.gif_random_count_var, "gif_width": self.gif_width_var, } for key, variable in simple_values.items(): if key in settings: variable.set(str(settings[key])) boolean_values = { "skip_existing": self.skip_existing_var, "create_plain": self.create_plain_var, "create_watermarked": self.create_watermarked_var, "clean_metadata": self.clean_metadata_var, "layer2_enabled": self.watermark2_enabled_var, } for key, variable in boolean_values.items(): if key in settings: variable.set(bool(settings[key])) for layer_name in ("Marca 1", "Marca 2"): saved = settings.get("layers", {}).get(layer_name) if isinstance(saved, dict): self.layer_states[layer_name].update(saved) def _save_default_settings(self) -> None: self._save_active_layer() settings = { "target": self.target_var.get(), "minimum": self.minimum_var.get(), "prefix": self.prefix_var.get(), "gif_mode": self.gif_mode_var.get(), "gif_variant": self.gif_variant_var.get(), "gif_random_count": self.gif_random_count_var.get(), "gif_width": self.gif_width_var.get(), "skip_existing": self.skip_existing_var.get(), "create_plain": self.create_plain_var.get(), "create_watermarked": self.create_watermarked_var.get(), "clean_metadata": self.clean_metadata_var.get(), "layer2_enabled": self.watermark2_enabled_var.get(), "layers": self.layer_states, } try: self.settings_file.write_text( json.dumps(settings, ensure_ascii=False, indent=2), encoding="utf-8" ) self.status_var.set("Configuracao padrao salva para a proxima abertura.") except OSError as exc: messagebox.showerror("Configuracao", str(exc)) def _read_presets(self) -> dict: if not self.presets_file.is_file(): return {} try: data = json.loads(self.presets_file.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} return data if isinstance(data, dict) else {} def _write_presets(self, presets: dict) -> None: self.presets_file.write_text( json.dumps(presets, ensure_ascii=False, indent=2), encoding="utf-8" ) def _refresh_preset_combo(self) -> None: presets = self._read_presets() self.preset_combo.configure(values=tuple(sorted(presets))) if self.preset_var.get() not in presets: self.preset_var.set("") def _save_preset(self) -> None: name = (simpledialog.askstring("Salvar marca", "Nome:", parent=self) or "").strip() if not name: return self._save_active_layer() presets = self._read_presets() presets[name] = { "layers": self.layer_states, "layer2_enabled": self.watermark2_enabled_var.get(), } try: self._write_presets(presets) except OSError as exc: messagebox.showerror("Marca d'agua", str(exc)) return self._refresh_preset_combo() self.preset_var.set(name) self.status_var.set(f"Marca '{name}' salva.") def _load_selected_preset(self, _event: object = None) -> None: name = self.preset_var.get() preset = self._read_presets().get(name) if not preset: return if "layers" in preset: for layer_name in ("Marca 1", "Marca 2"): if isinstance(preset["layers"].get(layer_name), dict): self.layer_states[layer_name].update(preset["layers"][layer_name]) self.watermark2_enabled_var.set(bool(preset.get("layer2_enabled", False))) else: # Import presets created by the one-watermark version. self.layer_states["Marca 1"].update( { "mode": preset.get("mode", "text"), "text": preset.get("text", ""), "image_path": preset.get("image_path", ""), "opacity": str(preset.get("opacity", "0.80")), "font_size": str(preset.get("font_size", "36")), "image_scale": str(preset.get("image_scale", "25")), "position": preset.get("position", "inferior direito"), "x_percent": str(preset.get("x_percent") or "50"), "y_percent": str(preset.get("y_percent") or "80"), } ) self.active_layer_name = "Marca 1" self.watermark_layer_var.set("Marca 1") self._load_active_layer() self.status_var.set(f"Marca '{name}' carregada.") def _delete_selected_preset(self) -> None: name = self.preset_var.get() if not name: return presets = self._read_presets() presets.pop(name, None) self._write_presets(presets) self._refresh_preset_combo() def _enable_drag_and_drop(self) -> None: if DND_FILES is None: return for widget in (self.file_list, self.drop_hint): try: widget.drop_target_register(DND_FILES) widget.dnd_bind("<>", self._on_drop) except (tk.TclError, AttributeError): pass def _on_drop(self, event: object) -> str: self._add_paths([Path(item) for item in self.tk.splitlist(getattr(event, "data", ""))]) return "break" def _add_paths(self, paths: list[Path]) -> None: found: list[Path] = [] for path in paths: if path.is_dir(): found.extend(video_files_in_folder(path)) elif is_video_path(path): found.append(path) if not found: messagebox.showwarning("Videos", "Nenhum video encontrado.") return self.input_paths = unique_video_paths(self.input_paths + found) self._refresh_file_list() if not self.output_dir_var.get().strip(): self.output_dir_var.set(str(self.input_paths[0].parent)) self.status_var.set(f"{len(self.input_paths)} video(s) na fila.") if self.editor_source_path is None: self._load_editor_frame(self.input_paths[0]) def _on_list_select(self, _event: object = None) -> None: selected = self.file_list.curselection() if selected and not self.processing: self._load_editor_frame(self.input_paths[selected[0]]) def _refresh_file_list(self) -> None: self.file_list.delete(0, tk.END) for path in self.input_paths: self.file_list.insert(tk.END, str(path)) def _choose_files(self) -> None: paths = filedialog.askopenfilenames( title="Escolha videos", filetypes=[("Videos", "*.mp4 *.mov *.mkv *.avi *.webm *.m4v *.wmv *.flv"), ("Todos", "*.*")], ) self._add_paths([Path(path) for path in paths]) def _choose_input_folder(self) -> None: path = filedialog.askdirectory(title="Escolha uma pasta com videos") if path: self._add_paths([Path(path)]) def _remove_selected(self) -> None: selected = set(self.file_list.curselection()) self.input_paths = [path for index, path in enumerate(self.input_paths) if index not in selected] self._refresh_file_list() self.status_var.set(f"{len(self.input_paths)} video(s) na fila.") def _clear_files(self) -> None: self.input_paths.clear() self.editor_source_path = None self.editor_canvas.delete("all") self.editor_items = [] self._refresh_file_list() self.status_var.set("Fila limpa.") def _choose_output_folder(self) -> None: initial = self.output_dir_var.get().strip() or None path = filedialog.askdirectory(title="Escolha a pasta de saida", initialdir=initial) if path: self.output_dir_var.set(path) def _start(self) -> None: try: paths = list(self.input_paths) output_text = self.output_dir_var.get().strip() output_parent = Path(output_text).expanduser() if output_text else None target = float(self.target_var.get().replace(",", ".")) minimum = float(self.minimum_var.get().replace(",", ".")) if not paths: raise VideoLoopError("Adicione pelo menos um video.") if output_parent is None or not output_parent.is_dir(): raise VideoLoopError("Escolha uma pasta de saida valida.") if not math.isfinite(target) or not math.isfinite(minimum): raise VideoLoopError("As duracoes precisam ser numeros validos.") if target <= 0 or minimum <= 0: raise VideoLoopError("As duracoes devem ser maiores que zero.") gif_mode = GIF_MODE_LABELS.get(self.gif_mode_var.get(), "none") gif_selected = set(self.file_list.curselection()) gif_count = int(self.gif_random_count_var.get()) gif_width_text = self.gif_width_var.get().strip() gif_width = int(gif_width_text) if gif_width_text else None if gif_mode == "manual" and not gif_selected: raise VideoLoopError("Selecione os videos que serao GIFs.") if gif_mode == "random" and gif_count < 1: raise VideoLoopError("Informe a quantidade de GIFs aleatorios.") watermarks = ( self._build_watermark_configs() if self.create_watermarked_var.get() else () ) if not self.create_plain_var.get() and not self.create_watermarked_var.get(): raise VideoLoopError("Marque Sem marca, Com marca ou ambos.") gif_variant = GIF_VARIANT_LABELS.get(self.gif_variant_var.get(), "plain") if gif_mode != "none" and gif_variant in {"watermarked", "both"} and not self.create_watermarked_var.get(): raise VideoLoopError("Ative a versao Com marca para criar esse GIF.") options = PipelineOptions( target_seconds=target, min_loop_seconds=minimum, prefix=sanitize_prefix(self.prefix_var.get()), create_plain=self.create_plain_var.get(), create_watermarked=self.create_watermarked_var.get(), watermark=watermarks[0] if watermarks else None, clean_metadata=self.clean_metadata_var.get(), gif_mode=gif_mode, gif_count=gif_count, gif_variant=gif_variant, gif_fps=24, gif_width=gif_width, skip_existing=self.skip_existing_var.get(), watermarks=watermarks, ) except (ValueError, VideoLoopError) as exc: messagebox.showerror("Nao foi possivel iniciar", str(exc)) return self.cancel_event.clear() self.processing = True self._set_processing_state(True) self.details_var.set("") self.status_var.set("Processando... acompanhe o video atual acima.") threading.Thread( target=self._worker, args=(paths, output_parent, options, gif_selected), daemon=True, ).start() def _set_processing_state(self, running: bool) -> None: state = "disabled" if running else "normal" for button in self.file_buttons: button.configure(state=state) self.output_button.configure(state=state) self.defaults_button.configure(state=state) self.start_button.configure(state="disabled" if running else "normal") self.cancel_button.configure(state="normal" if running else "disabled") for widget in self.editor_controls: try: widget.configure(state="disabled" if running else "normal") except tk.TclError: pass if running: self.progress.start(10) else: self.progress.stop() def _cancel(self) -> None: if self.processing: self.cancel_event.set() self.cancel_button.configure(state="disabled") self.status_var.set("Parada solicitada; terminando o video atual...") def _safe_after(self, func, *args) -> None: if self.closing: return try: if self.winfo_exists(): self.after(0, func, *args) except (tk.TclError, RuntimeError): pass def _post_status(self, text: str) -> None: self._safe_after(self.status_var.set, text) def _worker( self, paths: list[Path], output_parent: Path, options: PipelineOptions, gif_selected: set[int], ) -> None: try: report = process_pipeline( paths, output_parent, options, gif_selected_indices=gif_selected, watermark_indices=None, cancel_event=self.cancel_event, status=self._post_status, ) except Exception as exc: self._safe_after(self._show_batch_error, str(exc)) return self._safe_after(self._show_batch_result, report) def _show_batch_error(self, message: str) -> None: self.processing = False self._set_processing_state(False) self.status_var.set("Erro durante o processamento.") messagebox.showerror("Erro", message) def _show_batch_result(self, report: PipelineReport) -> None: self.processing = False self._set_processing_state(False) self.last_output_folder = report.output_folder self.open_folder_button.configure(state="normal") self.details_var.set( f"{len(report.created_videos)} video(s) | {len(report.created_gifs)} GIF(s) | " f"{len(report.skipped)} pulado(s) | {len(report.failures)} falha(s)" ) if report.created_videos or report.outputs: self._prepare_preview((report.created_videos or report.outputs)[-1]) if report.failures: details = "\n".join(f"{path.name}: {error}" for path, error in report.failures[:5]) messagebox.showwarning("Processamento parcial", details) elif not report.created_videos and not report.created_gifs and report.skipped: messagebox.showinfo( "Nada recriado", "Os arquivos ja existiam. Desmarque 'Pular existentes' para recriar.", ) else: messagebox.showinfo( "Concluido", f"Videos criados: {len(report.created_videos)}\n" f"GIFs criados: {len(report.created_gifs)}\n\n" f"Pasta: {report.output_folder}", ) def _clear_preview(self) -> None: self._stop_preview_playback() self.preview_frames = [] self.preview_index = 0 self.preview_photo = None for button in (self.previous_button, self.play_button, self.next_button): button.configure(state="disabled") self.preview_frame_label.configure(image="", text="Nenhum resultado carregado.") def _prepare_preview(self, output_path: Path) -> None: self._clear_preview() self._set_preview_mode("result") if self.preview_tempdir is not None: self.preview_tempdir.cleanup() self.preview_tempdir = tempfile.TemporaryDirectory(prefix="loop_preview_") folder = Path(self.preview_tempdir.name) self.preview_status_var.set("Gerando preview do resultado...") threading.Thread( target=self._preview_worker, args=(output_path, folder), daemon=True ).start() def _preview_worker(self, output_path: Path, folder: Path) -> None: try: ffmpeg = _find_executable("ffmpeg") command = [ ffmpeg, "-hide_banner", "-loglevel", "error", "-nostdin", "-i", str(output_path), "-vf", "fps=5,scale=640:640:force_original_aspect_ratio=decrease", "-frames:v", "300", "-y", str(folder / "frame_%04d.png"), ] result = subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) if result.returncode != 0: raise VideoLoopError(result.stderr.decode("utf-8", errors="replace").strip()) frames = sorted(folder.glob("frame_*.png")) if not frames: raise VideoLoopError("Nenhum quadro de preview foi gerado.") except Exception as exc: self._safe_after(self._preview_failed, str(exc)) return self._safe_after(self._preview_ready, frames) def _preview_failed(self, message: str) -> None: self.preview_status_var.set(f"Preview indisponivel: {message}") def _preview_ready(self, frames: list[Path]) -> None: self.preview_frames = frames self.preview_index = 0 for button in (self.previous_button, self.play_button, self.next_button): button.configure(state="normal") self._display_preview_frame() def _display_preview_frame(self) -> None: if not self.preview_frames: return try: path = self.preview_frames[self.preview_index] if Image is not None and ImageTk is not None: with Image.open(path) as image: image = image.convert("RGB") image.thumbnail((700, 520)) self.preview_photo = ImageTk.PhotoImage(image) else: self.preview_photo = tk.PhotoImage(file=str(path)) self.preview_frame_label.configure(image=self.preview_photo, text="") self.preview_status_var.set( f"Resultado: quadro {self.preview_index + 1}/{len(self.preview_frames)}" ) except (OSError, tk.TclError) as exc: self.preview_status_var.set(str(exc)) def _previous_preview(self) -> None: if self.preview_frames: self.preview_index = (self.preview_index - 1) % len(self.preview_frames) self._display_preview_frame() def _next_preview(self) -> None: if self.preview_frames: self.preview_index = (self.preview_index + 1) % len(self.preview_frames) self._display_preview_frame() def _toggle_preview(self) -> None: if not self.preview_frames: return if self.preview_playing: self._stop_preview_playback() else: self.preview_playing = True self.play_button.configure(text="Pausar") self._play_preview_frame() def _play_preview_frame(self) -> None: if not self.preview_playing or not self.preview_frames: return self._display_preview_frame() self.preview_index = (self.preview_index + 1) % len(self.preview_frames) self.preview_after_id = self.after(200, self._play_preview_frame) def _stop_preview_playback(self) -> None: self.preview_playing = False if self.preview_after_id is not None: try: self.after_cancel(self.preview_after_id) except tk.TclError: pass self.preview_after_id = None if hasattr(self, "play_button"): self.play_button.configure(text="Reproduzir") def _open_output_folder(self) -> None: if self.last_output_folder is None or not self.last_output_folder.exists(): return try: if os.name == "nt": os.startfile(str(self.last_output_folder)) elif sys.platform == "darwin": subprocess.Popen(["open", str(self.last_output_folder)]) else: subprocess.Popen(["xdg-open", str(self.last_output_folder)]) except OSError as exc: messagebox.showerror("Erro", str(exc)) def _on_close(self) -> None: self.closing = True self.cancel_event.set() self._stop_preview_playback() if self.preview_tempdir is not None: self.preview_tempdir.cleanup() self._save_default_settings() self.destroy() def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Encontra um loop visual em um video e o repete ate a duracao desejada." ) parser.add_argument("-i", "--input", type=Path, help="video de entrada") parser.add_argument("-o", "--output", type=Path, help="arquivo de saida") parser.add_argument( "-t", "--target", type=float, default=DEFAULT_TARGET_SECONDS, help="duracao minima desejada para a saida, em segundos (padrao: 20)", ) parser.add_argument( "-m", "--min-loop", type=float, default=DEFAULT_MIN_LOOP_SECONDS, help="duracao minima permitida para o loop, em segundos (padrao: 4)", ) parser.add_argument( "--analyze-only", action="store_true", help="apenas mostra o loop encontrado, sem renderizar a saida", ) return parser def run_cli(args: argparse.Namespace) -> int: if args.input is None: LoopApp().mainloop() return 0 if ( not math.isfinite(args.target) or not math.isfinite(args.min_loop) or args.target <= 0 or args.min_loop <= 0 ): print("target e min-loop devem ser maiores que zero.", file=sys.stderr) return 2 input_path = args.input output_path = args.output or default_output_path(input_path) try: info, candidate, fps = analyze_video( input_path, args.min_loop, status=lambda text: print(text, flush=True), ) print( f"Loop encontrado: inicio {candidate.start:.2f}s, " f"duracao {candidate.duration:.2f}s (analise em {fps:.1f} fps)." ) if args.analyze_only: return 0 repeats, output_duration = render_loop( input_path, output_path, candidate, args.target, status=lambda text: print(text, flush=True), ) except VideoLoopError as exc: print(f"Erro: {exc}", file=sys.stderr) return 1 print(f"Saida: {output_path}") print(f"Repeticoes: {repeats}; duracao final: {output_duration:.2f}s") return 0 def main() -> int: return run_cli(build_parser().parse_args()) if __name__ == "__main__": raise SystemExit(main())