Upload 24 files

.env

@@ -0,0 +1,2 @@
+OPENROUTER_API_KEY=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
+

Dockerfile

@@ -0,0 +1,70 @@
+# Base image with Python
+FROM python:3.10-slim
+
+# Set working directory
+WORKDIR /app
+
+# Environment variables
+ENV PYTHONUNBUFFERED=1
+ENV MODEL_DIR=/tmp/models/realvisxl_v4
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    git \
+    wget \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy all project files
+COPY . /app
+
+# Create model directory
+RUN mkdir -p ${MODEL_DIR}
+
+# --- Install Python dependencies ---
+RUN pip install --no-cache-dir \
+    annotated-types==0.7.0 \
+    anyio==4.11.0 \
+    certifi==2025.10.5 \
+    click==8.3.0 \
+    colorama==0.4.6 \
+    fastapi==0.119.0 \
+    h11==0.16.0 \
+    httpcore==1.0.9 \
+    httpx==0.28.1 \
+    idna==3.11 \
+    pydantic==2.12.2 \
+    pydantic_core==2.41.4 \
+    python-dotenv==1.1.1 \
+    sniffio==1.3.1 \
+    starlette==0.48.0 \
+    typing_extensions==4.15.0 \
+    typing-inspection==0.4.2 \
+    uvicorn==0.37.0 \
+    torch torchvision torchaudio --extra-index-url https://download.pytorch.org/whl/cpu \
+    diffusers==0.30.3 \
+    huggingface_hub==0.26.2 \
+    accelerate==1.1.1 \
+    safetensors==0.4.5 \
+    pillow==10.4.0
+
+# --- Pre-download model into /tmp/models/realvisxl_v4 ---
+RUN python -c "\
+import torch; \
+from diffusers import StableDiffusionXLPipeline; \
+from pathlib import Path; \
+model_dir = Path('${MODEL_DIR}'); \
+if not model_dir.exists(): \
+    print('Downloading model...'); \
+    pipe = StableDiffusionXLPipeline.from_pretrained('SG161222/RealVisXL_V4.0', torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32); \
+    pipe.save_pretrained(model_dir); \
+    print('Model downloaded to', model_dir); \
+else: \
+    print('Model already exists at', model_dir); \
+"
+
+# Expose the app port
+EXPOSE 8000
+
+# Command to run FastAPI server
+CMD ["uvicorn", "api.server:app", "--host", "0.0.0.0", "--port", "8000"]

api/server.py

