Image aur Video Generation interview questions & answers

Socho ek sculptor hai jo ek rough marble ke block se shuru karta hai jo actually sirf noise hai -- random bumps aur ridges -- aur baar-baar chheelta rehta hai jab tak ek shape na nikal aaye. Sculptor ko koi finished statue nahi diya jaata; use ek description di jaati hai ("ek baithi hui aurat, ek haath ghutne pe tika hua") aur sculptor har pass me thoda sa "not-that" hataata hai. Image aur video generation models bhi bilkul aise hi kaam karte hain: wo picture ko stroke-by-stroke nahi banaate, wo pure noise se shuru karte hain aur baar-baar usme se thoda sa hataate hain, har baar aapke text prompt se "is description jaisa aur" ki taraf nudge hote hue. Isi ko diffusion kehte hain, aur ye ek behaviour hai, koi ek specific architecture nahi -- kai alag-alag model families ise implement karti hain, aur behaviour hi wo cheez hai jo ek se doosre me carry hoti hai.

Production me, yehi behaviour wo wajah hai ki aapke app ko in teeno cheezon ke around plan karna padta hai, chahe aap koi bhi vendor call karo: ye kabhi apne aap ko exactly repeat nahi karta, ye chhote precise details ke baare me confidently galat hota hai, aur ise pata hi nahi hota ki result "correct" hai ya nahi -- iske andar check karne ke liye koi ground truth hoti hi nahi.

🌍 Real-world example: ek resume-builder feature user ko ek chhote prompt se profile banner generate karne deta hai ("minimal blue gradient banner, abstract geometric shapes, no text"). Isse teen baar generate karo aur teen alag banners milte hain -- sab prompt ke reasonably close, lekin koi bhi identical nahi, kyunki process har baar random noise ke ek fresh block se shuru hota hai. Agar ise banner ke andar user ka naam likhne ko kaha jaaye, toh wo bahut zyada chance se distorted, unreadable, near-text squiggles banayega, real letters nahi. Inme se koi bhi aapki integration ka bug nahi hai; dono hi model ka bilkul waisa behave karna hai jaisa diffusion behave karta hai. Fix ye hai ki iske around design karo -- pehle ke liye ek regenerate button do, aur doosre ke liye apne khud ke code se real text upar composite karo -- baar-baar re-prompt karke ummeed lagane ki bajaye.

💡 Diffusion = ek generation process jo random noise se shuru hota hai aur har step pe thoda sa hataata hai, aapke text prompt se steer hote hue, jab tak jo bacha wo ek image (ya video ke liye images ki ek sequence) jaisa na lage. "Diffusion model" isi noise-to-structure behaviour ko describe karta hai, kisi ek company ke product ko nahi -- kai alag model families isi tarah banti hain.

💡 Latent space = pixels ke full-resolution grid ko directly denoise karne ki bajaye (slow, aur zyaadatar detail redundant hoti hai), ye zyaadatar models image ke ek bahut chhote, compressed representation -- uske latent -- ko denoise karte hain, aur sirf last step me use wapas real pixels me expand karte hain. Aap API ke through is space ko kabhi directly touch nahi karte; yahi wajah hai ki generation raw pixels denoise karne se fast hota hai, aur ye ek sentence hai jo interview me term pehchaanne ke liye kaafi hai.

Same prompt har baar alag image kyun deta hai: starting point random noise hai, jo har baar jab aap API call karte ho ek random seed se fresh generate hota hai (jab tak aap khud ek seed pin na karo -- ye apne aap me ek alag topic hai). Alag starting noise, jab same text ki taraf denoise hota hai, toh ek alag point pe land karta hai jo phir bhi description ko match karta hai. Ye expected behaviour hai, koi inconsistency nahi jise debug karna ho.

Ye models kahan reliably break hote hain, aur kyun: model ne bahut saari images ke across statistical patterns seekhe hain -- ek haath usually kaisa dikhta hai, letters ek doosre ke paas usually kaise dikhte hain, kisi cheez ke "teen" usually kaisa dikhte hain -- na ki wo exact rules jo ek insaan apply karega (paanch ungliyaan, ek specific font, ek precise count). Isliye ye exactly wahaan sabse weak hai jahaan correctness exact aur unforgiving hai:

