higgsfield-faceless-video

Faceless video Create one finished video with a consistent visual style and narration. For an illustrated explainer use deliberate, diagram-led scenes; for a story or documentary adapt the pacing and visuals to the brief. Do not treat an explainer as a request for rapid cuts. Preserve choices and assets supplied by the user. Choose the requested mode before generating: illustrated explainers use ; fast-cut channel motion uses ; stills and picture stories use ; a sung children's music video uses ; explicitly requested talking characters use ; a requested presenter overlay uses alongside the mo…

\\t' read -r segment_number start_seconds end_seconds patch_kind patch_video; do\n [[ -n \"$segment_number\" ]] || continue\n if python3 - \"$start_seconds\" \"$CURSOR\" \u003c\u003c'PY'\nimport sys\nraise SystemExit(0 if float(sys.argv[1]) > float(sys.argv[2]) else 1)\nPY\n then\n CHUNK_NUMBER=$((CHUNK_NUMBER + 1))\n source_chunk=\"$TMP/chunk-$(printf '%03d' \"$CHUNK_NUMBER\").mp4\"\n source_duration=\"$(python3 - \"$start_seconds\" \"$CURSOR\" \u003c\u003c'PY'\nimport sys\nprint(float(sys.argv[1]) - float(sys.argv[2]))\nPY\n)\"\n normalize_source_range \"$CURSOR\" \"$source_duration\" \"$source_chunk\"\n printf \"file '%s'\\n\" \"$source_chunk\" >> \"$CONCAT_LIST\"\n fi\n\n segment_patch=\"$TMP/segment-$(printf '%03d' \"$segment_number\").mp4\"\n if [[ \"$patch_kind\" == \"patch\" ]]; then\n normalize_video \"$patch_video\" \"\" 10 \"$segment_patch\"\n else\n part_list=\"$TMP/segment-${segment_number}-parts.txt\"\n : > \"$part_list\"\n while IFS=

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

\\t' read -r part_segment part_number part_seconds part_video part_audio; do\n [[ \"$part_segment\" == \"$segment_number\" ]] || continue\n normalized_part=\"$TMP/segment-${segment_number}-part-${part_number}.mp4\"\n normalize_video \"$part_video\" \"$part_audio\" \"$part_seconds\" \"$normalized_part\"\n printf \"file '%s'\\n\" \"$normalized_part\" >> \"$part_list\"\n done \u003c \"$PARTS_TSV\"\n ffmpeg -v error -y -f concat -safe 0 -i \"$part_list\" -c copy \"$segment_patch\"\n fi\n patch_duration=\"$(ffprobe -v error -show_entries format=duration -of csv=p=0 \"$segment_patch\")\"\n python3 - \"$patch_duration\" \u003c\u003c'PY'\nimport sys\nduration = float(sys.argv[1])\nif abs(duration - 10) > 0.25:\n raise SystemExit(f\"ERROR: normalized patch is {duration:.3f}s, expected 10s\")\nPY\n CHUNK_NUMBER=$((CHUNK_NUMBER + 1))\n printf \"file '%s'\\n\" \"$segment_patch\" >> \"$CONCAT_LIST\"\n python3 - \"$segment_number\" \"$start_seconds\" \"$end_seconds\" \"$patch_duration\" >> \"$PER_SEGMENT_JSON\" \u003c\u003c'PY'\nimport json\nimport sys\nprint(json.dumps({\n \"n\": int(sys.argv[1]),\n \"start_seconds\": int(sys.argv[2]),\n \"end_seconds\": int(sys.argv[3]),\n \"patch_seconds\": round(float(sys.argv[4]), 3),\n}))\nPY\n CURSOR=\"$end_seconds\"\ndone \u003c \"$SEGMENTS_TSV\"\n\nif python3 - \"$PROBED_SOURCE_SECONDS\" \"$CURSOR\" \u003c\u003c'PY'\nimport sys\nraise SystemExit(0 if float(sys.argv[1]) > float(sys.argv[2]) else 1)\nPY\nthen\n CHUNK_NUMBER=$((CHUNK_NUMBER + 1))\n source_chunk=\"$TMP/chunk-$(printf '%03d' \"$CHUNK_NUMBER\").mp4\"\n source_duration=\"$(python3 - \"$PROBED_SOURCE_SECONDS\" \"$CURSOR\" \u003c\u003c'PY'\nimport sys\nprint(float(sys.argv[1]) - float(sys.argv[2]))\nPY\n)\"\n normalize_source_range \"$CURSOR\" \"$source_duration\" \"$source_chunk\"\n printf \"file '%s'\\n\" \"$source_chunk\" >> \"$CONCAT_LIST\"\nfi\n\nffmpeg -v error -y -f concat -safe 0 -i \"$CONCAT_LIST\" -c copy -movflags +faststart \"$OUT\"\nACTUAL_SECONDS=\"$(ffprobe -v error -show_entries format=duration -of csv=p=0 \"$OUT\")\"\nSEGMENT_COUNT=\"$(wc -l \u003c \"$SEGMENTS_TSV\" | tr -d ' ')\"\nPATCH_COUNT=\"$SEGMENT_COUNT\"\n\npython3 - \\\n \"$OUT.assembly.json\" \\\n \"$SOURCE_URL\" \\\n \"$SOURCE_SECONDS\" \\\n \"$REQUESTED_SECONDS\" \\\n \"$ACTUAL_SECONDS\" \\\n \"$PATCH_COUNT\" \\\n \"$PER_SEGMENT_JSON\" \u003c\u003c'PY'\nimport json\nimport pathlib\nimport sys\n\n(\n output_path,\n source_video_url,\n source_seconds,\n requested_seconds,\n actual_seconds,\n patch_count,\n per_segment_path,\n) = sys.argv[1:]\nper_segment = [\n json.loads(line)\n for line in pathlib.Path(per_segment_path).read_text().splitlines()\n if line\n]\npayload = {\n \"manifest_version\": 2,\n \"script\": \"assemble_partial_edit.sh\",\n \"assembly_attempts\": 1,\n\n \"source_s\": float(source_seconds),\n \"requested_s\": float(requested_seconds),\n \"actual_s\": round(float(actual_seconds), 3),\n \"segments\": len(per_segment),\n \"patch_count\": int(patch_count),\n \"per_segment\": per_segment,\n}\npathlib.Path(output_path).write_text(json.dumps(payload, indent=2) + \"\\n\")\nPY\n\necho \"OK: $OUT (${ACTUAL_SECONDS}s); sidecar: $OUT.assembly.json\" >&2\n","content_type":"application/x-sh; charset=utf-8","language":"bash","size":10924,"content_sha256":"e1613a84350f07df475a21bbffa1e46e08160c2bfe412fbd2fb082d571d722ca"},{"filename":"scripts/assemble_slides.sh","content":"#!/usr/bin/env bash\n# Usage: assemble_slides.sh --out final.mp4 --audio narration.wav --blocks N --manifest frames.txt [--music bed.mp3]\nset -euo pipefail\n\nOUT=\"final.mp4\"; BREATH=\"0.2\"; MINSLIDE=\"1.0\"; MAXHOLD=\"2.5\"; MUSIC=\"\"; MVOL=\"0.10\"; BLOCKS=\"\"; REQUESTED_SECONDS=\"\"; DURATION_BLOCKS=\"\"; MANIFEST=\"\"; TIMELINE=\"\"; FRAMES_DIR=\"\"; ALLOWMM=0; AUDIO=\"\"; ARGS=()\nwhile [[ $# -gt 0 ]]; do\n case \"$1\" in\n --out) OUT=\"$2\"; shift 2 ;;\n --breath) BREATH=\"$2\"; shift 2 ;;\n --min-slide) MINSLIDE=\"$2\"; shift 2 ;;\n --max-hold) MAXHOLD=\"$2\"; shift 2 ;;\n --audio) AUDIO=\"$2\"; shift 2 ;;\n --music) MUSIC=\"$2\"; shift 2 ;;\n --music-vol) MVOL=\"$2\"; shift 2 ;;\n --blocks) BLOCKS=\"$2\"; shift 2 ;;\n --requested-seconds) REQUESTED_SECONDS=\"$2\"; shift 2 ;;\n --manifest) MANIFEST=\"$2\"; shift 2 ;;\n --timeline) TIMELINE=\"$2\"; shift 2 ;;\n --frames-dir) FRAMES_DIR=\"$2\"; shift 2 ;;\n --subs) echo \"ERROR: --subs was removed. Assemble first, then follow the bundled captions instructions using the clean voice takes and assembly receipt.\" >&2; exit 2 ;;\n --allow-mismatch) ALLOWMM=1; shift ;;\n -h|--help) grep '^#' \"$0\" | sed 's/^# \\{0,1\\}//'; exit 0 ;;\n *) ARGS+=(\"$1\"); shift ;;\n esac\ndone\nif [[ -n \"$REQUESTED_SECONDS\" ]]; then\n [[ \"$REQUESTED_SECONDS\" =~ ^[0-9]+$ ]] && (( REQUESTED_SECONDS > 0 )) || {\n echo \"ERROR: --requested-seconds must be a positive integer\" >&2\n exit 2\n }\n DURATION_BLOCKS=$(( (REQUESTED_SECONDS + 9) / 10 ))\nfi\nMVOL=\"$(awk -v v=\"$MVOL\" 'BEGIN{v=v+0; if(v\u003c0)v=0; if(v>0.2){print \"WARN: --music-vol clamped to 0.20 (voice stays 1.0)\" > \"/dev/stderr\"; v=0.2} printf \"%.3f\", v}')\"\nbed_gain_db() {\n local vmean bmean trim\n vmean=\"$(ffmpeg -hide_banner -i \"$1\" -af volumedetect -f null - 2>&1 | grep -Eo 'mean_volume: *-?[0-9.]+' | grep -Eo '\\-?[0-9.]+' | head -1)\"\n bmean=\"$(ffmpeg -hide_banner -i \"$2\" -af volumedetect -f null - 2>&1 | grep -Eo 'mean_volume: *-?[0-9.]+' | grep -Eo '\\-?[0-9.]+' | head -1)\"\n awk -v v=\"${vmean:--20}\" -v b=\"${bmean:--20}\" -v mv=\"$3\" 'BEGIN{\n trim = 20*log(mv/0.10)/log(10);\n g = (v - 14) - b + trim;\n if (g > 0) g = 0; if (g \u003c -40) g = -40;\n printf \"%.2f\", g }'\n}\n\nfor b in ffmpeg ffprobe awk python3; do command -v \"$b\" >/dev/null 2>&1 || { echo \"ERROR: '$b' not found\" >&2; exit 1; }; done\n\nif [[ -n \"$AUDIO\" ]]; then\n [[ -f \"$AUDIO\" ]] || { echo \"ERROR: --audio not found: $AUDIO\" >&2; exit 1; }\n [[ -n \"$MUSIC\" && ! -f \"$MUSIC\" ]] && { echo \"ERROR: --music file not found: $MUSIC\" >&2; exit 1; }\n ASSEMBLY_ATTEMPT=1\n TMP=\"$(mktemp -d)\"\n trap 'rm -rf \"$TMP\"' EXIT\n FR_IMG=(); FR_DUR=(); FR_BEAT=(); FR_START=(); FR_END=()\n if [[ -n \"$TIMELINE\" ]]; then\n [[ -z \"$MANIFEST\" ]] || { echo \"ERROR: pass --timeline or --manifest, not both\" >&2; exit 1; }\n [[ -f \"$TIMELINE\" ]] || { echo \"ERROR: --timeline not found: $TIMELINE\" >&2; exit 1; }\n [[ -d \"$FRAMES_DIR\" ]] || { echo \"ERROR: --frames-dir is required with --timeline\" >&2; exit 1; }\n python3 - \"$TIMELINE\" > \"$TMP/timeline.tsv\" \u003c\u003c'PY'\nimport json\nimport sys\n\npayload = json.load(open(sys.argv[1], encoding=\"utf-8\"))\nif payload.get(\"manifest_version\") != 2:\n raise SystemExit(\"ERROR: scene timeline must use manifest_version 2\")\nframes = payload.get(\"frames\")\nif not isinstance(frames, list) or not frames:\n raise SystemExit(\"ERROR: scene timeline must contain frames\")\nprevious_end = 0.0\nprevious_frame_job_id = None\nprevious_frame_url = None\nprevious_output_slot = None\nfor expected, frame in enumerate(frames, start=1):\n if not isinstance(frame, dict) or frame.get(\"n\") != expected:\n raise SystemExit(f\"ERROR: timeline frame {expected} is invalid\")\n start = float(frame.get(\"start\", -1))\n end = float(frame.get(\"end\", -1))\n duration = float(frame.get(\"duration\", -1))\n if abs(start - previous_end) > 0.01:\n raise SystemExit(f\"ERROR: timeline gap/overlap before frame {expected}\")\n if duration \u003c= 0 or duration > 1.5 or abs((end - start) - duration) > 0.01:\n raise SystemExit(f\"ERROR: timeline frame {expected} duration is invalid\")\n if frame.get(\"output_slot\") != f\"frame-{expected:03d}\":\n raise SystemExit(f\"ERROR: timeline frame {expected} output_slot is invalid\")\n frame_job_id = frame.get(\"frame_job_id\")\n frame_url = frame.get(\"frame_url\")\n if not isinstance(frame_job_id, str) or not frame_job_id.strip():\n raise SystemExit(f\"ERROR: timeline frame {expected} frame_job_id is missing\")\n if not isinstance(frame_url, str) or not frame_url.startswith((\"http://\", \"https://\")):\n raise SystemExit(f\"ERROR: timeline frame {expected} frame_url is missing\")\n if frame.get(\"image_mode\") == \"variation\" and (\n frame.get(\"reference_output_slot\") != previous_output_slot\n or frame.get(\"reference_frame_job_id\") != previous_frame_job_id\n or frame.get(\"reference_frame_url\") != previous_frame_url\n ):\n raise SystemExit(f\"ERROR: timeline frame {expected} variation lineage is invalid\")\n print(expected, int(frame[\"beat_n\"]), f\"{start:.3f}\", f\"{end:.3f}\", f\"{duration:.3f}\", sep=\"\\t\")\n previous_end = end\n previous_frame_job_id = frame_job_id\n previous_frame_url = frame_url\n previous_output_slot = frame[\"output_slot\"]\nPY\n while IFS=

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