@@ -0,0 +1,403 @@
+# # api/server.py
+# from fastapi import FastAPI, HTTPException
+# from pydantic import BaseModel
+# import asyncio
+# from pipeline.pipeline_runner import run_pipeline_task
+
+# app = FastAPI(title="AI ADD Maker API", version="1.0")
+
+# # Request body schema
+# class GenerateRequest(BaseModel):
+#     idea: str
+
+# # Response schema (optional, for docs clarity)
+# class GenerateResponse(BaseModel):
+#     task_id: str
+#     scenes: list
+#     images: list
+#     video: dict
+#     music: dict
+#     final_output: dict
+
+# @app.post("/generate", response_model=GenerateResponse)
+# async def generate_video(request: GenerateRequest):
+#     """
+#     Trigger the AI Ad/Video pipeline with a product or idea description.
+#     Returns all intermediate and final outputs.
+#     """
+#     idea = request.idea.strip()
+#     if not idea:
+#         raise HTTPException(status_code=400, detail="Idea cannot be empty.")
+
+#     try:
+#         # Run the pipeline asynchronously
+#         result = await run_pipeline_task(idea)
+#         return result
+#     except Exception as e:
+#         raise HTTPException(status_code=500, detail=f"Pipeline failed: {str(e)}")
+
+# # Optional health check endpoint
+# @app.get("/health")
+# async def health_check():
+#     return {"status": "ok"}
+
+
+
+
+
+# # server.py
+# import uuid
+# import asyncio
+# from fastapi import FastAPI, HTTPException
+# from pydantic import BaseModel
+# from queue import QueueManager  # your queue.py module
+# import logging
+
+# # -------------------------------
+# # Setup logging
+# # -------------------------------
+# logging.basicConfig(
+#     level=logging.INFO,
+#     format="%(asctime)s [%(levelname)s] %(message)s"
+# )
+
+# # -------------------------------
+# # FastAPI app
+# # -------------------------------
+# app = FastAPI(title="ADD Maker Server", version="1.0")
+
+# # -------------------------------
+# # Pydantic models
+# # -------------------------------
+# class IdeaRequest(BaseModel):
+#     idea: str
+
+# class ConfirmationRequest(BaseModel):
+#     task_id: str
+#     confirm: bool
+
+# # -------------------------------
+# # Queue Manager Instance
+# # -------------------------------
+# queue_manager = QueueManager()
+
+# # In-memory task storage for confirmation
+# pending_confirmations = {}  # task_id -> asyncio.Event
+
+# # -------------------------------
+# # Helper function for confirmation
+# # -------------------------------
+# async def wait_for_confirmation(task_id: str, timeout: int = 120):
+#     """Wait for user confirmation or auto-confirm after timeout."""
+#     event = asyncio.Event()
+#     pending_confirmations[task_id] = event
+#     try:
+#         await asyncio.wait_for(event.wait(), timeout=timeout)
+#         logging.info(f"Task {task_id} confirmed by user.")
+#         return True
+#     except asyncio.TimeoutError:
+#         logging.info(f"Task {task_id} auto-confirmed after {timeout} seconds.")
+#         return True
+#     finally:
+#         pending_confirmations.pop(task_id, None)
+
+# # -------------------------------
+# # API Endpoints
+# # -------------------------------
+# @app.post("/submit_idea")
+# async def submit_idea(request: IdeaRequest):
+#     task_id = str(uuid.uuid4())
+#     logging.info(f"Received idea: {request.idea} | Task ID: {task_id}")
+
+#     # Push task to queue
+#     await queue_manager.enqueue({
+#         "task_id": task_id,
+#         "idea": request.idea
+#     })
+
+#     # Start confirmation wait in background
+#     asyncio.create_task(wait_for_confirmation(task_id))
+
+#     return {"status": "submitted", "task_id": task_id, "message": "Idea received, waiting for confirmation."}
+
+# @app.post("/confirm")
+# async def confirm_task(request: ConfirmationRequest):
+#     task_id = request.task_id
+#     if task_id not in pending_confirmations:
+#         raise HTTPException(status_code=404, detail="Task not pending confirmation or already confirmed.")
+
+#     if request.confirm:
+#         pending_confirmations[task_id].set()
+#         return {"status": "confirmed", "task_id": task_id}
+#     else:
+#         return {"status": "rejected", "task_id": task_id}
+
+# @app.get("/")
+# async def health_check():
+#     return {"status": "running"}
+
+# # -------------------------------
+# # Startup / Shutdown events
+# # -------------------------------
+# @app.on_event("startup")
+# async def startup_event():
+#     logging.info("Server starting up...")
+
+# @app.on_event("shutdown")
+# async def shutdown_event():
+#     logging.info("Server shutting down...")
+
+
+
+# # best
+# # server.py
+# import uuid
+# import asyncio
+# from fastapi import FastAPI, HTTPException
+# from pydantic import BaseModel
+# from services import queue_manager as queue_manager  # ✅ import your actual queue module
+# import logging
+# from fastapi.middleware.cors import CORSMiddleware
+
+
+# # -------------------------------
+# # Setup logging
+# # -------------------------------
+# logging.basicConfig(
+#     level=logging.INFO,
+#     format="%(asctime)s [%(levelname)s] %(message)s"
+# )
+
+# # -------------------------------
+# # FastAPI app
+# # -------------------------------
+# app = FastAPI(title="AI ADD Generator Server", version="1.0")
+# # Enable CORS for local testing
+# app.add_middleware(
+#     CORSMiddleware,
+#     allow_origins=["*"],  # Allow requests from anywhere (for testing)
+#     allow_credentials=True,
+#     allow_methods=["*"],  # Allow all HTTP methods
+#     allow_headers=["*"],  # Allow all headers
+# )
+
+# # -------------------------------
+# # Pydantic models
+# # -------------------------------
+# class IdeaRequest(BaseModel):
+#     idea: str
+
+# class ConfirmationRequest(BaseModel):
+#     task_id: str
+#     confirm: bool
+
+# # -------------------------------
+# # In-memory confirmation tracker
+# # -------------------------------
+# pending_confirmations = {}  # task_id -> asyncio.Event
+
+# # -------------------------------
+# # Helper function for confirmation
+# # -------------------------------
+# async def wait_for_confirmation(task_id: str, timeout: int = 120):
+#     """Wait for user confirmation or auto-confirm after timeout."""
+#     event = asyncio.Event()
+#     pending_confirmations[task_id] = event
+#     try:
+#         await asyncio.wait_for(event.wait(), timeout=timeout)
+#         logging.info(f"✅ Task {task_id} confirmed by user.")
+#         await queue_manager.confirm_task(task_id)
+#         return True
+#     except asyncio.TimeoutError:
+#         logging.info(f"⌛ Task {task_id} auto-confirmed after {timeout}s.")
+#         await queue_manager.confirm_task(task_id)
+#         return True
+#     finally:
+#         pending_confirmations.pop(task_id, None)
+
+# # -------------------------------
+# # API Endpoints
+# # -------------------------------
+
+# @app.post("/submit_idea")
+# async def submit_idea(request: IdeaRequest):
+#     """Receives a new ad idea and enqueues it."""
+#     task_id = await queue_manager.add_task(request.idea)
+#     logging.info(f"💡 New idea received | Task ID: {task_id}")
+
+#     # Start confirmation listener
+#     asyncio.create_task(wait_for_confirmation(task_id))
+
+#     return {
+#         "status": "submitted",
+#         "task_id": task_id,
+#         "message": "Idea received. Waiting for user confirmation after script generation."
+#     }
+
+
+# @app.post("/confirm")
+# async def confirm_task(request: ConfirmationRequest):
+#     """Confirms a paused task and continues the pipeline."""
+#     task_id = request.task_id
+#     if task_id not in pending_confirmations:
+#         raise HTTPException(status_code=404, detail="Task not pending confirmation or already confirmed.")
+
+#     if request.confirm:
+#         pending_confirmations[task_id].set()
+#         return {"status": "confirmed", "task_id": task_id}
+#     else:
+#         return {"status": "rejected", "task_id": task_id}
+
+
+# @app.get("/status/{task_id}")
+# async def get_status(task_id: str):
+#     """Check the current status of a task."""
+#     status = queue_manager.get_task_status(task_id)
+#     if not status:
+#         raise HTTPException(status_code=404, detail="Task not found.")
+#     return status
+
+
+# @app.get("/")
+# async def health_check():
+#     return {"status": "running", "message": "AI ADD Generator is live."}
+
+
+# # -------------------------------
+# # Startup / Shutdown events
+# # -------------------------------
+# @app.on_event("startup")
+# async def startup_event():
+#     logging.info("🚀 Server starting up...")
+#     queue_manager.start_worker()  # ✅ Start async worker loop
+
+# @app.on_event("shutdown")
+# async def shutdown_event():
+#     logging.info("🛑 Server shutting down...")
+
+
+
+import uuid
+import asyncio
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+from services import queue_manager  # ✅ import your actual queue module
+import logging
+from fastapi.middleware.cors import CORSMiddleware
+
+# -------------------------------
+# Setup logging
+# -------------------------------
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s"
+)
+
+# -------------------------------
+# FastAPI app
+# -------------------------------
+app = FastAPI(title="AI ADD Generator Server", version="1.0")
+
+# Enable CORS for local testing
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# -------------------------------
+# Pydantic models
+# -------------------------------
+class IdeaRequest(BaseModel):
+    idea: str
+
+class ConfirmationRequest(BaseModel):
+    task_id: str
+    confirm: bool
+
+# -------------------------------
+# In-memory confirmation tracker
+# -------------------------------
+pending_confirmations = {}  # task_id -> asyncio.Event
+script_results = {}         # task_id -> generated script for confirmation
+
+# -------------------------------
+# API Endpoints
+# -------------------------------
+@app.post("/submit_idea")
+async def submit_idea(request: IdeaRequest):
+    """Receives a new ad idea and enqueues it."""
+    task_id = await queue_manager.add_task(request.idea)
+    logging.info(f"💡 New idea received | Task ID: {task_id}")
+
+    # Start worker listener
+    asyncio.create_task(queue_manager.wait_for_script(task_id, script_results))
+
+    return {
+        "status": "submitted",
+        "task_id": task_id,
+        "message": "Idea received. Script will be generated shortly.",
+    }
+
+@app.post("/confirm")
+async def confirm_task(request: ConfirmationRequest):
+    """Confirms a paused task, generates story, and returns full JSON."""
+    task_id = request.task_id
+    task = queue_manager.get_task_status(task_id)
+    if not task:
+        raise HTTPException(status_code=404, detail="Task not found.")
+
+    if task["status"] != queue_manager.TaskStatus.WAITING_CONFIRMATION:
+        raise HTTPException(status_code=400, detail="Task not waiting for confirmation.")
+
+    if request.confirm:
+        # Confirm task
+        await queue_manager.confirm_task(task_id)
+        logging.info(f"✅ Task {task_id} confirmed by user.")
+
+        # Generate story immediately
+        script_result = task["result"]["script"]
+        story_result = await queue_manager.generate_story_after_confirm(script_result)
+        task["result"]["story_script"] = story_result
+        task["status"] = queue_manager.TaskStatus.COMPLETED
+
+        logging.info(f"🎬 Task {task_id} story generated and task completed.")
+        return {"status": "completed", "task": task}
+
+    else:
+        task["status"] = queue_manager.TaskStatus.FAILED
+        return {"status": "rejected", "task_id": task_id}
+
+
+@app.get("/status/{task_id}")
+async def get_status(task_id: str):
+    """Check the current status of a task."""
+    task = queue_manager.get_task_status(task_id)
+    if not task:
+        raise HTTPException(status_code=404, detail="Task not found.")
+
+    # If waiting confirmation, return script only
+    if task["status"] == queue_manager.TaskStatus.WAITING_CONFIRMATION:
+        return {"status": task["status"], "script": task["result"]["script"]}
+
+    return task
+
+
+@app.get("/")
+async def health_check():
+    return {"status": "running", "message": "AI ADD Generator is live."}
+
+
+# -------------------------------
+# Startup / Shutdown events
+# -------------------------------
+@app.on_event("startup")
+async def startup_event():
+    logging.info("🚀 Server starting up...")
+    queue_manager.start_worker()
+
+@app.on_event("shutdown")
+async def shutdown_event():
+    logging.info("🛑 Server shutting down...")

core/assembler.py

@@ -0,0 +1,58 @@
+# core/assembler.py
+"""
+Assembler combines video + music and optionally overlays images/text,
+and produces final assembled output. Uses ffmpeg for merging audio/video.
+Returns a dict with final path and metadata.
+"""
+import subprocess
+import time
+from pathlib import Path
+from typing import Dict
+from config import OUTPUT_DIR
+
+OUTPUT_DIR = Path(OUTPUT_DIR)
+OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+
+def assemble_video(video_result: Dict, music_result: Dict, out_name: str = None) -> Dict:
+    """
+    video_result: dict returned from video_generator
+    music_result: dict returned from music_generator
+    Returns dict { "final_path":..., "duration":..., "meta":... }
+    """
+    t0 = time.time()
+    video_path = Path(video_result.get("video_path"))
+    music_path = Path(music_result.get("music_path"))
+    out_name = out_name or f"final_{int(time.time()*1000)}.mp4"
+    out_path = OUTPUT_DIR / out_name
+
+    # If ffmpeg is available, merge audio and video
+    if video_path.exists() and music_path.exists():
+        cmd = [
+            "ffmpeg", "-y",
+            "-i", str(video_path),
+            "-i", str(music_path),
+            "-c:v", "copy",  # copy video stream
+            "-c:a", "aac",
+            "-shortest",
+            str(out_path)
+        ]
+        try:
+            subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            duration = video_result.get("duration", 0.0)
+        except Exception as e:
+            print(f"[assembler] ffmpeg merge failed: {e}. Creating placeholder final output.")
+            with open(out_path, "wb") as f:
+                f.write(b"")
+            duration = 0.0
+    else:
+        # If audio or video missing, create a placeholder
+        with open(out_path, "wb") as f:
+            f.write(b"")
+        duration = 0.0
+
+    meta = {
+        "video_src": str(video_path),
+        "music_src": str(music_path),
+        "assembled_at": time.time() - t0
+    }
+    return {"final_path": str(out_path), "duration": duration, "meta": meta}

core/image_generator.py