• Haath aur ungliyaan -- ek ordinary object jiski structure extremely specific aur easily galat hone wali hai; model ke paas exactly paanch force karne wala koi built-in constraint nahi hai.
• Image ke andar chhota text -- letters ek diffusion model ke liye visual patterns hain, kisi alphabet ke characters nahi, isliye ye letter-shaped noise banaata hai, spelling nahi. Yahi wajah hai ki "no text" wala negative prompt bhi aksar halki lettering peeche chhod deta hai (agla topic isse cover karta hai).
• Exact logos aur brand marks -- model kisi logo ki jo style usne dekhi hai wo reproduce karta hai, byte-perfect copy nahi, isliye wo utna close hota hai ki asli cheez jaanne wale ko galat lagega.
• Objects count karna -- "exactly chaar apples" ek precise constraint hai; model ke paas ek strong sense hai ki apples ka ek plausible dher kaisa dikhta hai, ek counter nahi.

Iske around design karna, hamesha re-prompt karte rehna nahi, yahi actual skill hai: exact text aur logos ko generation ke baad apne khud ke code se upar daalo, generated image ko ek draft ya background maano, precise facts ka source nahi, aur user ko pehli try me ek specific outcome promise karne ki bajaye ek regenerate button do.

Aap ye models call karte ho, unhe train nahi karte. Ye chapter job ke us ~1% hisse ke baare me hai jo API ko touch karta hai: prompt bhejna, response handle karna, jo wapas aata hai use store karna, refusal aur cost ke liye design karna. Diffusion model ko scratch se train ya fine-tune karna ek alag, research-heavy specialisation hai jiske apne hardware aur data requirements hain, aur ye yahaan scope se bahar hai -- bilkul waise jaise Ch8 ne LLM fine-tune karna sikhaya tha, use pretrain karna sikhaye bina.

Chapter ka opening frame, aur ye is course me kahin aur se zyada important yahaan kyun hai: is se pehle ka har chapter text produce karta tha -- chhota, sasta, instant, aur screen pe display hote hi khatam. Ye chapter bytes produce karta hai: ek image ya video file jo badi hai, banane me slow hai, per-unit priced hai, legally loaded hai (use kaun aur kis liye use kar sakta hai), aur jo request khatam hone ke bahut baad tak bhi kahin disk pe exist karti rehti hai. Ek generate call kaam ka end nahi hai, ye kaam ki shuruaat ke zyaada kareeb hai -- jis moment se bytes exist karti hain, unhe store karna, resize karna, serve karna, aur unme kya hai uska jawaab dena aapka kaam ban jaata hai. Yahi ek shift -- response versus artefact -- iske around baaki poora chapter banaya gaya hai.

Kab use karein: ek generation model tab use karo jab aapko ek original image ya video chahiye jo abhi exist nahi karta, aur fine detail ki exact correctness requirement nahi hai -- ek mood board, ek marketing banner, ek background, concept art, ya ek placeholder jab tak asli asset pe ek designer kaam kar raha ho.

Kab NAHI use karein: ise tab mat use karo jab aapko kuch pixel-exact ya legally clean chahiye ho -- company ka logo, kisi specific insaan ki photo, ek real product ka screenshot, ya koi bhi image jahaan ek chhota factual detail (ek count, ek exact label, ek real brand mark) sahi hona zaroori hai. Ek stock photo ya asset library jeetti hai jab "iss jaisi ek photo" pehle se exist karti ho aur licensing terms shuru se hi clear aur known hon. Ek human designer jeetta hai jab brand mark, exact layout, ya legal sign-off actually matter karta ho aur ek galat pixel ek real cost ho, na ki sirf ek redo-aur-aage-badho wali pareshaani.

Standard definition (interview me bolo): 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.

