Image & Video Generation interview questions & answers

Picture a sculptor who starts with a rough block of marble that is really just noise -- random bumps and ridges -- and chips away, over and over, until a shape emerges. Nobody hands the sculptor a finished statue; they hand them a description ("a seated woman, one arm resting on her knee") and the sculptor removes a little bit of "not-that" at each pass. Image and video generation models work the same way: they do not paint a picture stroke by stroke, they start from pure noise and repeatedly remove a little of it, each time nudged by your text prompt toward "more like that description." This is diffusion, and it is a behaviour, not one specific architecture -- several different model families implement it, and the behaviour is what carries over between them.

In production, this behaviour is the reason your app has to plan around three things every one of these models does, whichever vendor you call: it never repeats itself exactly, it is confidently wrong about small precise details, and it has no idea whether the result is "correct" -- there is no ground truth inside it to check against.

🌍 Real-world example: a resume-builder feature lets a user generate a profile banner from a short prompt ("minimal blue gradient banner, abstract geometric shapes, no text"). Generate it three times and you get three different banners -- all reasonably close to the prompt, none identical, because the process starts from a fresh block of random noise each time. Ask it to write the user's name inside the banner and it will very likely produce distorted, unreadable, near-text squiggles instead of real letters. Neither of those is a bug in your integration; both are the model behaving exactly as diffusion behaves. The fix is to design around it -- offer a regenerate button for the first, and composite real text on top with your own code for the second -- not to keep re-prompting and hoping.

💡 Diffusion = a generation process that starts from random noise and removes a little of it at each step, steered toward your text prompt, until what remains looks like an image (or a sequence of images, for video). "Diffusion model" describes this noise-to-structure behaviour, not one company's product -- several different model families are built this way.

💡 Latent space = instead of denoising a full-resolution grid of pixels directly (slow, and most detail is redundant), most of these models denoise a much smaller, compressed representation of the image -- its latent -- and only expand that back into real pixels as the last step. You never touch this space directly through an API; it is why generation is faster than denoising raw pixels would be, and it is the one sentence you need to recognise the term in an interview.

Why the same prompt gives a different image every run: the starting point is random noise, generated fresh from a random seed each time you call the API (unless you deliberately pin one -- a separate topic on its own). Different starting noise, denoised toward the same text, lands on a different point that still matches the description. This is expected behaviour, not inconsistency to debug.

Where these models reliably break, and why: the model has learned statistical patterns across huge numbers of images -- what a hand usually looks like, what letters usually look like near each other, what "three" of something usually looks like -- not the exact rules a human would apply (five fingers, a specific font, a precise count). So it is weakest exactly where correctness is exact and unforgiving:

• Hands and fingers -- an ordinary object with an extremely specific, easy-to-get-wrong structure; the model has no built-in constraint forcing exactly five.
• Small text inside the image -- letters are visual patterns to a diffusion model, not characters from an alphabet, so it produces letter-shaped noise instead of spelling. This is the same reason a "no text" negative prompt often still leaves faint lettering behind (covered in the next topic).
• Exact logos and brand marks -- the model reproduces the style of a logo it has seen, not a byte-perfect copy, so it is close enough to look wrong to anyone who knows the real one.
• Counting objects -- "exactly four apples" is a precise constraint; the model has a strong sense of what a plausible pile of apples looks like, not a counter.

Designing around this, not re-prompting forever, is the actual skill: put exact text and logos on top with your own code after generation, treat a generated image as a draft or a background rather than a source of precise facts, and give the user a regenerate button instead of promising a specific outcome on the first try.

You call these models, you do not train them. This chapter is about the ~1% of the job that touches an API: sending a prompt, handling the response, storing what comes back, designing for refusal and cost. Training or fine-tuning a diffusion model from scratch is a distinct, research-heavy specialisation with its own hardware and data requirements, and it is out of scope here -- exactly the way Ch8 taught fine-tuning an LLM without teaching you to pretrain one.