@@ -0,0 +1,85 @@
+# core/image_generator.py
+import os
+import torch
+from diffusers import StableDiffusionXLPipeline
+from huggingface_hub import hf_hub_download
+from pathlib import Path
+from typing import List
+
+# ---------------- MODEL CONFIG ----------------
+MODEL_REPO = "SG161222/RealVisXL_V4.0"
+MODEL_FILENAME = "realvisxlV40_v40LightningBakedvae.safetensors"
+MODEL_DIR = Path("/tmp/models/realvisxl_v4")
+os.makedirs(MODEL_DIR, exist_ok=True)
+
+# ---------------- MODEL DOWNLOAD ----------------
+def download_model() -> Path:
+    """
+    Downloads RealVisXL V4.0 model if not present.
+    Returns the local model path.
+    """
+    model_path = MODEL_DIR / MODEL_FILENAME
+    if not model_path.exists():
+        print("[ImageGen] Downloading RealVisXL V4.0 model...")
+        model_path = hf_hub_download(
+            repo_id=MODEL_REPO,
+            filename=MODEL_FILENAME,
+            local_dir=str(MODEL_DIR),
+            force_download=False,
+        )
+        print(f"[ImageGen] Model downloaded to: {model_path}")
+    else:
+        print("[ImageGen] Model already exists. Skipping download.")
+    return model_path
+
+# ---------------- PIPELINE LOAD ----------------
+def load_pipeline() -> StableDiffusionXLPipeline:
+    """
+    Loads the RealVisXL V4.0 model for image generation.
+    """
+    model_path = download_model()
+    print("[ImageGen] Loading model into pipeline...")
+    pipe = StableDiffusionXLPipeline.from_single_file(
+        str(model_path),
+        torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
+    )
+    if torch.cuda.is_available():
+        pipe.to("cuda")
+    print("[ImageGen] Model ready.")
+    return pipe
+
+# ---------------- GLOBAL PIPELINE CACHE ----------------
+pipe: StableDiffusionXLPipeline | None = None
+
+# ---------------- IMAGE GENERATION ----------------
+def generate_images(prompt: str, seed: int = None, num_images: int = 3) -> List:
+    """
+    Generates high-quality images using RealVisXL V4.0.
+    Supports deterministic generation using a seed.
+    
+    Args:
+        prompt (str): Text prompt for image generation.
+        seed (int, optional): Seed for deterministic generation.
+        num_images (int): Number of images to generate.
+        
+    Returns:
+        List: Generated PIL images.
+    """
+    global pipe
+    if pipe is None:
+        pipe = load_pipeline()
+
+    print(f"[ImageGen] Generating {num_images} image(s) for prompt: '{prompt}' with seed={seed}")
+    images = []
+
+    for i in range(num_images):
+        generator = None
+        if seed is not None:
+            device = "cuda" if torch.cuda.is_available() else "cpu"
+            generator = torch.Generator(device).manual_seed(seed + i)  # slightly vary keyframes
+
+        result = pipe(prompt, num_inference_steps=30, generator=generator).images[0]
+        images.append(result)
+
+    print(f"[ImageGen] Generated {len(images)} images successfully.")
+    return images

core/music_generator.py

@@ -0,0 +1,35 @@
+# core/music_generator.py
+"""
+Music generator stub. You can plug in a TTS or music model (Coqui, MusicLM, etc.)
+Returns dict: {"music_path": ..., "duration": ..., "model": ...}
+"""
+import asyncio
+from pathlib import Path
+import time
+from concurrent.futures import ThreadPoolExecutor
+from config import OUTPUT_DIR
+from typing import Dict
+
+
+
+OUTPUT_DIR = Path(OUTPUT_DIR)
+OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+_executor = ThreadPoolExecutor(max_workers=1)
+
+def _sync_generate_music(idea: str, out_path: Path, bpm=100):
+    """
+    Replace with your real music/TTS generator.
+    For now this creates a placeholder file and metadata.
+    """
+    t0 = time.time()
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    with open(out_path, "wb") as f:
+        f.write(b"")  # replace with real audio bytes
+    meta = {"prompt": idea, "bpm": bpm, "time_taken": time.time() - t0}
+    return {"music_path": str(out_path), "duration": 3.0, "meta": meta}
+
+async def generate_music(idea: str) -> Dict:
+    loop = asyncio.get_event_loop()
+    out_path = OUTPUT_DIR / f"music_{int(time.time()*1000)}.wav"
+    result = await loop.run_in_executor(_executor, _sync_generate_music, idea, out_path)
+    return result

core/script_gen.py

@@ -0,0 +1,98 @@
+# # script_gen.py
+# import os
+# import asyncio
+# import httpx
+# import logging
+# from dotenv import load_dotenv
+
+# dotenv_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '.env')
+# load_dotenv(dotenv_path)
+# logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+# OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
+# MODEL_NAME = "deepseek/deepseek-r1-distill-llama-70b:free"
+# OPENROUTER_URL = f"https://openrouter.ai/api/v1/chat/completions"
+
+# async def generate_script(idea: str) -> str:
+#     """
+#     Generate a highly interactive ad script from user idea.
+#     Includes detailed expressions, actions, and minute details.
+#     """
+#     prompt = f"""
+# You are a professional ad writer.
+# Take this idea and generate a fun, interactive, and detailed ad script.
+# Include **minute expressions, subtle actions, emotions, and character reactions**.
+# Idea: {idea}
+# The script should be ready for storyboard creation.
+# """
+
+#     headers = {"Authorization": f"Bearer {OPENROUTER_API_KEY}"}
+#     payload = {
+#         "model": MODEL_NAME,
+#         "messages": [{"role": "user", "content": prompt}],
+#         "temperature": 0.8,
+#         "max_tokens": 1200
+#     }
+
+#     async with httpx.AsyncClient(timeout=120) as client:
+#         response = await client.post(OPENROUTER_URL, json=payload, headers=headers)
+#         response.raise_for_status()
+#         data = response.json()
+
+#     script = data["choices"][0]["message"]["content"]
+#     logging.info("Script generated successfully")
+#     return script
+
+
+
+# script_gen.py
+import os
+import asyncio
+import httpx
+import logging
+from dotenv import load_dotenv
+
+dotenv_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '.env')
+load_dotenv(dotenv_path)
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
+MODEL_NAME = "deepseek/deepseek-r1-distill-llama-70b:free"
+OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
+
+async def generate_script(idea: str) -> str:
+    """
+    Generate a short, funny, no-dialogue ad script based on a user idea.
+    The script should be written in plain text — simple narrative form,
+    no markdown, no camera angles, and no stage directions.
+    Just like:
+        A guy sits in a quiet library. He slowly opens a packet of Lay’s...
+    """
+    prompt = f"""
+You are a creative ad script writer.
+Write a short, funny, no-dialogue ad script from this idea: "{idea}".
+Keep it simple and cinematic.
+Use plain text only, with minimal expressions.
+No markdown, no numbering, no headings, no scene titles.
+Just write it like a mini story in clean text form.
+"""
+
+    headers = {"Authorization": f"Bearer {OPENROUTER_API_KEY}"}
+    payload = {
+        "model": MODEL_NAME,
+        "messages": [{"role": "user", "content": prompt}],
+        "temperature": 0.8,
+        "max_tokens": 600
+    }
+
+    async with httpx.AsyncClient(timeout=120) as client:
+        response = await client.post(OPENROUTER_URL, json=payload, headers=headers)
+        response.raise_for_status()
+        data = response.json()
+
+    script = data["choices"][0]["message"]["content"].strip()
+    logging.info("Simple script generated successfully")
+    return script
+
+# Example usage (for testing)
+# asyncio.run(generate_script("Lay’s funny ad in a library with no dialogues"))

core/script_generator.py