Text model se chat karna ek conversation hai -- aap keh sakte ho "actually, isko chhota kar do" aur wo poora context samajh jaata hai. Image model ko prompt dena zyada ek form bharne jaisa hai jisme fixed fields hain: subject, style, detail, lighting, composition. Aapne paanch minute pehle kya poocha tha uski koi back-and-forth memory nahi hoti (jab tak tool khud aapka last prompt context ki tarah re-send na kare) -- har generation call ek fresh form hai, aur har field me daale gaye words hi output ko steer karne wali akeli cheez hain.

Production me ye is tarah map hota hai: pehle prompt ko data ki tarah banao -- ek subject, style, detail, lighting, composition -- aur inhe sirf tabhi ek string me join karo jab aap API call kar rahe ho. Fields ko waise treat karo jaise form ke inputs ko karte -- ek badlo, baaki identical rakho, aur ab aap bata sakte ho ki output me kaunsa change kis field ki wajah se aaya. Ek saath teen fields badal do aur ek alag image milegi, lekin aapne kuch nahi seekha -- aap ye nahi bata sakte ki naye lighting words ne kiya, naye style words ne kiya, ya dono ke interaction ne.

Negative prompts un models pe field set ka doosra half hain jo inhe support karte hain: ek doosri string jo batati hai aapko kya NAHI chahiye -- "blurry, low quality, distorted, watermark, extra fingers, text". Inke baare me do cheezein baar-baar galat sikhai jaati hain:

1. Negative prompt koi delete button nahi hai. Ye output se ek concept ko waise remove nahi karta jaise sed ek line delete karta hai. Ye generation ke dauran model ki guidance ko us concept se DOOR shift karta hai -- ek steer, subtraction nahi. Kisi cheez ke liye ek strong positive signal (maan lo, ek prompt jo ek wooden signboard maangta hai) still ek related cheez ke against ek weak negative signal ("text") ko out-vote kar sakta hai, kyunki signboards, model ke training data me, usually text carry karte hain. Yahi wajah hai ki negative prompt me "no text" hone ke baad bhi aksar text aa jaata hai -- positive prompt implicitly ek associated object ke through use maang raha hota hai, aur negative field sirf ek nudge hai, veto nahi.
2. Har model me negative-prompt parameter hota hi nahi. Kuch models ise ek first-class field ki tarah accept karte hain; kuch sirf ek single prompt string accept karte hain aur expect karte hain ki aap jo chahte ho wo positively likho. Ye ek per-model capability hai jise aap lookup karte ho, waise hi jaise size ya quality knobs lookup karte ho -- kabhi assume mat karo ki ye exist karta hai.

Style words prompting se alag ek separate capability nahi hain -- wo usi form ka ek field hain. "Isko Van Gogh oil painting style me reimagine karo" aur "isko pixel art style me reimagine karo" do alag API features nahi hain; ye same style slot hai jisme alag value bhari gayi hai. Style transfer ko apni alag cheez samajhna hi wo wajah hai ki log ek alag tool pakadte hain jab already existing prompt me ek word change karna kaafi hota.

Neeche wala runnable half in prompt strings ko Python me build aur print karta hai -- koi API key nahi chahiye, kyunki ye pure string composition hai; iska ye depend nahi karta ki aap eventually kaunsa vendor call karte ho. Generation call khud ek separate, illustrative snippet hai, kyunki usko ek live API key chahiye jo is environment me nahi hai.

🌍 Real-world example: ek team "hamare anime-style outputs inconsistent lag rahe hain" debug karte waqt discover karti hai ki wo har test run ke beech subject wording, style wording AUR negative prompt teeno badal rahe the -- toh koi bhi ek generation pichhle se comparable nahi thi. Paanch me se chaar fields lock karke sirf style vary karne se teen din ke confused screenshots ek one-page comparison ban gaye.

💡 Prompt structure = subject / style / detail / lighting / composition ko separate fields ki tarah treat karna jinhe aap compose karte ho, ek sentence nahi jise har baar zero se rewrite karte ho.

💡 Negative prompt = ek doosra field (jahan supported ho) jo un concepts ko list karta hai jinse generation ke dauran door steer karna hai -- ek bias, removal instruction nahi.

