TAS-Comp-Bot-py/video.py at py · crackhex/TAS-Comp-Bot-py · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
import asyncio
import json
import random
import string
import subprocess


class FFmpegBuilder:
    def __init__(self):
        self._inputs: list[str] = []

        self._vcodec = "libx264"  # support libx264/libx265
        self._vmode = "crf"  # crf/cbr
        self._crf = 23
        self._cbr = 0
        self._vbv_maxrate = 0
        self._vbv_bufsize = 0
        self._vfilter = ""
        self._pix_fmt = "yuv420p"

        self._acodec = "aac"
        self._omit_abr = False
        self._abr = 128000
        self._afilter = ""

        self._maxsize = 0
        self._preparams = []
        self._postparams = []

        self._output = "".join(random.choices(string.ascii_letters + string.digits,
                                              k=12)) + ".mp4"  # default to 12 random alphanumeric digits

    def __str__(self):
        return str(self.build())

    def __repr__(self):
        return str(self.__dict__) + "\n" + str(self)

    def build(self):
        command = ["ffmpeg", "-y"]

        for param in self._preparams:
            command.append(f"-{param}")

        for input in self._inputs:
            command.extend(["-i", input])

        if self._vfilter != "":
            command.extend(["-vf", f"{self._vfilter},format={self._pix_fmt}"])
        else:
            command.extend(["-vf", f"format={self._pix_fmt}"])

        if self._afilter != "":
            command.extend(["-af", self._afilter])

        command.extend(["-c:v", self._vcodec])
        if self._vcodec != "copy" and self._vcodec != "vn":
            if self._vmode == "crf":
                command.extend(["-crf", str(self._crf)])
            elif self._vmode == "cbr":
                command.extend(["-b:v", str(self._cbr)])

            if self._vbv_maxrate != 0 and self._vbv_bufsize != 0:
                command.extend(["-maxrate", str(self._vbv_maxrate), "-bufsize", str(self._vbv_bufsize)])

        if self._acodec == "copy":
            command.extend(["-c:a", "copy"])
        elif self._abr == 0:
            command.extend(["-c:a", "an"])
        else:
            command.extend(["-c:a", self._acodec])
            if not self._omit_abr and self._acodec != "an":
                command.extend(["-b:a", str(self._abr)])

        if self._maxsize != 0:
            command.extend(["-fs", str(self._maxsize)])

        for param in self._postparams:
            command.append("-{param}")

        command.append(self._output)

        return command

    def run(self):
        return subprocess.run(self.build())

    async def run_async(self):
        proc = await asyncio.create_subprocess_exec(*self.build())
        return proc

    def input(self, *inputs: str):
        self._inputs.extend(inputs)
        return self

    def vcodec(self, codec: str):
        self._vcodec = codec
        return self

    def vcopy(self):
        self._vcodec = "copy"
        return self

    def novideo(self):
        self._vcodec = "vn"
        return self

    def crf(self, crf: int):
        self._vmode = "crf"
        self._crf = crf
        return self

    def cbr(self, bitrate: int):
        self._vmode = "cbr"
        self._cbr = bitrate
        return self

    def vbv(self, maxrate: int, bufsize: int = 0):
        if bufsize == 0:
            bufsize = 2 * maxrate
        self._vbv_maxrate = maxrate
        self._vbv_bufsize = bufsize
        return self

    def vfilter(self, filter: str):
        self._vfilter = filter
        return self

    def pix_fmt(self, pix_fmt: str):
        self._pix_fmt = pix_fmt
        return self

    def acodec(self, codec: str, omit_bitrate: bool = False):
        self._acodec = codec
        self._omit_abr = omit_bitrate
        return self

    def acopy(self):
        self._acodec = "copy"
        return self

    def noaudio(self):
        self._acodec = "an"
        return self

    def abr(self, bitrate: int):
        self._abr = bitrate
        return self

    def afilter(self, filter: str):
        self._afilter = filter
        return self

    def maxsize(self, size: int):
        self._maxsize = size
        return self

    def preparams(self, *params: str):
        self._preparams.extend(params)
        return self

    def postparams(self, *params: str):
        self._postparams.extend(params)
        return self

    def output(self, output: str):
        self._output = output
        return self


async def ffprobe(file: str):
    proc = await asyncio.create_subprocess_exec(
        *["ffprobe", file, "-v", "-8", "-show_entries", "format:stream", "-of", "json"], stdout=subprocess.PIPE)
    out, err = await proc.communicate()
    return json.loads(out)