@@ -0,0 +1,95 @@
+# core/script_generator.py
+import asyncio
+import uuid
+from typing import List, Dict
+from config import OPENROUTER_API_KEY
+from core.seed_manager import SeedManager
+import httpx
+import json
+
+# Initialize seed manager
+seed_manager = SeedManager()
+
+# ---------------- OPENROUTER LLM CALL ----------------
+async def _call_openrouter_llm(prompt: str) -> str:
+    """
+    Calls OpenRouter LLM to generate proposed video script.
+    Returns the raw text script.
+    """
+    url = "https://api.openrouter.ai/v1/chat/completions"
+    headers = {"Authorization": f"Bearer {OPENROUTER_API_KEY}"}
+    payload = {
+        "model": "gpt-4.1-mini",  # powerful and suitable for script generation
+        "messages": [
+            {"role": "system", "content": "You are a professional creative video script writer."},
+            {"role": "user", "content": prompt}
+        ],
+        "max_tokens": 1500,
+        "temperature": 0.7
+    }
+
+    async with httpx.AsyncClient(timeout=60) as client:
+        response = await client.post(url, json=payload, headers=headers)
+        response.raise_for_status()
+        data = response.json()
+        # OpenRouter returns message content in choices[0].message.content
+        return data["choices"][0]["message"]["content"]
+
+# ---------------- SCRIPT PROCESSING ----------------
+def parse_script_to_scenes(script_text: str) -> List[Dict]:
+    """
+    Converts a script text into scene + keyframe JSON.
+    Each scene may have multiple keyframes.
+    Assigns unique scene_ids and seeds.
+    """
+    scenes_json = []
+    scene_counter = 1
+    keyframe_counter = 1
+
+    lines = [line.strip() for line in script_text.split("\n") if line.strip()]
+    for line in lines:
+        # Generate a unique scene_id and seed for this scene
+        scene_id = scene_counter
+        seed = seed_manager.generate_seed(scene_id)
+
+        # We assume each line is a keyframe
+        scenes_json.append({
+            "scene": scene_counter,
+            "scene_id": scene_id,
+            "keyframe_number": keyframe_counter,
+            "description": line,
+            "camera": "default",  # can be improved later
+            "seed": seed
+        })
+
+        keyframe_counter += 1
+        scene_counter += 1
+
+    return scenes_json
+
+# ---------------- MAIN FUNCTION ----------------
+async def generate_script_async(idea: str, user_confirmed: bool = True) -> List[Dict]:
+    """
+    Full pipeline for script generation:
+    1. Generates proposed script from LLM
+    2. Waits for user confirmation
+    3. Converts confirmed script into scene + keyframe JSON
+    """
+    prompt = f"Create a professional video script for: {idea}. Write each scene in one line."
+    raw_script = await _call_openrouter_llm(prompt)
+
+    # Here you can integrate actual user confirmation in your frontend
+    if not user_confirmed:
+        return [{"proposed_script": raw_script}]
+
+    # Convert approved script into structured scene/keyframe JSON
+    scenes = parse_script_to_scenes(raw_script)
+    return scenes
+
+def generate_script(idea: str, user_confirmed: bool = True) -> List[Dict]:
+    """
+    Synchronous wrapper for pipeline integration.
+    """
+    return asyncio.get_event_loop().run_until_complete(
+        generate_script_async(idea, user_confirmed)
+    )

core/seed_manager.py

@@ -0,0 +1,65 @@
+# core/seed_manager.py
+import json
+import random
+import threading
+from pathlib import Path
+
+DATA_DIR = Path(__file__).resolve().parent.parent / "data"
+SEED_FILE = DATA_DIR / "seeds.json"
+SEED_FILE.parent.mkdir(parents=True, exist_ok=True)
+
+_lock = threading.Lock()
+
+class SeedManager:
+    """
+    Persistent and thread-safe seed manager for reproducible image/video generation.
+    Maps: scene_id (str/int) -> seed (int)
+    """
+
+    def __init__(self, path: Path = SEED_FILE):
+        self.path = path
+        self._load()
+
+    def _load(self):
+        """Load seeds from disk safely."""
+        with _lock:
+            if self.path.exists():
+                try:
+                    with open(self.path, "r", encoding="utf-8") as f:
+                        self._store = json.load(f)
+                except Exception:
+                    self._store = {}
+            else:
+                self._store = {}
+
+    def _save(self):
+        """Save seeds to disk safely."""
+        with _lock:
+            with open(self.path, "w", encoding="utf-8") as f:
+                json.dump(self._store, f, indent=2)
+
+    def get_seed(self, scene_id: str | int) -> int | None:
+        """Return existing seed for a scene_id or None if not set."""
+        return self._store.get(str(scene_id))
+
+    def store_seed(self, scene_id: str | int, seed: int):
+        """Store a seed persistently for a given scene_id."""
+        self._store[str(scene_id)] = int(seed)
+        self._save()
+
+    def ensure_seed(self, scene_id: str | int) -> int:
+        """
+        Return existing seed or generate/store a new one.
+        Guarantees deterministic reproducible seed for the same scene_id.
+        """
+        s = self.get_seed(scene_id)
+        if s is None:
+            s = random.randint(0, 2**31 - 1)
+            self.store_seed(scene_id, s)
+        return s
+
+    def reset(self):
+        """Clear all seeds (useful for testing or new projects)."""
+        with _lock:
+            self._store = {}
+            self._save()

core/story_script.py

