admaker / core /story_script.py
karthikeya1212's picture
Update core/story_script.py
82f04d1 verified
raw
history blame
20 kB
# # # 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.")
print(parsed_story)
return parsed_story