Unified Voice & Images Pipeline

Diagnosis first:** if your current ffmpeg command is pointed at the built-in microphone, that's almost certainly your noise/buzz source — a mic can only capture whatever leaks into the room plus its own hiss, and it's structurally impossible for that to sound clean. The fix isn't "denoise better," it's "stop touching the mic."

The trusted free tool: BlackHole. It's an open-source, MIT-licensed virtual audio driver — it creates a device that other apps' output (including your browser) can be routed into digitally, with zero air/mic involved:

brew install blackhole-2ch

Setup: 1. Open Audio MIDI Setup → "+" → Create Multi-Output Device, and check both your headphones and BlackHole 2ch. You'll still hear everything through your headphones while it's simultaneously sent to BlackHole for capture — this directly solves the "no speakers, headphones-only" problem. One known quirk: when using a multi-output device, the built-in/headphone output needs to be enabled and listed as the top device in the list, or macOS can misbehave. 2. Match sample rates. This is very likely your buzzing culprit. Apple's own guidance on combined devices is explicit: all devices in the aggregate/multi-output device need to have the same sample rate, matching whichever device is set as the clock source, and if the devices aren't hardware-synced, you need to enable Drift Correction for every device except the clock source to compensate for clock drift between them. Skipping this step is the #1 cause of the clicking/buzzing you're describing in exactly this kind of setup. 3. Point ffmpeg at BlackHole instead of the mic: bash ffmpeg -f avfoundation -list_devices true -i "" # find BlackHole's index ffmpeg -f avfoundation -i ":N" -ac 2 -ar 48000 recording.wav 4. Stop it with a graceful signal (Ctrl-C / SIGINT, or send q), not kill -9 — ffmpeg needs to close the file header properly, and an abrupt kill can leave a corrupted/glitchy tail that sounds exactly like buzzing.

Once you're capturing from BlackHole, hiss and buzz from the room are simply not possible anymore — it's a digital signal, not sound traveling through air into a mic.

**

Even better than any live recording, when it's an option: grab the actual file instead of re-recording it.** Live loopback capture means the agent has to sit and "listen" in real time and depends on getting the audio setup exactly right. Two cleaner alternatives:

• yt-dlp (free, open source, very actively maintained, supports 1000+ sites): if the website's audio comes from a supported platform, skip capture entirely — bash yt-dlp -x --audio-format mp3 --audio-quality 0 "URL" This is a byte-for-byte copy of the source stream, so buzzing/hiss simply can't occur, and there's no real-time waiting.
• Browser automation with network interception (Playwright, driven headlessly by the agent): for a custom site not covered by yt-dlp, the agent opens the page and grabs the audio file straight out of the network response instead of "listening" to playback. Same benefit — it's the original bytes.

Reserve BlackHole + ffmpeg for cases with no discrete file to grab — DRM streams, live calls, WebRTC.

**

How the agent can verify quality without ears:**

ffmpeg -i recording.wav -af astats -f null -

This flags near-total silence (usually means the wrong input device got selected) or clipping at 0 dBFS. If there's still faint hum after the sample-rate fix, afftdn (denoise) or highpass=100 (cuts mains hum) can clean it up — but treat that as a last resort, since denoising can shave consonants and actually hurt transcription accuracy more than it helps.

Best default: whisper.cpp, invoked via its whisper-cli binary. It's a pure C/C++ port of Whisper — no PyTorch, no heavy Python ML stack — so an agent can shell out to it without spinning up a big environment. On Apple Silicon it can run the encoder on the Neural Engine via Core ML, which is more than 3x faster than CPU-only execution, and it supports quantized models (q5_0, q8_0) that cut RAM/disk footprint substantially. It also has a built-in Voice Activity Detection mode — passing --vad skips silent stretches before they hit the model, which speeds up long recordings significantly.

Typical agent recipe:

# whisper wants 16kHz mono wav — convert first
ffmpeg -i input.mp3 -ar 16000 -ac 1 -c:a pcm_s16le input.wav

# transcribe (--vad skips silence, speeds up long recordings)
whisper-cli -m models/ggml-base.en.bin -f input.wav --vad --output-txt --output-srt

If you're on Apple Silicon and don't mind a thin Python layer: mlx-whisper. It runs on the GPU through Apple's MLX/Metal stack, pip install mlx-whisper, and is a two-line call:

import mlx_whisper
text = mlx_whisper.transcribe("audio.mp3", path_or_hf_repo="mlx-community/whisper-small")["text"]

Also very light on CPU since the actual compute happens on the GPU.