The chapter's opening frame, and why it matters more here than anywhere else in this course: every chapter before this one produced text -- small, cheap, instant, and gone the moment it is displayed on screen. This chapter produces bytes: an image or video file that is large, slow to make, priced per unit, legally loaded (who may use it, and for what), and that keeps existing on a disk somewhere long after the request that made it has finished. A generate call is not the end of the work, it is closer to the beginning -- from the moment the bytes exist, you own storing them, resizing them, serving them, and answering for what they contain. That single shift -- response versus artefact -- is what the rest of this chapter is built around.

When to use it: reach for a generation model when you need an original image or video that does not exist yet and exact correctness of fine detail is not the requirement -- a mood board, a marketing banner, a background, concept art, a placeholder while a designer works on the real asset.

When NOT to use it: do not reach for it when you need something that must be pixel-exact or legally clean -- a company logo, a specific person's photo, a screenshot of a real product, or any image where a small factual detail (a count, an exact label, a real brand mark) has to be right. A stock photo or asset library wins when "a photo like this" already exists and licensing terms are clear and known upfront. A human designer wins when the brand mark, the exact layout, or legal sign-off actually matters and a wrong pixel is a real cost, not a redo-and-move-on annoyance.

Standard definition: A diffusion model generates media by starting from random noise and iteratively removing it over a series of steps, guided at each step by a text (or image) conditioning signal, until the result resembles the description; most implementations perform this denoising in a compressed latent space rather than on raw pixels for speed, then decode the result back to pixels at the end. Because the starting noise is random and the model has learned statistical patterns rather than exact rules, output varies between runs on the same prompt and is least reliable on precise fine detail such as hands, embedded text, exact logos, and object counts.

Chatting with a text model is a conversation -- you can say "actually, make it shorter" and it understands the whole context. Prompting an image model is closer to filling out a form with fixed fields: subject, style, detail, lighting, composition. There is no back-and-forth memory of what you asked five minutes ago (unless the tool explicitly re-sends your last prompt as context) -- every generation call is a fresh form, and the words you put in each field are the only thing steering the output.

In production, that maps to: build the prompt as data first -- a subject, style, detail, lighting, composition -- and only join it into one string at the moment you call the API. Treat the fields the same way you would treat a form's inputs: change one, keep the rest identical, and you can now tell which field caused which change in the output. Change three fields at once and get a different image, and you have learned nothing -- you cannot tell whether the new lighting words, the new style words, or their interaction did it.

Negative prompts are the second half of the field set on models that support them: a second string listing what you do NOT want -- "blurry, low quality, distorted, watermark, extra fingers, text". Two things about them get taught wrong constantly:

1. A negative prompt is not a delete button. It does not remove a concept from the output the way sed deletes a line. It shifts the model's guidance AWAY from that concept during generation -- a steer, not a subtraction. A strong positive signal for something (say, a prompt that asks for a wooden signboard) can still out-vote a weak negative signal against a related thing ("text") because signboards, in the model's training data, usually carry text. That is exactly why "no text" in the negative prompt often still produces text -- the positive prompt is implicitly asking for it through an associated object, and the negative field is only a nudge, not a veto.
2. Not every model has a negative-prompt parameter at all. Some accept it as a first-class field; others only accept a single prompt string and expect you to phrase what you want positively instead. This is a per-model capability you look up, the same way you look up size or quality knobs -- never assume it exists.

Style words are not a separate capability from prompting -- they are one field in the same form. "Reimagine this in Van Gogh oil painting style" and "reimagine this in pixel art style" are not two different API features; they are the same style slot filled with a different value. Treating style transfer as its own thing is why people reach for a separate tool when changing one word in the prompt they already have would do it.

The runnable half below builds and prints these prompt strings in Python -- no API key needed, because it is pure string composition; nothing about it depends on which vendor you eventually call. The generation call itself is a separate, illustrative snippet, because it needs a live API key this environment does not have.

🌍 Real-world example: a team debugging "our anime-style outputs look inconsistent" discovers they were changing subject wording, style wording AND the negative prompt between every test run -- so no single generation was comparable to the last one. Locking four of five fields and varying only style turned three days of confused screenshots into a one-page comparison.

