catbox webm maker

webm.py

import subprocess
import argparse
import requests
import tempfile
import decimal
import pathlib
import shutil
import urllib.parse

scale_factor = 100
crop_window = {"width": (0, 100), "height": (0, 100)}


def make_filter(name, *args):
    return f"{name}=" + ":".join(map(str, args))


video_filters = ", ".join(
    [
        make_filter(
            "crop",
            f"in_w*({crop_window['width'][1] - crop_window['width'][0]}/100)",
            f"in_h*({crop_window['height'][1] - crop_window['height'][0]}/100)",
            f"in_w*({crop_window['width'][0]}/100)",
            f"in_h*({crop_window['height'][0]}/100)",
        ),
        make_filter(
            "scale",
            f"in_w*({scale_factor}/100)",
            f"in_h*({scale_factor}/100)",
        ),
    ]
)

extra_video_args = ["-filter:v", video_filters]

MAX_FILESIZE = 1024 * 1024 * 4 * 95 // 100  # 4MB with 95% fudge factor


def upload_catbox(fileobj):
    response = requests.post(
        "https://catbox.moe/user/api.php",
        {
            "reqtype": "fileupload",
        },
        files={
            "fileToUpload": fileobj,
        },
    )
    response.raise_for_status()
    return response.text


def render_preview(input_file, start_timestamp, duration):
    with tempfile.TemporaryDirectory() as tempdir:
        preview_file = pathlib.Path(tempdir) / "preview.mkv"
        subprocess.run(
            [
                "ffmpeg",
                "-ss",
                start_timestamp,
                "-i",
                input_file,
                "-t",
                str(duration),
                *extra_video_args,
                "-preset",
                "ultrafast",
                "-y",
                preview_file,
            ],
            check=True,
        )
        subprocess.run(["ffplay", "-loop", "0", preview_file])


def negotiate_time_range(input_file):
    start_timestamp = "0"
    duration = "1"
    while True:
        start_timestamp = (
            input(f"Start timestamp (currently {start_timestamp!r})? ")
            or start_timestamp
        )
        duration = input(f"Duration (currently {duration!r})? ") or duration
        render_preview(input_file, start_timestamp, duration)
        while True:
            accepted = input(
                f"Okay with start={start_timestamp!r} and duration={duration!r} (N/y)? "
            )
            if accepted.lower() == "y":
                duration_whole, *_ = duration.split(".", 1)
                duration_seconds = decimal.Decimal(
                    sum(
                        int(value) * (60**idx)
                        for idx, value in enumerate(duration_whole.split(":")[::-1])
                    )
                )
                if "." in duration:
                    duration_seconds += decimal.Decimal(
                        "0." + duration.split(".", 1)[1]
                    )
                return start_timestamp, duration_seconds
            elif accepted.lower() == "n":
                break


def upload_audio(input_file, start_timestamp, duration):
    with tempfile.TemporaryDirectory() as tempdir:
        audio_file = pathlib.Path(tempdir) / "audio.mp3"
        subprocess.run(
            [
                "ffmpeg",
                "-ss",
                start_timestamp,
                "-i",
                input_file,
                "-t",
                str(duration),
                "-vn",
                "-c:a",
                "libmp3lame",
                "-q:a",
                "0",
                "-y",
                audio_file,
            ],
            check=True,
        )
        with audio_file.open("rb") as f:
            return upload_catbox(f)


def create_video(input_file, start_timestamp, duration, sound_url):
    with tempfile.TemporaryDirectory() as tempdir:
        pass_log_file = pathlib.Path(tempdir) / "pass.log"
        video_file = pathlib.Path(tempdir) / "output.webm"
        for pass_number in ("1", "2"):
            subprocess.run(
                [
                    "ffmpeg",
                    "-ss",
                    start_timestamp,
                    "-i",
                    input_file,
                    "-t",
                    str(duration),
                    "-an",
                    "-sn",
                    *extra_video_args,
                    "-b:v",
                    str(int(MAX_FILESIZE / duration) * 8),
                    # "-crf", "4",
                    "-threads",
                    "4",
                    "-pass",
                    pass_number,
                    "-passlogfile",
                    pass_log_file,
                    "-y",
                    video_file,
                ],
                check=True,
            )
        video_name = input("Video name? ")
        escaped_sound_url = urllib.parse.quote(
            sound_url.replace("https://", ""), safe=""
        )
        destination_file = (
            input_file.parent / f"{video_name} [sound={escaped_sound_url}].webm"
        )
        shutil.copy(video_file, destination_file)
        return destination_file


def get_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("input_file", type=pathlib.Path)
    return parser.parse_args()


def main():
    args = get_args()

    start_timestamp, duration = negotiate_time_range(args.input_file)
    sound_url = upload_audio(args.input_file, start_timestamp, duration)
    output_file = create_video(args.input_file, start_timestamp, duration, sound_url)


if __name__ == "__main__":
    main()