From f22233c4802a9a5aa89068e63d118ebf2428b72a Mon Sep 17 00:00:00 2001 From: eaikarensantos-gif Date: Fri, 12 Jun 2026 00:12:42 -0300 Subject: [PATCH 1/4] Improve render.py robustness and performance - Surface ffmpeg stderr on failure instead of swallowing it - Match output fps to the source instead of forcing 24fps - Probe each source once (HDR + orientation + fps) with a cache - Encode segments in parallel (up to 4 ffmpeg processes) - Detect libass support by probing -filters instead of hardcoding paths - Refuse lossless concat when segment resolutions mismatch - Escape subtitle paths safely for the filtergraph (brackets, quotes) - Use -map 0:a? so audioless sources don't fail the composite Co-Authored-By: Claude Fable 5 --- helpers/render.py | 201 ++++++++++++++++++++++++++++++++++++---------- 1 file changed, 159 insertions(+), 42 deletions(-) diff --git a/helpers/render.py b/helpers/render.py index 0d02cffa..3959a72d 100644 --- a/helpers/render.py +++ b/helpers/render.py @@ -26,6 +26,7 @@ import re import subprocess import sys +from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path try: @@ -55,6 +56,43 @@ def auto_grade_for_clip(video, start=0.0, duration=None, verbose=False): # type "Alignment=2,MarginV=90" ) +# -------- ffmpeg binary detection ------------------------------------------- +# Some ffmpeg builds (e.g. plain brew ffmpeg) lack libass, so the `subtitles` +# filter is unavailable. Detect a libass-capable binary by probing -filters +# instead of hardcoding install paths. +import os as _os +import shutil as _shutil + + +def _has_subtitles_filter(binary: str) -> bool: + try: + out = subprocess.run( + [binary, "-hide_banner", "-filters"], + capture_output=True, text=True, timeout=15, + ) + return " subtitles " in out.stdout + except Exception: + return False + + +def _detect_ffmpeg() -> tuple[str, str]: + """Return (ffmpeg, ffmpeg_for_subtitles).""" + default = _shutil.which("ffmpeg") or "ffmpeg" + if _has_subtitles_filter(default): + return default, default + # Known alternate locations for full builds (Apple Silicon / Intel brew). + for candidate in ( + "/opt/homebrew/opt/ffmpeg-full/bin/ffmpeg", + "/usr/local/opt/ffmpeg-full/bin/ffmpeg", + ): + if _os.path.exists(candidate) and _has_subtitles_filter(candidate): + return default, candidate + return default, default + + +FFMPEG, FFMPEG_SUBS = _detect_ffmpeg() + + # -------- Helpers ------------------------------------------------------------ @@ -64,6 +102,16 @@ def run(cmd: list[str], quiet: bool = False) -> None: subprocess.run(cmd, check=True) +def run_ffmpeg(cmd: list[str]) -> None: + """Run an ffmpeg command quietly, but surface stderr if it fails.""" + proc = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) + if proc.returncode != 0: + stderr = proc.stderr.decode(errors="replace") if proc.stderr else "" + tail = "\n".join(stderr.strip().splitlines()[-15:]) + print(f"\nffmpeg failed (exit {proc.returncode}):\n $ {' '.join(str(c) for c in cmd)}\n{tail}", file=sys.stderr) + raise subprocess.CalledProcessError(proc.returncode, cmd, stderr=stderr) + + def resolve_grade_filter(grade_field: str | None) -> str: """The EDL's 'grade' field can be a preset name, a raw ffmpeg filter, or 'auto'. @@ -116,34 +164,52 @@ def resolve_path(maybe_path: str, base: Path) -> Path: "format=yuv420p" ) - -def is_hdr_source(video: Path) -> bool: - """Return True if the source uses a PQ or HLG transfer function.""" - try: - out = subprocess.run( - ["ffprobe", "-v", "error", "-select_streams", "v:0", - "-show_entries", "stream=color_transfer", - "-of", "default=noprint_wrappers=1:nokey=1", str(video)], - capture_output=True, text=True, check=True, - ) - return out.stdout.strip() in HDR_TRANSFERS - except subprocess.CalledProcessError: - return False +# One ffprobe per source — segments usually share sources, so cache the result. +_PROBE_CACHE: dict[str, dict] = {} -def is_portrait_source(video: Path) -> bool: - """Return True if the video's height > width (portrait / vertical).""" +def probe_source(video: Path) -> dict: + """Return {'hdr': bool, 'portrait': bool, 'fps': float|None} for a source.""" + key = str(video.resolve()) + if key in _PROBE_CACHE: + return _PROBE_CACHE[key] + info = {"hdr": False, "portrait": False, "fps": None} try: out = subprocess.run( ["ffprobe", "-v", "error", "-select_streams", "v:0", - "-show_entries", "stream=width,height", - "-of", "csv=p=0", str(video)], + "-show_entries", "stream=width,height,color_transfer,r_frame_rate", + "-of", "json", str(video)], capture_output=True, text=True, check=True, ) - w, h = map(int, out.stdout.strip().split(",")) - return h > w + stream = json.loads(out.stdout)["streams"][0] + info["hdr"] = stream.get("color_transfer", "") in HDR_TRANSFERS + w, h = int(stream.get("width", 0)), int(stream.get("height", 0)) + info["portrait"] = h > w + rate = stream.get("r_frame_rate", "") + if "/" in rate: + num, den = rate.split("/") + if float(den) > 0: + info["fps"] = float(num) / float(den) except Exception: - return False + pass + _PROBE_CACHE[key] = info + return info + + +def project_fps(edl: dict, edit_dir: Path) -> float: + """Pick one fps for the whole render: the first source's native rate. + + Forcing a fixed 24 on 30/60fps footage drops/duplicates frames; matching + the source keeps motion smooth. All segments share one rate so the + lossless concat stays valid. + """ + for r in edl.get("ranges", []): + src = resolve_path(edl["sources"][r["source"]], edit_dir) + fps = probe_source(src).get("fps") + if fps: + # Cap at 60 to keep encode times sane on slo-mo sources. + return min(round(fps, 3), 60.0) + return 24.0 # -------- Per-segment extraction (Rule 2 + Rule 3) -------------------------- @@ -157,6 +223,7 @@ def extract_segment( out_path: Path, preview: bool = False, draft: bool = False, + fps: float = 24.0, ) -> None: """Extract a cut range as its own MP4 with grade + 30ms audio fades baked in. @@ -170,14 +237,15 @@ def extract_segment( """ out_path.parent.mkdir(parents=True, exist_ok=True) - portrait = is_portrait_source(source) + info = probe_source(source) + portrait = info["portrait"] if draft: scale = "scale=-2:1280" if portrait else "scale=1280:-2" else: scale = "scale=-2:1920" if portrait else "scale=1920:-2" vf_parts: list[str] = [] - if is_hdr_source(source): + if info["hdr"]: vf_parts.append(TONEMAP_CHAIN) vf_parts.append(scale) if grade_filter: @@ -196,19 +264,19 @@ def extract_segment( preset, crf = "fast", "20" cmd = [ - "ffmpeg", "-y", + FFMPEG, "-y", "-ss", f"{seg_start:.3f}", "-i", str(source), "-t", f"{duration:.3f}", "-vf", vf, "-af", af, "-c:v", "libx264", "-preset", preset, "-crf", crf, - "-pix_fmt", "yuv420p", "-r", "24", + "-pix_fmt", "yuv420p", "-r", f"{fps:g}", "-c:a", "aac", "-b:a", "192k", "-ar", "48000", "-movflags", "+faststart", str(out_path), ] - subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) + run_ffmpeg(cmd) def extract_all_segments( @@ -234,10 +302,12 @@ def extract_all_segments( ranges = edl["ranges"] sources = edl["sources"] - seg_paths: list[Path] = [] - print(f"extracting {len(ranges)} segment(s) → {clips_dir.name}/") + fps = project_fps(edl, edit_dir) + print(f"extracting {len(ranges)} segment(s) → {clips_dir.name}/ ({fps:g} fps)") if is_auto: print(" (auto-grade per segment: analyzing each range)") + + jobs: list[tuple[int, Path, float, float, str, Path]] = [] for i, r in enumerate(ranges): src_name = r["source"] src_path = resolve_path(sources[src_name], edit_dir) @@ -255,23 +325,60 @@ def extract_all_segments( print(f" [{i:02d}] {src_name} {start:7.2f}-{end:7.2f} ({duration:5.2f}s) {note}") if is_auto: print(f" grade: {seg_filter or '(none)'}") - extract_segment(src_path, start, duration, seg_filter, out_path, preview=preview, draft=draft) - seg_paths.append(out_path) + jobs.append((i, src_path, start, duration, seg_filter, out_path)) + + # Encode segments in parallel — each is an independent ffmpeg process. + workers = min(4, max(1, len(jobs))) + with ThreadPoolExecutor(max_workers=workers) as pool: + futures = { + pool.submit( + extract_segment, src, start, dur, flt, out, + preview=preview, draft=draft, fps=fps, + ): i + for i, src, start, dur, flt, out in jobs + } + for fut in as_completed(futures): + fut.result() # re-raise the first failure - return seg_paths + return [job[5] for job in jobs] # -------- Lossless concat ---------------------------------------------------- +def _segment_signature(path: Path) -> tuple: + """Resolution of an encoded segment — must be uniform for -c copy concat.""" + try: + out = subprocess.run( + ["ffprobe", "-v", "error", "-select_streams", "v:0", + "-show_entries", "stream=width,height", + "-of", "csv=p=0", str(path)], + capture_output=True, text=True, check=True, + ) + return tuple(out.stdout.strip().split(",")) + except Exception: + return ("?",) + + def concat_segments(segment_paths: list[Path], out_path: Path, edit_dir: Path) -> None: """Lossless concat via the concat demuxer. No re-encode.""" + # -c copy silently produces a broken file if segments differ in + # resolution/orientation (e.g. portrait + landscape sources in one EDL). + sigs = {p.name: _segment_signature(p) for p in segment_paths} + if len(set(sigs.values())) > 1: + detail = "\n".join(f" {name}: {'x'.join(sig)}" for name, sig in sigs.items()) + sys.exit( + "error: segments have mismatched resolutions — lossless concat would " + f"produce a corrupt file:\n{detail}\n" + "Reframe sources to a common orientation (helpers/reframe.py) first." + ) + out_path.parent.mkdir(parents=True, exist_ok=True) concat_list = edit_dir / "_concat.txt" concat_list.write_text("".join(f"file '{p.resolve()}'\n" for p in segment_paths)) cmd = [ - "ffmpeg", "-y", + FFMPEG, "-y", "-f", "concat", "-safe", "0", "-i", str(concat_list), "-c", "copy", @@ -279,7 +386,7 @@ def concat_segments(segment_paths: list[Path], out_path: Path, edit_dir: Path) - str(out_path), ] print(f"concat → {out_path.name}") - subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) + run_ffmpeg(cmd) concat_list.unlink(missing_ok=True) @@ -404,7 +511,7 @@ def measure_loudness(video_path: Path) -> dict[str, str] | None: f"loudnorm=I={LOUDNORM_I}:TP={LOUDNORM_TP}:LRA={LOUDNORM_LRA}:print_format=json" ) cmd = [ - "ffmpeg", "-y", "-hide_banner", "-nostats", + FFMPEG, "-y", "-hide_banner", "-nostats", "-i", str(video_path), "-af", filter_str, "-vn", "-f", "null", "-", @@ -445,7 +552,7 @@ def apply_loudnorm_two_pass( # One-pass approximation — faster, slightly less accurate. filter_str = f"loudnorm=I={LOUDNORM_I}:TP={LOUDNORM_TP}:LRA={LOUDNORM_LRA}" cmd = [ - "ffmpeg", "-y", "-hide_banner", "-nostats", + FFMPEG, "-y", "-hide_banner", "-nostats", "-i", str(input_path), "-c:v", "copy", "-af", filter_str, @@ -454,7 +561,7 @@ def apply_loudnorm_two_pass( str(output_path), ] print(f" loudnorm (1-pass preview) → {output_path.name}") - subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) + run_ffmpeg(cmd) return True # Full two-pass @@ -477,7 +584,7 @@ def apply_loudnorm_two_pass( f":linear=true" ) cmd = [ - "ffmpeg", "-y", "-hide_banner", "-nostats", + FFMPEG, "-y", "-hide_banner", "-nostats", "-i", str(input_path), "-c:v", "copy", "-af", filter_str, @@ -486,7 +593,7 @@ def apply_loudnorm_two_pass( str(output_path), ] print(f" loudnorm pass 2: normalizing → {output_path.name}") - subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) + run_ffmpeg(cmd) return True @@ -509,7 +616,7 @@ def build_final_composite( if not has_overlays and not has_subs: # Nothing to do — just rename/copy base to final name - run(["ffmpeg", "-y", "-i", str(base_path), "-c", "copy", str(out_path)], quiet=True) + run([FFMPEG, "-y", "-i", str(base_path), "-c", "copy", str(out_path)], quiet=True) return inputs: list[str] = ["-i", str(base_path)] @@ -537,7 +644,15 @@ def build_final_composite( # Subtitles LAST — Rule 1 if has_subs: - subs_abs = str(subtitles_path.resolve()).replace(":", r"\:").replace("'", r"\'") + # Filtergraph escaping: the surrounding single quotes protect [ ] , ; + # at the graph level; ':' still needs escaping at the filter-arg level, + # and a literal ' must be emitted as '\'' (close, escaped quote, reopen). + subs_abs = ( + str(subtitles_path.resolve()) + .replace("\\", "/") + .replace(":", r"\:") + .replace("'", r"'\''") + ) filter_parts.append( f"{current}subtitles='{subs_abs}':force_style='{SUB_FORCE_STYLE}'[outv]" ) @@ -552,12 +667,14 @@ def build_final_composite( filter_complex = ";".join(filter_parts) + # Use FFMPEG_SUBS if subtitles are being burned (requires libass) + _ffmpeg_bin = FFMPEG_SUBS if has_subs else FFMPEG cmd = [ - "ffmpeg", "-y", + _ffmpeg_bin, "-y", *inputs, "-filter_complex", filter_complex, "-map", out_label, - "-map", "0:a", + "-map", "0:a?", "-c:v", "libx264", "-preset", "fast", "-crf", "18", "-pix_fmt", "yuv420p", "-c:a", "copy", @@ -566,7 +683,7 @@ def build_final_composite( ] print(f"compositing → {out_path.name}") print(f" overlays: {len(overlays)}, subtitles: {'yes' if has_subs else 'no'}") - subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) + run_ffmpeg(cmd) # -------- Main --------------------------------------------------------------- From 2716683153e9b0f2b694df0c77cc8e72d6ccb27b Mon Sep 17 00:00:00 2001 From: eaikarensantos-gif Date: Fri, 12 Jun 2026 00:12:51 -0300 Subject: [PATCH 2/4] Add new editing helpers (batch, denoise, reframe, subtitles, etc.) Co-Authored-By: Claude Fable 5 --- helpers/batch.py | 177 ++++++++++++++++++++++++++ helpers/burn_subs_pil.py | 195 +++++++++++++++++++++++++++++ helpers/check_gaps.py | 87 +++++++++++++ helpers/correct_transcript.py | 101 +++++++++++++++ helpers/denoise.py | 115 +++++++++++++++++ helpers/detect_breaths.py | 227 ++++++++++++++++++++++++++++++++++ helpers/export_formats.py | 115 +++++++++++++++++ helpers/reframe.py | 198 +++++++++++++++++++++++++++++ helpers/thumbnail.py | 206 ++++++++++++++++++++++++++++++ 9 files changed, 1421 insertions(+) create mode 100644 helpers/batch.py create mode 100644 helpers/burn_subs_pil.py create mode 100644 helpers/check_gaps.py create mode 100644 helpers/correct_transcript.py create mode 100644 helpers/denoise.py create mode 100644 helpers/detect_breaths.py create mode 100644 helpers/export_formats.py create mode 100644 helpers/reframe.py create mode 100644 helpers/thumbnail.py diff --git a/helpers/batch.py b/helpers/batch.py new file mode 100644 index 00000000..6b121668 --- /dev/null +++ b/helpers/batch.py @@ -0,0 +1,177 @@ +#!/usr/bin/env python3 +""" +batch.py — processa múltiplos vídeos em sequência: + 1. Transcreve (Scribe, cached) + 2. Corrige português (Claude API) + 3. Detecta gaps internos (check_gaps) + 4. Gera EDL via sub-agente Claude + 5. Renderiza + 6. Exporta formatos + +Uso: + python3 batch.py [--task remove-silence] [--formats reels feed] + python3 batch.py /path/to/videos --task remove-silence --formats reels landscape +""" + +import argparse +import json +import os +import subprocess +import sys +from pathlib import Path + + +HELPERS = Path(__file__).parent + + +def run(cmd, description=""): + if description: + print(f" → {description}") + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + print(f" ⚠️ ERRO: {result.stderr[-300:]}", file=sys.stderr) + return result.returncode == 0, result.stdout, result.stderr + + +def find_videos(videos_dir, extensions=(".mp4", ".mov", ".MP4", ".MOV")): + return sorted([ + str(p) for p in Path(videos_dir).iterdir() + if p.suffix in extensions and not p.stem.startswith(".") + ]) + + +def transcribe(video_path, edit_dir): + ok, out, _ = run([ + sys.executable, str(HELPERS / "transcribe.py"), video_path, + "--edit-dir", edit_dir + ], "transcrever") + return ok + + +def correct_transcript(transcript_path): + corrected = str(Path(transcript_path).with_suffix(".corrected.json")) + if os.path.exists(corrected): + print(f" ↩ correção já existe: {corrected}") + return corrected + ok, _, _ = run([ + sys.executable, str(HELPERS / "correct_transcript.py"), + transcript_path, "-o", corrected + ], "corrigir português") + return corrected if ok else transcript_path + + +def check_gaps_report(edl_path, transcripts_dir): + ok, out, _ = run([ + sys.executable, str(HELPERS / "check_gaps.py"), + edl_path, transcripts_dir, "--min-ms", "300" + ], "checar gaps internos") + return out + + +def render(edl_path, out_path): + ok, _, _ = run([ + sys.executable, str(HELPERS / "render.py"), + edl_path, "-o", out_path + ], f"renderizar → {out_path}") + return ok + + +def export(final_path, formats, out_dir): + run([ + sys.executable, str(HELPERS / "export_formats.py"), + final_path, + "--formats", *formats, + "--out-dir", out_dir + ], f"exportar {formats}") + + +def process_video(video_path, edit_dir, task, formats): + stem = Path(video_path).stem + print(f"\n{'='*60}") + print(f" {stem}") + print(f"{'='*60}") + + # 1. transcrever + transcripts_dir = os.path.join(edit_dir, "transcripts") + os.makedirs(transcripts_dir, exist_ok=True) + transcript_path = os.path.join(transcripts_dir, f"{stem}.json") + + if os.path.exists(transcript_path): + print(f" ↩ transcrição já existe") + else: + if not transcribe(video_path, edit_dir): + print(f" ✗ falha na transcrição — pulando") + return False + + # 2. corrigir português + corrected_path = correct_transcript(transcript_path) + + # 3. pack transcripts + run([sys.executable, str(HELPERS / "pack_transcripts.py"), + "--edit-dir", edit_dir], "pack transcripts") + + print(f"\n ⚠️ EDL manual necessário para {stem}") + print(f" Transcrição disponível em: {corrected_path}") + print(f" Use Claude para gerar o EDL e depois rode:") + print(f" python3 render.py -o {edit_dir}/final_{stem}.mp4") + + edl_path = os.path.join(edit_dir, f"edl_{stem}.json") + if not os.path.exists(edl_path): + print(f" ↩ EDL não encontrado — aguardando geração manual") + return False + + # 4. checar gaps + gaps_report = check_gaps_report(edl_path, transcripts_dir) + if "gap" in gaps_report: + print(f"\n ⚠️ GAPS DETECTADOS:\n{gaps_report}") + + # 5. renderizar + final_path = os.path.join(edit_dir, f"final_{stem}.mp4") + if not render(edl_path, final_path): + return False + + # 6. exportar formatos + if formats: + exports_dir = os.path.join(edit_dir, "exports") + os.makedirs(exports_dir, exist_ok=True) + export(final_path, formats, exports_dir) + + print(f"\n ✓ {stem} concluído → {final_path}") + return True + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("videos_dir") + parser.add_argument("--task", default="remove-silence", + help="Tarefa a executar (padrão: remove-silence)") + parser.add_argument("--formats", nargs="*", + default=[], + help="Formatos de export (reels feed stories landscape)") + parser.add_argument("--edit-dir", default=None) + args = parser.parse_args() + + videos = find_videos(args.videos_dir) + if not videos: + sys.exit(f"Nenhum vídeo encontrado em {args.videos_dir}") + + edit_dir = args.edit_dir or os.path.join(args.videos_dir, "edit") + os.makedirs(edit_dir, exist_ok=True) + + print(f"Batch: {len(videos)} vídeo(s) em {args.videos_dir}") + print(f"Task: {args.task}") + print(f"Formatos: {args.formats or 'nenhum'}") + + results = {"ok": [], "fail": []} + for vp in videos: + ok = process_video(vp, edit_dir, args.task, args.formats) + (results["ok"] if ok else results["fail"]).append(Path(vp).stem) + + print(f"\n{'='*60}") + print(f"Batch concluído: {len(results['ok'])} ok, {len(results['fail'])} falhas") + if results["fail"]: + print(f"Falhas: {results['fail']}") + + +if __name__ == "__main__": + main() diff --git a/helpers/burn_subs_pil.py b/helpers/burn_subs_pil.py new file mode 100644 index 00000000..84ebe850 --- /dev/null +++ b/helpers/burn_subs_pil.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python3 +"""Gera uma faixa de legendas como vídeo RGBA usando PIL. +Não requer libass nem freetype no ffmpeg. + +Uso: + python burn_subs_pil.py [fps] +""" +from __future__ import annotations +import re, sys, subprocess, tempfile, os +from pathlib import Path +from PIL import Image, ImageDraw, ImageFont + +W, H = 1080, 1920 +FONT_SIZE = 60 +MARGIN_BOTTOM = int(H * 0.30) # 30% do fundo — safe zone Reels/TikTok +MAX_LINE_W = int(W * 0.82) +PAD = 12 +OUTLINE = 3 + + +def ts_to_s(ts: str) -> float: + h, m, rest = ts.split(":") + s, ms = rest.split(",") + return int(h) * 3600 + int(m) * 60 + int(s) + int(ms) / 1000 + + +def parse_srt(path: str) -> list[tuple[float, float, str]]: + text = Path(path).read_text(encoding="utf-8") + blocks = re.split(r"\n{2,}", text.strip()) + cues: list[tuple[float, float, str]] = [] + for block in blocks: + lines = block.strip().splitlines() + if len(lines) < 2: + continue + # find timestamp line + ts_line = None + text_lines: list[str] = [] + for i, line in enumerate(lines): + if "-->" in line: + ts_line = line + text_lines = lines[i + 1:] + break + if ts_line is None: + continue + m = re.match(r"(\S+)\s+-->\s+(\S+)", ts_line) + if not m: + continue + start = ts_to_s(m.group(1)) + end = ts_to_s(m.group(2)) + cue_text = " ".join(text_lines).strip() + if cue_text: + cues.append((start, end, cue_text)) + return cues + + +def load_font(size: int) -> ImageFont.FreeTypeFont: + candidates = [ + "/System/Library/Fonts/Helvetica.ttc", + "/System/Library/Fonts/HelveticaNeue.ttc", + "/System/Library/Fonts/SFNS.ttf", + "/System/Library/Fonts/SFNSDisplay.ttf", + "/Library/Fonts/Arial Bold.ttf", + "/Library/Fonts/Arial.ttf", + ] + for path in candidates: + if os.path.exists(path): + try: + return ImageFont.truetype(path, size) + except Exception: + continue + return ImageFont.load_default() + + +def render_cue_png(text: str, font: ImageFont.FreeTypeFont, out_path: str) -> None: + """Renderiza uma imagem RGBA 1080x1920 com a legenda no fundo.""" + img = Image.new("RGBA", (W, H), (0, 0, 0, 0)) + draw = ImageDraw.Draw(img) + + # Wrap se necessário + words = text.split() + lines: list[str] = [] + current = "" + for word in words: + test = (current + " " + word).strip() + bbox = font.getbbox(test) + if bbox[2] - bbox[0] > MAX_LINE_W and current: + lines.append(current) + current = word + else: + current = test + if current: + lines.append(current) + + line_bboxes = [font.getbbox(l) for l in lines] + line_heights = [bb[3] - bb[1] for bb in line_bboxes] + total_h = sum(line_heights) + 4 * (len(lines) - 1) + max_w = max(bb[2] - bb[0] for bb in line_bboxes) + + y0 = H - MARGIN_BOTTOM - total_h + x0 = (W - max_w) // 2 + + # Fundo semi-transparente + draw.rectangle( + [x0 - PAD, y0 - PAD, x0 + max_w + PAD, y0 + total_h + PAD], + fill=(0, 0, 0, 180), + ) + + y = y0 + for line, bb, lh in zip(lines, line_bboxes, line_heights): + lw = bb[2] - bb[0] + x = (W - lw) // 2 + # Contorno preto + for dx in range(-OUTLINE, OUTLINE + 1): + for dy in range(-OUTLINE, OUTLINE + 1): + if dx or dy: + draw.text((x + dx, y + dy), line, font=font, fill=(0, 0, 0, 255)) + # Texto branco + draw.text((x, y), line, font=font, fill=(255, 255, 255, 255)) + y += lh + 4 + + img.save(out_path, "PNG") + + +def generate_subtitle_video( + srt_path: str, duration_s: float, out_path: str, fps: int = 24 +) -> None: + cues = parse_srt(srt_path) + font = load_font(FONT_SIZE) + + with tempfile.TemporaryDirectory() as tmp: + # Gera um PNG por cue único + png_cache: dict[str, str] = {} + for i, (_, _, text) in enumerate(cues): + if text not in png_cache: + p = os.path.join(tmp, f"cue_{i:04d}.png") + render_cue_png(text, font, p) + png_cache[text] = p + + # PNG transparente (silêncio / sem legenda) + blank_path = os.path.join(tmp, "blank.png") + Image.new("RGBA", (W, H), (0, 0, 0, 0)).save(blank_path, "PNG") + + # Monta timeline: (t_start, t_end, png_path) + timeline: list[tuple[float, float, str]] = [] + prev = 0.0 + for start, end, text in sorted(cues, key=lambda c: c[0]): + if start > prev + 0.001: + timeline.append((prev, start, blank_path)) + timeline.append((start, end, png_cache[text])) + prev = end + if prev < duration_s: + timeline.append((prev, duration_s, blank_path)) + + # Concat list para ffmpeg + concat_file = os.path.join(tmp, "concat.txt") + with open(concat_file, "w") as f: + for t_start, t_end, png_path in timeline: + dur = max(1 / fps, t_end - t_start) + f.write(f"file '{png_path}'\n") + f.write(f"duration {dur:.4f}\n") + # Último frame precisa ser repetido (bug concat demuxer) + if timeline: + f.write(f"file '{timeline[-1][2]}'\n") + + cmd = [ + "ffmpeg", "-y", + "-f", "concat", "-safe", "0", + "-i", concat_file, + "-vf", f"scale={W}:{H},format=rgba", + "-r", str(fps), + "-c:v", "libx264", + "-pix_fmt", "yuva420p", + "-preset", "fast", + "-crf", "18", + out_path, + ] + print(f" gerando faixa de legendas → {Path(out_path).name}") + result = subprocess.run(cmd, capture_output=True) + if result.returncode != 0: + # Fallback: yuv420p se yuva420p não suportado + cmd[cmd.index("yuva420p")] = "yuv420p" + result2 = subprocess.run(cmd, capture_output=True) + if result2.returncode != 0: + print("AVISO: não foi possível gerar faixa de legendas:") + print(result2.stderr.decode()[-500:]) + return + print(f" OK: {Path(out_path).name}") + + +if __name__ == "__main__": + srt = sys.argv[1] + dur = float(sys.argv[2]) + out = sys.argv[3] + fps = int(sys.argv[4]) if len(sys.argv) > 4 else 24 + generate_subtitle_video(srt, dur, out, fps) diff --git a/helpers/check_gaps.py b/helpers/check_gaps.py new file mode 100644 index 00000000..df879829 --- /dev/null +++ b/helpers/check_gaps.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +""" +check_gaps.py — detecta gaps >= MIN_MS dentro de cada segmento do EDL. +Uso: + python3 check_gaps.py [--min-ms 300] +""" + +import json +import sys +import argparse +from pathlib import Path + +def load_words(transcript_path): + with open(transcript_path) as f: + d = json.load(f) + return [w for w in d.get("words", []) if w.get("type") == "word"] + +def check_edl(edl_path, transcripts_dir, min_ms=300): + min_s = min_ms / 1000.0 + + with open(edl_path) as f: + edl = json.load(f) + + # index transcripts by source name + transcripts = {} + for source_name in edl["sources"]: + candidates = list(Path(transcripts_dir).glob(f"{source_name}.json")) + if candidates: + transcripts[source_name] = load_words(str(candidates[0])) + + problems = [] + + for i, seg in enumerate(edl["ranges"]): + src = seg["source"] + seg_start = seg["start"] + seg_end = seg["end"] + words = transcripts.get(src, []) + + # words that fall inside this segment + seg_words = [w for w in words + if w["start"] >= seg_start - 0.1 + and w["end"] <= seg_end + 0.1] + + if len(seg_words) < 2: + continue + + for j in range(len(seg_words) - 1): + gap_start = seg_words[j]["end"] + gap_end = seg_words[j + 1]["start"] + gap_s = gap_end - gap_start + if gap_s >= min_s: + problems.append({ + "seg_index": i, + "beat": seg.get("beat", f"seg_{i}"), + "source_range": f"{seg_start:.2f}-{seg_end:.2f}", + "gap_start": gap_start, + "gap_end": gap_end, + "gap_ms": int(gap_s * 1000), + "before": seg_words[j]["text"], + "after": seg_words[j + 1]["text"], + }) + + return problems + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("edl") + parser.add_argument("transcripts_dir") + parser.add_argument("--min-ms", type=int, default=300) + args = parser.parse_args() + + problems = check_edl(args.edl, args.transcripts_dir, args.min_ms) + + if not problems: + print(f"✅ Nenhum gap >= {args.min_ms}ms encontrado dentro dos segmentos.") + return + + print(f"⚠️ {len(problems)} gap(s) >= {args.min_ms}ms dentro dos segmentos:\n") + for p in problems: + print(f" [{p['seg_index']:02d}] {p['beat']:<22} " + f"gap {p['gap_ms']}ms " + f"entre {p['before']!r} ({p['gap_start']:.3f}s) " + f"→ {p['after']!r} ({p['gap_end']:.3f}s) " + f"[source {p['source_range']}]") + +if __name__ == "__main__": + main() diff --git a/helpers/correct_transcript.py b/helpers/correct_transcript.py new file mode 100644 index 00000000..277d181f --- /dev/null +++ b/helpers/correct_transcript.py @@ -0,0 +1,101 @@ +#!/usr/bin/env python3 +""" +correct_transcript.py — corrige erros ortográficos e de reconhecimento de fala +no JSON do Scribe usando Claude API, preservando os timestamps palavra por palavra. + +Uso: + python3 correct_transcript.py [-o corrected.json] +""" + +import argparse +import json +import os +import sys +from pathlib import Path + +try: + import anthropic +except ImportError: + sys.exit("anthropic não instalado. Rode: pip3 install anthropic") + + +SYSTEM_PROMPT = """Você é um revisor de transcrições automáticas em português brasileiro. +Receberá uma lista de palavras transcritas por ASR (reconhecimento de fala automático). +Sua tarefa: +1. Corrigir erros ortográficos mantendo o tom coloquial da fala (ex: "tá" é aceitável, não mude para "está") +2. Corrigir nomes próprios e siglas (ex: "fiap" → "FIAP", "linkedin" → "LinkedIn", "mec" → "MEC") +3. Corrigir palavras cortadas pelo ASR se o contexto deixar claro (ex: "Laboratória" → "Laboratoria" se for nome de empresa) +4. NÃO mudar palavras informais naturais da fala (tá, né, pra, vamo, a gente, etc.) +5. NÃO alterar a ordem das palavras +6. NÃO adicionar nem remover palavras +7. Retornar EXATAMENTE o mesmo número de palavras, uma por linha, na mesma ordem + +Retorne APENAS as palavras corrigidas, uma por linha, sem numeração, sem explicações.""" + + +def correct_words(words, api_key=None): + client = anthropic.Anthropic(api_key=api_key or os.environ.get("ANTHROPIC_API_KEY")) + + # só palavras reais (ignora audio_event e spacing) + word_tokens = [(i, w) for i, w in enumerate(words) if w.get("type") == "word"] + if not word_tokens: + return words + + texts = [w["text"] for _, w in word_tokens] + + # envia em batches de 200 para não ultrapassar context + batch_size = 200 + corrected_texts = [] + + for start in range(0, len(texts), batch_size): + batch = texts[start:start + batch_size] + prompt = "\n".join(batch) + + response = client.messages.create( + model="claude-haiku-4-5", + max_tokens=2048, + system=SYSTEM_PROMPT, + messages=[{"role": "user", "content": prompt}] + ) + + fixed = response.content[0].text.strip().split("\n") + + # garante mesmo número de palavras + if len(fixed) != len(batch): + print(f" ⚠️ batch {start//batch_size+1}: retornou {len(fixed)} palavras, esperado {len(batch)} — mantendo original", file=sys.stderr) + corrected_texts.extend(batch) + else: + corrected_texts.extend(fixed) + + # aplica correções de volta + result = list(words) + for (orig_idx, _), corrected_text in zip(word_tokens, corrected_texts): + if result[orig_idx]["text"] != corrected_text: + print(f" {result[orig_idx]['text']!r} → {corrected_text!r}") + result[orig_idx] = dict(result[orig_idx], text=corrected_text) + + return result + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("transcript", help="arquivo JSON do Scribe") + parser.add_argument("-o", "--output", default=None) + parser.add_argument("--api-key", default=None) + args = parser.parse_args() + + with open(args.transcript) as f: + data = json.load(f) + + print(f" corrigindo {args.transcript}...") + data["words"] = correct_words(data["words"], api_key=args.api_key) + + out_path = args.output or str(Path(args.transcript).with_suffix(".corrected.json")) + with open(out_path, "w", ensure_ascii=False) as f: + json.dump(data, f, ensure_ascii=False, indent=2) + + print(f"\ndone: {out_path}") + + +if __name__ == "__main__": + main() diff --git a/helpers/denoise.py b/helpers/denoise.py new file mode 100644 index 00000000..a08ec0a5 --- /dev/null +++ b/helpers/denoise.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +""" +denoise.py — redução de ruído de áudio em vídeos. + +Modos: + light — ruído ambiente leve (ar condicionado, sala) + medium — microfone com hiss moderado (padrão) + strong — vento, ruído forte (como o 163131) + custom — passa filtros ffmpeg diretamente via --filter + +Uso: + python3 denoise.py [--mode medium] [-o output.mp4] + python3 denoise.py --filter "afftdn=nf=-25" -o saida.mp4 +""" + +import argparse +import subprocess +import os +import sys +from pathlib import Path + + +MODES = { + "light": { + "desc": "ruído leve (sala, AC)", + # anlmdn: Non-Local Means Denoising (gentil, preserva voz) + "filter": "anlmdn=s=0.00005:p=0.002:r=0.002:m=15", + }, + "medium": { + "desc": "ruído moderado (microfone, hiss) — padrão", + # afftdn: FFT-based denoising + highpass para remover sub-bass + "filter": "afftdn=nf=-20,highpass=f=80", + }, + "strong": { + "desc": "ruído forte (vento, microfone ruim)", + # arnndn: RNN-based denoising (mais agressivo) + # + afftdn para o que sobrar + # + highpass para fundos + "filter": "arnndn=m=bd.rnnn,afftdn=nf=-15,highpass=f=100", + }, + "wind": { + "desc": "vento intenso (como nos primeiros 45s do 163131)", + "filter": "highpass=f=200,lowpass=f=8000,arnndn=m=bd.rnnn,afftdn=nf=-10", + }, +} + +# modelo arnndn (baixa automaticamente se não existir) +ARNNDN_MODEL_URL = "https://github.com/GregorR/rnnoise-models/raw/master/beguiling-drafter-2018-08-30/bd.rnnn" + + +def ensure_arnndn_model(): + model_dir = Path.home() / ".local" / "share" / "arnndn" + model_path = model_dir / "bd.rnnn" + if not model_path.exists(): + model_dir.mkdir(parents=True, exist_ok=True) + print(" baixando modelo arnndn...") + subprocess.run(["curl", "-sL", ARNNDN_MODEL_URL, "-o", str(model_path)], check=True) + return str(model_path) + + +def denoise(video_path, audio_filter, out_path): + cmd = [ + "ffmpeg", "-i", video_path, + "-af", audio_filter, + "-c:v", "copy", # vídeo sem re-encode + "-c:a", "aac", "-b:a", "192k", + out_path, "-y", "-loglevel", "error" + ] + result = subprocess.run(cmd) + return result.returncode == 0 + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("video") + parser.add_argument("--mode", choices=list(MODES.keys()), default="medium") + parser.add_argument("--filter", dest="custom_filter", default=None, + help="Filtro ffmpeg customizado (substitui --mode)") + parser.add_argument("-o", "--output", default=None) + args = parser.parse_args() + + video_path = args.video + if not os.path.exists(video_path): + sys.exit(f"Arquivo não encontrado: {video_path}") + + # resolve filtro + if args.custom_filter: + audio_filter = args.custom_filter + mode_desc = "custom" + else: + mode = MODES[args.mode] + audio_filter = mode["filter"] + mode_desc = f"{args.mode} — {mode['desc']}" + + # se o modo usa arnndn, garante o modelo + if "arnndn" in audio_filter: + model_path = ensure_arnndn_model() + audio_filter = audio_filter.replace("m=bd.rnnn", f"m={model_path}") + + stem = Path(video_path).stem + out_path = args.output or str(Path(video_path).parent / f"{stem}_denoised.mp4") + + print(f" modo: {mode_desc}") + print(f" filtro: {audio_filter}") + print(f" → {out_path}") + + if denoise(video_path, audio_filter, out_path): + size_mb = os.path.getsize(out_path) / 1024 / 1024 + print(f"\ndone: {out_path} ({size_mb:.1f} MB)") + else: + sys.exit("Erro na redução de ruído.") + + +if __name__ == "__main__": + main() diff --git a/helpers/detect_breaths.py b/helpers/detect_breaths.py new file mode 100644 index 00000000..3811448b --- /dev/null +++ b/helpers/detect_breaths.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +""" +detect_breaths.py — detecta respiros e sons de respiração no áudio de um vídeo +usando análise de amplitude via ffmpeg, independente do Scribe. + +Estratégia: + 1. Extrai áudio em mono + 2. Calcula RMS por janelas de 50ms + 3. Identifica regiões de "atividade baixa mas não-silêncio" (respiros) + usando limiares configuráveis + 4. Filtra regiões muito curtas ou muito longas + 5. Reporta timestamps e duração de cada respiro detectado + +Integração com EDL: + - Se --edl passado, verifica quais respiros estão dentro de segmentos + - Gera lista de cortes sugeridos + +Uso: + python3 detect_breaths.py [--edl edl.json] [--min-ms 80] [--max-ms 800] +""" + +import argparse +import json +import os +import subprocess +import sys +import tempfile +from pathlib import Path + +import numpy as np + + +def extract_audio_samples(video_path, sample_rate=16000): + """Extrai áudio como PCM mono via ffmpeg, retorna array numpy.""" + with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp: + tmp_path = tmp.name + + subprocess.run([ + "ffmpeg", "-i", video_path, + "-ac", "1", "-ar", str(sample_rate), + "-f", "wav", tmp_path, + "-y", "-loglevel", "error" + ], check=True) + + # lê o WAV manualmente (evita dependência de scipy) + with open(tmp_path, "rb") as f: + f.read(44) # pula header WAV de 44 bytes + raw = f.read() + os.unlink(tmp_path) + + samples = np.frombuffer(raw, dtype=np.int16).astype(np.float32) / 32768.0 + return samples, sample_rate + + +def compute_rms(samples, sr, window_ms=50): + """Calcula RMS por janelas de window_ms milissegundos.""" + window = int(sr * window_ms / 1000) + n_windows = len(samples) // window + rms = np.array([ + np.sqrt(np.mean(samples[i*window:(i+1)*window]**2)) + for i in range(n_windows) + ]) + return rms, window_ms / 1000.0 # retorna RMS e duração de cada janela em s + + +def detect_breath_regions(rms, window_s, + silence_threshold=0.005, + breath_min=0.008, + breath_max=0.12, + min_duration_ms=80, + max_duration_ms=800): + """ + Identifica regiões de respiro: + - amplitude entre silence_threshold e breath_max (não é silêncio, não é fala) + - duração entre min_duration_ms e max_duration_ms + """ + min_windows = int(min_duration_ms / (window_s * 1000)) + max_windows = int(max_duration_ms / (window_s * 1000)) + + # classifica cada janela + is_breath = (rms >= silence_threshold) & (rms <= breath_max) + + # encontra regiões contíguas + regions = [] + in_region = False + start = 0 + + for i, b in enumerate(is_breath): + if b and not in_region: + in_region = True + start = i + elif not b and in_region: + in_region = False + duration = i - start + if min_windows <= duration <= max_windows: + t_start = start * window_s + t_end = i * window_s + avg_rms = float(np.mean(rms[start:i])) + regions.append({ + "start": round(t_start, 3), + "end": round(t_end, 3), + "duration_ms": int((t_end - t_start) * 1000), + "avg_rms": round(avg_rms, 5), + }) + + # fecha região aberta no final + if in_region: + duration = len(is_breath) - start + if min_windows <= duration <= max_windows: + t_start = start * window_s + t_end = len(is_breath) * window_s + avg_rms = float(np.mean(rms[start:])) + regions.append({ + "start": round(t_start, 3), + "end": round(t_end, 3), + "duration_ms": int((t_end - t_start) * 1000), + "avg_rms": round(avg_rms, 5), + }) + + return regions + + +def load_edl_segments(edl_path): + with open(edl_path) as f: + edl = json.load(f) + return edl.get("ranges", []) + + +def filter_in_segments(breath_regions, segments): + """Retorna respiros que estão dentro de algum segmento do EDL.""" + inside = [] + for b in breath_regions: + for seg in segments: + # verifica sobreposição + if b["start"] < seg["end"] and b["end"] > seg["start"]: + inside.append({ + **b, + "segment_beat": seg.get("beat", "?"), + "segment_range": f"{seg['start']:.2f}-{seg['end']:.2f}", + }) + break + return inside + + +def calibrate_thresholds(rms): + """Estima limiares automaticamente baseado na distribuição do RMS.""" + p5 = float(np.percentile(rms, 5)) # silêncio + p30 = float(np.percentile(rms, 30)) # fala baixa / respiros + p70 = float(np.percentile(rms, 70)) # fala normal + + silence_threshold = max(p5 * 2, 0.003) + breath_max = min(p30 * 1.5, p70 * 0.5, 0.15) + + return silence_threshold, breath_max + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("video") + parser.add_argument("--edl", default=None, help="EDL JSON para filtrar por segmento") + parser.add_argument("--min-ms", type=int, default=80) + parser.add_argument("--max-ms", type=int, default=800) + parser.add_argument("--silence-threshold", type=float, default=None, + help="RMS mínimo para não ser silêncio (auto se omitido)") + parser.add_argument("--breath-max", type=float, default=None, + help="RMS máximo para ser respiro, não fala (auto se omitido)") + parser.add_argument("--window-ms", type=int, default=50) + args = parser.parse_args() + + if not os.path.exists(args.video): + sys.exit(f"Arquivo não encontrado: {args.video}") + + print(f" extraindo áudio de {args.video}...") + samples, sr = extract_audio_samples(args.video) + + print(f" calculando RMS ({args.window_ms}ms por janela)...") + rms, window_s = compute_rms(samples, sr, args.window_ms) + + # calibração automática + silence_thr, breath_max = calibrate_thresholds(rms) + if args.silence_threshold is not None: + silence_thr = args.silence_threshold + if args.breath_max is not None: + breath_max = args.breath_max + + print(f" limiares: silêncio < {silence_thr:.4f} | respiro < {breath_max:.4f}") + + breaths = detect_breath_regions( + rms, window_s, + silence_threshold=silence_thr, + breath_max=breath_max, + min_duration_ms=args.min_ms, + max_duration_ms=args.max_ms, + ) + + print(f" {len(breaths)} evento(s) detectado(s) no total\n") + + if args.edl: + segments = load_edl_segments(args.edl) + breaths = filter_in_segments(breaths, segments) + print(f" {len(breaths)} dentro de segmentos do EDL:\n") + label = "beat" + else: + label = None + + if not breaths: + print(" ✅ nenhum respiro detectado nos segmentos.") + return + + for b in breaths: + extra = "" + if label: + extra = f" [{b['segment_beat']:<20} src {b['segment_range']}]" + print(f" {b['start']:7.3f}s – {b['end']:7.3f}s " + f"({b['duration_ms']}ms rms={b['avg_rms']:.4f}){extra}") + + # salva JSON + stem = Path(args.video).stem + out_json = str(Path(args.video).parent / "edit" / f"breaths_{stem}.json") + os.makedirs(os.path.dirname(out_json), exist_ok=True) + with open(out_json, "w") as f: + json.dump(breaths, f, indent=2) + print(f"\n salvo: {out_json}") + + +if __name__ == "__main__": + main() diff --git a/helpers/export_formats.py b/helpers/export_formats.py new file mode 100644 index 00000000..b3a04a17 --- /dev/null +++ b/helpers/export_formats.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +""" +export_formats.py — exporta um vídeo final em múltiplos formatos de uma vez. + +Formatos disponíveis: + reels 1080x1920 9:16 vertical Instagram Reels / TikTok / Shorts + feed 1080x1080 1:1 quadrado Instagram Feed + stories 1080x1920 9:16 vertical Stories (com margens laterais se necessário) + landscape 1920x1080 16:9 horizontal YouTube / LinkedIn + +Uso: + python3 export_formats.py [--formats reels feed stories landscape] [--out-dir ] +""" + +import argparse +import subprocess +import os +import sys +from pathlib import Path + + +FORMATS = { + "reels": { + "w": 1080, "h": 1920, + "desc": "Instagram Reels / TikTok / YouTube Shorts (9:16)", + "crop": "scale=1080:1920:force_original_aspect_ratio=increase,crop=1080:1920", + }, + "feed": { + "w": 1080, "h": 1080, + "desc": "Instagram Feed (1:1)", + "crop": "scale=1080:1080:force_original_aspect_ratio=increase,crop=1080:1080", + }, + "stories": { + "w": 1080, "h": 1920, + "desc": "Stories com margens (9:16 com padding)", + # encaixa o vídeo dentro de 1080x1920 com barras pretas se necessário + "crop": "scale=1080:1920:force_original_aspect_ratio=decrease,pad=1080:1920:(ow-iw)/2:(oh-ih)/2:black", + }, + "landscape": { + "w": 1920, "h": 1080, + "desc": "YouTube / LinkedIn (16:9)", + "crop": "scale=1920:1080:force_original_aspect_ratio=increase,crop=1920:1080", + }, +} + + +def get_video_info(video_path): + result = subprocess.run([ + "ffprobe", "-v", "quiet", "-print_format", "json", + "-show_streams", video_path + ], capture_output=True, text=True) + import json + data = json.loads(result.stdout) + vs = next(s for s in data["streams"] if s["codec_type"] == "video") + return int(vs["width"]), int(vs["height"]) + + +def export_format(video_path, fmt_name, fmt, out_dir, crf=23): + stem = Path(video_path).stem + out_path = os.path.join(out_dir, f"{stem}_{fmt_name}.mp4") + + vf = fmt["crop"] + + cmd = [ + "ffmpeg", "-i", video_path, + "-vf", vf, + "-c:v", "libx264", "-crf", str(crf), "-preset", "fast", + "-c:a", "aac", "-b:a", "192k", + "-movflags", "+faststart", + out_path, "-y", "-loglevel", "error" + ] + + print(f" [{fmt_name}] {fmt['desc']} → {out_path}") + result = subprocess.run(cmd) + if result.returncode != 0: + print(f" ⚠️ erro ao exportar {fmt_name}", file=sys.stderr) + return False + + size_mb = os.path.getsize(out_path) / 1024 / 1024 + print(f" ✓ {size_mb:.1f} MB") + return True + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("video") + parser.add_argument("--formats", nargs="+", + choices=list(FORMATS.keys()), + default=["reels", "feed", "landscape"], + help="Formatos a exportar (padrão: reels feed landscape)") + parser.add_argument("--out-dir", default=None) + parser.add_argument("--crf", type=int, default=23, + help="CRF de qualidade (18=alta, 28=comprimido, padrão 23)") + args = parser.parse_args() + + video_path = args.video + if not os.path.exists(video_path): + sys.exit(f"Arquivo não encontrado: {video_path}") + + out_dir = args.out_dir or str(Path(video_path).parent / "exports") + os.makedirs(out_dir, exist_ok=True) + + w, h = get_video_info(video_path) + print(f" fonte: {w}x{h} → exportando {len(args.formats)} formato(s)...\n") + + ok = 0 + for fmt_name in args.formats: + if export_format(video_path, fmt_name, FORMATS[fmt_name], out_dir, args.crf): + ok += 1 + + print(f"\ndone: {ok}/{len(args.formats)} formatos em {out_dir}/") + + +if __name__ == "__main__": + main() diff --git a/helpers/reframe.py b/helpers/reframe.py new file mode 100644 index 00000000..d1cb754a --- /dev/null +++ b/helpers/reframe.py @@ -0,0 +1,198 @@ +#!/usr/bin/env python3 +""" +reframe.py — recorta vídeo horizontal (16:9) para vertical (9:16) +seguindo o rosto do speaker via rastreamento de face com OpenCV. + +Estratégia: + 1. Amostra frames a cada 0.5s para detectar posição X do rosto + 2. Suaviza a trajetória horizontal (moving average) para evitar jumps + 3. Gera um filtro ffmpeg crop+scale dinâmico via arquivo de expressões + +Uso: + python3 reframe.py [-o output.mp4] [--target reels|stories] +""" + +import argparse +import subprocess +import tempfile +import os +import sys +from pathlib import Path + +import cv2 +import numpy as np + + +TARGETS = { + "reels": {"w": 1080, "h": 1920}, + "stories": {"w": 1080, "h": 1920}, + "square": {"w": 1080, "h": 1080}, +} + + +def get_video_info(video_path): + result = subprocess.run([ + "ffprobe", "-v", "quiet", "-print_format", "json", + "-show_streams", "-show_format", video_path + ], capture_output=True, text=True) + import json + data = json.loads(result.stdout) + vs = next(s for s in data["streams"] if s["codec_type"] == "video") + fps_str = vs.get("avg_frame_rate", "30/1") + num, den = map(int, fps_str.split("/")) + fps = num / den if den else 30 + return int(vs["width"]), int(vs["height"]), fps, float(data["format"]["duration"]) + + +def detect_face_positions(video_path, sample_interval=0.5): + """ + Retorna lista de (timestamp, center_x_normalized) onde center_x é 0..1 + relativo à largura do vídeo. Se nenhum rosto: retorna None para esse frame. + """ + cap = cv2.VideoCapture(video_path) + if not cap.isOpened(): + sys.exit(f"Não consegui abrir {video_path}") + + fps = cap.get(cv2.CAP_PROP_FPS) or 30 + total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) + w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + + cv_data = cv2.data.haarcascades + face_cascade = cv2.CascadeClassifier(cv_data + "haarcascade_frontalface_default.xml") + + positions = [] + frame_interval = max(1, int(fps * sample_interval)) + frame_idx = 0 + + while True: + ret, frame = cap.read() + if not ret: + break + + if frame_idx % frame_interval == 0: + ts = frame_idx / fps + gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) + faces = face_cascade.detectMultiScale(gray, 1.1, 5, minSize=(60, 60)) + + if len(faces) > 0: + fx, fy, fw, fh = max(faces, key=lambda r: r[2]*r[3]) + cx = (fx + fw / 2) / w # normalizado 0..1 + else: + cx = None + + positions.append((ts, cx)) + + frame_idx += 1 + + cap.release() + return positions, w, h + + +def smooth_positions(positions, window=10): + """ + Suaviza os valores de cx com moving average. + Preenche None com o último valor conhecido. + """ + # preenche nulos com interpolação + xs = [] + last = 0.5 # default: centro + for ts, cx in positions: + if cx is not None: + last = cx + xs.append(last) + + # moving average + kernel = np.ones(window) / window + smoothed = np.convolve(xs, kernel, mode="same") + return [(positions[i][0], float(smoothed[i])) for i in range(len(positions))] + + +def build_crop_filter(smooth_pos, src_w, src_h, target_w, target_h): + """ + Gera filtro ffmpeg com crop dinâmico baseado nos keyframes de posição. + Para simplificar, usa a mediana da posição (crop estático suavizado). + Para tracking real frame-a-frame seria necessário usar vf=sendcmd. + """ + # aspect ratio do crop necessário no source + crop_h = src_h + crop_w = int(crop_h * target_w / target_h) + crop_w = min(crop_w, src_w) + + if crop_w >= src_w: + # já é mais estreito que necessário (ex: source quadrado) + crop_w = src_w + crop_h = int(src_w * target_h / target_w) + + # posição X do centro de interesse (mediana suavizada) + median_cx = float(np.median([cx for _, cx in smooth_pos])) + center_x = int(median_cx * src_w) + + # calcula x do crop garantindo que não sai do frame + x = center_x - crop_w // 2 + x = max(0, min(x, src_w - crop_w)) + y = 0 # crop vertical: topo + + vf = ( + f"crop={crop_w}:{crop_h}:{x}:{y}," + f"scale={target_w}:{target_h}" + ) + return vf, x, center_x + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("video") + parser.add_argument("-o", "--output", default=None) + parser.add_argument("--target", choices=list(TARGETS.keys()), default="reels") + parser.add_argument("--sample-interval", type=float, default=0.5) + parser.add_argument("--smooth-window", type=int, default=15, + help="Janela de suavização em frames amostrados (padrão 15)") + args = parser.parse_args() + + video_path = args.video + if not os.path.exists(video_path): + sys.exit(f"Arquivo não encontrado: {video_path}") + + target = TARGETS[args.target] + stem = Path(video_path).stem + out_path = args.output or str(Path(video_path).parent / f"{stem}_{args.target}.mp4") + + print(f" detectando rostos a cada {args.sample_interval}s...") + positions, src_w, src_h = detect_face_positions(video_path, args.sample_interval) + + detected = sum(1 for _, cx in positions if cx is not None) + print(f" {detected}/{len(positions)} frames com rosto detectado") + + if detected == 0: + print(" ⚠️ nenhum rosto detectado — usando crop central") + smooth_pos = [(ts, 0.5) for ts, _ in positions] + else: + smooth_pos = smooth_positions(positions, args.smooth_window) + + vf, crop_x, center_x = build_crop_filter( + smooth_pos, src_w, src_h, + target["w"], target["h"] + ) + + print(f" crop: x={crop_x} center_x={center_x}/{src_w}") + print(f" filtro: {vf}") + print(f" → {out_path}") + + cmd = [ + "ffmpeg", "-i", video_path, + "-vf", vf, + "-c:v", "libx264", "-crf", "20", "-preset", "fast", + "-c:a", "copy", + out_path, "-y", "-loglevel", "error" + ] + result = subprocess.run(cmd) + if result.returncode == 0: + size_mb = os.path.getsize(out_path) / 1024 / 1024 + print(f"\ndone: {out_path} ({size_mb:.1f} MB)") + else: + sys.exit("Erro no reframe.") + + +if __name__ == "__main__": + main() diff --git a/helpers/thumbnail.py b/helpers/thumbnail.py new file mode 100644 index 00000000..a61d08f0 --- /dev/null +++ b/helpers/thumbnail.py @@ -0,0 +1,206 @@ +#!/usr/bin/env python3 +""" +thumbnail.py — extrai os melhores frames de um vídeo como sugestões de thumbnail. + +Pontuação por frame: + - rosto detectado e centralizado (+++) + - olhos abertos (+++) + - boca fechada (++) + - nitidez / sem blur (+) + - rosto grande o suficiente (+) + +Uso: + python3 thumbnail.py [--out-dir ] [--top N] [--interval 0.5] +""" + +import argparse +import subprocess +import tempfile +import os +import sys +from pathlib import Path + +import cv2 +import numpy as np +from PIL import Image, ImageDraw, ImageFont + + +def extract_frames(video_path, interval_s=0.5, out_dir=None): + """Extrai frames em intervalos regulares via ffmpeg.""" + result = subprocess.run( + ["ffprobe", "-v", "quiet", "-show_entries", "format=duration", + "-of", "csv=p=0", video_path], + capture_output=True, text=True + ) + duration = float(result.stdout.strip() or 0) + if duration == 0: + sys.exit(f"Não consegui ler a duração de {video_path}") + + frames = [] + t = 0.0 + while t < duration: + out_path = os.path.join(out_dir, f"frame_{t:.3f}.jpg") + subprocess.run([ + "ffmpeg", "-ss", str(t), "-i", video_path, + "-vframes", "1", "-q:v", "2", + "-vf", "scale=iw*min(1080/iw\\,1920/ih):ih*min(1080/iw\\,1920/ih)", + out_path, "-y", "-loglevel", "error" + ], check=False) + if os.path.exists(out_path): + frames.append((t, out_path)) + t = round(t + interval_s, 3) + return frames + + +def sharpness(gray): + """Laplacian variance — maior = mais nítido.""" + return cv2.Laplacian(gray, cv2.CV_64F).var() + + +def score_frame(img_path, face_cascade, eye_cascade, mouth_cascade): + img = cv2.imread(img_path) + if img is None: + return 0, None + gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) + h, w = gray.shape + + faces = face_cascade.detectMultiScale(gray, 1.1, 5, minSize=(80, 80)) + if len(faces) == 0: + return sharpness(gray) * 0.1, None # sem rosto = penaliza + + # pega o maior rosto + x, y, fw, fh = max(faces, key=lambda r: r[2] * r[3]) + face_roi = gray[y:y+fh, x:x+fw] + face_img = img[y:y+fh, x:x+fw] + + score = 0.0 + + # tamanho do rosto relativo ao frame + face_fraction = (fw * fh) / (w * h) + score += face_fraction * 30 + + # centralização horizontal + cx = x + fw / 2 + center_score = 1 - abs(cx / w - 0.5) * 2 # 1.0 = centrado + score += center_score * 20 + + # posição vertical (rosto no terço superior-central) + cy = y + fh / 2 + vert_ideal = 0.35 + vert_score = max(0, 1 - abs(cy / h - vert_ideal) * 3) + score += vert_score * 10 + + # olhos detectados + eyes = eye_cascade.detectMultiScale(face_roi, 1.1, 3, minSize=(20, 20)) + score += len(eyes) * 15 + + # boca: penaliza se detectada grande (boca aberta) + mouths = mouth_cascade.detectMultiScale( + face_roi[int(fh*0.5):], 1.1, 3, minSize=(30, 15)) + if len(mouths) > 0: + mouth_area = max(m[2]*m[3] for m in mouths) / (fw * fh) + score -= mouth_area * 100 # boca aberta = penaliza + + # nitidez + sharp = sharpness(face_roi) + score += min(sharp / 100, 20) # cap em 20 pts + + return score, (x, y, fw, fh) + + +def make_contact_sheet(top_frames, video_name, out_path): + """Gera um PNG com os top frames lado a lado.""" + n = len(top_frames) + if n == 0: + return + thumbs = [] + for ts, img_path, sc, _ in top_frames: + img = Image.open(img_path).convert("RGB") + img.thumbnail((320, 320)) + thumbs.append((ts, img, sc)) + + cols = min(n, 5) + rows = (n + cols - 1) // cols + tw, th = thumbs[0][1].size + pad = 10 + header = 50 + sheet_w = cols * tw + (cols + 1) * pad + sheet_h = rows * (th + 30) + (rows + 1) * pad + header + + sheet = Image.new("RGB", (sheet_w, sheet_h), (20, 20, 20)) + draw = ImageDraw.Draw(sheet) + try: + font = ImageFont.truetype("/System/Library/Fonts/Helvetica.ttc", 14) + font_title = ImageFont.truetype("/System/Library/Fonts/Helvetica.ttc", 18) + except Exception: + font = font_title = ImageFont.load_default() + + draw.text((pad, 12), f"Thumbnails sugeridos — {video_name}", fill=(255,255,255), font=font_title) + + for i, (ts, img, sc) in enumerate(thumbs): + col = i % cols + row = i // cols + x = pad + col * (tw + pad) + y = header + pad + row * (th + 30 + pad) + sheet.paste(img, (x, y)) + label = f"#{i+1} {ts:.1f}s (score {sc:.0f})" + draw.text((x, y + th + 4), label, fill=(180, 180, 180), font=font) + + sheet.save(out_path) + print(f" contact sheet → {out_path}") + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("video") + parser.add_argument("--out-dir", default=None) + parser.add_argument("--top", type=int, default=5) + parser.add_argument("--interval", type=float, default=0.5, + help="Intervalo entre frames em segundos (padrão 0.5)") + args = parser.parse_args() + + video_path = args.video + video_name = Path(video_path).stem + + out_dir = args.out_dir or str(Path(video_path).parent / "edit" / "thumbnails") + os.makedirs(out_dir, exist_ok=True) + + cv_data = cv2.data.haarcascades + face_cascade = cv2.CascadeClassifier(cv_data + "haarcascade_frontalface_default.xml") + eye_cascade = cv2.CascadeClassifier(cv_data + "haarcascade_eye.xml") + mouth_cascade = cv2.CascadeClassifier(cv_data + "haarcascade_smile.xml") + + print(f" extraindo frames a cada {args.interval}s...") + with tempfile.TemporaryDirectory() as tmp: + frames = extract_frames(video_path, args.interval, tmp) + print(f" {len(frames)} frames — pontuando...") + + scored = [] + for ts, img_path in frames: + sc, face_box = score_frame(img_path, face_cascade, eye_cascade, mouth_cascade) + scored.append((ts, img_path, sc, face_box)) + + scored.sort(key=lambda x: -x[2]) + top = scored[:args.top] + + # salva os top frames + saved = [] + for rank, (ts, img_path, sc, box) in enumerate(top, 1): + dst = os.path.join(out_dir, f"thumb_{rank:02d}_{ts:.1f}s.jpg") + img = cv2.imread(img_path) + if img is not None: + cv2.imwrite(dst, img, [cv2.IMWRITE_JPEG_QUALITY, 95]) + saved.append((ts, dst, sc, box)) + print(f" #{rank} t={ts:.1f}s score={sc:.1f} → {dst}") + + make_contact_sheet( + [(ts, p, sc, b) for ts, p, sc, b in saved], + video_name, + os.path.join(out_dir, f"{video_name}_thumbnails.png") + ) + + print(f"\ndone: {args.top} thumbnails em {out_dir}/") + + +if __name__ == "__main__": + main() From 7219cdad1947809e285d4c4d44c63c7019693b3b Mon Sep 17 00:00:00 2001 From: eaikarensantos-gif Date: Fri, 12 Jun 2026 00:12:51 -0300 Subject: [PATCH 3/4] Add news_server with cached RSS, shared client, and claude-opus-4-8 - Reuse one Anthropic client instead of per-request instantiation - Cache TechCrunch headlines for 5 minutes - Update model to claude-opus-4-8 - Disable Flask debug mode by default (opt in via FLASK_DEBUG=1) Co-Authored-By: Claude Fable 5 --- news_server.py | 133 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 133 insertions(+) create mode 100644 news_server.py diff --git a/news_server.py b/news_server.py new file mode 100644 index 00000000..4f0f5a7c --- /dev/null +++ b/news_server.py @@ -0,0 +1,133 @@ +""" +news_server.py — Servidor Flask para geração de conteúdo a partir de manchetes +Uso: python3 news_server.py +Acesse: http://localhost:5050 +""" + +import os, json, time +from flask import Flask, jsonify, request, send_from_directory +import feedparser +import anthropic + +app = Flask(__name__, static_folder="static") + +TECHCRUNCH_RSS = "https://techcrunch.com/feed/" +HEADLINES_CACHE_TTL = 300 # seconds +_headlines_cache: dict = {"at": 0.0, "items": []} + +# Client is created once; resolves ANTHROPIC_API_KEY from the environment. +_client = anthropic.Anthropic() if os.environ.get("ANTHROPIC_API_KEY") else None + +PROMPTS = { + "linkedin": """Você é um criador de conteúdo especialista em LinkedIn para o mercado tech brasileiro. +Com base na manchete abaixo, escreva um post profissional para LinkedIn. +Regras: português brasileiro, tom reflexivo e profissional, 150-250 palavras, termine com 2-3 hashtags relevantes, sem emojis excessivos. +Manchete: {headline} +Resumo: {summary}""", + + "instagram": """Você é um criador de conteúdo para Instagram focado em tecnologia. +Com base na manchete abaixo, escreva uma legenda para Instagram. +Regras: português brasileiro, engajante e direto, 80-120 palavras, emojis estratégicos, 5-8 hashtags no final. +Manchete: {headline} +Resumo: {summary}""", + + "tiktok": """Você é um roteirista de TikTok sobre tecnologia para público brasileiro jovem. +Com base na manchete abaixo, escreva uma legenda + gancho para vídeo TikTok. +Regras: português brasileiro coloquial, gancho forte na primeira linha, máximo 80 palavras, call to action no final. +Manchete: {headline} +Resumo: {summary}""", + + "twitter": """Você é um criador de threads no X (Twitter) sobre tecnologia. +Com base na manchete abaixo, escreva uma thread em português brasileiro. +Regras: 5 tweets numerados (1/5, 2/5...), máximo 280 caracteres cada, primeiro tweet com gancho forte, último com opinião/conclusão. +Manchete: {headline} +Resumo: {summary}""", + + "stories": """Você é um criador de conteúdo para Stories do Instagram sobre tecnologia. +Com base na manchete abaixo, escreva um roteiro de 3-5 stories sequenciais. +Regras: português brasileiro, cada story marcado [Story 1], [Story 2] etc, texto curto por story (max 2 linhas), inclua sugestão de fundo/visual entre parênteses. +Manchete: {headline} +Resumo: {summary}""", + + "roteiro": """Você é um roteirista de vídeos curtos (Reels/TikTok/Shorts) sobre tecnologia. +Com base na manchete abaixo, escreva um roteiro narrado de 60 segundos. +Regras: português brasileiro, marcações [GANCHO], [DESENVOLVIMENTO], [CONCLUSÃO], inclua sugestões visuais entre colchetes, ritmo dinâmico. +Manchete: {headline} +Resumo: {summary}""", +} + +FORMAT_LABELS = { + "linkedin": "LinkedIn", + "instagram": "Instagram", + "tiktok": "TikTok", + "twitter": "Thread X/Twitter", + "stories": "Stories", + "roteiro": "Roteiro de Vídeo", +} + + +@app.route("/") +def index(): + return send_from_directory(".", "news.html") + + +@app.route("/api/headlines") +def headlines(): + try: + now = time.time() + if _headlines_cache["items"] and now - _headlines_cache["at"] < HEADLINES_CACHE_TTL: + return jsonify({"ok": True, "items": _headlines_cache["items"], "cached": True}) + feed = feedparser.parse(TECHCRUNCH_RSS) + items = [] + for entry in feed.entries[:20]: + items.append({ + "title": entry.get("title", ""), + "summary": entry.get("summary", "")[:400], + "link": entry.get("link", ""), + "published": entry.get("published", ""), + "image": next( + (m.get("url") for m in entry.get("media_content", []) if m.get("url")), + None + ), + }) + _headlines_cache["at"] = now + _headlines_cache["items"] = items + return jsonify({"ok": True, "items": items}) + except Exception as e: + return jsonify({"ok": False, "error": str(e)}), 500 + + +@app.route("/api/generate", methods=["POST"]) +def generate(): + data = request.json or {} + headline = data.get("headline", "").strip() + summary = data.get("summary", "").strip() + fmt = data.get("format", "linkedin") + + if not headline: + return jsonify({"ok": False, "error": "headline obrigatória"}), 400 + + if fmt not in PROMPTS: + return jsonify({"ok": False, "error": f"formato inválido: {fmt}"}), 400 + + if _client is None: + return jsonify({"ok": False, "error": "ANTHROPIC_API_KEY não configurada"}), 500 + + try: + prompt = PROMPTS[fmt].format(headline=headline, summary=summary) + + message = _client.messages.create( + model="claude-opus-4-8", + max_tokens=1024, + messages=[{"role": "user", "content": prompt}], + ) + content = message.content[0].text + return jsonify({"ok": True, "content": content, "format": FORMAT_LABELS.get(fmt, fmt)}) + except Exception as e: + return jsonify({"ok": False, "error": str(e)}), 500 + + +if __name__ == "__main__": + debug = os.environ.get("FLASK_DEBUG", "") == "1" + print("🚀 Servidor rodando em http://localhost:5050") + app.run(port=5050, debug=debug) From f44ac89dd45edc49a199bd2ca7935277bb877441 Mon Sep 17 00:00:00 2001 From: eaikarensantos-gif Date: Fri, 12 Jun 2026 00:12:51 -0300 Subject: [PATCH 4/4] Ignore media files and working folders in repo root Co-Authored-By: Claude Fable 5 --- .gitignore | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/.gitignore b/.gitignore index ea1791ad..80763f6a 100644 --- a/.gitignore +++ b/.gitignore @@ -58,3 +58,27 @@ master.srt takes_packed.md edl.json project.md + +# Raw footage, renders, music, SFX — never commit large media binaries +*.mp4 +*.mov +*.MOV +*.mkv +*.avi +*.wav +*.mp3 +*.m4a +*.aac +*.flac + +# Media/working folders dropped into the repo root +background/ +musica/ +luts/ +presets/ +WHOOSH/ +SFX Reels/ +CONTABILIZEI/ +drive-download-*/ +#03 Sound Effects*/ +\#03 Sound Effects*/