Mass converter for Telegram video sticker

process.py

import logging

from pathlib import Path
from subprocess import run

input_dir = Path("input/")
output_dir = Path("output/")
max_duration = 2.95
max_size = 1024 * 256

logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", 
    level=logging.INFO
)

def get_duration(path: Path) -> float:
    result = run(
        [
            "ffprobe", 
            "-v", "error",
            "-show_entries", "format=duration",
            "-of", "default=noprint_wrappers=1:nokey=1",
            str(path),
        ],
        capture_output=True,
        universal_newlines=True,
    )

    return float(0) if result.stdout == "N/A\n" else float(result.stdout)


def convert(path: Path, out_path: Path):
    duration = get_duration(path)
    logging.info(f"duration = {duration}")
   
    prefix = []
    suffix = []
    filters = [
        "scale=512:512",
    ]

    if duration == 0:
        prefix += ["-loop", "1"]
        suffix += ["-frames:v", "1"]
    else:
        # 최대 길이를 초과하는 동영상은 최대 길이에 맞게끔 빠른 재생하기
        pts = max_duration / duration
        if pts < 1:
            logging.info(f"PTS *= ({pts})")
            filters.append(f"setpts=PTS*{pts}")

    run([
        "ffmpeg",
        "-hide_banner",
        "-loglevel", "error",
        *prefix,
        "-i", str(path),
        "-c:v", "libvpx-vp9",
        "-row-mt", "1",
        "-vf", ','.join(filters),
        *suffix,
        str(out_path)
    ])


def compress(path: Path, out_path: Path):
    run([
        "ffmpeg",
        "-hide_banner",
        "-loglevel", "error",
        "-c:v", "libvpx-vp9",
        "-i", str(path),
        "-c:v", "libvpx-vp9",
        "-b:v", "0",
        "-crf", "30",
        "-pass", "1",
        "-deadline", "best",
        "-row-mt", "1",
        "-f", "null",
        "/dev/null"
    ])

    run([
        "ffmpeg",
        "-hide_banner",
        "-loglevel", "error",
        "-c:v", "libvpx-vp9",
        "-i", str(path),
        "-c:v", "libvpx-vp9",
        "-b:v", "0",
        "-crf", "30",
        "-pass", "2",
        "-deadline", "best",
        "-row-mt", "1",
        str(out_path)
    ])

for path in input_dir.glob("*"):
    out_path = output_dir.joinpath(path.stem + ".webm")
    logging.info(f"checking {path.name}")
    
    if out_path.is_file():
        logging.info("already converted, skip the ffmpeg process")
    else:
        convert(path, out_path)

    # 출력 파일의 크기가 최대 파일 크기보다 크다면 2-pass 압축하기
    compressed_path = output_dir.joinpath(f"{out_path.stem}_passed{out_path.suffix}")
    out_stat = out_path.stat()
    if out_stat.st_size > max_size:
        if compressed_path.is_file():
            logging.info("already compressed, skip the 2-pass process")
        else:
            logging.info(f"filesize is too large, attempt to 2-pass ({out_stat.st_size / 1024}kb)")
            compress(out_path, compressed_path)
            compressed_stat = compressed_path.stat()
            logging.info(f"{out_stat.st_size / 1024}kb -> {compressed_stat.st_size / 1024}kb")