@@ -0,0 +1,593 @@
+# # # story_script.py
+# # import asyncio
+# # import json
+# # import logging
+# # import random
+
+# # logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+# # async def generate_story(script: str) -> dict:
+# #     """
+# #     Convert the ad script into a structured storyboard JSON format:
+# #     - Characters
+# #     - Scenes with keyframes, camera instructions, and music
+# #     """
+# #     # Sample character extraction (for simplicity, can be improved)
+# #     characters = [{"name": name, "seed": idx+1} for idx, name in enumerate(["MainGuy", "Friend1", "Friend2"])]
+
+# #     # Split script into lines and create scenes (basic heuristic, can be improved)
+# #     lines = [line.strip() for line in script.split("\n") if line.strip()]
+# #     scenes = {}
+# #     for idx, line in enumerate(lines, start=1):
+# #         char = characters[idx % len(characters)]["name"]
+# #         seed = characters[idx % len(characters)]["seed"]
+# #         scenes[f"scene{idx}"] = {
+# #             "character": char,
+# #             "scene": line,
+# #             "keyframes": [
+# #                 {
+# #                     "seed": seed,
+# #                     "keyframe1": f"{char} in action based on script line: '{line[:40]}...'",
+# #                     "keyframe2": f"{char} expressive close-up reacting to: '{line[:40]}...'"
+# #                 }
+# #             ],
+# #             "camera": "Medium shot with dynamic zoom-ins",
+# #             "music": "Appropriate upbeat or dramatic tune based on action"
+# #         }
+
+# #     storyboard = {
+# #         "characters": characters,
+# #         **scenes
+# #     }
+
+# #     logging.info("Story script generated successfully")
+# #     return storyboard
+
+
+# # # best
+# # # story_script.py
+# # import os
+# # import asyncio
+# # import httpx
+# # import logging
+# # from dotenv import load_dotenv
+# # from dotenv import load_dotenv
+
+# # dotenv_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '.env')
+# # load_dotenv(dotenv_path)
+
+# # load_dotenv()
+# # logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+# # OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
+# # MODEL_NAME = "deepseek/deepseek-r1-distill-llama-70b:free"
+# # OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
+
+# # async def generate_story(script: str) -> dict:
+# #     """
+# #     Convert the ad script into a structured storyboard JSON using AI.
+# #     The JSON format includes:
+# #     - characters (with seeds)
+# #     - scenes (character, scene description, keyframes, camera, music)
+# #     """
+# #     prompt = f"""
+# # You are a professional ad storyboard generator.
+# # Take this ad script and convert it into a **storyboard JSON**.
+# # Follow this format exactly:
+
+# # ADD {{
+# #   "characters": [
+# #     {{"name": "MainGuy", "seed": 1}},
+# #     {{"name": "Friend1", "seed": 2}},
+# #     {{"name": "Friend2", "seed": 3}}
+# #   ],
+# #   "scene1": {{
+# #     "character": "MainGuy",
+# #     "scene": "Description of the scene",
+# #     "keyframes": [
+# #       {{
+# #         "seed": 1,
+# #         "keyframe1": "First keyframe description",
+# #         "keyframe2": "Second keyframe description"
+# #       }}
+# #     ],
+# #     "camera": "Camera instructions",
+# #     "music": "Music instructions"
+# #   }}
+# #   ...
+# # }}
+
+# # Ensure:
+# # - Use the **script lines as scenes**.
+# # - Assign characters logically to actions.
+# # - Provide **keyframes, camera, and music**.
+# # - Return **valid JSON only**, no extra text.
+# # Script: 
+# # \"\"\"{script}\"\"\"
+# # """
+
+# #     headers = {"Authorization": f"Bearer {OPENROUTER_API_KEY}"}
+# #     payload = {
+# #         "model": MODEL_NAME,
+# #         "messages": [{"role": "user", "content": prompt}],
+# #         "temperature": 0.8,
+# #         "max_tokens": 1200
+# #     }
+
+# #     async with httpx.AsyncClient(timeout=120) as client:
+# #         response = await client.post(OPENROUTER_URL, json=payload, headers=headers)
+# #         response.raise_for_status()
+# #         data = response.json()
+
+# #     # AI returns the JSON as a string
+# #     story_json_str = data["choices"][0]["message"]["content"]
+
+# #     # Remove possible extra text before/after JSON (some AI outputs might wrap with "ADD {...}")
+# #     if story_json_str.startswith("ADD"):
+# #         story_json_str = story_json_str[story_json_str.find("{"):]
+
+# #     # Convert string to dict
+# #     try:
+# #         story_dict = eval(story_json_str)  # safe because AI returns JSON-like dict
+# #     except Exception as e:
+# #         logging.error(f"Failed to parse story JSON: {e}")
+# #         story_dict = {}
+
+# #     logging.info("Story script generated successfully using AI")
+# #     return story_dict
+
+
+
+
+# # import os
+# # import asyncio
+# # import httpx
+# # import logging
+# # import json
+# # from dotenv import load_dotenv
+
+# # dotenv_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '.env')
+# # load_dotenv(dotenv_path)
+
+# # logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+# # OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
+# # MODEL_NAME = "deepseek/deepseek-r1-distill-llama-70b:free"
+# # OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
+
+# # async def generate_story(script: str) -> dict:
+# #     """
+# #     Convert the ad script into a structured storyboard JSON using AI.
+# #     The JSON format includes:
+# #     - characters (with seeds)
+# #     - scenes (character, scene description, keyframes, camera, music)
+# #     """
+# #     prompt = f"""
+# # You are a professional ad storyboard generator.
+# # Take this ad script and convert it into a **storyboard JSON**.
+# # Follow this format exactly:
+
+# # ADD {{
+# #   "characters": [
+# #     {{"name": "MainGuy", "seed": 1}},
+# #     {{"name": "Friend1", "seed": 2}},
+# #     {{"name": "Friend2", "seed": 3}}
+# #   ],
+# #   "scene1": {{
+# #     "character": "MainGuy",
+# #     "scene": "Description of the scene",
+# #     "keyframes": [
+# #       {{
+# #         "seed": 1,
+# #         "keyframe1": "First keyframe description",
+# #         "keyframe2": "Second keyframe description"
+# #       }}
+# #     ],
+# #     "camera": "Camera instructions",
+# #     "music": "Music instructions"
+# #   }}
+# #   ...
+# # }}
+
+# # Ensure:
+# # - Use the **script lines as scenes**.
+# # - Assign characters logically to actions.
+# # - Provide **keyframes, camera, and music**.
+# # - Return **valid JSON only**, no extra text, no markdown, no ``` fences.
+# # Script: 
+# # \"\"\"{script}\"\"\"
+# # """
+
+# #     headers = {"Authorization": f"Bearer {OPENROUTER_API_KEY}"}
+# #     payload = {
+# #         "model": MODEL_NAME,
+# #         "messages": [{"role": "user", "content": prompt}],
+# #         "temperature": 0.8,
+# #         "max_tokens": 1500
+# #     }
+
+# #     async with httpx.AsyncClient(timeout=120) as client:
+# #         response = await client.post(OPENROUTER_URL, json=payload, headers=headers)
+# #         response.raise_for_status()
+# #         data = response.json()
+
+# #     story_json_str = data["choices"][0]["message"]["content"]
+
+# #     # Clean unwanted wrappers
+# #     story_json_str = story_json_str.strip()
+# #     if story_json_str.startswith("ADD"):
+# #         story_json_str = story_json_str[story_json_str.find("{"):]
+
+# #     # Remove markdown fences
+# #     story_json_str = story_json_str.replace("```json", "").replace("```", "").strip()
+
+# #     # Parse safely
+# #     try:
+# #         story_dict = json.loads(story_json_str)
+# #     except json.JSONDecodeError as e:
+# #         logging.error(f"❌ Failed to parse story JSON properly: {e}")
+# #         story_dict = {"raw_output": story_json_str}
+
+# #     logging.info("Story script generated successfully using AI")
+# #     return story_dict
+
+
+# import os
+# import asyncio
+# import httpx
+# import logging
+# import json
+# import re
+# from dotenv import load_dotenv
+
+# dotenv_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '.env')
+# load_dotenv(dotenv_path)
+
+# logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+# OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
+# MODEL_NAME = "deepseek/deepseek-r1-distill-llama-70b:free"
+# OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
+
+
+# async def generate_story(script: str) -> dict:
+#     """
+#     Convert the ad script into a structured storyboard JSON using AI.
+#     Returns:
+#     {
+#         "raw_output": "<original AI string>",
+#         "parsed_output": <dict | None>
+#     }
+#     """
+#     prompt = f"""
+# You are a professional ad storyboard generator.
+# Take this ad script and convert it into a **storyboard JSON**.
+# Follow this format exactly:
+
+# ADD {{
+#   "characters": [
+#     {{"name": "MainGuy-", "seed": 1}},
+#     {{"name": "Friend1", "seed": 2}},
+#     {{"name": "Friend2", "seed": 3}}
+#   ],
+#   "scene1": {{
+#     "character": "MainGuy",
+#     "scene": "Description of the scene",
+#     "keyframes": [
+#       {{
+#         "seed": 1,
+#         "keyframe1": "First keyframe description",
+#         "keyframe2": "Second keyframe description"
+#       }}
+#     ],
+#     "camera": "Camera instructions",
+#     "music": "Music instructions"
+#   }}
+# }}
+
+# Ensure:
+# - Each scene corresponds to a line in the script.
+# - Assign logical characters.
+# - Return only valid JSON (no markdown or explanations).
+# Script:
+# \"\"\"{script}\"\"\"
+# """
+
+#     headers = {"Authorization": f"Bearer {OPENROUTER_API_KEY}"}
+#     payload = {
+#         "model": MODEL_NAME,
+#         "messages": [{"role": "user", "content": prompt}],
+#         "temperature": 0.8,
+#         "max_tokens": 1500
+#     }
+
+#     async with httpx.AsyncClient(timeout=120) as client:
+#         response = await client.post(OPENROUTER_URL, json=payload, headers=headers)
+#         response.raise_for_status()
+#         data = response.json()
+
+#     story_json_str = data["choices"][0]["message"]["content"].strip()
+#     raw_output = story_json_str
+
+#     # --- Cleaning Stage ---
+#     story_json_str = story_json_str.strip()
+#     if story_json_str.startswith("ADD"):
+#         story_json_str = story_json_str[story_json_str.find("{"):]
+
+#     # Remove markdown code fences and artifacts
+#     story_json_str = story_json_str.replace("```json", "").replace("```", "").strip()
+
+#     # Remove unwanted triple quotes, trailing commas, and unescaped slashes
+#     story_json_str = re.sub(r',\s*}', '}', story_json_str)
+#     story_json_str = re.sub(r',\s*\]', ']', story_json_str)
+#     story_json_str = story_json_str.replace('\\"', '"').replace("\\'", "'")
+
+#     parsed_story = None
+
+#     # --- Parsing Stage ---
+#     try:
+#         parsed_story = json.loads(story_json_str)
+#     except json.JSONDecodeError:
+#         try:
+#             # Handle double-encoded or escaped JSON
+#             cleaned_str = bytes(story_json_str, "utf-8").decode("unicode_escape")
+#             parsed_story = json.loads(cleaned_str)
+#         except Exception as e:
+#             logging.error(f"❌ JSON parse failed after cleaning: {e}")
+#             parsed_story = None
+
+#     logging.info("✅ Storyboard generation completed")
+#     return parsed_story
+
+
+##best best
+# import os
+# import asyncio
+# import httpx
+# import logging
+# import json
+# import re
+# from dotenv import load_dotenv
+
+# dotenv_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '.env')
+# load_dotenv(dotenv_path)
+
+# logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+# OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
+# MODEL_NAME = "deepseek/deepseek-r1-distill-llama-70b:free"
+# OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
+
+
+# async def generate_story(script: str) -> dict:
+#     """
+#     Convert the ad script into a structured storyboard JSON using AI.
+#     Always returns:
+#     {
+#         "raw_output": "<original text from AI>",
+#         "parsed_output": <dict or None>
+#     }
+#     """
+#     prompt = f"""
+# You are a professional ad storyboard generator.
+# Convert this ad script into a **storyboard JSON** only — no extra text.
+
+# Format example:
+# {{
+#   "characters": [
+#     {{"name": "MainGuy","Description":"complete decsripiton of the character", "seed": 1}},
+#     {{"name": "Dog","Description":"complete decsripiton of the character", "seed": 2}}
+#   ],
+#   "scene1": {{
+#     "character": "MainGuy",
+#     "scene": "Man wakes up late and rushes outside",
+#     "keyframes": [
+#       {{"seed": 1, "keyframe1": "Man stepping in puddle", "keyframe2": "Reaction close-up"}}
+#     ],
+#     "camera": "Medium shot with soft lighting",
+#     "music": "Playful upbeat tune"
+#   }}
+# }}
+# Script:
+# \"\"\"{script}\"\"\"
+# Return **only valid JSON**, no markdown or commentary.
+# """
+
+#     headers = {"Authorization": f"Bearer {OPENROUTER_API_KEY}"}
+#     payload = {
+#         "model": MODEL_NAME,
+#         "messages": [{"role": "user", "content": prompt}],
+#         "temperature": 0.7,
+#         "max_tokens": 1500
+#     }
+
+#     async with httpx.AsyncClient(timeout=120) as client:
+#         response = await client.post(OPENROUTER_URL, json=payload, headers=headers)
+#         response.raise_for_status()
+#         data = response.json()
+
+#     story_json_str = data["choices"][0]["message"]["content"].strip()
+#     raw_output = story_json_str
+
+#     # --- Clean the output ---
+#     story_json_str = re.sub(r"^ADD\s*", "", story_json_str)
+#     story_json_str = story_json_str.replace("```json", "").replace("```", "")
+#     story_json_str = story_json_str.replace("“", "\"").replace("”", "\"").replace("’", "'")
+#     story_json_str = re.sub(r",\s*([}\]])", r"\1", story_json_str)  # remove trailing commas
+#     story_json_str = story_json_str.strip()
+
+#     # --- Parse the JSON safely ---
+#     parsed_story = None
+#     try:
+#         parsed_story = json.loads(story_json_str)
+#     except json.JSONDecodeError as e:
+#         logging.warning(f"JSON parse failed: {e}")
+#         try:
+#             # try to find JSON substring in case AI wrapped it with text
+#             match = re.search(r"\{.*\}", story_json_str, re.DOTALL)
+#             if match:
+#                 parsed_story = json.loads(match.group(0))
+#         except Exception as e2:
+#             logging.error(f"Final parsing failed: {e2}")
+#             parsed_story = None
+
+#     logging.info("✅ Storyboard generation completed")
+#     return parsed_story
+
+
+
+
+
+import os
+import asyncio
+import httpx
+import logging
+import json
+import re
+from dotenv import load_dotenv
+
+# --- Load environment variables ---
+dotenv_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '.env')
+load_dotenv(dotenv_path)
+
+# --- Configure logging ---
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+# --- API Constants ---
+OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
+MODEL_NAME = "deepseek/deepseek-r1-distill-llama-70b:free"
+OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
+
+
+async def generate_story(script: str) -> dict:
+    """
+    Converts ad script into a structured storyboard JSON.
+    Each scene = only one action.
+    Reuses character seeds for consistency.
+    Always returns a non-null dictionary.
+    """
+
+    prompt = f"""
+You are a professional storyboard generator for AI video production.
+
+Convert the given ad script into a JSON storyboard format. 
+Each scene must represent only ONE clear action or emotional beat. 
+Do not include any explanations, markdown, or text outside JSON.
+
+### STRICT JSON FORMAT:
+{{
+  "characters": [
+    {{"name": "Man","description":"average build, brown hair, casual outfit","seed":1}},
+    {{"name": "Librarian","description":"stern woman, cat-eye glasses, neat bun","seed":2}}
+  ],
+  "scene1": {{
+    "character": "Man",
+    "scene": "Man enters the library holding a bag of potato chips",
+    "keyframes": [
+      {{
+        "seed": 1,
+        "keyframe1": "Man walking into a quiet library, holding a bag of potato chips, warm sunlight from windows, calm mood",
+        "keyframe2": "Side shot of man sitting down at a wooden table, casual expression, sunlight glows behind him"
+      }}
+    ],
+    "camera": "Medium wide shot with soft natural lighting",
+    "music": "Gentle ambient tune"
+  }}
+}}
+
+Rules:
+- Each scene = ONE clear action only.
+- Use each character’s 'seed' consistently across scenes.
+- Each keyframe describes two cinematic angles of that action.
+- Keep descriptions detailed but realistic.
+- Return valid JSON only — no extra text, comments, or markdown.
+
+Script:
+\"\"\"{script}\"\"\"
+"""
+
+    headers = {"Authorization": f"Bearer {OPENROUTER_API_KEY}"}
+    payload = {
+        "model": MODEL_NAME,
+        "messages": [{"role": "user", "content": prompt}],
+        "temperature": 0.7,
+        "max_tokens": 1800
+    }
+
+    async with httpx.AsyncClient(timeout=180) as client:
+        try:
+            response = await client.post(OPENROUTER_URL, json=payload, headers=headers)
+            response.raise_for_status()
+            data = response.json()
+        except Exception as e:
+            logging.error(f"API request failed: {e}")
+            return {"error": "API request failed", "details": str(e)}
+
+    story_json_str = data["choices"][0]["message"]["content"].strip()
+
+    # --- Clean the model output ---
+    story_json_str = re.sub(r"```(?:json)?", "", story_json_str)
+    story_json_str = story_json_str.replace("“", "\"").replace("”", "\"").replace("’", "'")
+    story_json_str = re.sub(r",\s*([}\]])", r"\1", story_json_str).strip()
+
+    parsed_story = None
+    try:
+        parsed_story = json.loads(story_json_str)
+    except json.JSONDecodeError as e:
+        logging.warning(f"Initial JSON parse failed: {e}")
+        match = re.search(r"\{.*\}", story_json_str, re.DOTALL)
+        if match:
+            try:
+                parsed_story = json.loads(match.group(0))
+            except Exception as e2:
+                logging.error(f"Fallback parse failed: {e2}")
+
+    # --- Final fallback: minimal structure ---
+    if not parsed_story:
+        logging.warning("Model output invalid, generating fallback JSON.")
+        parsed_story = {
+            "characters": [],
+            "scene1": {
+                "character": "Unknown",
+                "scene": "Failed to parse script properly",
+                "keyframes": [
+                    {
+                        "seed": 0,
+                        "keyframe1": "Generic placeholder image",
+                        "keyframe2": "Generic placeholder image"
+                    }
+                ],
+                "camera": "Static fallback frame",
+                "music": "None"
+            }
+        }
+
+    # --- Ensure consistent seeds & single-action scenes ---
+    characters = {c.get("name"): c for c in parsed_story.get("characters", [])}
+    for key, scene in parsed_story.items():
+        if not key.startswith("scene"):
+            continue
+
+        char_name = scene.get("character")
+        seed = characters.get(char_name, {}).get("seed", 0)
+        desc = characters.get(char_name, {}).get("description", "")
+
+        # Limit to one clear action
+        scene_text = scene.get("scene", "")
+        scene["scene"] = re.split(r"[,.] and |, then |;| but | while ", scene_text)[0].strip()
+
+        # Keyframe corrections
+        for kf in scene.get("keyframes", []):
+            kf["seed"] = seed
+            for i, k in enumerate(["keyframe1", "keyframe2"]):
+                if not kf.get(k):
+                    angle = "wide shot" if i == 0 else "close-up"
+                    kf[k] = (
+                        f"{char_name} ({desc}) performing '{scene['scene']}', "
+                        f"{angle}, cinematic tone, photorealistic lighting"
+                    )
+
+    logging.info("✅ Storyboard generated successfully with consistent single-action scenes.")
+    return parsed_story
+
+

