src/youtube/utils.py

   1 from os import unlink
   2 import subprocess
   3 from tempfile import NamedTemporaryFile
   4
   5
   6 def video_from_image(img_path, duration, fps=25):
   7     tmp = NamedTemporaryFile(prefix='image', suffix='.mkv', delete=False)
   8     tmp.close()
   9     subprocess.run(
  10         ['ffmpeg', '-y', '-loop', '1', '-t', str(duration), '-i', img_path, '-c:v', 'libx264', '-vf', f'fps={fps},format=yuv420p', tmp.name], check=True)
  11     return tmp.name
  12
  13
  14 def cut_video(video_path, duration):
  15     tmp = NamedTemporaryFile(prefix='cut', suffix='.mkv', delete=False)
  16     tmp.close()
  17     subprocess.run(
  18         ['ffmpeg', '-y', '-i', video_path, '-t', str(duration), tmp.name], check=True)
  19     return tmp.name
  20
  21
  22 def ffmpeg_concat(paths, suffix):
  23     filelist = NamedTemporaryFile(prefix='concat', suffix='.txt')
  24     for path in paths:
  25         filelist.write(f"file '{path}'\n".encode('utf-8'))
  26     filelist.flush()
  27
  28     output = NamedTemporaryFile(prefix='concat', suffix=suffix, delete=False)
  29     output.close()
  30
  31     subprocess.run(
  32         ['ffmpeg', '-y', '-safe', '0', '-f', 'concat', '-i', filelist.name, output.name],
  33         check=True)
  34
  35     filelist.close()
  36     return output.name
  37
  38
  39 def concat_videos(paths):
  40     return ffmpeg_concat(paths, '.mkv')
  41
  42 def concat_audio(paths):
  43     std_paths = [
  44         standardize_audio(p)
  45         for p in paths
  46     ]
  47     output = ffmpeg_concat(std_paths, '.flac')
  48     for p in std_paths:
  49         unlink(p)
  50     return output
  51
  52 def standardize_audio(p):
  53     output = NamedTemporaryFile(prefix='standarize', suffix='.flac', delete=False)
  54     output.close()
  55     subprocess.run(
  56         ['ffmpeg', '-y', '-i', p, '-sample_fmt', 's16', '-acodec', 'flac', '-ac', '2', '-ar', '44100', output.name],
  57         check=True)
  58     return output.name
  59
  60 def standardize_video(p):
  61     output = NamedTemporaryFile(prefix='standarize', suffix='.mkv', delete=False)
  62     output.close()
  63     subprocess.run(
  64         ['ffmpeg', '-y', '-i', p, output.name],
  65         check=True)
  66     return output.name
  67
  68
  69
  70 def mux(channels, output_path=None):
  71     if not output_path:
  72         output = NamedTemporaryFile(prefix='concat', suffix='.mkv', delete=False)
  73         output.close()
  74         output_path = output.name
  75     args = ['ffmpeg']
  76     for c in channels:
  77         args.extend(['-i', c])
  78     args.extend([
  79         '-y', output_path])
  80     subprocess.run(args, check=True)
  81     return output_path
  82
  83
  84 def get_duration(path):
  85     return float(
  86         subprocess.run(
  87             [
  88                 "ffprobe",
  89                 "-i",
  90                 path,
  91                 "-show_entries",
  92                 "format=duration",
  93                 "-v",
  94                 "quiet",
  95                 "-of",
  96                 "csv=p=0",
  97             ],
  98             capture_output=True,
  99             text=True,
 100             check=True,
 101         ).stdout
 102     )
 103
 104
 105 def get_framerate(path):
 106     rates = subprocess.run(
 107             [
 108                 "ffprobe",
 109                 "-i",
 110                 path,
 111                 "-show_entries",
 112                 "stream=r_frame_rate",
 113                 "-v",
 114                 "quiet",
 115                 "-of",
 116                 "csv=p=0",
 117             ],
 118             capture_output=True,
 119             text=True,
 120             check=True,
 121         ).stdout.strip().split('\n')
 122     for rate in rates:
 123         a, b = rate.split('/')
 124         if b == '1':
 125             return int(a)