Files
the-silent-gods-hearthfire/06_AI动漫制作/05_生产素材/正片/S01E01_H3_V01/assemble.py
T

83 lines
5.7 KiB
Python

"""Scene-level conform. Only accepted shots qualify for delivery unless --draft."""
import argparse,json,subprocess,sys,shutil
from pathlib import Path
from production import ROOT,SAMPLE,read,write,cmd,probe,FFMPEG,FFPROBE,log
MANIFEST=read(ROOT/'source_manifest.json')
LINES={d['id']:d for d in MANIFEST['dialogue']}
def edit_spec(shot):
f=ROOT/'shots'/shot['id']/'accepted.json'
decision=read(f) if f.exists() else {}
return float(decision.get('edit_in_seconds',0)),float(decision.get('edit_seconds',shot['seconds']))
def edit_duration(shot):return edit_spec(shot)[1]
def stamp(s):
ms=round(s*1000);h,ms=divmod(ms,3600000);m,ms=divmod(ms,60000);sec,ms=divmod(ms,1000)
return f'{h:02}:{m:02}:{sec:02},{ms:03}'
def escaped_filter_path(p):return p.as_posix().replace(':','\\:').replace("'","\\'")
def subtitles(shots):
entries=[];offset=0
for shot in shots:
f=ROOT/'shots'/shot['id'];ids=shot['dialogue_ids'];start_in,seconds=edit_spec(shot)
expected=[LINES[x]['text'] for x in ids]
if expected:
speech=f/'speech_check_small.json' if (f/'speech_check_small.json').exists() else f/'speech_check.json'
segments=read(speech).get('segments',[]) if speech.exists() else []
if shot['scene']==6:
st,en=(.2,3.2) if shot['id'].endswith('C02') else (2.65,4.95)
elif segments:
st=max(0,segments[0]['start']-start_in-.08);en=min(seconds,segments[-1]['end']-start_in+.12)
else:st,en=.25,max(.5,seconds-.25)
if en-st<.4:st,en=.25,seconds-.2
total=sum(max(1,len(t)) for t in expected);cursor=st
for line in expected:
duration=(en-st)*max(1,len(line))/total
entries.append((offset+cursor,offset+cursor+duration,line));cursor+=duration
offset+=seconds
return '\n\n'.join(f'{i+1}\n{stamp(a)} --> {stamp(b)}\n{t}' for i,(a,b,t) in enumerate(entries))+'\n'
def normalized(shot,draft=False):
f=ROOT/'shots'/shot['id'];assert (f/'outputs.json').exists(),f'{shot["id"]}: still rendering'
if not draft:assert (f/'accepted.json').exists(),f'{shot["id"]}: visual/dialogue review outstanding'
if (f/'screen_overlay.json').exists() and not draft:
assert (f/'post_outputs.json').exists(),f'{shot["id"]}: essential screen text still pending'
output_file=f/'post_outputs.json' if (f/'post_outputs.json').exists() else f/'outputs.json'
source=Path(read(output_file)[0]);out=f/'conform.mkv';start_in,seconds=edit_spec(shot)
reframe=read(f/'reframe.json') if (f/'reframe.json').exists() else {}
config={'source':str(source),'mtime':source.stat().st_mtime,'in':start_in,'seconds':seconds,'reframe':reframe,'version':3}
if out.exists() and (f/'conform_source.json').exists() and read(f/'conform_source.json')==config:return out
vf='scale=1280:720:force_original_aspect_ratio=increase,crop=1280:720,setsar=1,fps=24,setpts=PTS-STARTPTS'
if reframe.get('crop_top'):
top=int(reframe['crop_top']);vf=f'crop=iw:ih-{top}:0:{top},'+vf
af=f'aresample=48000:async=1:first_pts=0,apad,atrim=end_sample={round(seconds*48000)},asetpts=PTS-STARTPTS,afade=t=in:d=0.012,afade=t=out:st={max(0,seconds-.025)}:d=0.025'
cmd([FFMPEG,'-hide_banner','-loglevel','error','-y','-ss',start_in,'-i',source,'-vf',vf,'-af',af,'-t',seconds,'-map','0:v:0','-map','0:a:0','-c:v','libx264','-crf','18','-preset','medium','-pix_fmt','yuv420p','-fps_mode','cfr','-c:a','pcm_s16le','-ar','48000','-ac','2',out])
info=probe(out);assert abs(float(info['format']['duration'])-seconds)<.06
write(f/'conform_source.json',config);return out
def scene(number,draft=False):
shots=[s for s in MANIFEST['shots'] if s['scene']==number];assert shots
dest=ROOT/'scenes'/f'SC{number:02}';dest.mkdir(parents=True,exist_ok=True)
if number==6:
out=dest/'SC06_正片.mp4'
shutil.copy2(SAMPLE/'EP01_泉堂祝福_40秒试看.mp4',out)
write(dest/'probe.json',probe(out))
write(dest/'assembly.json',{'output':str(out),'duration':40,'draft':False,'reused_final_sample':True,'shots':[s['id'] for s in shots]})
return out
sources=[normalized(s,draft) for s in shots]
playlist=dest/'concat.txt';playlist.write_text(''.join("file '"+p.as_posix().replace("'","'\\''")+"'\n" for p in sources),encoding='utf-8')
srt=dest/'台词.srt';srt.write_text(subtitles(shots),encoding='utf-8-sig')
duration=sum(edit_duration(s) for s in shots)
out=dest/(f'SC{number:02}_分场检查.mp4' if draft else f'SC{number:02}_正片.mp4')
vf="subtitles='"+escaped_filter_path(srt)+"':force_style='FontName=Microsoft YaHei,FontSize=20,Outline=1,Shadow=0,MarginV=28'" if any(s['dialogue_ids'] for s in shots) else 'null'
cmd([FFMPEG,'-hide_banner','-loglevel','error','-y','-f','concat','-safe','0','-i',playlist,'-vf',vf,'-t',duration,'-c:v','libx264','-preset','medium','-crf','18','-pix_fmt','yuv420p','-fps_mode','cfr','-r','24','-c:a','aac','-b:a','192k','-ar','48000','-movflags','+faststart',out])
info=probe(out);video=next(x for x in info['streams'] if x['codec_type']=='video')
assert video['avg_frame_rate']=='24/1'
assert abs(float(info['format']['duration'])-duration)<.06
write(dest/'probe.json',info)
cmd([FFMPEG,'-hide_banner','-loglevel','error','-y','-i',out,'-vf',f'fps=1/{duration/12},scale=480:-1,tile=4x3','-frames:v',1,dest/'review.jpg'])
write(dest/'assembly.json',{'output':str(out),'duration':duration,'draft':draft,'shots':[s['id'] for s in shots]})
log('scene_assembled',scene=number,draft=draft,output=str(out),duration=duration)
print(str(out));return out
if __name__=='__main__':
p=argparse.ArgumentParser();p.add_argument('--scene',type=int,required=True);p.add_argument('--draft',action='store_true');a=p.parse_args();scene(a.scene,a.draft)