Amalfa_Creative_Studio / backend /app /analysis_flows.py
sushilideaclan01's picture
added a new flow
4c69ac5
"""
Alternative creative analysis flow: Cross-vertical inspiration.
Uses reference images (creativity examples) and returns ad creatives JSON directly,
ready for image generation with product reference images.
"""
import base64
import json
import logging
import mimetypes
import os
import random
from pathlib import Path
from typing import Optional
from app.llm import call_llm_vision, extract_json
log = logging.getLogger("uvicorn.error")
# Structured output schema (all flows). strict=False avoids additionalProperties requirement on nested objects.
CREATIVE_OUTPUT_JSON_SCHEMA = {
"type": "json_schema",
"json_schema": {
"name": "creative_output",
"strict": False,
"schema": {
"type": "object",
"properties": {
"creatives": {
"type": "array",
"minItems": 50,
"maxItems": 50,
"items": {
"type": "object",
"properties": {
"creative_number": {"type": "integer"},
"archetype": {"type": "string"},
"visual_strategy": {
"type": "object",
"properties": {
"scene_prompt": {"type": "string"},
"shooting_angle": {"type": "string"},
"color_world": {"type": "string"},
"creative_type": {"type": "string"},
"best_platform": {"type": "string"},
},
"required": ["scene_prompt"],
"additionalProperties": False,
},
"text_on_image": {
"type": "object",
"properties": {
"headline_serif": {"type": "string"},
"headline_script": {"type": "string"},
"product_name": {"type": "string"},
"body": {"type": "string"},
"cta": {"type": "string"},
"price_original": {"type": "string"},
"price_final": {"type": "string"},
},
"required": ["headline_serif", "body", "cta"],
"additionalProperties": False,
},
"title": {"type": "string"},
},
"required": ["creative_number", "archetype", "visual_strategy", "text_on_image", "title"],
"additionalProperties": False,
},
},
},
"required": ["creatives"],
"additionalProperties": False,
},
},
}
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp"}
def _file_to_data_url(path: Path) -> str:
"""Convert local image file to data URL for vision API."""
mime, _ = mimetypes.guess_type(str(path))
if mime is None:
mime = "image/jpeg"
b64 = base64.b64encode(path.read_bytes()).decode("utf-8")
return f"data:{mime};base64,{b64}"
def _pick_random_images(folder: str, k: int = 5, seed: Optional[int] = None) -> list[Path]:
"""Pick k random image files from folder. Raises ValueError if folder empty."""
path = Path(folder)
if not path.is_dir():
return []
files = [p for p in path.iterdir() if p.is_file() and p.suffix.lower() in IMAGE_EXTS]
if not files:
return []
rng = random.Random(seed)
k = min(k, len(files))
return rng.sample(files, k)
def _get_creativity_examples_dir() -> str:
"""Get creativity examples dir from env, with fallback to backend/data/creativity_examples."""
creativity_dir = os.environ.get("CREATIVITY_EXAMPLES_DIR", "").strip()
if creativity_dir:
return creativity_dir
# Fallback: backend/data/creativity_examples relative to this file
fallback = Path(__file__).resolve().parent.parent / "data" / "creativity_examples"
return str(fallback) if fallback.is_dir() else ""
def _get_reference_image_urls(image_urls_override: list[str] | None = None) -> list[str]:
"""
Get reference images for Cross-vertical and Archetype analysis.
Uses CREATIVITY_EXAMPLES_DIR only (creative inspiration). Product images
are NOT used here—they are used only for image generation (Replicate).
"""
if image_urls_override is not None and len(image_urls_override) > 0:
return image_urls_override[:5]
creativity_dir = _get_creativity_examples_dir()
if creativity_dir:
paths = _pick_random_images(creativity_dir, k=5, seed=None)
if paths:
urls = [_file_to_data_url(p) for p in paths]
log.info("creativity_examples: using %d images from %s", len(urls), creativity_dir)
return urls
log.info("creativity_examples: no images found in %s (folder empty or no supported formats)", creativity_dir)
return []
def analyze_product_cross_vertical(
product_data: dict,
target_audience: list[str] | None = None,
image_urls: list[str] | None = None,
) -> dict:
"""
Deep marketing analysis with cross-vertical inspiration.
Returns 50 ad creatives as JSON directly, ready for image generation with product references.
"""
category = product_data.get("category") or "Necklace"
product_name = product_data.get("product_name", "")
price = product_data.get("price", "")
ref_urls = _get_reference_image_urls(image_urls)
log.info("cross_vertical: starting analysis, category=%s, image_count=%d", category, len(ref_urls))
if ref_urls:
log.info("creativity_examples: used for cross_vertical flow (dir=%s)", _get_creativity_examples_dir() or "override")
audience_block = ""
if target_audience and len(target_audience) > 0:
audience_list = ", ".join(target_audience)
audience_block = f"""
TARGET AUDIENCES (tailor strategies to these segments):
{audience_list}
"""
prompt = f"""you are an elite creative person with 190+ IQ
Your creativity think ONLY in:
- Scroll stopping power
- Emotional tension
- Identity signaling
- Conversion psychology
- Visual hierarchy
Category: {category}
##Creative Direction Note:
While the product falls under the jewelry category, our creative thinking should not be limited to this niche. The fact that it's jewelry is just context — not a constraint.
We want ideas that break category boundaries. Pull inspiration from completely different industries, emotions, cultures, psychology, storytelling, status symbols, rituals, movements, or even abstract concepts.
Think beyond what's typically done in jewelry advertising. Explore unexpected angles, fresh narratives, and unconventional hooks that wouldn't normally be associated with necklaces.
The goal is to create concepts so unique and imaginative that they feel impossible to generate through conventional thinking.
{audience_block}
━━━━━━━━━━━━━━━━━━
OBJECTIVE
━━━━━━━━━━━━━━━━━━
Generate EXACTLY 50 high-converting ad creatives as JSON.
Affiliate marketing is performance-based. Low-production, realistic visuals preferred.
Each creative: ONE dominant motivator, mobile-first, strong contrast. Avoid futuristic/tech, clutter, fake authority.
━━━━━━━━━━━━━━━━━━
CREATIVE RULES
━━━━━━━━━━━━━━━━━━
Do NOT use "elegance" or "elegant" in any creative (titles, headlines, body, etc.). Use other descriptive terms.
Each creative must:
- Focus on ONE dominant motivator only
- Be mobile-first
- Use strong contrast
- Avoid clutter
- Avoid fake authority
- Avoid fabricated claims
- Avoid impersonation
- Avoid unrealistic guarantees
- Avoid abstract sci-fi or unrealistic imagery
- Feel photographable in real life
Increase emotional intensity beyond typical jewelry ads.
Avoid soft, safe romance tones.
Lean into tension, declaration, or transformation.
Reject weak, safe, or predictable ideas.
If a concept feels like a typical jewelry ad, discard it and replace it with a stronger one.
Provide concise, execution-ready output.
No explanations.
No reasoning.
No fluff.
━━━━━━━━━━━━━━━━━━
PRODUCT DATA (use for scene_prompt—describe size, shape, material accurately for image generation)
━━━━━━━━━━━━━━━━━━
{json.dumps(product_data, indent=2)}
━━━━━━━━━━━━━━━━━━
JEWELRY PLACEMENT (MANDATORY for PRODUCT creatives)
━━━━━━━━━━━━━━━━━━
In every scene_prompt you MUST specify WHERE the jewelry appears in the frame, in a way that is logical for the product type and the concept:
- Necklace / pendant / chain: worn at collarbone/neck, or laid flat on surface for flat lay; never floating or in a random spot.
- Earrings: on the model's ears, or placed on a surface for flat lay; never floating.
- Bracelet / bangle: on wrist, or on surface for flat lay.
- Ring: on finger, or centered on surface for flat lay.
- Any product: lifestyle shot = worn on body in the correct position; flat lay = product on surface, centered or composed clearly; close-up = product fills frame in a natural way (e.g. neck for necklace, ear for earrings).
Placement must match the shot type and concept. The image should look like a real photo: jewelry in a believable, on-concept position.
Do NOT leave placement vague (e.g. "jewelry in the scene"). Always state the exact placement (e.g. "necklace at collarbone", "earrings on model's ears", "bracelet on wrist", "ring on finger", "pendant centered on marble surface").
━━━━━━━━━━━━━━━━━━
OUTPUT SCHEMA (structured JSON)
━━━━━━━━━━━━━━━━━━
Return 50 creatives. Each creative has:
- creative_number: 1–50
- archetype: the creative archetype or concept type
- title: short concept name
- visual_strategy: scene_prompt (describe background, lighting, framing AND explicit logical placement of the jewelry as above; end with: Ensure the jewelry is the dominant focal point in sharp focus. Preserve the product exactly as shown. Square 1:1, 1080x1080px.), shooting_angle, color_world, creative_type (PRODUCT or NO_PRODUCT), best_platform
- text_on_image: headline_serif (UPPERCASE), headline_script, body (1–2 lines), cta, price_original/price_final where relevant (use ₹ format, product "{product_name}", price "{price}")
Include price in exactly 17 creatives. Most creatives should be PRODUCT; up to 8 NO_PRODUCT.
"""
content: list[dict] = [{"type": "text", "text": prompt}]
for url in ref_urls:
content.append({"type": "image_url", "image_url": {"url": url}})
messages = [{"role": "user", "content": content}]
raw = call_llm_vision(
messages=messages,
model="gpt-4o",
temperature=0.85,
response_format=CREATIVE_OUTPUT_JSON_SCHEMA,
)
result = raw or ""
if _is_llm_refusal(result) and ref_urls:
log.warning("cross_vertical: LLM refused with images, retrying without reference images")
content = [{"type": "text", "text": prompt}]
raw = call_llm_vision(
messages=[{"role": "user", "content": content}],
model="gpt-4o",
temperature=0.85,
response_format=CREATIVE_OUTPUT_JSON_SCHEMA,
)
result = raw or ""
if _is_llm_refusal(result):
log.warning("cross_vertical: LLM returned refusal")
raise ValueError(
"Creative analysis was blocked. Try with different images in CREATIVITY_EXAMPLES_DIR, "
"or ensure the folder contains only ad/creative inspiration images."
)
try:
out = json.loads(result)
except json.JSONDecodeError:
out = extract_json(result)
creatives = out.get("creatives", [])
log.info("cross_vertical: analysis complete, creatives_count=%d", len(creatives))
return {"creatives": creatives}
def _is_llm_refusal(text: str) -> bool:
"""Detect if the LLM returned a refusal/cannot-assist response."""
if not text or len(text) < 20:
return False
lower = text.lower().strip()
refusal_phrases = (
"i'm sorry",
"i am sorry",
"i can't assist",
"i cannot assist",
"i'm unable",
"i am unable",
"i can't help",
"i cannot help",
"i'm not able",
"as an ai",
"as a language model",
"i don't have the ability",
)
return any(p in lower for p in refusal_phrases) and len(text) < 500