Verstack Local
841a95c56a
Save the tested release workflows, RAM media processing, program discovery, FLIRT library, and shared-object maintenance. Add Git exclusions, file attributes, and instructions for a later push using a forwarded SSH agent.
112 lines
8.1 KiB
Python
112 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
"""FFmpeg/ffprobe previews with retained source identities and visible codec failures."""
|
|
import hashlib,io,itertools,json,pathlib,subprocess,sys
|
|
from collections import deque
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from parallelism import worker_count
|
|
from media_extract import sniff
|
|
REVISION=3
|
|
|
|
def image_preview(path):
|
|
from PIL import Image,ImageOps
|
|
with Image.open(path) as source:
|
|
if source.width*source.height>16777216:raise ValueError('media dimensions exceed preview limit')
|
|
info={'format':{'format_name':source.format},'streams':[{'codec_type':'video','width':source.width,'height':source.height}]}
|
|
source.load()
|
|
thumb=ImageOps.exif_transpose(source).convert('RGBA')
|
|
thumb.thumbnail((320,240))
|
|
encoded=io.BytesIO();thumb.save(encoded,format='WEBP',quality=80)
|
|
encoded.seek(0)
|
|
with Image.open(encoded) as readback:
|
|
readback.load()
|
|
if readback.size!=thumb.size:raise ValueError('thumbnail readback dimensions differ')
|
|
metadata={'width':readback.width,'height':readback.height,'format':readback.format}
|
|
return info,metadata,encoded.getvalue()
|
|
|
|
def probe(tool,path):
|
|
p=subprocess.run([tool,'-v','error','-protocol_whitelist','file,pipe','-show_streams','-show_format','-of','json',str(path)],capture_output=True,timeout=30,check=True)
|
|
if len(p.stdout)>1024*1024:raise ValueError('media metadata exceeds bound')
|
|
return json.loads(p.stdout)
|
|
|
|
from progress import report
|
|
|
|
def run(request):
|
|
source=pathlib.Path(request['input_dir']);output=pathlib.Path(request['output_dir']);settings=request.get('settings') or {}
|
|
ffmpeg=settings.get('ffmpeg','ffmpeg');ffprobe=settings.get('ffprobe','ffprobe');budget=request['workspace_bytes']//2;used=0;records=[];verified={}
|
|
paths=[p for p in sorted(source.rglob('*')) if p.is_file() and not p.is_symlink()]
|
|
def candidates():
|
|
for path in paths:
|
|
with path.open('rb') as f:ext=sniff(f.read(32))
|
|
included=settings.get('include_extensions')
|
|
if ext is not None and (included is None or ext in included):yield path,ext
|
|
workers=worker_count(request,len(paths),512*1024**2)
|
|
# Image codecs release the GIL. Keep only one worker's decoded image per
|
|
# future, then commit in source order under the shared output byte budget.
|
|
# Drain each image group before starting FFmpeg so their CPU budgets do not overlap.
|
|
def prepared():
|
|
with ThreadPoolExecutor(max_workers=workers) as pool:
|
|
for images,group in itertools.groupby(candidates(),key=lambda item:item[1] in ('.png','.jpg','.webp')):
|
|
pending=deque()
|
|
for path,ext in group:
|
|
if images:
|
|
pending.append((path,ext,pool.submit(image_preview,path)))
|
|
if len(pending)>=workers:yield pending.popleft()
|
|
else:yield path,ext,None
|
|
while pending:yield pending.popleft()
|
|
threads=str(min(8,worker_count(request,8,128*1024**2)))
|
|
for index,(path,ext,future) in enumerate(prepared()):
|
|
report('Preparing previews',index,len(paths),'files',f'{workers} image workers · {path.name}')
|
|
rel=path.relative_to(source).as_posix()
|
|
with path.open('rb') as f:digest=hashlib.file_digest(f,'sha256').hexdigest()
|
|
row={'source':rel,'source_sha256':digest,'processing_revision':REVISION};dest=None
|
|
try:
|
|
if digest in verified:
|
|
row.update(verified[digest]);records.append(row);continue
|
|
if ext in ('.png','.jpg','.webp'):
|
|
dest=output/'previews'/(rel+'.webp');dest.parent.mkdir(parents=True,exist_ok=True)
|
|
info,preview,encoded=future.result()
|
|
if len(encoded)>=budget-used:raise ValueError('preview exceeded byte budget')
|
|
dest.write_bytes(encoded);used+=len(encoded)
|
|
row.update(status='verified',inspection=info,preview=dest.relative_to(output).as_posix(),preview_metadata=preview)
|
|
records.append(row);continue
|
|
info=probe(ffprobe,path);row['inspection']=info
|
|
streams=info.get('streams',[]);video=next((s for s in streams if s['codec_type']=='video'),None)
|
|
audio=next((s for s in streams if s['codec_type']=='audio'),None)
|
|
if video and video.get('width',0)*video.get('height',0)>16777216:raise ValueError('media dimensions exceed preview limit')
|
|
if ext=='.mp4' and video and video.get('codec_name')=='h264' and video.get('pix_fmt') in ('yuv420p','yuvj420p') and (audio is None or audio.get('codec_name')=='aac'):
|
|
subprocess.run([ffmpeg,'-nostdin','-v','error','-xerror','-threads',threads,'-filter_threads','1','-protocol_whitelist','file,pipe','-i',str(path),'-f','null','-'],stdout=subprocess.DEVNULL,stderr=subprocess.PIPE,check=True,timeout=300)
|
|
row.update(status='verified_source',reason='Browser-compatible H.264/AAC source passed complete decode; conversion is unnecessary')
|
|
verified[digest]={k:v for k,v in row.items() if k not in ('source','source_sha256','processing_revision')}
|
|
records.append(row);continue
|
|
base=output/'previews'/rel;base.parent.mkdir(parents=True,exist_ok=True)
|
|
image=ext in ('.png','.jpg','.webp')
|
|
dest=base.with_name(base.name+('.webp' if image else '.mp4' if video else '.wav'))
|
|
args=[ffmpeg,'-nostdin','-v','error','-max_alloc','268435456','-threads',threads,'-filter_threads','1','-protocol_whitelist','file,pipe','-i',str(path)]
|
|
if image:args+=['-frames:v','1','-vf','scale=320:240:force_original_aspect_ratio=decrease','-c:v','libwebp']
|
|
elif video:args+=['-map','0:v:0','-map','0:a:0?','-vf','scale=1280:720:force_original_aspect_ratio=decrease:force_divisible_by=2','-c:v','libx264','-preset','fast','-pix_fmt','yuv420p','-c:a','aac','-movflags','+faststart']
|
|
elif audio:args+=['-map','0:a:0','-c:a','pcm_s16le']
|
|
else:raise ValueError('no supported audio/video stream')
|
|
remaining=budget-used
|
|
if remaining<1024:raise ValueError('preview budget exhausted')
|
|
args+=['-threads',threads,'-fs',str(remaining),str(dest)]
|
|
subprocess.run(args,stdout=subprocess.DEVNULL,stderr=subprocess.PIPE,check=True,timeout=300)
|
|
size=dest.stat().st_size;used+=size
|
|
if size>=remaining:raise ValueError('preview exceeded byte budget')
|
|
preview=probe(ffprobe,dest)
|
|
if not image:
|
|
duration=float(info.get('format',{}).get('duration',0));converted=float(preview.get('format',{}).get('duration',0))
|
|
if duration and abs(converted-duration)>max(.25,duration*.01):raise ValueError('preview duration differs from source')
|
|
subprocess.run([ffmpeg,'-nostdin','-v','error','-xerror','-threads',threads,'-filter_threads','1','-protocol_whitelist','file,pipe','-i',str(dest),'-f','null','-'],stdout=subprocess.DEVNULL,stderr=subprocess.PIPE,check=True,timeout=300)
|
|
row.update(status='verified',preview=dest.relative_to(output).as_posix(),preview_metadata=preview)
|
|
verified[digest]={k:v for k,v in row.items() if k not in ('source','source_sha256','processing_revision')}
|
|
except (ValueError,OSError,subprocess.SubprocessError) as error:
|
|
if dest is not None:dest.unlink(missing_ok=True)
|
|
row.update(status='failed',error=str(error))
|
|
records.append(row)
|
|
report('Preparing previews',len(paths),len(paths),'files','Previews checked',force=True)
|
|
failed=any(r['status']=='failed' for r in records)
|
|
(output/'preview-evidence.json').write_text(json.dumps({'schema':1,'previews':records},indent=2))
|
|
return {'protocol':1,'layer':'derived','coverage':'partial' if failed else 'complete','files':[p.relative_to(output).as_posix() for p in sorted(output.rglob('*')) if p.is_file()],'warnings':['Some media previews failed; original assets remain retained.'] if failed else []}
|
|
if __name__=='__main__':
|
|
request=json.loads(pathlib.Path(sys.argv[1]).read_text());pathlib.Path(request['result_file']).write_text(json.dumps(run(request)))
|