Standard definition (interview me bolo): 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.

Kab use karein: pehle apne HI prompt fields pe iterate karo -- ek slot badlo, result dekho, agla badlo -- jab bhi current image close ho lekin ek aspect (lighting, style, framing) galat ho; ye sabse sasta fix hai aur ye ek mental map banane ka bhi ekmatra tareeka hai ki har field asal me kya control karta hai.

Kab NAHI use karein / Trade-off: agar wording ke das iterations ne ek specific, localised problem (ek face, ek logo, ek hand) fix nahi kiya, toh prompt-iterate karna band karo aur tool switch karo -- ya toh image ko baad me edit karo masked inpainting se (is chapter ka ek baad wala topic) ek local fix ke liye, ya text/logos khud composite karo ek image library se, model se exact characters render karwane ke bajaye (model ko kabhi reliably spell karne ke liye train nahi kiya gaya), ya model hi badal do agar failure structural hai (wo model consistently wo pose/style nahi kar pata jo aapko chahiye). Prompt wording ek aisi capability fix nahi kar sakti jo model ke paas hai hi nahi.

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.")

Kya bana rahe hain: is chapter ke har image aur video topic ne pipeline ke ek-ek piece ko alag-alag sikhaya -- bytes save karna, resize karna, ek slideshow banana, ek key ke neeche store karna, ek job poll karna. Ye project un sab ko us ek cheez me jodta hai jise ek user actually click karta hai: Generate. Ek prompt jaata hai; bees second ki engineering hoti hai; doosri taraf se ek real image, ek real video aur ek real gallery entry aati hai. Ye project FREE hai, aur ye maan kar chalta hai ki aapne is chapter me aur kuch nahi padha -- neeche har step apna idea zero se dobara derive karta hai.

\U0001f30d Real-world example: ek resume-builder ke "cover-letter illustration generate karo" button pe, user ko ek click dikhta hai. Uske peeche: ek provider call jo refuse kar sakti hai, bytes jinhe trust karne se pehle verify karna hai, ek resize taaki gallery card phone pe ek megabyte na bheje, ek metadata row taaki chhe mahine baad ka ek support ticket answer ho sake, aur -- agar feature ek chhoti intro video bhi promise karta hai -- ek job jo HTTP response se pehle khatam nahi ho sakti. Generate call us poore sentence ka sabse chhota hissa hai. Baaki sab engineering hai jo ye project banata hai.

\U0001f4a1 Pipeline = wo ordered sequence jisse ek request actually guzarti hai -- generate, verify, store, derive, list, poll -- na ki "API call karo", jo isme sirf pehli link hai.

Standard definition (interview me bolo): 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.

⚠ Ek plain sentence ki kya chala aur kya nahi: image-generation aur video-generation provider calls illustrative hain -- is machine pe koi key nahi hai, isi chapter ke baaki har coding topic ki tarah. Neeche ki baaki har line -- bytes download aur verify karna, derivatives banana, ffmpeg se slideshow encode karna, metadata store likhna aur padhna, gallery list karna, aur job-status endpoint ka submit/poll/idempotency logic -- real me chali, do baar, aur dono runs me har byte count identical aaya.

Step 1 -- Architecture: paanch endpoints, ek async branch

Aise feature ke peeche brief hamesha kuch aisa hota hai: ek prompt se image generate karo, optionally use ek chhoti video me badlo, aur user ko sab kuch browse karne do jo unhone banaya. Kisi bhi ek vendor ke product naam hataa ke, ye paanch endpoints me map hota hai -- aur har ek ka shape hi wo cheez hai jo ye project actually sikhata hai, na ki kis company ne use banaya:

POST /generate/image     text -> image. Itna slow ki user wait kare, itna fast ki inline answer ho jaaye.
POST /generate/video     image -> video. Ek HTTP request ke liye bahut slow -- ek job id return karta hai, video nahi.
GET  /task/{id}          slow half ka poll target: /generate/video ka job abhi kya kar raha hai?
GET  /gallery            is user ne jo bhi generate kiya, metadata store se padha gaya.
DELETE /content/{id}     ek asset hataao -- storage key AUR uski metadata row dono, kabhi akela nahi.