\\t' read -r frame_number beat_number frame_start frame_end frame_duration; do\n FR_IMG+=(\"${FRAMES_DIR}/frame$(printf '%03d' \"$frame_number\").png\")\n FR_DUR+=(\"$frame_duration\")\n FR_BEAT+=(\"$beat_number\")\n FR_START+=(\"$frame_start\")\n FR_END+=(\"$frame_end\")\n done \u003c \"$TMP/timeline.tsv\"\n elif [[ -n \"$MANIFEST\" ]]; then\n [[ -f \"$MANIFEST\" ]] || { echo \"ERROR: --manifest not found: $MANIFEST\" >&2; exit 1; }\n while read -r im du _extra; do\n [[ -z \"$im\" || \"$im\" == \\#* ]] && continue\n [[ -n \"$du\" ]] || { echo \"ERROR: --audio manifest line needs 'image \u003cseconds>': '$im'\" >&2; exit 1; }\n FR_IMG+=(\"$im\"); FR_DUR+=(\"$du\"); FR_BEAT+=(\"0\"); FR_START+=(\"0\"); FR_END+=(\"0\")\n done \u003c \"$MANIFEST\"\n else\n (( ${#ARGS[@]} >= 2 && ${#ARGS[@]} % 2 == 0 )) || { echo \"ERROR: pass image/duration pairs or --manifest in --audio mode\" >&2; exit 1; }\n for ((k=0;k\u003c${#ARGS[@]};k+=2)); do\n FR_IMG+=(\"${ARGS[k]}\"); FR_DUR+=(\"${ARGS[k+1]}\")\n FR_BEAT+=(\"0\"); FR_START+=(\"0\"); FR_END+=(\"0\")\n done\n fi\n frames=${#FR_IMG[@]}\n [[ -z \"$BLOCKS\" ]] && { echo \"ERROR: --blocks N is REQUIRED (the frame count).\" >&2; exit 1; }\n (( frames == BLOCKS )) || { echo \"ERROR: --blocks $BLOCKS but $frames frames supplied — fix the manifest; do NOT ship a partial video.\" >&2; exit 1; }\n for im in \"${FR_IMG[@]}\"; do\n [[ -s \"$im\" ]] || { echo \"ERROR: frame image not found or empty: $im\" >&2; exit 1; }\n done\n ADUR=\"$(ffprobe -v error -show_entries format=duration -of csv=p=0 \"$AUDIO\")\"\n MINFRAMES=\"$(awk -v a=\"$ADUR\" 'BEGIN{ m=int(a/1.5); if(m\u003c1)m=1; print m }')\"\n (( frames \u003c MINFRAMES )) && { echo \"ERROR: only ${frames} frames for ${ADUR}s of narration — that is a slideshow. Frame-by-frame needs >= ${MINFRAMES} frames (one every ~1.5s); most are cheap micro-variation edits of the previous frame. Re-cut the timeline denser and regenerate the missing frames.\" >&2; exit 1; }\n UNIQ=\"$(python3 -c 'import hashlib,sys; print(len({hashlib.sha256(open(p,\"rb\").read()).digest() for p in sys.argv[1:]}))' \"${FR_IMG[@]}\")\"\n (( UNIQ \u003c MINFRAMES )) && { echo \"ERROR: ${frames} frame entries but only ${UNIQ} DISTINCT pictures for ${ADUR}s of narration — repeating one image is a gallery, not frame-by-frame. Need >= ${MINFRAMES} distinct pictures; most are cheap micro-variation EDITS of the previous frame (one reference image + one changed detail), not full re-renders.\" >&2; exit 1; }\n echo \" density: ${frames} entries / ${UNIQ} distinct pictures / floor ${MINFRAMES}\" >&2\n DIMS=\"$(ffprobe -v error -select_streams v:0 -show_entries stream=width,height -of csv=p=0 \"${FR_IMG[0]}\")\"\n W=\"${DIMS%%,*}\"; H=\"${DIMS##*,}\"; W=$(( (W/2)*2 )); H=$(( (H/2)*2 )); AR_W=\"$W\"; AR_H=\"$H\"\n if (( W > 1920 || H > 1920 )); then\n read -r W H \u003c\u003c\u003c \"$(awk -v w=\"$W\" -v h=\"$H\" 'BEGIN{ m=(w>h)?w:h; s=1920/m; printf \"%d %d\", int(w*s/2)*2, int(h*s/2)*2 }')\"\n echo \" output capped to ${W}x${H} (1080p-class)\" >&2\n fi\n TAILHOLD=0.3\n PRESUM=\"$(awk 'BEGIN{s=0} {s+=$1} END{printf \"%.3f\", s}' \u003c\u003c\u003c \"$(printf '%s\\n' \"${FR_DUR[@]}\")\")\"\n LASTI=$((frames-1))\n NEED=\"$(awk -v a=\"$ADUR\" -v s=\"$PRESUM\" -v th=\"$TAILHOLD\" -v last=\"${FR_DUR[$LASTI]}\" 'BEGIN{\n d=(a+th)-s; cap=1.5-last;\n if(d\u003c0)d=0; if(cap\u003c0)cap=0; if(d>cap)d=cap;\n printf \"%.3f\", d\n }')\"\n if awk -v n=\"$NEED\" 'BEGIN{exit (n>0.01)?0:1}'; then\n FR_DUR[$LASTI]=\"$(awk -v d=\"${FR_DUR[$LASTI]}\" -v n=\"$NEED\" 'BEGIN{printf \"%.3f\", d+n}')\"\n echo \" tail hold: last frame extended by ${NEED}s so the narration is never clipped\" >&2\n fi\n FPS=30; LIST=\"$TMP/frames.ffconcat\"; printf 'ffconcat version 1.0\\n' > \"$LIST\"; SUM=0; PBJSON=(); PREVNUM=-1\n echo \"[1/3] ${frames} frames -> ${W}x${H} @ ${FPS}fps; narration ${ADUR}s laid over the cut\" >&2\n for ((i=0;i\u003cframes;i++)); do\n im=\"${FR_IMG[i]}\"; du=\"${FR_DUR[i]}\"\n [[ \"$im\" != *\"'\"* && \"$im\" != *

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

\\n'* ]] || { echo \"ERROR: frame image path contains an unsupported quote or newline: $im\" >&2; exit 1; }\n im=\"$(realpath \"$im\")\"\n FR_IMG[i]=\"$im\"\n fn=\"$(basename \"$im\" | sed 's/\\.[^.]*$//' | grep -Eo '[0-9]+' | tail -1 || true)\"\n if [[ -n \"$fn\" ]]; then\n if (( 10#$fn \u003c= PREVNUM )) && [[ \"$ALLOWMM\" != \"1\" ]]; then\n echo \"ERROR: frame order broken at manifest line $((i+1)): $(basename \"$im\") (number ${fn}) does not come after ${PREVNUM}. The manifest must list frameNNN in STRICT ascending timeline order — rebuild it by looping the numbered timeline 1..N, not by ls or job-finish order (--allow-mismatch to override for unnumbered files).\" >&2\n exit 1\n fi\n PREVNUM=$(( 10#$fn ))\n fi\n awk -v d=\"$du\" 'BEGIN{exit (d+0>0)?0:1}' || { echo \"ERROR: frame $((i+1)) has a non-positive duration '$du'\" >&2; exit 1; }\n awk -v d=\"$du\" -v m=\"$MAXHOLD\" 'BEGIN{exit (d+0>m+0)?0:1}' && { echo \"ERROR: frame $((i+1)) ($im) holds ${du}s (max ${MAXHOLD}s) — slideshow feel; split it into micro-variation frames.\" >&2; exit 1; }\n ID=\"$(ffprobe -v error -select_streams v:0 -show_entries stream=width,height -of csv=p=0 \"$im\")\"\n IW=\"${ID%%,*}\"; IH=\"${ID##*,}\"\n awk -v w=\"$IW\" -v h=\"$IH\" -v W=\"$AR_W\" -v H=\"$AR_H\" 'BEGIN{ a=w/h; A=W/H; d=a-A; if(d\u003c0)d=-d; exit (d/A\u003c=0.02)?0:1 }' || {\n echo \"ERROR: frame $(basename \"$im\") is ${IW}x${IH} — not the video aspect (${AR_W}x${AR_H}). Asset sheets are refs, NEVER frames: regenerate in the chosen aspect_ratio.\" >&2; exit 1; }\n printf \"file '%s'\\nduration %s\\n\" \"$im\" \"$du\" >> \"$LIST\"\n SUM=\"$(awk -v s=\"$SUM\" -v d=\"$du\" 'BEGIN{printf \"%.3f\", s+d}')\"\n PBJSON[i]=\"$(printf '{\"n\":%d,\"beat_n\":%s,\"start\":%s,\"end\":%s,\"image\":\"%s\",\"dur\":%s}' \"$((i+1))\" \"${FR_BEAT[i]}\" \"${FR_START[i]}\" \"${FR_END[i]}\" \"$(basename \"$im\")\" \"$du\")\"\n done\n awk -v s=\"$SUM\" -v a=\"$ADUR\" -v th=\"$TAILHOLD\" 'BEGIN{x=s-a; exit (x>=-0.2 && x\u003c=th+0.2)?0:1}' || {\n echo \"ERROR: frame durations sum to ${SUM}s against ${ADUR}s of narration — outside [-0.2s, +$(awk -v th=\"$TAILHOLD\" 'BEGIN{printf \"%.1f\", th+0.2}')s]. The pictures would drift off the words. Re-cut the frame timeline from the Whisper segment starts: a frame's duration is next_segment_start - this_segment_start, never an estimate.\" >&2; exit 1; }\n printf \"file '%s'\\n\" \"${FR_IMG[$LASTI]}\" >> \"$LIST\"\n VSILENT=\"$TMP/vid.mp4\"\n echo \"[2/3] encode ${frames} timed frames in one pass -> visual track\" >&2\n ffmpeg -y -loglevel error -f concat -safe 0 -i \"$LIST\" \\\n -vf \"scale=${W}:${H}:force_original_aspect_ratio=decrease,pad=${W}:${H}:(ow-iw)/2:(oh-ih)/2:black,fps=${FPS},format=yuv420p,setsar=1\" \\\n -t \"$SUM\" -an -c:v libx264 -preset veryfast -crf 20 -movflags +faststart \"$VSILENT\"\n QUIET=\"$(ffmpeg -i \"$AUDIO\" -af \"silencedetect=noise=-40dB:d=2\" -f null - 2>&1 | grep -c 'silence_start: 0' || true)\"\n ATOT=\"$(ffmpeg -i \"$AUDIO\" -af \"volumedetect\" -f null - 2>&1 | awk -F': ' '/mean_volume/{print $2+0}')\"\n awk -v m=\"$ATOT\" 'BEGIN{exit (m \u003c -70)?0:1}' && { echo \"ERROR: --audio is effectively silent (mean ${ATOT}dB) — bad narration track; do NOT deliver.\" >&2; exit 1; }\n echo \"[3/3] lay narration + finalize -> $OUT\" >&2\n ln2b() { ffmpeg -hide_banner -i \"$1\" -af \"loudnorm=I=-16:TP=-1.5:LRA=11:print_format=json\" -f null - 2>&1 \\\n | awk '/\"input_i\"/{i=$3}/\"input_tp\"/{t=$3}/\"input_lra\"/{l=$3}/\"input_thresh\"/{h=$3}/\"target_offset\"/{o=$3}{gsub(/[\",]/,\"\",i);gsub(/[\",]/,\"\",t);gsub(/[\",]/,\"\",l);gsub(/[\",]/,\"\",h);gsub(/[\",]/,\"\",o)}END{printf \"measured_I=%s:measured_TP=%s:measured_LRA=%s:measured_thresh=%s:offset=%s\", i,t,l,h,o}'; }\n NARRN=\"$TMP/narr_matched.wav\"\n if ffmpeg -y -loglevel error -i \"$AUDIO\" -af \"loudnorm=I=-19:TP=-1.5:LRA=11\" -ar 44100 -ac 2 \"$NARRN\" 2>/dev/null; then\n echo \" narration level-matched to -19 LUFS before the bed\" >&2\n AUDIO=\"$NARRN\"\n fi\n if [[ -n \"$MUSIC\" ]]; then\n BEDG=\"$(bed_gain_db \"$AUDIO\" \"$MUSIC\" \"$MVOL\")\"\n echo \" music bed: measured against the speech -> ${BEDG} dB (then ducked under the voice)\" >&2\n ffmpeg -y -loglevel error -i \"$VSILENT\" -i \"$AUDIO\" -stream_loop -1 -i \"$MUSIC\" \\\n -filter_complex \"[1:a]aformat=sample_rates=44100:channel_layouts=stereo,asplit=2[vo][key];[2:a]aformat=sample_rates=44100:channel_layouts=stereo,volume=${BEDG}dB[m];[m][key]sidechaincompress=threshold=0.02:ratio=8:attack=10:release=300[duck];[vo][duck]amix=inputs=2:duration=first:normalize=0[a]\" \\\n -map 0:v -map \"[a]\" -shortest -c:v copy -c:a aac -b:a 192k \"$TMP/mix.mp4\"\n else\n ffmpeg -y -loglevel error -i \"$VSILENT\" -i \"$AUDIO\" \\\n -filter_complex \"[1:a]aformat=sample_rates=44100:channel_layouts=stereo[a]\" \\\n -map 0:v -map \"[a]\" -shortest -c:v copy -c:a aac -b:a 192k \"$TMP/mix.mp4\"\n fi\n LN=\"$(ln2b \"$TMP/mix.mp4\")\"\n OUT_CANDIDATE=\"$TMP/final_candidate.mp4\"\n if [[ \"$LN\" == *measured_I=-* ]]; then\n ffmpeg -y -loglevel error -i \"$TMP/mix.mp4\" -af \"loudnorm=I=-16:TP=-1.5:LRA=11:${LN}:linear=true\" -c:v copy -c:a aac -b:a 192k \"$OUT_CANDIDATE\"\n else\n ffmpeg -y -loglevel error -i \"$TMP/mix.mp4\" -af \"loudnorm=I=-16:TP=-1.5:LRA=11\" -c:v copy -c:a aac -b:a 192k \"$OUT_CANDIDATE\"\n fi\n DERR=\"$( (ffmpeg -v error -xerror -i \"$OUT_CANDIDATE\" -f null - ) 2>&1 | head -3 || true)\"\n [[ -n \"$DERR\" ]] && { echo \"ERROR: final failed decode validation:\" >&2; echo \"$DERR\" >&2; exit 1; }\n POSTER=\"${OUT%.mp4}_poster.jpg\"\n POSTER_CANDIDATE=\"$TMP/final_poster.jpg\"\n ffmpeg -y -loglevel error -ss 1 -i \"$OUT_CANDIDATE\" -frames:v 1 -q:v 2 \"$POSTER_CANDIDATE\" || true\n [[ -s \"$POSTER_CANDIDATE\" ]] || ffmpeg -y -loglevel error -i \"$OUT_CANDIDATE\" -frames:v 1 -q:v 2 \"$POSTER_CANDIDATE\" || true\n FDUR=\"$(ffprobe -v error -show_entries format=duration -of csv=p=0 \"$OUT_CANDIDATE\")\"\n SIDE=\"${OUT}.assembly.json\"\n SIDE_CANDIDATE=\"$TMP/final.assembly.json\"\n { printf '{\"manifest_version\":2,\"timeline_version\":2,\"assembly_attempts\":%d,\"script\":\"assemble_slides.sh\",\"mode\":\"frame-by-frame\",\"out\":\"%s\",\"frames\":%d,\"narration_s\":%s,\"actual_s\":%s' \\\n \"$ASSEMBLY_ATTEMPT\" \"$(basename \"$OUT\")\" \"$frames\" \"$ADUR\" \"$FDUR\"\n if [[ -n \"$REQUESTED_SECONDS\" ]]; then\n printf ',\"requested_s\":%d,\"blocks\":%d' \"$REQUESTED_SECONDS\" \"$DURATION_BLOCKS\"\n fi\n printf ',\"width\":%s,\"height\":%s,\"fps\":\"%s\",\"music\":%s,\"music_vol\":%s,\"gate\":\"one continuous narration over the cut, per-frame durations sum to narration -0.2/+0.6s (deliberate tail hold on the last frame), max-hold %ss, all frames in aspect, 1080p cap, ducked bed, 2-pass linear loudnorm -16, full decode\",\"per_frame\":[' \\\n \"$W\" \"$H\" \"$FPS\" \"$([[ -n \"$MUSIC\" ]] && echo true || echo false)\" \"$MVOL\" \"$MAXHOLD\"\n for ((i=0;i\u003cframes;i++)); do printf '%s%s' \"${PBJSON[i]}\" \"$([[ $i -lt $((frames-1)) ]] && echo ,)\"; done\n printf '],\"ts\":\"%s\"}\\n' \"$(date -u +%Y-%m-%dT%H:%M:%SZ)\"; } > \"$SIDE_CANDIDATE\"\n mv \"$OUT_CANDIDATE\" \"$OUT\"\n [[ -s \"$POSTER_CANDIDATE\" ]] && mv \"$POSTER_CANDIDATE\" \"$POSTER\"\n mv \"$SIDE_CANDIDATE\" \"$SIDE\"\n echo \"ASSEMBLY-SIDECAR: $SIDE\" >&2\n [[ -s \"$POSTER\" ]] && echo \"POSTER: $POSTER\" >&2\n echo \"DONE: $OUT (ONE file, ${frames} frames, narration ${ADUR}s, actual ${FDUR}s; decode-validated)\" >&2\n exit 0\nfi\n\nif [[ -n \"$MANIFEST\" ]]; then\n [[ -f \"$MANIFEST\" ]] || { echo \"ERROR: --manifest not found: $MANIFEST\" >&2; exit 1; }\n while read -r f1 f2 f3 _extra; do\n [[ -z \"$f1\" || \"$f1\" == \\#* ]] && continue\n if [[ -n \"$f3\" ]]; then ARGS+=(\"$f1|$f2\" \"$f3\")\n elif [[ -n \"$f2\" ]]; then ARGS+=(\"$f1\" \"$f2\")\n else echo \"ERROR: manifest line has no voice file: '$f1'\" >&2; exit 1; fi\n done \u003c \"$MANIFEST\"\nfi\nfor b in ffmpeg ffprobe awk; do command -v \"$b\" >/dev/null 2>&1 || { echo \"ERROR: '$b' not found\" >&2; exit 1; }; done\nn=${#ARGS[@]}\n(( n >= 2 && n % 2 == 0 )) || { echo \"ERROR: pass image/voice PAIRS (or --manifest)\" >&2; exit 1; }\npairs=$(( n / 2 ))\nif [[ -n \"$BLOCKS\" ]] && (( pairs != BLOCKS )); then\n echo \"ERROR: --blocks $BLOCKS declared but $pairs pairs supplied — a missing/extra beat. Fix the manifest; do NOT assemble a partial video.\" >&2; exit 1\nfi\n[[ -z \"$BLOCKS\" ]] && { echo \"ERROR: --blocks N is REQUIRED — pass the expected beat count (a dropped beat must FAIL, not ship; short runs included).\" >&2; exit 1; }\n[[ -n \"$MUSIC\" && ! -f \"$MUSIC\" ]] && { echo \"ERROR: --music file not found: $MUSIC\" >&2; exit 1; }\n\nTMP=\"$(mktemp -d)\"; trap 'rm -rf \"$TMP\"' EXIT\nPBJSON=() # per-beat sidecar entries (machine-readable assembly manifest)\nimg0=\"${ARGS[0]%%|*}\" # first image (strip the X2 \"|imgB\" part if line 1 is a triple)\nDIMS=\"$(ffprobe -v error -select_streams v:0 -show_entries stream=width,height -of csv=p=0 \"$img0\")\"\nW=\"${DIMS%%,*}\"; H=\"${DIMS##*,}\"\nW=$(( (W/2)*2 )); H=$(( (H/2)*2 ))\nAR_W=\"$W\"; AR_H=\"$H\" # the video's aspect — every slide image must match it\nif (( W > 1920 || H > 1920 )); then\n read -r W H \u003c\u003c\u003c \"$(awk -v w=\"$W\" -v h=\"$H\" 'BEGIN{ m=(w>h)?w:h; s=1920/m; w2=int(w*s/2)*2; h2=int(h*s/2)*2; print w2, h2 }')\"\n echo \" output capped to ${W}x${H} (1080p-class; long side \u003c=1920)\" >&2\nfi\nFPS=30\necho \"[1/3] ${pairs} slides -> ${W}x${H} @ ${FPS}fps; slide = its take's length (+${BREATH}s breath, min ${MINSLIDE}s)\" >&2\n\nLIST=\"$TMP/list.txt\"; : > \"$LIST\"\nBOUNDS=\"$TMP/bounds.txt\"; : > \"$BOUNDS\"\nTCUM=0\nfor ((i=0;i\u003cpairs;i++)); do\n imgspec=\"${ARGS[i*2]}\"; voice=\"${ARGS[i*2+1]}\"\n imgA=\"${imgspec%%|*}\"; imgB=\"\"\n [[ \"$imgspec\" == *\"|\"* ]] && imgB=\"${imgspec##*|}\"\n [[ -f \"$imgA\" ]] || { echo \"ERROR: image not found: $imgA\" >&2; exit 1; }\n [[ -n \"$imgB\" && ! -f \"$imgB\" ]] && { echo \"ERROR: image not found: $imgB\" >&2; exit 1; }\n [[ -f \"$voice\" ]] || { echo \"ERROR: voice not found: $voice\" >&2; exit 1; }\n cn=\"$(basename \"$imgA\" | sed 's/\\.[^.]*$//' | grep -Eo '[0-9]+' | tail -1 || true)\"\n vn=\"$(basename \"$voice\" | sed 's/\\.[^.]*$//' | grep -Eo '[0-9]+' | tail -1 || true)\"\n if [[ -z \"$imgB\" && -n \"$cn\" && -n \"$vn\" ]] && (( 10#$cn != 10#$vn )); then\n if [[ \"$ALLOWMM\" == \"1\" ]]; then\n echo \"WARN: beat $((i+1)) mixes numbers: $(basename \"$imgA\") + $(basename \"$voice\") (--allow-mismatch).\" >&2\n else\n echo \"ERROR: beat $((i+1)) mixes numbers: $(basename \"$imgA\") + $(basename \"$voice\") — the voice would land on the WRONG picture. Fix the manifest order (or pass --allow-mismatch only for truly unnumbered files).\" >&2\n exit 1\n fi\n fi\n for im in \"$imgA\" ${imgB:+\"$imgB\"}; do\n ID=\"$(ffprobe -v error -select_streams v:0 -show_entries stream=width,height -of csv=p=0 \"$im\")\"\n IW=\"${ID%%,*}\"; IH=\"${ID##*,}\"\n awk -v w=\"$IW\" -v h=\"$IH\" -v W=\"$AR_W\" -v H=\"$AR_H\" 'BEGIN{ a=w/h; A=W/H; d=a-A; if(d\u003c0)d=-d; exit (d/A\u003c=0.02)?0:1 }' || {\n echo \"ERROR: slide image $(basename \"$im\") is ${IW}x${IH} — not the video aspect (${AR_W}x${AR_H} from slide 1). Asset sheets are refs, NEVER slides: regenerate this beat's FRAME in the chosen aspect_ratio.\" >&2\n exit 1; }\n done\n A=\"$(ffprobe -v error -show_entries format=duration -of csv=p=0 \"$voice\")\"\n DUR=\"$(awk -v a=\"$A\" -v br=\"$BREATH\" -v mn=\"$MINSLIDE\" 'BEGIN{d=a+br; if(d\u003cmn)d=mn; printf \"%.3f\", d}')\"\n HOLD=\"$DUR\"; [[ -n \"$imgB\" ]] && HOLD=\"$(awk -v d=\"$DUR\" 'BEGIN{printf \"%.3f\", d/2}')\"\n awk -v h=\"$HOLD\" -v m=\"$MAXHOLD\" 'BEGIN{exit (h>m)?0:1}' && {\n echo \"ERROR: beat $((i+1)) holds one frame ${HOLD}s (max ${MAXHOLD}s) — slideshow feel. Split the beat (X2 pair / micro-variation) or tighten the phrase; long holds do not ship.\" >&2\n exit 1; }\n seg=\"$TMP/s_$(printf '%03d' \"$i\").mp4\"\n VFILT=\"scale=${W}:${H}:force_original_aspect_ratio=decrease,pad=${W}:${H}:(ow-iw)/2:(oh-ih)/2:black,fps=${FPS},format=yuv420p,setsar=1\"\n if [[ -n \"$imgB\" ]]; then\n D1=\"$(awk -v d=\"$DUR\" 'BEGIN{printf \"%.3f\", d/2}')\"\n D2=\"$(awk -v d=\"$DUR\" -v h=\"$D1\" 'BEGIN{printf \"%.3f\", d-h}')\"\n ffmpeg -y -loglevel error -loop 1 -t \"$D1\" -i \"$imgA\" -loop 1 -t \"$D2\" -i \"$imgB\" -i \"$voice\" \\\n -filter_complex \"[0:v]${VFILT}[va];[1:v]${VFILT}[vb];[va][vb]concat=n=2:v=1:a=0[v];[2:a]aformat=sample_rates=44100:channel_layouts=stereo,apad[a]\" \\\n -map \"[v]\" -map \"[a]\" -t \"$DUR\" -c:v libx264 -preset veryfast -crf 20 -c:a aac -b:a 192k \"$seg\"\n printf ' beat %03d: take %.2fs -> slide %ss (TWO frames, cut at %ss) (at %ss)\\n' \"$((i+1))\" \"$A\" \"$DUR\" \"$D1\" \"$TCUM\" >&2\n else\n ffmpeg -y -loglevel error -loop 1 -i \"$imgA\" -i \"$voice\" \\\n -filter_complex \"[0:v]${VFILT}[v];[1:a]aformat=sample_rates=44100:channel_layouts=stereo,apad[a]\" \\\n -map \"[v]\" -map \"[a]\" -t \"$DUR\" -c:v libx264 -preset veryfast -crf 20 -c:a aac -b:a 192k \"$seg\"\n printf ' beat %03d: take %.2fs -> slide %ss (at %ss)\\n' \"$((i+1))\" \"$A\" \"$DUR\" \"$TCUM\" >&2\n fi\n echo \"file '$seg'\" >> \"$LIST\"\n TEND=\"$(awk -v c=\"$TCUM\" -v d=\"$DUR\" 'BEGIN{printf \"%.3f\", c+d}')\"\n printf '%s %s %s\\n' \"$((i+1))\" \"$TCUM\" \"$TEND\" >> \"$BOUNDS\"\n PBJSON[i]=\"$(printf '{\"n\":%d,\"images\":\"%s\",\"voice\":\"%s\",\"take_s\":%.2f,\"slide_s\":%s,\"start_s\":%s,\"x2\":%s}' \\\n \"$((i+1))\" \"$(basename \"$imgA\")$([[ -n \"$imgB\" ]] && echo \"+$(basename \"$imgB\")\")\" \"$(basename \"$voice\")\" \\\n \"$A\" \"$DUR\" \"$TCUM\" \"$([[ -n \"$imgB\" ]] && echo true || echo false)\")\"\n TCUM=\"$TEND\"\ndone\n\nVOTRACK=\"$TMP/joined.mp4\"\necho \"[2/3] concat ${pairs} slides -> single track\" >&2\nffmpeg -y -loglevel error -f concat -safe 0 -i \"$LIST\" -c:v libx264 -preset veryfast -crf 20 -c:a aac -movflags +faststart \"$VOTRACK\"\n\nQUIET=\"$(ffmpeg -i \"$VOTRACK\" -af \"silencedetect=noise=-18dB:d=0.9\" -f null - 2>&1 | grep -Eo 'silence_(start|end): *[0-9.]+' || true)\"\nEMPTY=\"$(awk -v BF=\"$BOUNDS\" '\n BEGIN { while ((getline line \u003c BF) > 0) { split(line,b,\" \"); idx[++nb]=b[1]; st[nb]=b[2]; en[nb]=b[3] } }\n /silence_start/ { s[++k]=$NF+0; next }\n /silence_end/ { e[k]=$NF+0 }\n END {\n for (i=1;i\u003c=nb;i++) {\n ws=st[i]+0.15; we=en[i]-0.35; if (we\u003c=ws) continue;\n for (j=1;j\u003c=k;j++) { ee=(e[j]>0)?e[j]:1e9;\n if (s[j]\u003c=ws && ee>=we) { printf \"%s%s\", (out++?\",\":\"\"), idx[i]; break } } } }' \u003c\u003c\u003c \"$QUIET\")\"\nif [[ -n \"$EMPTY\" ]]; then\n echo \"ERROR: beats [$EMPTY] have NO narration in their segments — a voice file is silent/mispaired. Fix the inputs and re-run; do NOT deliver.\" >&2\n exit 1\nfi\necho \" narration present in all ${pairs} segments\" >&2\n\necho \"[3/3] finalize -> $OUT\" >&2\nln2_args() {\n ffmpeg -hide_banner -i \"$1\" -af \"loudnorm=I=-16:TP=-1.5:LRA=11:print_format=json\" -f null - 2>&1 \\\n | awk '/\"input_i\"/{i=$3}/\"input_tp\"/{t=$3}/\"input_lra\"/{l=$3}/\"input_thresh\"/{h=$3}/\"target_offset\"/{o=$3}\n {gsub(/[\",]/,\"\",i);gsub(/[\",]/,\"\",t);gsub(/[\",]/,\"\",l);gsub(/[\",]/,\"\",h);gsub(/[\",]/,\"\",o)}\n END{printf \"measured_I=%s:measured_TP=%s:measured_LRA=%s:measured_thresh=%s:offset=%s\", i,t,l,h,o}'\n}\nif [[ -n \"$MUSIC\" ]]; then\n MIXED=\"$TMP/mixed.mp4\"\n ffmpeg -y -loglevel error -i \"$VOTRACK\" -stream_loop -1 -i \"$MUSIC\" \\\n -filter_complex \"[0:a]aformat=sample_rates=44100:channel_layouts=stereo,asplit=2[vo][key];[1:a]aformat=sample_rates=44100:channel_layouts=stereo,volume=${MVOL}[m];[m][key]sidechaincompress=threshold=0.02:ratio=8:attack=10:release=300[duck];[vo][duck]amix=inputs=2:duration=first:normalize=0[a]\" \\\n -map 0:v -map \"[a]\" -shortest -c:v copy -c:a aac -b:a 192k \"$MIXED\"\n SRC=\"$MIXED\"\nelse\n SRC=\"$VOTRACK\"\nfi\nLNARGS=\"$(ln2_args \"$SRC\")\"\nif [[ \"$LNARGS\" == *measured_I=-* ]]; then\n ffmpeg -y -loglevel error -i \"$SRC\" -af \"loudnorm=I=-16:TP=-1.5:LRA=11:${LNARGS}:linear=true\" -c:v copy -c:a aac -b:a 192k \"$OUT\"\nelse\n echo \"WARN: loudness measurement failed — falling back to single-pass loudnorm.\" >&2\n ffmpeg -y -loglevel error -i \"$SRC\" -af \"loudnorm=I=-16:TP=-1.5:LRA=11\" -c:v copy -c:a aac -b:a 192k \"$OUT\"\nfi\n\nFDUR=\"$(ffprobe -v error -show_entries format=duration -of csv=p=0 \"$OUT\")\"\nawk -v d=\"$FDUR\" -v e=\"$TCUM\" 'BEGIN{x=d-e; if(x\u003c0)x=-x; exit (x\u003c=1.5)?0:1}' || {\n echo \"ERROR: final duration ${FDUR}s != expected ${TCUM}s — do NOT deliver; investigate the inputs.\" >&2; exit 1; }\nADUR=\"$(ffprobe -v error -select_streams a:0 -show_entries stream=duration -of csv=p=0 \"$OUT\" | head -1)\"\n[[ -z \"$ADUR\" || \"$ADUR\" == \"N/A\" ]] && { echo \"ERROR: final has no readable audio stream — do NOT deliver.\" >&2; exit 1; }\nDERR=\"$( (ffmpeg -v error -xerror -i \"$OUT\" -f null - ) 2>&1 | head -3 || true)\"\n[[ -n \"$DERR\" ]] && { echo \"ERROR: final failed decode validation — corrupted stream; do NOT deliver:\" >&2; echo \"$DERR\" >&2; exit 1; }\n\n\nPOSTER=\"${OUT%.mp4}_poster.jpg\"\nffmpeg -y -loglevel error -ss 1 -i \"$OUT\" -frames:v 1 -q:v 2 \"$POSTER\" || true\nif [[ ! -s \"$POSTER\" ]]; then\n ffmpeg -y -loglevel error -i \"$OUT\" -frames:v 1 -q:v 2 \"$POSTER\" || true\nfi\n[[ -s \"$POSTER\" ]] && echo \"POSTER: $POSTER\" >&2 || echo \"WARN: poster extraction failed (no $POSTER)\" >&2\n\nSIDE=\"${OUT}.assembly.json\"\n{\n printf '{\"script\":\"assemble_slides.sh\",\"out\":\"%s\",\"beats\":%d,\"total_s\":%s,\"actual_s\":%s' \\\n \"$(basename \"$OUT\")\" \"$pairs\" \"$TCUM\" \"$FDUR\"\n if [[ -n \"$REQUESTED_SECONDS\" ]]; then\n printf ',\"requested_s\":%d,\"blocks\":%d' \"$REQUESTED_SECONDS\" \"$DURATION_BLOCKS\"\n fi\n printf ',\"width\":%s,\"height\":%s,\"fps\":\"%s\",\"breath\":%s,\"min_slide\":%s,\"music\":%s,\"music_vol\":%s,\"gate\":\"slide = take duration + breath (min slide), all slides in the video aspect (asset sheets rejected), 1080p-class cap, narration in every segment (silencedetect -18dB), 2-pass linear loudnorm -16, duration match, full decode\",\"per_beat\":[' \\\n \"$W\" \"$H\" \"$FPS\" \"$BREATH\" \"$MINSLIDE\" \\\n \"$([[ -n \"$MUSIC\" ]] && echo true || echo false)\" \"$MVOL\"\n for ((i=0;i\u003cpairs;i++)); do printf '%s%s' \"${PBJSON[i]}\" \"$([[ $i -lt $((pairs-1)) ]] && echo ,)\"; done\n printf '],\"ts\":\"%s\"}\\n' \"$(date -u +%Y-%m-%dT%H:%M:%SZ)\"\n} > \"$SIDE\"\necho \"ASSEMBLY-SIDECAR: $SIDE\" >&2\necho \"DONE: $OUT (ONE file, ${pairs} slides, ${TCUM}s total, actual ${FDUR}s; decode-validated)\" >&2\n","content_type":"application/x-sh; charset=utf-8","language":"bash","size":26736,"content_sha256":"096a0b6564d94e2751e78d0991c3546c81c73e758a4ecae401c00f9c21b903fb"},{"filename":"scripts/audio_to_captions.py","content":"#!/usr/bin/env python3\n\"\"\"Build timed captions using authored text and Whisper's word clock.\n\nWhisper supplies timing only when a script manifest is available. Caption text\nis aligned back to the exact ``vo_line``/``phrase`` values, preventing\ntranscription substitutions from being burned into generated videos.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport difflib\nimport json\nimport math\nimport os\nimport re\nimport sys\nfrom pathlib import Path\nfrom typing import Any\n\n\nNORMALIZE_PATTERN = re.compile(r\"[^\\w]+\", re.UNICODE)\nTOKEN_PATTERN = re.compile(r\"[^\\W_]+(?:[-'’][^\\W_]+)*[.,!?;:]?\", re.UNICODE)\nNONVERBAL_DIRECTION_PATTERN = re.compile(r\"\\[[^\\]]+\\]\")\n\n\ndef _normalize(value: str) -> str:\n return NORMALIZE_PATTERN.sub(\"\", value.casefold())\n\n\ndef _load_authored_rows(script_path: Path) -> list[list[str]]:\n \"\"\"Authored tokens, kept GROUPED BY ROW (one row = one block or beat).\n\n The grouping is what lets the per-block path align each line inside its own\n block window: a word the STT swallowed can then only be interpolated across\n that block's speech, never across the silence between two blocks.\n \"\"\"\n payload = json.loads(script_path.read_text(encoding=\"utf-8\"))\n if not isinstance(payload, dict):\n raise ValueError(\"script manifest must be an object\")\n\n rows = payload.get(\"beats\")\n fields = (\"phrase\", \"vo_line\")\n if not isinstance(rows, list):\n rows = payload.get(\"blocks\")\n fields = (\"vo_line\",)\n if not isinstance(rows, list) or not rows:\n raise ValueError(\"script manifest must contain blocks or beats\")\n\n grouped: list[list[str]] = []\n for index, row in enumerate(rows, start=1):\n if not isinstance(row, dict):\n raise ValueError(f\"script row {index} must be an object\")\n text = next(\n (\n row[field].strip()\n for field in fields\n if isinstance(row.get(field), str) and row[field].strip()\n ),\n \"\",\n )\n if not text:\n raise ValueError(f\"script row {index} has no authored narration\")\n spoken_text = NONVERBAL_DIRECTION_PATTERN.sub(\"\", text)\n grouped.append(TOKEN_PATTERN.findall(spoken_text))\n if not any(grouped):\n raise ValueError(\"script manifest contains no authored narration words\")\n return grouped\n\n\ndef _load_authored_tokens(script_path: Path) -> list[str]:\n return [token for row in _load_authored_rows(script_path) for token in row]\n\n\ndef _timed_script_span(\n script_tokens: list[str],\n *,\n start: float,\n end: float,\n) -> list[dict[str, Any]]:\n if not script_tokens:\n return []\n duration = max(end - start, 0.05 * len(script_tokens))\n weights = [max(len(_normalize(token)), 1) for token in script_tokens]\n total_weight = sum(weights)\n timed: list[dict[str, Any]] = []\n cursor = start\n for index, (token, weight) in enumerate(zip(script_tokens, weights)):\n token_end = (\n end\n if index == len(script_tokens) - 1\n else cursor + duration * weight / total_weight\n )\n timed.append(\n {\n \"word\": token,\n \"start\": round(cursor, 3),\n \"end\": round(max(token_end, cursor + 0.01), 3),\n }\n )\n cursor = token_end\n return timed\n\n\ndef align_words_to_script(\n words: list[dict[str, Any]],\n script_tokens: list[str],\n *,\n minimum_similarity: float = 0.55,\n span_floor: float = 0.0,\n span_ceiling: float | None = None,\n) -> tuple[list[dict[str, Any]], float]:\n \"\"\"Return exact script tokens carrying timings derived from Whisper words.\n\n `span_floor` / `span_ceiling` bound where an INTERPOLATED span may live when\n the STT dropped words at the very start or end of this text (VAD eats quiet\n openers like \"you know?\"). Without them a dropped opener is spread from 0.0 —\n or from the previous block's last word — and the caption shows up seconds\n before it is spoken. Callers that align one block at a time pass that block's\n speech window.\n \"\"\"\n whisper_tokens = [_normalize(str(word[\"word\"])) for word in words]\n normalized_script_tokens = [_normalize(token) for token in script_tokens]\n matcher = difflib.SequenceMatcher(\n a=whisper_tokens,\n b=normalized_script_tokens,\n autojunk=False,\n )\n similarity = matcher.ratio()\n if similarity \u003c minimum_similarity:\n raise ValueError(\n f\"Whisper/script similarity {similarity:.3f} is below \"\n f\"{minimum_similarity:.3f}; refusing to burn unverified captions\"\n )\n\n aligned: list[dict[str, Any]] = []\n for tag, whisper_start, whisper_end, script_start, script_end in matcher.get_opcodes():\n authored_span = script_tokens[script_start:script_end]\n if not authored_span:\n continue\n\n if tag == \"equal\":\n for offset, token in enumerate(authored_span):\n timed_word = words[whisper_start + offset]\n aligned.append(\n {\n \"word\": token,\n \"start\": float(timed_word[\"start\"]),\n \"end\": float(timed_word[\"end\"]),\n }\n )\n continue\n\n if whisper_end > whisper_start:\n span_start = float(words[whisper_start][\"start\"])\n span_end = float(words[whisper_end - 1][\"end\"])\n else:\n previous_end = (\n float(words[whisper_start - 1][\"end\"])\n if whisper_start > 0\n else span_floor\n )\n next_start = (\n float(words[whisper_start][\"start\"])\n if whisper_start \u003c len(words)\n else previous_end + 0.05 * len(authored_span)\n )\n span_start = max(previous_end, span_floor)\n span_end = max(next_start, span_start + 0.05 * len(authored_span))\n if span_ceiling is not None:\n span_end = min(span_end, max(span_ceiling, span_start + 0.01))\n aligned.extend(\n _timed_script_span(\n authored_span,\n start=span_start,\n end=span_end,\n )\n )\n\n if [_normalize(str(word[\"word\"])) for word in aligned] != normalized_script_tokens:\n raise ValueError(\"caption alignment did not preserve the complete authored script\")\n return aligned, similarity\n\n\ndef group_captions(\n words: list[dict[str, Any]],\n *,\n max_words: int = 5,\n max_gap: float = 0.5,\n max_chars: int = 32,\n min_duration: float = 0.0,\n max_cps: float = 0.0,\n) -> list[dict[str, Any]]:\n if not all(math.isfinite(value) and value >= 0 for value in (min_duration, max_cps)):\n raise ValueError(\"caption readability limits must be finite and nonnegative\")\n if min_duration or max_cps:\n return _readable_captions(\n words,\n max_words=max_words,\n max_gap=max_gap,\n max_chars=max_chars,\n min_duration=min_duration,\n max_cps=max_cps,\n )\n captions: list[dict[str, Any]] = []\n current: list[dict[str, Any]] = []\n\n def _flush() -> None:\n if not current:\n return\n text = \" \".join(str(word[\"word\"]).strip() for word in current).strip()\n text = re.sub(r\"\\s+([,.!?;:])\", r\"\\1\", text)\n captions.append(\n {\n \"text\": text,\n \"start\": round(float(current[0][\"start\"]), 2),\n \"end\": round(float(current[-1][\"end\"]), 2),\n }\n )\n current.clear()\n\n for word in words:\n if current:\n gap = float(word[\"start\"]) - float(current[-1][\"end\"])\n projected_text = \" \".join(\n [*(str(item[\"word\"]) for item in current), str(word[\"word\"])]\n )\n if (\n gap >= max_gap\n or len(current) >= max_words\n or len(projected_text) > max_chars\n ):\n _flush()\n current.append(word)\n if re.search(r\"[.!?]$\", str(word[\"word\"]).strip()):\n _flush()\n _flush()\n return captions\n\n\ndef _readable_captions(\n words: list[dict[str, Any]],\n *,\n max_words: int,\n max_gap: float,\n max_chars: int,\n min_duration: float,\n max_cps: float,\n) -> list[dict[str, Any]]:\n \"\"\"Partition speech on existing word boundaries, never by stretching cues.\n\n A greedy four-word cut can strand a 0.26s final word. Look ahead within\n each sentence/pause-delimited span to find a readable partition instead.\n Keep the fewest cues; on ties prefer the longer opening phrase.\n \"\"\"\n if max_words \u003c 1 or max_chars \u003c 1 or not math.isfinite(max_gap) or max_gap \u003c= 0:\n raise ValueError(\"caption word, character and gap limits must be positive\")\n spans: list[list[dict[str, Any]]] = []\n for word in words:\n start, end = float(word[\"start\"]), float(word[\"end\"])\n if not math.isfinite(start) or not math.isfinite(end) or start \u003c 0 or end \u003c start:\n raise ValueError(\"caption word timings must be finite, ordered and nonnegative\")\n if not spans or (\n start - float(spans[-1][-1][\"end\"]) >= max_gap\n or re.search(r\"[.!?]$\", str(spans[-1][-1][\"word\"]).strip())\n ):\n spans.append([])\n spans[-1].append(word)\n\n captions: list[dict[str, Any]] = []\n for span in spans:\n count = len(span)\n costs = [count + 1] * count + [0]\n choices: dict[int, tuple[int, dict[str, Any]]] = {}\n for start_index in range(count - 1, -1, -1):\n for stop in range(start_index + 1, min(count, start_index + max_words) + 1):\n text = \" \".join(str(word[\"word\"]).strip() for word in span[start_index:stop])\n text = re.sub(r\"\\s+([,.!?;:])\", r\"\\1\", text)\n if len(text) > max_chars:\n break\n start = round(float(span[start_index][\"start\"]), 2)\n end = round(float(span[stop - 1][\"end\"]), 2)\n duration = end - start\n if duration \u003c= 0 or duration + 1e-6 \u003c min_duration:\n continue\n if max_cps and len(text) > max_cps * duration + 1e-6:\n continue\n if costs[stop] \u003c count + 1 and 1 + costs[stop] \u003c= costs[start_index]:\n costs[start_index] = 1 + costs[stop]\n choices[start_index] = (stop, {\"text\": text, \"start\": start, \"end\": end})\n if 0 not in choices:\n raise ValueError(\n f\"no readable caption grouping for speech at {span[0]['start']}s \"\n f\"under {max_words} words/{max_chars} chars, {min_duration}s minimum \"\n f\"and {max_cps or 'unlimited'} chars/s; inspect the word timings \"\n \"or speech pace, do not invent times or extend through a pause\"\n )\n index = 0\n while index \u003c count:\n index, caption = choices[index]\n captions.append(caption)\n return captions\n\n\ndef _timestamp(seconds: float) -> str:\n total_milliseconds = int(round(seconds * 1000))\n hours, remaining_milliseconds = divmod(total_milliseconds, 3_600_000)\n minutes, remaining_milliseconds = divmod(remaining_milliseconds, 60_000)\n whole_seconds, milliseconds = divmod(remaining_milliseconds, 1000)\n return (\n f\"{hours:02d}:{minutes:02d}:{whole_seconds:02d},\"\n f\"{milliseconds:03d}\"\n )\n\n\ndef to_srt(captions: list[dict[str, Any]]) -> str:\n return \"\\n\".join(\n f\"{index}\\n{_timestamp(caption['start'])} --> \"\n f\"{_timestamp(caption['end'])}\\n{caption['text']}\\n\"\n for index, caption in enumerate(captions, start=1)\n )\n\n\ndef _words_faster(\n path: Path,\n model_size: str,\n *,\n language: str | None = None,\n) -> list[dict[str, Any]]:\n from faster_whisper import WhisperModel\n\n model = WhisperModel(model_size, device=\"cpu\", compute_type=\"int8\", local_files_only=True)\n # vad_filter drops music/room-tone stretches instead of hallucinating words\n # over them; condition_on_previous_text=False stops one bad segment from\n # dragging the rest of the transcript out of sync (the classic caption\n # drift). beam_size=5 recovers words a greedy pass swallows.\n segments, _info = model.transcribe(\n str(path),\n word_timestamps=True,\n vad_filter=True,\n condition_on_previous_text=False,\n beam_size=5,\n language=language,\n )\n words: list[dict[str, Any]] = []\n for segment in segments:\n for word in segment.words or []:\n words.append(\n {\n \"word\": str(word.word).strip(),\n \"start\": float(word.start),\n \"end\": float(word.end),\n }\n )\n return words\n\n\ndef _speech_only(path: Path) -> Path | None:\n \"\"\"Voice-band copy of a MIXED track (video with music + SFX).\n\n Whisper loses words and drifts when a bed and diegetic SFX sit under the\n narration. Band-passing to the voice range and levelling first recovers\n most of them. Returns None when ffmpeg is unavailable — callers fall back\n to the original file.\n \"\"\"\n import shutil\n import subprocess\n import tempfile\n\n if shutil.which(\"ffmpeg\") is None:\n return None\n target = Path(tempfile.mkdtemp(prefix=\"caps_\")) / \"speech.wav\"\n command = [\n \"ffmpeg\",\n \"-y\",\n \"-loglevel\",\n \"error\",\n \"-i\",\n str(path),\n \"-vn\",\n \"-af\",\n \"highpass=f=180,lowpass=f=3800,dynaudnorm=f=250:g=15\",\n \"-ac\",\n \"1\",\n \"-ar\",\n \"16000\",\n str(target),\n ]\n try:\n subprocess.run(command, check=True, capture_output=True)\n except (OSError, subprocess.CalledProcessError):\n shutil.rmtree(target.parent)\n return None\n if target.is_file() and target.stat().st_size > 0:\n return target\n shutil.rmtree(target.parent)\n return None\n\n\ndef get_words(\n path: Path,\n model_size: str,\n *,\n language: str | None = None,\n mixed_input: bool = False,\n) -> tuple[list[dict[str, Any]], str]:\n source = path\n if mixed_input:\n cleaned = _speech_only(path)\n if cleaned is not None:\n source = cleaned\n print(\"(band-passed the mixed track before STT)\", file=sys.stderr)\n try:\n return _words_faster(source, model_size, language=language), \"faster-whisper\"\n finally:\n if source != path:\n import shutil\n shutil.rmtree(source.parent)\n\n\ndef _load_blocks(sidecar_path: Path) -> list[dict[str, Any]]:\n payload = json.loads(sidecar_path.read_text(encoding=\"utf-8\"))\n blocks = payload.get(\"per_block\")\n if not isinstance(blocks, list) or not blocks:\n raise ValueError(\"assembly sidecar has no per_block array\")\n timed = [\n block\n for block in blocks\n if isinstance(block, dict)\n and isinstance(block.get(\"voice\"), str)\n and block[\"voice\"]\n and float(block.get(\"speech_s\") or 0) > 0\n ]\n if not timed:\n raise ValueError(\n \"assembly sidecar carries no voiced blocks (song mode has no narration \"\n \"to caption)\"\n )\n for block in timed:\n if \"speech_abs_s\" not in block:\n raise ValueError(\n \"assembly sidecar predates speech_abs_s — re-run the assembler \"\n \"so captions can be timed from the clean voice files\"\n )\n return timed\n\n\ndef words_from_blocks(\n blocks: list[dict[str, Any]],\n voice_dir: Path,\n model_size: str,\n *,\n language: str | None = None,\n) -> list[dict[str, Any]]:\n \"\"\"Time words on the CLEAN per-block voice files, then shift into the\n finished timeline: t_final = speech_abs_s + (t_word - lead_silence_s).\n\n This is the accurate path: no music or SFX under the narration, and the\n offsets come from the assembler's own centring math instead of being\n re-estimated from the mix.\n \"\"\"\n groups: list[dict[str, Any]] = []\n for block in blocks:\n voice_path = voice_dir / str(block[\"voice\"])\n if not voice_path.is_file():\n raise ValueError(f\"voice file missing for block {block.get('n')}: {voice_path}\")\n offset = float(block[\"speech_abs_s\"])\n lead = float(block.get(\"lead_silence_s\") or 0.0)\n block_words = _words_faster(voice_path, model_size, language=language)\n if not block_words:\n raise ValueError(\n f\"no words transcribed for block {block.get('n')} ({voice_path.name})\"\n )\n shifted: list[dict[str, Any]] = []\n for word in block_words:\n start = offset + (float(word[\"start\"]) - lead)\n end = offset + (float(word[\"end\"]) - lead)\n shifted.append(\n {\n \"word\": word[\"word\"],\n \"start\": round(max(start, 0.0), 3),\n \"end\": round(max(end, start + 0.01), 3),\n }\n )\n shifted.sort(key=lambda word: float(word[\"start\"]))\n speech_seconds = float(block.get(\"speech_s\") or 0.0)\n speech_end = (\n offset + speech_seconds\n if speech_seconds > 0\n else float(shifted[-1][\"end\"])\n )\n groups.append(\n {\n \"n\": block.get(\"n\"),\n \"words\": shifted,\n \"speech_start\": round(offset, 3),\n \"speech_end\": round(max(speech_end, float(shifted[-1][\"end\"])), 3),\n }\n )\n return groups\n\n\ndef align_blocks_to_script(\n groups: list[dict[str, Any]],\n rows: list[list[str]],\n *,\n minimum_similarity: float,\n) -> tuple[list[dict[str, Any]], float]:\n \"\"\"Align the authored text ONE BLOCK AT A TIME, inside that block's window.\n\n Aligning the whole timeline at once lets a missing opener borrow the silence\n between two blocks, so the caption fires before the speech. Per block, an\n interpolated span is clamped to `[speech_start, speech_end]` of the block that\n actually contains those words.\n \"\"\"\n aligned: list[dict[str, Any]] = []\n weighted_similarity = 0.0\n token_count = 0\n worst_similarity = 1.0\n worst_block: Any = None\n if len(groups) != len(rows):\n raise ValueError(\"script and audio block counts differ\")\n for group, tokens in zip(groups, rows):\n if not tokens:\n raise ValueError(f\"block {group['n']} has no authored narration\")\n try:\n block_aligned, block_similarity = align_words_to_script(\n group[\"words\"],\n tokens,\n minimum_similarity=minimum_similarity,\n span_floor=float(group[\"speech_start\"]),\n span_ceiling=float(group[\"speech_end\"]),\n )\n except ValueError as error:\n raise ValueError(f\"block {group['n']}: {error}\") from error\n aligned.extend(block_aligned)\n weighted_similarity += block_similarity * len(tokens)\n token_count += len(tokens)\n if block_similarity \u003c worst_similarity:\n worst_similarity = block_similarity\n worst_block = group[\"n\"]\n if worst_block is not None and worst_similarity \u003c 0.9:\n print(\n f\"WARN: block {worst_block} matched only {worst_similarity:.2%} of its \"\n \"authored line — that block's word timings are partly interpolated \"\n \"(they stay inside its own speech window). Consider --model medium.\",\n file=sys.stderr,\n )\n return aligned, (weighted_similarity / token_count if token_count else 1.0)\n\n\ndef _resolve_script_path(explicit_path: Path | None) -> Path | None:\n if explicit_path is not None:\n return explicit_path\n default_path = Path(\"script_manifest.json\")\n return default_path if default_path.is_file() else None\n\n\ndef main() -> int:\n parser = argparse.ArgumentParser()\n parser.add_argument(\"audio\", type=Path)\n parser.add_argument(\"--json\", type=Path)\n parser.add_argument(\"--srt\", type=Path)\n parser.add_argument(\n \"--script\",\n type=Path,\n help=(\n \"Authored script manifest. Defaults to ./script_manifest.json \"\n \"when present.\"\n ),\n )\n parser.add_argument(\n \"--per-block\",\n type=Path,\n help=(\n \"final.mp4.assembly.json — time the words on the CLEAN per-block \"\n \"voice files and shift them by the assembler's own offsets. The \"\n \"accurate path: no music/SFX under the narration.\"\n ),\n )\n parser.add_argument(\n \"--voice-dir\",\n type=Path,\n default=Path(\".\"),\n help=\"Directory holding the voiceNN.wav files named in the sidecar.\",\n )\n parser.add_argument(\n \"--language\",\n help=\"Language hint (e.g. en, ru). Stops the STT guessing on short takes.\",\n )\n parser.add_argument(\n \"--mixed\",\n action=\"store_true\",\n help=(\n \"The input carries music/SFX under the speech — band-pass to the \"\n \"voice range before STT. Implied when captioning a finished video.\"\n ),\n )\n parser.add_argument(\"--max-words\", type=int, default=5)\n parser.add_argument(\"--max-chars\", type=int, default=32)\n parser.add_argument(\"--max-gap\", type=float, default=0.4)\n parser.add_argument(\n \"--min-duration\", type=float, default=0.0,\n help=\"Minimum readable cue seconds; regroup using word timings (0 disables).\",\n )\n parser.add_argument(\n \"--max-cps\", type=float, default=0.0,\n help=\"Maximum characters per second per cue (0 disables).\",\n )\n parser.add_argument(\"--model\", default=\"small\")\n parser.add_argument(\"--minimum-similarity\", type=float, default=0.75)\n arguments = parser.parse_args()\n\n groups: list[dict[str, Any]] | None = None\n if arguments.per_block is not None:\n try:\n blocks = _load_blocks(arguments.per_block)\n groups = words_from_blocks(\n blocks,\n arguments.voice_dir,\n arguments.model,\n language=arguments.language,\n )\n words = [word for group in groups for word in group[\"words\"]]\n except (OSError, json.JSONDecodeError, ValueError) as error:\n print(f\"ERROR: per-block timing failed: {error}\", file=sys.stderr)\n return 1\n provider = \"faster-whisper/per-block\"\n else:\n words, provider = get_words(\n arguments.audio,\n arguments.model,\n language=arguments.language,\n mixed_input=arguments.mixed,\n )\n\n script_path = _resolve_script_path(arguments.script)\n similarity: float | None = None\n if script_path is not None:\n try:\n rows = _load_authored_rows(script_path)\n if groups is not None and len(rows) == len(groups):\n # Per-block: each line is aligned inside its own speech window.\n words, similarity = align_blocks_to_script(\n groups,\n rows,\n minimum_similarity=arguments.minimum_similarity,\n )\n else:\n if groups is not None:\n raise ValueError(\"script and audio block counts differ\")\n words, similarity = align_words_to_script(\n words,\n [token for row in rows for token in row],\n minimum_similarity=arguments.minimum_similarity,\n )\n except (OSError, json.JSONDecodeError, ValueError) as error:\n print(f\"ERROR: authored caption alignment failed: {error}\", file=sys.stderr)\n return 1\n if similarity is not None and similarity \u003c 0.9:\n print(\n f\"WARN: the STT only matched {similarity:.2%} of the authored script — \"\n \"wording is correct (taken from the manifest) but some timings are \"\n \"interpolated. Prefer --per-block on clean voice files, or a larger \"\n \"--model, before shipping.\",\n file=sys.stderr,\n )\n\n try:\n captions = group_captions(\n words,\n max_words=arguments.max_words,\n max_gap=arguments.max_gap,\n max_chars=arguments.max_chars,\n min_duration=arguments.min_duration,\n max_cps=arguments.max_cps,\n )\n except ValueError as error:\n print(f\"ERROR: caption grouping failed: {error}\", file=sys.stderr)\n return 1\n\n # Coverage report: every timed word must survive into a caption, and a\n # suspiciously thin transcript (words per second of covered speech) means\n # the STT swallowed words — the caller must not ship that silently.\n caption_words = sum(len(str(caption[\"text\"]).split()) for caption in captions)\n span = (\n float(captions[-1][\"end\"]) - float(captions[0][\"start\"]) if captions else 0.0\n )\n density = caption_words / span if span > 0 else 0.0\n if caption_words != len(words):\n print(\n f\"ERROR: {len(words)} timed words but {caption_words} made it into \"\n \"captions — refusing to write a lossy caption file.\",\n file=sys.stderr,\n )\n return 1\n if script_path is None and span > 5 and density \u003c 1.6:\n print(\n f\"WARN: only {density:.2f} words/second over {span:.1f}s — the STT likely \"\n \"dropped words. Re-run with --model medium (and --language), or pass \"\n \"--script/--per-block, before burning.\",\n file=sys.stderr,\n )\n\n print(\n f\"provider={provider} words={len(words)} captions={len(captions)} \"\n f\"caption_words={caption_words} density={density:.2f}/s \"\n f\"script_aligned={script_path is not None} similarity={similarity}\",\n file=sys.stderr,\n )\n\n output = {\n \"words\": words,\n \"captions\": captions,\n \"script_aligned\": script_path is not None,\n \"similarity\": similarity,\n \"timed_words\": len(words),\n \"caption_words\": caption_words,\n \"words_per_second\": round(density, 2),\n \"provider\": provider,\n }\n serialized_output = json.dumps(output, indent=2, ensure_ascii=False)\n print(serialized_output)\n if arguments.json is not None:\n arguments.json.write_text(serialized_output, encoding=\"utf-8\")\n if arguments.srt is not None:\n arguments.srt.write_text(to_srt(captions), encoding=\"utf-8\")\n return 0\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":26796,"content_sha256":"8ae958107d5146dc358498423915d00e5195f98889ba6b53117b9b9baa174e9a"},{"filename":"scripts/bind_scene_frame_results.py","content":"#!/usr/bin/env python3\n\"\"\"Bind generated image job provenance to a deterministic scene timeline.\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport json\nfrom pathlib import Path\nfrom typing import Any\nfrom urllib.parse import urlparse\n\n\ndef _is_http_url(value: Any) -> bool:\n if not isinstance(value, str):\n return False\n parsed = urlparse(value)\n return parsed.scheme in {\"http\", \"https\"} and bool(parsed.netloc)\n\n\ndef _job_id(value: Any) -> str | None:\n if isinstance(value, str) and value.strip():\n return value.strip()\n return None\n\n\ndef _result_url(value: Any) -> str | None:\n if not isinstance(value, dict):\n return None\n for candidate in (\n value.get(\"url\"),\n value.get(\"result_url\"),\n value.get(\"preview\"),\n value.get(\"output_url\"),\n ):\n if _is_http_url(candidate):\n return candidate\n result = value.get(\"result\")\n if isinstance(result, dict):\n return _result_url(result)\n return None\n\n\ndef _records(payload: Any) -> list[dict[str, Any]]:\n if isinstance(payload, list):\n return [item for item in payload if isinstance(item, dict)]\n if not isinstance(payload, dict):\n return []\n records: list[dict[str, Any]] = []\n for field in (\n \"frames\",\n \"requests\",\n \"results\",\n \"generation_provenance\",\n ):\n value = payload.get(field)\n if isinstance(value, list):\n records.extend(item for item in value if isinstance(item, dict))\n return records\n\n\ndef _frame_results(payloads: list[Any]) -> dict[str, dict[str, Any]]:\n job_urls: dict[str, str] = {}\n job_references: dict[str, list[str]] = {}\n slots: dict[str, dict[str, Any]] = {}\n records = [record for payload in payloads for record in _records(payload)]\n for record in records:\n job_id = _job_id(record.get(\"job_id\") or record.get(\"id\"))\n url = _result_url(record)\n if job_id is not None and url is not None and record.get(\"status\") == \"completed\":\n job_urls[job_id] = url\n params = record.get(\"params\")\n if isinstance(params, dict) and isinstance(params.get(\"medias\"), list):\n job_references[job_id] = [\n media[\"data\"][\"id\"] for media in params[\"medias\"]\n if isinstance(media, dict) and media.get(\"role\") == \"image\"\n and isinstance(media.get(\"data\"), dict)\n and media[\"data\"].get(\"type\") == \"image_job\"\n and isinstance(media[\"data\"].get(\"id\"), str)\n ]\n\n for record in records:\n output_slot = record.get(\"output_slot\")\n if not isinstance(output_slot, str) or not output_slot.strip():\n continue\n job_ids = record.get(\"job_ids\")\n job_id = _job_id(record.get(\"job_id\") or record.get(\"id\"))\n if job_id is None and isinstance(job_ids, list) and len(job_ids) == 1:\n job_id = _job_id(job_ids[0])\n url = _result_url(record) or (job_urls.get(job_id) if job_id else None)\n if job_id is not None and url is not None and job_urls.get(job_id) == url:\n slots[output_slot.strip()] = {\"job_id\": job_id, \"url\": url,\n \"reference_job_ids\": job_references.get(job_id, [])}\n return slots\n\n\ndef bind_scene_frame_results(\n manifest: dict[str, Any],\n payloads: list[Any],\n) -> dict[str, Any]:\n frames = manifest.get(\"frames\")\n if manifest.get(\"manifest_version\") != 2 or not isinstance(frames, list):\n raise ValueError(\"scene manifest must contain manifest_version 2 frames\")\n results = _frame_results(payloads)\n bound_frames: list[dict[str, Any]] = []\n previous: dict[str, Any] | None = None\n for index, raw_frame in enumerate(frames, start=1):\n if not isinstance(raw_frame, dict):\n raise ValueError(f\"frame {index} must be an object\")\n output_slot = raw_frame.get(\"output_slot\")\n if output_slot != f\"frame-{index:03d}\":\n raise ValueError(f\"frame {index} output_slot is invalid\")\n result = results.get(output_slot)\n if result is None:\n raise ValueError(\n f\"frame {index} has no completed job_id/url result for {output_slot}\"\n )\n frame = {\n **raw_frame,\n \"frame_job_id\": result[\"job_id\"],\n \"frame_url\": result[\"url\"],\n }\n if frame.get(\"image_mode\") == \"variation\":\n if previous is None:\n raise ValueError(\"the first frame cannot be a variation\")\n if frame.get(\"reference_output_slot\") != previous.get(\"output_slot\"):\n raise ValueError(\n f\"frame {index} reference_output_slot must point to frame {index - 1}\"\n )\n if previous[\"frame_job_id\"] not in result[\"reference_job_ids\"]:\n raise ValueError(\n f\"frame {index} job receipt does not reference the previous image job; \"\n \"pass that job ID to generation and include the completed job receipt\"\n )\n frame[\"reference_frame_job_id\"] = previous[\"frame_job_id\"]\n frame[\"reference_frame_url\"] = previous[\"frame_url\"]\n else:\n frame.pop(\"reference_frame_job_id\", None)\n frame.pop(\"reference_frame_url\", None)\n bound_frames.append(frame)\n previous = frame\n return {**manifest, \"frames\": bound_frames}\n\n\ndef _main() -> int:\n parser = argparse.ArgumentParser()\n parser.add_argument(\"--manifest\", required=True, type=Path)\n parser.add_argument(\"--results\", required=True, action=\"append\", type=Path)\n parser.add_argument(\"--out\", required=True, type=Path)\n arguments = parser.parse_args()\n manifest = json.loads(arguments.manifest.read_text(encoding=\"utf-8\"))\n payloads = [\n json.loads(path.read_text(encoding=\"utf-8\"))\n for path in arguments.results\n ]\n bound = bind_scene_frame_results(manifest, payloads)\n temporary = arguments.out.with_suffix(arguments.out.suffix + \".tmp\")\n temporary.write_text(\n json.dumps(bound, indent=2, ensure_ascii=False) + \"\\n\",\n encoding=\"utf-8\",\n )\n temporary.replace(arguments.out)\n print(json.dumps({\"valid\": True, \"frames\": len(bound[\"frames\"])}))\n return 0\n\n\nif __name__ == \"__main__\":\n raise SystemExit(_main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":6423,"content_sha256":"e2c699cd626ad61537316e17a336d66d048314896215bf89d1c2d5cd6e8b1a0f"},{"filename":"scripts/build_scene_timeline.py","content":"#!/usr/bin/env python3\n\"\"\"Build the deterministic scene-based manifest v2 from narration timestamps.\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport json\nimport math\nimport re\nfrom difflib import SequenceMatcher\nfrom pathlib import Path\nfrom typing import Any\n\n\nWORD_PATTERN = re.compile(r\"[\\w]+(?:[-'’][\\w]+)*\", re.UNICODE)\nMIN_FRAME_SECONDS = 0.7\nTARGET_MAX_FRAME_SECONDS = 1.2\nHARD_MAX_FRAME_SECONDS = 1.5\nREQUESTED_DURATION_TOLERANCE_RATIO = 0.10\nREQUESTED_DURATION_MINIMUM_TOLERANCE_SECONDS = 3.0\n\n\ndef _tokens(value: str) -> list[str]:\n return [token.casefold() for token in WORD_PATTERN.findall(value)]\n\n\ndef _word_items(payload: Any) -> list[dict[str, Any]]:\n if isinstance(payload, list):\n candidates = payload\n elif isinstance(payload, dict) and isinstance(payload.get(\"words\"), list):\n candidates = payload[\"words\"]\n elif isinstance(payload, dict) and isinstance(payload.get(\"segments\"), list):\n candidates = [\n word\n for segment in payload[\"segments\"]\n if isinstance(segment, dict)\n for word in segment.get(\"words\", [])\n if isinstance(word, dict)\n ]\n else:\n raise ValueError(\"timestamps must contain a words array\")\n words = []\n for item in candidates:\n if not isinstance(item, dict):\n continue\n text = str(item.get(\"word\") or item.get(\"text\") or \"\").strip()\n start = item.get(\"start\")\n end = item.get(\"end\")\n if (\n text\n and isinstance(start, (int, float))\n and isinstance(end, (int, float))\n and 0 \u003c= start \u003c end\n ):\n words.append({\"word\": text, \"start\": float(start), \"end\": float(end)})\n if not words:\n raise ValueError(\"timestamps contain no usable timed words\")\n return words\n\n\ndef _beat_phrase(beat: dict[str, Any]) -> str:\n return str(beat.get(\"phrase\") or beat.get(\"vo_line\") or \"\").strip()\n\n\ndef _aligned_beat_spans(\n beats: list[dict[str, Any]],\n words: list[dict[str, Any]],\n audio_duration: float,\n) -> tuple[list[tuple[float, float]], float]:\n authored_tokens = [\n token\n for beat in beats\n for token in _tokens(_beat_phrase(beat))\n ]\n spoken_tokens = [\n token\n for word in words\n for token in _tokens(str(word[\"word\"]))\n ]\n matcher = SequenceMatcher(a=authored_tokens, b=spoken_tokens, autojunk=False)\n similarity = matcher.ratio()\n if similarity \u003c 0.5:\n raise ValueError(\n f\"narration/script similarity {similarity:.3f} is below 0.500\"\n )\n authored_to_spoken: dict[int, int] = {}\n for block in matcher.get_matching_blocks():\n for offset in range(block.size):\n authored_to_spoken[block.a + offset] = block.b + offset\n\n raw_spans: list[tuple[float, float]] = []\n authored_offset = 0\n for beat in beats:\n count = len(_tokens(_beat_phrase(beat)))\n spoken_indexes = [\n authored_to_spoken[index]\n for index in range(authored_offset, authored_offset + count)\n if index in authored_to_spoken\n ]\n if not spoken_indexes:\n raise ValueError(f\"beat {beat.get('n')} has no aligned narration words\")\n raw_spans.append(\n (\n words[min(spoken_indexes)][\"start\"],\n words[max(spoken_indexes)][\"end\"],\n )\n )\n authored_offset += count\n\n boundaries = [0.0]\n for index in range(len(raw_spans) - 1):\n left_end = raw_spans[index][1]\n right_start = raw_spans[index + 1][0]\n boundaries.append(max(boundaries[-1], (left_end + right_start) / 2))\n boundaries.append(audio_duration)\n spans = [\n (boundaries[index], boundaries[index + 1])\n for index in range(len(beats))\n ]\n return spans, similarity\n\n\ndef _frame_count(duration: float) -> int:\n if duration \u003c= 0:\n raise ValueError(\"beat span must be positive\")\n count = max(1, math.ceil(duration / TARGET_MAX_FRAME_SECONDS))\n while count > 1 and duration / count \u003c MIN_FRAME_SECONDS:\n count -= 1\n if duration / count > HARD_MAX_FRAME_SECONDS:\n raise ValueError(\n f\"beat span {duration:.3f}s cannot satisfy the {HARD_MAX_FRAME_SECONDS}s \"\n \"hard frame hold limit\"\n )\n return count\n\n\ndef build_scene_manifest(\n script: dict[str, Any],\n timestamps: Any,\n audio_duration: float,\n requested_duration_seconds: float | None = None,\n) -> dict[str, Any]:\n beats = script.get(\"beats\")\n if not isinstance(beats, list) or not beats:\n raise ValueError(\"script manifest must contain a non-empty beats array\")\n if audio_duration \u003c= 0:\n raise ValueError(\"audio_duration must be positive\")\n if requested_duration_seconds is not None:\n if requested_duration_seconds \u003c= 0:\n raise ValueError(\"requested_duration_seconds must be positive\")\n tolerance_seconds = max(\n REQUESTED_DURATION_MINIMUM_TOLERANCE_SECONDS,\n requested_duration_seconds * REQUESTED_DURATION_TOLERANCE_RATIO,\n )\n if abs(audio_duration - requested_duration_seconds) > tolerance_seconds:\n raise ValueError(\n f\"narration duration {audio_duration:.3f}s is outside the allowed \"\n f\"range for {requested_duration_seconds:.3f}s requested duration\"\n )\n normalized_beats = []\n for index, raw_beat in enumerate(beats, start=1):\n if not isinstance(raw_beat, dict) or raw_beat.get(\"n\") != index:\n raise ValueError(f\"beat {index} is invalid\")\n if not _beat_phrase(raw_beat):\n raise ValueError(f\"beat {index} has no phrase\")\n normalized_beats.append(raw_beat)\n\n words = _word_items(timestamps)\n spans, similarity = _aligned_beat_spans(\n normalized_beats,\n words,\n audio_duration,\n )\n frames: list[dict[str, Any]] = []\n variation_depth = 0\n if len(normalized_beats) != len(spans):\n raise ValueError(\"beat and timing counts differ\")\n for beat, (start, end) in zip(normalized_beats, spans):\n duration = end - start\n count = _frame_count(duration)\n for part in range(count):\n frame_start = start + duration * part / count\n frame_end = start + duration * (part + 1) / count\n frame_number = len(frames) + 1\n wants_variation = part > 0 or (\n bool(frames) and beat.get(\"image_mode\") == \"variation\"\n )\n is_variation = (\n bool(frames)\n and wants_variation\n and variation_depth \u003c 2\n )\n variation_depth = variation_depth + 1 if is_variation else 0\n frame = {\n \"n\": frame_number,\n \"beat_n\": beat[\"n\"],\n \"start\": round(frame_start, 3),\n \"end\": round(frame_end, 3),\n \"duration\": round(frame_end - frame_start, 3),\n \"output_slot\": f\"frame-{frame_number:03d}\",\n \"image_mode\": \"variation\" if is_variation else \"new\",\n \"generation_wave\": variation_depth,\n }\n if is_variation:\n frame[\"variation_of_frame\"] = frame_number - 1\n frame[\"reference_output_slot\"] = f\"frame-{frame_number - 1:03d}\"\n frame[\"change_only\"] = (\n str(beat.get(\"change_only\") or \"advance the named action\")\n )\n frames.append(frame)\n\n generation_waves = [\n {\n \"wave\": wave,\n \"output_slots\": [\n frame[\"output_slot\"]\n for frame in frames\n if frame[\"generation_wave\"] == wave\n ],\n }\n for wave in range(3)\n if any(frame[\"generation_wave\"] == wave for frame in frames)\n ]\n\n return {\n **script,\n \"manifest_version\": 2,\n \"narration\": {\n \"audio_duration_seconds\": round(audio_duration, 3),\n \"alignment_similarity\": round(similarity, 4),\n \"timeline_source\": \"whisper_word_timestamps\",\n },\n \"frames\": frames,\n \"generation_waves\": generation_waves,\n }\n\n\ndef _main() -> int:\n parser = argparse.ArgumentParser()\n parser.add_argument(\"--script\", required=True, type=Path)\n parser.add_argument(\"--timestamps\", required=True, type=Path)\n parser.add_argument(\"--audio-duration\", required=True, type=float)\n parser.add_argument(\"--requested-duration\", type=float)\n parser.add_argument(\"--out\", required=True, type=Path)\n arguments = parser.parse_args()\n script = json.loads(arguments.script.read_text())\n timestamps = json.loads(arguments.timestamps.read_text())\n manifest = build_scene_manifest(\n script,\n timestamps,\n arguments.audio_duration,\n arguments.requested_duration,\n )\n temporary = arguments.out.with_suffix(arguments.out.suffix + \".tmp\")\n temporary.write_text(json.dumps(manifest, indent=2, ensure_ascii=False) + \"\\n\")\n temporary.replace(arguments.out)\n print(\n json.dumps(\n {\n \"valid\": True,\n \"manifest_version\": 2,\n \"beats\": len(manifest[\"beats\"]),\n \"frames\": len(manifest[\"frames\"]),\n \"duration_seconds\": arguments.audio_duration,\n }\n )\n )\n return 0\n\n\nif __name__ == \"__main__\":\n raise SystemExit(_main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":9524,"content_sha256":"16544b9a2b5a34e2cd5ee8ebe54a6b6c7350dff8d8338dd0232e46ef9711eb10"},{"filename":"scripts/burn_captions.sh","content":"#!/usr/bin/env bash\n# burn_captions.sh — burn CLEAN CAPS subtitles onto a finished video from an .srt,\n# using ffmpeg + libass. Explainer-style but PLATELESS: bold WHITE text, black\n# OUTLINE + soft drop shadow, bottom-centre. No paper, NO background box.\n# Text is forced to UPPERCASE (Unicode-aware: works for Cyrillic etc. via python3).\n# Exact timing (from a Whisper-built srt), zero jitter.\n#\n# Sizing: ffmpeg converts SRT with a 384x288 subtitle grid (PlayResY=288), so values\n# are fractions of 288: fontsize 13 ≈ 4.5% of frame height (small), marginv 34 ≈\n# bottom 12%. Do NOT pass marginv ≥ 90 — that parks captions mid-frame.\n#\n# Usage:\n# scripts/burn_captions.sh --in video.mp4 --srt caps.srt --out final_subbed.mp4\n# [--font \"TikTok Sans\"] [--fontsdir DIR] [--fontsize 13] [--marginv 34]\n# [--outline 2] [--shadow 1]\n#\n# Font: put a bold face at scripts/fonts/ — default family \"TikTok Sans\"\n# (TikTokSans-Bold.ttf, SIL OFL, Latin+Cyrillic+Greek; see fonts/README.md for the\n# curl one-liner). Binaries are not committed; without a .ttf libass falls back to a\n# system font (the script warns — the look will differ).\n# Requires: ffmpeg (with libass); python3 recommended for non-ASCII uppercasing.\nset -euo pipefail\n\nIN=\"\"; SRT=\"\"; OUT=\"final_subbed.mp4\"; FONT=\"TikTok Sans\"; FONTSDIR=\"\"; FSIZE=13; MV=34; OUTLINE=2; SHADOW=1; NOCAPS=0\nwhile [[ $# -gt 0 ]]; do\n case \"$1\" in\n --in) IN=\"$2\"; shift 2 ;;\n --srt) SRT=\"$2\"; shift 2 ;;\n --out) OUT=\"$2\"; shift 2 ;;\n --font) FONT=\"$2\"; shift 2 ;;\n --fontsdir) FONTSDIR=\"$2\"; shift 2 ;;\n --fontsize) FSIZE=\"$2\"; shift 2 ;;\n --marginv) MV=\"$2\"; shift 2 ;;\n --outline) OUTLINE=\"$2\"; shift 2 ;;\n --shadow) SHADOW=\"$2\"; shift 2 ;;\n --no-caps) NOCAPS=1; shift ;;\n -h|--help) grep '^#' \"$0\" | sed 's/^# \\{0,1\\}//'; exit 0 ;;\n *) echo \"unknown arg: $1\" >&2; exit 1 ;;\n esac\ndone\n[[ -n \"$IN\" && -f \"$IN\" ]] || { echo \"ERROR: --in video not found\" >&2; exit 1; }\n[[ -n \"$SRT\" && -f \"$SRT\" ]] || { echo \"ERROR: --srt file not found\" >&2; exit 1; }\ncommand -v ffmpeg >/dev/null 2>&1 || { echo \"ERROR: ffmpeg not found\" >&2; exit 1; }\n\nSCRIPT_DIR=\"$(cd \"$(dirname \"$0\")\" && pwd)\"\n[[ -z \"$FONTSDIR\" && -d \"$SCRIPT_DIR/fonts\" ]] && FONTSDIR=\"$SCRIPT_DIR/fonts\"\nif [[ -n \"$FONTSDIR\" ]] && ! ls \"$FONTSDIR\"/*.[to]tf >/dev/null 2>&1; then\n echo \"WARN: no .ttf/.otf in $FONTSDIR — libass will fall back to a system font (look may differ). Drop TikTokSans-Bold.ttf there (fonts/README.md).\" >&2\nfi\n\n# Force UPPERCASE in the srt copy. python3 = Unicode-correct (Cyrillic etc.);\n# awk toupper is ASCII-only on mawk/BSD awk, so it is only the last-resort fallback.\n# --no-caps keeps the srt as authored (natural sentence case — the UGC-natural look).\nTMP=\"$(mktemp -d)\"; trap 'rm -rf \"$TMP\"' EXIT\nif [[ \"$NOCAPS\" == \"1\" ]]; then\n cp \"$SRT\" \"$TMP/caps_upper.srt\"\nelif command -v python3 >/dev/null 2>&1; then\n python3 -c 'import sys; sys.stdout.write(open(sys.argv[1], encoding=\"utf-8\", errors=\"replace\").read().upper())' \"$SRT\" > \"$TMP/caps_upper.srt\"\nelse\n echo \"WARN: python3 not found — using awk toupper (ASCII-only: non-Latin text may stay lowercase).\" >&2\n awk 'BEGIN{RS=\"\";FS=\"\\n\"} {print toupper($0) \"\\n\"}' \"$SRT\" > \"$TMP/caps_upper.srt\" 2>/dev/null || cp \"$SRT\" \"$TMP/caps_upper.srt\"\nfi\n\n# ASS colours are &HAABBGGRR. White fill, black outline, semi-soft shadow.\n# BorderStyle=1 => outline+shadow (NO opaque box). No BackColour plate.\nSTYLE=\"FontName=${FONT},Fontsize=${FSIZE},PrimaryColour=&H00FFFFFF&,OutlineColour=&H00000000&,BorderStyle=1,Outline=${OUTLINE},Shadow=${SHADOW},Alignment=2,MarginV=${MV},Bold=1\"\n\nFONTS_ARG=\"\"; [[ -n \"$FONTSDIR\" ]] && FONTS_ARG=\":fontsdir=${FONTSDIR}\"\nffmpeg -y -loglevel error -i \"$IN\" -vf \"subtitles=${TMP}/caps_upper.srt${FONTS_ARG}:force_style='${STYLE}'\" \\\n -c:v libx264 -preset veryfast -crf 20 -c:a copy \"$OUT\"\necho \"DONE -> $OUT (clean CAPS, no plate, small bottom captions)\"\n","content_type":"application/x-sh; charset=utf-8","language":"bash","size":3937,"content_sha256":"fefe706786c2eddc07597fb05c234bafec9e12ca18f8a26b8cca090ba8cc1b94"},{"filename":"scripts/check_voice.sh","content":"#!/usr/bin/env bash\n# Usage: check_voice.sh voice.wav window_seconds\nset -euo pipefail\n[[ $# -eq 2 && -s \"$1\" ]] || { echo 'Usage: check_voice.sh voice.wav window_seconds' >&2; exit 2; }\ncommand -v ffmpeg >/dev/null && command -v ffprobe >/dev/null && command -v awk >/dev/null || { echo 'Missing audio tools' >&2; exit 2; }\nawk -v t=\"$2\" 'BEGIN{exit !(t>3 && t\u003c=10)}' || { echo 'Spoken window must be greater than 3 and at most 10 seconds' >&2; exit 2; }\nffprobe -v error -select_streams a:0 -show_entries stream=codec_type -of csv=p=0 \"$1\" | grep -qx audio || { echo 'No readable audio stream' >&2; exit 1; }\nend=$(ffmpeg -v error -i \"$1\" -af 'aformat=channel_layouts=mono,aresample=16000,asetnsamples=n=1600:p=0,astats=metadata=1:reset=1,ametadata=mode=print:key=lavfi.astats.Overall.RMS_level:file=-' -f null - 2>/dev/null | awk -F= '/RMS_level/{n++;v=$2;if(v!=\"-inf\" && v+0> -34)last=n} END{printf \"%.2f\",last*0.1}')\nawk -v t=\"$2\" -v s=\"$end\" 'BEGIN{printf \"Speech ends at %.2fs; required %.2f–%.2fs\\n\",s,t-2,t-.2; exit !(s>=t-2 && s\u003c=t-.2)}'\n","content_type":"application/x-sh; charset=utf-8","language":"bash","size":1050,"content_sha256":"623d0c1d44036614bd77c8fdb05273adc23e8dcb2dfa120fd8709dfa2ed0bbc0"},{"filename":"scripts/measure_narration_takes.py","content":"#!/usr/bin/env python3\n\"\"\"Measure per-block narration against its authored script.\"\"\"\nfrom __future__ import annotations\n\nimport argparse\nimport json\nimport math\nimport subprocess\nimport sys\nfrom pathlib import Path\nfrom typing import Any, Callable\n\nTARGET_MIN = 7.8\nHARD_MIN = 7.2\nHARD_MAX = 9.5\nADAPTIVE_WORD_MIN = 17\nDEFAULT_WORD_MAX = 23\n\n\ndef _parse_blocks(value: str) -> set[int]:\n if not value.strip():\n return set()\n try:\n blocks = {int(item.strip()) for item in value.split(\",\") if item.strip()}\n except ValueError as error:\n raise argparse.ArgumentTypeError(\"block lists contain integers only\") from error\n if any(block \u003c 1 for block in blocks):\n raise argparse.ArgumentTypeError(\"block numbers start at 1\")\n return blocks\n\n\ndef _default_metrics_script() -> Path:\n candidate = Path(__file__).resolve().with_name(\"speech_metrics.sh\")\n if not candidate.is_file():\n raise FileNotFoundError(\"speech_metrics.sh is missing\")\n return candidate\n\n\ndef _recommended_words(words: int, speech: float) -> int:\n if speech \u003c= 0:\n return words\n if speech > HARD_MAX:\n target = round(words * 9.0 / speech)\n return max(ADAPTIVE_WORD_MIN, min(words - 1, target))\n if speech \u003c HARD_MIN:\n target = math.ceil(words * 8.5 / speech)\n return min(DEFAULT_WORD_MAX, max(words + 1, target))\n return words\n\n\ndef _classify(metric: dict[str, Any], *, accept_soft: bool) -> tuple[str, bool]:\n speech = float(metric[\"speech\"])\n pauses = int(metric.get(\"pauses\", 0))\n rate = metric.get(\"rate\")\n if pauses:\n return \"pausey\", False\n if speech > HARD_MAX:\n return \"too_long\", False\n if speech \u003c HARD_MIN:\n return \"too_short\", False\n if rate == \"RUSHED\":\n return \"rushed\", False\n if rate == \"SLOW\":\n return \"slow\", False\n if speech \u003c TARGET_MIN:\n return (\"soft_pass\" if accept_soft else \"soft_retry\"), accept_soft\n return \"pass\", True\n\n\ndef measure(\n script: dict[str, Any],\n voice_dir: Path,\n metrics_script: Path,\n *,\n accept_soft_blocks: set[int] | None = None,\n runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,\n) -> dict[str, Any]:\n accept_soft_blocks = accept_soft_blocks or set()\n blocks = script.get(\"blocks\")\n if not isinstance(blocks, list) or not blocks:\n raise ValueError(\"script must contain a non-empty blocks array\")\n\n takes: list[dict[str, Any]] = []\n retry_blocks: list[int] = []\n overlong_blocks: list[int] = []\n for index, block in enumerate(blocks, start=1):\n if not isinstance(block, dict) or not isinstance(block.get(\"vo_line\"), str):\n raise ValueError(f\"block {index} has no vo_line\")\n voice = voice_dir / f\"voice{index:02d}.wav\"\n if not voice.is_file():\n takes.append({\"block\": index, \"state\": \"missing\", \"file\": str(voice)})\n retry_blocks.append(index)\n continue\n\n completed = runner(\n [\n str(metrics_script),\n str(voice),\n \"--text\",\n block[\"vo_line\"],\n \"--json\",\n ],\n check=False,\n capture_output=True,\n text=True,\n )\n if completed.returncode != 0:\n detail = (completed.stderr or completed.stdout).strip()[:500]\n raise RuntimeError(f\"speech metrics failed for block {index}: {detail}\")\n metric = json.loads(completed.stdout)\n state, accepted = _classify(\n metric, accept_soft=index in accept_soft_blocks\n )\n take = {\"block\": index, \"state\": state, **metric}\n if state in {\"too_long\", \"too_short\"}:\n take[\"recommended_words\"] = _recommended_words(\n int(metric[\"words\"]), float(metric[\"speech\"])\n )\n takes.append(take)\n if not accepted:\n retry_blocks.append(index)\n if state == \"too_long\":\n overlong_blocks.append(index)\n\n result: dict[str, Any] = {\n \"valid\": not retry_blocks,\n \"retry_blocks\": retry_blocks,\n \"overlong_blocks\": overlong_blocks,\n \"takes\": takes,\n }\n if overlong_blocks:\n joined = \",\".join(str(block) for block in overlong_blocks)\n result[\"revalidate_with\"] = f\"--duration-retry-blocks {joined}\"\n return result\n\n\ndef main() -> int:\n parser = argparse.ArgumentParser()\n parser.add_argument(\"--script\", required=True, type=Path)\n parser.add_argument(\"--voice-dir\", required=True, type=Path)\n parser.add_argument(\"--metrics-script\", type=Path, default=None)\n parser.add_argument(\n \"--accept-soft-blocks\",\n type=_parse_blocks,\n default=set(),\n help=\"Comma-separated blocks already retried once in the 7.2-7.8s soft band\",\n )\n arguments = parser.parse_args()\n try:\n script = json.loads(arguments.script.read_text(encoding=\"utf-8\"))\n result = measure(\n script,\n arguments.voice_dir,\n arguments.metrics_script or _default_metrics_script(),\n accept_soft_blocks=arguments.accept_soft_blocks,\n )\n except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as error:\n print(json.dumps({\"valid\": False, \"errors\": [str(error)]}))\n return 2\n print(json.dumps(result, ensure_ascii=False, indent=2))\n return 0 if result[\"valid\"] else 1\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":5494,"content_sha256":"657c471001f574c9f0e627442fb62ae072b6d836ad7e643c328eb17fd5d25a38"},{"filename":"scripts/presenter_composite.sh","content":"#!/usr/bin/env bash\n# Usage: presenter_composite.sh background.mp4 presenter.mp4 output.mp4 [--style cutout|badge|fullframe] [--pos br|bl|tr|tl]\nset -euo pipefail\n\nSTYLE=\"cutout\"; POS=\"br\"; KEY=\"auto\"; CUT_H_FRAC=\"0.64\"; CUT_W_FRAC=\"0.72\"\nBADGE_FRAC=\"0.355\"; BADGE_CROP_Y=\"40\"; MARGIN_FRAC=\"\"; DISC=\"0xE8DFD0\"\nARGS=()\nwhile [[ $# -gt 0 ]]; do\n case \"$1\" in\n --style) STYLE=\"$2\"; shift 2 ;;\n --pos) POS=\"$2\"; shift 2 ;;\n --key) KEY=\"$2\"; shift 2 ;;\n --cut-h-frac) CUT_H_FRAC=\"$2\"; shift 2 ;;\n --cut-w-frac) CUT_W_FRAC=\"$2\"; shift 2 ;;\n --badge-frac) BADGE_FRAC=\"$2\"; shift 2 ;;\n --badge-crop-y) BADGE_CROP_Y=\"$2\"; shift 2 ;;\n --margin-frac) MARGIN_FRAC=\"$2\"; shift 2 ;;\n --disc-color) DISC=\"$2\"; shift 2 ;;\n -h|--help) grep '^#' \"$0\" | sed 's/^# \\{0,1\\}//'; exit 0 ;;\n *) ARGS+=(\"$1\"); shift ;;\n esac\ndone\n[[ ${#ARGS[@]} -eq 3 ]] || { echo \"usage: presenter_composite.sh BLOCK TALK OUT [--style cutout|badge] (see --help)\" >&2; exit 2; }\nBLOCK=\"${ARGS[0]}\"; TALK=\"${ARGS[1]}\"; OUT=\"${ARGS[2]}\"\n[[ \"$STYLE\" == \"cutout\" || \"$STYLE\" == \"badge\" || \"$STYLE\" == \"fullframe\" ]] || { echo \"bad --style '$STYLE'\" >&2; exit 2; }\n[[ \"$POS\" == \"br\" || \"$POS\" == \"bl\" || \"$POS\" == \"tr\" || \"$POS\" == \"tl\" ]] || { echo \"bad --pos '$POS' (br|bl|tr|tl)\" >&2; exit 2; }\nif [[ \"$STYLE\" == \"cutout\" && ( \"$POS\" == \"tr\" || \"$POS\" == \"tl\" ) ]]; then\n echo \"cutout is bottom-only (torso is cut at the frame edge) — use --pos br|bl or --style badge\" >&2; exit 2\nfi\nfor f in \"$BLOCK\" \"$TALK\"; do [[ -f \"$f\" ]] || { echo \"missing input: $f\" >&2; exit 2; }; done\n\nprobe() { ffprobe -v error -select_streams v:0 -show_entries stream=\"$2\" -of csv=p=0 \"$1\"; }\nBW=$(probe \"$BLOCK\" width); BH=$(probe \"$BLOCK\" height)\nTW=$(probe \"$TALK\" width); TH=$(probe \"$TALK\" height)\n\nCHROMA_CORE=\"format=gbrp,split[ks][km];[km]geq=r='255*gt(g(X,Y)*100,42*(r(X,Y)+g(X,Y)+b(X,Y)))*gt(g(X,Y),r(X,Y)+10)*gt(g(X,Y),b(X,Y)+10)':g='0':b='0',extractplanes=r,dilation,gblur=sigma=1.2,lutyuv=y=negval[ka];[ks][ka]alphamerge,despill=type=green\"\nCORE=\"\"; ROUTE=\"\"\nif [[ \"$KEY\" == \"auto\" ]]; then\n SAMPLES=$(for XY in \"8:8\" \"iw-48:8\" \"8:ih/4\" \"iw-48:ih/4\" \"iw/2-20:8\"; do\n X=\"${XY%%:*}\"; Y=\"${XY##*:}\"\n ffmpeg -v error -ss 1 -i \"$TALK\" -vf \"crop=40:40:${X}:${Y},scale=1:1\" -frames:v 1 -f rawvideo -pix_fmt rgb24 - 2>/dev/null | od -An -tu1 | awk '{print $1, $2, $3; exit}'\n done)\n read -r MR MG MB SIM \u003c\u003c\u003c\"$(echo \"$SAMPLES\" | awk '\n {r[NR]=$1;g[NR]=$2;b[NR]=$3;sr+=$1;sg+=$2;sb+=$3;n++}\n END{mr=sr/n;mg=sg/n;mb=sb/n;mx=0;\n for(i=1;i\u003c=n;i++){d=sqrt((r[i]-mr)^2+(g[i]-mg)^2+(b[i]-mb)^2)/441.67; if(d>mx)mx=d}\n s=mx+0.04; if(s\u003c0.08)s=0.08; if(s>0.15)s=0.15;\n printf \"%d %d %d %.3f\", mr, mg, mb, s}')\"\n KEY=$(printf \"0x%02X%02X%02X\" \"$MR\" \"$MG\" \"$MB\")\n if [[ \"$MG\" -lt 80 || \"$MG\" -le \"$MR\" || \"$MG\" -le \"$MB\" ]]; then\n echo \"presenter_composite: background ($MR,$MG,$MB) is not green — is the talk clip on a green screen?\" >&2; exit 1\n fi\n SAT=$(( MG - (MR > MB ? MR : MB) ))\n if [[ \"$SAT\" -ge 60 && \"$MR\" -lt 30 ]]; then\n SIM=$(awk -v s=\"$SIM\" 'BEGIN{print (s\u003c0.11)?\"0.110\":s}') # chromakey needs a bit more reach on soft hair edges\n CORE=\"chromakey=${KEY}:${SIM}:0.05,despill=type=green\"; ROUTE=\"chromakey\"\n echo \"KEY: $KEY -> chromakey=${KEY}:${SIM}:0.05\"\n else\n CORE=\"$CHROMA_CORE\"; ROUTE=\"chromaticity\"\n echo \"KEY: $KEY (SAT=$SAT R=$MR) -> chromaticity key (g_norm>0.42)\"\n fi\nelse\n CORE=\"chromakey=${KEY}:0.13:0.06,despill=type=green\"; ROUTE=\"chromakey-manual\"\n echo \"KEY: $KEY -> $CORE\"\nfi\n\nrun_qc() { # $1 = CORE filter chain, $2 = bg alpha max -> prints metrics; rc 0/1\n local png rc temp_dir; temp_dir=$(mktemp -d) || return 1; png=\"$temp_dir/matte.png\"\n ffmpeg -y -v error -ss 2 -i \"$TALK\" -filter_complex \"[0:v]$1\" -frames:v 1 \"$png\" || { rmdir \"$temp_dir\"; return 1; }\n QC=$(python3 - \"$png\" \"$2\" \u003c\u003c'PY'\nimport sys\nfrom PIL import Image\nimport numpy as np\nim = np.array(Image.open(sys.argv[1]).convert(\"RGBA\"))\nbgmax = float(sys.argv[2])\nh, w = im.shape[:2]\na = im[:, :, 3].astype(int)\npts = [(8, 8), (w - 48, 8), (8, h // 4), (w - 48, h // 4), (w // 2 - 20, 8)]\nbg = float(np.mean([a[y:y + 40, x:x + 40].mean() for (x, y) in pts]))\nface = float(a[int(h * 0.30):int(h * 0.55), int(w * 0.40):int(w * 0.60)].mean())\ntorso = float(a[int(h * 0.60):int(h * 0.85), int(w * 0.35):int(w * 0.65)].mean())\nok = bg \u003c bgmax and face > 245 and torso > 245\nprint(f\"{'PASS' if ok else 'FAIL'} bg_alpha={bg:.0f} face_alpha={face:.0f} torso_alpha={torso:.0f}\")\nsys.exit(0 if ok else 1)\nPY\n); rc=$?\n rm -f \"$png\"; rmdir \"$temp_dir\"; echo \"$QC\"; return $rc\n}\nFALLBACK=0\nif QCRES=$(run_qc \"$CORE\" 16); then\n echo \"QC: $QCRES\"\nelse\n echo \"QC: $QCRES\"\n BGV=$(sed -E 's/.*bg_alpha=([0-9]+).*/\\1/' \u003c\u003c\u003c\"$QCRES\")\n if [[ \"$ROUTE\" == chromakey* && \"$BGV\" -lt 16 ]]; then\n echo \"QC: person alpha failed on chromakey with a clean bg -> auto-fallback to chromaticity key\" >&2\n CORE=\"$CHROMA_CORE\"; ROUTE=\"chromaticity-fallback\"; FALLBACK=1\n if QCRES=$(run_qc \"$CORE\" 40); then\n echo \"QC(fallback): $QCRES\"\n else\n echo \"QC(fallback): $QCRES\"\n echo \"presenter_composite: matte QC FAILED on both keys — regenerate this talking clip (bad green), do not tune thresholds\" >&2; exit 3\n fi\n else\n echo \"presenter_composite: matte QC FAILED — regenerate this talking clip (bad green), do not tune thresholds\" >&2; exit 3\n fi\nfi\n\neven() { awk -v x=\"$1\" 'BEGIN{ v=int(x+0.5); if (v%2) v=v+1; print v }'; }\n\nVOICE=\"${OUT%.mp4}_voice.wav\"\nACHAIN=\"loudnorm=I=-16:TP=-1.5:LRA=11,aformat=sample_rates=44100:channel_layouts=stereo\"\n\nif [[ \"$STYLE\" == \"fullframe\" ]]; then\n FH=$(even \"$BH\")\n FILTER=\"[1:v]${CORE},scale=-2:${FH}[p];[0:v][p]overlay=(W-w)/2:H-h:shortest=1[v];[1:a]${ACHAIN}[a]\"\nelif [[ \"$STYLE\" == \"cutout\" ]]; then\n CH=$(even \"$(awk -v h=\"$BH\" -v f=\"$CUT_H_FRAC\" 'BEGIN{print h*f}')\")\n CW_KEEP=\"$CUT_W_FRAC\"\n CW_OFF=$(awk -v f=\"$CUT_W_FRAC\" 'BEGIN{printf \"%.4f\", (1-f)/2}')\n case \"$POS\" in\n br) OXY=\"x=W-w+6:y=H-h+2\" ;;\n bl) OXY=\"x=-6:y=H-h+2\" ;;\n esac\n FILTER=\"[1:v]${CORE},crop=iw*${CW_KEEP}:ih:iw*${CW_OFF}:0,scale=-2:${CH}[p];[0:v][p]overlay=${OXY}:shortest=1[v];[1:a]${ACHAIN}[a]\"\nelse\n S=$(( TW \u003c TH ? TW : TH ))\n MAXY=$(( TH - S )); CY=$(( BADGE_CROP_Y > MAXY ? MAXY : BADGE_CROP_Y ))\n CX=$(( (TW - S) / 2 ))\n C=$(( S / 2 )); R=$(awk -v s=\"$S\" 'BEGIN{print int(s*0.4375)}'); R2=$(( R * R ))\n BMIN=$(( BW \u003c BH ? BW : BH ))\n BS=$(even \"$(awk -v h=\"$BMIN\" -v f=\"$BADGE_FRAC\" 'BEGIN{print h*f}')\")\n px() { awk -v d=\"$1\" -v f=\"$2\" 'BEGIN{print int(d*f+0.5)}'; }\n if [[ -n \"$MARGIN_FRAC\" ]]; then\n MXL=$(px \"$BH\" \"$MARGIN_FRAC\"); MXR=$MXL; MYT=$MXL; MYB=$MXL\n elif (( BW >= BH )); then # landscape: broadcast title-safe\n MXL=$(px \"$BW\" 0.05); MXR=$MXL; MYT=$(px \"$BH\" 0.07); MYB=$MYT\n else\n MXL=$(px \"$BW\" 0.112); MYT=$(px \"$BH\" 0.131); MYB=$(px \"$BH\" 0.167)\n if [[ \"$POS\" == \"br\" || \"$POS\" == \"bl\" ]]; then\n MXR=$(px \"$BW\" 0.218) # clear of the bottom-right button rail\n else\n MXR=$(px \"$BW\" 0.112)\n fi\n fi\n case \"$POS\" in\n br) OXY=\"x=W-w-${MXR}:y=H-h-${MYB}\" ;;\n bl) OXY=\"x=${MXL}:y=H-h-${MYB}\" ;;\n tr) OXY=\"x=W-w-${MXR}:y=${MYT}\" ;;\n tl) OXY=\"x=${MXL}:y=${MYT}\" ;;\n esac\n FILTER=\"color=c=white:s=${S}x${S},format=gray,geq=lum='if(lte((X-${C})*(X-${C})+(Y-${C})*(Y-${C}),${R2}),255,0)',split[circ][circ2];[1:v]${CORE},crop=${S}:${S}:${CX}:${CY},format=rgba,split[pkA][pkB];[pkA]alphaextract[pa];[pa][circ]blend=all_mode=multiply[clipmask];[pkB][clipmask]alphamerge[personclip];color=c=${DISC}:s=${S}x${S}[bg];[bg][circ2]alphamerge[disc];[disc][personclip]overlay,scale=${BS}:${BS}[badge];[0:v][badge]overlay=${OXY}:shortest=1[v];[1:a]${ACHAIN}[a]\"\nfi\n\nffmpeg -y -v error -i \"$BLOCK\" -i \"$TALK\" -filter_complex \"$FILTER\" \\\n -map \"[v]\" -map \"[a]\" -c:v libx264 -crf 18 -preset veryfast -pix_fmt yuv420p -c:a aac \"$OUT\"\nffmpeg -y -v error -i \"$TALK\" -vn -af \"$ACHAIN\" -c:a pcm_s16le \"$VOICE\"\n\nBGA=$(sed -E 's/.*bg_alpha=([0-9]+).*/\\1/' \u003c\u003c\u003c\"$QCRES\")\nFCA=$(sed -E 's/.*face_alpha=([0-9]+).*/\\1/' \u003c\u003c\u003c\"$QCRES\")\nTSA=$(sed -E 's/.*torso_alpha=([0-9]+).*/\\1/' \u003c\u003c\u003c\"$QCRES\")\nSIDECAR=\"${OUT}.qc.json\"\nprintf '{\"script\":\"presenter_composite.sh\",\"result\":\"PASS\",\"style\":\"%s\",\"pos\":\"%s\",\"key\":\"%s\",\"route\":\"%s\",\"fallback\":%s,\"bg_alpha\":%s,\"face_alpha\":%s,\"torso_alpha\":%s,\"block\":\"%s\",\"talk\":\"%s\",\"out\":\"%s\",\"ts\":\"%s\"}\\n' \\\n \"$STYLE\" \"$POS\" \"$KEY\" \"$ROUTE\" \"$FALLBACK\" \"$BGA\" \"$FCA\" \"$TSA\" \\\n \"$(basename \"$BLOCK\")\" \"$(basename \"$TALK\")\" \"$(basename \"$OUT\")\" \\\n \"$(date -u +%Y-%m-%dT%H:%M:%SZ)\" > \"$SIDECAR\"\n\necho \"OUT: $OUT\"\necho \"VOICE: $VOICE\"\necho \"QC-SIDECAR: $SIDECAR\"\n","content_type":"application/x-sh; charset=utf-8","language":"bash","size":8607,"content_sha256":"3233bada18a9565279948676d5742e805d4a453cc8d26b200aa2644c14db842d"},{"filename":"scripts/speech_metrics.sh","content":"#!/usr/bin/env bash\n# Usage: speech_metrics.sh audio.wav --text \"spoken text\" --json\nset -euo pipefail\n\nFILE=\"\"; PAUSE=\"0.8\"; NOISE=\"-45\"; JSON=0; TEXT=\"\"; WORDS=\"\"; MAXWPS=\"2.9\"\nwhile [[ $# -gt 0 ]]; do\n case \"$1\" in\n --pause) PAUSE=\"$2\"; shift 2 ;;\n --noise) NOISE=\"$2\"; shift 2 ;;\n --text) TEXT=\"$2\"; shift 2 ;;\n --words) WORDS=\"$2\"; shift 2 ;;\n --max-wps) MAXWPS=\"$2\"; shift 2 ;;\n --json) JSON=1; shift ;;\n -h|--help) grep '^#' \"$0\" | sed 's/^# \\{0,1\\}//'; exit 0 ;;\n *) FILE=\"$1\"; shift ;;\n esac\ndone\n[[ -n \"$FILE\" && -f \"$FILE\" ]] || { echo \"ERROR: pass an audio file\" >&2; exit 1; }\nfor b in ffmpeg ffprobe awk; do command -v \"$b\" >/dev/null 2>&1 || { echo \"ERROR: '$b' not found\" >&2; exit 1; }; done\n\nDUR=\"$(ffprobe -v error -show_entries format=duration -of csv=p=0 \"$FILE\")\"\nEVENTS=\"$(ffmpeg -i \"$FILE\" -af \"silencedetect=noise=${NOISE}dB:d=0.25\" -f null - 2>&1 \\\n | grep -Eo 'silence_(start|end): *-?[0-9.]+' || true)\"\n\nread -r SS SE \u003c\u003c\u003c \"$(awk -v A=\"$DUR\" '\n /silence_start/ { s=$NF+0; if(first==\"\"){first=s}; last=s; open=1; next }\n /silence_end/ { e=$NF+0; if(first!=\"\" && first\u003c0.1 && ssdone==\"\"){ss=e; ssdone=1};\n e_last=e; s_before=last; open=0 }\n END {\n if (open==1) se=last;\n else if (e_last>0 && A-e_last\u003c0.2) se=s_before;\n else se=A;\n printf \"%.3f %.3f\", ss+0, se+0 }' \u003c\u003c\u003c \"$EVENTS\")\"\n[[ -z \"${SS:-}\" ]] && SS=0; [[ -z \"${SE:-}\" || \"$SE\" == \"0.000\" ]] && SE=\"$DUR\"\nSPEECH=\"$(awk -v a=\"$SS\" -v b=\"$SE\" 'BEGIN{d=b-a; if(d\u003c0)d=0; printf \"%.3f\", d}')\"\nawk -v s=\"$SPEECH\" 'BEGIN{exit (s\u003c0.3)?0:1}' && { SS=0; SE=\"$DUR\"; SPEECH=\"$DUR\"; }\n\nread -r PN PMAX \u003c\u003c\u003c \"$(awk -v lo=\"$SS\" -v hi=\"$SE\" -v thr=\"$PAUSE\" '\n /silence_start/ { s=$NF+0; open=1; next }\n /silence_end/ { e=$NF+0; if(open==1 && s>lo+0.1 && e\u003chi-0.1 && e-s>=thr){n++; if(e-s>mx)mx=e-s}; open=0 }\n END{ printf \"%d %.2f\", n+0, mx+0 }' \u003c\u003c\u003c \"$EVENTS\")\"\n\nif [[ -z \"$WORDS\" && -n \"$TEXT\" ]]; then\n WORDS=\"$(printf '%s' \"$TEXT\" | sed -e 's/\\[[^][]*\\]//g' -e 's/([^()]*)//g' \\\n | tr -cs \"[:alnum:]'-\" ' ' | awk '{print NF}')\"\nfi\nRATE=\"\"; WPS=\"\"\nif [[ -n \"$WORDS\" ]]; then\n read -r WPS RATE \u003c\u003c\u003c \"$(awk -v w=\"$WORDS\" -v s=\"$SPEECH\" -v mx=\"$MAXWPS\" 'BEGIN{\n if (s\u003c=0) { print \"0.00 unknown\"; exit }\n r=w/s; printf \"%.2f %s\", r, (r>mx ? \"RUSHED\" : (r\u003c2.1 ? \"SLOW\" : \"ok\")) }')\"\nfi\n\nif [[ \"$JSON\" == \"1\" ]]; then\n printf '{\"file\":\"%s\",\"duration\":%.3f,\"speech_start\":%.3f,\"speech_end\":%.3f,\"speech\":%.3f,\"pauses\":%d,\"longest_pause\":%.2f' \\\n \"$(basename \"$FILE\")\" \"$DUR\" \"$SS\" \"$SE\" \"$SPEECH\" \"${PN:-0}\" \"${PMAX:-0}\"\n [[ -n \"$WORDS\" ]] && printf ',\"words\":%d,\"wps\":%s,\"rate\":\"%s\"' \"$WORDS\" \"$WPS\" \"$RATE\"\n printf '}\\n'\nelse\n printf 'duration=%.3f speech_start=%.3f speech_end=%.3f speech=%.3f pauses=%d longest_pause=%.2f' \\\n \"$DUR\" \"$SS\" \"$SE\" \"$SPEECH\" \"${PN:-0}\" \"${PMAX:-0}\"\n [[ -n \"$WORDS\" ]] && printf ' words=%d wps=%s rate=%s' \"$WORDS\" \"$WPS\" \"$RATE\"\n printf '\\n'\nfi\nif [[ \"$RATE\" == \"RUSHED\" ]]; then\n echo \"WARN: ${WPS} words/sec is auctioneer pace — REWRITE the line shorter (target \u003c= ${MAXWPS}) and regenerate. Do NOT ship it and do NOT slow the audio down.\" >&2\nfi\n","content_type":"application/x-sh; charset=utf-8","language":"bash","size":3206,"content_sha256":"0408aabe795045bdb4e16c31d5a83a155deeefd5a67617f1b861bf3a4d812d3e"},{"filename":"scripts/tests/test_assembly.py","content":"#!/usr/bin/env python3\n\"\"\"Exercise assembly using generated local fixtures; no account is required.\"\"\"\nimport json\nimport shutil\nimport subprocess\nimport tempfile\nfrom pathlib import Path\n\n\ndef run(*args, **kwargs):\n return subprocess.run(args, check=True, capture_output=True, text=True, **kwargs)\n\n\ndef main():\n for tool in (\"ffmpeg\", \"ffprobe\", \"bash\"):\n if not shutil.which(tool):\n raise RuntimeError(f\"missing {tool}\")\n script = Path(__file__).resolve().parents[1] / \"assemble_blocks.sh\"\n with tempfile.TemporaryDirectory(prefix=\"faceless-test-\") as temp:\n folder = Path(temp)\n for index, duration in [(1, 8.8), (2, 4.2)]:\n run(\"ffmpeg\", \"-v\", \"error\", \"-y\", \"-f\", \"lavfi\", \"-i\",\n \"testsrc2=size=320x180:rate=24:duration=10\", \"-c:v\", \"libx264\",\n \"-pix_fmt\", \"yuv420p\", str(folder / f\"clip{index:02}.mp4\"))\n run(\"ffmpeg\", \"-v\", \"error\", \"-y\", \"-f\", \"lavfi\", \"-i\",\n f\"sine=frequency=220:duration={duration}\", \"-ar\", \"24000\",\n str(folder / f\"voice{index:02}.wav\"))\n checker = script.parent / \"check_voice.sh\"\n run(\"bash\", str(checker), str(folder / \"voice01.wav\"), \"10\")\n wrong_window = subprocess.run([\"bash\", str(checker), str(folder / \"voice02.wav\"), \"10\"], capture_output=True)\n assert wrong_window.returncode != 0\n output = folder / \"final.mp4\"\n pairs = [str(folder / f\"{kind}{index:02}.{extension}\")\n for index in (1, 2) for kind, extension in ((\"clip\", \"mp4\"), (\"voice\", \"wav\"))]\n result = run(\"bash\", str(script), \"--out\", str(output), \"--last-seconds\", \"5\", *pairs)\n assert \"mixes numbers\" not in result.stderr\n duration = float(run(\"ffprobe\", \"-v\", \"error\", \"-show_entries\", \"format=duration\",\n \"-of\", \"default=nw=1:nk=1\", str(output)).stdout)\n assert abs(duration - 15) \u003c 0.2, duration\n receipt = json.loads(Path(str(output) + \".assembly.json\").read_text())\n assert receipt[\"blocks\"] == 2 and receipt[\"total_s\"] == 15\n failed = subprocess.run([\"bash\", str(script), \"--out\", str(folder / \"invalid.mp4\"),\n pairs[0], pairs[3]], capture_output=True, text=True)\n assert failed.returncode != 0 and \"DURATION GATE FAILED\" in failed.stderr\n assert not (folder / \"invalid.mp4\").exists()\n print(\"assembly: ordered pairs, short final block and duration gate passed\")\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":2519,"content_sha256":"b51674006f0fcc77bdaa0dee5e31330afcc5ee2cc8ea793397ac33bb48f3cb1f"},{"filename":"scripts/tests/test_captions.py","content":"#!/usr/bin/env python3\n\"\"\"Check authored caption alignment without downloading a model.\"\"\"\nimport importlib.util\nfrom pathlib import Path\n\npath = Path(__file__).resolve().parents[1] / \"audio_to_captions.py\"\nspec = importlib.util.spec_from_file_location(\"captions\", path)\nmodule = importlib.util.module_from_spec(spec)\nspec.loader.exec_module(module)\ngroups = [{\"n\": 1, \"speech_start\": 0, \"speech_end\": 2,\n \"words\": [{\"word\": \"hello\", \"start\": 0.1, \"end\": 0.5},\n {\"word\": \"world\", \"start\": 0.6, \"end\": 1.0}]}]\naligned, score = module.align_blocks_to_script(groups, [[\"hello\", \"world\"]], minimum_similarity=0.75)\nassert score == 1.0 and [word[\"word\"] for word in aligned] == [\"hello\", \"world\"]\ntry:\n module.align_blocks_to_script(groups, [], minimum_similarity=0.75)\nexcept ValueError:\n pass\nelse:\n raise AssertionError(\"mismatched blocks were accepted\")\nprint(\"captions: authored alignment and block count passed\")\n","content_type":"text/x-python; charset=utf-8","language":"python","size":953,"content_sha256":"5011ba3606095f8926f51591d36728f9f73d8dc3149fa15b35aabdff4b359dff"},{"filename":"scripts/tests/test_channel.py","content":"#!/usr/bin/env python3\n\"\"\"Exercise channel script, lock and narration checks without external calls.\"\"\"\nimport importlib.util\nimport json\nimport subprocess\nimport tempfile\nfrom pathlib import Path\n\nroot = Path(__file__).resolve().parents[1]\n\ndef load(name):\n spec = importlib.util.spec_from_file_location(name, root / (name + '.py'))\n module = importlib.util.module_from_spec(spec)\n spec.loader.exec_module(module)\n return module\n\n\ndef words(n, offset):\n alphabet = 'abcdefghijklmnopqrstuvwxyz'\n return ' '.join('word' + alphabet[(i + offset) % 26] + alphabet[((i + offset) // 26) % 26] + alphabet[((i + offset) // 676) % 26]\n for i in range(n))\n\n\ndef shots(offset):\n return [f'{size}: {words(3, offset + i * 10)}' for i, size in enumerate(['WIDE', 'MEDIUM', 'CU', 'DETAIL', 'ECU'])]\n\n\ndef main():\n validator = load('validate_motion_script')\n script = {'topic': 'Why plants grow', 'genre': 'education', 'animation_mode': 'fully_animated',\n 'style': 'Paper', 'cast': [],\n 'through_line': {'name': 'seed', 'asset': 'seed_prop', 'progression': 'grows', 'resolution': 'plant'},\n 'arc': {'hook': 1, 'build': [], 'turn': 2, 'payoff': 3},\n 'blocks': [{'n': n, 'arc_role': role,\n 'vo_line': (words(6, 0) + '. ' + words(16, 300) if n == 1 else words(22, 1000 * n)),\n 'location': location, 'through_line_state': f'stage {n}',\n 'shots': shots(n * 400), 'assets_used': [location, 'seed_prop']}\n for n, role, location in [(1, 'hook', 'soil'), (2, 'turn', 'garden'), (3, 'payoff', 'greenhouse')]],\n 'sources': ['https://example.com/plants']}\n result = validator._validate(script, 30)\n assert result['valid'], result\n with tempfile.TemporaryDirectory() as tmp:\n path = Path(tmp) / 'script.json'\n path.write_text(json.dumps(script))\n command = ['python3', str(root / 'validate_motion_script.py'), '--script', str(path), '--duration-seconds', '30']\n assert subprocess.run(command, capture_output=True).returncode == 0\n assert (path.parent / 'script.lock').is_file()\n assert subprocess.run(['python3', str(root / 'validate_motion_script.py'), '--script', str(path),\n '--duration-seconds', '10', '--verify-lock'], capture_output=True).returncode != 0\n changed = json.loads(path.read_text())\n changed['blocks'][1]['vo_line'] += ' changed'\n path.write_text(json.dumps(changed))\n assert subprocess.run(command + ['--verify-lock'], capture_output=True).returncode != 0\n digest = validator._narration_digest(script)\n script['blocks'][1]['vo_line'] += ' changed'\n assert validator._narration_digest(script) != digest\n script['blocks'][1]['assets_used'] = ['garden']\n assert not validator._validate(script, 30)['valid']\n print('channel script validation, narration lock and missing-asset gate passed')\n\n\nif __name__ == '__main__':\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":3053,"content_sha256":"b0405fce4e8692bdfcb3261c917dcf06b9aa26cce61497b468ecb1f9bf449fc4"},{"filename":"scripts/tests/test_modes.py","content":"#!/usr/bin/env python3\n\"\"\"Exercise picture, song, edit and presenter assembly with local fixtures.\"\"\"\nimport json\nimport hashlib\nimport shutil\nimport subprocess\nimport sys\nimport tempfile\nfrom pathlib import Path\n\n\ndef run(*args, check=True):\n return subprocess.run(args, check=check, stdout=subprocess.PIPE, stderr=subprocess.PIPE)\n\n\ndef run_script(name, *args, check=True):\n return run(\"bash\", str(SCRIPTS / name), *(str(arg) for arg in args), check=check)\n\n\ndef duration(path):\n return float(run(\"ffprobe\", \"-v\", \"error\", \"-show_entries\", \"format=duration\",\n \"-of\", \"default=nw=1:nk=1\", str(path)).stdout)\n\n\ndef color(path, second):\n output = run(\"ffmpeg\", \"-v\", \"error\", \"-ss\", str(second), \"-i\", str(path),\n \"-frames:v\", \"1\", \"-vf\", \"scale=1:1\", \"-pix_fmt\", \"rgb24\", \"-f\", \"rawvideo\", \"-\").stdout\n return tuple(output[:3])\n\n\nSCRIPTS = Path(__file__).resolve().parents[1]\n\n\ndef main():\n for tool in (\"bash\", \"ffmpeg\", \"ffprobe\", \"python3\"):\n if not shutil.which(tool):\n raise RuntimeError(f\"missing {tool}\")\n with tempfile.TemporaryDirectory(prefix=\"faceless-modes-\") as tmp:\n p = Path(tmp)\n clip, source, song = p / \"clip.mp4\", p / \"source.mp4\", p / \"song.wav\"\n run(\"ffmpeg\", \"-v\", \"error\", \"-y\", \"-f\", \"lavfi\", \"-i\", \"testsrc2=size=320x180:rate=24:duration=10\",\n \"-f\", \"lavfi\", \"-i\", \"sine=frequency=440:duration=10\", \"-c:v\", \"libx264\", \"-pix_fmt\", \"yuv420p\",\n \"-c:a\", \"aac\", \"-shortest\", str(clip))\n run(\"ffmpeg\", \"-v\", \"error\", \"-y\", \"-f\", \"lavfi\", \"-i\", \"color=red:size=320x180:rate=24:duration=30\",\n \"-f\", \"lavfi\", \"-i\", \"sine=frequency=220:duration=30\", \"-t\", \"30\", \"-c:v\", \"libx264\", \"-c:a\", \"aac\", str(source))\n run(\"ffmpeg\", \"-v\", \"error\", \"-y\", \"-f\", \"lavfi\", \"-i\", \"sine=frequency=330:duration=60\",\n \"-af\", \"volume=3\", \"-ar\", \"24000\", str(song))\n manifest = p / \"clips.txt\"\n manifest.write_text((str(clip) + \"\\n\") * 6)\n result = p / \"song.mp4\"\n run_script(\"assemble_final.sh\", \"--out\", result, \"--blocks\", 6, \"--manifest\", manifest, \"--song\", song)\n assert abs(duration(result) - 60) \u003c 0.2\n negative = run_script(\"assemble_final.sh\", \"--out\", p / \"bad-song.mp4\", \"--blocks\", 5,\n \"--manifest\", manifest, \"--song\", song, check=False)\n assert negative.returncode != 0\n\n dialogue = p / \"dialogue01.wav\"\n run(\"ffmpeg\", \"-v\", \"error\", \"-y\", \"-f\", \"lavfi\", \"-i\", \"sine=frequency=400:duration=8.2\",\n \"-af\", \"adelay=1000:all=1,apad\", \"-t\", \"10\", str(dialogue))\n talking_script = p / \"talking.json\"\n talking_script.write_text(json.dumps({\"talking_characters\": True, \"blocks\": [{\"block_kind\": \"dialogue\"}]}))\n talking_out = p / \"talking.mp4\"\n assert run_script(\"assemble_final.sh\", \"--out\", talking_out, \"--blocks\", 1,\n \"--script\", talking_script, \"--allow-mismatch\", clip, dialogue, check=False).returncode != 0\n (p / \"script.lock\").write_text(json.dumps({\"narration_sha256\": hashlib.sha256(b\"\").hexdigest(),\n \"block_count\": 1, \"duration_seconds\": 10}))\n run_script(\"assemble_final.sh\", \"--out\", talking_out, \"--blocks\", 1,\n \"--script\", talking_script, \"--allow-mismatch\", clip, dialogue)\n timing = json.loads(Path(str(talking_out) + \".assembly.json\").read_text())[\"per_block\"][0]\n assert abs(timing[\"speech_abs_s\"] - 1.0) \u003c 0.15, timing\n assert timing[\"voice\"] == \"dialogue01.wav\", timing\n muted_out = p / \"talking-muted.mp4\"\n run_script(\"assemble_final.sh\", \"--out\", muted_out, \"--blocks\", 1, \"--sfx-vol\", 0,\n \"--script\", talking_script, \"--allow-mismatch\", clip, dialogue)\n def audio_bytes(path):\n return run(\"ffmpeg\", \"-v\", \"error\", \"-i\", str(path), \"-vn\", \"-f\", \"s16le\", \"-\").stdout\n assert audio_bytes(talking_out) == audio_bytes(muted_out), \"dialogue clip audio was mixed twice\"\n (p / \"script.lock\").unlink()\n soft = p / \"soft.wav\"\n run(\"ffmpeg\", \"-v\", \"error\", \"-y\", \"-f\", \"lavfi\", \"-i\", \"sine=frequency=300:duration=7.5\", \"-af\", \"volume=3\", str(soft))\n assert run_script(\"assemble_final.sh\", \"--out\", p / \"soft.mp4\", \"--blocks\", 1,\n \"--allow-mismatch\", clip, soft, check=False).returncode != 0\n accepted = run_script(\"assemble_final.sh\", \"--out\", p / \"soft.mp4\", \"--blocks\", 1,\n \"--accept-soft-blocks\", \"1\", \"--allow-mismatch\", clip, soft, check=False)\n assert b\"accepted retried voice 1\" in accepted.stderr\n voice = p / \"voice.wav\"\n run(\"ffmpeg\", \"-v\", \"error\", \"-y\", \"-f\", \"lavfi\", \"-i\", \"sine=frequency=300:duration=8.8\", str(voice))\n frames = []\n for i in range(1, 9):\n frame = p / f\"frame{i:03}.png\"\n run(\"ffmpeg\", \"-v\", \"error\", \"-y\", \"-ss\", str(i), \"-i\", str(clip), \"-frames:v\", \"1\", str(frame))\n frames.append(frame)\n pictures = p / \"frames.txt\"\n pictures.write_text(\"\".join(f\"{frame} 1.1\\n\" for frame in frames))\n stills = p / \"stills.mp4\"\n run_script(\"assemble_slides.sh\", \"--out\", stills, \"--audio\", voice, \"--blocks\", 8, \"--manifest\", pictures)\n assert abs(duration(stills) - 8.8) \u003c 0.3\n pictures.write_text(\"\".join(f\"{frames[0]} 1.1\\n\" for _ in frames))\n assert run_script(\"assemble_slides.sh\", \"--out\", p / \"bad-stills.mp4\", \"--audio\", voice,\n \"--blocks\", 8, \"--manifest\", pictures, check=False).returncode != 0\n\n patch = p / \"patch.mp4\"\n run(\"ffmpeg\", \"-v\", \"error\", \"-y\", \"-f\", \"lavfi\", \"-i\", \"color=blue:size=320x180:rate=24:duration=10\",\n \"-f\", \"lavfi\", \"-i\", \"sine=frequency=330:duration=10\", \"-t\", \"10\", \"-c:v\", \"libx264\", \"-c:a\", \"aac\", str(patch))\n patches = p / \"patches.json\"\n patches.write_text(json.dumps([{\"start_seconds\": 10, \"end_seconds\": 20, \"patch_video\": str(patch)}]))\n edited = p / \"edited.mp4\"\n run_script(\"assemble_partial_edit.sh\", \"--source\", source, \"--source-seconds\", 30,\n \"--requested-seconds\", 30, \"--manifest\", patches, \"--out\", edited)\n assert abs(duration(edited) - 30) \u003c 0.2\n assert color(edited, 5)[0] > 200 and color(edited, 15)[2] > 200 and color(edited, 25)[0] > 200\n patches.write_text(json.dumps([{\"start_seconds\": 5, \"end_seconds\": 15, \"patch_video\": str(patch)}]))\n assert run_script(\"assemble_partial_edit.sh\", \"--source\", source, \"--source-seconds\", 30,\n \"--requested-seconds\", 30, \"--manifest\", patches, \"--out\", p / \"bad-edit.mp4\", check=False).returncode != 0\n\n if \"--skip-presenter\" not in sys.argv:\n talk = p / \"talk.mp4\"\n run(\"ffmpeg\", \"-v\", \"error\", \"-y\", \"-f\", \"lavfi\", \"-i\", \"color=green:s=320x480:r=24:d=10\",\n \"-f\", \"lavfi\", \"-i\", \"sine=frequency=220:duration=10\", \"-filter_complex\",\n \"[0:v]drawbox=x=110:y=100:w=100:h=320:color=red:t=fill[v]\", \"-map\", \"[v]\", \"-map\", \"1:a\",\n \"-c:v\", \"libx264\", \"-pix_fmt\", \"yuv420p\", \"-c:a\", \"aac\", \"-shortest\", str(talk))\n reference = p / \"reference.png\"\n run(\"ffmpeg\", \"-v\", \"error\", \"-y\", \"-i\", str(talk), \"-frames:v\", \"1\", str(reference))\n assert run_script(\"trim_static_head.sh\", talk, p / \"frozen.mp4\", reference, check=False).returncode != 0\n composite = p / \"presenter.mp4\"\n run_script(\"presenter_composite.sh\", clip, talk, composite, \"--style\", \"cutout\")\n assert json.loads(Path(str(composite) + \".qc.json\").read_text())[\"result\"] == \"PASS\"\n assert (p / \"presenter_voice.wav\").exists()\n print(\"picture, song and edit fixtures passed\" + (\"; presenter skipped\" if \"--skip-presenter\" in sys.argv else \"; presenter passed\"))\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":7966,"content_sha256":"865eb3e46d583c8e04fe5d55b2f9ff16bbb04156a9687b13a61fea137a121098"},{"filename":"scripts/tests/test_song.py","content":"#!/usr/bin/env python3\n\"\"\"Check duration and language-aware prompt validation.\"\"\"\nimport importlib.util\nfrom pathlib import Path\n\npath = Path(__file__).resolve().parents[1] / \"validate_song_prompt.py\"\nspec = importlib.util.spec_from_file_location(\"song\", path)\nmodule = importlib.util.module_from_spec(spec)\nspec.loader.exec_module(module)\nbase = (\"A full 1-minute song, 100 BPM, steady constant beat, 4/4, no tempo changes, \"\n \"8 syllables, one syllable per beat, beat one, 4-bar phrases, chorus repeats verbatim. \")\nfor language, line in ((\"en\", \"The little bird can sing today\"),\n (\"ru\", \"На лугу растёт зелёный сад\"),\n (\"ja\", \"春の風が吹いています\")):\n result = module.validate(base + '\"' + line + '\"', duration_seconds=60, lyrics_language=language)\n assert result[\"valid\"], result\n assert result[\"manual_meter_review_required\"] == (language == \"ja\")\nassert not module.validate_duration(90)[\"valid\"]\nassert not module.validate(base, duration_seconds=60)[\"valid\"]\nprint(\"song duration, lyric and language checks passed\")\n","content_type":"text/x-python; charset=utf-8","language":"python","size":1116,"content_sha256":"044ef7c579a72e5e7f8350fd9cc74eb2556e90bfd09695c436893200ac772036"},{"filename":"scripts/tests/test_timeline.py","content":"#!/usr/bin/env python3\n\"\"\"Check timed scene planning and generated frame binding.\"\"\"\nimport importlib.util\nfrom pathlib import Path\n\nroot = Path(__file__).resolve().parents[1]\n\ndef load(name):\n spec = importlib.util.spec_from_file_location(name, root / (name + \".py\"))\n module = importlib.util.module_from_spec(spec)\n spec.loader.exec_module(module)\n return module\n\n\ndef main():\n timeline = load(\"build_scene_timeline\")\n binding = load(\"bind_scene_frame_results\")\n script = {\"beats\": [{\"n\": 1, \"phrase\": \"Seeds need water\"},\n {\"n\": 2, \"phrase\": \"Leaves catch sunlight\"}]}\n words = {\"words\": [{\"word\": word, \"start\": i * .8, \"end\": i * .8 + .6}\n for i, word in enumerate(\"Seeds need water Leaves catch sunlight\".split())]}\n plan = timeline.build_scene_manifest(script, words, 5.0, 5.0)\n frames = plan[\"frames\"]\n assert frames and all(f[\"duration\"] \u003c= 1.5 for f in frames)\n results = {\"results\": [{\"output_slot\": f[\"output_slot\"], \"job_id\": f\"id-{f['n']}\",\n \"result_url\": f\"https://example.test/{f['n']}.png\", \"status\": \"completed\",\n \"params\": {\"medias\": [{\"role\": \"image\", \"data\": {\"type\": \"image_job\",\n \"id\": f\"id-{f['n']-1}\"}}]}}\n for f in frames]}\n bound = binding.bind_scene_frame_results(plan, [results])\n for previous, current in zip(bound[\"frames\"], bound[\"frames\"][1:]):\n if current[\"image_mode\"] == \"variation\":\n assert current[\"reference_frame_job_id\"] == previous[\"frame_job_id\"]\n variation = next(i for i, f in enumerate(frames) if f[\"image_mode\"] == \"variation\")\n incorrect = {\"results\": [dict(r) for r in results[\"results\"]]}\n incorrect[\"results\"][variation] = {**incorrect[\"results\"][variation], \"params\": {\"medias\": []}}\n try:\n binding.bind_scene_frame_results(plan, [incorrect])\n except ValueError:\n pass\n else:\n raise AssertionError(\"unverified variation reference was accepted\")\n try:\n binding.bind_scene_frame_results(plan, [{\"results\": results[\"results\"][:-1]}])\n except ValueError:\n pass\n else:\n raise AssertionError(\"missing frame was accepted\")\n try:\n timeline.build_scene_manifest(script, words, 5.0, 20.0)\n except ValueError:\n pass\n else:\n raise AssertionError(\"wrong duration was accepted\")\n print(\"timeline alignment, frame binding and failure gates passed\")\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":2551,"content_sha256":"da541bb28954d5c4cba24d393fdedb5a71492f4649817d18a38f0aaade708934"},{"filename":"scripts/trim_static_head.sh","content":"#!/usr/bin/env bash\n# Usage: trim_static_head.sh input.mp4 output.mp4 reference.png\nset -euo pipefail\nIN=\"$1\"; OUT=\"$2\"; REF=\"${3:-}\"\nTRIM=$(python3 - \"$IN\" \"$REF\" \u003c\u003c'PY'\nimport subprocess, sys\nimport numpy as np\ninp, ref = sys.argv[1], sys.argv[2]\nprobe = subprocess.run([\"ffprobe\",\"-v\",\"error\",\"-select_streams\",\"v:0\",\n \"-show_entries\",\"stream=width,height\",\"-of\",\"csv=p=0\",inp],\n capture_output=True,text=True,check=True).stdout.strip().split(\",\")\nw0,h0=int(probe[0]),int(probe[1])\nh = int(round(h0 * 240 / w0)); h -= h % 2\n\nif ref:\n def rgb0(path):\n r = subprocess.run([\"ffmpeg\",\"-v\",\"error\",\"-i\",path,\"-frames:v\",\"1\",\n \"-vf\",f\"scale=240:{h}\",\"-f\",\"rawvideo\",\"-pix_fmt\",\"rgb24\",\"-\"],\n capture_output=True, check=True).stdout\n return np.frombuffer(r[:h*240*3],dtype=np.uint8).reshape(h,240,3).astype(int)\n diff = float(np.abs(rgb0(inp)-rgb0(ref)).mean())\n if diff >= 20:\n print(f\"SEEDCHK: clean (frame0-vs-ref diff {diff:.1f}) 0.0\", file=sys.stderr)\n print(\"0.0\"); sys.exit()\n print(f\"SEEDCHK: SEEDED (frame0-vs-ref diff {diff:.1f})\", file=sys.stderr)\n\np = subprocess.run([\"ffmpeg\",\"-v\",\"error\",\"-t\",\"1.6\",\"-i\",inp,\n \"-vf\",\"fps=10,scale=240:-2,format=gray\",\"-f\",\"rawvideo\",\"-\"],\n capture_output=True, check=True)\nbuf = p.stdout\nn = len(buf)//(240*h)\nfr = np.frombuffer(buf[:n*240*h],dtype=np.uint8).reshape(n,h,240).astype(int)\nd = [float(np.abs(fr[i]-fr[0]).mean()) for i in range(1,n)]\nif not d: print(\"0.0\"); sys.exit()\npeak = max(d)\nif peak \u003c 5: # no motion in the probe window\n if ref:\n print(\"DEAD: clip frozen through the probe window — regenerate this block\", file=sys.stderr)\n sys.exit(2)\n print(\"0.0\"); sys.exit()\nthr = min(max(peak*0.6, 8), 15)\nt = 0.0\nfor i,v in enumerate(d):\n if v > thr:\n t = (i+1)/10.0\n break\nprint(f\"{t:.1f}\")\nPY\n)\necho \"TRIM: ${TRIM}s\"\nif [[ \"$TRIM\" == \"0.0\" ]]; then cp \"$IN\" \"$OUT\"; exit 0; fi\nDUR=$(ffprobe -v error -show_entries format=duration -of csv=p=0 \"$IN\")\nffmpeg -y -v error -ss \"$TRIM\" -i \"$IN\" -filter_complex \\\n \"[0:v]tpad=stop_mode=clone:stop_duration=${TRIM}[v];[0:a]apad=pad_dur=${TRIM}[a]\" \\\n -map \"[v]\" -map \"[a]\" -t \"$DUR\" -c:v libx264 -crf 18 -preset veryfast -pix_fmt yuv420p -c:a aac \"$OUT\"\n","content_type":"application/x-sh; charset=utf-8","language":"bash","size":2269,"content_sha256":"aced652ae5cba68a83b9d08a704797fe457722c2ee1912cb3f2b1d1e2d9d9a03"},{"filename":"scripts/validate_motion_script.py","content":"#!/usr/bin/env python3\n\"\"\"Validate fully-animated faceless script gates before generation.\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport hashlib\nimport json\nimport math\nimport re\nimport sys\nfrom pathlib import Path\nfrom typing import Any\nfrom urllib.parse import urlparse\n\n\nWORD_PATTERN = re.compile(r\"[^\\W_]+(?:[-'’][^\\W_]+)*\", re.UNICODE)\nTTS_WRAPPER_PATTERN = re.compile(r\"^\\s*\\[|\\[\\s*00:\\d{2}(?::\\d{2})?-\", re.UNICODE)\nARC_ROLES = {\"hook\", \"build\", \"turn\", \"payoff\"}\n\nDEFAULT_WORD_BUDGET = (20, 23)\nDURATION_RETRY_WORD_MIN = 17\nSPEECH_TARGET_WINDOW = (7.8, 9.5)\nSPEECH_HARD_WINDOW = (7.2, 9.5)\nWORD_BUDGETS = {\"kids\": (17, 21)} # a child needs room to follow, so Kids sits one\n\nMAX_SENTENCES_PER_LINE = 2\nSENTENCE_SPLIT_PATTERN = re.compile(r\"[.!?]+(?:\\s|$)\", re.UNICODE)\n\nDIGIT_PATTERN = re.compile(r\"\\d\", re.UNICODE)\n\nDIALOGUE_WORD_BUDGET = (16, 22)\nBLOCK_KINDS = {\"narration\", \"dialogue\"}\nMAX_DIALOGUE_SPEAKERS = 2\n\nFILLER_PATTERN = re.compile(\n r\"\\b(?:you\\s+know|y'?know|i\\s+mean|sort\\s+of|kinda|basically|um+|uh+|erm)\\b\",\n re.IGNORECASE | re.UNICODE,\n)\n\nREPEAT_WINDOW = 6\nREPEAT_STOPWORDS = {\n \"a\", \"an\", \"and\", \"are\", \"as\", \"at\", \"be\", \"but\", \"by\", \"can\", \"did\", \"do\",\n \"does\", \"for\", \"from\", \"had\", \"has\", \"have\", \"he\", \"her\", \"here\", \"his\",\n \"how\", \"i\", \"if\", \"in\", \"into\", \"is\", \"it\", \"its\", \"just\", \"me\", \"my\", \"no\",\n \"not\", \"of\", \"on\", \"one\", \"or\", \"our\", \"out\", \"over\", \"she\", \"so\", \"than\",\n \"that\", \"the\", \"their\", \"them\", \"then\", \"there\", \"these\", \"they\", \"this\",\n \"those\", \"to\", \"up\", \"was\", \"we\", \"were\", \"what\", \"when\", \"why\", \"will\",\n \"with\", \"would\", \"you\", \"your\",\n}\n\n\nSHARED_PHRASE_WORDS = 5\nMAX_REPORTED_SHARED_PHRASES = 6\n\nSHOT_SIZE_VOCABULARY = {\n \"ecu\": \"ECU\",\n \"xcu\": \"ECU\",\n \"extreme close-up\": \"ECU\",\n \"close-up\": \"CU\",\n \"closeup\": \"CU\",\n \"cu\": \"CU\",\n \"macro\": \"MACRO\",\n \"detail\": \"DETAIL\",\n \"insert\": \"DETAIL\",\n \"medium\": \"MEDIUM\",\n \"mid\": \"MEDIUM\",\n \"ms\": \"MEDIUM\",\n \"ots\": \"OTS\",\n \"over-the-shoulder\": \"OTS\",\n \"two-shot\": \"TWO-SHOT\",\n \"wide\": \"WIDE\",\n \"full-spread\": \"WIDE\",\n \"full spread\": \"WIDE\",\n \"establishing\": \"WIDE\",\n \"master\": \"WIDE\",\n \"overhead\": \"TOP\",\n \"top-down\": \"TOP\",\n \"aerial\": \"TOP\",\n}\nESTABLISHING_SHOT_SIZES = {\"WIDE\", \"TOP\"}\nSHOT_SIZE_LEAD_CHARACTERS = 48\nSHARED_SHOT_WORDS = 8\nNEAR_BLOCK_DISTANCE = 3\nHOOK_SENTENCE_WORDS = 8\nHOOK_SENTENCE_GENRES = {\"education\", \"history\"}\n_SHOT_SIZE_PATTERN = re.compile(\n \"|\".join(\n rf\"\\b{re.escape(key)}\\b\"\n for key in sorted(SHOT_SIZE_VOCABULARY, key=len, reverse=True)\n ),\n re.IGNORECASE,\n)\n\n\nNUMBER_WORDS = {\n \"zero\", \"one\", \"two\", \"three\", \"four\", \"five\", \"six\", \"seven\", \"eight\", \"nine\",\n \"ten\", \"eleven\", \"twelve\", \"thirteen\", \"fourteen\", \"fifteen\", \"sixteen\",\n \"seventeen\", \"eighteen\", \"nineteen\", \"twenty\", \"thirty\", \"forty\", \"fifty\",\n \"sixty\", \"seventy\", \"eighty\", \"ninety\", \"hundred\", \"thousand\", \"million\",\n \"billion\", \"trillion\", \"first\", \"second\", \"third\", \"fourth\", \"fifth\", \"sixth\",\n \"seventh\", \"eighth\", \"ninth\", \"tenth\", \"half\", \"quarter\", \"dozen\",\n}\n\n\ndef _word_count(text: str) -> int:\n return len(WORD_PATTERN.findall(text))\n\n\ndef _normalised_words(text: str) -> list[str]:\n return [token.casefold() for token in WORD_PATTERN.findall(text)]\n\n\ndef _longest_shared_run(left: list[str], right: list[str]) -> list[str]:\n \"\"\"Longest verbatim word run present in both lines.\"\"\"\n best_end = 0\n best_length = 0\n previous = [0] * (len(right) + 1)\n for left_index in range(1, len(left) + 1):\n current = [0] * (len(right) + 1)\n for right_index in range(1, len(right) + 1):\n if left[left_index - 1] == right[right_index - 1]:\n current[right_index] = previous[right_index - 1] + 1\n if current[right_index] > best_length:\n best_length = current[right_index]\n best_end = left_index\n previous = current\n if best_length \u003c SHARED_PHRASE_WORDS:\n return []\n return left[best_end - best_length : best_end]\n\n\ndef _shot_size(text: str) -> str | None:\n \"\"\"Canonical framing size of a shot, read from the head of its description.\"\"\"\n match = _SHOT_SIZE_PATTERN.search(text[:SHOT_SIZE_LEAD_CHARACTERS])\n if match is None:\n return None\n return SHOT_SIZE_VOCABULARY[match.group(0).casefold()]\n\n\ndef _first_sentence(text: str) -> str:\n parts = [part for part in SENTENCE_SPLIT_PATTERN.split(text.strip()) if part.strip()]\n return parts[0] if parts else text.strip()\n\n\ndef _narration_digest(script: Any) -> str:\n \"\"\"Digest of the narration only, so formatting is free and words are locked.\"\"\"\n blocks = script.get(\"blocks\") if isinstance(script, dict) else None\n lines = []\n if isinstance(blocks, list):\n for block in blocks:\n line = block.get(\"vo_line\") if isinstance(block, dict) else None\n lines.append(\" \".join(_normalised_words(line if isinstance(line, str) else \"\")))\n return hashlib.sha256(\"\\n\".join(lines).encode(\"utf-8\")).hexdigest()\n\n\ndef _script_lock(script: Any, duration_seconds: float) -> dict[str, Any]:\n blocks = script.get(\"blocks\") if isinstance(script, dict) else None\n return {\n \"narration_sha256\": _narration_digest(script),\n \"block_count\": len(blocks) if isinstance(blocks, list) else 0,\n \"duration_seconds\": duration_seconds,\n }\n\n\ndef _sentence_count(text: str) -> int:\n parts = [part for part in SENTENCE_SPLIT_PATTERN.split(text.strip()) if part.strip()]\n return max(1, len(parts))\n\n\ndef _filler_hits(text: str) -> list[str]:\n seen: list[str] = []\n for match in FILLER_PATTERN.finditer(text):\n phrase = \" \".join(match.group(0).split()).casefold()\n if phrase not in seen:\n seen.append(phrase)\n return seen\n\n\ndef _near_repeats(text: str) -> list[str]:\n tokens = [token.casefold() for token in WORD_PATTERN.findall(text)]\n repeats: list[str] = []\n for index, token in enumerate(tokens):\n if token in REPEAT_STOPWORDS or token in NUMBER_WORDS or len(token) \u003c 3:\n continue\n if token in tokens[index + 1 : index + REPEAT_WINDOW] and token not in repeats:\n repeats.append(token)\n return repeats\n\n\ndef _is_http_url(value: Any) -> bool:\n if not isinstance(value, str):\n return False\n parsed = urlparse(value)\n return parsed.scheme in {\"http\", \"https\"} and bool(parsed.netloc)\n\n\ndef _validate(\n script: dict,\n duration_seconds: float,\n word_budget: tuple[int, int] | None = None,\n duration_retry_blocks: set[int] | None = None,\n) -> dict:\n errors: list[str] = []\n invalid_blocks: list[dict[str, Any]] = []\n if not isinstance(script, dict):\n return {\n \"valid\": False,\n \"errors\": [\"script must be an object\"],\n \"invalid_blocks\": [],\n }\n\n duration_retry_blocks = duration_retry_blocks or set()\n genre = script.get(\"genre\")\n if genre not in {\"education\", \"history\", \"kids\", \"storytelling\"}:\n errors.append(\n \"genre must be education, history, kids, or storytelling\"\n )\n\n talking_characters = script.get(\"talking_characters\", False)\n if not isinstance(talking_characters, bool):\n errors.append(\"talking_characters must be a boolean\")\n talking_characters = False\n if talking_characters and genre != \"kids\":\n errors.append(\"talking_characters is a Kids-only direction\")\n\n expected_blocks = math.ceil(duration_seconds / 10)\n blocks = script.get(\"blocks\")\n if not isinstance(blocks, list) or not blocks:\n return {\n \"valid\": False,\n \"errors\": [*errors, \"script must contain a non-empty blocks array\"],\n \"invalid_blocks\": [],\n }\n if len(blocks) != expected_blocks:\n errors.append(\n f\"block count {len(blocks)} does not match expected {expected_blocks}\"\n )\n unknown_retry_blocks = sorted(\n block for block in duration_retry_blocks if block > len(blocks)\n )\n if unknown_retry_blocks:\n errors.append(\n \"duration retry blocks do not exist: \"\n + \", \".join(str(block) for block in unknown_retry_blocks)\n )\n\n through_line = script.get(\"through_line\")\n if not isinstance(through_line, dict):\n errors.append(\"through_line must be an object\")\n through_line = {}\n for field in (\"name\", \"asset\", \"progression\", \"resolution\"):\n value = through_line.get(field)\n if not isinstance(value, str) or not value.strip():\n errors.append(f\"through_line.{field} is required\")\n through_line_asset = through_line.get(\"asset\")\n\n cast = script.get(\"cast\")\n cast_members: list[tuple[str, str]] = []\n if not isinstance(cast, list):\n errors.append(\n \"cast must be an array of {name, asset} — the characters Phase 2 generated; \"\n \"use [] only for a video with no characters\"\n )\n else:\n for position, member in enumerate(cast, start=1):\n if not isinstance(member, dict):\n errors.append(f\"cast[{position}] must be an object with name and asset\")\n continue\n name = member.get(\"name\")\n asset = member.get(\"asset\")\n if not isinstance(name, str) or not name.strip():\n errors.append(f\"cast[{position}].name is required\")\n if not isinstance(asset, str) or not asset.strip():\n errors.append(f\"cast[{position}].asset is required\")\n if isinstance(name, str) and name.strip() and isinstance(asset, str) and asset.strip():\n cast_members.append((name.strip(), asset.strip()))\n\n arc = script.get(\"arc\")\n errors_before_arc = len(errors)\n if not isinstance(arc, dict):\n errors.append(\"arc must be an object\")\n arc = {}\n if arc.get(\"hook\") != 1:\n errors.append(\"arc.hook must be block 1\")\n if arc.get(\"payoff\") != expected_blocks:\n errors.append(f\"arc.payoff must be block {expected_blocks}\")\n if expected_blocks >= 3 and not isinstance(arc.get(\"turn\"), int):\n errors.append(\"arc.turn must name the turn block\")\n if not isinstance(arc.get(\"build\"), list):\n errors.append(\"arc.build must be an array of build block numbers\")\n if len(errors) > errors_before_arc:\n build_blocks = [\n index\n for index, block in enumerate(blocks, start=1)\n if isinstance(block, dict) and block.get(\"arc_role\") == \"build\"\n ]\n errors.append(\n \"expected arc shape, block numbers as integers and nothing else: \"\n '{\"hook\": 1, \"build\": '\n + json.dumps(build_blocks or [2])\n + ', \"turn\": \u003cblock number>, \"payoff\": '\n + str(expected_blocks)\n + \"} — there are no hook_block/build_blocks/turn_block keys and no prose \"\n \"in these fields; the arc text belongs in the blocks themselves\"\n )\n\n sources = script.get(\"sources\")\n if not isinstance(sources, list):\n errors.append(\"sources must be an array\")\n elif any(not _is_http_url(source) for source in sources):\n errors.append(\"sources must contain only absolute HTTP URLs\")\n elif genre in {\"education\", \"history\"} and not sources:\n errors.append(\"factual Education and History scripts require source URLs\")\n\n expected_shot_count = 4 if genre == \"kids\" else 5\n locations: list[str | None] = []\n arc_roles: list[str | None] = []\n narration_lines: list[tuple[int, list[str]]] = []\n block_shot_sizes: list[tuple[int, str | None]] = []\n block_shot_texts: list[tuple[int, int, list[str]]] = []\n block_sizes_by_number: dict[int, tuple[str | None, ...]] = {}\n block_renders: list[tuple[int, str, tuple[str, ...]]] = []\n for index, raw_block in enumerate(blocks, start=1):\n if not isinstance(raw_block, dict):\n invalid_blocks.append(\n {\"n\": index, \"errors\": [\"block must be an object\"]}\n )\n locations.append(None)\n arc_roles.append(None)\n continue\n\n block_errors: list[str] = []\n if raw_block.get(\"n\") != index:\n block_errors.append(f\"n must be {index}\")\n\n voice_line = raw_block.get(\"vo_line\")\n word_count = _word_count(voice_line) if isinstance(voice_line, str) else 0\n if isinstance(voice_line, str) and voice_line.strip():\n narration_lines.append((index, _normalised_words(voice_line)))\n if not isinstance(voice_line, str) or not voice_line.strip():\n block_errors.append(\"vo_line is required\")\n elif TTS_WRAPPER_PATTERN.search(voice_line):\n block_errors.append(\n \"vo_line must contain authored narration only, without \"\n \"TTS delivery instructions or timecodes\"\n )\n block_duration_seconds = 10.0\n if index == expected_blocks and duration_seconds % 10:\n block_duration_seconds = duration_seconds % 10\n\n block_kind = raw_block.get(\"block_kind\")\n if talking_characters:\n expected_kind = \"narration\" if index % 2 else \"dialogue\"\n if block_kind is None:\n block_errors.append(\n f\"block_kind is required on a talking_characters run \"\n f\"(expected '{expected_kind}')\"\n )\n elif block_kind not in BLOCK_KINDS:\n block_errors.append(\"block_kind must be narration or dialogue\")\n elif block_kind != expected_kind:\n block_errors.append(\n f\"block_kind must alternate: block {index} must be \"\n f\"'{expected_kind}'\"\n )\n speakers = raw_block.get(\"speakers\")\n if block_kind == \"dialogue\":\n if not isinstance(speakers, list) or not speakers:\n block_errors.append(\n \"a dialogue block must name its speakers (1-2 characters)\"\n )\n elif len(speakers) > MAX_DIALOGUE_SPEAKERS:\n block_errors.append(\n f\"a dialogue block takes at most {MAX_DIALOGUE_SPEAKERS} \"\n \"speaking characters\"\n )\n elif speakers:\n block_errors.append(\n \"a narration block has no on-screen speakers — the narrator is \"\n \"off screen and every mouth stays closed\"\n )\n elif block_kind is not None:\n block_errors.append(\n \"block_kind belongs to talking_characters runs only\"\n )\n\n if talking_characters and block_kind == \"dialogue\":\n budget_minimum, budget_maximum = DIALOGUE_WORD_BUDGET\n elif word_budget is not None:\n budget_minimum, budget_maximum = word_budget\n else:\n budget_minimum, budget_maximum = WORD_BUDGETS.get(\n genre if isinstance(genre, str) else \"\",\n DEFAULT_WORD_BUDGET,\n )\n if index in duration_retry_blocks:\n budget_minimum = min(budget_minimum, DURATION_RETRY_WORD_MIN)\n minimum_words = max(1, math.floor(budget_minimum * block_duration_seconds / 10))\n maximum_words = max(\n minimum_words, math.ceil(budget_maximum * block_duration_seconds / 10)\n )\n if (\n isinstance(voice_line, str)\n and voice_line.strip()\n and not minimum_words \u003c= word_count \u003c= maximum_words\n ):\n block_errors.append(\n f\"vo_line must contain {minimum_words}-{maximum_words} words\"\n )\n if isinstance(voice_line, str) and voice_line.strip():\n filler = _filler_hits(voice_line)\n if filler:\n block_errors.append(\n \"vo_line must not use conversational filler: \"\n + \", \".join(f'\"{phrase}\"' for phrase in filler)\n + \" — reach the word count with content, not padding\"\n )\n sentences = _sentence_count(voice_line)\n if sentences > MAX_SENTENCES_PER_LINE:\n block_errors.append(\n f\"vo_line packs {sentences} sentences into one block — keep it to \"\n f\"{MAX_SENTENCES_PER_LINE}; every extra period costs ~0.7s of dead \"\n \"air and the voice rushes the words to compensate\"\n )\n if DIGIT_PATTERN.search(voice_line):\n block_errors.append(\n \"vo_line contains digits — write numbers, dates and years as words \"\n '(\"nineteen seventy-seven\"), they are spoken longer than written'\n )\n repeats = _near_repeats(voice_line)\n if repeats:\n block_errors.append(\n \"vo_line repeats \"\n + \", \".join(f'\"{token}\"' for token in repeats)\n + f\" within {REPEAT_WINDOW} words — stacked epithets read as \"\n \"padding; keep one modifier per thing\"\n )\n\n arc_role = raw_block.get(\"arc_role\")\n if arc_role not in ARC_ROLES:\n block_errors.append(\"arc_role must be hook, build, turn, or payoff\")\n arc_roles.append(None)\n else:\n arc_roles.append(arc_role)\n\n location = raw_block.get(\"location\")\n if not isinstance(location, str) or not location.strip():\n block_errors.append(\"location is required\")\n locations.append(None)\n else:\n locations.append(location.strip())\n\n through_line_state = raw_block.get(\"through_line_state\")\n if not isinstance(through_line_state, str) or not through_line_state.strip():\n block_errors.append(\"through_line_state is required\")\n\n shots = raw_block.get(\"shots\")\n if not isinstance(shots, list) or len(shots) != expected_shot_count:\n block_errors.append(\n f\"shots must contain exactly {expected_shot_count} entries\"\n )\n elif any(not isinstance(shot, str) or not shot.strip() for shot in shots):\n block_errors.append(\"every shot must be a non-empty string\")\n else:\n sizes: list[str | None] = []\n for shot_index, shot in enumerate(shots, start=1):\n size = _shot_size(shot)\n sizes.append(size)\n if size is None:\n block_errors.append(\n f\"shot {shot_index} does not open with a framing size — start it \"\n \"with one of WIDE / FULL-SPREAD / OVERHEAD / MEDIUM / OTS / CU / \"\n \"ECU / MACRO / DETAIL, as in 'CU: the slat railing, dotted airflow \"\n \"lines drifting past'\"\n )\n elif size == \"OTS\" and \"shoulder\" not in shot.casefold():\n block_errors.append(\n f\"shot {shot_index} is an OTS but names no shoulder — an OTS needs \"\n \"a visible character whose shoulder or head sits in the foreground; \"\n \"over an object or a diagram it is just a medium shot\"\n )\n for shot_index in range(1, len(sizes)):\n previous, current = sizes[shot_index - 1], sizes[shot_index]\n if previous is not None and previous == current:\n block_errors.append(\n f\"shots {shot_index} and {shot_index + 1} are both {current} — \"\n \"every shot differs in SIZE and ANGLE from its neighbour, or the \"\n \"block reads as one held plate\"\n )\n block_shot_sizes.append((index, sizes[0] if sizes else None))\n block_sizes_by_number[index] = tuple(sizes)\n for shot_index, shot in enumerate(shots, start=1):\n block_shot_texts.append((index, shot_index, _normalised_words(shot)))\n\n assets_used = raw_block.get(\"assets_used\")\n if not isinstance(assets_used, list) or not assets_used:\n block_errors.append(\"assets_used must be a non-empty array\")\n else:\n if len(assets_used) > 7:\n block_errors.append(\"assets_used must contain at most 7 entries\")\n if (\n isinstance(through_line_asset, str)\n and through_line_asset not in assets_used\n ):\n block_errors.append(\n \"assets_used must include through_line.asset in every block\"\n )\n block_tokens = set(_normalised_words(voice_line if isinstance(voice_line, str) else \"\"))\n for shot in shots if isinstance(shots, list) else []:\n if isinstance(shot, str):\n block_tokens.update(_normalised_words(shot))\n for member_name, member_asset in cast_members:\n member_tokens = set(_normalised_words(member_name))\n if not member_tokens or not member_tokens \u003c= block_tokens:\n continue\n if member_asset not in assets_used:\n block_errors.append(\n f\"this block names {member_name} but assets_used does not carry \"\n f\"{member_asset} — a named character without their asset is a new \"\n f\"character, so attach it or take the name out of the block\"\n )\n if isinstance(location, str) and location:\n block_renders.append(\n (\n index,\n location,\n tuple(\n str(asset)\n for asset in assets_used\n if asset != through_line_asset\n ),\n )\n )\n\n if block_errors:\n invalid_blocks.append(\n {\n \"n\": raw_block.get(\"n\", index),\n \"word_count\": word_count,\n \"errors\": block_errors,\n }\n )\n\n if arc_roles:\n if arc_roles[0] != \"hook\":\n errors.append(\"the first block arc_role must be hook\")\n if arc_roles[-1] != \"payoff\":\n errors.append(\"the final block arc_role must be payoff\")\n if expected_blocks >= 3 and \"turn\" not in arc_roles[1:-1]:\n errors.append(\"the arc must contain a turn before the payoff\")\n if expected_blocks >= 4 and \"build\" not in arc_roles[1:-1]:\n errors.append(\"the arc must contain a build block\")\n if isinstance(arc.get(\"turn\"), int):\n turn_index = arc[\"turn\"] - 1\n if (\n turn_index \u003c 0\n or turn_index >= len(arc_roles)\n or arc_roles[turn_index] != \"turn\"\n ):\n errors.append(\"arc.turn must point to a block with arc_role turn\")\n if isinstance(arc.get(\"build\"), list):\n declared_build_blocks = arc[\"build\"]\n actual_build_blocks = [\n index\n for index, role in enumerate(arc_roles, start=1)\n if role == \"build\"\n ]\n if declared_build_blocks != actual_build_blocks:\n errors.append(\n \"arc.build must exactly match blocks with arc_role build\"\n )\n\n shared_phrase_pairs = 0\n for left_position in range(len(narration_lines)):\n left_number, left_words = narration_lines[left_position]\n for right_position in range(left_position + 1, len(narration_lines)):\n right_number, right_words = narration_lines[right_position]\n phrase = _longest_shared_run(left_words, right_words)\n if not phrase:\n continue\n shared_phrase_pairs += 1\n if shared_phrase_pairs \u003c= MAX_REPORTED_SHARED_PHRASES:\n errors.append(\n f\"blocks {left_number} and {right_number} share the verbatim phrase \"\n f\"'{' '.join(phrase)}' ({len(phrase)} words) — a line is never brought \"\n \"up to the word budget by appending or reusing a sentence; rewrite the \"\n \"short line with one more FACT and leave the other one alone\"\n )\n if shared_phrase_pairs > MAX_REPORTED_SHARED_PHRASES:\n errors.append(\n f\"{shared_phrase_pairs} block pairs share a verbatim phrase in total; the \"\n f\"first {MAX_REPORTED_SHARED_PHRASES} are listed above\"\n )\n\n seen_locations: set[str] = set()\n first_size_by_block = dict(block_shot_sizes)\n for position, location in enumerate(locations):\n if not isinstance(location, str) or not location:\n continue\n opening_size = first_size_by_block.get(position + 1)\n if location in seen_locations and opening_size in ESTABLISHING_SHOT_SIZES:\n errors.append(\n f\"block {position + 1} returns to location {location!r} and opens on \"\n f\"{opening_size} — only the FIRST block of a new location establishes it; \"\n \"a later block there opens on a fresh close, medium or coverage angle\"\n )\n seen_locations.add(location)\n\n repeated_shots = 0\n for left_position in range(len(block_shot_texts)):\n left_block, left_index, left_words = block_shot_texts[left_position]\n for right_position in range(left_position + 1, len(block_shot_texts)):\n right_block, right_index, right_words = block_shot_texts[right_position]\n if left_block == right_block:\n continue\n phrase = _longest_shared_run(left_words, right_words)\n if len(phrase) \u003c SHARED_SHOT_WORDS:\n continue\n repeated_shots += 1\n if repeated_shots \u003c= MAX_REPORTED_SHARED_PHRASES:\n errors.append(\n f\"block {left_block} shot {left_index} and block {right_block} shot \"\n f\"{right_index} share the verbatim phrase '{' '.join(phrase)}' \"\n f\"({len(phrase)} words) — the same plate described twice is the same \"\n \"plate; give the later block a different subject, size or angle\"\n )\n if repeated_shots > MAX_REPORTED_SHARED_PHRASES:\n errors.append(\n f\"{repeated_shots} shot pairs repeat a verbatim phrase in total; the first \"\n f\"{MAX_REPORTED_SHARED_PHRASES} are listed above\"\n )\n\n for left_position in range(len(block_renders)):\n left_block, left_location, left_assets = block_renders[left_position]\n for right_position in range(left_position + 1, len(block_renders)):\n right_block, right_location, right_assets = block_renders[right_position]\n if left_location != right_location or left_assets != right_assets:\n continue\n distance = right_block - left_block\n same_sizes = block_sizes_by_number.get(left_block) == block_sizes_by_number.get(\n right_block\n )\n if distance > NEAR_BLOCK_DISTANCE and not same_sizes:\n continue\n listed = \", \".join(left_assets) if left_assets else \"no assets beyond the through-line\"\n reason = (\n f\"only {distance} blocks apart\"\n if distance \u003c= NEAR_BLOCK_DISTANCE\n else \"and the same framing sizes in the same order\"\n )\n errors.append(\n f\"blocks {left_block} and {right_block} render from the same inputs — \"\n f\"location {left_location!r} with the same assets in the same order \"\n f\"({listed}), {reason}. The narration differs but the clip will not: give \"\n \"the later block a different location, a different asset order, or a \"\n \"different set of framing sizes\"\n )\n\n if genre in HOOK_SENTENCE_GENRES and narration_lines:\n hook_number, _ = narration_lines[0]\n first_line = blocks[0].get(\"vo_line\") if isinstance(blocks[0], dict) else None\n if hook_number == 1 and isinstance(first_line, str) and first_line.strip():\n opening = _first_sentence(first_line)\n opening_words = _word_count(opening)\n if opening_words > HOOK_SENTENCE_WORDS:\n errors.append(\n f\"the cold open is {opening_words} words before its first full stop \"\n f\"({opening.strip()!r}) — block 1 opens on a sentence of at most \"\n f\"{HOOK_SENTENCE_WORDS} words, the raw fact stated flat, and only then \"\n \"fills out to the block's word budget\"\n )\n\n for index in range(2, len(locations)):\n current = locations[index]\n if current and current == locations[index - 1] == locations[index - 2]:\n errors.append(\n f\"location {current!r} is used in more than 2 consecutive blocks\"\n )\n\n return {\n \"valid\": not errors and not invalid_blocks,\n \"expected_block_count\": expected_blocks,\n \"block_count\": len(blocks),\n \"errors\": errors,\n \"invalid_blocks\": invalid_blocks,\n }\n\n\ndef main() -> int:\n parser = argparse.ArgumentParser()\n parser.add_argument(\"--script\", required=True, type=Path)\n parser.add_argument(\"--duration-seconds\", required=True, type=float)\n parser.add_argument(\"--words-min\", type=int, default=None)\n parser.add_argument(\"--words-max\", type=int, default=None)\n parser.add_argument(\n \"--duration-retry-blocks\",\n type=str,\n default=\"\",\n help=(\n \"Comma-separated block numbers measured above the 9.5s hard ceiling. \"\n \"Only those blocks may use the adaptive 17-word retry floor.\"\n ),\n )\n parser.add_argument(\n \"--lock\",\n type=Path,\n default=None,\n help=(\n \"Where the script lock lives (default: script.lock beside --script). \"\n \"A passing run writes it; assemble_final.sh refuses to work with narration \"\n \"whose digest does not match it, which is what makes an edit after \"\n \"SCRIPT LOCK impossible instead of merely forbidden.\"\n ),\n )\n parser.add_argument(\n \"--verify-lock\",\n action=\"store_true\",\n help=(\n \"Only compare the narration against an existing lock and exit; no gates are \"\n \"run. This is the mode assemble_final.sh calls.\"\n ),\n )\n arguments = parser.parse_args()\n if arguments.duration_seconds \u003c= 0:\n parser.error(\"--duration-seconds must be positive\")\n\n word_budget = None\n if arguments.words_min is not None or arguments.words_max is not None:\n if arguments.words_min is None or arguments.words_max is None:\n parser.error(\"--words-min and --words-max go together\")\n if arguments.words_min > arguments.words_max:\n parser.error(\"--words-min must not exceed --words-max\")\n word_budget = (arguments.words_min, arguments.words_max)\n try:\n duration_retry_blocks = {\n int(item.strip())\n for item in arguments.duration_retry_blocks.split(\",\")\n if item.strip()\n }\n except ValueError:\n parser.error(\"--duration-retry-blocks contains integers only\")\n if any(block \u003c 1 for block in duration_retry_blocks):\n parser.error(\"--duration-retry-blocks uses 1-based block numbers\")\n\n try:\n script = json.loads(arguments.script.read_text(encoding=\"utf-8\"))\n except (OSError, json.JSONDecodeError) as error:\n print(\n json.dumps(\n {\"valid\": False, \"errors\": [str(error)], \"invalid_blocks\": []},\n ensure_ascii=False,\n )\n )\n return 1\n\n lock_path = arguments.lock or arguments.script.with_name(\"script.lock\")\n\n if arguments.verify_lock:\n expected = _script_lock(script, arguments.duration_seconds)\n try:\n recorded = json.loads(lock_path.read_text(encoding=\"utf-8\"))\n except (OSError, json.JSONDecodeError) as error:\n print(\n json.dumps(\n {\n \"lock_ok\": False,\n \"errors\": [\n f\"no usable script lock at {lock_path}: {error} — run \"\n \"validate_motion_script.py on the manifest first\"\n ],\n },\n ensure_ascii=False,\n )\n )\n return 1\n matches = all(recorded.get(key) == expected[key] for key in (\n \"narration_sha256\", \"block_count\", \"duration_seconds\"\n ))\n payload: dict[str, Any] = {\n \"lock_ok\": matches,\n \"narration_sha256\": expected[\"narration_sha256\"],\n \"locked_sha256\": recorded.get(\"narration_sha256\"),\n }\n if not matches:\n payload[\"errors\"] = [\n \"the narration, block count or duration no longer matches the locked \"\n \"script. Revalidate the script and regenerate any affected takes.\"\n ]\n print(json.dumps(payload, ensure_ascii=False, indent=2))\n return 0 if matches else 1\n\n result = _validate(\n script,\n arguments.duration_seconds,\n word_budget,\n duration_retry_blocks,\n )\n if result[\"valid\"]:\n lock = _script_lock(script, arguments.duration_seconds)\n try:\n lock_path.write_text(\n json.dumps(lock, ensure_ascii=False, indent=2) + \"\\n\", encoding=\"utf-8\"\n )\n result[\"lock\"] = {\"path\": str(lock_path), **lock}\n except OSError as error:\n result[\"valid\"] = False\n result[\"errors\"].append(f\"could not write the script lock to {lock_path}: {error}\")\n print(json.dumps(result, ensure_ascii=False, indent=2))\n return 0 if result[\"valid\"] else 1\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":34275,"content_sha256":"6b712a296e8a46abce1b60457c2f0ccd38d503ab2a7652135c5e08bf4d2167db"},{"filename":"scripts/validate_song_prompt.py","content":"#!/usr/bin/env python3\n\"\"\"Validate the requested song duration and prompt.\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport json\nimport re\nimport sys\n\nMAX_PROMPT_CHARS = 2048\nALLOWED_BPM = {100, 112}\nTARGET_SYLLABLES = 8\nALLOWED_DURATION_SECONDS = frozenset({60, 120})\n\nREQUIRED_CLAUSES: tuple[tuple[str, str], ...] = (\n (r\"steady|constant\", \"state that the beat is STEADY AND CONSTANT from first second to last\"),\n (r\"\\b4/4\\b\", \"state the simple 4/4 metre\"),\n (r\"no\\s+tempo\\s+changes?\", 'state \"NO tempo changes\"'),\n (r\"8\\s+syllables|eight\\s+syllables\", \"state the 8-syllable line law\"),\n (r\"one\\s+syllable\\s+per\\s+beat\", 'state \"one syllable per beat\"'),\n (r\"beat\\s+one\", \"state that every line starts on beat one of its bar\"),\n (r\"4-?bar|four-?bar\", \"state the 4-bar phrase law\"),\n (r\"chorus\", \"name the chorus sections\"),\n (r\"verbatim|repeat\", \"state that the chorus repeats verbatim\"),\n)\n\nBANNED_TEMPO = (\n \"bridge\",\n \"slow down\",\n \"slowdown\",\n \"speed up\",\n \"speedup\",\n \"ritardando\",\n \"accelerando\",\n \"tempo change\",\n \"key change\",\n \"breakdown\",\n \"half-time\",\n \"double-time\",\n \"drop\",\n)\nBANNED_GROOVE = (\n \"four-on-the-floor\",\n \"four on the floor\",\n \"syncopation\",\n \"syncopated\",\n \"swung\",\n \"shuffle groove\",\n)\n\nVOWEL_GROUPS = re.compile(r\"[aeiouy]+\", re.IGNORECASE)\nWORD = re.compile(r\"[^\\W\\d_]+(?:['’][^\\W\\d_]+)*\", re.UNICODE)\n\n\ndef _syllables(word: str) -> int:\n \"\"\"Cheap English syllable estimate — good enough to catch a wandering line.\"\"\"\n word = word.lower().strip(\"'\")\n if not word:\n return 0\n groups = VOWEL_GROUPS.findall(word)\n count = len(groups)\n if word.endswith(\"e\") and count > 1 and not word.endswith((\"le\", \"ee\", \"ye\")):\n count -= 1\n return max(1, count)\n\n\ndef _line_syllables(line: str) -> int:\n return sum(_syllables(word) for word in WORD.findall(line))\n\n\ndef _lyric_lines(prompt: str) -> list[str]:\n lines: list[str] = []\n for quoted in re.findall(r'\"([^\"]{10,})\"', prompt):\n for part in re.split(r\"\\s*(?:/|\\||\\n)\\s*\", quoted):\n part = part.strip(\" ,.;:!?-\")\n if len(WORD.findall(part)) >= 3 or (not any(c.isspace() for c in part) and sum(c.isalpha() for c in part) >= 4):\n lines.append(part)\n return lines\n\n\nNEGATED = re.compile(\n r\"\\bno\\s+(?:tempo\\s+changes?|key\\s+changes?|slow\\s?downs?|speed\\s?ups?|breakdowns?|drops?|bridges?)\",\n re.IGNORECASE,\n)\nSYLLABLE_TOLERANCE = 1\nMAX_OFF_METER_SHARE = 0.25\nGROSS_SYLLABLE_MISS = 4\n\n\ndef validate_duration(duration_seconds: int) -> dict:\n valid = type(duration_seconds) is int and duration_seconds in ALLOWED_DURATION_SECONDS\n return {\n \"valid\": valid,\n \"duration_seconds\": duration_seconds,\n \"errors\": [] if valid else [\n \"Song mode supports only 60 or 120 seconds. Keep the original requested \"\n \"duration; ask for a supported length when clarification is available, \"\n \"otherwise stop and explain the unsupported duration. Do not generate a shorter song.\"\n ],\n }\n\n\ndef validate(\n prompt: str,\n *,\n duration_seconds: int,\n target_syllables: int = TARGET_SYLLABLES,\n lyrics_language: str = \"en\",\n) -> dict:\n errors = validate_duration(duration_seconds)[\"errors\"]\n lowered = prompt.casefold()\n scannable = NEGATED.sub(\" \", lowered)\n\n declared_minutes = re.search(r\"\\ba full (\\d+)-minute\\b\", lowered)\n if declared_minutes is None or int(declared_minutes.group(1)) * 60 != duration_seconds:\n errors.append(\n \"Start the song prompt with 'A full 1-minute' or 'A full 2-minute' \"\n \"matching the original requested duration; do not replace it with a shorter length\"\n )\n\n if len(prompt) > MAX_PROMPT_CHARS:\n errors.append(\n f\"prompt is {len(prompt)} characters, over the {MAX_PROMPT_CHARS} limit — \"\n \"trim instrument adjectives first, never the meter laws or the lyrics\"\n )\n\n bpm_values = {int(value) for value in re.findall(r\"(\\d{2,3})\\s*bpm\", lowered)}\n if not bpm_values:\n errors.append(\"no BPM stated — name one constant tempo (100 gentle / 112 dance-along)\")\n elif len(bpm_values) > 1:\n errors.append(\n \"more than one BPM in the prompt (\"\n + \", \".join(str(value) for value in sorted(bpm_values))\n + \") — one constant tempo only\"\n )\n elif not bpm_values \u003c= ALLOWED_BPM:\n errors.append(\n f\"BPM {bpm_values.pop()} is untested — use 100 (gentle) or 112 (dance-along)\"\n )\n\n for pattern, message in REQUIRED_CLAUSES:\n if not re.search(pattern, lowered):\n errors.append(f\"missing clause: {message}\")\n\n for phrase in BANNED_TEMPO:\n if phrase in scannable:\n errors.append(\n f'banned tempo word \"{phrase}\" — dynamics come from orchestration '\n \"(fuller final chorus, extra claps), never from tempo\"\n )\n for phrase in BANNED_GROOVE:\n if phrase in scannable:\n errors.append(\n f'banned groove jargon \"{phrase}\" — describe claps and instruments plainly'\n )\n\n lyrics = _lyric_lines(prompt)\n if not lyrics:\n errors.append(\"no quoted lyric lines found — the verses and chorus go in the prompt\")\n elif lyrics_language in {\"en\", \"ru\"}:\n count_line = _line_syllables if lyrics_language == \"en\" else lambda line: len(re.findall(r\"[аеёиоуыэюя]\", line.casefold()))\n off = [\n f'\"{line}\" ({count_line(line)})'\n for line in lyrics\n if abs(count_line(line) - target_syllables) > SYLLABLE_TOLERANCE\n ]\n gross = [\n f'\"{line}\" ({count_line(line)})'\n for line in lyrics\n if abs(count_line(line) - target_syllables) >= GROSS_SYLLABLE_MISS\n ]\n if gross:\n errors.append(\n f\"lyric line(s) nowhere near the {target_syllables}-syllable meter — they \"\n \"cannot land one syllable per beat: \"\n + \"; \".join(gross[:6])\n + (\" …\" if len(gross) > 6 else \"\")\n )\n elif len(off) > max(1, int(len(lyrics) * MAX_OFF_METER_SHARE)):\n errors.append(\n f\"{len(off)} of {len(lyrics)} lyric lines miss the {target_syllables}-syllable \"\n \"meter by two or more — the wandering meter is what makes the beat sound \"\n \"broken; rewrite them: \"\n + \"; \".join(off[:6])\n + (\" …\" if len(off) > 6 else \"\")\n )\n\n return {\n \"valid\": not errors,\n \"duration_seconds\": duration_seconds,\n \"characters\": len(prompt),\n \"bpm\": sorted(bpm_values),\n \"lyric_lines\": len(lyrics),\n \"manual_meter_review_required\": lyrics_language not in {\"en\", \"ru\"},\n \"errors\": errors,\n }\n\n\ndef main() -> int:\n parser = argparse.ArgumentParser(description=__doc__)\n parser.add_argument(\"--duration-seconds\", type=int, required=True,\n help=\"original requested duration, exactly 60 or 120 seconds\")\n mode = parser.add_mutually_exclusive_group(required=True)\n mode.add_argument(\"--prompt\", help=\"file holding the song prompt\")\n mode.add_argument(\"--duration-only\", action=\"store_true\",\n help=\"check the request before writing lyrics or generating assets\")\n parser.add_argument(\"--syllables\", type=int, default=TARGET_SYLLABLES)\n parser.add_argument(\"--lyrics-language\", default=\"en\", help=\"ISO language code; other languages require manual meter review\")\n parser.add_argument(\"--json\", action=\"store_true\")\n args = parser.parse_args()\n\n result = validate_duration(args.duration_seconds)\n if result[\"valid\"] and not args.duration_only:\n with open(args.prompt, encoding=\"utf-8\") as handle:\n prompt = handle.read()\n result = validate(prompt, duration_seconds=args.duration_seconds,\n target_syllables=args.syllables, lyrics_language=args.lyrics_language.casefold())\n if args.json:\n print(json.dumps(result, indent=2, ensure_ascii=False))\n else:\n print(f\"valid: {result['valid']} duration seconds: {result['duration_seconds']}\")\n if \"characters\" in result:\n print(f\"chars: {result['characters']} bpm: {result['bpm']} \"\n f\"lyric lines: {result['lyric_lines']}\")\n if result.get(\"manual_meter_review_required\"):\n print(\"Manual lyric meter review required for this language.\")\n for error in result[\"errors\"]:\n print(f\" - {error}\")\n return 0 if result[\"valid\"] else 1\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":8792,"content_sha256":"298c20de7fc967e56ecf9528b1f23cb1cf9a73107bf39d9badf925eebe7be3f5"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"Faceless video","type":"text"}]},{"type":"paragraph","content":[{"text":"Create one finished video with a consistent visual style and narration. For an illustrated explainer use deliberate, diagram-led scenes; for a story or documentary adapt the pacing and visuals to the brief. Do not treat an explainer as a request for rapid cuts. Preserve choices and assets supplied by the user.","type":"text"}]},{"type":"paragraph","content":[{"text":"Choose the requested mode before generating: illustrated explainers use ","type":"text"},{"text":"references/production.md","type":"text","marks":[{"type":"code_inline"}]},{"text":"; fast-cut channel motion uses ","type":"text"},{"text":"references/channel-motion.md","type":"text","marks":[{"type":"code_inline"}]},{"text":"; stills and picture stories use ","type":"text"},{"text":"references/picture-story.md","type":"text","marks":[{"type":"code_inline"}]},{"text":"; a sung children's music video uses ","type":"text"},{"text":"references/song.md","type":"text","marks":[{"type":"code_inline"}]},{"text":"; explicitly requested talking characters use ","type":"text"},{"text":"references/talking-characters.md","type":"text","marks":[{"type":"code_inline"}]},{"text":"; a requested presenter overlay uses ","type":"text"},{"text":"references/presenter.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" alongside the motion workflow; edits to a previously generated video use ","type":"text"},{"text":"references/editing.md","type":"text","marks":[{"type":"code_inline"}]},{"text":". A request for background music alone is not a song request. Never infer a presenter from an incidental photo.","type":"text"}]},{"type":"paragraph","content":[{"text":"When the brief supplies ","type":"text"},{"text":"animation_mode","type":"text","marks":[{"type":"code_inline"}]},{"text":", treat ","type":"text"},{"text":"fully_animated","type":"text","marks":[{"type":"code_inline"}]},{"text":" as motion and ","type":"text"},{"text":"scene_based","type":"text","marks":[{"type":"code_inline"}]},{"text":" as stills without asking again. Resolve song intent from the complete brief, including topic and script; an explicitly sung children's video uses song mode even when a voice is supplied. Preserve the production mode when continuing or editing an existing video.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Before spending credits","type":"text"}]},{"type":"paragraph","content":[{"text":"Resolve missing topic, approximate duration, aspect ratio, language, visual style, voice, music and subtitle choices. Never invent a preset or voice ID. Use ","type":"text"},{"text":"higgsfield preset list video-explainer --json","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"higgsfield voices list --json","type":"text","marks":[{"type":"code_inline"}]},{"text":" for current choices, and resolve a selected preset with ","type":"text"},{"text":"higgsfield preset resolve video-explainer \u003cid> --json","type":"text","marks":[{"type":"code_inline"}]},{"text":". Confirm photo roles before using reference images. Honor the requested duration; never silently round it up.","type":"text"}]},{"type":"paragraph","content":[{"text":"Find this skill's installed directory and retain its absolute path as ","type":"text"},{"text":"SKILL_DIR","type":"text","marks":[{"type":"code_inline"}]},{"text":". Keep all work in a persistent session directory, not the skill directory. Before generation verify ","type":"text"},{"text":"higgsfield account status","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"ffmpeg","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"ffprobe","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"python3","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"curl","type":"text","marks":[{"type":"code_inline"}]},{"text":", and ","type":"text"},{"text":"awk","type":"text","marks":[{"type":"code_inline"}]},{"text":", and inspect the live schemas using ","type":"text"},{"text":"higgsfield model get","type":"text","marks":[{"type":"code_inline"}]},{"text":" for each planned model. Estimate generation costs with ","type":"text"},{"text":"higgsfield generate cost","type":"text","marks":[{"type":"code_inline"}]},{"text":" where supported; if an estimate is unavailable, disclose that before generation. If any required stage, model, dependency, or requested feature is unavailable, explain and stop before paid work. Do not call ","type":"text"},{"text":"explainer_video","type":"text","marks":[{"type":"code_inline"}]},{"text":" or ","type":"text"},{"text":"video_explainer","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]},{"type":"paragraph","content":[{"text":"Read ","type":"text"},{"text":"references/production.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" for the generation, assembly, QC, recovery and delivery procedure. For subtitles, read ","type":"text"},{"text":"references/captions.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" and check its dependencies before paid generation. Read ","type":"text"},{"text":"references/creative.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" for style, assets, long-form planning and review, ","type":"text"},{"text":"references/styles.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" for style guidance, and ","type":"text"},{"text":"references/prompts.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" when writing the scene plan. The executable helper ","type":"text"},{"text":"scripts/assemble_blocks.sh","type":"text","marks":[{"type":"code_inline"}]},{"text":" assembles ordered video/audio pairs without changing speech speed; read its usage before running it. The previous skill name is supported for existing installations.","type":"text"}]}]},"metadata":{"date":"2026-11-02","name":"higgsfield-faceless-video","author":"@skillopedia","source":{"stars":1854,"repo_name":"skills","origin_url":"https://github.com/higgsfield-ai/skills/blob/HEAD/higgsfield-faceless-video/SKILL.md","repo_owner":"higgsfield-ai","body_sha256":"1ec7a771e5bf022134df37e1a9c1ac48e9cfb11d11d1a2b1db354b4d7971e715","cluster_key":"6393b09f5f072434c77ec0088142790967e4cc319b08c3116374df930dcf3335","clean_bundle":{"format":"clean-skill-bundle-v1","source":"higgsfield-ai/skills/higgsfield-faceless-video/SKILL.md","text_bytes":232911,"source_commit":"4d0757471a704dac8729ea3c2ef0f3f330e0d9b6","skills_sh_rank":3555,"attachment_count":32,"text_attachments":32,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"higgsfield-faceless-video/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"0.14.0","category":"media-content","category_label":"Media"},"exact_dupes_collapsed_into_this":0},"version":"0.14.0","category":"media-content","import_tag":"skills-sh-top20k-2026-11","description":"Create a finished faceless explainer, documentary, illustrated story or children's music video, or edit a previously created faceless video. Use when: \"make a faceless video\", \"create a narrated explainer\", \"make a children's song video\", \"edit this faceless video\". NOT for: a single silent clip, ads or UGC, unrelated footage edits, or script-only requests.\n","allowed-tools":"Bash","argument-hint":"[topic or source] [duration] [language] [aspect ratio]"}},"renderedAt":1793671930351}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.