src/youtube/utils.py

   1 import subprocess
   2 from tempfile import NamedTemporaryFile
   3
   4
   5 def video_from_image(img_path, duration, fps=25):
   6     tmp = NamedTemporaryFile(prefix='image', suffix='.mkv', delete=False)
   7     tmp.close()
   8     subprocess.run(
   9         ['ffmpeg', '-y', '-framerate', f'1/{duration}',  '-i', img_path, '-c:v', 'libx264', '-vf', f'fps={fps},format=yuv420p', tmp.name], check=True)
  10     return tmp.name
  11
  12
  13 def cut_video(video_path, duration):
  14     tmp = NamedTemporaryFile(prefix='cut', suffix='.mkv', delete=False)
  15     tmp.close()
  16     subprocess.run(
  17         ['ffmpeg', '-y', '-i', video_path, '-t', str(duration), tmp.name], check=True)
  18     return tmp.name
  19
  20
  21 def concat_videos(paths):
  22     filelist = NamedTemporaryFile(prefix='concat', suffix='.txt')
  23     for path in paths:
  24         filelist.write(f"file '{path}'\n".encode('utf-8'))
  25     filelist.flush()
  26
  27     output = NamedTemporaryFile(prefix='concat', suffix='.mkv', delete=False)
  28     output.close()
  29
  30     subprocess.run(
  31         ['ffmpeg', '-y', '-safe', '0', '-f', 'concat', '-i', filelist.name, '-c', 'copy', output.name],
  32         check=True)
  33
  34     filelist.close()
  35     return output.name
  36
  37
  38 def mux(channels, output_path=None):
  39     if not output_path:
  40         output = NamedTemporaryFile(prefix='concat', suffix='.mkv', delete=False)
  41         output.close()
  42         output_path = output.name
  43     args = ['ffmpeg']
  44     for c in channels:
  45         args.extend(['-i', c])
  46     args.extend(['-shortest', '-y', output_path])
  47     subprocess.run(args, check=True)
  48     return output_path
  49
  50
  51 def get_duration(path):
  52     return float(
  53         subprocess.run(
  54             [
  55                 "ffprobe",
  56                 "-i",
  57                 path,
  58                 "-show_entries",
  59                 "format=duration",
  60                 "-v",
  61                 "quiet",
  62                 "-of",
  63                 "csv=p=0",
  64             ],
  65             capture_output=True,
  66             text=True,
  67             check=True,
  68         ).stdout
  69     )
  70
  71
  72 def get_framerate(path):
  73     rates = subprocess.run(
  74             [
  75                 "ffprobe",
  76                 "-i",
  77                 path,
  78                 "-show_entries",
  79                 "stream=r_frame_rate",
  80                 "-v",
  81                 "quiet",
  82                 "-of",
  83                 "csv=p=0",
  84             ],
  85             capture_output=True,
  86             text=True,
  87             check=True,
  88         ).stdout.strip().split('\n')
  89     for rate in rates:
  90         a, b = rate.split('/')
  91         if b == '1':
  92             return int(a)