💡 Prompt structure = treating subject / style / detail / lighting / composition as separate fields you compose, not one sentence you rewrite from scratch each time.

💡 Negative prompt = a second field (where supported) listing concepts to steer away from during generation -- a bias, not a removal instruction.

Standard definition: Image prompting is field-based, not conversational: a prompt is composed from subject, style, detail, lighting and composition, and the discipline that makes iteration legible is changing exactly one field per generation while holding the rest fixed; a negative prompt, where the model supports one, steers generation away from listed concepts but does not guarantee their absence, which is why a strong positive association (a sign implies text) can still produce the very thing the negative prompt named.

When to use it: iterate on your OWN prompt fields first -- change one slot, look at the result, change the next -- whenever the current image is close but one aspect (lighting, style, framing) is wrong; this is the cheapest fix and it is also the only way to build a mental map of what each field actually controls.

When NOT to use it / Trade-off: if ten iterations of wording have not fixed a specific, localised problem (a face, a logo, a hand), stop prompt-iterating and switch tools -- either edit the image afterwards with masked inpainting (a later topic in this chapter) for a local fix, composite text/logos yourself with an image library rather than asking the model to render exact characters (the model was never trained to spell reliably), or change the model entirely if the failure is structural (that model consistently cannot do the pose/style you need). Prompt wording cannot fix a capability the model does not have.

def build_prompt(subject, style, detail, lighting, composition):
    parts = [subject, style, detail, lighting, composition]
    return ", ".join(p for p in parts if p)


base = dict(
    subject="a samurai warrior standing in a bamboo forest",
    style="anime style",
    detail="detailed armor, intricate patterns",
    lighting="golden hour lighting",
    composition="low angle shot, wide composition",
)

print("== base prompt ==")
print(build_prompt(**base))

print()
print("== change ONE slot, keep the rest fixed ==")
variants = {
    "style":       {**base, "style": "watercolor illustration style"},
    "lighting":    {**base, "lighting": "dramatic rim lighting at dusk"},
    "composition": {**base, "composition": "close-up shot, centered composition"},
}
for slot, v in variants.items():
    changed = v[slot]
    print(f"[changed only: {slot} -> {changed!r}]")
    print(build_prompt(**v))
    print()

print("== negative prompt: steers away from, does not delete ==")
negative_prompt = "blurry, low quality, distorted, watermark, extra fingers, text"
print(f"positive: {build_prompt(**base)}")
print(f"negative: {negative_prompt}")
print("note: some models have no negative-prompt parameter at all -- check per model.")

print()
print("== why 'no text' in the negative prompt often still produces text ==")
attempt = {**base, "detail": "a wooden signboard behind him"}
print(f"positive: {build_prompt(**attempt)}")
print(f"negative: {negative_prompt}")
print("the positive prompt asks for a signboard (things with signboards usually carry text)")
print("the negative prompt only lowers the WEIGHT of 'text' -- it is not a delete instruction,")
print("so a strong positive signal for a sign can still out-vote a weak negative signal against text.")

What we're building: every image and video topic in this chapter taught one piece of the pipeline in isolation -- saving bytes, resizing, building a slideshow, storing under a key, polling a job. This project wires all of it into the one thing a user actually clicks: Generate. A prompt goes in; twenty seconds of engineering happen; a real image, a real video and a real gallery entry come out the other side. This project is FREE, and it assumes you have read nothing else in this chapter -- every step below re-derives its own idea from the ground up.

\U0001f30d Real-world example: a resume-builder's "generate a cover-letter illustration" button looks, to a user, like one click. Behind it: a provider call that might refuse, bytes that must be verified before they are trusted, a resize so the gallery card does not ship a megabyte to a phone, a metadata row so a support ticket six months later can be answered, and -- if the feature also promises a short intro video -- a job that cannot finish before the HTTP response does. The generate call is the smallest piece of that sentence. Everything else is the engineering this project builds.

\U0001f4a1 Pipeline = the ordered sequence a request actually goes through -- generate, verify, store, derive, list, poll -- as opposed to "call the API," which is only the first link in it.

