from dotenv import load_dotenv from openai import AzureOpenAI import json import os import requests from typing import List, Optional from pydantic import BaseModel, Field, ValidationError, field_validator from src.marketing.vector_db_collection.controller import search_handler from src.marketing.vector_db_collection.schema import SearchRequest import re, json import duckdb from datetime import datetime import time from PIL import Image from io import BytesIO load_dotenv(override=True) import base64 API_KEY = os.getenv("AZURE_OPENAI_API_KEY") ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT") DEPLOYMENT = os.getenv("AZURE_OPENAI_DEPLOYMENT") API_VERSION = os.getenv("AZURE_OPENAI_API_VERSION_CHAT") IDEOGRAM_URL = os.getenv("IDEOGRAM_URL") IDEOGRAM_API_KEY = os.getenv("IDEOGRAM_API_KEY") BASE_URL = os.getenv("BASE_URL", "https://hubwallet-dev.dreamztesting.com/") AZURE_ENDPOINT_IMAGE = os.getenv("AZURE_OPENAI_ENDPOINT_IMAGE") DEPLOYMENT_NAME_IMAGE = os.getenv("AZURE_OPENAI_DEPLOYMENT_IMAGE") API_VERSION_IMAGE = os.getenv("AZURE_OPENAI_API_VERSION_IMAGE") AZURE_API_KEY_IMAGE= os.getenv("AZURE_OPENAI_API_KEY_IMAGE") # Get the root directory of your project dynamically BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) # Construct the static path based on the root directory STATIC_PATH = os.path.join(BASE_DIR, "menu_design", "designer", "static", "generated_images") # Ensure the directory exists os.makedirs(STATIC_PATH, exist_ok=True) # The usual start client = AzureOpenAI( api_key=API_KEY, api_version="2024-08-01-preview", azure_endpoint=ENDPOINT, azure_deployment=DEPLOYMENT ) # ---- 1) Schema ---- class PromoEntities(BaseModel): promotion_type: Optional[str] = Field( default=None, description="High-level type like 'BOGO', 'Flat Discount', 'Cashback', 'Bundle', 'Clearance', 'Seasonal', etc." ) products: Optional[str] = Field(default=None, description="Product or category names mentioned.") offer_details: Optional[str] = Field( default=None, description="Concrete offer phrasing (e.g., 'Buy 2 get 1 free', '20% off orders over $99, code SAVE20')." ) style: Optional[str] = Field( default=None, description="Visual/voice style cues (e.g., 'minimal', 'bold', 'modern', 'retro', 'luxury')." ) color_theme: Optional[str] = Field( default=None, description="Colors or palettes mentioned or implied (e.g., 'red & gold', 'pastel green')." ) # ---- 1) Schema ---- class PromoCopy(BaseModel): title: Optional[str] = Field( default=None, description="Weekend Sale!, Burger Bliss" ) tagline: Optional[str] = Field( default=None, description="SAVE20, Save Now, Order Now" ) cta: Optional[str] = Field( default=None, description="Call to action for the campaign" ) discount: Optional[str] = Field( default=None, description="Discount details (e.g., '20% off', 'Buy 1 Get 1')." ) class ChatRequest(BaseModel): message: str = Field( ..., description="The user's message to the chat." ) session_id: str = Field( ..., description="The unique identifier for the chat session." ) class Template(BaseModel): template_id: str template_name: str template_url:str class ChatState(BaseModel): history: List[dict] = Field( default_factory=list, description="Chat history with user messages and assistant responses." ) missing_entities: List[str] = Field( default_factory=list, description="List of missing entities." ) entities: Optional[dict] = Field( default_factory=dict, description="Extracted promotional entities." ) entities_captured: bool = Field( default=False, description="Indicates whether all entities have been captured." ) intent: Optional[str] = Field( default=None, description="Current intent" ) intents: List[str] = Field( default_factory=list, description="List of detected intents." ) copy: Optional[dict] = Field( default_factory=dict, description="Generated promotional copy." ) image_url: Optional[str] = Field( default=None, description="Generated image URL." ) template: Optional[list[dict]] = Field(default_factory=list) state: Optional[dict[str,bool]] = Field( default=None, description="Current state of the chat." ) # @field_validator("template", mode="before") # def coerce_items(cls, v): # if v in (None, "", (), []): # return [] # if isinstance(v, dict): # return [v] # return v class MenuDesignGenerator: def __init__(self, session_id: str, duckdb_path: str = "menu_state.duckdb"): # self.session = {} # self.session[session_id] = ChatState() # --- DuckDB setup --- self._con = duckdb.connect(duckdb_path) # Enable JSON functions (safe to call repeatedly) self._con.execute("INSTALL json; LOAD json;") # Create table if it doesn't exist self._con.execute(""" CREATE TABLE IF NOT EXISTS chat_state ( session_id TEXT PRIMARY KEY, state JSON, updated_at TIMESTAMP DEFAULT now() ) """) # --- in-memory cache (optional) --- self.session = {} # Try to load existing state for this session_id; else create fresh existing = self._load_state(session_id) if existing is None: self.session[session_id] = ChatState() self._save_state(session_id, self.session[session_id]) else: self.session[session_id] = existing self._session_id = session_id # -------- DuckDB helpers -------- def _save_state(self, session_id: str, state: ChatState) -> None: """Upsert the ChatState for a session into DuckDB.""" state_json = state.model_dump_json() # Use a simple DELETE+INSERT to emulate upsert (portable across DuckDB versions) self._con.execute("BEGIN") try: self._con.execute("DELETE FROM chat_state WHERE session_id = ?", [session_id]) self._con.execute( "INSERT INTO chat_state (session_id, state, updated_at) VALUES (?, ?, ?)", [session_id, state_json, datetime.utcnow()] ) self._con.execute("COMMIT") except Exception: self._con.execute("ROLLBACK") raise def _load_state(self, session_id: str) -> Optional[ChatState]: """Fetch ChatState by session_id from DuckDB; return None if not found.""" row = self._con.execute( "SELECT state FROM chat_state WHERE session_id = ?", [session_id] ).fetchone() if not row: return None # row[0] is a JSON string (DuckDB returns TEXT for JSON); rehydrate Pydantic return ChatState.model_validate_json(row[0]) def _persist(self): """Persist the current session’s state.""" self._save_state(self._session_id, self.session[self._session_id]) def classify_intent_from_history( self, history ) -> str: """ Classify the user's intent from a chat history and return one of: - general - input_collection - input_confirmation - copy_generation - copy_regeneration - copy_confirmation - image_generation - image_regeneration - image_confirmation - template_search Parameters ---------- history : str | list[str] | list[dict] Conversation so far. Accepts: - a single string (latest user message or entire transcript), - a list of strings (ordered turns), - a list of {'role': 'user'|'assistant'|'system', 'content': '...'} dicts. model : str Chat model name (OpenAI or Azure-compatible). api_key_env : str Env var name holding the API key. Returns ------- str One of the intents above. If parsing fails, defaults to 'input_collection'. """ # ---- normalize history into a readable transcript for the model ---- if isinstance(history, str): transcript = history.strip() elif isinstance(history, list): if history and isinstance(history[0], dict) and "content" in history[0]: lines = [] for m in history: role = m.get("role", "user").lower() lines.append(f"{role}: {m.get('content','').strip()}") transcript = "\n".join(lines) else: transcript = "\n".join([str(x).strip() for x in history]) else: transcript = str(history) # ---- allowed label set & JSON schema to force a valid enum ---- labels = [ "general", "input_collection", "input_confirmation", "copy_generation", "copy_regeneration", "copy_confirmation", "image_generation", "image_regeneration", "image_confirmation", "template_search" ] schema = { "name": "Intent", "schema": { "type": "object", "additionalProperties": False, "properties": { "intent": {"type": "string", "enum": labels}, "rationale": {"type": "string"}, }, "required": ["intent"], }, } # ---- system guidance + compact rubric ---- sys = ( "You are an intent classifier for a creative assistant. " "Read the conversation history and output JSON with the single key 'intent'. " "Choose only from:\n" "general = general conversation or inquiries.\n" "input_collection = user is supplying requirements or asking what info you need.\n" "input_confirmation = user confirms their inputs/brief are final.\n" "copy_generation = user asks to create new wording/text/copy.\n" "copy_regeneration = user asks to retry/rewrite/modify existing copy.\n" "copy_confirmation = user approves/accepts the copy as final.\n" "image_generation = user asks to create a new image/visual.\n" "image_regeneration = user asks to modify/regenerate an image just made.\n" "image_confirmation = user approves/accepts the image as final.\n" "If ambiguous, pick the best single label." ) # ---- few-shot coverage (very compact) ---- fewshots = [ ("user: We’re launching a monsoon sale for sandals. What details do you need from me?\nassistant: I can collect brand, tone, products, colors.", {"intent": "input_collection", "rationale": "User asks what info to provide."}), ("user: Okay, that’s all the info. Proceed with this brief.\nassistant: Noted.", {"intent": "input_confirmation", "rationale": "User confirms inputs are final."}), ("user: Write a Facebook post announcing 20% off backpacks in a cheerful tone.", {"intent": "copy_generation", "rationale": "Direct request to create copy."}), ("user: Rewrite that caption, make it shorter and punchier.", {"intent": "copy_regeneration", "rationale": "User wants a revision of existing copy."}), ("user: Perfect, keep that caption as final.", {"intent": "copy_confirmation", "rationale": "User approves copy."}), ("user: Generate a square food background with no text.", {"intent": "image_generation", "rationale": "New image request."}), ("user: Try the same image but brighter colors and more contrast.", {"intent": "image_regeneration", "rationale": "Modify existing image."}), ("user: Looks great—export that image as final.", {"intent": "image_confirmation", "rationale": "User approves image."}), ("user: Hi! how are you?", {"intent": "general", "rationale": "User initiates general conversation."}), ("user: Find me some templates", {"intent": "template_search", "rationale": "User asks to find templates."}), ("user: show templates", {"intent": "template_search", "rationale": "User asks to find templates."}), ("user: generate some templates", {"intent": "template_search", "rationale": "User asks to find templates."}), ] # ---- build messages ---- messages = [{"role": "system", "content": sys}] for u, a in fewshots: messages.append({"role": "user", "content": f"History:\n{u}\n\nReturn JSON."}) messages.append({"role": "assistant", "content": json.dumps(a, ensure_ascii=False)}) messages.append({"role": "user", "content": f"History:\n{transcript}\n\nReturn JSON."}) # for msg in history: # messages.append({"role":msg['role'],"content":msg['content']}) # ---- call model ---- try: resp = client.chat.completions.create( model="gpt0", temperature=0, response_format={"type": "json_schema", "json_schema": schema}, messages=messages, ) content = resp.choices[0].message.content except Exception as e: # Fallback: try basic json_object format resp = client.chat.completions.create( model="gpt0", temperature=0, response_format={"type": "json_object"}, messages=messages, ) content = resp.choices[0].message.content # ---- parse robustly & return ---- try: data = json.loads(content) except json.JSONDecodeError: # strip accidental code fences content = re.sub(r"^```json|```$", "", content.strip(), flags=re.IGNORECASE | re.MULTILINE) data = json.loads(content) intent = str(data.get("intent", "")).strip() return intent if intent in labels else "input_collection" def _reply_markdown( self, message: str ) -> str: """ Take a rough bot reply and return a polished version in Markdown. - Tone: professional, friendly, concise - Output: Markdown only (no code fences) """ system_prompt = ( "You are an expert promotional-material assistant. " "Rewrite the user's draft reply so it is professional, friendly, and concise. " "Preserve intent and factual details, improve clarity, and format as Markdown. " "Use short sentences, clear structure, and bullets only when they help. " "Return ONLY the improved reply in Markdown—no code fences, no extra commentary." ) user_prompt = f"Draft reply to improve:\n\n{message}" resp = client.chat.completions.create( model="gpt0", temperature=0.2, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], ) out = resp.choices[0].message.content.strip() # Strip accidental code fences if the model adds them out = re.sub(r"^```[a-zA-Z]*\s*", "", out) out = re.sub(r"\s*```$", "", out) return out def answer_user_from_history( self, history, ): """ Produce the next assistant reply for a promo-material workflow, based on conversation history. Returns a structured JSON dict with: - assistant_reply: concise text to show the user - next_intent: one of [input_collection, input_confirmation, copy_generation, copy_regeneration, copy_confirmation, image_generation, image_regeneration, image_confirmation] - state_write: key-value updates the app should persist (inputs/preferences) - copy: a dictionary containing (title, tagline, cta, offer details) all in max 2 words - image_requests: list of image generation prompts (each: {purpose, prompt, size, variations}) - template_suggestions: optional list of templates {name, rationale} - missing_inputs: array of strings describing remaining info needed `history` can be: - a single string (transcript or latest user message), - a list[str] (ordered turns), - a list[{'role': 'user'|'assistant'|'system', 'content': '...'}]. The function does NOT create images; it returns prompts your app can pass to an image generator. """ import os, json, re from openai import OpenAI # ---- Normalize history to a readable transcript ---- if isinstance(history, str): transcript = history.strip() elif isinstance(history, list): if history and isinstance(history[0], dict) and "content" in history[0]: lines = [] for m in history: role = m.get("role", "user").lower() lines.append(f"{role}: {m.get('content','').strip()}") transcript = "\n".join(lines) else: transcript = "\n".join([str(x).strip() for x in history]) else: transcript = str(history) # ---- System prompt (user-provided) + tight response rules ---- system_prompt = ( "You are an expert promotional material designer that helps users create promotional materials by " "guiding them through a structured, step-by-step process, maintaining state and context, and facilitating " "inputs, copywriting, image generation, and template selection. Handle users’ requests and changes " "efficiently, ensuring every step is completed before moving forward.\n\n" "Key Guidelines:\n" "- Maintain State: Keep track of all inputs and user preferences. Handle users changing their minds, revisiting stages, " " or providing inputs out of order.\n" "- Conversational Tone: Keep responses concise, conversational, and focused while staying on task.\n" "- Critical Task Completion: Always complete promised tasks (e.g., generating copy, images) in the same response.\n" "- Adaptive Engagement: Ask for clarifications if intent is unclear. Detect and adapt to whether a user is providing " " inputs, modifying previous data, or requesting suggestions.\n\n" "Return STRICT JSON only with the schema provided. No markdown, no extra commentary." ) # ---- Output JSON schema (contract for your app) ---- intents = [ "input_collection", "input_confirmation", "copy_generation", "copy_regeneration", "copy_confirmation", "image_generation", "image_regeneration", "image_confirmation", ] json_schema = { "name": "PromoAssistantReply", "schema": { "type": "object", "additionalProperties": False, "properties": { "assistant_reply": {"type": "string"}, "next_intent": {"type": "string", "enum": intents}, "state_write": { "type": "object", "additionalProperties": True, "properties": { "brand": {"type": ["string","null"]}, "promotion_type": {"type": ["string","null"]}, "products": {"type": "array", "items": {"type": "string"}}, "offer_details": {"type": ["string","null"]}, "style": {"type": "array", "items": {"type": "string"}}, "color_theme": {"type": "array", "items": {"type": "string"}}, "tone": {"type": ["string","null"]}, "channel": {"type": ["string","null"]}, "size_preferences": {"type": ["string","null"]}, "notes": {"type": ["string","null"]} } }, "copy": { "type": "object", "items": { "type": "object", "additionalProperties": False, "properties": { "label": {"type": "string"}, "text": {"type": "string"} }, "required": ["label","text"] } }, }, "required": ["assistant_reply", "next_intent", "state_write", "copy", "image_requests", "template_suggestions", "missing_inputs"] } } # ---- Instruction to the model on how to act (rubric + mini playbook) ---- rubric = ( "You will read the conversation history and produce the next response.\n" "Rules:\n" "1) Be concise and conversational in 'assistant_reply'.\n" "2) If the user asked for copy, generate 2–4 short, high-quality options in 'copy'.\n" "3) If the user asked for images, include at least one 'image_requests' entry with a clear prompt.\n" " Prefer square 1080x1080 for Facebook/Instagram unless user specified otherwise.\n" "4) Update 'state_write' with any inputs you can infer (brand, products, style, color_theme, etc.).\n" "5) If something is missing, list it in 'missing_inputs' and ask briefly for it in 'assistant_reply'.\n" "6) Choose 'next_intent' from the enum to reflect what just happened or what you're prompting next.\n" "7) Never output markdown. Only return valid JSON as per schema.\n" ) # ---- Build messages ---- messages = [{"role": "system", "content": system_prompt}] messages.append({"role": "system", "content": rubric}) for chat in history: messages.append({"role": chat['role'], "content": chat['content']}) # ---- Call the model ---- try: resp = client.chat.completions.create( model="gpt0", temperature=0, response_format={"type": "json_schema", "json_schema": json_schema}, messages=messages, ) content = resp.choices[0].message.content except Exception: # fallback to json_object if schema not supported resp = client.chat.completions.create( model="gpt0", temperature=0, response_format={"type": "json_object"}, messages=messages, ) content = resp.choices[0].message.content # ---- Parse & return dict ---- try: return json.loads(content) except json.JSONDecodeError: content = re.sub(r"^```json|```$", "", content.strip(), flags=re.IGNORECASE | re.MULTILINE) return json.loads(content) def general_reply( self, history, ) -> str: """ Simple LLM-driven reply generator for general/possibly-irrelevant user inputs, using the provided conversation history. No rules or few-shots—LLM decides. Returns a single short reply string. """ # ---- normalize history into a compact transcript ---- if isinstance(history, str): transcript = history.strip() elif isinstance(history, list) and history and isinstance(history[0], dict) and "content" in history[0]: transcript = "\n".join(f"{m.get('role','user')}: {m.get('content','').strip()}" for m in history) elif isinstance(history, list): transcript = "\n".join(str(x).strip() for x in history) else: transcript = str(history) system_prompt = ( "You are an expert promotional material designer that helps users create promotional materials by " "guiding them through a structured, step-by-step process, maintaining state and context, and facilitating " "inputs, copywriting, image generation, and template selection. Handle users’ requests and changes " "efficiently, ensuring every step is completed before moving forward.\n\n" "Give answer to user based on the history" "If it is a general input from user then reply accodingly and politely state that you are a copy & design agent for promotional materials. Offer help " "with copy, image prompts, or templates, or offer to continue the existing discussion if context exists." ) user_prompt = ( "Conversation history:\n" f"{transcript}\n\n" "Write ONE short, helpful reply to the user's LAST message, following the system guidelines above. " "Keep it concise and conversational." ) resp = client.chat.completions.create( model="gpt0", temperature=0.3, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], ) return resp.choices[0].message.content.strip() # ---- 3) LLM call + validation ---- def extract_promo_entities( self, history: list, entities: list ) -> PromoEntities: """ Returns a PromoEntities object with normalized fields. Works with OpenAI; for Azure OpenAI, initialize the `client` with your Azure base_url and api_key. """ # ---- 2) Prompt builder (few-shot + constraints) ---- SYS = """You are an expert information extractor. Return ONLY valid JSON following the provided schema. No markdown. No extra keys. Do not assume data. Return null if not found. User can provide answer like any or all for exmaple any color, all product, choose yourself. Take that as answer and if needed generate answer based on that input. Do not change or modify data unless user asked to. Example 70% off upto $100 does not mean 70% off. If a field is unknown, use null (for scalars) or [] (for arrays). Examples: Promotion Type: Sales, Product Launch, Holiday Special, BOGO, nothing specific Products: ceramic mugs, swimwear, beach towels, all items, all product Style: warm, cozy, hand-drawn, bold, minimalist, modern, contemporary, anything Color Theme: white and blue, gold and black, any color """ # Optional: enforce a JSON schema at the API level (models that support response_format) json_schema = { "name": "PromoEntities", "schema": { "type": "object", "additionalProperties": False, "properties": { "promotion_type": {"type": ["string", "null"]}, "products": {"type": ["string", "null"]}, "offer_details": {"type": ["string", "null"]}, "style": {"type": ["string", "null"]}, "color_theme": {"type": ["string", "null"]}, }, "required": ["promotion_type", "products", "offer_details", "style", "color_theme"] } } # Build few-shot messages messages = [{"role": "system", "content": SYS}] for i, obj in enumerate(history): messages.append({"role": obj['role'], "content": obj['content']}) line = ", ".join(f"{k}={v}" for k, v in entities.items()) messages.append({"role": "user","content":f"existing entities are, {line}"}) # for i in range(0, len(FEWSHOTS), 2): # messages.append({"role": "user", "content": build_user_prompt(FEWSHOTS[i])}) # messages.append({"role": "assistant", "content": PromoEntities(**FEWSHOTS[i+1]).model_dump_json()}) # messages.append({"role": "user", "content": build_user_prompt(text)}) # Call the model resp = client.chat.completions.create( model="gpt0", temperature=0, response_format={"type": "json_schema", "json_schema": json_schema}, # fallback: use {"type":"json_object"} messages=messages, ) raw = resp.choices[0].message.content try: obj = PromoEntities.model_validate_json(raw) except ValidationError: # Minimal, robust fallback if model returns slightly off JSON import json, re cleaned = raw.strip() cleaned = re.sub(r"^```json|```$", "", cleaned, flags=re.IGNORECASE | re.MULTILINE).strip() print(f"Validation error: {cleaned}") obj = PromoEntities(**json.loads(cleaned)) # Normalize a bit # obj.products = obj.products # obj.style = obj.style # obj.color_theme = obj.color_theme if obj.offer_details: obj.offer_details = obj.offer_details.strip() if obj.promotion_type: obj.promotion_type = obj.promotion_type.strip() return obj def generate_copy_from_history( self, history, ) -> dict: """ Generate ultra-brief promo copy (max 2 words each) from conversation history. Returns dict: {title, tagline, cta, offer} """ # Normalize history into a single transcript string if isinstance(history, str): transcript = history.strip() elif isinstance(history, list): if history and isinstance(history[0], dict) and "content" in history[0]: transcript = "\n".join(f"{m.get('role','user')}: {m.get('content','').strip()}" for m in history) else: transcript = "\n".join(str(x).strip() for x in history) else: transcript = str(history) system_prompt = ( "You are an expert copywriter that guides users through structured steps, " "maintains state, and generates concise copy. " "Output STRICT JSON only with keys: title, tagline, cta, offer. " "Each value MUST be 1–2 words, no punctuation-heavy phrases. " "Do not chnage any details such as offer, promotion type, etc unless user have asked to. Keep them as it is" "Infer from the conversation history if details are missing." ) user_prompt = ( "Conversation history:\n" f"{transcript}\n\n" "Return JSON with exactly these keys and 1–2 words each: " "title, tagline, cta, offer." ) messages = [{"role": "system", "content": system_prompt}] for msg in history: messages.append({"role": msg['role'], "content": msg['content']}) resp = client.chat.completions.create( model="gpt0", temperature=0, response_format={"type": "json_object"}, messages=messages, ) raw = resp.choices[0].message.content.strip() # Parse and sanitize to enforce max 2 words try: data = json.loads(re.sub(r"^```json|```$", "", raw, flags=re.IGNORECASE | re.MULTILINE).strip()) except json.JSONDecodeError: data = {} return { "title": data.get("title"), "tagline": data.get("tagline"), "cta": data.get("cta"), "offer": data.get("offer"), } def generate_image_prompt(self, history: list, last_msg:str) -> str: """ Generate a prompt for image generation based on conversation history. """ # Normalize history into a single transcript string if isinstance(history, str): transcript = history.strip() elif isinstance(history, list): if history and isinstance(history[0], dict) and "content" in history[0]: transcript = "\n".join(f"{m.get('role','user')}: {m.get('content','').strip()}" for m in history) else: transcript = "\n".join(str(x).strip() for x in history) else: transcript = str(history) transcript += f"\nLast mesasge of the user: {last_msg}" # return f"We need a background image for design editor for banner design. Generate an background image without any foreground element like text or logo based on the following description:\n{transcript}" prompt = f""" Create a detailed image prompt for an image with the following details: {transcript} The image should be realistic with proper background and foreground elements relevant to the user input, with a mild black overlay, and without any text. Focus on creating a visually appealing image. The image would be a food menu design image for one or multiple products as provided. From the reference only take the graphics design excluding the texts and copy. IMPORTANT: Do not add text and copy in the image. Do not add any promotional info or offer details from the user input. Generate a detailed prompt that I can use to create this image, in not more than 300 chars. """ # Make the API call # response = self.llm.completions.create( # model=self.MODEL, # prompt=prompt, # max_tokens=500, # temperature=0.7 # ) # # Extract the generated prompt # image_prompt = response.choices[0].text.strip() response = client.chat.completions.create( messages=[ { "role": "system", "content": "You are a helpful assistant.", }, { "role": "user", "content": prompt, } ], max_tokens=500, temperature=0.7, model="gpt0", ) # Extract and clean the image prompt image_prompt = response.choices[0].message.content.strip() + "\n IMPORTANT: The image should NOT include any text in the image. You should only generate image. Any text in the image would mean it is not qualified. Generate this image formenu design for restaurant for print" # Log the generated prompt print(f"Generated image prompt: {image_prompt}") return image_prompt def generate_image(self, image_prompt:str): # Generate unique filename with timestamp and session ID timestamp = int(time.time()) image_filename = f"promo_image_{self._session_id}_{timestamp}.png" image_filepath = os.path.join(STATIC_PATH, image_filename) sample_ref_1 = os.path.join(STATIC_PATH, "Food_Menu_Special_Price_Promotion.jpg") sample_ref_2 = os.path.join(STATIC_PATH, "Black_Modern_Restaurant.jpg") sample_ref_3 = os.path.join(STATIC_PATH, "delicious_food_menu_design_Instagram_story.jpg") sample_ref_4 = os.path.join(STATIC_PATH, "hot_pizza.png") try: # Step 1: Call Ideogram API response = requests.post( IDEOGRAM_URL, headers={ "Api-Key": IDEOGRAM_API_KEY, }, data={ "prompt": (None, image_prompt), 'aspect_ratio': (None, "1x1"), 'magic_prompt': (None, "ON"), 'negative_prompt': (None, "Texts, logos, brands, annoucement.No text allowed in the image. Stricly no text. No text in the image. No text in the image. No text in the image. No text in the image. No text in the image."), }, files={ 'style_reference_images': ('Food_Menu_Special_Price_Promotion.jpg', open(sample_ref_1, 'rb')), 'style_reference_images': ('Black_Modern_Restaurant.jpg', open(sample_ref_2, 'rb')), 'style_reference_images': ('delicious_food_menu_design_Instagram_story.png', open(sample_ref_3, 'rb')), 'style_reference_images': ('hot_pizza.png', open(sample_ref_4, 'rb')) } ) print(f"Received response from Ideogram generate API: {response.status_code} - {response}") # Step 2: Check response if response.status_code != 200: return False, None, f"Error from Ideogram API: {response.text}" generation_url = response.json().get('data')[0].get('url') if not generation_url: return False, None, "No image URL returned by Ideogram API." except Exception as e: print(f"{e}") # Step 3: Download image image_response = requests.get(generation_url) if image_response.status_code != 200: return False, None, f"Failed to download image: {image_response.status_code} - {image_response.text}" img = Image.open(BytesIO(image_response.content)) img.save(image_filepath) print(f"Image saved locally at: {image_filepath}") # Step 4: Create the image URL # Use the global BASE_URL variable, ensuring it ends with a trailing slash base_url = BASE_URL if BASE_URL.endswith('/') else BASE_URL + '/' image_url = f"{base_url}static/generated_images/{image_filename}" # Log the final image URL print(f"Image URL set to: {image_url}") return True, image_url, None def generate_promo_image(self, prompt: str) -> tuple[bool, str | None, str | None]: """ Simple: call Azure OpenAI Images, save PNG under STATIC_PATH, return public URL. Expects globals: AZURE_ENDPOINT, AZURE_API_KEY, API_VERSION, DEPLOYMENT_NAME, STATIC_PATH (disk path to static/generated_images), BASE_URL (site base, e.g., https://example.com/) """ # --- 1) Generate image via Azure Images API (b64 or URL) --- url = f"https://{AZURE_ENDPOINT_IMAGE}.openai.azure.com/openai/deployments/{DEPLOYMENT_NAME_IMAGE}/images/generations?api-version={API_VERSION_IMAGE}" # params = {"api-version": AZURE_API_KEY_IMAGE} headers = { "Content-Type": "application/json", "Authorization": f"Bearer {AZURE_API_KEY_IMAGE}", } payload = { "prompt": prompt, "size": "1024x1024", "quality": "medium", "output_compression": 100, "output_format": "png", "n": 1, } r = requests.post(url, headers=headers, json=payload, timeout=120) if r.status_code != 200: return False, None, f"Generation failed: {r.status_code} - {r.text}" data = r.json().get("data", []) if not data: return False, None, "No image data returned." item = data[0] # --- 2) Prepare filename/paths --- os.makedirs(STATIC_PATH, exist_ok=True) timestamp = int(time.time()) image_filename = f"promo_image_{self._session_id}_{timestamp}.png" image_filepath = os.path.join(STATIC_PATH, image_filename) # --- 3) Save image (URL or base64) --- try: if item.get("url") or item.get("content_url"): generation_url = item.get("url") or item.get("content_url") image_response = requests.get(generation_url, timeout=120) if image_response.status_code != 200: return False, None, f"Failed to download image: {image_response.status_code} - {image_response.text}" img = Image.open(BytesIO(image_response.content)) img.save(image_filepath) elif item.get("b64_json"): png_bytes = base64.b64decode(item["b64_json"]) Image.open(BytesIO(png_bytes)).save(image_filepath) else: return False, None, "Unsupported image payload (no url or b64_json)." except Exception as e: return False, None, f"Failed to save image: {e}" # # --- 4) Build public URL --- base_url = BASE_URL if BASE_URL.endswith("/") else BASE_URL + "/" image_url = f"{base_url}static/generated_images/{image_filename}" return True, image_url, None def select_templates(self,user_requirements,top_k=5): try: input_requirement = f"""promotion type is :{user_requirements['promotion_type']}, offer details is {user_requirements['offer_details']}, style is {user_requirements['style']}, color theme is {user_requirements['color_theme']} """ request = SearchRequest( query=input_requirement, collection_name="design_template", limit=top_k ) response = search_handler(request) # calling the search_handler function # response = search_handler(user_requirements) # Process the top-k results of the filtering arry selected_results = response.results return_array_of_result = [ { "template_url": r.template_url, "template_id": r.template_id, "template_name": r.template_name } for r in selected_results ] # print("templat_result: ",return_array_of_result ) return return_array_of_result except Exception as e: print(f"Error in select_templates: {e}") return [] def chat(self, request: ChatRequest) -> str: state: ChatState = self.session[request.session_id] # print("====================================",state,"======================") # if len(state.history) == 0: # state.history.append({"role": "system", "content": system_prompt}) state.history.append({"role": "user", "content": request.message}) intent = self.classify_intent_from_history(state.history) state.intents.append(intent) # print(extract_promo_entities(messages)) # extracted_info = {} state.missing_entities = [] print("=============== INTENT IS =========") print(intent) print("=============== INTENT IS =========") if intent == "general": answer = self.general_reply(state.history) state.history.append({"role": "assistant", "content": answer}) response = { "assistant_reply": answer, "next_intent": "input_collection", "intent": "input_collection", "state_write": { "entities": state.entities, "missing_entities": state.missing_entities }, "copy": state.copy or {}, "image_url": state.image_url or None, "template_suggestions": state.template or [], "missing_inputs": state.missing_entities } self._persist() return response for entity in self.extract_promo_entities(state.history, state.entities): # print(f"{entity[0]}: {entity[1]}") if entity[1] is None: state.missing_entities.append(entity[0]) else: state.entities[entity[0]] = entity[1] # print(state.missing_entities) if state.missing_entities: state.history.append({"role": "assistant", "content": f"Please provide the following missing information:\n" + '\n'.join(state.missing_entities) +"\n"+f"We have the following information:\n " + "\n".join(f"{key}: {value}" for key, value in state.entities.items())}) self.session[request.session_id] = state # print("===========================",state,"======================") # state.missing_entities = [entity + " (bold, modern, minimalist, contemporary)" if entity.lower() == "style" else entity for entity in state.entities] reply = f"Please provide the following missing information:\n" + '\n'.join(state.missing_entities) +"\n"+f"We have the following information:\n " + "\n".join(f"{key}: {value}" for key, value in state.entities.items()) if "style" in state.missing_entities: reply = reply + " ### some examples of style may include modern, bold, minimalist, vintage, contemporary" md_reply = self._reply_markdown(reply) response = { "assistant_reply": md_reply, "next_intent": "input_collection", "intent": "input_collection", "state_write": { "entities": state.entities, "missing_entities": state.missing_entities }, "copy": state.copy or {}, "image_url": state.image_url or None, "template_suggestions": state.template or [], "missing_inputs": state.missing_entities } self._persist() return response else: # for copy_ent in promo_copy(self.history): # print(copy_ent) if not state.entities_captured: state.entities_captured = True state.missing_entities = [] self.session[request.session_id] = state # print("===========================",state,"======================") reply = f"All required information has been provided. Please check and confirm the following details: \n" + "\n".join(f"{key}: {value}" for key, value in state.entities.items()) md_reply = self._reply_markdown(reply) state.history.append({"role":'assistant',"content":md_reply}) response = { "assistant_reply": md_reply, "next_intent": "input_confirmation", "intent": "input_confirmation", "state_write": { "entities": state.entities, "missing_entities": state.missing_entities }, "copy": state.copy or {}, "image_url": state.image_url or None, "template_suggestions": state.template or [], "missing_inputs": state.missing_entities } self._persist() return response else: # Will check for intent intent = self.classify_intent_from_history(state.history) print("detected intent",intent) if intent == "input_confirmation" or "input_confirmation" in state.intents: if "copy_generation" not in state.intents or len(state.copy.items()) == 0: copy = self.generate_copy_from_history(state.history) print("generated copy:", copy) formatted_copy = "\n".join(f"{key}: {value}" for key, value in copy.items()) state.intents.append(intent) state.intents.append("copy_generation") state.copy = copy state.history.append({"role": "assistant", "content": f"Great! Here is the generated copy based on the provided information:\n{formatted_copy}. Do you want to continue with image generation or want any modification?"}) self.session[request.session_id] = state assistant_reply = f"Great! Here is the generated copy based on the provided information:\n{formatted_copy}. Do you want to continue with image generation or want any modification?" md_assistant_reply = self._reply_markdown(assistant_reply) response = { "assistant_reply": md_assistant_reply, "next_intent": "copy_confirmation", "intent": "copy_generation", "state_write": { "entities": state.entities, "missing_entities": state.missing_entities }, "copy": state.copy or {}, "image_url": state.image_url or None, "template_suggestions": state.template or [], "missing_inputs": state.missing_entities } self._persist() return response if intent == "copy_generation" or intent == "copy_regeneration": copy = self.generate_copy_from_history(state.history) print("generated copy:", copy) formatted_copy = "\n".join(f"{key}: {value}" for key, value in copy.items()) copy_output = f"Here is the generated copy based on the provided information:\n{formatted_copy}." if intent == "copy_generation": copy_output += " Do you want to continue with image generation or want any modification?" state.intents.append(intent) state.copy = copy state.history.append({"role": "assistant", "content": copy_output}) self.session[request.session_id] = state md_copy_output = self._reply_markdown(copy_output) response = { "assistant_reply": md_copy_output, "next_intent": "copy_confirmation", "intent": intent, "state_write": { "entities": state.entities, "missing_entities": state.missing_entities }, "copy": copy, "image_url": state.image_url or None, "template_suggestions": state.template or [], "missing_inputs": state.missing_entities } self._persist() return response if intent == "copy_confirmation" or "copy_confirmation" in state.intents: print(state.intents) if "image_generation" not in state.intents: print("Will generate image here") image_prompt = self.generate_image_prompt(state.history, request.message) image_url = self.generate_image(image_prompt) # image_url = self.generate_promo_image(image_prompt) # print(f"========= image generation {image_url[2]}") state.history.append({"role":"assistant","content":f"We have generated an image for you based on your inputs. Should we proceed for template search or you want any modification? "}) state.intents.append(intent) state.intents.append("image_generation") state.image_url = image_url[1] self.session[request.session_id] = state response = { "assistant_reply": f"We have generated an image for you based on your inputs. Should we proceed for template search or you want any modification? ", "next_intent": "image_confirmation", "intent": "image_generation", "state_write": { "entities": state.entities, "missing_entities": state.missing_entities }, "copy": state.copy or {}, "image_url": state.image_url or None, "template_suggestions": state.template or [], "missing_inputs": state.missing_entities } self._persist() return response if intent == "image_generation" or intent == "image_regeneration": image_prompt = self.generate_image_prompt(state.history, request.message) image_url = self.generate_image(image_prompt) # image_url = self.generate_promo_image(image_prompt) # print(f"========= image generation {image_url[2]}") state.history.append({"role":"assistant","content":f"We have generated an image for you based on your inputs. Should we proceed for template search or you want any modification? "}) state.intents.append(intent) state.image_url = image_url[1] self.session[request.session_id] = state response = { "assistant_reply": "We have generated an image for you based on your inputs. Should we proceed for template search or you want any modification? ", "next_intent": "image_confirmation", "intent": intent, "state_write": { "entities": state.entities, "missing_entities": state.missing_entities }, "copy": state.copy or {}, "image_url": state.image_url or None, "template_suggestions": state.template or [], "missing_inputs": state.missing_entities } self._persist() return response if intent == "image_confirmation" or "image_confirmation" in state.intents: if "template_search" not in state.intents: templates = self.select_templates(state.entities) state.template = templates state.intent = intent state.intent = "template_search" formatted_templates = "\n".join(f"{i+1}. {template['template_name']}" for i, template in enumerate(templates)) if state.copy is None or len(state.copy.items()) == 0: copy = self.generate_copy_from_history(state.history) state.copy=copy response = { "assistant_reply": f"Here are some templates that might work for you:\n", "next_intent": "template_selection", "intent": "template_search", "state_write": { "entities": state.entities, "missing_entities": state.missing_entities }, "copy": state.copy or {}, "image_url": state.image_url or None, "template_suggestions": state.template or [], "missing_inputs": state.missing_entities } state.history.append({"role": "assistant", "content": f"Here are some templates that might work for you: \n{formatted_templates}"}) self._persist() return response if intent == "template_search": templates = self.select_templates(state.entities) state.template = templates state.intent = intent formatted_templates = "\n".join(f"{i+1}. {template['template_name']}" for i, template in enumerate(templates)) if state.copy is None or len(state.copy.items()) == 0: copy = self.generate_copy_from_history(state.history) state.copy=copy state.history.append({"role":"assistant","content":f"Here are some templates that might work for you:\n"}) response = { "assistant_reply": f"Here are some templates that might work for you:\n", "next_intent": "template_selection", "intent": "template_search", "state_write": { "entities": state.entities, "missing_entities": state.missing_entities }, "copy": state.copy or {}, "image_url": state.image_url or None, "template_suggestions": state.template or [], "missing_inputs": state.missing_entities } state.history.append({"role": "assistant", "content": f"Here are some templates that might work for you: \n{formatted_templates}"}) self._persist() return response if intent == "input_collection": state.history.append({"role": "assistant", "content": f"Does the updated information look correct?" + "\n" + "\n".join(f"{key}: {value}" for key, value in state.entities.items())}) self.session[request.session_id] = state reply = f"Does the updated information look correct?\n" + "\n".join(f"{key}: {value}" for key, value in state.entities.items()) md_reply = self._reply_markdown(reply) state.history.append({"role":"assistant","content":md_reply}) response = { "assistant_reply": md_reply, "next_intent": "input_confirmation", "intent": intent, "state_write": { "entities": state.entities, "missing_entities": state.missing_entities }, "copy": state.copy or {}, "image_url": state.image_url or None, "template_suggestions": state.template or [], "missing_inputs": state.missing_entities } self._persist() return response