core/video_generator.py

@@ -0,0 +1,93 @@
+import os
+import torch
+from pathlib import Path
+from huggingface_hub import hf_hub_download
+from diffusers import AnimateDiffPipeline, MotionAdapter
+from typing import List
+from PIL import Image
+
+# ---------------- MODEL CONFIG ----------------
+MODEL_REPO = "ByteDance/AnimateDiff-Lightning"
+MODEL_FILENAME = "animatediff_lightning_8step_comfyui.safetensors"
+MODEL_DIR = Path("/tmp/models/animatediff_lightning")
+os.makedirs(MODEL_DIR, exist_ok=True)
+
+# ---------------- MODEL DOWNLOAD ----------------
+def download_model() -> Path:
+    model_path = MODEL_DIR / MODEL_FILENAME
+    if not model_path.exists():
+        print("[VideoGen] Downloading AnimateDiff Lightning 8-step...")
+        model_path = hf_hub_download(
+            repo_id=MODEL_REPO,
+            filename=MODEL_FILENAME,
+            local_dir=str(MODEL_DIR),
+            force_download=False,
+        )
+        print(f"[VideoGen] Model downloaded to: {model_path}")
+    else:
+        print("[VideoGen] AnimateDiff model already exists.")
+    return model_path
+
+# ---------------- PIPELINE LOAD ----------------
+def load_pipeline() -> AnimateDiffPipeline:
+    model_path = download_model()
+    print("[VideoGen] Loading AnimateDiff pipeline...")
+
+    adapter = MotionAdapter.from_single_file(str(model_path))
+    pipe = AnimateDiffPipeline.from_pretrained(
+        "emilianJR/epiCRealism",
+        motion_adapter=adapter,
+        torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
+    )
+
+    if torch.cuda.is_available():
+        pipe.to("cuda")
+
+    print("[VideoGen] AnimateDiff ready.")
+    return pipe
+
+# ---------------- GLOBAL PIPELINE CACHE ----------------
+pipe: AnimateDiffPipeline | None = None
+
+# ---------------- VIDEO GENERATION ----------------
+def generate_video(
+    keyframe_images: List[Image.Image],
+    seed: int = None,
+    num_frames: int = 16
+) -> List[Image.Image]:
+    """
+    Generates a short video by interpolating between input keyframe images.
+    
+    Args:
+        keyframe_images (List[PIL.Image]): List of PIL images representing keyframes.
+        seed (int, optional): Seed for deterministic generation.
+        num_frames (int): Total number of frames in the generated video.
+
+    Returns:
+        List[PIL.Image]: Interpolated video frames.
+    """
+    global pipe
+    if pipe is None:
+        pipe = load_pipeline()
+
+    if len(keyframe_images) < 2:
+        raise ValueError("At least 2 keyframe images are required to generate a video.")
+
+    print(f"[VideoGen] Generating video from {len(keyframe_images)} keyframes, {num_frames} frames, seed={seed}")
+
+    generator = None
+    if seed is not None:
+        device = "cuda" if torch.cuda.is_available() else "cpu"
+        generator = torch.Generator(device).manual_seed(seed)
+
+    # AnimateDiff expects init_images for interpolation between keyframes
+    video_frames = pipe(
+        init_images=keyframe_images,
+        num_frames=num_frames,
+        guidance_scale=1.0,
+        num_inference_steps=8,
+        generator=generator
+    ).frames
+
+    print("[VideoGen] Video generated successfully.")
+    return video_frames

