Python Media Optimizer & Auto-Transcriber (AV1 + EBU R128 + faster-whisper)

optimize_and_archive_videos.py

#!/usr/bin/env python3
"""
Pipeline de Otimização, Deduplicação e Conversão em Lote para AV1 + EBU R128 Loudnorm.
Compatível com Linux, Windows e macOS.
"""

import argparse
import hashlib
import os
import subprocess
import sys
from pathlib import Path

VIDEO_EXTENSIONS = {".mp4", ".mkv", ".avi", ".mov", ".wmv", ".webm", ".flv", ".ts"}


def calculate_md5(filepath: Path, chunk_size: int = 8192) -> str:
    hasher = hashlib.md5()
    with open(filepath, "rb") as f:
        while chunk := f.read(chunk_size):
            hasher.update(chunk)
    return hasher.hexdigest()


def remove_duplicates(directory: Path):
    print(f"\n[Deduplicação] Varrendo pasta: {directory}...")
    seen_hashes = {}
    duplicates_removed = 0
    bytes_saved = 0

    for item in sorted(directory.rglob("*")):
        if item.is_file() and item.suffix.lower() in VIDEO_EXTENSIONS:
            file_hash = calculate_md5(item)
            if file_hash in seen_hashes:
                original = seen_hashes[file_hash]
                size = item.stat().st_size
                print(f"  [Duplicado] Removendo: {item.name}\n              (Cópia de: {original.name})")
                try:
                    item.unlink()
                    duplicates_removed += 1
                    bytes_saved += size
                except Exception as e:
                    print(f"  [Erro ao remover]: {e}")
            else:
                seen_hashes[file_hash] = item

    print(f"[Deduplicação] Concluída. {duplicates_removed} duplicatas removidas ({bytes_saved / (1024*1024):.1f} MB liberados).")


def convert_video(input_path: Path, crf: int = 34, preset: int = 6, transcribe: bool = True):
    input_path = Path(input_path).resolve()
    if not input_path.exists():
        return

    output_path = input_path.with_suffix(".mkv")
    is_same_file = (input_path == output_path)

    if is_same_file:
        target_output = input_path.with_name(f"{input_path.stem}_av1.mkv")
    else:
        target_output = output_path

    orig_size = input_path.stat().st_size

    print(f"\n[Conversão AV1] Processando: {input_path.name} ({orig_size / (1024*1024):.1f} MB)...")

    cmd = [
        "ffmpeg",
        "-y",
        "-i", str(input_path),
        "-af", "loudnorm=I=-16:TP=-1.5:LRA=11",
        "-c:v", "libsvtav1",
        "-crf", str(crf),
        "-preset", str(preset),
        "-c:a", "libopus",
        "-b:a", "48k",
        str(target_output),
    ]

    try:
        proc = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
        if proc.returncode != 0:
            print(f"  [Erro na Conversão]: {proc.stderr[-500:]}")
            if target_output.exists():
                target_output.unlink()
            return

        new_size = target_output.stat().st_size
        saved_mb = (orig_size - new_size) / (1024 * 1024)
        pct = (1 - (new_size / orig_size)) * 100

        print(f"  [Sucesso] Novo tamanho: {new_size / (1024*1024):.1f} MB ({pct:.1f}% menor, {saved_mb:.1f} MB economizados)")

        # Remove original após validação
        if is_same_file:
            input_path.unlink()
            target_output.rename(input_path)
            final_file = input_path
        else:
            input_path.unlink()
            final_file = target_output

        # Chama transcrição se solicitado
        if transcribe:
            script_dir = Path(__file__).resolve().parent
            transcribe_script = script_dir / "transcribe_and_subtitle.py"
            if transcribe_script.exists():
                subprocess.run([sys.executable, str(transcribe_script), str(final_file)])

    except Exception as e:
        print(f"  [Exceção]: {e}")


def process_directory(directory: Path, crf: int = 34, preset: int = 6, transcribe: bool = True):
    directory = Path(directory).resolve()
    remove_duplicates(directory)

    videos = [p for p in sorted(directory.rglob("*")) if p.is_file() and p.suffix.lower() in VIDEO_EXTENSIONS]
    print(f"\n[Fila] {len(videos)} vídeos encontrados para conversão.")

    for vid in videos:
        # Pula arquivos que já são o resultado de conversões AV1
        if "_30s_" in vid.name:
            continue
        convert_video(vid, crf=crf, preset=preset, transcribe=transcribe)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Otimiza, deduplica e converte vídeos em lote para AV1 com áudio normalizado.")
    parser.add_argument("--dir", help="Diretório para processar recursivamente")
    parser.add_argument("--file", help="Arquivo único para converter")
    parser.add_argument("--crf", type=int, default=34, help="Fator CRF para AV1 (padrão: 34)")
    parser.add_argument("--preset", type=int, default=6, help="Preset SVT-AV1 (padrão: 6)")
    parser.add_argument("--no-transcribe", action="store_true", help="Não gerar transcrição e legendas")

    args = parser.parse_args()
    if args.file:
        convert_video(Path(args.file), crf=args.crf, preset=args.preset, transcribe=not args.no_transcribe)
    elif args.dir:
        process_directory(Path(args.dir), crf=args.crf, preset=args.preset, transcribe=not args.no_transcribe)
    else:
        parser.print_help()