List nahi, shape dekho: in paanch me se exactly ek asynchronous hai, aur ye structural wajah se hai, stylistic nahi -- image generation seconds ka hai, video generation minutes ka, aur koi HTTP client minutes tak connection open nahi rakhta. Is ek fact ke neeche ki har cheez -- job id, poll target, submit pe idempotency key -- isi ek asymmetry ki wajah se exist karti hai. DELETE /content/{id} neeche nahi banaya gaya (ye ek os.remove plus ek metadata-row filter hai, aisa kuch nahi jo is chapter ne pehle cover na kiya ho), lekin ye list pe isi wajah se hai jis wajah se checklist code se zyada matter karti hai: use bhoolna ka matlab hai user ka "delete" ek orphaned blob chhod deta hai jo hamesha ke liye storage cost karta rehta hai.

Step 2 -- Prompt se 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()          # abhi ke abhi poora decode force karta hai, baad me lazy failure nahi
    ...

Function exactly yahaan kyun split hua hai, kisi zyada convenient jagah pe kyun nahi: illustrative_generate_call is poore project ka WAHI ek part hai jo kisi provider se baat karta hai, aur ye wahi ek function hai jo ye project kabhi nahi chalata. raw_bytes line ke neeche ki har cheez real hai, kyunki wo real hai chaahe wo bytes kisi bhi provider ne banaye hon, wo ek download karne wale URL ki tarah aaye ya decode karne wale base64 ki tarah, aur call ki koi cost lagi ho ya nahi. Function ko yahaan split karna wahi lesson hai jo calling-an-image-api aur media-storage-and-urls ne alag-alag sikhaya, ab code ke apne shape se enforce ho raha hai: aap jo OWN karte ho uska SHAPE is baat pe depend nahi karta ki aapne kis vendor ko call kiya.

Verified real run: download-and-store half ko ek real JPEG ke actual bytes diye gaye, ek generated image ka stand-in, aur usne diya stored image: key=user/u_7c19/325a2a1b-...jpg bytes=130786 dims=1280x639 mime=image/jpeg -- ek real decode, disk pe ek real file is chapter ke storage topic ke banaye hue user/<id>/<asset-id>.jpg key convention ke neeche, aur uske saath likhi ek real metadata row (prompt, seed, model, mime, dimensions, byte count -- us row me kabhi khud bytes nahi).

Step 3 -- Pillow se derivatives: master kabhi serve nahi hoti

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

Do derivatives kyun, "resize on request" endpoint kyun nahi: dono derivatives ko ek baar, generation time pe, banana aur master ke saath cache karna wahi poora lesson hai jo sizes-formats-and-delivery ne measure kiya -- master archival copy hai, aur pixel dimensions strong knob hain, isliye dono derivatives pehle resize karte hain aur quality ko doosre number pe set karte hain, ulta nahi. Verified real run: 130,786-byte master ne ek 50,214-byte web derivative diya (640x319, 2.6x chhota) aur ek 12,873-byte thumb (320x160, 10.16x chhota) -- koi bhi number invented nahi hai, dono usi source pixels ko encode karne se aaye jo is run ne actually decode kiye.

Step 4 -- ffmpeg se ek slideshow video: koi video-generation call hi nahi

lines.append(f"file '{p}'"); lines.append("duration 1.5")   # har slide ke liye
lines.append(f"file '{slide_paths[-1]}'")                    # repeat -- isi chapter ka measured quirk
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]