pipeline/pipeline.py

@@ -0,0 +1,49 @@
+# pipeline.py
+import asyncio
+import logging
+from api.server import pending_confirmations  # Access the confirmation events
+
+# Import your modules
+import core.script_gen as script_gen
+import core.story_script as story_script
+# import core.image_gen as image_gen
+# import core.video_gen as video_gen
+# import core.music_gen as music_gen
+# import core.assemble as assemble
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s"
+)
+
+async def run_pipeline(task: dict):
+    task_id = task["task_id"]
+    idea = task["idea"]
+
+    logging.info(f"[Pipeline] Starting script generation for task {task_id}")
+    script = await script_gen.generate_script(idea)  # Async script generation
+
+    logging.info(f"[Pipeline] Waiting for user confirmation for task {task_id}")
+    # Wait for confirmation (manual or auto)
+    if task_id in pending_confirmations:
+        await pending_confirmations[task_id].wait()
+    else:
+        logging.info(f"[Pipeline] No pending confirmation found, auto-confirming task {task_id}")
+
+    logging.info(f"[Pipeline] Generating story script for task {task_id}")
+    story = await story_script.generate_story(script)
+    final_output = story  # Placeholder for final output
+    # logging.info(f"[Pipeline] Generating images for task {task_id}")
+    # images = await image_gen.generate_images(story)
+
+    # logging.info(f"[Pipeline] Generating video for task {task_id}")
+    # video = await video_gen.generate_video(images)
+
+    # logging.info(f"[Pipeline] Generating music/audio for task {task_id}")
+    # audio = await music_gen.generate_music(story)
+
+    # logging.info(f"[Pipeline] Assembling final output for task {task_id}")
+    # final_output = await assemble.create_final(video, audio)
+
+    logging.info(f"[Pipeline] Task {task_id} completed. Output: {final_output}")
+    return final_output

services/queue_manager.py