organize_and_name_meetings.py

#!/usr/bin/env python3
"""
Organizador e Renomeador Inteligente de Reuniões e Vídeos.
Lê os arquivos de transcrição (.txt), extrai a data e os temas técnicos principais,
cria pastas modulares individuais e renomeia o trio (.mkv, .srt, .txt) automaticamente.
"""

import argparse
import os
import re
import shutil
import sys
from pathlib import Path


def extract_keywords_from_transcript(txt_path: Path) -> str:
    """Gera o título semântico da gravação utilizando Inteligência Artificial Generativa baseada na transcrição."""
    if not txt_path.exists() or txt_path.stat().st_size == 0:
        return "Gravacao de Midia"

    with open(txt_path, "r", encoding="utf-8", errors="ignore") as f:
        content = f.read(4000)

    api_key = os.environ.get("GEMINI_API_KEY")
    if api_key:
        try:
            import json, urllib.request
            url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash:generateContent?key={api_key}"
            prompt = f"Gere um título descritivo, profissional e conciso de 3 a 5 palavras em Português para uma reunião técnica com base nesta transcrição:\n\n{content}\n\nRetorne EXCLUSIVAMENTE o título limpo, sem pontuação final ou aspas."
            data = json.dumps({"contents": [{"parts": [{"text": prompt}]}]}).encode("utf-8")
            req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"})
            with urllib.request.urlopen(req, timeout=5) as resp:
                res_data = json.loads(resp.read().decode())
                title = res_data["candidates"][0]["content"]["parts"][0]["text"].strip()
                clean_title = re.sub(r'[\\/*?:"<>|]', "", title)
                if clean_title:
                    return clean_title
        except Exception:
            pass

    # Processamento Semântico NLP Local (Fallback Inteligente Dinâmico)
    words = re.findall(r"\b[a-zA-ZáéíóúãõçÁÉÍÓÚÃÕÇ]{4,}\b", content)
    freq = {}
    stopwords = {"para", "como", "esta", "está", "isso", "esse", "essa", "fazer", "vamos", "pode", "onde", "quando", "sobre", "muito", "gente", "entao", "então", "aqui", "mais", "tinha", "outro", "outra"}
    for w in words:
        if w.lower() not in stopwords:
            freq[w.capitalize()] = freq.get(w.capitalize(), 0) + 1
    top = sorted(freq.items(), key=lambda x: x[1], reverse=True)[:3]
    return " - ".join([t[0] for t in top]) if top else "Alinhamento Tecnico"


def extract_date_from_filename(filename: str) -> str:
    match = re.search(r"(202\d{5})", filename)
    if match:
        raw_date = match.group(1)
        return f"{raw_date[:4]}-{raw_date[4:6]}-{raw_date[6:8]}"
    return "2026-08-20"


def organize_meeting_folder(directory: Path):
    directory = Path(directory).resolve()
    print(f"\n[Organização] Varrendo pasta: {directory}...")

    # Localiza arquivos .mkv que estejam soltos na raiz da pasta
    mkv_files = [f for f in directory.glob("*.mkv") if f.is_file()]

    if not mkv_files:
        print("  ✓ Nenhum arquivo de vídeo solto necessitando organização.")
        return

    for mkv in mkv_files:
        base_name = mkv.stem
        txt_file = mkv.with_suffix(".txt")
        srt_file = mkv.with_suffix(".srt")

        date_str = extract_date_from_filename(mkv.name)
        topic_str = extract_keywords_from_transcript(txt_file)

        # Se o nome original já for descritivo (ex: Se Vira nos 30)
        if "se vira nos 30" in mkv.name.lower():
            descriptive_name = f"{date_str} - Se Vira nos 30 - {topic_str}"
        elif "teams" in mkv.name.lower():
            descriptive_name = f"{date_str} - Microsoft Teams - {topic_str}"
        else:
            descriptive_name = f"{date_str} - {topic_str}"

        # Cria pasta individual dedicada
        target_folder = directory / descriptive_name
        target_folder.mkdir(parents=True, exist_ok=True)

        print(f"  -> Criando pasta modular: {descriptive_name}")

        # Move e renomeia o trio de arquivos
        for ext in [".mkv", ".srt", ".txt"]:
            src = mkv.with_suffix(ext)
            if src.exists():
                dst = target_folder / f"{descriptive_name}{ext}"
                shutil.move(str(src), str(dst))
                print(f"     [Movido]: {src.name} -> {dst.name}")

    print("\n🎉 Organização e renomeação concluídas com sucesso!")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Organiza reuniões em pastas modulares e renomeia com base nas legendas.")
    parser.add_argument("directory", help="Diretório contendo vídeos e legendas")
    args = parser.parse_args()

    organize_meeting_folder(Path(args.directory))