Standard definition: An AI content-generation platform is not one API call, it is a pipeline: a generation call that can refuse or return bytes in different shapes, a download-and-verify step before anything is trusted, object storage addressed by a key convention with a metadata row beside it, purpose-built derivatives so nothing serves the master file directly, a way to build video content with no video-generation model at all when a slideshow will do, a gallery that reads the metadata store rather than touching storage bytes, and a submit/poll job shape for whichever half of the pipeline cannot finish inside one HTTP request.

⚠ One plain sentence on what ran and what did not: the image-generation and video-generation provider calls are illustrative -- no key exists on this machine, matching every other coding topic in this chapter. Every other line below -- downloading and verifying bytes, building derivatives, encoding the slideshow with ffmpeg, writing and reading the metadata store, listing the gallery, and the job-status endpoint's submit/poll/idempotency logic -- ran for real, twice, and every byte count came out identical across both runs.

Step 1 -- The architecture: five endpoints, one async branch

The brief behind a feature like this is always some version of: generate an image from a prompt, optionally turn it into a short video, and let the user browse everything they made. Stripped of any one vendor's product names, that maps to five endpoints -- and the shape of each one is what this project actually teaches, not which company built it:

POST /generate/image     text -> image. Slow enough to make the user wait, fast enough to answer inline.
POST /generate/video     image -> video. Too slow for one HTTP request -- returns a job id, not a video.
GET  /task/{id}          the slow half's poll target: what is /generate/video's job doing right now?
GET  /gallery            everything this user has generated, read from the metadata store.
DELETE /content/{id}     remove one asset -- delete the storage key AND its metadata row, never one alone.

Look at the shape, not the list: exactly one of these five is asynchronous, and it is asynchronous for a structural reason, not a stylistic one -- image generation is seconds, video generation is minutes, and no HTTP client holds a connection open for minutes. Everything downstream of that one fact -- the job id, the poll target, the idempotency key on the submit -- exists because of this single asymmetry. DELETE /content/{id} is not built below (it is one os.remove plus one metadata-row filter, nothing this chapter has not already covered), but it belongs on the list for the same reason the checklist matters more than the code: forgetting it means a user's "delete" leaves an orphaned blob paying for storage forever.

Step 2 -- Prompt to image: illustrative call, REAL download-and-store

def illustrative_generate_call(prompt, seed):
    # ---- ILLUSTRATIVE (requires an API key; not executed) ----
    response = image_client.generate(prompt=prompt, model="<current-image-model-id>", seed=seed)
    raw_bytes = response.image_bytes  # or download(response.url) -- provider-shaped
    raise NotImplementedError("illustrative only -- no provider key on this machine")

def download_and_store(raw_bytes, user_id, prompt, seed, model_id):
    if not raw_bytes:
        raise ValueError("received 0 bytes -- refusing to store")
    img = Image.open(io.BytesIO(raw_bytes))
    img.load()          # forces a full decode now, not a lazy failure later
    ...

Why the function is split exactly here, and not somewhere more convenient: illustrative_generate_call is the ONLY part of this entire project that talks to a provider, and it is the only function this project never runs. Everything below the raw_bytes line is real, because it is real regardless of which provider produced those bytes, whether they arrived as a URL to download or base64 to decode, and whether the call cost anything at all. Splitting the function here is the same lesson calling-an-image-api and media-storage-and-urls taught separately, now enforced by the code's own shape: the SHAPE of what you own does not change based on which vendor you called.

Verified real run: the download-and-store half was fed the actual bytes of a real JPEG standing in for a generated image, and produced stored image: key=user/u_7c19/325a2a1b-...jpg bytes=130786 dims=1280x639 mime=image/jpeg -- a real decode, a real file on disk under the user/<id>/<asset-id>.jpg key convention this chapter's storage topic established, and a real metadata row written beside it (prompt, seed, model, mime, dimensions, byte count -- never the bytes themselves in that row).

Step 3 -- Derivatives with Pillow: the master is never served

