Init

entity/ffmpeg.py

@@ -0,0 +1,111 @@
+
+class FfmpegTask(object):
+
+    def __init__(self, input_file, task_type='copy', output_file=''):
+        if type(input_file) is str:
+            self.input_file = [input_file]
+        elif type(input_file) is list:
+            self.input_file = input_file
+        else:
+            self.input_file = []
+        self.task_type = task_type
+        self.output_file = output_file
+        self.mute = True
+        self.speed = 1
+        self.subtitles = []
+        self.luts = []
+        self.audios = []
+        self.overlays = []
+        self.annexb = False
+
+    def __repr__(self):
+        _str = f'FfmpegTask(input_file={self.input_file}, task_type={self.task_type}'
+        if len(self.luts) > 0:
+            _str += f', luts={self.luts}'
+        if len(self.audios) > 0:
+            _str += f', audios={self.audios}'
+        if len(self.overlays) > 0:
+            _str += f', overlays={self.overlays}'
+        if self.annexb:
+            _str += f', annexb={self.annexb}'
+        if self.mute:
+            _str += f', mute={self.mute}'
+        return _str + ')'
+
+    def analyze_input_render_tasks(self):
+        for i in self.input_file:
+            if type(i) is str:
+                continue
+            elif type(i) is FfmpegTask:
+                if i.need_run():
+                    yield i
+
+    def need_run(self):
+        """
+        判断是否需要运行
+        :rtype: bool
+        :return:
+        """
+        if self.annexb:
+            return True
+        # TODO: copy from url
+        return not self.check_can_copy()
+
+    def add_inputs(self, *inputs):
+        self.input_file.extend(inputs)
+
+    def add_overlay(self, *overlays):
+        self.overlays.extend(overlays)
+        self.correct_task_type()
+
+    def add_audios(self, *audios):
+        self.audios.extend(audios)
+        self.correct_task_type()
+        self.check_audio_track()
+
+    def add_lut(self, *luts):
+        self.luts.extend(luts)
+        self.correct_task_type()
+
+    def get_output_file(self):
+        if self.task_type == 'copy':
+            return self.input_file
+        return self.output_file
+
+    def correct_task_type(self):
+        if self.check_can_copy():
+            self.task_type = 'copy'
+        elif self.check_can_concat():
+            self.task_type = 'concat'
+        else:
+            self.task_type = 'encode'
+
+    def check_can_concat(self):
+        if len(self.luts) > 0:
+            return False
+        if len(self.overlays) > 0:
+            return False
+        if len(self.subtitles) > 0:
+            return False
+        if self.speed != 1:
+            return False
+        return True
+
+    def check_can_copy(self):
+        if len(self.luts) > 0:
+            return False
+        if len(self.overlays) > 0:
+            return False
+        if len(self.subtitles) > 0:
+            return False
+        if self.speed != 1:
+            return False
+        if len(self.audios) > 1:
+            return False
+        if len(self.input_file) > 1:
+            return False
+        return True
+
+    def check_audio_track(self):
+        if len(self.audios) > 0:
+            self.mute = False