transcribe_and_subtitle.py

#!/usr/bin/env python3
"""
Transcrição de Áudio e Geração de Legendas com faster-whisper e FFmpeg.
Compatível com Linux, Windows e macOS.
Gera arquivos .txt (para IA), .srt (legendas) e embute a legenda no container .mkv.
"""

import argparse
import os
import subprocess
import sys
from pathlib import Path

try:
    from faster_whisper import WhisperModel
except ImportError:
    print("Erro: faster-whisper não encontrado. Instale com: pip install faster-whisper")
    sys.exit(1)


def format_timestamp(seconds: float) -> str:
    h = int(seconds // 3600)
    m = int((seconds % 3600) // 60)
    s = int(seconds % 60)
    ms = int((seconds - int(seconds)) * 1000)
    return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"


def transcribe_file(
    media_path: Path,
    model_size: str = "small",
    language: str = "pt",
    embed_subtitle: bool = True,
    device: str = "cpu",
    compute_type: str = "int8",
):
    media_path = Path(media_path).resolve()
    if not media_path.exists():
        print(f"Erro: Arquivo não encontrado: {media_path}")
        return False

    base_path = media_path.with_suffix("")
    txt_file = base_path.with_suffix(".txt")
    srt_file = base_path.with_suffix(".srt")

    print(f"\n[Whisper] Carregando modelo '{model_size}' no dispositivo '{device}'...")
    model = WhisperModel(model_size, device=device, compute_type=compute_type)

    print(f"[Whisper] Transcrevendo: {media_path.name} (Idioma: {language})...")
    segments, info = model.transcribe(str(media_path), language=language, beam_size=5)

    print(f"[Whisper] Idioma confirmado: {info.language} (Probabilidade: {info.language_probability:.2%})")

    count = 0
    with open(txt_file, "w", encoding="utf-8") as f_txt, open(srt_file, "w", encoding="utf-8") as f_srt:
        for idx, seg in enumerate(segments, start=1):
            text = seg.text.strip()
            f_txt.write(f"{text}\n")
            f_srt.write(f"{idx}\n{format_timestamp(seg.start)} --> {format_timestamp(seg.end)}\n{text}\n\n")
            count += 1
            if count % 15 == 0:
                print(f"  [{format_timestamp(seg.start)}] {text}")

    print(f"[Whisper] Concluído! {count} segmentos transcritos.")
    print(f"  -> Texto IA: {txt_file.name}")
    print(f"  -> Legenda:  {srt_file.name}")

    if embed_subtitle and media_path.suffix.lower() == ".mkv":
        temp_output = media_path.with_name(f"{media_path.stem}_subtitled.mkv")
        cmd = [
            "ffmpeg",
            "-y",
            "-i",
            str(media_path),
            "-i",
            str(srt_file),
            "-c",
            "copy",
            "-c:s",
            "srt",
            "-metadata:s:s:0",
            f"language={language}",
            "-metadata:s:s:0",
            'title=Português (Brasil)',
            str(temp_output),
        ]
        try:
            res = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            if res.returncode == 0 and temp_output.exists() and temp_output.stat().st_size > 0:
                temp_output.replace(media_path)
                print(f"[FFmpeg] Legenda embutida com sucesso em: {media_path.name}")
        except Exception as e:
            print(f"[Aviso] Não foi possível embutir legenda via FFmpeg: {e}")

    return True


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Transcreve áudio/vídeo e gera legendas em Português.")
    parser.add_argument("file", help="Caminho do arquivo de mídia (.mkv, .mp4, .wav, .mp3, etc.)")
    parser.add_argument("--model", default="small", help="Modelo Whisper (tiny, base, small, medium, large-v3)")
    parser.add_argument("--lang", default="pt", help="Código do idioma (padrão: pt)")
    parser.add_argument("--no-embed", action="store_true", help="Não embutir legenda no container MKV")
    parser.add_argument("--device", default="cpu", help="Dispositivo (cpu ou cuda)")

    args = parser.parse_args()
    transcribe_file(
        Path(args.file),
        model_size=args.model,
        language=args.lang,
        embed_subtitle=not args.no_embed,
        device=args.device,
    )