Skip to content

zimscraperlib.video.encoding

Functions:

  • reencode

    Runs ffmpeg with given ffmpeg_args

reencode

reencode(
    src_path: Path,
    dst_path: Path,
    ffmpeg_args: list[str],
    threads: int | None = 1,
    *,
    delete_src: bool = False,
    failsafe: bool = True,
    existing_tmp_path: Path | None = None,
) -> tuple[bool, CompletedProcess[str]]

Runs ffmpeg with given ffmpeg_args

Arguments - src_path - Path to source file dst_path - Path to destination file ffmpeg_args - A list of ffmpeg arguments threads - Number of encoding threads used by ffmpeg delete_src - Delete source file after convertion failsafe - Run in failsafe mode

Source code in src/zimscraperlib/video/encoding.py
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def reencode(
    src_path: pathlib.Path,
    dst_path: pathlib.Path,
    ffmpeg_args: list[str],
    threads: int | None = 1,
    *,
    delete_src: bool = False,
    failsafe: bool = True,
    existing_tmp_path: pathlib.Path | None = None,
) -> tuple[bool, subprocess.CompletedProcess[str]]:
    """Runs ffmpeg with given ffmpeg_args

    Arguments -
        src_path - Path to source file
        dst_path - Path to destination file
        ffmpeg_args - A list of ffmpeg arguments
        threads - Number of encoding threads used by ffmpeg
        delete_src - Delete source file after convertion
        failsafe - Run in failsafe mode
    """

    with path_from(existing_tmp_path or tempfile.TemporaryDirectory()) as tmp_dir:
        tmp_path = pathlib.Path(tmp_dir).joinpath(f"video.tmp{dst_path.suffix}")
        args = _build_ffmpeg_args(
            src_path=src_path,
            tmp_path=tmp_path,
            ffmpeg_args=ffmpeg_args,
            threads=threads,
        )
        logger.debug(
            f"Encode {src_path} -> {dst_path} video format = {dst_path.suffix}"
        )
        logger.debug(nicer_args_join(args))
        ffmpeg = subprocess.run(
            args,
            stderr=subprocess.STDOUT,
            stdout=subprocess.PIPE,
            text=True,
            check=False,
        )
        if not failsafe:
            ffmpeg.check_returncode()
        if ffmpeg.returncode == 0:
            if delete_src:
                src_path.unlink()
            shutil.copy(tmp_path, dst_path)
        return ffmpeg.returncode == 0, ffmpeg