web_img = img.convert("RGB").resize((w // 2, h // 2), Image.LANCZOS)
web_img.save(web_path, "JPEG", quality=85)          # dimension knob first
thumb_img = img.convert("RGB").copy()
thumb_img.thumbnail((320, 320), Image.LANCZOS)
thumb_img.save(thumb_path, "JPEG", quality=85)

Why two derivatives and not a "resize on request" endpoint: building both derivatives once, at generation time, and caching them beside the master is the entire lesson sizes-formats-and-delivery measured -- the master is the archival copy, and the pixel dimensions are the strong knob, so both derivatives resize first and set quality second rather than the other way round. Verified real run: the 130,786-byte master produced a 50,214-byte web derivative (640x319, 2.6x smaller) and a 12,873-byte thumb (320x160, 10.16x smaller) -- neither number is invented, both came from encoding the identical source pixels this run actually decoded.

Step 4 -- A slideshow video with ffmpeg: no video-generation call at all

lines.append(f"file '{p}'"); lines.append("duration 1.5")   # per slide
lines.append(f"file '{slide_paths[-1]}'")                    # repeat -- the quirk this chapter measured
cmd = [FFMPEG, "-f", "concat", "-safe", "0", "-i", list_path,
       "-f", "lavfi", "-i", "sine=frequency=330:duration=4.5",
       "-c:v", "libx264", "-c:a", "aac", "-shortest", video_path]

Why this project's "video generation" is ffmpeg, not a provider call: the brief only needs three static slides -- a storyboard, not real motion -- and video-without-a-video-model established exactly when that substitution is correct: genuinely static content, priced at zero generation cost, fully deterministic. The concat list repeats the final slide's filename with no trailing duration line, because the demuxer times each entry against the file that FOLLOWS it -- skip the repeat and the last slide's requested length is silently dropped, with no error and no warning. Verified real run: return_code=0 bytes=44922 build_seconds=0.366 duration=00:00:04.50 resolution=1280x720 audio_present=True -- a real MP4 this project can open, built from three Pillow-drawn captioned frames and a lavfi sine tone, with zero video-generation API calls anywhere in the chain.

Step 5 -- Store bytes + metadata, then a gallery that never touches storage

def write_metadata_row(row):
    with open(METADATA_FILE, "a", encoding="utf-8") as f:
        f.write(json.dumps(row) + "\n")

def list_gallery(user_id):
    rows = [r for r in read_metadata_rows() if r["user_id"] == user_id]
    rows.sort(key=lambda r: r["created_at"])
    return [{"asset_id": r["asset_id"], "kind": r["kind"], "storage_key": r["storage_key"],
              "bytes": r["bytes"], "created_at": r["created_at"]} for r in rows]

A JSON-lines file stands in for the database table on this machine, and object_storage_standin/ stands in for the bucket -- the split is real regardless of which two products sit behind those two roles. Why list_gallery never opens a file under object_storage_standin/: a gallery endpoint's job is to answer "what does this user have," not "here are the bytes" -- exactly the row-holds-the-key-not-the-bytes lesson media-storage-and-urls measured, now called from three lines of route logic instead of stated as a rule. Verified real run: after the image and the video were each written as one metadata row, list_gallery("u_7c19") returned exactly 2 items, sorted by created_at, each carrying its storage_key and byte count and nothing resembling a signed URL -- that URL gets minted only when a browser actually asks to view one asset, never stored in the row itself.

Step 6 -- A job-status endpoint for the slow half

def generate_video_endpoint(user_id, image_asset_id, idempotency_key=None):
    if idempotency_key and idempotency_key in JOBS.get("_idempotency", {}):
        return {"job_id": JOBS["_idempotency"][idempotency_key], "status": "accepted"}
    job_id = str(uuid.uuid4())
    JOBS[job_id] = {"status": "queued", ...}
    ...
    return {"job_id": job_id, "status": "accepted"}

def task_status_endpoint(job_id):
    job = JOBS.get(job_id)
    if job is None:
        return {"status": "not_found"}, 404
    return {"job_id": job_id, "status": job["status"]}, 200

This is the shape async-jobs-and-polling taught in isolation, called here from POST /generate/video's and GET /task/{id}'s actual position in this project's architecture. Why the idempotency key matters more here than almost anywhere else in this course: video is the single most expensive artefact this chapter names, priced by duration -- a client that times out waiting for /generate/video and retries the click must NOT start a second, separately-billed job. Verified real run: the endpoint went queued -> succeeded across two real calls to task_status_endpoint, and calling generate_video_endpoint twice with the identical idempotency key "req-8841" returned the same job_id both times -- one job was ever created, even though the submit function itself was called twice.

What this project genuinely proves, and what it does not

Proves: every piece downstream of the generation call was built and run for real, twice, with byte-identical results both times -- download-and-verify, two Pillow derivatives at measured size ratios, an ffmpeg slideshow with a real duration and a real audio stream, a metadata store a gallery endpoint can read without touching storage, and a job-status endpoint whose idempotency key genuinely prevented a second job on a retried submit.

Does not (say so before an interviewer finds it): the image- and video-generation provider calls are illustrative -- no key exists on this machine, and swapping a real provider in changes nothing about the pipeline built around it, because that pipeline was written to the SHAPE (bytes in, job id out) rather than to one vendor's field names. There is no real database, no real object-storage SDK and no real HTTP server here -- a JSON-lines file and a local directory tree stand in for them, and generate_video_endpoint/task_status_endpoint are plain functions with the exact bodies a FastAPI route would have, called directly rather than over a network. And DELETE /content/{id} is named in the architecture but not built -- it is a smaller problem than everything above it, not a harder one.

import io
import json
import os
import subprocess
import time
import uuid
from datetime import datetime, timezone

import imageio_ffmpeg
from PIL import Image, ImageDraw

HERE = os.path.dirname(os.path.abspath(__file__))
STORAGE_ROOT = os.path.join(HERE, "object_storage_standin")
FFMPEG = imageio_ffmpeg.get_ffmpeg_exe()
USER_ID = "u_7c19"


# ---- Step 2: prompt -> image. Generate call is ILLUSTRATIVE (no key here). ----
def illustrative_generate_call(prompt, seed):
    """
    # ---- ILLUSTRATIVE (requires an API key; not executed) ----
    response = image_client.generate(
        prompt=prompt,
        model="<current-image-model-id>",  # looked up at call time, never hardcoded
        seed=seed,
    )
    raw_bytes = response.image_bytes  # or download(response.url) -- provider-shaped
    """
    raise NotImplementedError("illustrative only -- no provider key on this machine")


def download_and_store(raw_bytes, user_id, prompt, seed, model_id):
    """REAL: the half every provider response hits regardless of its shape."""
    if not raw_bytes:
        raise ValueError("received 0 bytes -- refusing to store")
    img = Image.open(io.BytesIO(raw_bytes))
    img.load()  # force full decode now, not lazily later
    width, height = img.size
    mime = Image.MIME.get(img.format)

    asset_id = str(uuid.uuid4())
    key = f"user/{user_id}/{asset_id}.jpg"
    dest_path = os.path.join(STORAGE_ROOT, *key.split("/"))
    os.makedirs(os.path.dirname(dest_path), exist_ok=True)
    with open(dest_path, "wb") as f:
        f.write(raw_bytes)
    stored_size = os.path.getsize(dest_path)

    metadata = {
        "asset_id": asset_id, "user_id": user_id, "kind": "image", "storage_key": key,
        "prompt": prompt, "seed": seed, "model": model_id, "mime": mime,
        "width": width, "height": height, "bytes": stored_size,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    write_metadata_row(metadata)
    return metadata


# ---- Step 5: the metadata store. JSON-lines file stands in for the DB table. ----
METADATA_FILE = os.path.join(HERE, "asset_metadata.jsonl")

def write_metadata_row(row):
    with open(METADATA_FILE, "a", encoding="utf-8") as f:
        f.write(json.dumps(row) + "\n")

def read_metadata_rows():
    if not os.path.exists(METADATA_FILE):
        return []
    with open(METADATA_FILE, encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]


# ---- Step 3: derivatives with Pillow. Master is never served directly. ----
def make_derivatives(source_path, user_id, asset_id):
    img = Image.open(source_path)
    img.load()
    w, h = img.size
    derivatives = {}

    web_img = img.convert("RGB").resize((w // 2, h // 2), Image.LANCZOS)
    web_key = f"user/{user_id}/{asset_id}_web.jpg"
    web_path = os.path.join(STORAGE_ROOT, *web_key.split("/"))
    os.makedirs(os.path.dirname(web_path), exist_ok=True)
    web_img.save(web_path, "JPEG", quality=85)
    derivatives["web"] = {"key": web_key, "bytes": os.path.getsize(web_path),
                           "width": web_img.width, "height": web_img.height}

    thumb_img = img.convert("RGB").copy()
    thumb_img.thumbnail((320, 320), Image.LANCZOS)
    thumb_key = f"user/{user_id}/{asset_id}_thumb.jpg"
    thumb_path = os.path.join(STORAGE_ROOT, *thumb_key.split("/"))
    thumb_img.save(thumb_path, "JPEG", quality=85)
    derivatives["thumb"] = {"key": thumb_key, "bytes": os.path.getsize(thumb_path),
                             "width": thumb_img.width, "height": thumb_img.height}

    return os.path.getsize(source_path), derivatives


# ---- Step 4: slideshow video with ffmpeg -- no video-generation call at all. ----
def build_slideshow(user_id, asset_id):
    colors = [(35, 55, 95), (95, 35, 55), (35, 90, 60)]
    captions = ["Step 1: Prompt", "Step 2: Generate", "Step 3: Publish"]
    slide_paths = []
    for i, (c, cap) in enumerate(zip(colors, captions)):
        img = Image.new("RGB", (1280, 720), c)
        d = ImageDraw.Draw(img)
        d.rectangle([40, 40, 1240, 680], outline=(255, 255, 255), width=6)
        d.text((480, 340), cap, fill=(255, 255, 255))
        p = os.path.join(HERE, f"cap_slide{i}.png")
        img.save(p)
        slide_paths.append(p)

    list_path = os.path.join(HERE, "cap_slides.txt")
    lines = []
    for p in slide_paths:
        lines.append(f"file '{p}'")
        lines.append("duration 1.5")
    lines.append(f"file '{slide_paths[-1]}'")  # repeat -- last-file duration quirk
    with open(list_path, "w") as f:
        f.write("\n".join(lines) + "\n")

    video_key = f"user/{user_id}/{asset_id}.mp4"
    video_path = os.path.join(STORAGE_ROOT, *video_key.split("/"))
    os.makedirs(os.path.dirname(video_path), exist_ok=True)

    t0 = time.perf_counter()
    cmd = [FFMPEG, "-y", "-f", "concat", "-safe", "0", "-i", list_path,
           "-f", "lavfi", "-i", "sine=frequency=330:duration=4.5",
           "-vf", "scale=1280:720,format=yuv420p",
           "-c:v", "libx264", "-c:a", "aac", "-shortest", video_path]
    r = subprocess.run(cmd, capture_output=True, text=True)
    build_seconds = time.perf_counter() - t0

    probe = subprocess.run([FFMPEG, "-i", video_path], capture_output=True, text=True)
    import re
    dur = re.search(r"Duration:\s*([\d:.]+)", probe.stderr)
    res = re.search(r"Video:.*?(\d{2,5}x\d{2,5})", probe.stderr)
    return {"video_key": video_key, "return_code": r.returncode,
            "bytes": os.path.getsize(video_path), "build_seconds": round(build_seconds, 3),
            "duration": dur.group(1) if dur else None, "resolution": res.group(1) if res else None,
            "audio_present": "Audio:" in probe.stderr}


# ---- Step 5: gallery listing -- reads metadata only, never touches storage bytes. ----
def list_gallery(user_id):
    rows = [r for r in read_metadata_rows() if r["user_id"] == user_id]
    rows.sort(key=lambda r: r["created_at"])
    return [{"asset_id": r["asset_id"], "kind": r["kind"], "storage_key": r["storage_key"],
              "bytes": r["bytes"], "created_at": r["created_at"]} for r in rows]


# ---- Step 6: job-status shape for the slow half. Plain functions == the route bodies. ----
JOBS = {}

def generate_video_endpoint(user_id, image_asset_id, idempotency_key=None):
    """Shape of POST /generate/video -- submits, returns a job id immediately."""
    if idempotency_key and idempotency_key in JOBS.get("_idempotency", {}):
        return {"job_id": JOBS["_idempotency"][idempotency_key], "status": "accepted"}
    job_id = str(uuid.uuid4())
    JOBS[job_id] = {"status": "queued", "user_id": user_id, "source_asset": image_asset_id,
                     "submitted_at": time.monotonic()}
    JOBS.setdefault("_idempotency", {})
    if idempotency_key:
        JOBS["_idempotency"][idempotency_key] = job_id
    return {"job_id": job_id, "status": "accepted"}

def task_status_endpoint(job_id):
    """Shape of GET /task/{id} -- the slow half's poll target."""
    job = JOBS.get(job_id)
    if job is None:
        return {"status": "not_found"}, 404
    return {"job_id": job_id, "status": job["status"]}, 200

def mark_job_succeeded(job_id, result_key):
    JOBS[job_id]["status"] = "succeeded"
    JOBS[job_id]["result_key"] = result_key


def main():
    with open(os.path.join(os.path.dirname(HERE), "ch12_assets", "photo_nasa.jpg"), "rb") as f:
        stand_in_bytes = f.read()  # stand-in for the provider's response bytes
    image_row = download_and_store(stand_in_bytes, USER_ID,
        prompt="a red panda mascot for a job-search blog post", seed=99182,
        model_id="<current-image-model-id>")
    print(f"stored image: key={image_row['storage_key']} bytes={image_row['bytes']} "
          f"dims={image_row['width']}x{image_row['height']} mime={image_row['mime']}")

    src_path = os.path.join(STORAGE_ROOT, *image_row["storage_key"].split("/"))
    original_bytes, derivatives = make_derivatives(src_path, USER_ID, image_row["asset_id"])
    for name, d in derivatives.items():
        ratio = round(original_bytes / d["bytes"], 2)
        print(f"{name}: bytes={d['bytes']} dims={d['width']}x{d['height']} ({ratio}x smaller)")

    video_info = build_slideshow(USER_ID, image_row["asset_id"])
    print(f"video: return_code={video_info['return_code']} bytes={video_info['bytes']} "
          f"build_seconds={video_info['build_seconds']} duration={video_info['duration']} "
          f"resolution={video_info['resolution']} audio_present={video_info['audio_present']}")
    write_metadata_row({"asset_id": image_row["asset_id"] + "-video", "user_id": USER_ID,
        "kind": "video", "storage_key": video_info["video_key"],
        "prompt": "3-slide storyboard for the job-search blog post", "seed": None,
        "model": "ffmpeg-concat (no video-generation model)", "mime": "video/mp4",
        "bytes": video_info["bytes"], "created_at": datetime.now(timezone.utc).isoformat()})

    gallery = list_gallery(USER_ID)
    print(f"gallery item count: {len(gallery)}")

    submit_resp = generate_video_endpoint(USER_ID, image_row["asset_id"], idempotency_key="req-8841")
    print(f"POST /generate/video -> {submit_resp}")
    status_resp, code = task_status_endpoint(submit_resp["job_id"])
    print(f"GET /task/{{id}} (before finish) -> {code} {status_resp}")
    mark_job_succeeded(submit_resp["job_id"], video_info["video_key"])
    status_resp2, code2 = task_status_endpoint(submit_resp["job_id"])
    print(f"GET /task/{{id}} (after finish)  -> {code2} {status_resp2}")
    retry_resp = generate_video_endpoint(USER_ID, image_row["asset_id"], idempotency_key="req-8841")
    print(f"POST /generate/video retry (same idempotency key) -> {retry_resp} "
          f"(same job_id={retry_resp['job_id'] == submit_resp['job_id']})")


if __name__ == "__main__":
    main()