loop-studio / loop_video.py
menochoppa's picture
Upload loop_video.py with huggingface_hub
86c5e73 verified
Raw
History Blame Contribute Delete
162 kB
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("<<ComboboxSelected>>", 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("<KeyRelease>", self._update_editor_overlay)
self.watermark_position_combo.bind(
"<<ComboboxSelected>>", 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("<<ComboboxSelected>>", 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("<Button-1>", self._editor_press)
self.preview_editor_canvas.bind("<B1-Motion>", 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("<<Drop>>", 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("<<ListboxSelect>>", 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("<<ComboboxSelected>>", 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("<<ComboboxSelected>>", 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("<Button-1>", self._editor_press)
self.editor_canvas.bind("<B1-Motion>", self._editor_drag)
self.editor_canvas.bind("<MouseWheel>", self._editor_wheel)
self.editor_canvas.bind("<Configure>", 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("<<ComboboxSelected>>", 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("<KeyRelease>", 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("<<Drop>>", 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())