@@ -0,0 +1,401 @@
+# # services/queue_manager.py
+# import os
+# import uuid
+# import asyncio
+# import pickle
+# from collections import deque
+# from concurrent.futures import ThreadPoolExecutor
+# from typing import Callable, Any
+
+# # ---------------- CONFIG ----------------
+# RAM_QUEUE_LIMIT = 10
+# DISK_QUEUE_DIR = "task_queue_disk"
+# os.makedirs(DISK_QUEUE_DIR, exist_ok=True)
+
+# # ---------------- INTERNAL STORAGE ----------------
+# ram_queue = deque()
+# disk_queue_files = deque()
+# futures = {}
+# executor = ThreadPoolExecutor(max_workers=2)  # Can process 2 tasks concurrently
+# processing = False
+# lock = asyncio.Lock()
+
+
+# class QueueManager:
+#     def __init__(self):
+#         self.ram_queue = ram_queue
+#         self.disk_queue_files = disk_queue_files
+#         self.futures = futures
+#         self.executor = executor
+#         self.processing = False
+#         self.lock = lock
+
+#     async def _process_queue(self):
+#         """Process tasks one at a time in strict FIFO order."""
+#         async with self.lock:
+#             if self.processing:
+#                 return
+#             self.processing = True
+
+#         try:
+#             while self.ram_queue or self.disk_queue_files:
+#                 if not self.ram_queue and self.disk_queue_files:
+#                     # Move one task from disk to RAM
+#                     file_path = self.disk_queue_files.popleft()
+#                     with open(file_path, "rb") as f:
+#                         task_data = pickle.load(f)
+#                     os.remove(file_path)
+#                     self.ram_queue.append(task_data)
+#                     print(f"[Queue] Loaded task from disk to RAM.")
+
+#                 if not self.ram_queue:
+#                     break
+
+#                 task_id, func, args, kwargs = self.ram_queue.popleft()
+#                 print(f"[Queue] Processing task {task_id}...")
+
+#                 loop = asyncio.get_event_loop()
+#                 # Run in executor to avoid blocking event loop
+#                 result = await loop.run_in_executor(self.executor, func, *args, **kwargs)
+
+#                 # Set result in future
+#                 fut = self.futures.get(task_id)
+#                 if fut and not fut.done():
+#                     fut.set_result(result)
+#                     print(f"[Queue] Task {task_id} completed.")
+
+#         finally:
+#             self.processing = False
+
+#     def enqueue_task(self, func: Callable, *args, **kwargs) -> str:
+#         """
+#         Add a task to the queue.
+#         func: Callable function to execute.
+#         args, kwargs: Arguments for the function.
+#         Returns a unique task_id.
+#         """
+#         task_id = str(uuid.uuid4())
+#         loop = asyncio.get_event_loop()
+#         fut = loop.create_future()
+#         self.futures[task_id] = fut
+
+#         task_data = (task_id, func, args, kwargs)
+
+#         if len(self.ram_queue) < RAM_QUEUE_LIMIT:
+#             self.ram_queue.append(task_data)
+#             print(f"[Queue] Task {task_id} added to RAM queue.")
+#         else:
+#             file_path = os.path.join(DISK_QUEUE_DIR, f"{task_id}.pkl")
+#             with open(file_path, "wb") as f:
+#                 pickle.dump(task_data, f)
+#             self.disk_queue_files.append(file_path)
+#             print(f"[Queue] Task {task_id} saved to disk.")
+
+#         # Start processing
+#         loop.create_task(self._process_queue())
+#         return task_id
+
+#     def get_future(self, task_id: str):
+#         """Get the future object for a task_id."""
+#         return self.futures.get(task_id)
+
+#     def get_queue_status(self) -> dict:
+#         """Return current queue info."""
+#         total_q = len(self.ram_queue) + len(self.disk_queue_files)
+#         return {
+#             "status": "free" if total_q == 0 else "busy",
+#             "queue_length": total_q,
+#             "ram_queue": len(self.ram_queue),
+#             "disk_queue": len(self.disk_queue_files),
+#             "processing": self.processing
+#         }
+
+# # # Singleton instance for the pipeline.pipeline to use
+# # queue_manager = QueueManager()
+
+
+
+# # best
+# # queue.py
+# import asyncio
+# import uuid
+# from typing import Dict, Any, Optional
+# from enum import Enum
+
+# # === Import all stage functions ===
+# # queue_manager.py
+# # queue_manager.py inside core/
+# from core.script_gen import generate_script
+# from core.story_script import generate_story
+# # from core.image_gen import generate_images
+# # from core.video_gen import generate_video
+# # from core.music_gen import generate_music
+# # from core.assemble import assemble_final_video
+
+
+
+# # -------------------------------------------------------------
+# # ENUMS AND GLOBALS
+# # -------------------------------------------------------------
+# class TaskStatus(str, Enum):
+#     PENDING = "pending"
+#     RUNNING = "running"
+#     WAITING_CONFIRMATION = "waiting_for_confirmation"
+#     CONFIRMED = "confirmed"
+#     COMPLETED = "completed"
+#     FAILED = "failed"
+
+
+# # All active tasks are tracked here (in-memory)
+# tasks: Dict[str, Dict[str, Any]] = {}
+
+# # Async queue for orderly execution
+# task_queue = asyncio.Queue()
+
+
+# # -------------------------------------------------------------
+# # ADD NEW TASK
+# # -------------------------------------------------------------
+# async def add_task(idea: str) -> str:
+#     """
+#     Adds a new ad generation task to the queue and returns its task ID.
+#     """
+#     task_id = str(uuid.uuid4())
+
+#     tasks[task_id] = {
+#         "id": task_id,
+#         "idea": idea,
+#         "status": TaskStatus.PENDING,
+#         "result": None,
+#         "confirmation_required": False
+#     }
+
+#     await task_queue.put(task_id)
+#     print(f"🧩 Task added to queue: {task_id}")
+
+#     return task_id
+
+
+# # -------------------------------------------------------------
+# # MAIN WORKER LOOP
+# # -------------------------------------------------------------
+# async def worker():
+#     """
+#     Continuously consumes tasks from the queue one-by-one.
+#     Each task runs through all stages in sequence.
+#     """
+#     while True:
+#         task_id = await task_queue.get()
+#         task = tasks.get(task_id)
+#         if not task:
+#             task_queue.task_done()
+#             continue
+
+#         try:
+#             print(f"🚀 Starting task: {task_id}")
+#             task["status"] = TaskStatus.RUNNING
+
+#             # === STEP 1: Script generation ===
+#             script_result = await generate_script(task["idea"])
+#             task["result"] = {"script": script_result}
+#             task["status"] = TaskStatus.WAITING_CONFIRMATION
+#             task["confirmation_required"] = True
+
+#             print(f"✋ Task {task_id} waiting for confirmation after script stage.")
+
+#             # Wait until user confirms externally
+#             while task["status"] == TaskStatus.WAITING_CONFIRMATION:
+#                 await asyncio.sleep(1)
+
+#             # === STEP 2: Story generation ===
+#             if task["status"] == TaskStatus.CONFIRMED:
+#                 print(f"🎬 Task {task_id} confirmed. Continuing pipeline.pipeline...")
+
+#                 story_result = await generate_story(script_result)
+#                 # images = await generate_images(story_result)
+#                 # video = await generate_video(images)
+#                 # music = await generate_music(story_result)
+#                 # final_output = await assemble_final_video(video, music)
+
+#                 task["result"].update({
+#                     "story_script": story_result,
+#                     # "images": images,
+#                     # "video": video,
+#                     # "music": music,
+#                     # "final_output": final_output
+#                 })
+
+#                 task["status"] = TaskStatus.COMPLETED
+#                 print(f"✅ Task {task_id} completed successfully!")
+
+#         except Exception as e:
+#             task["status"] = TaskStatus.FAILED
+#             task["result"] = {"error": str(e)}
+#             print(f"❌ Task {task_id} failed with error: {e}")
+
+#         finally:
+#             task_queue.task_done()
+
+
+# # -------------------------------------------------------------
+# # CONFIRMATION HANDLER
+# # -------------------------------------------------------------
+# async def confirm_task(task_id: str) -> Dict[str, Any]:
+#     """
+#     Confirms a paused task and resumes the rest of the pipeline.pipeline.
+#     """
+#     task = tasks.get(task_id)
+#     if not task:
+#         return {"error": "Invalid task ID."}
+
+#     if task["status"] != TaskStatus.WAITING_CONFIRMATION:
+#         return {"error": "Task is not waiting for confirmation."}
+
+#     task["status"] = TaskStatus.CONFIRMED
+#     task["confirmation_required"] = False
+#     print(f"👍 Task {task_id} confirmed. Resuming pipeline.pipeline...")
+
+#     return {"message": f"Task {task_id} confirmed and resumed."}
+
+
+# # -------------------------------------------------------------
+# # STATUS CHECK
+# # -------------------------------------------------------------
+# def get_task_status(task_id: str) -> Optional[Dict[str, Any]]:
+#     """
+#     Returns the full details of a task including stage results.
+#     """
+#     return tasks.get(task_id)
+
+
+# # -------------------------------------------------------------
+# # START WORKER ON APP STARTUP
+# # -------------------------------------------------------------
+# def start_worker():
+#     """
+#     Starts the asynchronous background worker loop.
+#     Must be called once when FastAPI app starts.
+#     """
+#     loop = asyncio.get_event_loop()
+#     loop.create_task(worker())
+#     print("⚙️  Worker loop started.")
+
+
+
+import asyncio
+import uuid
+from typing import Dict, Any, Optional
+from enum import Enum
+from core.script_gen import generate_script
+from core.story_script import generate_story
+# from core.image_gen import generate_images
+# from core.video_gen import generate_video
+# from core.music_gen import generate_music
+# from core.assemble import assemble_final_video
+
+# -------------------------------------------------------------
+# ENUMS AND GLOBALS
+# -------------------------------------------------------------
+class TaskStatus(str, Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    WAITING_CONFIRMATION = "waiting_for_confirmation"
+    CONFIRMED = "confirmed"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+tasks: Dict[str, Dict[str, Any]] = {}
+task_queue = asyncio.Queue()
+
+# -------------------------------------------------------------
+# ADD NEW TASK
+# -------------------------------------------------------------
+async def add_task(idea: str) -> str:
+    task_id = str(uuid.uuid4())
+    tasks[task_id] = {
+        "id": task_id,
+        "idea": idea,
+        "status": TaskStatus.PENDING,
+        "result": None,
+        "confirmation_required": False
+    }
+    await task_queue.put(task_id)
+    print(f"🧩 Task added to queue: {task_id}")
+    return task_id
+
+# -------------------------------------------------------------
+# WAIT FOR SCRIPT FOR CONFIRMATION
+# -------------------------------------------------------------
+async def wait_for_script(task_id: str, script_results: dict):
+    task = tasks.get(task_id)
+    if not task:
+        return
+
+    task["status"] = TaskStatus.RUNNING
+    # Generate script
+    script_result = await generate_script(task["idea"])
+    task["result"] = {"script": script_result}
+    task["status"] = TaskStatus.WAITING_CONFIRMATION
+    task["confirmation_required"] = True
+
+    # Keep script accessible for server endpoint
+    script_results[task_id] = script_result
+    print(f"✋ Task {task_id} waiting for confirmation. Script ready.")
+
+# -------------------------------------------------------------
+# GENERATE STORY AFTER CONFIRMATION
+# -------------------------------------------------------------
+async def generate_story_after_confirm(script: str):
+    story_result = await generate_story(script)
+    return story_result
+
+# -------------------------------------------------------------
+# WORKER LOOP (Optional: future stages)
+# -------------------------------------------------------------
+async def worker():
+    while True:
+        task_id = await task_queue.get()
+        task = tasks.get(task_id)
+        if not task:
+            task_queue.task_done()
+            continue
+        try:
+            # Already handled script in wait_for_script
+            # Future stages like images/video/music can go here
+            pass
+        except Exception as e:
+            task["status"] = TaskStatus.FAILED
+            task["result"] = {"error": str(e)}
+            print(f"❌ Task {task_id} failed with error: {e}")
+        finally:
+            task_queue.task_done()
+
+# -------------------------------------------------------------
+# CONFIRM TASK
+# -------------------------------------------------------------
+async def confirm_task(task_id: str):
+    task = tasks.get(task_id)
+    if not task:
+        return {"error": "Invalid task ID."}
+
+    if task["status"] != TaskStatus.WAITING_CONFIRMATION:
+        return {"error": "Task is not waiting for confirmation."}
+
+    task["status"] = TaskStatus.CONFIRMED
+    task["confirmation_required"] = False
+    print(f"👍 Task {task_id} confirmed. Ready for story generation...")
+    return {"message": f"Task {task_id} confirmed."}
+
+# -------------------------------------------------------------
+# GET TASK STATUS
+# -------------------------------------------------------------
+def get_task_status(task_id: str) -> Optional[Dict[str, Any]]:
+    return tasks.get(task_id)
+
+# -------------------------------------------------------------
+# START WORKER
+# -------------------------------------------------------------
+def start_worker():
+    loop = asyncio.get_event_loop()
+    loop.create_task(worker())
+    print("⚙️ Worker loop started.")