""" ================================================================================ Storigo Content Generator - Professional Edition v7.0 - Synchronous image generation during slide creation - Direct image URL assignment (no null values) - One slide processed completely before moving to next - Professional error handling and logging - GUARANTEED: Every slide with is_image=1 gets an image URL or None ================================================================================ """ import os import time import random import re import json import requests from langchain_groq import ChatGroq from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser, PydanticOutputParser from pydantic import BaseModel, Field, validator from typing import List, Optional, Dict, Union, Callable, Any from collections import deque from threading import Lock import logging from langchain_ollama import OllamaLLM from langchain_ollama import ChatOllama # Import the professional synchronous image generator from storigo_image_generator import fetch_image_for_slide # -------------------------------------------------------------------------- # Configuration # -------------------------------------------------------------------------- GROQ_API_KEY = "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" # Setup logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - [%(levelname)s] - (StorigoGen) - %(message)s' ) logger = logging.getLogger(__name__) # -------------------------------------------------------------------------- # Rate Limiter Class # -------------------------------------------------------------------------- class SmartRateLimiter: """Intelligent rate limiter for Groq API""" def __init__(self, max_requests_per_minute=25): self.max_requests = max_requests_per_minute self.request_times = deque() self.lock = Lock() self.total_requests = 0 self.total_waits = 0 def wait_if_needed(self): """Wait if approaching rate limits""" with self.lock: now = time.time() # Remove old requests while self.request_times and now - self.request_times[0] > 60: self.request_times.popleft() # Check if we need to wait if len(self.request_times) >= self.max_requests: sleep_time = 60 - (now - self.request_times[0]) + 1 logger.warning(f"โณ Rate limit protection: waiting {sleep_time:.1f}s") self.total_waits += 1 time.sleep(sleep_time) # Clean up again after waiting while self.request_times and time.time() - self.request_times[0] > 60: self.request_times.popleft() self.request_times.append(time.time()) self.total_requests += 1 rate_limiter = SmartRateLimiter(max_requests_per_minute=25) # -------------------------------------------------------------------------- # Pydantic Models # -------------------------------------------------------------------------- class SlideContent(BaseModel): type: str = Field("flash") subheading: Optional[str] = Field(None) paragraphs: List[str] = Field(...) visualization_suggestion: str = Field(...) image: Optional[str] = Field(default=None) @validator('image', pre=True, always=True) def validate_image_path(cls, v): """Ensure image is either None or a valid string URL""" if v is None: return None if isinstance(v, str): stripped = v.strip() return stripped if stripped and stripped != "None" else None return None class Config: arbitrary_types_allowed = True validate_assignment = True class MCQContent(BaseModel): type: str = Field("Question") question: str = Field(..., description="The multiple-choice question") options: List[str] = Field(..., description="A list of 4 answer options") correct_answer: str = Field(..., description="The single letter 'a', 'b', 'c', or 'd' corresponding to the correct answer") class StorigoContent(BaseModel): slides: Dict[str, SlideContent] = Field(...) token_count: int = 0 class StorigoContentMCQMid(BaseModel): slides: Dict[str, Union[SlideContent, MCQContent]] = Field(...) token_count: int = 0 # -------------------------------------------------------------------------- # Utility Functions # -------------------------------------------------------------------------- def count_tokens(text: str) -> int: """Simple token counter based on word count""" tokens = re.findall(r'\w+', text) return len(tokens) def quick_json_fix(ai_message) -> str: """Fix common JSON formatting issues from LLM responses""" if hasattr(ai_message, 'content'): text = ai_message.content else: text = str(ai_message) text = text.strip() # Remove markdown code blocks if "```" in text: text = re.sub(r'^```json?\s*', '', text, flags=re.MULTILINE) text = re.sub(r'```$', '', text, flags=re.MULTILINE) # Extract the outermost JSON object or array start_obj = text.find('{') start_arr = text.find('[') start = -1 if start_obj != -1 and start_arr != -1: start = min(start_obj, start_arr) else: start = max(start_obj, start_arr) end_obj = text.rfind('}') end_arr = text.rfind(']') end = max(end_obj, end_arr) + 1 if start != -1 and end != 0: text = text[start:end] # --- Robust Regex Cleaning --- # 1. Fix missing commas between key-value pairs or array items across newlines text = re.sub(r'(\"|\]|\})\s*\n\s*(\"|\[|\{)', r'\1,\n\2', text) # 2. Fix missing commas between key-value pairs (value followed by "next_key":) # Matches: "val" "key": OR 123 "key": OR null "key": OR true "key": text = re.sub(r'(\"|null|true|false|\d+|\]|\})\s*(\"[\w\-_]+\"\s*:)', r'\1,\n\2', text) # 3. Fix missing commas in arrays or between values on the same line # Matches: "item1" "item2" OR "item1" [ OR } "item" text = re.sub(r'(\"|\]|\})\s+(\"|\[|\{)', r'\1,\2', text) # 4. Fix trailing commas before closing braces/brackets text = re.sub(r',\s*([\]\}])', r'\1', text) # 5. Remove any non-JSON noise that might have slipped through at the beginning/end text = text.strip() return text # -------------------------------------------------------------------------- # Main Generation Function - PROFESSIONAL VERSION # -------------------------------------------------------------------------- def generate_slide_content_from_prompt( prompt: str, num_slides: int, num_mcqs: int, is_image: bool, is_question: str, question_position: str, GPU: int, progress_cb: Optional[Callable[[Dict[str, Any]], None]] = None, ) -> Union[StorigoContent, StorigoContentMCQMid]: """ Generate professional slide content with INLINE image generation PROFESSIONAL APPROACH: 1. Generate slide content 2. For EACH slide, immediately fetch and assign image 3. Process one slide completely before moving to next 4. Return final result with all images assigned Args: prompt: Main topic prompt num_slides: Number of slides to generate num_mcqs: Number of MCQs is_image: Whether to generate images is_question: Whether to generate MCQs question_position: Position of questions ("end" or "1") GPU: Legacy parameter Returns: StorigoContent or StorigoContentMCQMid with images properly assigned """ generation_start = time.time() try: logger.info("=" * 80) logger.info("๐Ÿš€ STORIGO PROFESSIONAL CONTENT GENERATOR v7.0") logger.info("=" * 80) logger.info(f"๐Ÿ“‹ Configuration:") logger.info(f" - Slides: {num_slides}") logger.info(f" - MCQs: {num_mcqs}") logger.info(f" - Images: {is_image}") logger.info(f" - Prompt: {prompt[:60]}...") logger.info("=" * 80) # Initialize API conditionally based on GPU logger.info(f"๐Ÿ”ง Initializing API setup for GPU={GPU}") if str(GPU).strip() == "1": try: from langchain_community.chat_models import ChatOllama except ImportError: from langchain_ollama import ChatOllama # llama3.1:8b llm = ChatOllama( base_url='http://127.0.0.1:11434', model="llama3.1:8b", format="json", temperature=0.7, num_predict=8192, num_ctx=8192, max_tokens=4096, request_timeout=60, max_retries=3 ) logger.info("โœ… Ollama API (llama3.1:8b) ready") else: llm = ChatGroq( model_name='llama-3.1-8b-instant', groq_api_key=GROQ_API_KEY, temperature=0.7, max_tokens=4096, request_timeout=60, max_retries=3 ) logger.info("โœ… Groq API ready") if progress_cb: progress_cb({"type": "status", "message": "Initializing generator..."}) # ====================================================================== # STEP 1: Generate Slide Content # ====================================================================== logger.info("\n" + "=" * 80) logger.info("๐Ÿ“ STEP 1: Generating Slide Content") logger.info("=" * 80) slide_content_template = """ Based on the following prompt, generate professional content for exactly {num_slides} slides. Each slide MUST include: - A clear, descriptive subheading - 2-3 informative paragraphs (very detailed, well-written, and engaging.) - A "visualization_suggestion" (exactly 6 to 10 words). This MUST describe a real-world photograph of physical people or objects. DO NOT use abstract concepts, slide titles, or text. CRITICAL for visualization_suggestion: - Use ONLY concrete, visual, searchable terms with specific details - Think: "What would I search on a stock photo website with exact scene details?" - Include specific elements: people (e.g., "diverse team"), objects (e.g., "laptop computer"), actions (e.g., "discussing strategy"), settings (e.g., "modern office") - Good examples: "diverse business team discussing strategy in modern conference room", "data analyst examining colorful charts on computer screen", "professional kitchen staff preparing fresh ingredients in commercial kitchen" - Bad examples: "concept of leadership", "abstract representation", "team meeting" (too vague) - EACH visualization suggestion MUST BE COMPLETELY UNIQUE across all slides - no overlapping concepts, objects, or scenes - If slides are related, vary all elements significantly: different people, objects, actions, and settings Prompt: {prompt} Make content professional with: - Thought-provoking insights - Relevant examples or statistics - Industry trends - Clear, actionable takeaways Return ONLY a valid JSON object with this exact structure (no explanations, no markdown, no schema definitions): {{ "slides": {{ "slide_1": {{ "subheading": "Introduction to Kitchen Safety", "paragraphs": [ "Kitchen safety is paramount in any food service environment.", "Proper protocols protect staff and ensure food quality." ], "visualization_suggestion": "professional kitchen staff wearing safety gear in commercial kitchen", "image": null }}, "slide_2": {{ "subheading": "Food Handling Best Practices", "paragraphs": [ "Proper food handling prevents contamination and foodborne illness.", "Key practices include temperature control and cross-contamination prevention." ], "visualization_suggestion": "chef demonstrating proper food handling techniques with fresh ingredients", "image": null }} }}, "token_count": 0 }} Generate exactly {num_slides} slides numbered slide_1 through slide_{num_slides}. """ if progress_cb: progress_cb({"type": "stage", "stage": "slide_text", "message": "Generating slide text..."}) slide_prompt = ChatPromptTemplate.from_template(slide_content_template) slide_chain = ( { "prompt": lambda x: x["prompt"], "num_slides": lambda x: x["num_slides"] } | slide_prompt | llm | quick_json_fix ) # Generate with rate limiting rate_limiter.wait_if_needed() slide_gen_start = time.time() raw_result = slide_chain.invoke({"prompt": prompt, "num_slides": num_slides}) slide_gen_time = time.time() - slide_gen_start # Parse the JSON response try: # First attempt: standard parsing after quick_json_fix result_data = json.loads(raw_result) result = StorigoContent(**result_data) logger.info(f"โœ… Generated {len(result.slides)} slides in {slide_gen_time:.2f}s") except (json.JSONDecodeError, Exception) as e: logger.warning(f"โš ๏ธ JSON parsing failed on first attempt: {str(e)}") # Second attempt: Deep cleaning logger.info("๐Ÿ”ง Attempting deep clean of JSON...") try: # Remove common problematic patterns that quick_json_fix might have missed # or that were introduced by the fix itself deep_cleaned = raw_result # 1. Broadly try to fix structure first deep_cleaned = quick_json_fix(deep_cleaned) # 2. Final attempt at basic structural fixes if quick_json_fix missed anything deep_cleaned = re.sub(r'(\"|\]|\})\s*\n\s*(\"|\[|\{)', r'\1,\n\2', deep_cleaned) deep_cleaned = re.sub(r'(\"|null|true|false|\d+|\]|\})\s*(\"[\w\-_]+\"\s*:)', r'\1,\n\2', deep_cleaned) deep_cleaned = re.sub(r',\s*([\]\}])', r'\1', deep_cleaned) result_data = json.loads(deep_cleaned) result = StorigoContent(**result_data) logger.info(f"โœ… Generated {len(result.slides)} slides after deep clean") except Exception as deep_e: logger.error(f"โŒ JSON parsing failed even after deep clean: {str(deep_e)}") logger.error(f"Raw response (first 1000 chars): {raw_result[:1000]}") # Write failure to a temp file for analysis try: with open("failed_layout.json", "w", encoding="utf-8") as f: f.write(raw_result) logger.info("๐Ÿ“‚ Saved failed JSON to failed_layout.json for inspection") except: pass raise Exception(f"Failed to parse slide content JSON: {str(e)}") # Sort slides by slide number (numerical order) ordered_slides = dict(sorted(result.slides.items(), key=lambda x: int(x[0].split('_')[1]))) # Validate content for slide_key, slide_content in ordered_slides.items(): # Ensure minimum paragraphs while len(slide_content.paragraphs) < 2: slide_content.paragraphs.append("Additional information for this slide.") # Ensure visualization suggestion exists if not slide_content.visualization_suggestion or not slide_content.visualization_suggestion.strip(): slide_content.visualization_suggestion = "professional business concept" logger.warning(f"โš ๏ธ {slide_key}: Empty visualization, using default") if progress_cb: progress_cb({ "type": "slide", "slide_key": slide_key, "subheading": slide_content.subheading, "paragraphs": slide_content.paragraphs, "visualization_suggestion": slide_content.visualization_suggestion, "message": f"{slide_key} prepared" }) # ====================================================================== # STEP 2: Generate Images INLINE (One by One) # ====================================================================== if is_image: logger.info("\n" + "=" * 80) logger.info("๐ŸŽจ STEP 2: Generating Images (INLINE - One by One)") logger.info("=" * 80) if progress_cb: progress_cb({"type": "stage", "stage": "images", "message": "Generating images..."}) success_count = 0 failed_count = 0 for slide_key in ordered_slides.keys(): slide_content = ordered_slides[slide_key] try: viz_prompt = slide_content.visualization_suggestion logger.info(f"\n๐Ÿ“ธ Processing {slide_key}...") logger.info(f" Visualization: '{viz_prompt}'") # CRITICAL: Fetch image synchronously image_start = time.time() image_url = fetch_image_for_slide(slide_key, viz_prompt) image_time = time.time() - image_start if image_url: # ASSIGN IMAGE URL ordered_slides[slide_key].image = image_url success_count += 1 logger.info(f" โœ… Image assigned: {image_url} ({image_time:.2f}s)") if progress_cb: progress_cb({"type": "image", "slide_key": slide_key, "image": image_url, "message": f"{slide_key} image ready"}) else: # NO IMAGE FOUND ordered_slides[slide_key].image = None failed_count += 1 logger.warning(f" โš ๏ธ No image found ({image_time:.2f}s)") if progress_cb: progress_cb({"type": "image", "slide_key": slide_key, "image": None, "message": f"{slide_key} image not found"}) except Exception as e: logger.error(f" โŒ Error: {str(e)}") ordered_slides[slide_key].image = None failed_count += 1 if progress_cb: progress_cb({"type": "image", "slide_key": slide_key, "image": None, "message": f"{slide_key} image error"}) logger.info("\n" + "=" * 80) logger.info(f"๐ŸŽจ Image Generation Summary:") logger.info(f" - Success: {success_count}/{len(ordered_slides)}") logger.info(f" - Failed: {failed_count}/{len(ordered_slides)}") logger.info("=" * 80) else: # No images requested - set all to None logger.info("\n๐Ÿ“ท Image generation disabled (is_image=False)") for slide_key in ordered_slides.keys(): ordered_slides[slide_key].image = None # Update result with ordered slides result.slides = ordered_slides # ====================================================================== # STEP 3: Calculate Token Count # ====================================================================== token_count = 0 for slide_content in result.slides.values(): text = f"{slide_content.subheading} {' '.join(slide_content.paragraphs)} {slide_content.visualization_suggestion}" token_count += count_tokens(text) # ====================================================================== # STEP 4: Generate MCQs (if requested) # ====================================================================== if is_question and is_question not in ["0", "false", "False"]: logger.info("\n" + "=" * 80) logger.info(f"โ“ STEP 4: Generating {num_mcqs} MCQs (Batch Mode)") logger.info("=" * 80) if progress_cb: progress_cb({"type": "stage", "stage": "mcq", "message": f"Generating {num_mcqs} MCQs..."}) time.sleep(1.5) # Rate limiting # Build context from slides (Truncated to avoid 413 error) all_context = [] for key, slide in result.slides.items(): title = slide.subheading # Keep only first 300 characters of paragraphs to avoid "Request Entity Too Large" paras_text = ' '.join(slide.paragraphs) if len(paras_text) > 300: summary = paras_text[:300] + "..." else: summary = paras_text all_context.append(f"**{title}**\n{summary}") full_context = "\n\n".join(all_context) mcq_template = """ You are an expert educational content creator. Generate exactly {num_mcqs} high-quality MCQs. **Requirements:** 1. Test comprehension of key concepts 2. Distribute across different topics 3. Each question has exactly 4 options 4. Only one clearly correct answer 5. Avoid trivial questions 6. CRITICAL: For `correct_answer`, you MUST ONLY output "a", "b", "c", or "d". (a = first option, b = second option, c = third option, d = fourth option). NEVER output 'T', 'F', '1', '2', '3', '4'. NEVER output the full text of the option. ONLY the single letter! **Content:** {context} **Output Format:** Return ONLY a valid JSON object with a single key "mcqs" containing an array. DO NOT return a raw array. It must be identical to this example: {{ "mcqs": [ {{ "type": "Question", "question": "What is the capital of France?", "options": ["Paris", "London", "Berlin", "Madrid"], "correct_answer": "a" }}, {{ "type": "Question", "question": "What is 2 + 2?", "options": ["3", "4", "5", "6"], "correct_answer": "b" }} ] }} Generate exactly {num_mcqs} questions now: """ try: mcq_prompt = ChatPromptTemplate.from_template(mcq_template) rate_limiter.wait_if_needed() mcq_start = time.time() mcq_result = (mcq_prompt | llm).invoke({ "context": full_context, "num_mcqs": num_mcqs }) mcq_time = time.time() - mcq_start # Parse MCQs with improved resiliency content = mcq_result.content.strip() logger.info(f"๐Ÿ” Raw LLM Output: {content[:800]}") # Robustly extract JSON (object or array) json_match = re.search(r'(\{.*\}|\[.*\])', content, re.DOTALL) if json_match: content = json_match.group(0) # Automatically fix common formatting issues content = quick_json_fix(content) try: mcq_data_parsed = json.loads(content) except Exception as syntax_err: logger.error(f"Failed to decode MCQ JSON syntax: {syntax_err}") # Try bare objects if needed if content.strip().startswith('{') and content.strip().endswith('}'): try: mcq_data_parsed = json.loads(f"[{content.strip()}]") except: raise syntax_err else: raise syntax_err # Extract the list of MCQs if isinstance(mcq_data_parsed, dict): # Look for "mcqs" key or any list value if "mcqs" in mcq_data_parsed and isinstance(mcq_data_parsed["mcqs"], list): mcq_list = mcq_data_parsed["mcqs"] else: found_list = next((v for v in mcq_data_parsed.values() if isinstance(v, list)), None) mcq_list = found_list if found_list is not None else [mcq_data_parsed] elif isinstance(mcq_data_parsed, list): mcq_list = mcq_data_parsed else: mcq_list = [mcq_data_parsed] mcqs = {} for idx, mcq_data in enumerate(mcq_list[:num_mcqs]): try: if 'type' not in mcq_data: mcq_data['type'] = 'Question' mcqs[f"mcq_{idx + 1}"] = MCQContent(**mcq_data) logger.info(f"โœ… MCQ {idx + 1}: {mcq_data['question'][:50]}...") if progress_cb: progress_cb({ "type": "mcq", "mcq_key": f"mcq_{idx + 1}", "question": mcq_data.get("question", ""), "options": mcq_data.get("options", []), "correct_answer": mcq_data.get("correct_answer", ""), "message": f"MCQ {idx + 1} ready" }) except Exception as e: logger.warning(f"โš ๏ธ Failed to parse MCQ {idx + 1}: {e}") logger.info(f"โœ… Generated {len(mcqs)}/{num_mcqs} MCQs in {mcq_time:.2f}s") # Calculate MCQ token count for mcq in mcqs.values(): text = f"{mcq.question} {' '.join(mcq.options)} {mcq.correct_answer}" token_count += count_tokens(text) # Combine slides and MCQs final_content = {} if question_position == "1" or question_position == 1: # Distribute MCQs densely or sparsely logger.info("๐Ÿ“ Distributing MCQs throughout slides") slide_keys = list(result.slides.keys()) num_s = len(slide_keys) num_m = len(mcqs) # Calculate how many MCQs per slide or interval if num_m >= num_s: # More or equal MCQs than slides: distribute across each slide # and append remaining at the end mcq_keys = list(mcqs.keys()) for i, s_key in enumerate(slide_keys): final_content[s_key] = result.slides[s_key] if i < num_m: final_content[mcq_keys[i]] = mcqs[mcq_keys[i]] # Add any leftover MCQs for j in range(num_s, num_m): final_content[mcq_keys[j]] = mcqs[mcq_keys[j]] else: # Fewer MCQs than slides: spacing them out interval = max(1, num_s // num_m) mcq_counter = 0 mcq_keys = list(mcqs.keys()) for i, s_key in enumerate(slide_keys): final_content[s_key] = result.slides[s_key] if mcq_counter < num_m and (i + 1) % interval == 0: final_content[mcq_keys[mcq_counter]] = mcqs[mcq_keys[mcq_counter]] mcq_counter += 1 # Ensure no MCQs were missed (e.g. due to interval math) while mcq_counter < num_m: final_content[mcq_keys[mcq_counter]] = mcqs[mcq_keys[mcq_counter]] mcq_counter += 1 else: # All MCQs at end logger.info("๐Ÿ“ Placing all MCQs at end") for key in result.slides.keys(): final_content[key] = result.slides[key] for mcq_key, mcq_content in mcqs.items(): final_content[mcq_key] = mcq_content generation_time = time.time() - generation_start logger.info("\n" + "=" * 80) logger.info("๐ŸŽ‰ GENERATION COMPLETE") logger.info("=" * 80) logger.info(f"๐Ÿ“Š Final Statistics:") logger.info(f" - Total time: {generation_time:.2f}s") logger.info(f" - Slides: {len(result.slides)}") logger.info(f" - MCQs: {len(mcqs)}") logger.info(f" - Total items: {len(final_content)}") logger.info(f" - Token count: {token_count}") if is_image: logger.info(f" - Images assigned: {success_count}/{len(result.slides)}") logger.info("=" * 80) return StorigoContentMCQMid(slides=final_content, token_count=token_count) except Exception as e: logger.error(f"โŒ MCQ generation failed: {str(e)}") # Return slides only return StorigoContent(slides=result.slides, token_count=token_count) else: # No MCQs requested generation_time = time.time() - generation_start logger.info("\n" + "=" * 80) logger.info("๐ŸŽ‰ GENERATION COMPLETE") logger.info("=" * 80) logger.info(f"๐Ÿ“Š Final Statistics:") logger.info(f" - Total time: {generation_time:.2f}s") logger.info(f" - Slides: {len(result.slides)}") logger.info(f" - Token count: {token_count}") if is_image: logger.info(f" - Images assigned: {success_count}/{len(result.slides)}") logger.info("=" * 80) return StorigoContent(slides=result.slides, token_count=token_count) except Exception as e: logger.error("=" * 80) logger.error(f"โŒ FATAL ERROR") logger.error(f"Error: {str(e)}") logger.error("=" * 80) import traceback logger.error(traceback.format_exc()) raise Exception(f"Content generation failed: {str(e)}") # -------------------------------------------------------------------------- # Testing # -------------------------------------------------------------------------- if __name__ == "__main__": print("\n" + "=" * 80) print("๐Ÿงช TESTING STORIGO CONTENT GENERATOR v7.0") print("=" * 80 + "\n") test_result = generate_slide_content_from_prompt( prompt="Kitchen safety and food handling best practices", num_slides=3, num_mcqs=2, is_image=True, is_question="1", question_position="end", GPU=1 ) print("\n" + "=" * 80) print("๐Ÿ“‹ TEST RESULTS") print("=" * 80) slide_count = sum(1 for k in test_result.slides.keys() if k.startswith('slide_')) mcq_count = sum(1 for k in test_result.slides.keys() if k.startswith('mcq_')) images_with_url = sum(1 for k, v in test_result.slides.items() if k.startswith('slide_') and hasattr(v, 'image') and v.image) print(f"Total items: {len(test_result.slides)}") print(f" - Slides: {slide_count}") print(f" - MCQs: {mcq_count}") print(f" - Images assigned: {images_with_url}/{slide_count}") print(f"Token count: {test_result.token_count}") print("\n๐Ÿ“ธ Image URLs:") for key, content in test_result.slides.items(): if key.startswith('slide_'): img = getattr(content, 'image', None) status = "โœ…" if img else "โš ๏ธ" print(f" {status} {key}: {img if img else 'None'}") print("\n" + "=" * 80) print("โœ… Test completed successfully!") print("=" * 80)