If you want zero local compute at all: a free-tier hosted Whisper endpoint (e.g. Groq's) does the heavy lifting on their servers — the agent just makes a lightweight HTTP call. Trade-off: needs network access and you're bound by whatever the free-tier limits are, so it's "free" with an asterisk rather than fully offline-free.

For fully automated pipelines, the pattern is simple: agent watches a folder → converts new audio to 16kHz mono wav → runs whisper-cli → saves .txt/.srt next to the source file. I can write that script for you if useful.

Tool Selection

Recommended Default: whisper.cpp

Pure C/C++ port of Whisper — no PyTorch, no heavy Python ML stack. Agent shells out to it directly.

# One-time setup
brew install whisper-cpp
# Or build from source:
# git clone https://github.com/ggml-org/whisper.cpp && cd whisper.cpp
# cmake -B build && cmake --build build -j --config Release
# ./models/download-ggml-model.sh base.en

# Convert to whisper's required format (16kHz mono WAV)
ffmpeg -i input.mp3 -ar 16000 -ac 1 -c:a pcm_s16le input.wav

# Transcribe — English
whisper-cli -m models/ggml-base.en.bin -f input.wav --vad --output-txt --output-srt

# Transcribe — auto-detect language
whisper-cli -m models/ggml-base.bin -f input.wav --vad --language auto --output-txt --output-srt

# Transcribe — word-level timestamps (for precise image matching)
whisper-cli -m models/ggml-base.en.bin -f input.wav --vad --output-words --output-json

Key flags: - --vad — Voice Activity Detection. Skips silence before it hits the model. Dramatically faster on recordings with dead air. - --language auto — auto-detect language (use ggml-base.bin not ggml-base.en.bin for multilingual). - --output-words — word-level timestamps, essential for precise image-to-transcript alignment. - --output-json — structured output the orchestrator can parse directly.

Alternative: mlx-whisper (Apple Silicon GPU)

import mlx_whisper

# Simple transcription
result = mlx_whisper.transcribe(
    "audio.mp3",
    path_or_hf_repo="mlx-community/whisper-small"
)
text = result["text"]
segments = result["segments"]  # [{"start": 0.0, "end": 4.2, "text": "..."}, ...]

Very light on CPU since actual compute happens on Metal GPU. Good for Python-native pipelines.

Agent Auto-Transcription Recipe

#!/usr/bin/env bash
# transcribe_audio.sh — agent drops audio in, gets .txt + .srt + .json out
set -euo pipefail

INPUT="${1:?Usage: $0 <audio_file> [language]}"
LANG="${2:-en}"
MODEL_DIR="${WHISPER_MODEL_DIR:-$HOME/whisper.cpp/models}"

# Pick model: .en for English-only (faster), base for multilingual
if [ "$LANG" = "en" ]; then
    MODEL="$MODEL_DIR/ggml-base.en.bin"
else
    MODEL="$MODEL_DIR/ggml-base.bin"
    LANG="auto"  # let whisper detect
fi

[ -f "$MODEL" ] || { echo "Model not found: $MODEL"; exit 1; }

# Convert to 16kHz mono wav
BASE="$(basename "$INPUT" | sed 's/\.[^.]*$//')"
OUT_DIR="$(dirname "$INPUT")"
WAV="$OUT_DIR/${BASE}_16k.wav"

ffmpeg -y -i "$INPUT" -ar 16000 -ac 1 -c:a pcm_s16le "$WAV" -loglevel error

# Transcribe
LANG_ARGS=""
[ "$LANG" != "en" ] && LANG_ARGS="--language $LANG"

whisper-cli -m "$MODEL" -f "$WAV" --vad \
    --output-txt --output-srt --output-json \
    $LANG_ARGS

# Move outputs next to input
for ext in txt srt json; do
    [ -f "${WAV}.${ext}" ] && mv -f "${WAV}.${ext}" "$OUT_DIR/${BASE}.${ext}"
done

# Verify
if [ -f "$OUT_DIR/${BASE}.txt" ] && [ -s "$OUT_DIR/${BASE}.txt" ]; then
    WORDS=$(wc -w < "$OUT_DIR/${BASE}.txt")
    echo "Transcribed: $OUT_DIR/${BASE}.txt ($WORDS words)"
    [ -f "$OUT_DIR/${BASE}.srt" ] && echo "Subtitles:   $OUT_DIR/${BASE}.srt"
    [ -f "$OUT_DIR/${BASE}.json" ] && echo "JSON:        $OUT_DIR/${BASE}.json"
else
    echo "WARNING: Transcription produced empty output. Check audio quality."
    # Quality diagnostic
    ffmpeg -i "$INPUT" -af astats -f null - 2>&1 | grep "RMS level dB" | tail -1
    exit 1
fi

# Cleanup intermediate wav
rm -f "$WAV"

Transcript Verification Protocol

After generating a transcript, the agent MUST verify it before using it in the pipeline:

VERIFICATION STEPS (mandatory):

1. FILE EXISTS AND NON-EMPTY
   [ -s "output.txt" ] || FAIL "Empty transcript"

2. WORD COUNT SANITY CHECK
   Expected: ~2.5 words/second for English, ~2.0 words/second for Arabic.
   Duration (seconds) * rate = expected word count.
   If actual < 50% of expected → likely wrong input device or silence.
   If actual > 200% of expected → likely hallucination (whisper repeats on silence).

3. HALLUCINATION DETECTION
   Whisper is known to hallucinate on silent/near-silent input.
   Red flags:
   - Same phrase repeated 3+ times in a row
   - "Thank you for watching" or "Subscribe" appearing in non-YouTube audio
   - Transcript much longer than audio duration would suggest
   Agent check:
   grep -c 'Thank you for watching\|Please subscribe\|Thanks for watching' output.txt
   # If count > 0 and audio is not from YouTube → HALLUCINATION, re-run with --vad

4. LANGUAGE MATCH
   If expected language is known, verify transcript contains that language's characters.
   For Arabic: grep -c '[\u0600-\u06FF]' output.txt should be > 0.
   For English: mostly ASCII.

5. TIMESTAMP CONTINUITY (SRT/JSON)
   Segments should be sequential, non-overlapping, and cover most of the audio duration.
   Gaps > 5s between segments → potential missed content (unless silence/music).

6. SPOT-CHECK (if agent has audio playback or VLM)
   Pick 3 random segments. Compare transcript text against actual audio.
   If 2+ segments are wrong → re-transcribe with larger model (small → medium → large).

Model Size Selection Guide

Use .en variants when you know the audio is English-only — they are faster and more accurate for English.

#!/usr/bin/env bash
#
# record_and_transcribe.sh
# Records website/system audio via BlackHole loopback, then transcribes it
# locally and for free with whisper.cpp. No microphone involved anywhere,
# so hiss/buzz from the room is structurally impossible.
#
# Can also download audio directly via yt-dlp (--download URL) to skip
# live capture entirely when the source is a supported site.
#
# ---------------------------------------------------------------------------
# ONE-TIME SETUP (do this before running the script):
#
#   1. Install the tools:
#        brew install ffmpeg blackhole-2ch
#
#   2. Build whisper.cpp and grab a model:
#        git clone https://github.com/ggml-org/whisper.cpp
#        cd whisper.cpp
#        cmake -B build && cmake --build build -j --config Release
#        ./models/download-ggml-model.sh base.en
#      whisper-cli then lives at: whisper.cpp/build/bin/whisper-cli
#
#   3. In Audio MIDI Setup (Applications > Utilities):
#        - "+" -> Create Multi-Output Device
#        - Check both your headphones AND "BlackHole 2ch"
#        - Keep your headphones/built-in output listed at the TOP
#        - Give every device in it the SAME sample rate (e.g. 48000 Hz)
#        - Tick "Drift Correction" for every device except the clock source
#      This is what fixes the buzzing/clicking.
#
#   4. Optional but recommended, lets this script auto-switch your output:
#        brew install switchaudio-osx
#
#   5. Optional, for --download mode:
#        brew install yt-dlp
#
# ---------------------------------------------------------------------------
# USAGE:
#   ./record_and_transcribe.sh [name] [duration_seconds]
#   ./record_and_transcribe.sh --download URL [name]
#
#   name               Label for the output files (default: timestamp)
#   duration_seconds   Auto-stop after N seconds. Omit to record until
#                       you press Ctrl+C.
#   --download URL     Download audio from URL via yt-dlp instead of
#                       recording from BlackHole. Skips live capture.
#
# CONFIG (optional environment variables, or just edit the defaults below):
#   WHISPER_BIN, WHISPER_MODEL, WHISPER_LANG, OUTPUT_DIR, MULTI_OUTPUT_NAME
# ---------------------------------------------------------------------------

set -euo pipefail

WHISPER_BIN="${WHISPER_BIN:-whisper-cli}"
WHISPER_MODEL="${WHISPER_MODEL:-$HOME/whisper.cpp/models/ggml-base.en.bin}"
WHISPER_LANG="${WHISPER_LANG:-en}"  # ponytail: "auto" for auto-detect, "en" for English-only
OUTPUT_DIR="${OUTPUT_DIR:-$HOME/Recordings}"
MULTI_OUTPUT_NAME="${MULTI_OUTPUT_NAME:-Multi-Output Device}"
SAMPLE_RATE=48000
CHANNELS=2

#!/usr/bin/env bash
#
# record_and_transcribe.sh
# Records website/system audio via BlackHole loopback, then transcribes it
# locally and for free with whisper.cpp. No microphone involved anywhere,
# so hiss/buzz from the room is structurally impossible.
#
# Can also download audio directly via yt-dlp (--download URL) to skip
# live capture entirely when the source is a supported site.
#
# ---------------------------------------------------------------------------
# ONE-TIME SETUP (do this before running the script):
#
#   1. Install the tools:
#        brew install ffmpeg blackhole-2ch
#
#   2. Build whisper.cpp and grab a model:
#        git clone https://github.com/ggml-org/whisper.cpp
#        cd whisper.cpp
#        cmake -B build && cmake --build build -j --config Release
#        ./models/download-ggml-model.sh base.en
#      whisper-cli then lives at: whisper.cpp/build/bin/whisper-cli
#
#   3. In Audio MIDI Setup (Applications > Utilities):
#        - "+" -> Create Multi-Output Device
#        - Check both your headphones AND "BlackHole 2ch"
#        - Keep your headphones/built-in output listed at the TOP
#        - Give every device in it the SAME sample rate (e.g. 48000 Hz)
#        - Tick "Drift Correction" for every device except the clock source
#      This is what fixes the buzzing/clicking.
#
#   4. Optional but recommended, lets this script auto-switch your output:
#        brew install switchaudio-osx
#
#   5. Optional, for --download mode:
#        brew install yt-dlp
#
# ---------------------------------------------------------------------------
# USAGE:
#   ./record_and_transcribe.sh [name] [duration_seconds]
#   ./record_and_transcribe.sh --download URL [name]
#
#   name               Label for the output files (default: timestamp)
#   duration_seconds   Auto-stop after N seconds. Omit to record until
#                       you press Ctrl+C.
#   --download URL     Download audio from URL via yt-dlp instead of
#                       recording from BlackHole. Skips live capture.
#
# CONFIG (optional environment variables, or just edit the defaults below):
#   WHISPER_BIN, WHISPER_MODEL, WHISPER_LANG, OUTPUT_DIR, MULTI_OUTPUT_NAME
# ---------------------------------------------------------------------------

set -euo pipefail

WHISPER_BIN="${WHISPER_BIN:-whisper-cli}"
WHISPER_MODEL="${WHISPER_MODEL:-$HOME/whisper.cpp/models/ggml-base.en.bin}"
WHISPER_LANG="${WHISPER_LANG:-en}"  # ponytail: "auto" for auto-detect, "en" for English-only
OUTPUT_DIR="${OUTPUT_DIR:-$HOME/Recordings}"
MULTI_OUTPUT_NAME="${MULTI_OUTPUT_NAME:-Multi-Output Device}"
SAMPLE_RATE=48000
CHANNELS=2

# ---------- Parse arguments ----------
DOWNLOAD_URL=""
if [ "${1:-}" = "--download" ]; then
  DOWNLOAD_URL="${2:?Usage: $0 --download URL [name]}"
  NAME="${3:-download_$(date +%Y%m%d_%H%M%S)}"
  DURATION=""
else
  NAME="${1:-recording_$(date +%Y%m%d_%H%M%S)}"
  DURATION="${2:-}"
fi

# ---------- Preflight checks ----------
command -v ffmpeg >/dev/null 2>&1 || { echo "ffmpeg not found. Install with: brew install ffmpeg"; exit 1; }
command -v "$WHISPER_BIN" >/dev/null 2>&1 || { echo "'$WHISPER_BIN' not found on PATH. Set WHISPER_BIN to its full path, or build whisper.cpp first (see script header)."; exit 1; }
[ -f "$WHISPER_MODEL" ] || { echo "Whisper model not found at: $WHISPER_MODEL. Set WHISPER_MODEL, or run whisper.cpp's models/download-ggml-model.sh"; exit 1; }

mkdir -p "$OUTPUT_DIR"

RAW_WAV="$OUTPUT_DIR/${NAME}_raw.wav"
WHISPER_WAV="$OUTPUT_DIR/${NAME}_16k.wav"

# ---------- Cleanup trap — restores audio output even on unexpected exit ----------
ORIGINAL_OUTPUT=""
cleanup() {
  if command -v SwitchAudioSource >/dev/null 2>&1 && [ -n "$ORIGINAL_OUTPUT" ]; then
    SwitchAudioSource -s "$ORIGINAL_OUTPUT" >/dev/null 2>&1 || true
    echo "Restored system output to '$ORIGINAL_OUTPUT'."
  fi
}
trap cleanup EXIT

# ---------- Download mode (yt-dlp) ----------
if [ -n "$DOWNLOAD_URL" ]; then
  command -v yt-dlp >/dev/null 2>&1 || { echo "yt-dlp not found. Install with: brew install yt-dlp"; exit 1; }
  echo "Downloading audio from: $DOWNLOAD_URL"
  yt-dlp -x --audio-format wav --audio-quality 0 -o "$RAW_WAV" "$DOWNLOAD_URL"

  if [ ! -s "$RAW_WAV" ]; then
    echo "Download failed or produced empty file."
    exit 1
  fi
  echo ""
  echo "Downloaded: $RAW_WAV"
else
  # ---------- Find BlackHole's avfoundation index ----------
  DEVICE_LIST="$(ffmpeg -f avfoundation -list_devices true -i "" 2>&1)"

  # ponytail: only search the audio-devices section to avoid grabbing a video index.
  # ceiling: if ffmpeg changes its section header text, the sed range may need updating.
  AUDIO_SECTION="$(echo "$DEVICE_LIST" | sed -n '/AVFoundation audio devices/,/^$/p')"
  BLACKHOLE_INDEX="$(echo "$AUDIO_SECTION" | grep -i "BlackHole" | head -1 | sed -E 's/.*\[([0-9]+)\].*/\1/')"

  if [ -z "$BLACKHOLE_INDEX" ] || ! [[ "$BLACKHOLE_INDEX" =~ ^[0-9]+$ ]]; then
    echo "Could not find a BlackHole audio device. Is it installed? brew install blackhole-2ch"
    echo "Full audio device list:"
    echo "$AUDIO_SECTION"
    exit 1
  fi
  echo "Found BlackHole at avfoundation audio index $BLACKHOLE_INDEX"

  # ---------- Try to auto-switch system output to the Multi-Output Device ----------
  if command -v SwitchAudioSource >/dev/null 2>&1; then
    ORIGINAL_OUTPUT="$(SwitchAudioSource -c)"
    if SwitchAudioSource -s "$MULTI_OUTPUT_NAME" >/dev/null 2>&1; then
      echo "Switched system output to '$MULTI_OUTPUT_NAME'."
    else
      echo "Could not switch to '$MULTI_OUTPUT_NAME' automatically — set it manually in System Settings > Sound before continuing."
    fi
  else
    echo "Tip: 'brew install switchaudio-osx' lets this script switch your output device automatically."
    echo "For now, make sure System Settings > Sound > Output is set to your Multi-Output Device."
  fi

  echo ""
  if [ -n "$DURATION" ]; then
    echo "Recording for ${DURATION}s from BlackHole. Start playing the website audio now."
  else
    echo "Recording from BlackHole. Start playing the website audio now, then press Ctrl+C when done."
  fi
  echo ""

  # ponytail: ${DURATION:+-t "$DURATION"} is bash 3.2-safe. The old DURATION_ARGS=()
  # pattern crashes on macOS stock bash under set -u (unbound variable on empty array).
  set +e  # ffmpeg returns non-zero on Ctrl-C; that's expected
  ffmpeg -f avfoundation -i ":$BLACKHOLE_INDEX" \
    ${DURATION:+-t "$DURATION"} \
    -ac "$CHANNELS" -ar "$SAMPLE_RATE" -y "$RAW_WAV" \
    -loglevel warning
  set -e

  if [ ! -s "$RAW_WAV" ]; then
    echo "No audio was captured. Check that BlackHole/Multi-Output routing is set up correctly."
    exit 1
  fi

  echo ""
  echo "Recording saved: $RAW_WAV"
fi

# ---------- Quick quality check (no ears required) ----------
echo "Checking audio level..."
STATS="$(ffmpeg -i "$RAW_WAV" -af astats -f null - 2>&1)"
RMS="$(echo "$STATS" | grep "RMS level dB" | tail -1 | grep -oE '\-?[0-9]+\.[0-9]+' | head -1)"
if [ -n "$RMS" ] && [ "$RMS" != "-inf" ]; then
  echo "Overall RMS level: ${RMS} dB"
  # ponytail: awk numeric comparison; the -inf guard above prevents awk coercing it to 0
  awk -v rms="$RMS" 'BEGIN { if (rms+0 < -50) print "WARNING: this looks close to silent — double check routing before trusting the transcript." }'
else
  if [ "$RMS" = "-inf" ]; then
    echo "WARNING: RMS is -inf (total silence). Wrong input device or nothing was playing."
  else
    echo "Could not read level stats (file may be very short)."
  fi
fi

# ---------- Convert to 16kHz mono for whisper.cpp ----------
echo "Converting to 16kHz mono..."
ffmpeg -y -i "$RAW_WAV" -ar 16000 -ac 1 -c:a pcm_s16le "$WHISPER_WAV" -loglevel error

if [ ! -s "$WHISPER_WAV" ]; then
  echo "Conversion to 16kHz mono failed."
  exit 1
fi

# ---------- Transcribe ----------
# --vad: skip silence before it hits the model (faster on recordings with dead air)
LANG_ARGS=""
if [ "$WHISPER_LANG" != "en" ]; then
  LANG_ARGS="--language $WHISPER_LANG"
fi

echo "Transcribing with whisper.cpp (lang=$WHISPER_LANG)..."
"$WHISPER_BIN" -m "$WHISPER_MODEL" -f "$WHISPER_WAV" \
  --vad --output-txt --output-srt $LANG_ARGS

# whisper.cpp writes "<input>.txt" / "<input>.srt" next to the input file by default
mv -f "${WHISPER_WAV}.txt" "$OUTPUT_DIR/${NAME}.txt" 2>/dev/null || true
mv -f "${WHISPER_WAV}.srt" "$OUTPUT_DIR/${NAME}.srt" 2>/dev/null || true

echo ""
# ---------- Verify outputs exist ----------
if [ -f "$OUTPUT_DIR/${NAME}.txt" ] || [ -f "$OUTPUT_DIR/${NAME}.srt" ]; then
  echo "Done."
  echo "  Raw audio:  $RAW_WAV"
  [ -f "$OUTPUT_DIR/${NAME}.txt" ] && echo "  Transcript: $OUTPUT_DIR/${NAME}.txt"
  [ -f "$OUTPUT_DIR/${NAME}.srt" ] && echo "  Subtitles:  $OUTPUT_DIR/${NAME}.srt"
else
  echo "WARNING: whisper ran but no .txt or .srt output was found."
  echo "  Check whisper-cli output above for errors."
  echo "  Raw audio is still at: $RAW_WAV"
  exit 1
fi

One script, two modes: record (BlackHole loopback capture) or download (yt-dlp, no live capture needed). It finds BlackHole automatically, records, does a quick silence/level check, converts to 16kHz mono, and transcribes with whisper-cli — all in one call.

Before first run:

brew install ffmpeg blackhole-2ch switchaudio-osx yt-dlp

plus building whisper.cpp and grabbing a model (full steps are in the script's header comments). Set up the Multi-Output Device once in Audio MIDI Setup as described earlier, matching sample rates.

Then it's just:

./record_and_transcribe.sh                          # records until Ctrl+C
./record_and_transcribe.sh interview 120            # named, auto-stops at 120s
./record_and_transcribe.sh --download "URL"         # download instead of recording
./record_and_transcribe.sh --download "URL" podcast # download with custom name
WHISPER_LANG=auto ./record_and_transcribe.sh        # auto-detect language

If switchaudio-osx is installed, it'll flip your system output to the Multi-Output Device for you and switch it back when done (even on unexpected exit, via a cleanup trap) — so the agent can run this fully hands-off without you touching Sound settings each time.

Before first run:

brew install ffmpeg blackhole-2ch switchaudio-osx yt-dlp

Then:

./record_and_transcribe.sh                          # records until Ctrl+C
./record_and_transcribe.sh interview 120            # named, auto-stops at 120s
./record_and_transcribe.sh --download "URL"         # download instead of recording
./record_and_transcribe.sh --download "URL" podcast # download with custom name
WHISPER_LANG=auto ./record_and_transcribe.sh        # auto-detect language

Tool Selection

PATH A: Pure Language — edge-tts (Recommended Default)

Microsoft Neural TTS via free public endpoint. No API key. No account. No credit card.

# Install
pip install edge-tts

# CLI usage
edge-tts --voice en-US-JennyNeural --text "Your script text here" --write-media /tmp/out.mp3
edge-tts --voice ar-EG-SalmaNeural --text "النص العربي هنا" --write-media /tmp/out_ar.mp3

# List all available voices
edge-tts --list-voices

Recommended voices: - English: en-US-JennyNeural, en-US-GuyNeural, en-GB-SoniaNeural - Arabic: ar-EG-SalmaNeural, ar-SA-HamedNeural, ar-AE-HamdanNeural, ar-JO-SanaNeural

Python integration:

import edge_tts
import asyncio
import re
import hashlib


async def speak_pure(text: str, voice: str = "en-US-JennyNeural") -> str | None:
    """Generate speech from text using edge-tts. Returns output file path."""
    if not text or not text.strip():
        return None
    text = text.strip()
    output_path = "/tmp/edge_" + hashlib.md5((text + voice).encode()).hexdigest()[:8] + ".mp3"
    for attempt in range(2):
        try:
            comm = edge_tts.Communicate(text, voice)
            await asyncio.wait_for(comm.save(output_path), timeout=15.0)
            return output_path
        except Exception:
            if attempt == 0:
                await asyncio.sleep(2)
    return None  # both attempts failed


def auto_detect_voice(text: str) -> str:
    """Pick voice based on Unicode content."""
    if re.search(r'[\u0600-\u06FF]', text):
        return "ar-EG-SalmaNeural"
    return "en-US-JennyNeural"


async def speak_auto(text: str) -> str | None:
    """Auto-detect language and generate speech."""
    return await speak_pure(text, auto_detect_voice(text))

PATH B: Mixed Language — FreeTTS SSML (Arabic + English in one audio)

FreeTTS is the only free, no-signup, no-API-key service that supports full W3C SSML including <lang> switching. Backend is Microsoft Azure Cognitive Services.

SSML template for mixed Arabic-English:

<speak version="1.0" xml:lang="ar-SA">
  <voice name="ar-SA-ZariyahNeural">
    النص العربي هنا
    <lang xml:lang="en-US">English term</lang>
    يستمر النص العربي
    <break time="300ms"/>
    <lang xml:lang="en-US">ICU</lang>
    هنا
  </voice>
</speak>

Python call:

import requests


def speak_mixed_ssml(ssml_text: str) -> dict | None:
    """Generate mixed-language speech via FreeTTS SSML API."""
    url = "https://freetts.org/api/tts"
    payload = {
        "text": ssml_text,
        "voice": "ar-SA-ZariyahNeural",
        "format": "mp3"
    }
    for attempt in range(2):
        try:
            r = requests.post(url, json=payload, timeout=20)
            r.raise_for_status()
            return r.json()
        except Exception:
            if attempt == 0:
                import time
                time.sleep(2)
    return None

Constraints: - 1,000 characters max per request. For longer text, use Split-and-Stitch (below). - Free tier has a short audio watermark at end. PRO ($19/mo) removes it. - No CORS headers — if calling from browser, proxy through your backend.

PATH C: Split-and-Stitch (Long Scripts / Heavy Mixing)

Use when SSML fails, scripts exceed TTS length limits, or heavy code-switching makes single-file generation unreliable.

Workflow: 1. Pre-process script: - Highlight all English terms in Arabic text. - Extract them as pure-English segments. - Add light tashkeel to remaining Arabic text. - Convert numbers to the voice language that will read them (or spell out). 2. Generate audio per segment: - Arabic segments: ar-EG-SalmaNeural (edge-tts) or ar-SA-ZariyahNeural (FreeTTS) - English segments: en-US-JennyNeural (edge-tts) 3. Name convention: [ORDER]_[LANG]_[TOPIC].mp3 - Example: 01_AR_intro.mp3, 02_EN_brandname.mp3, 03_AR_explanation.mp3 4. Stitch with ffmpeg: bash # Create concat list for f in /tmp/segments/*.mp3; do echo "file '$f'"; done > /tmp/concat.txt # Concatenate with normalized loudness ffmpeg -f concat -safe 0 -i /tmp/concat.txt \ -af "loudnorm=I=-16:LRA=11:TP=-1.5" \ -ar 48000 -c:a libmp3lame -q:a 2 /tmp/final_voice.mp3 5. Silence gaps: 0.3s for mid-sentence switches, 0.5-0.8s for paragraph breaks.

PATH D: Browser — Web Speech API (Client-Side)

For browser-based apps with free client-side speech. Not for production video pipelines.

function speakMixed(text) {
  const voices = speechSynthesis.getVoices();
  if (!voices.length) {
    speechSynthesis.onvoiceschanged = () => speakMixed(text);
    return;
  }

  const parts = text.match(
    /[\u0600-\u06FF\u0750-\u077F\u08A0-\u08FF\uFB50-\uFDFF\uFE70-\uFEFF]+|[^\u0600-\u06FF\u0750-\u077F\u08A0-\u08FF\uFB50-\uFDFF\uFE70-\uFEFF]+/g
  ) || [];

  parts.forEach((part, index) => {
    const trimmed = part.trim();
    if (!trimmed) return;

    const isArabic = /[\u0600-\u06FF]/.test(trimmed);
    const utt = new SpeechSynthesisUtterance(trimmed);
    utt.lang = isArabic ? 'ar-SA' : 'en-US';
    utt.voice = voices.find(v =>
      v.lang && v.lang.toLowerCase().startsWith(isArabic ? 'ar' : 'en')
    ) || null;
    utt.rate = isArabic ? 0.95 : 1.0;
    utt.pitch = 1.0;

    utt.onerror = (e) => console.error(`Speech error segment ${index}:`, e.error);
    speechSynthesis.speak(utt);
  });
}

// Mobile Safari fix — must be triggered by user gesture
document.addEventListener('click', () => {
  if (speechSynthesis.paused) speechSynthesis.resume();
}, { once: true });

Voice Generation Anti-Error Protocols

BEFORE GENERATION (prevention):
[ ] Text not empty or whitespace-only
[ ] Language mode decided: pure (Path A) or mixed (Path B)?
[ ] If mixed: SSML properly formed with <lang> around every foreign term?
[ ] If mixed: FreeTTS request under 1,000 chars?
[ ] If pure: correct voice selected via Unicode detection?
[ ] Text length checked; split into chunks if over limit
[ ] Output path absolute and writable
[ ] For Arabic: text lightly diacritized or undiacritized (avoid heavy tashkeel)
[ ] Timeout set (10-20s) and retry logic active

AFTER GENERATION (verification):
[ ] File exists and size > 1 KB?
[ ] Duration reasonable? (~0.6-0.8s per 10 English chars / 8-10 Arabic chars)
[ ] No cut-off at end of audio?
[ ] For Path B: Arabic + English terms both pronounced correctly?
[ ] For Path B: <lang> tag actually switched pronunciation?
[ ] If any check fails → regenerate with different voice or switch to fallback engine
[ ] Log: timestamp + engine + voice + text length + file size + success/failure

FALLBACK CHAIN:
edge-tts → FreeTTS SSML → Kokoro-82M (English) / Piper (Arabic) → macOS 'say' command

1. Why no SSML mixed-language in Box 1?

edge-tts does not support <lang> or <voice> switching for seamless mixed-language audio.

edge-tts is a free wrapper around Microsoft's public Edge browser TTS endpoint. While it accepts basic SSML like <prosody> and <break>, it does not support <lang xml:lang="en-US"> or <voice> switches inside a single request. The engine reads the entire text with one voice only — if you give it Arabic text with an English voice, it butchers the Arabic; if you give it English text with an Arabic voice, it butchers the English.

That is why Box 1 is locked to pure Arabic OR pure English per generation. For mixed text, you must either: - Split into pure segments (Box 3), or - Use a service that supports SSML <lang> switching (Box 2 or the new FreeTTS option below).

2. Can an AI agent use Google Cloud / Azure free tiers automatically?

No — and this is a hard limit, not a prompt-engineering problem.

Bottom line: There is no legitimate way for an AI agent to "get" an Azure or Google API key without your manual account creation. Any claim otherwise is either a scam or a terms-of-service violation.

The Discovery: FreeTTS (freetts.org)

After searching, FreeTTS is the only free, no-signup, no-API-key service that supports full W3C SSML including mixed-language markup. It proxies Microsoft Azure Cognitive Services Neural TTS under the hood.

The catch: 1,000 characters per request is small — fine for agent responses, but long scripts need chunking (see Box 3).

📦 PROMPT BOX 1 — FREE No-Key AI-Agent Voice Pipeline

(For: AI agents that need to speak responses automatically. Covers pure language via edge-tts AND mixed language via FreeTTS SSML. Zero accounts, zero cards, zero API keys.)

MISSION: Generate natural, fluent, human-like speech for every AI-agent response using ONLY free, no-key tools. Two paths: (A) edge-tts for pure Arabic or pure English single-language output; (B) FreeTTS API for mixed Arabic-English output with full SSML <lang> switching. Never sound robotic.

CORE QUALITY MANDATE (always-on):
Correct speech words, pure correct sound, fluent and natural human-like delivery — never AI-sounding.

═══════════════════════════════════════════════════════════════
PATH A: PURE LANGUAGE — edge-tts (Microsoft Neural, free public endpoint)
═══════════════════════════════════════════════════════════════

Install: pip install edge-tts
No API key. No account. No credit card.

Recommended Voices:
- English: en-US-JennyNeural, en-US-GuyNeural, en-GB-SoniaNeural
- Arabic: ar-EG-SalmaNeural, ar-SA-HamedNeural, ar-AE-HamdanNeural, ar-JO-SanaNeural

CLI:
edge-tts --voice <VOICE> --text "<TEXT>" --write-media /tmp/out.mp3

Python:
import edge_tts, asyncio, re, hashlib, os

VOICE_CACHE = {}
async def load_voices():
    voices = await edge_tts.list_voices()
    for v in voices:
        if v["ShortName"] in ["en-US-JennyNeural","en-US-GuyNeural","en-GB-SoniaNeural",
                              "ar-EG-SalmaNeural","ar-SA-HamedNeural","ar-AE-HamdanNeural","ar-JO-SanaNeural"]:
            VOICE_CACHE[v["ShortName"]] = v
    return VOICE_CACHE

async def speak_pure(text, voice="en-US-JennyNeural"):
    if not text or not text.strip():
        return None
    safe_text = text.strip().replace('"', '\\"')
    output_path = "/tmp/edge_" + hashlib.md5((text+voice).encode()).hexdigest()[:8] + ".mp3"
    try:
        communicate = edge_tts.Communicate(safe_text, voice)
        await asyncio.wait_for(communicate.save(output_path), timeout=15.0)
        return output_path
    except Exception:
        await asyncio.sleep(2)
        try:
            communicate = edge_tts.Communicate(safe_text, voice)
            await asyncio.wait_for(communicate.save(output_path), timeout=15.0)
            return output_path
        except:
            return await fallback_speak(text, voice)

async def fallback_speak(text, voice):
    is_ar = bool(re.search(r'[\u0600-\u06FF]', text))
    # Fallback: Kokoro-82M for English, Piper for Arabic
    return "fallback_" + ("ar.wav" if is_ar else "en.wav")

Language Detection for Path A:
- Contains Arabic Unicode (\u0600-\u06FF): use ar-EG-SalmaNeural (default)
- Else: use en-US-JennyNeural (default)

═══════════════════════════════════════════════════════════════
PATH B: MIXED LANGUAGE — FreeTTS API (Full SSML, no key, proxies Azure)
═══════════════════════════════════════════════════════════════

Endpoint: POST https://freetts.org/api/tts
No API key. No signup. 15 req/min. 1,000 chars/request. 15,000 chars/month.

SSML Template for Mixed Arabic-English:
<speak version="1.0" xml:lang="ar-SA">
  <voice name="ar-SA-ZariyahNeural">
    النص العربي هنا
    <lang xml:lang="en-US">English term</lang>
    يستمر النص العربي
    <break time="300ms"/>
    <lang xml:lang="en-US">ICU</lang>
    هنا
  </voice>
</speak>

Python call:
import requests, json

def speak_mixed_ssml(ssml_text):
    url = "https://freetts.org/api/tts"
    payload = {
        "text": ssml_text,  # complete SSML document
        "voice": "ar-SA-ZariyahNeural",
        "format": "mp3"
    }
    try:
        r = requests.post(url, json=payload, timeout=20)
        r.raise_for_status()
        data = r.json()
        # data contains file_id or direct URL
        return data
    except Exception as e:
        # Retry once
        import time; time.sleep(2)
        r = requests.post(url, json=payload, timeout=20)
        return r.json()

IMPORTANT — FreeTTS Constraints:
- 1,000 characters max per request. If SSML exceeds this, split into chunks and stitch (see Box 3).
- No CORS headers by default. If calling from browser frontend, proxy through your backend.
- Free tier is personal use with a short audio watermark tag at end. For commercial/watermark-free, PRO is $19/mo.
- Backend is Azure Cognitive Services — <lang>, <voice>, <break>, <prosody>, <phoneme> tags should work, but VERIFY first.

═══════════════════════════════════════════════════════════════
ANTI-ERROR PROTOCOLS (Both Paths)
═══════════════════════════════════════════════════════════════
- Never send empty/whitespace text → skip or return silent success.
- Always escape internal quotes in text. For Arabic, ensure UTF-8 encoding.
- Validate voice name exists (cache edge-tts voice list at startup; FreeTTS uses fixed Azure voice names).
- Split long texts (>800 chars for edge-tts, >1000 chars for FreeTTS) into sentences/chunks.
- Add try/except; retry once with 2s delay, then fallback.
- Use absolute writable paths for output files.
- Do not mix rate/pitch parameters unless tested; keep defaults for stability.
- For Arabic: prefer lightly diacritized or undiacritized text; avoid heavy code-switching in one sentence if not using SSML Path B.
- Set timeout (10–20s) and enforce it.

═══════════════════════════════════════════════════════════════
VERIFICATION CHECKLIST (Every Generation)
═══════════════════════════════════════════════════════════════
[ ] File exists and size > 1 KB?
[ ] File extension correct (.mp3) and playable?
[ ] Duration reasonable (~0.6–0.8s per 10 English chars / 8–10 Arabic chars)?
[ ] Voice matches intended language/accent (log exact voice name)?
[ ] No cut-off at end of audio?
[ ] For Path B: play first 3–5 seconds — confirm Arabic + English terms both pronounced correctly?
[ ] For Path B: confirm <lang> tag actually switched pronunciation (spot-check one English term)?
[ ] If any check fails → regenerate with different voice or switch to fallback engine.
[ ] Log: timestamp + engine (edge-tts/FreeTTS) + voice + text length + file size + success/failure.

═══════════════════════════════════════════════════════════════
PREVENTION CHECKLIST — Before Any Generation
═══════════════════════════════════════════════════════════════
[ ] Text not empty or whitespace-only
[ ] Language mode decided: pure (Path A) or mixed (Path B)?
[ ] If mixed: SSML properly formed with <lang xml:lang="en-US"> around every English term?
[ ] If mixed: FreeTTS request under 1,000 chars?
[ ] If pure: correct voice selected via Unicode detection?
[ ] Voice name validated against cached list
[ ] Text length checked; split into chunks if over limit
[ ] Output path absolute and writable
[ ] For Arabic: text lightly diacritized or undiacritized
[ ] Timeout set (10–20s) and retry logic active
[ ] Fallback engine configured (Kokoro-82M for English, Piper for Arabic)

GOAL: Every response spoken with natural, pure, fluent, human-like quality — completely free, zero signup, with maximum reliability.

📦 PROMPT BOX 2 — Full Cloud SSML Production

For: When you eventually have API keys. Azure / Google Cloud / Amazon Polly — seamless mixed-language in one file with full control.

MISSION: Build a production-ready SSML pipeline that generates seamless mixed Arabic-English audio in one file. Use when API keys are available and perfect mid-sentence language switching is required.

CORE QUALITY MANDATE (always-on):
Correct speech words, pure correct sound, fluent and natural human-like delivery — never AI-sounding.

PRIMARY ENGINES:
- Microsoft Azure Neural TTS (ar-SA-ZariyahNeural, en-US-JennyNeural)
- Google Cloud TTS (ar-XA-Wavenet-A, en-US-Wavenet-F)
- Amazon Polly (Zeina for Arabic, Joanna for English)

SSML TEMPLATES:

Azure / Google Cloud (single voice, language switch):
<speak version="1.0" xml:lang="ar-SA">
  <voice name="ar-SA-ZariyahNeural">
    النص العربي هنا
    <lang xml:lang="en-US">English term</lang>
    يستمر النص
    <prosody rate="medium" pitch="default">نص مهم</prosody>
    <break time="300ms"/>
    النص التالي
  </voice>
</speak>

Amazon Polly (voice switch per segment):
<speak>
  <voice name="Zeina">النص العربي</voice>
  <voice name="Joanna">English term</voice>
  <voice name="Zeina">يستمر النص</voice>
</speak>

Phoneme override for stubborn terms:
<phoneme alphabet="ipa" ph="ˈmaɪ.kro.soft">Microsoft</phoneme>

WORKFLOW:
1. Collect all English terms from scripts.
2. Wrap each in <lang xml:lang="en-US">...</lang> (Azure/Google) or separate <voice> (Polly).
3. Add LIGHT tashkeel (diacritics) to Arabic text only where ambiguity exists — do not over-diacritize.
4. Decide number format: Arabic numerals (١٢٣) read naturally by Arabic voices; English numerals (123) need <say-as> or <lang> wrapping.
5. Spell out URLs phonetically or segment with SSML; never let the engine guess pronunciation.

ANTI-ERROR PROTOCOLS:
- Always validate SSML against provider schema before API call (Azure Speech Studio, Google Cloud console, Polly console).
- Wrap ALL English terms — missing one term ruins the entire segment's professionalism.
- Use <prosody rate="medium" pitch="default"> to prevent flat robotic delivery.
- Keep <break time="300ms"/> between sentences for breathing room.
- Cache IPA phoneme entries for recurring brand names to avoid re-typing.
- If SSML becomes too complex (>50 language switches in one file), switch to Box 3 (Split-and-Stitch) for reliability.
- Set API timeout to 20–30 seconds; retry once on network failure.
- Log raw SSML for debugging failed generations.

VERIFICATION CHECKLIST (audit every generation):
[ ] SSML validated against provider schema (no malformed tags)?
[ ] 30-second preview generated and manually reviewed?
[ ] English pronunciation accurate inside <lang> tags?
[ ] Arabic fluency natural after returning from English switch?
[ ] Transition smoothness between language switches acceptable?
[ ] Arabic diacritics present on ambiguous words (e.g., فَصْل vs فصل)?
[ ] Numbers read in intended language (test one Arabic numeral + one English numeral)?
[ ] No audio cut-offs at segment ends?
[ ] File size and duration reasonable for text length?
[ ] If one word sounds off → switch <lang> to <phoneme> with IPA spelling, then regenerate.
[ ] Log: provider + voice names + text hash + duration + success/failure.

PREVENTION CHECKLIST — Before batch generation:
[ ] All English terms wrapped in <lang xml:lang="en-US"> tags (or phoneme/IPA spelled)
[ ] Arabic text has light tashkeel/diacritics on ambiguous words only
[ ] Numbers format decided and tested (Arabic numerals vs English SSML <say-as>)
[ ] URLs either spelled out phonetically or segmented with SSML tags
[ ] Prosody tags (rate/pitch) added to prevent flat robotic delivery
[ ] <break> tags placed between sentences (300ms standard)
[ ] 30-second preview generated and manually reviewed — fix before batching
[ ] If >50 language switches in one file, redirect to Box 3 workflow
[ ] Provider quota and API key validity confirmed
[ ] API timeout set (20–30s) and retry logic active

GOAL: Seamless mixed-language audio with perfect pronunciation in both languages, professional broadcast quality.

📦 PROMPT BOX 3 — Long Scripts & Browser Apps — Advanced Workflows

For: Long-form content, heavy code-switching, or browser-based apps where Box 1 or Box 2 are insufficient.

MISSION: Handle long-form content, heavy code-switching, or browser-based apps. Provides two methods: Web Speech API for browsers, and Split-and-Stitch for production. Both can use edge-tts (Box 1) or FreeTTS/Azure SSML (Box 2) as underlying engines per segment.

CORE QUALITY MANDATE (always-on):
Correct speech words, pure correct sound, fluent and natural human-like delivery — never AI-sounding.

═══════════════════════════════════════════════════════════════
METHOD A: Web Speech API (Browser Translator / Learning Apps)
═══════════════════════════════════════════════════════════════
Use when building browser-based apps with free client-side speech.

Enhanced Implementation:
function speakMixed(text) {
  const voices = speechSynthesis.getVoices();
  if (!voices.length) {
    speechSynthesis.onvoiceschanged = () => speakMixed(text);
    return;
  }

  const parts = text.match(/[\u0600-\u06FF\u0750-\u077F\u08A0-\u08FF\uFB50-\uFDFF\uFE70-\uFEFF]+|[^\u0600-\u06FF\u0750-\u077F\u08A0-\u08FF\uFB50-\uFDFF\uFE70-\uFEFF]+/g) || [];

  parts.forEach((part, index) => {
    const trimmed = part.trim();
    if (!trimmed) return;

    const isArabic = /[\u0600-\u06FF]/.test(trimmed);
    const utt = new SpeechSynthesisUtterance(trimmed);
    utt.lang = isArabic ? 'ar-SA' : 'en-US';
    utt.voice = voices.find(v => v.lang && v.lang.toLowerCase().startsWith(isArabic ? 'ar' : 'en')) || null;
    utt.rate = isArabic ? 0.95 : 1.0;
    utt.pitch = 1.0;

    utt.onstart = () => console.log(`Speaking [${isArabic ? 'AR' : 'EN'}]:`, trimmed.substring(0, 30));
    utt.onerror = (e) => console.error(`Speech error segment ${index}:`, e.error);
    speechSynthesis.speak(utt);
  });
}

// Mobile Safari fix
document.addEventListener('click', () => {
  if (speechSynthesis.paused) speechSynthesis.resume();
}, { once: true });

═══════════════════════════════════════════════════════════════
METHOD B: Split-and-Stitch (Long Scripts / Heavy Mixing)
═══════════════════════════════════════════════════════════════
Use when SSML fails, scripts exceed TTS length limits, or heavy mixing makes single-file generation unreliable.

Workflow:
1. Pre-process script:
   - Highlight all English terms.
   - Extract them as pure-English segments.
   - Add light tashkeel to remaining Arabic text.
   - Convert numbers to the voice language that will read them (or spell out).
2. Generate audio per segment using best engine:
   - Arabic segments: ar-EG-SalmaNeural (edge-tts) or ar-SA-ZariyahNeural (FreeTTS/Azure)
   - English segments: en-US-JennyNeural (edge-tts) or ElevenLabs Multilingual v2
3. Name convention: [ORDER]_[LANG]_[TOPIC].mp3
   Example: 01_AR_intro.mp3, 02_EN_brandname.mp3, 03_AR_explanation.mp3
4. Stitch in editor/DAW:
   - Align in original order.
   - Silence gaps: 0.3s for mid-sentence switches, 0.5–0.8s for paragraph breaks.
   - Normalize loudness: -16 LUFS stereo / -19 LUFS mono.
   - Apply light compression if dynamics differ between engines.
5. Export final mix.

ANTI-ERROR PROTOCOLS:
- Web Speech: Always load voices asynchronously before first call; cache available voices.
- Web Speech: Test on Chrome, Safari, Firefox — voice catalogs differ significantly.
- Web Speech: Handle mobile autoplay with touch/click listener.
- Split-and-Stitch: Never leave mixed sentences in segments — purity is essential.
- Split-and-Stitch: If one segment fails, regenerate only that segment — do not re-render entire timeline.
- Split-and-Stitch: Keep individual segment backups before flattening final mix.
- For both methods: decide number pronunciation language upfront.
- For long scripts: if total text >5000 chars, break into chapters/sections before segmenting.

VERIFICATION CHECKLIST (Method A — Web Speech):
[ ] Voices loaded and cached before user interaction?
[ ] Tested edge cases: "السعر 50 دولار", "موقع Master1.vip", "ICU ward"?
[ ] No unnatural pauses between queued segments?
[ ] Works on mobile (iOS Safari tested)?
[ ] Fallback if no Arabic voice found on user's OS?

VERIFICATION CHECKLIST (Method B — Split-and-Stitch):
[ ] Segment count matches language switches in original script?
[ ] Each segment duration reasonable for word count (flag outliers)?
[ ] No gaps or overlaps at boundaries?
[ ] Perceived loudness consistent across Arabic and English segments?
[ ] Full script read-along test: audio aligns with original text word-for-word?
[ ] Tested on headphones AND speakers?
[ ] Master cue sheet exists with durations and gap timings?
[ ] Individual segment backups kept before final mix flattening?

PREVENTION CHECKLIST — Before deployment:
[ ] Arabic text has light tashkeel on ambiguous words
[ ] Numbers format decided: which voice reads them?
[ ] URLs spelled out phonetically or pre-processed
[ ] For Web Speech: voices loaded, fallback configured, mobile handler added
[ ] For Split-and-Stitch: segment naming convention applied, cue sheet created
[ ] Audio levels normalized across all segments (target LUFS matched)
[ ] Silence gaps set: 0.3s inline, 0.6s paragraph
[ ] Full track previewed while reading along
[ ] Backups of individual segments kept before final mix

GOAL: Perfect multi-language audio for any length or platform — browser apps via Web Speech API, professional long-form via Split-and-Stitch — with flawless reliability and broadcast-quality output.

Summary of what changed:

Complete AI-agent pipeline: image understanding → transcript matching → voice generation → video assembly.
Covers: vision preprocessing, image-to-transcript matching, voice-to-text (transcription), text-to-voice (TTS), zoompan rendering, and end-to-end coherence.

MERGED PROMPT BOX 1 — Visualized Model (Full Pipeline)

ROLE: You are a Cinematic Content Extraction & Matching Engine. 
Your output feeds a voice-over generator and a video editor. 
Accuracy > creativity. You must SEE before you SPEAK.

INPUT PER TURN: 
- One transcript chunk (with start/end timestamps)
- A candidate pool of images labeled A, B, C, D... (attached in this same message)
- Previous frame extraction (from prior turn, for continuity)

STEP 0 — ANTI-ADVERSARY CHECK (new from doc):
Before analyzing, state: "I have received [N] images labeled A through [N]."
If any image shows a visible human face, real or illustrated, flag it immediately 
with [FACE RULE VIOLATION: Image X] and exclude it from selection.

STEP 1 — DESCRIBE BEFORE DECIDE (new from doc):
For EACH candidate image, output a description table BEFORE making any selection:
| Image ID | Dominant Colors (max 3) | How Many Hands? | Visible Text? (Y/N + transcript) | Face Visible? (Y/N) | Primary Action | Confidence (High/Med/Low) |

STEP 2 — GROUND-TRUTH CHECK (new from doc):
Answer these exactly:
- Which image has the most visible text matching the transcript chunk?
- Which image shows an action that directly illustrates the transcript verb?
- Are any two images nearly identical? If yes, note [NEAR-DUPLICATE: X and Y].
- Is there any image where no visual element relates to the transcript? Note [NO MATCH: X].

STEP 3 — SELECTION & COORDINATES (enhanced with focal point):
Output valid JSON only:
{
  "chosen_image": "B",
  "confidence": 0.91,
  "has_visible_face": false,
  "focal_point": {"x": 0.35, "y": 0.62, "reason": "hands demonstrating grip technique"},
  "match_type": "direct_action | mood_match | text_match | no_match",
  "reasoning": "Image B shows hands adjusting a mask surface, which matches the transcript line about preparation. No face is visible. Focal point is on the hands for zoompan."
}

RULE: If no image scores above 0.7 confidence or if all candidates violate constraints, 
set "chosen_image": null and "match_type": "no_match". Do not force a pick.

STEP 4 — VOICE-OVER ALIGNMENT:
Original Transcript: "[paste chunk]"
Visual Accuracy Score (0-10): [score]
Mismatches: [bullet list, or "None"]
Aligned Rewrite (max 25 words for short-form, 60 for long-form): "[rewrite]"
SEO Keywords Woven In: [list which keywords appear and where]

STEP 5 — VOICE CUE:
| Tone | Pace (WPS) | Emphasis Word | SFX Gap | Pronunciation Alert |

MERGED PROMPT BOX 4 — Lightweight Vision Preprocessor (New, bridges non-visual gap)

Use this with a small local VLM (Qwen2-VL 3B, LLaVA-Phi3, or even cloud Gemini flash) to generate the structured data that feeds Prompt Box 3.

ROLE: You are a Vision Preprocessor. Your job is to look at an image once 
and store everything a text-only model will need later. Be exhaustive; 
creativity is forbidden.

INPUT: One image file + image_id

OUTPUT — STRICT JSON:
{
  "image_id": "string",
  "caption": "Detailed description: subject, action, setting, mood, lighting. Max 40 words.",
  "has_face": true/false,
  "has_text_overlay": true/false,
  "text_transcription": "exact text if any, else null",
  "dominant_colors": ["color1", "color2", "color3"],
  "primary_action": "verb phrase, e.g., 'figure grips table edge'",
  "hand_count": 0/1/2/null,
  "focal_point": {"x": 0.0-1.0, "y": 0.0-1.0, "description": "what is at this coordinate"},
  "setting": "interior/exterior/abstract",
  "lighting_mood": "high-key/low-key/chiaroscuro/flat/neon",
  "faceless_compliance": {
    "mask_present": true/false,
    "facial_features_visible": true/false,
    "notes": "any ambiguity about face visibility"
  },
  "confidence": "High/Med/Low",
  "ambiguities": ["list anything unclear, or empty array"]
}

EXTRACTION RULES:
- For faceless/masked figures: describe mask angle, head tilt, shoulder tension, hand gestures. Never infer emotion from a missing face.
- If text is stylized/low-contrast, transcribe what you can read and mark the rest [ILLEGIBLE].
- If the image is a manga panel, note panel borders, speed lines, and screentone separately from content.
- Focal point must be on the narrative center of interest, not geometric center.

MERGED PROMPT BOX 5 — Adversarial Quality Test (New from doc)

Run this weekly or when adding a new model to your pipeline.

ROLE: Adversarial Test Designer

Generate a test batch of 5 image sets. Each set contains:
- 1 correct match (aligns perfectly with a transcript chunk)
- 1 near-miss (same keywords, wrong visual content)
- 1 face-rule violation (visible face, to test compliance)
- 1 low-quality/compressed image (to test artifact handling)
- 1 abstract/mood-only image (no clear action, to test "no_match" option)

For each set, provide:
| Set ID | Transcript Chunk | Images | Expected Behavior |
|--------|------------------|--------|-------------------|
| T01 | "He waited in silence" | [correct: silhouette], [near-miss: "silent library" with person reading], [violation: close-up portrait], [lowq: blurry dark frame], [abstract: empty chair] | Choose silhouette; reject portrait; flag lowq; possibly choose abstract or null |

ADMINISTER THE TEST:
Feed each set to your production vision model using Prompt Box 1.
Score: +1 for correct pick, +1 for rejecting violation, +1 for flagging lowq, 
+1 for using "no_match" appropriately. -2 for picking violation or near-miss.

PASS THRESHOLD: 80% or higher. If failed, tune Step 0/Step 2 in Prompt Box 1.

MERGED PROMPT BOX 3 — Non-Visualized Model (Orchestrator)

This replaces my previous non-visual prompt. It now explicitly references the embedding/caption architecture from the document.

ROLE: You are a Video Pipeline Orchestrator. You do NOT see images. 
You reason over structured data produced by a vision preprocessor.

INPUT SCHEMA (you will receive this from SQLite/JSON store):
{
  "video_project": "masked_assembly_ep07",
  "transcript_chunks": [
    {"id": 1, "start": 0.0, "end": 4.2, "text": "He adjusted the mask before entering.", "seo_keywords": ["psychological thriller","faceless"]}
  ],
  "candidate_pool": [
    {"image_id": "img_042", "caption": "masked figure, left hand touching mask surface, low light, no face visible", "has_face": false, "dominant_colors": ["black","red"], "clip_embedding_available": true, "last_used": "ep06"}
  ]
}

YOUR ORCHESTRATION TASKS:

TASK 1 — TRANSCRIPT PREP
If the audio has no transcript at all, generate one first:
- Run whisper.cpp: whisper-cli -m models/ggml-base.en.bin -f audio.wav --vad --output-txt --output-srt
- For non-English: add --language auto
- For word-level timing: add --output-words
If timestamps are not word-level, instruct the system to run whisper.cpp 
with --output-words before proceeding.

TASK 2 — MATCHING (choose one method and state which):

METHOD A: EMBEDDING SIMILARITY (recommended for volume)
Instruct the system to:
- Encode each transcript chunk text with CLIP text encoder
- Encode each candidate caption with CLIP text encoder  
- Compute cosine similarity
- Return top-3 matches per chunk

METHOD B: CAPTION KEYWORD MATCH (fallback if no CLIP)
Instruct the system to:
- Extract nouns/verbs from transcript chunk
- Match against candidate captions
- Return top-3 matches per chunk

TASK 3 — COMPLIANCE FILTER
For each top match, enforce:
- has_face must be false (hard reject if true)
- If last_used is within the last 2 videos, deprioritize (avoid overuse)
- If dominant_colors clash with previous frame's palette, flag [COLOR JUMP]

TASK 4 — GAP HANDLING
If no candidate scores above threshold for a chunk, output:
{"chunk_id": N, "image_id": null, "action": "GENERATE_FILLER or EXTEND_PREVIOUS"}

TASK 5 — VOICE GENERATION
For each finalized voice-over text chunk, generate audio:
- Pure language: edge-tts (Path A — see Voice Generation section)
- Mixed Arabic-English: FreeTTS SSML (Path B — see Voice Generation section)
- Long scripts (>1000 chars): Split-and-Stitch method
Store generated audio path in timeline entry.

TASK 6 — OUTPUT TIMELINE
Produce final JSON timeline for the video editor:
[
  {"chunk_id": 1, "start": 0.0, "end": 4.2, "image_id": "img_042", 
   "match_method": "clip_embedding", "compliance": "pass", 
   "voice_over": "He adjusted the mask before entering.",
   "voice_audio": "/tmp/vo_0001.mp3", "voice_engine": "edge-tts"}
]

ANTI-HALLUCINATION RULES:
- Never invent image details not present in the candidate_pool schema.
- If a candidate's caption is ambiguous, flag [AMBIGUOUS: img_X] and request human review.
- SEO keywords must appear in the voice_over naturally; do not append them as tags.

MERGED PROMPT BOX 2 — End-to-End Coherence Pass (New from doc)

Run this AFTER all chunks are matched. Feed the model the full sequence, not individual chunks.

ROLE: Sequence Coherence Auditor

INPUT: 
- Full ordered list of chosen images with their frame IDs
- Full transcript
- Per-frame JSON outputs from previous step

TASK: Review the entire sequence as a viewer would experience it. Output:

---REPETITION CHECK---
Any image used more than once? List occurrences.
Any visual motif repeated too soon? Flag [REPETITION FATIGUE].

---JUMP CHECK---
Any transition where lighting, setting, or character posture changes abruptly 
without narrative justification? Flag [JARRING JUMP] and suggest a bridge frame.

---FACELESS CONTINUITY---
Confirm every frame has has_visible_face: false. 
Any frame missing this field or flagged true → [COMPLIANCE FAILURE].

---PACING AUDIT---
Mark frames where the voice-over is dense but the image is static → suggest [HOLD + ZOOMPAN].
Mark frames where the image is complex but the voice-over is sparse → suggest [EXTEND DWELL TIME].

---FINAL SEQUENCE JSON---
Output the corrected timeline as JSON array:
[
  {"frame_id": "06_003", "image": "B", "start": 0.0, "end": 4.5, 
   "focal_point": {"x":0.35,"y":0.62}, "zoompan": "slow_push_in", 
   "voice_over": "...", "compliance": "pass"}
]

How all sections connect end-to-end:

┌─────────────────────────────────────────────────────────────────────────┐
│                        AUDIO INPUT                                     │
│  (raw audio file with no transcript)                                   │
└──────────────────────────┬──────────────────────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────────────────┐
│  Section 6: TRANSCRIPTION (Voice → Text)                               │
│  whisper-cli --vad --output-txt --output-srt --output-json             │
│  → Produces: transcript chunks with timestamps                         │
│  → Verification: word count, hallucination check, language match       │
└──────────────────────────┬──────────────────────────────────────────────┘
                           │
           ┌───────────────┼───────────────┐
           ▼                               ▼
┌──────────────────────┐     ┌──────────────────────┐
│  Prompt Box 4        │     │  Prompt Box 1        │
│  Vision Preprocessor │     │  Visual Model        │
│  (run once per img)  │     │  (per chunk)         │
└──────────┬───────────┘     └──────────┬───────────┘
           │                            │
           ▼                            │
┌──────────────────────┐                │
│  SQLite Store        │◄───────────────┘
│  (image_metadata,    │    focal_point{x,y}
│   transcript_chunks) │
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│  Prompt Box 3        │
│  Orchestrator        │
│  (matching + gaps)   │
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│  Section 7: TTS      │
│  (Text → Speech)     │
│  edge-tts / FreeTTS  │
│  → voice audio files │
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│  Prompt Box 2        │
│  Coherence Auditor   │
│  (sequence check)    │
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│  FFmpeg Zoompan      │
│  Engine              │
│  → render segments   │
│  → concatenate       │
│  → final video       │
└──────────────────────┘

Both the following are standalone Python scripts. The SQLite bridge requires numpy (for embeddings) and sqlite3 (built-in). The Zoompan engine is pure stdlib.

┌─────────────────┐     ┌──────────────────────┐     ┌─────────────────┐
│  Prompt Box 1   │     │   Prompt Box 4       │     │  Prompt Box 3   │
│  Visual Model   │     │  Vision Preprocessor │     │  Orchestrator   │
│  (per chunk)    │     │  (run once per img)  │     │  (text-only LLM)│
└────────┬────────┘     └──────────┬───────────┘     └────────┬────────┘
         │                         │                          │
    focal_point{x,y}         →  SQLite (File 1)  ←    queries by embedding
         │                         │                          │
         └─────────────────────────┼──────────────────────────┘
                                   │
                    ┌──────────────▼──────────────┐
                    │   video_timeline table      │
                    │   (sequence + zoompan)      │
                    └──────────────┬──────────────┘
                                   │
                    ┌──────────────▼──────────────┐
                    │  FFmpeg Zoompan Engine      │
                    │       (File 2)              │
                    │  → generates bash script    │
                    │  → renders all segments     │
                    │  → concatenates final video │
                    └─────────────────────────────┘

If you use a visualized model (Gemini/Claude/Grok): Use Prompt Box 1 per chunk → Prompt Box 2 at sequence end → Prompt Box 5 weekly for QA.

If you use a non-visualized model (text-only LLM on CPU): Use Prompt Box 4 (run once per image via small VLM or cloud API) → store in SQLite → Prompt Box 3 (orchestrator) for every video → Prompt Box 2 (coherence pass, but now the orchestrator runs it as a script calling the text model).

Human spot-check: The document recommends checking 5-10% of outputs manually. I recommend applying this to Prompt Box 2's output (the final timeline JSON) — it's cheaper to audit one JSON array than 50 individual frame decisions.

What it does: - Defines the full database schema that sits between Prompt Box 4 (Vision Preprocessor) and Prompt Box 3 (Orchestrator). - Handles CLIP embedding similarity search with recency/overuse penalties. - Runs the Prompt Box 2 coherence audit (repetition, color jumps, compliance failures) automatically. - Stores every image once, queries it forever — no pixels touched again after preprocessing.

Key tables: | Table | Purpose | |---|---| | image_metadata | One row per image. Stores caption, faceless compliance, focal point, CLIP embedding, use count, last used project. | | transcript_chunks | One row per voice-over segment. Stores text, timestamps, SEO keywords, matched image, compliance status. | | video_timeline | Final render timeline. Links chunks to images with zoompan params and render status. | | adversarial_tests | Logs from Prompt Box 5 quality tests. |

How to use it:

from video_pipeline_sqlite import VideoPipelineDB
db = VideoPipelineDB("masked_assembly.db")

# 1. After Vision Preprocessor (Prompt Box 4) runs:
db.insert_image_metadata(preprocessor_json_output)

# 2. During Orchestrator (Prompt Box 3) matching:
matches = db.find_matches_by_embedding(query_embedding, project="masked_assembly")

# 3. After timeline is built:
issues = db.audit_timeline(project_id="ep07")
# Returns: repetition_fatigue, color_jumps, compliance_failures

"""
SQLite Schema & Pipeline Bridge
Connects Vision Preprocessor (Prompt Box 4) → Orchestrator (Prompt Box 3)
"""

import sqlite3
import json
import numpy as np
from typing import Optional, List, Dict, Any, Tuple
from dataclasses import dataclass
from datetime import datetime

# ============================================================
# 1. DATABASE SCHEMA
# ============================================================

SCHEMA_SQL = """
-- Main image metadata table (output of Vision Preprocessor / Prompt Box 4)
CREATE TABLE IF NOT EXISTS image_metadata (
    image_id            TEXT PRIMARY KEY,
    project             TEXT NOT NULL,
    file_path           TEXT NOT NULL UNIQUE,
    caption             TEXT NOT NULL,           -- max 40 words description
    has_face            INTEGER NOT NULL DEFAULT 0 CHECK (has_face IN (0,1)),
    has_text_overlay    INTEGER NOT NULL DEFAULT 0 CHECK (has_text_overlay IN (0,1)),
    text_transcription  TEXT,                    -- exact text, or NULL
    dominant_colors     TEXT NOT NULL,           -- JSON array ["black","red"]
    primary_action      TEXT,                    -- verb phrase
    hand_count          INTEGER CHECK (hand_count IN (0,1,2)),
    focal_point_x       REAL CHECK (focal_point_x BETWEEN 0.0 AND 1.0),
    focal_point_y       REAL CHECK (focal_point_y BETWEEN 0.0 AND 1.0),
    focal_description   TEXT,                    -- what is at focal point
    setting             TEXT CHECK (setting IN ('interior','exterior','abstract')),
    lighting_mood       TEXT CHECK (lighting_mood IN ('high-key','low-key','chiaroscuro','flat','neon')),
    mask_present        INTEGER DEFAULT 0 CHECK (mask_present IN (0,1)),
    facial_features_visible INTEGER DEFAULT 0 CHECK (facial_features_visible IN (0,1)),
    compliance_notes    TEXT,                    -- faceless ambiguity notes
    confidence          TEXT CHECK (confidence IN ('High','Med','Low')),
    ambiguities         TEXT NOT NULL DEFAULT '[]', -- JSON array
    clip_embedding      BLOB,                    -- 512-dim float32 numpy bytes
    clip_embedding_dim  INTEGER DEFAULT 512,
    created_at          TEXT NOT NULL DEFAULT (datetime('now')),
    last_used_project   TEXT,                    -- last video that used this image
    last_used_at        TEXT,                    -- datetime of last use
    use_count           INTEGER NOT NULL DEFAULT 0,
    is_archived         INTEGER NOT NULL DEFAULT 0 CHECK (is_archived IN (0,1))
);

-- Video project registry
CREATE TABLE IF NOT EXISTS video_projects (
    project_id          TEXT PRIMARY KEY,
    project_name        TEXT NOT NULL,
    platform            TEXT,                    -- youtube/shorts/tiktok/reels
    style_preset        TEXT DEFAULT 'manga_thriller',
    created_at          TEXT NOT NULL DEFAULT (datetime('now')),
    status              TEXT DEFAULT 'draft' CHECK (status IN ('draft','rendering','published','archived'))
);

-- Transcript chunks (input to orchestrator)
CREATE TABLE IF NOT EXISTS transcript_chunks (
    chunk_id            INTEGER PRIMARY KEY AUTOINCREMENT,
    project_id          TEXT NOT NULL REFERENCES video_projects(project_id),
    sequence_order      INTEGER NOT NULL,
    start_time          REAL NOT NULL,           -- seconds
    end_time            REAL NOT NULL,
    duration            REAL GENERATED ALWAYS AS (end_time - start_time) STORED,
    text                TEXT NOT NULL,
    seo_keywords        TEXT NOT NULL DEFAULT '[]', -- JSON array
    voice_tone          TEXT,
    pace_wps            REAL,                    -- words per second target
    emphasis_word       TEXT,
    sfx_gap             INTEGER DEFAULT 0 CHECK (sfx_gap IN (0,1)),
    pronunciation_alert TEXT,
    matched_image_id    TEXT REFERENCES image_metadata(image_id),
    match_confidence    REAL,
    match_method        TEXT CHECK (match_method IN ('clip_embedding','caption_keyword','manual','no_match')),
    zoompan_preset      TEXT DEFAULT 'none',
    final_voice_over    TEXT,
    voice_audio_path    TEXT,                    -- path to generated TTS audio
    voice_engine        TEXT,                    -- edge-tts / freetts / kokoro / piper
    compliance_status   TEXT DEFAULT 'pending' CHECK (compliance_status IN ('pending','pass','fail','flagged')),
    created_at          TEXT NOT NULL DEFAULT (datetime('now'))
);

-- Final timeline (output of orchestrator + coherence pass)
CREATE TABLE IF NOT EXISTS video_timeline (
    timeline_id         INTEGER PRIMARY KEY AUTOINCREMENT,
    project_id          TEXT NOT NULL REFERENCES video_projects(project_id),
    chunk_id            INTEGER NOT NULL REFERENCES transcript_chunks(chunk_id),
    sequence_order      INTEGER NOT NULL,
    image_id            TEXT REFERENCES image_metadata(image_id),
    start_time          REAL NOT NULL,
    end_time            REAL NOT NULL,
    duration            REAL GENERATED ALWAYS AS (end_time - start_time) STORED,
    focal_point_x       REAL,
    focal_point_y       REAL,
    zoompan_params      TEXT,                    -- JSON of FFmpeg params
    voice_over          TEXT NOT NULL,
    voice_audio_path    TEXT,                    -- path to TTS audio file
    voice_engine        TEXT,                    -- which TTS engine was used
    match_method        TEXT,
    compliance          TEXT DEFAULT 'pending' CHECK (compliance IN ('pass','fail','flagged')),
    coherence_notes     TEXT,                    -- from Prompt Box 2 audit
    render_status       TEXT DEFAULT 'pending' CHECK (render_status IN ('pending','queued','rendering','done','error'))
);

-- Adversarial test log (Prompt Box 5)
CREATE TABLE IF NOT EXISTS adversarial_tests (
    test_id             INTEGER PRIMARY KEY AUTOINCREMENT,
    test_date           TEXT NOT NULL DEFAULT (datetime('now')),
    model_version       TEXT,
    set_id              TEXT NOT NULL,
    transcript_chunk    TEXT NOT NULL,
    expected_behavior   TEXT NOT NULL,
    actual_behavior     TEXT NOT NULL,
    score               INTEGER NOT NULL,
    passed              INTEGER NOT NULL CHECK (passed IN (0,1))
);

-- Indexes for performance
CREATE INDEX IF NOT EXISTS idx_img_project ON image_metadata(project);
CREATE INDEX IF NOT EXISTS idx_img_faceless ON image_metadata(has_face, is_archived);
CREATE INDEX IF NOT EXISTS idx_img_last_used ON image_metadata(last_used_at);
CREATE INDEX IF NOT EXISTS idx_chunk_project ON transcript_chunks(project_id, sequence_order);
CREATE INDEX IF NOT EXISTS idx_timeline_project ON video_timeline(project_id, sequence_order);
"""

# ============================================================
# 2. DATABASE CLASS
# ============================================================

class VideoPipelineDB:
    def __init__(self, db_path: str = "video_pipeline.db"):
        self.db_path = db_path
        self.conn = sqlite3.connect(db_path)
        self.conn.row_factory = sqlite3.Row
        self._init_schema()

    def _init_schema(self):
        self.conn.executescript(SCHEMA_SQL)
        self.conn.commit()

    # --------------------------------------------------------
    # VISION PREPROCESSOR INSERT (Prompt Box 4 output)
    # --------------------------------------------------------
    def insert_image_metadata(self, data: Dict[str, Any]) -> str:
        """
        Insert output from Vision Preprocessor (Prompt Box 4).
        data keys match the JSON schema exactly.
        """
        focal = data.get("focal_point", {})
        faceless = data.get("faceless_compliance", {})

        # Convert numpy embedding to bytes if present
        embedding_bytes = None
        embedding_dim = 512
        clip_emb = data.get("clip_embedding")
        if clip_emb is not None:
            emb = np.array(clip_emb, dtype=np.float32)
            embedding_bytes = emb.tobytes()
            embedding_dim = len(clip_emb)

        sql = """
        INSERT OR REPLACE INTO image_metadata (
            image_id, project, file_path, caption, has_face, has_text_overlay,
            text_transcription, dominant_colors, primary_action, hand_count,
            focal_point_x, focal_point_y, focal_description, setting, lighting_mood,
            mask_present, facial_features_visible, compliance_notes, confidence,
            ambiguities, clip_embedding, clip_embedding_dim
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """
        self.conn.execute(sql, (
            data["image_id"],
            data.get("project", "default"),
            data.get("file_path", ""),
            data["caption"],
            int(data.get("has_face", False)),
            int(data.get("has_text_overlay", False)),
            data.get("text_transcription"),
            json.dumps(data.get("dominant_colors", [])),
            data.get("primary_action"),
            data.get("hand_count"),
            focal.get("x"),
            focal.get("y"),
            focal.get("description"),
            data.get("setting"),
            data.get("lighting_mood"),
            int(faceless.get("mask_present", False)),
            int(faceless.get("facial_features_visible", False)),
            faceless.get("notes"),
            data.get("confidence", "Med"),
            json.dumps(data.get("ambiguities", [])),
            embedding_bytes,
            embedding_dim
        ))
        self.conn.commit()
        return data["image_id"]

    # --------------------------------------------------------
    # ORCHESTRATOR: CLIP Embedding Match (Method A)
    # --------------------------------------------------------
    def find_matches_by_embedding(
        self,
        query_embedding: np.ndarray,
        project: str,
        top_k: int = 3,
        exclude_recent_projects: List[str] = None,
        min_confidence: str = "Med"
    ) -> List[Dict[str, Any]]:
        """
        Find top-K image matches using CLIP embedding cosine similarity.
        Hard-filter: has_face=0, is_archived=0, confidence >= min_confidence.
        Soft-filter: deprioritize images used in exclude_recent_projects.
        """
        confidence_rank = {"High": 3, "Med": 2, "Low": 1}
        min_rank = confidence_rank.get(min_confidence, 2)

        cursor = self.conn.execute("""
            SELECT image_id, file_path, caption, has_face, dominant_colors,
                   focal_point_x, focal_point_y, last_used_project, use_count,
                   clip_embedding, confidence
            FROM image_metadata
            WHERE project = ? AND has_face = 0 AND is_archived = 0
        """, (project,))

        results = []
        q_norm = query_embedding / (np.linalg.norm(query_embedding) + 1e-9)

        for row in cursor:
            if row["clip_embedding"] is None:
                continue
            emb = np.frombuffer(row["clip_embedding"], dtype=np.float32)
            emb_norm = emb / (np.linalg.norm(emb) + 1e-9)
            similarity = float(np.dot(q_norm, emb_norm))

            # Confidence gate
            if confidence_rank.get(row["confidence"], 0) < min_rank:
                continue

            # Deprioritize recently used
            recency_penalty = 0.0
            if exclude_recent_projects and row["last_used_project"] in exclude_recent_projects:
                recency_penalty = 0.15

            # Overuse penalty
            overuse_penalty = min(row["use_count"] * 0.02, 0.1)

            adjusted_score = similarity - recency_penalty - overuse_penalty

            results.append({
                "image_id": row["image_id"],
                "file_path": row["file_path"],
                "caption": row["caption"],
                "focal_point": {"x": row["focal_point_x"], "y": row["focal_point_y"]},
                "dominant_colors": json.loads(row["dominant_colors"]),
                "similarity": similarity,
                "adjusted_score": adjusted_score,
                "last_used": row["last_used_project"],
                "use_count": row["use_count"],
                "match_method": "clip_embedding"
            })

        results.sort(key=lambda x: x["adjusted_score"], reverse=True)
        return results[:top_k]

    # --------------------------------------------------------
    # ORCHESTRATOR: Caption Keyword Match (Method B fallback)
    # --------------------------------------------------------
    def find_matches_by_caption(
        self,
        keywords: List[str],
        project: str,
        top_k: int = 3,
        exclude_recent_projects: List[str] = None
    ) -> List[Dict[str, Any]]:
        """
        Fallback keyword match against captions.
        Hard-filter: has_face=0, is_archived=0.
        """
        cursor = self.conn.execute("""
            SELECT image_id, file_path, caption, dominant_colors,
                   focal_point_x, focal_point_y, last_used_project, use_count
            FROM image_metadata
            WHERE project = ? AND has_face = 0 AND is_archived = 0
        """, (project,))

        results = []
        keyword_set = set(k.lower() for k in keywords)

        for row in cursor:
            caption_words = set(row["caption"].lower().split())
            match_count = len(keyword_set & caption_words)
            score = match_count / max(len(keyword_set), 1)

            recency_penalty = 0.0
            if exclude_recent_projects and row["last_used_project"] in exclude_recent_projects:
                recency_penalty = 0.15

            overuse_penalty = min(row["use_count"] * 0.02, 0.1)
            adjusted_score = score - recency_penalty - overuse_penalty

            results.append({
                "image_id": row["image_id"],
                "file_path": row["file_path"],
                "caption": row["caption"],
                "focal_point": {"x": row["focal_point_x"], "y": row["focal_point_y"]},
                "dominant_colors": json.loads(row["dominant_colors"]),
                "similarity": score,
                "adjusted_score": adjusted_score,
                "last_used": row["last_used_project"],
                "use_count": row["use_count"],
                "match_method": "caption_keyword"
            })

        results.sort(key=lambda x: x["adjusted_score"], reverse=True)
        return results[:top_k]

    # --------------------------------------------------------
    # MARK IMAGE AS USED (update after timeline is built)
    # --------------------------------------------------------
    def mark_image_used(self, image_id: str, project_id: str):
        self.conn.execute("""
            UPDATE image_metadata
            SET last_used_project = ?, last_used_at = datetime('now'), use_count = use_count + 1
            WHERE image_id = ?
        """, (project_id, image_id))
        self.conn.commit()

    # --------------------------------------------------------
    # COHERENCE PASS: Detect repetition, color jumps, lighting jumps
    # --------------------------------------------------------
    def audit_timeline(self, project_id: str) -> Dict[str, Any]:
        """
        Run Prompt Box 2 checks on the built timeline.
        Returns: repetition_flags, color_jumps, compliance_failures, jarring_jumps
        """
        timeline = self.conn.execute("""
            SELECT t.sequence_order, t.image_id, t.compliance, t.voice_over,
                   i.dominant_colors, i.has_face, i.focal_point_x, i.focal_point_y,
                   i.lighting_mood
            FROM video_timeline t
            LEFT JOIN image_metadata i ON t.image_id = i.image_id
            WHERE t.project_id = ?
            ORDER BY t.sequence_order
        """, (project_id,)).fetchall()

        issues = {
            "repetition_fatigue": [],
            "color_jumps": [],
            "compliance_failures": [],
            "jarring_jumps": []
        }

        image_usage = {}
        prev_colors = None
        prev_lighting = None

        for i, row in enumerate(timeline):
            img_id = row["image_id"]

            # Repetition check
            if img_id in image_usage:
                issues["repetition_fatigue"].append({
                    "sequence": row["sequence_order"],
                    "image_id": img_id,
                    "previous_use": image_usage[img_id]
                })
            image_usage[img_id] = row["sequence_order"]

            # Compliance check
            if row["has_face"] == 1:
                issues["compliance_failures"].append({
                    "sequence": row["sequence_order"],
                    "image_id": img_id,
                    "issue": "FACE_VISIBLE"
                })

            # Color jump check
            curr_colors = set(json.loads(row["dominant_colors"]) if row["dominant_colors"] else [])
            if prev_colors and not curr_colors.intersection(prev_colors):
                issues["color_jumps"].append({
                    "sequence": row["sequence_order"],
                    "from_colors": list(prev_colors),
                    "to_colors": list(curr_colors)
                })

            # Jarring lighting jump check
            curr_lighting = row["lighting_mood"]
            if prev_lighting and curr_lighting and prev_lighting != curr_lighting:
                # Flag abrupt mood shifts (e.g. high-key → low-key)
                harsh_pairs = {("high-key", "low-key"), ("low-key", "high-key"),
                               ("neon", "chiaroscuro"), ("chiaroscuro", "neon")}
                if (prev_lighting, curr_lighting) in harsh_pairs:
                    issues["jarring_jumps"].append({
                        "sequence": row["sequence_order"],
                        "from_lighting": prev_lighting,
                        "to_lighting": curr_lighting
                    })

            prev_colors = curr_colors
            prev_lighting = curr_lighting

        return issues

    def close(self):
        self.conn.close()


# ============================================================
# 3. USAGE EXAMPLE
# ============================================================

if __name__ == "__main__":
    db = VideoPipelineDB("masked_assembly.db")

    # Example: Insert vision preprocessor output
    sample_image = {
        "image_id": "ep07_frame_042",
        "project": "masked_assembly",
        "file_path": "/assets/ep07/frame_042.png",
        "caption": "masked figure left hand touching mask surface low light no face visible",
        "has_face": False,
        "has_text_overlay": False,
        "text_transcription": None,
        "dominant_colors": ["black", "deep_red", "pale_white"],
        "primary_action": "figure touches mask surface",
        "hand_count": 1,
        "focal_point": {"x": 0.35, "y": 0.62, "description": "left hand on mask"},
        "setting": "interior",
        "lighting_mood": "low-key",
        "faceless_compliance": {
            "mask_present": True,
            "facial_features_visible": False,
            "notes": "clean mask, no eyes mouth or nose visible"
        },
        "confidence": "High",
        "ambiguities": [],
        "clip_embedding": np.random.randn(512).astype(np.float32).tolist()  # placeholder
    }
    db.insert_image_metadata(sample_image)
    print(f"Inserted: {sample_image['image_id']}")

    # Example: Match by embedding
    query = np.random.randn(512).astype(np.float32)
    matches = db.find_matches_by_embedding(
        query, project="masked_assembly", top_k=3,
        exclude_recent_projects=["ep06"]
    )
    print(f"\nTop matches: {matches}")

    db.close()

Full class and __main__ example are in the Database Class box above. Keep schema + methods + usage together so no line is dropped.

What it does: - Consumes focal_point: {x, y} from Prompt Box 1 visual model output. - Converts normalized coordinates (0.0-1.0) into exact FFmpeg zoompan viewport pixel math. - Generates 8 motion presets: push_in, pull_out, pan_left/right/up/down, drift, hold. - Builds single-clip render commands and full bash scripts for timeline concatenation. - Includes an auto-effect selector that picks motion based on primary_action + lighting_mood from the vision preprocessor.

Example output:

ffmpeg -y -loop 1 -framerate 30 -i '/assets/ep07/frame_042.png' -i '/tmp/vo_0000.wav' \
  -filter_complex "apad" -shortest -map 0:v:0 -map 1:a:0 \
  -vf "zoompan=z='1.0000+0.3600*(0.5-0.5*cos(on/(d-1)*PI))':\
       x='403+(0)*(0.5-0.5*cos(on/(d-1)*PI))':\
       y='334+(0)*(0.5-0.5*cos(on/(d-1)*PI))':\
       d=135:s=1920x1080,format=yuv420p" \
  -c:v libx264 -crf 18 -preset slow -pix_fmt yuv420p -r 30 \
  -c:a aac -b:a 192k -ar 48000 '/tmp/seg_0000.mp4'

Auto-effect selector:

engine = ZoompanEngine()
effect = engine.suggest_effect(
    primary_action="figure grips table edge",
    lighting_mood="chiaroscuro",
    is_manga_panel=True
)
# → Returns "push_in" (high tension + manga = zoom for impact)

"""
FFmpeg Zoompan Parameter Generator
Consumes focal_point {x, y} from visual model output → produces Ken Burns FFmpeg filters.
Supports: push_in, pull_out, pan_left, pan_right, pan_up, pan_down, hold, drift.
"""

import json
import math
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass


@dataclass
class ZoompanConfig:
    """Configuration for a single zoompan effect."""
    effect: str              # push_in | pull_out | pan_left | pan_right | pan_up | pan_down | hold | drift
    duration: float          # seconds
    fps: int = 30
    start_zoom: float = 1.0  # 1.0 = no zoom
    end_zoom: float = 1.0
    start_x: float = 0.5     # 0.0-1.0, relative to frame
    start_y: float = 0.5
    end_x: float = 0.5
    end_y: float = 0.5
    easing: str = "ease_in_out"  # linear | ease_in | ease_out | ease_in_out
    output_width: int = 1920
    output_height: int = 1080

    def total_frames(self) -> int:
        return int(self.duration * self.fps)


class ZoompanEngine:
    """
    Generates FFmpeg zoompan filter expressions from focal points.

    How zoompan works in FFmpeg:
    - zoompan filter pans/zooms on a single input image over time.
    - z = zoom level (1.0 = 100%, 2.0 = 200% = 2x closer)
    - x, y = top-left corner of the viewport (in input image pixels)
    - d = duration in frames
    - s = output size (WxH)

    The trick: focal_point (0.0-1.0) must be converted to pixel coordinates
    that keep the focal point CENTERED in the output viewport.
    """

    def __init__(self, input_width: int = 1920, input_height: int = 1080):
        self.input_w = input_width
        self.input_h = input_height

    # --------------------------------------------------------
    # CORE MATH: Focal Point → Viewport Coordinates
    # --------------------------------------------------------
    def focal_to_viewport(
        self,
        focal_x: float,      # 0.0-1.0 from visual model
        focal_y: float,      # 0.0-1.0 from visual model
        zoom_level: float    # current zoom (1.0 = no zoom)
    ) -> Tuple[int, int]:
        """
        Convert normalized focal point to top-left viewport coordinates
        that keep the focal point centered in the output frame.
        """
        viewport_w = self.input_w / zoom_level
        viewport_h = self.input_h / zoom_level

        x = (focal_x * self.input_w) - (viewport_w / 2)
        y = (focal_y * self.input_h) - (viewport_h / 2)

        # Clamp so viewport stays within image bounds
        max_x = self.input_w - viewport_w
        max_y = self.input_h - viewport_h

        x = max(0, min(x, max_x))
        y = max(0, min(y, max_y))

        return int(x), int(y)

    # --------------------------------------------------------
    # EASING FUNCTIONS (for smooth motion)
    # --------------------------------------------------------
    @staticmethod
    def ease(t: float, mode: str) -> float:
        """Easing function: t in [0,1], returns eased value in [0,1]."""
        if mode == "linear":
            return t
        elif mode == "ease_in":
            return t * t
        elif mode == "ease_out":
            return 1 - (1 - t) * (1 - t)
        elif mode == "ease_in_out":
            return 0.5 - 0.5 * math.cos(t * math.pi)  # smooth cosine ease
        else:
            return t

    # --------------------------------------------------------
    # EFFECT PRESETS
    # --------------------------------------------------------
    def build_effect(
        self,
        focal_x: float,
        focal_y: float,
        effect: str,
        duration: float,
        fps: int = 30,
        intensity: float = 1.0,   # 0.5 = subtle, 1.0 = standard, 2.0 = dramatic
        easing: str = "ease_in_out",
        output_size: Tuple[int, int] = (1920, 1080)
    ) -> ZoompanConfig:
        """
        Build a ZoompanConfig from effect name and focal point.
        Intensity scales the zoom range and pan distance.
        """
        out_w, out_h = output_size

        # Default: hold (no motion, just frame on focal point)
        cfg = ZoompanConfig(
            effect=effect, duration=duration, fps=fps,
            start_x=focal_x, start_y=focal_y,
            end_x=focal_x, end_y=focal_y,
            start_zoom=1.0, end_zoom=1.0,
            easing=easing,
            output_width=out_w, output_height=out_h
        )

        if effect == "push_in":
            cfg.start_zoom = 1.0
            cfg.end_zoom = 1.0 + (0.3 * intensity)

        elif effect == "pull_out":
            cfg.start_zoom = 1.0 + (0.3 * intensity)
            cfg.end_zoom = 1.0

        elif effect == "pan_left":
            cfg.start_zoom = 1.0 + (0.1 * intensity)
            cfg.end_zoom = 1.0 + (0.1 * intensity)
            cfg.start_x = min(focal_x + 0.15 * intensity, 0.9)
            cfg.end_x = max(focal_x - 0.15 * intensity, 0.1)
            cfg.start_y = focal_y
            cfg.end_y = focal_y

        elif effect == "pan_right":
            cfg.start_zoom = 1.0 + (0.1 * intensity)
            cfg.end_zoom = 1.0 + (0.1 * intensity)
            cfg.start_x = max(focal_x - 0.15 * intensity, 0.1)
            cfg.end_x = min(focal_x + 0.15 * intensity, 0.9)
            cfg.start_y = focal_y
            cfg.end_y = focal_y

        elif effect == "pan_up":
            cfg.start_zoom = 1.0 + (0.1 * intensity)
            cfg.end_zoom = 1.0 + (0.1 * intensity)
            cfg.start_x = focal_x
            cfg.end_x = focal_x
            cfg.start_y = min(focal_y + 0.15 * intensity, 0.9)
            cfg.end_y = max(focal_y - 0.15 * intensity, 0.1)

        elif effect == "pan_down":
            cfg.start_zoom = 1.0 + (0.1 * intensity)
            cfg.end_zoom = 1.0 + (0.1 * intensity)
            cfg.start_x = focal_x
            cfg.end_x = focal_x
            cfg.start_y = max(focal_y - 0.15 * intensity, 0.1)
            cfg.end_y = min(focal_y + 0.15 * intensity, 0.9)

        elif effect == "drift":
            cfg.start_zoom = 1.0 + (0.05 * intensity)
            cfg.end_zoom = 1.0 + (0.15 * intensity)
            cfg.start_x = max(focal_x - 0.1 * intensity, 0.1)
            cfg.start_y = max(focal_y - 0.05 * intensity, 0.1)
            cfg.end_x = min(focal_x + 0.1 * intensity, 0.9)
            cfg.end_y = min(focal_y + 0.05 * intensity, 0.9)

        elif effect == "hold":
            pass  # no motion

        return cfg

    # --------------------------------------------------------
    # GENERATE FFmpeg zoompan FILTER EXPRESSION
    # --------------------------------------------------------
    def generate_zoompan_expression(self, cfg: ZoompanConfig) -> str:
        """Generate the FFmpeg zoompan filter string."""
        total_frames = cfg.total_frames()

        start_vx, start_vy = self.focal_to_viewport(cfg.start_x, cfg.start_y, cfg.start_zoom)
        end_vx, end_vy = self.focal_to_viewport(cfg.end_x, cfg.end_y, cfg.end_zoom)

        # Zoom expression
        if abs(cfg.end_zoom - cfg.start_zoom) < 0.001:
            zoom_expr = f"{cfg.start_zoom:.4f}"
        else:
            zoom_expr = (
                f"{cfg.start_zoom:.4f}+"
                f"({cfg.end_zoom - cfg.start_zoom:.4f})*"
                f"({self._easing_expr(cfg.easing)})"
            )

        # X expression
        if abs(end_vx - start_vx) < 1:
            x_expr = f"{start_vx}"
        else:
            x_expr = (
                f"{start_vx}+"
                f"({end_vx - start_vx})*"
                f"({self._easing_expr(cfg.easing)})"
            )

        # Y expression
        if abs(end_vy - start_vy) < 1:
            y_expr = f"{start_vy}"
        else:
            y_expr = (
                f"{start_vy}+"
                f"({end_vy - start_vy})*"
                f"({self._easing_expr(cfg.easing)})"
            )

        filter_str = (
            f"zoompan=z='{zoom_expr}':"
            f"x='{x_expr}':"
            f"y='{y_expr}':"
            f"d={total_frames}:"
            f"s={cfg.output_width}x{cfg.output_height}"
        )

        return filter_str

    def _easing_expr(self, easing: str) -> str:
        """Convert easing name to FFmpeg expression using on/d."""
        t = "on/(d-1)"
        if easing == "linear":
            return t
        elif easing == "ease_in":
            return f"{t}*{t}"
        elif easing == "ease_out":
            return f"1-(1-{t})*(1-{t})"
        elif easing == "ease_in_out":
            return f"0.5-0.5*cos({t}*PI)"
        return t

    # --------------------------------------------------------
    # FULL RENDER COMMAND BUILDER
    # --------------------------------------------------------
    def build_render_command(
        self,
        image_path: str,
        cfg: ZoompanConfig,
        output_path: str,
        audio_path: Optional[str] = None,
        pix_fmt: str = "yuv420p",
        codec: str = "libx264",
        crf: int = 18,
        preset: str = "slow"
    ) -> str:
        """
        Build a complete FFmpeg command for rendering a single image
        with zoompan into a video clip.
        """
        zoompan_filter = self.generate_zoompan_expression(cfg)

        cmd_parts = [
            "ffmpeg -y",
            "-loop 1",
            f"-framerate {cfg.fps}",     # ponytail: explicit framerate prevents VFR trap
            f"-i '{image_path}'",
        ]

        if audio_path:
            cmd_parts.append(f"-i '{audio_path}'")
            # ponytail: apad + -shortest prevents A/V drift on image-loop + audio mux
            cmd_parts.append("-filter_complex apad")
            cmd_parts.append("-shortest")
            cmd_parts.append("-map 0:v:0 -map 1:a:0")
        else:
            cmd_parts.append(f"-t {cfg.duration}")

        cmd_parts.extend([
            f"-vf '{zoompan_filter},format={pix_fmt}'",
            f"-c:v {codec}",
            f"-crf {crf}",
            f"-preset {preset}",
            f"-pix_fmt {pix_fmt}",
            f"-r {cfg.fps}",
        ])

        if audio_path:
            cmd_parts.append("-c:a aac -b:a 192k -ar 48000")

        cmd_parts.append(f"'{output_path}'")

        return " \\
  ".join(cmd_parts)

    # --------------------------------------------------------
    # BATCH: Full Timeline → FFmpeg concat script
    # --------------------------------------------------------
    def build_timeline_render(
        self,
        timeline: List[Dict[str, Any]],
        output_path: str,
        temp_dir: str = "/tmp/render_segments",
        fps: int = 30,
        output_size: Tuple[int, int] = (1920, 1080)
    ) -> str:
        """
        Build a full video from a timeline of image clips.
        Returns: bash script string that renders all segments then concatenates.
        """
        lines = [
            "#!/bin/bash",
            "set -euo pipefail",
            f"mkdir -p {temp_dir}",
            ""
        ]

        concat_list = []

        for i, item in enumerate(timeline):
            fp = item.get("focal_point", {"x": 0.5, "y": 0.5})
            cfg = self.build_effect(
                focal_x=fp["x"],
                focal_y=fp["y"],
                effect=item.get("effect", "hold"),
                duration=item["duration"],
                fps=fps,
                intensity=item.get("intensity", 1.0),
                easing=item.get("easing", "ease_in_out"),
                output_size=output_size
            )

            seg_path = f"{temp_dir}/seg_{i:04d}.mp4"
            concat_list.append(seg_path)

            cmd = self.build_render_command(
                image_path=item["image_path"],
                cfg=cfg,
                output_path=seg_path,
                audio_path=item.get("audio_path")
            )
            lines.append(f'echo "[{i+1}/{len(timeline)}] Rendering segment {i}..."')
            lines.append(cmd)
            lines.append("")

        # Build concat list file
        concat_file = f"{temp_dir}/concat_list.txt"
        lines.append(f"cat > {concat_file} << 'EOF'")
        for seg in concat_list:
            lines.append(f"file '{seg}'")
        lines.append("EOF")
        lines.append("")

        # Final concat command
        lines.append(f'echo "Concatenating {len(timeline)} segments..."')
        lines.append(
            f"ffmpeg -y -f concat -safe 0 -i {concat_file} "
            f"-c copy '{output_path}'"
        )
        lines.append("")
        lines.append(f'echo "Done: {output_path}"')

        return "\n".join(lines)

    # --------------------------------------------------------
    # AUTO-EFFECT SELECTOR (based on content type)
    # --------------------------------------------------------
    @staticmethod
    def suggest_effect(
        primary_action: str,
        lighting_mood: str,
        has_text_overlay: bool = False,
        is_manga_panel: bool = False
    ) -> str:
        """
        Suggest a zoompan effect based on image content metadata.
        This connects the vision preprocessor output to motion choice.
        """
        action_lower = primary_action.lower() if primary_action else ""

        # High tension / action → push_in for intensity
        if any(w in action_lower for w in ["grip", "clench", "strike", "lunge", "grasp"]):
            return "push_in"

        # Reveal / discovery → pull_out
        if any(w in action_lower for w in ["reveal", "open", "uncover", "turn", "look up"]):
            return "pull_out"

        # Movement verbs → pan in direction of action
        # ponytail: check direction keywords first, then generic "walk" falls through to drift
        if "move left" in action_lower or "walk left" in action_lower:
            return "pan_left"
        if "move right" in action_lower or "walk right" in action_lower:
            return "pan_right"

        # Text-heavy / infographic → hold for readability
        if has_text_overlay:
            return "hold"

        # Manga speed lines → push_in for impact
        if is_manga_panel:
            return "push_in"

        # Low-key / chiaroscuro → slow drift for atmosphere
        if lighting_mood in ["low-key", "chiaroscuro"]:
            return "drift"

        # Default
        return "drift"


# ============================================================
# USAGE EXAMPLES
# ============================================================

if __name__ == "__main__":
    engine = ZoompanEngine(input_width=1920, input_height=1080)

    # Example 1: Single clip from Prompt Box 1 output
    focal = {"x": 0.35, "y": 0.62}  # from visual model
    cfg = engine.build_effect(
        focal_x=focal["x"],
        focal_y=focal["y"],
        effect="push_in",
        duration=4.5,
        intensity=1.2,
        easing="ease_in_out"
    )

    filter_expr = engine.generate_zoompan_expression(cfg)
    print("=== Single Filter Expression ===")
    print(filter_expr)
    print()

    cmd = engine.build_render_command(
        image_path="/assets/ep07/frame_042.png",
        cfg=cfg,
        output_path="/tmp/seg_0000.mp4",
        audio_path="/tmp/vo_0000.wav"
    )
    print("=== Single Render Command ===")
    print(cmd)
    print()

    # Example 2: Auto-suggest effect from content
    suggested = engine.suggest_effect(
        primary_action="figure grips table edge",
        lighting_mood="chiaroscuro",
        is_manga_panel=True
    )
    print(f"=== Auto-suggested effect: {suggested} ===")
    print()

    # Example 3: Full timeline render script
    timeline = [
        {
            "image_path": "/assets/ep07/frame_042.png",
            "focal_point": {"x": 0.35, "y": 0.62},
            "effect": "push_in",
            "duration": 4.5,
            "intensity": 1.2,
            "audio_path": "/tmp/vo_0000.wav"
        },
        {
            "image_path": "/assets/ep07/frame_043.png",
            "focal_point": {"x": 0.5, "y": 0.4},
            "effect": "drift",
            "duration": 3.0,
            "intensity": 0.8
        },
        {
            "image_path": "/assets/ep07/frame_044.png",
            "focal_point": {"x": 0.2, "y": 0.7},
            "effect": "hold",
            "duration": 2.5,
            "intensity": 1.0
        }
    ]

    bash_script = engine.build_timeline_render(
        timeline=timeline,
        output_path="/tmp/ep07_final.mp4",
        temp_dir="/tmp/render_segments"
    )
    print("=== Full Timeline Bash Script ===")
    print(bash_script)

The zoompan class, effect presets, render commands, and usage examples live in the code box above (full File 2 source, uncut).

Key Differences

Do You Need Automation Command Tools?

No. Vision models do not require external automation command tools like Zapier (the likely intended meaning of "zoopman") or similar no-code connectors. Images are passed directly inside the multimodal prompt using the platform's native image API, public URL, or base64 data URL.

Non-vision models do not need them either. They consume the structured output produced by a vision preprocessor or embedding store. The only "commands" are ordinary API calls — for example, POST /chat/completions with a JSON payload of text and metadata, or local SQLite queries.

Use an automation connector only if you are wiring the pipeline into a larger no-code workflow. Even then, the tool is orchestrating API calls and file moves; it is not generating vision understanding itself.

When to Use Each

• Use a vision model when you need per-image quality control, face-rule checks, text-in-image transcription, or focal-point selection.
• Use a non-vision model when you have already preprocessed images into structured metadata and need fast, cheap, large-scale matching and timeline orchestration.
• Use both together for cost-effective scale: a small local VLM as preprocessor (Prompt Box 4) + a text-only LLM as orchestrator (Prompt Box 3).