Is project ki "video generation" ffmpeg kyun hai, provider call nahi: brief ko sirf teen static slides chahiye -- ek storyboard, real motion nahi -- aur video-without-a-video-model ne exactly ye establish kiya ki ye substitution kab sahi hai: genuinely static content, zero generation cost pe, fully deterministic. Concat list final slide ka filename bina trailing duration line ke repeat karti hai, kyunki demuxer har entry ko us file ke against time karta hai jo uske BAAD aati hai -- repeat skip karo aur last slide ki maangi hui length chupke se drop ho jaati hai, bina kisi error ya warning ke. Verified real run: return_code=0 bytes=44922 build_seconds=0.366 duration=00:00:04.50 resolution=1280x720 audio_present=True -- ek real MP4 jise ye project khol sakta hai, teen Pillow-drawn captioned frames aur ek lavfi sine tone se bana, poori chain me kahin bhi zero video-generation API calls ke saath.

Step 5 -- Bytes + metadata store karo, phir ek gallery jo storage ko kabhi touch nahi karti

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]

Ek JSON-lines file is machine pe database table ka stand-in hai, aur object_storage_standin/ bucket ka -- ye split real hai chaahe in do roles ke peeche koi bhi do products hon. list_gallery object_storage_standin/ ke neeche koi file kyun kabhi nahi kholta: ek gallery endpoint ka kaam "is user ke paas kya hai" batana hai, "ye rahe bytes" nahi -- exactly wahi row-holds-the-key-not-the-bytes lesson jo media-storage-and-urls ne measure kiya, ab ek rule ki tarah bataye jaane ke bajaye teen lines ki route logic se call ho raha hai. Verified real run: image aur video dono ek-ek metadata row ki tarah likhe jaane ke baad, list_gallery("u_7c19") ne exactly 2 items return kiye, created_at se sorted, har ek apni storage_key aur byte count ke saath aur kisi signed URL jaisi cheez ke bina -- wo URL sirf tab mint hoti hai jab ek browser actually ek asset dekhne ke liye poochta hai, row me khud kabhi store nahi hoti.

Step 6 -- Slow half ke liye ek job-status endpoint

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

Ye wahi shape hai jo async-jobs-and-polling ne isolation me sikhaya, yahaan POST /generate/video aur GET /task/{id} ki is project ke architecture me actual jagah se call ho raha hai. Yahaan idempotency key is chapter me kahin bhi se zyada kyun matter karti hai: video is chapter ka named sabse expensive artefact hai, duration ke hisaab se priced -- ek client jo /generate/video ka wait karte-karte timeout ho jaata hai aur click retry karta hai, use ek doosri, alag-billed job start NAHI karni chahiye. Verified real run: endpoint task_status_endpoint ki do real calls me queued -> succeeded gaya, aur generate_video_endpoint ko identical idempotency key "req-8841" ke saath do baar call karne pe dono baar same job_id mila -- ek hi job kabhi banaya gaya, chaahe submit function khud do baar call hui ho.

Ye project genuinely kya prove karta hai, aur kya nahi

Prove karta hai: generation call ke neeche ki har cheez real me banayi aur chalayi gayi, do baar, dono baar byte-identical results ke saath -- download-and-verify, do Pillow derivatives measured size ratios pe, ek ffmpeg slideshow ek real duration aur ek real audio stream ke saath, ek metadata store jise gallery endpoint storage touch kiye bina padh sakta hai, aur ek job-status endpoint jiski idempotency key ne genuinely ek retried submit pe doosri job banne se roka.

Nahi prove karta (interviewer se pehle khud bata do): image- aur video-generation provider calls illustrative hain -- is machine pe koi key nahi hai, aur ek real provider ko swap karne se uske around banaayi gayi pipeline me kuch nahi badalta, kyunki wo pipeline SHAPE ke hisaab se likhi gayi (bytes in, job id out), kisi ek vendor ke field names ke hisaab se nahi. Yahaan koi real database nahi hai, koi real object-storage SDK nahi hai aur koi real HTTP server nahi hai -- ek JSON-lines file aur ek local directory tree unke liye stand-in hain, aur generate_video_endpoint/task_status_endpoint plain functions hain us exact body ke saath jo ek FastAPI route ki hoti, network ke bajaye directly call kiye gaye. Aur DELETE /content/{id} architecture me named hai lekin banaya nahi gaya -- ye upar ki har cheez se chhota problem hai, harder nahi.

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()