{py:load {video}
import os, subprocess, json, random, time, re
in_path = os.getenv('FILE_1')
if not in_path:
print('no input file passed in')
raise SystemExit(1)
def detect_ext(p):
base = os.path.basename(p)
_, ext = os.path.splitext(base)
if ext:
return ext.lower()
try:
fmt = subprocess.check_output(
['ffprobe','-v','error','-show_entries','format=format_name','-of','default=noprint_wrappers=1:nokey=1', p]
).decode().strip()
if 'mp4' in fmt:
return '.mp4'
if 'mov' in fmt:
return '.mov'
if 'matroska' in fmt or 'webm' in fmt or 'mkv' in fmt:
return '.mkv'
if 'avi' in fmt:
return '.avi'
except Exception:
pass
return '.mp4'
num = int(time.time()*1000) ^ random.randint(0, 1_000_000)
ext = detect_ext(in_path) or '.mp4'
out_path = f'/home/notsocoder/output/concat_{num}{ext}'
def ffprobe_duration(p):
cmd = ['ffprobe','-v','error','-show_entries','format=duration','-of','default=noprint_wrappers=1:nokey=1',p]
try:
dur = float(subprocess.check_output(cmd).decode().strip() or 0.0)
except Exception:
dur = 0.0
return max(0.0, dur)
dur = ffprobe_duration(in_path)
if dur == 0:
print('no duration found')
raise SystemExit(0)
seg_dir = '/home/notsocoder/input/segs'
rev_dir = '/home/notsocoder/input/rev'
os.makedirs(seg_dir, exist_ok=True)
os.makedirs(rev_dir, exist_ok=True)
subprocess.check_call([
'ffmpeg','-y','-loglevel','error','-i', in_path,
'-f','segment','-segment_time','1',
'-reset_timestamps','1',
os.path.join(seg_dir, f'f_%04d{ext}')])
forward = sorted([f for f in os.listdir(seg_dir) if f.startswith('f_') and f.endswith(ext)])
reverse = []
for f in forward:
fin = os.path.join(seg_dir, f)
fout = os.path.join(rev_dir, f.replace('f_','r_'))
subprocess.check_call([
'ffmpeg','-loglevel','error','-y','-i', fin,
'-vf','reverse','-af','areverse',
fout
])
reverse.append(fout)
pairs = []
for i, f in enumerate(forward):
fpath = os.path.join(seg_dir, f)
rpath = reverse[i] if i < len(reverse) else None
pairs.append(fpath)
if rpath:
pairs.append(rpath)
list_path = f'/home/notsocoder/input/list_{num}.txt'
with open(list_path,'w') as fw:
for p in pairs:
fw.write(f"file '{p}'\n")
subprocess.check_call([
'ffmpeg','-y','-loglevel','error','-f','concat','-safe','0','-i', list_path, out_path])
print('created:', out_path)}