import os import re import logging import tempfile import shutil import subprocess import asyncio import boto3 import json from typing import List, Dict, Optional, Tuple from concurrent.futures import ThreadPoolExecutor, as_completed from botocore.exceptions import ClientError from openai import OpenAI, APIStatusError, APIConnectionError, RateLimitError from dotenv import load_dotenv from playwright.async_api import async_playwright from src.state import VideoGenerationState from src.tools.audio_utils import audio_fn_from_string, trim_audio_to_max_duration, pad_audio_to_duration from src.tools.template_utils import ( render_template, format_code_with_pygments, get_pygments_css, format_bullet_points, ) from src.tools.prompt_utils import get_tts_narration_prompt, shorten_narration_text from src.tools.imp_word_highlight import process_all_bullets_for_highlighting from src.nodes.slide_creation.image_create import generate_infographic_img from src.nodes.slide_creation.code_generation import generate_code_example from src.tools.assets_utils import get_assets_dir, sanitize_filename from src.three_d_merger import get_complete_html_page from src.model_config import get_client_for_task, get_model_for_task, get_service_for_task load_dotenv() OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") logger = logging.getLogger(__name__) openai_client = None client = None DEFAULT_MODEL = "gpt-4o-mini" if OPENAI_API_KEY: try: openai_client = OpenAI(api_key=OPENAI_API_KEY) client = openai_client logger.info("Using OpenAI for LLM") except Exception as e: logger.warning(f"Failed to initialize OpenAI: {e}") if not client: logger.error("No valid LLM client available") class SlideCreationNode: """ Node responsible for creating presentation slide videos. User personalization is read directly from state.user_profile, which is populated upstream by the router node via UserInfoRetriever. All outputs are saved to S3 only - no local storage for ECS deployment. """ def __init__(self, bucket_name="tech-learn-state", enable_highlighting=False, enable_images=True, enable_visualizations=True, enable_maths=False): self.bucket_name = bucket_name self.s3_client = boto3.client("s3") self.logger = logging.getLogger(__name__) logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) self.client = client self.enable_highlighting = enable_highlighting self.enable_images = enable_images self.enable_visualizations = enable_visualizations self.enable_maths = enable_maths self.AUDIO_DELAY_MS = 2500 self.PLAYWRIGHT_AWAIT_DELAY = 50 self.cached_presentation_content = None def _extract_personalization_context(self, user_profile: Dict) -> str: """ Builds a personalization context string from state.user_profile. This replaces the old Pinecone merge logic that was happening inline inside generate_presentation_text. """ if not user_profile: return "" parts = [] name = user_profile.get("user_name") age = user_profile.get("age") role = user_profile.get("role") level = user_profile.get("experience_level") traits = user_profile.get("traits", []) hobbies = user_profile.get("hobbies", []) interests = user_profile.get("interests", []) skills = user_profile.get("skills", []) if name or role or age: line = "STUDENT PROFILE:" if name: line += f"\n- Name: {name}" if age: line += f"\n- Age: {age}" if role: line += f"\n- Role: {role}" parts.append(line) if level and level != "not_specified": parts.append(f"- Technical Level: {level}") if traits: parts.append(f"- Personality: {', '.join(traits[:3])}") hobbies_and_interests = list(set(hobbies + interests)) if hobbies_and_interests: parts.append(f"- Interests/Hobbies: {', '.join(hobbies_and_interests[:4])}") if skills: parts.append(f"- Known Skills: {', '.join(skills[:5])}") if parts: parts.append("\nUse these personal details in your examples and explanations to make content relatable.\n") return "\n".join(parts) def _call_llm_with_fallback(self, messages, temperature=0.5, max_tokens=2000, model=None, task_name="content_generation"): import time if model is None: try: model = get_model_for_task(task_name) task_client = get_client_for_task(task_name) service = get_service_for_task(task_name) self.logger.info(f"Using {service} ({model}) for task: {task_name}") except Exception as e: self.logger.warning(f"Could not get task-specific model for {task_name}: {e}. Using defaults.") model = DEFAULT_MODEL task_client = None else: task_client = None if task_client: try: if service == "Anthropic": system_msg = next((m['content'] for m in messages if m['role'] == 'system'), None) user_messages = [m for m in messages if m['role'] != 'system'] if system_msg: response = task_client.messages.create( model=model, max_tokens=max_tokens, temperature=temperature, system=system_msg, messages=user_messages ) else: response = task_client.messages.create( model=model, max_tokens=max_tokens, temperature=temperature, messages=user_messages ) class MockResponse: def __init__(self, content): self.choices = [type('obj', (object,), {'message': type('obj', (object,), {'content': content})})()] response = MockResponse(response.content[0].text) else: response = task_client.chat.completions.create( model=model, messages=messages, temperature=temperature, max_tokens=max_tokens, ) self.logger.info(f"Successfully called {task_name} model: {model}") return response except Exception as e: self.logger.warning(f"Primary LLM failed for {task_name}: {e}. Falling back to Gemini.") try: from google import genai import os gemini_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY")) prompt = "\n".join([m["content"] for m in messages]) gemini_response = gemini_client.models.generate_content( model="gemini-2.5-flash", contents=prompt ) class MockResponse: def __init__(self, content): self.choices = [type('obj', (object,), { 'message': type('obj', (object,), {'content': content}) })()] return MockResponse(gemini_response.text) except Exception as gemini_error: self.logger.error(f"Gemini fallback failed: {gemini_error}") raise RuntimeError("All LLM providers failed (OpenAI + Gemini)") def _upload_to_s3(self, local_path: str, s3_key: str) -> str: try: self.logger.info(f"Uploading {local_path} to S3 as {s3_key}") self.s3_client.upload_file(local_path, self.bucket_name, s3_key) return f"s3://{self.bucket_name}/{s3_key}" except ClientError as e: self.logger.error(f"Failed to upload {local_path} to S3: {e}") raise def generate_presentation_text( self, topic: str, programming_language: str, user_profile: Dict, ) -> str: """ Generates the presentation script using an LLM. Personalization comes entirely from user_profile (state.user_profile), which was built by UserInfoRetriever upstream. No Pinecone calls here. """ try: personalization_context = self._extract_personalization_context(user_profile) code_instructions = "" if self.is_programming_topic(topic, programming_language): code_instructions = """ **PRIMARY GOAL: Generate Foundational Content** - This is a programming topic. Your task is to create the conceptual slides. - DO NOT generate a final 'Complete Example' slide. A separate process will create that. - **You MUST include small, inline code snippets (1-3 lines) on at least one of the conceptual slides** to help explain the concepts. """ else: code_instructions = """ **NON-CODE TOPIC INSTRUCTIONS:** - This is a conceptual topic. Focus on clear explanations, not code. """ system_prompt = f""" You are an expert technical educator and presentation designer. You MUST follow all user instructions. Your task is to generate 3-5 slides for the topic: '{topic}'. You MUST NOT generate a final slide with a complete code example. **CRITICAL OUTPUT RULES:** - Output ONLY the slides. NO thinking process, NO tags, NO explanation. - Start your response directly with "Slide 1:" - Follow the exact format specified in the user prompt. **RULE 1: VISUALIZATION (AI-DECIDED - CRITICAL)** - The visualization flag for this request is: **{"ENABLED" if self.enable_visualizations else "DISABLED"}**. - `if self.enable_visualizations == False`: You **MUST NOT** add any `[VISUALIZATION_PLACEHOLDER]` tags, regardless of the topic. - `if self.enable_visualizations == True`: You **MUST** follow this logic: - First, you MUST classify the topic: '{topic}'. - Is this topic one of the following: mathematical concepts, physics, complex data structures/algorithms, Machine Learning, system design, architecture patterns, component lifecycles, state management, data flow, or technical workflows? - **If YES:** You **MUST** select EXACTLY ONE slide and place `[VISUALIZATION_PLACEHOLDER]` at the START of its 'Content:'. - **If NO:** You **MUST NOT** add this placeholder. **RULE 2: IMAGE (FLAG-DECIDED - CRITICAL)** - The image flag for this request is: **{"ENABLED" if self.enable_images else "DISABLED"}**. - `if self.enable_images == True`: You **MUST** select EXACTLY ONE slide and place `[IMAGE_PLACEHOLDER]` at the START of its 'Content:'. This is mandatory. - `if self.enable_images == False`: You **MUST NOT** add any `[IMAGE_PLACEHOLDER]` tags. **RULE 3: NO OVERLAP (MANDATORY)** - The `[IMAGE_PLACEHOLDER]` and `[VISUALIZATION_PLACEHOLDER]` tags MUST NOT be on the same slide. **RULE 4: MATH EQUATIONS (FLAG-DECIDED - CRITICAL)** - The math flag for this request is: **{"ENABLED" if self.enable_maths else "DISABLED"}**. - `if self.enable_maths == True`: Use KaTeX. Block equations: `$$...$$`. Inline: `$ ... $`. DO NOT escape backslashes. - `if self.enable_maths == False`: You **MUST NOT** add any math equations. """ user_prompt = f"""{personalization_context}TOPIC: {topic} PROGRAMMING LANGUAGE: {programming_language} TASK: Create a 3-5 slide presentation to explain the foundational concepts of '{topic}'. Follow all rules in the system prompt precisely. **CRITICAL FORMAT REQUIREMENT:** - Output ONLY the slides in the exact format shown below. - NO thinking tags like ... - NO preamble, explanation, or commentary before or after the slides. - Start directly with "Slide 1:" {code_instructions} SLIDE FORMAT (MANDATORY - FOLLOW EXACTLY): Slide 1: Title: Content: - First bullet point (15-30 words, substantive content) - Second bullet point (15-30 words, substantive content) - Third bullet point (15-30 words, substantive content) - Fourth bullet point (15-30 words, substantive content) (Continue for 3-5 slides total) CONTENT RULES: - Each slide MUST have EXACTLY 4 bullet points. - Each bullet MUST start with "- " (dash and space). - Slide 1: Introduction and overview. - Slides 2-4: Core concepts. - Last Slide: Summary and key takeaways. Begin generating slides now (start with "Slide 1:"):""" completion = self._call_llm_with_fallback( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=0.6, max_tokens=2000, ) response = completion.choices[0].message.content or "" image_placeholder_count = response.count("[IMAGE_PLACEHOLDER]") viz_placeholder_count = response.count("[VISUALIZATION_PLACEHOLDER]") expected_image_count = 1 if self.enable_images else 0 needs_retry = False retry_reason = "" if image_placeholder_count != expected_image_count: retry_reason = f"LLM response failed image validation (Expected {expected_image_count}, Got {image_placeholder_count})." needs_retry = True elif self.enable_visualizations and viz_placeholder_count > 1: retry_reason = f"LLM response failed visualization validation (Flag ENABLED, Got {viz_placeholder_count}, expected 0 or 1)." needs_retry = True elif not self.enable_visualizations and viz_placeholder_count > 0: retry_reason = f"LLM response failed visualization validation (Flag DISABLED, Got {viz_placeholder_count}, expected 0)." needs_retry = True elif "[IMAGE_PLACEHOLDER][VISUALIZATION_PLACEHOLDER]" in response.replace("\n", "").replace(" ", "") or \ "[VISUALIZATION_PLACEHOLDER][IMAGE_PLACEHOLDER]" in response.replace("\n", "").replace(" ", ""): retry_reason = "LLM response failed validation (Placeholders on the same slide)." needs_retry = True viz_keywords = ["vector", "neural network", "gradient descent", "matrix", "algorithm", "data structure", "recursion", "linear regression"] topic_lower = topic.lower() is_viz_topic = any(keyword in topic_lower for keyword in viz_keywords) if not needs_retry and self.enable_visualizations and viz_placeholder_count == 0 and is_viz_topic: retry_reason = f"LLM failed visualization logic. Topic '{topic}' IS a visualization topic and REQUIRES a `[VISUALIZATION_PLACEHOLDER]`." needs_retry = True if not response or "Slide 1:" not in response or needs_retry: self.logger.warning(f"{retry_reason} Retrying with explicit correction.") correction_message = f""" Your last response FAILED. REASON: {retry_reason} YOUR (FAILED) RESPONSE: {response} --- This is UNACCEPTABLE. You MUST follow the rules from the system prompt. - **Rule 1 (Viz):** The visualization flag is `{"ENABLED" if self.enable_visualizations else "DISABLED"}`. You added {viz_placeholder_count} viz tags. - **Rule 2 (Image):** The image flag is `{"ENABLED" if self.enable_images else "DISABLED"}`. You MUST add {expected_image_count} image tags. You added {image_placeholder_count}. - **Rule 3 (No Overlap):** They cannot be on the same slide. Generate the slides again now, and this time, follow ALL rules. """ completion = self._call_llm_with_fallback( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, {"role": "assistant", "content": response}, {"role": "user", "content": correction_message}, ], temperature=0.5, max_tokens=2000, ) response = completion.choices[0].message.content or "" self.logger.info( f"Generated presentation text with {response.count('Slide ')} slides. " f"(Images: {response.count('[IMAGE_PLACEHOLDER]')}, Viz: {response.count('[VISUALIZATION_PLACEHOLDER]')})" ) return response except Exception as e: self.logger.error(f"Error generating presentation text: {e}", exc_info=True) return "Slide 1:\nTitle: Error in Generation\n\nContent:\n- Could not generate presentation content." def parse_slides(self, content: str) -> List[Dict]: slides = [] content = re.sub(r'.*?', '', content, flags=re.DOTALL) content = content.strip() pattern = r"Slide\s*(\d+):\s*\nTitle:\s*(.*?)\s*\n\n(.*?)(?=\n\nSlide\s*\d+:|\Z)" matches = re.findall(pattern, content, re.DOTALL) if not matches: self.logger.warning("Could not parse slides using primary pattern. Trying fallback.") pattern_fallback = r"Slide\s*(\d+)[:.]?\s*\n?Title[:.]?\s*(.*?)\n+(.*?)(?=\nSlide\s*\d+|\Z)" matches = re.findall(pattern_fallback, content, re.DOTALL | re.IGNORECASE) if not matches: pattern_alt1 = r"Slide\s*(\d+):\s*([^\n]+)\n+(.*?)(?=Slide\s*\d+:|\Z)" matches = re.findall(pattern_alt1, content, re.DOTALL | re.IGNORECASE) if not matches: pattern_alt2 = r"#+\s*Slide\s*(\d+)[:\s]+([^\n]+)\n+(.*?)(?=#+\s*Slide|\Z)" matches = re.findall(pattern_alt2, content, re.DOTALL | re.IGNORECASE) if not matches: pattern_alt3 = r"\*?\*?(\d+)[.)]\s*\*?\*?\s*([^\n*]+)\*?\*?\n+(.*?)(?=\*?\*?\d+[.)]|\Z)" matches = re.findall(pattern_alt3, content, re.DOTALL) if not matches: bullets = re.findall(r'^[-*]\s*(.+?)(?=\n[-*]|\n\n|\Z)', content, re.MULTILINE | re.DOTALL) if len(bullets) >= 3: for i, bullet in enumerate(bullets[:5], 1): sentences = bullet.split('.') title = sentences[0].strip()[:60] bullet_content = '. '.join(sentences[1:]).strip() if len(sentences) > 1 else bullet matches.append((str(i), title, f"- {bullet_content}")) if not matches: paragraphs = [p.strip() for p in content.split('\n\n') if p.strip() and len(p.strip()) > 20] if len(paragraphs) >= 2: for i, para in enumerate(paragraphs[:4], 1): title = para.split('.')[0][:50] if '.' in para else f"Key Point {i}" matches.append((str(i), title, f"- {para}")) else: return [{"number": "1", "title": "Generated Content", "content": content.strip(), "type": "content"}] for num, title, body in matches: cleaned_content = re.sub(r"Content:\s*", "", body).strip() if "[VISUALIZATION_PLACEHOLDER]" in cleaned_content: slide_type = "visualization" cleaned_content = cleaned_content.replace("[VISUALIZATION_PLACEHOLDER]", "").strip() elif "[IMAGE_PLACEHOLDER]" in cleaned_content: slide_type = "image" cleaned_content = cleaned_content.replace("[IMAGE_PLACEHOLDER]", "").strip() else: slide_type = "content" lines = cleaned_content.split('\n') filtered_lines = [] for line in lines: stripped = line.strip() if not stripped.startswith('-') or (stripped.startswith('-') and len(stripped) > 2 and stripped[1:].strip()): filtered_lines.append(line) cleaned_content = '\n'.join(filtered_lines).strip() if cleaned_content and not cleaned_content.startswith('-'): lines = cleaned_content.split('\n') cleaned_content = '\n'.join( f"- {line.strip()}" if line.strip() and not line.strip().startswith('-') else line for line in lines if line.strip() ) if cleaned_content: slides.append({ "number": num.strip(), "title": title.strip().strip("*_#"), "content": cleaned_content, "type": slide_type, }) if not slides: return [{"number": "1", "title": "Generated Content", "content": content.strip(), "type": "content"}] return slides def get_audio_duration(self, audio_path: str) -> float: if not os.path.exists(audio_path): self.logger.warning(f"Audio file not found at {audio_path}. Cannot get duration.") return 10.0 try: cmd = [ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", audio_path, ] result = subprocess.run(cmd, capture_output=True, text=True, check=True) return float(result.stdout.strip()) except Exception as e: self.logger.error(f"Could not get audio duration for {audio_path}: {e}", exc_info=True) return 10.0 async def render_slide_to_video(self, html_path: str, output_video_path: str, duration: float) -> None: render_dir = tempfile.mkdtemp() try: async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context( viewport={"width": 1920, "height": 1080}, record_video_dir=render_dir, record_video_size={"width": 1920, "height": 1080}, device_scale_factor=1, ) page = await context.new_page() await page.goto(f"file:///{os.path.abspath(html_path)}") await page.wait_for_load_state("networkidle") await page.wait_for_timeout(self.PLAYWRIGHT_AWAIT_DELAY) await asyncio.sleep(duration) await context.close() await browser.close() webm_files = [f for f in os.listdir(render_dir) if f.endswith(".webm")] if webm_files: shutil.move(os.path.join(render_dir, webm_files[0]), output_video_path) else: self.logger.error(f"Playwright did not generate a video file for {html_path}") finally: shutil.rmtree(render_dir) def add_audio_to_video(self, video_path: str, audio_path: str, output_path: str, audio_delay_ms: int = 0): if not os.path.exists(video_path): self.logger.error(f"Input video not found: {video_path}") return if not os.path.exists(audio_path): self.logger.error(f"Input audio not found: {audio_path}, creating silent video.") cmd = [ "ffmpeg", "-i", video_path, "-f", "lavfi", "-i", "anullsrc=channel_layout=mono:sample_rate=44100", "-c:v", "copy", "-c:a", "aac", "-shortest", "-y", output_path, ] subprocess.run(cmd, check=True, capture_output=True, text=True) return cmd = ["ffmpeg", "-i", video_path, "-i", audio_path] filter_complex_parts = [] if audio_delay_ms > 0: filter_complex_parts.append(f"[1:a]adelay={audio_delay_ms}|{audio_delay_ms}[aud]") else: filter_complex_parts.append("[1:a]acopy[aud]") cmd.extend([ "-filter_complex", "".join(filter_complex_parts), "-map", "0:v:0", "-map", "[aud]", "-c:v", "libx264", "-c:a", "aac", "-preset", "fast", "-crf", "23", "-y", output_path, ]) try: subprocess.run(cmd, check=True, capture_output=True, text=True) self.logger.info(f"Successfully created video with audio: {os.path.basename(output_path)}") except subprocess.CalledProcessError as e: self.logger.error(f"ffmpeg failed while adding audio to {os.path.basename(video_path)}: {e.stderr}") shutil.copy(video_path, output_path) def concatenate_videos(self, video_paths: List[str], output_path: str): self.logger.info(f"Concatenating {len(video_paths)} videos...") valid_videos = [v for v in video_paths if os.path.exists(v)] if len(valid_videos) < len(video_paths): self.logger.warning("Some slide videos were missing and will be skipped in concatenation.") if not valid_videos: raise ValueError("No valid video paths provided for concatenation.") cmd = ["ffmpeg"] filter_complex_parts = [] for i, path in enumerate(valid_videos): cmd.extend(["-i", path]) filter_complex_parts.append(f"[{i}:v:0][{i}:a:0]") filter_complex_string = "".join(filter_complex_parts) + f"concat=n={len(valid_videos)}:v=1:a=1[v][a]" cmd.extend([ "-filter_complex", filter_complex_string, "-map", "[v]", "-map", "[a]", "-c:v", "libx264", "-c:a", "aac", "-preset", "fast", "-crf", "23", "-y", output_path, ]) try: subprocess.run(cmd, check=True, capture_output=True, text=True) self.logger.info(f"Successfully concatenated videos into {output_path}") except subprocess.CalledProcessError as e: self.logger.error(f"ffmpeg concatenation failed: {e.stderr}") raise def _get_animation_delays(self, num_bullets: int, base_delay_s: float = 2.0, stagger_s: float = 0.0) -> Tuple[Dict, int]: delays = {} total_animation_time_s = base_delay_s for i in range(num_bullets): delay = base_delay_s + i * stagger_s delays[f"bullet_{i+1}_delay"] = f"{delay:.1f}s" total_animation_time_s = delay audio_delay_ms = int((total_animation_time_s + 1.0) * 1000) return delays, audio_delay_ms def _create_slide_video( self, folder_path: str, slide_name: str, template_name: Optional[str], template_data: Optional[Dict], audio_path: Optional[str], audio_delay_ms: int = 0, fixed_duration_s: Optional[float] = None, html_content: Optional[str] = None, add_post_delay: bool = True, ) -> Optional[str]: html_path = os.path.join(folder_path, f"{slide_name}.html") if html_content: with open(html_path, "w", encoding="utf-8") as f: f.write(html_content) elif template_name and template_data: template_path = f"src/template/slide/{template_name}" rendered_html = render_template(template_path, template_data) with open(html_path, "w", encoding="utf-8") as f: f.write(rendered_html) else: self.logger.error(f"Cannot create slide {slide_name}: No html_content or template_data provided.") return None audio_duration = self.get_audio_duration(audio_path) if audio_path else 0 pre_audio_delay_s = audio_delay_ms / 1000.0 post_audio_delay_s = 0.0 if not add_post_delay else 2.0 video_duration = ( fixed_duration_s if fixed_duration_s is not None else pre_audio_delay_s + audio_duration + post_audio_delay_s ) self.logger.info( f"Slide {slide_name}: " f"delay={pre_audio_delay_s:.1f}s + audio={audio_duration:.2f}s + post={post_audio_delay_s:.1f}s " f"= total={video_duration:.2f}s" ) video_path_silent = os.path.join(folder_path, f"{slide_name}_silent.webm") asyncio.run(self.render_slide_to_video(html_path, video_path_silent, duration=video_duration)) final_video_path = os.path.join(folder_path, f"{slide_name}.mp4") if audio_path: self.add_audio_to_video(video_path_silent, audio_path, final_video_path, audio_delay_ms) else: cmd = [ "ffmpeg", "-i", video_path_silent, "-f", "lavfi", "-i", "anullsrc=channel_layout=mono:sample_rate=44100", "-c:v", "libx264", "-c:a", "aac", "-shortest", "-y", final_video_path, ] subprocess.run(cmd, check=True, capture_output=True, text=True) if os.path.exists(video_path_silent): os.remove(video_path_silent) return final_video_path if os.path.exists(final_video_path) else None def is_programming_topic(self, topic, programming_language): if programming_language: lang_lower = programming_language.lower().strip() if lang_lower in ["none", "general", "", "n/a", "not applicable"]: return False programming_languages = [ "python", "javascript", "java", "c++", "c#", "ruby", "go", "rust", "typescript", "php", "swift", "kotlin", "r", "matlab", "sql", "html", "css", "react", "vue", "angular", "node", "django", "flask", ] if any(lang in lang_lower for lang in programming_languages): return True code_keywords = [ "programming", "code", "coding", "development", "algorithm", "function", "class", "method", "implementation", "script", "software", "api", "framework", "library", "syntax", "variable", "loop", "conditional", "debugging", "testing", ] return any(keyword in topic.lower() for keyword in code_keywords) def extract_inline_code(self, content: str) -> Tuple[str, Optional[str], Optional[str]]: code_pattern = r"(- Example:\s*\n)?\s*```(\w+)?\n(.*?)```" match = re.search(code_pattern, content, re.DOTALL | re.IGNORECASE) if not match: return content, None, None language = match.group(2) or "python" code = match.group(3).strip() if len(code.split("\n")) <= 7 and len(code) <= 400: cleaned_content = content.replace(match.group(0), "").strip() cleaned_content = "\n".join( line for line in cleaned_content.split("\n") if line.strip() and not line.strip() == "-" ) return cleaned_content, code, language return content, None, None def _generate_slide_audio( self, slide_data: Dict, folder_path: str, state: VideoGenerationState, slide_index: int ) -> Tuple[int, str]: slide = slide_data["slide"] previous_slide_title = slide_data.get("previous_slide_title") previous_slide_summary = slide_data.get("previous_slide_summary") self.logger.info(f"Generating audio for Slide {slide['number']} (Index: {slide_index}): {slide['title']}") narration_script = slide["content"] if slide["type"] == "visualization": narration_script = f"Here is a visualization of {slide['title']}. {slide['content']}" elif slide["type"] == "image": narration_script = f"Here is an image illustrating {slide['title']}. {slide['content']}" narration_text = self._generate_tts_optimized_narration( content=narration_script, title=slide["title"], topic=state.topic, state=state, previous_slide_title=previous_slide_title, previous_slide_summary=previous_slide_summary ) audio_path = audio_fn_from_string( input_text=narration_text, folder_path=folder_path, file_name_prefix=f"slide_{slide['number']}", target_language=(state.target_language or "english").lower(), tts_gender=state.tts_gender or "male", tts_voice_name=state.tts_voice or "Puck", toggle_hinglish=state.toggle_hinglish or False, ) return slide_index, audio_path def _generate_tts_optimized_narration( self, content: str, title: str, topic: str, state: VideoGenerationState, previous_slide_title: Optional[str] = None, previous_slide_summary: Optional[str] = None ) -> str: user_name = None target_audience = "beginners" if state.user_profile: user_name = state.user_profile.get("user_name") level = state.user_profile.get("experience_level", "") if level and level != "not_specified": target_audience = level tts_prompt = get_tts_narration_prompt( slide_content=content, slide_title=title, topic=topic, target_audience=target_audience, user_name=user_name, previous_slide_title=previous_slide_title, previous_slide_summary=previous_slide_summary ) try: narration_completion = self._call_llm_with_fallback( messages=[ { "role": "system", "content": "You are an expert at creating natural, conversational narration scripts optimized for text-to-speech systems. Output ONLY the narration text, no thinking tags or explanations." }, {"role": "user", "content": tts_prompt} ], temperature=0.7, max_tokens=300, task_name="content_generation" ) narration_text = narration_completion.choices[0].message.content or "" narration_text = re.sub(r'.*?', '', narration_text, flags=re.DOTALL).strip() if not narration_text or len(narration_text) < 20: self.logger.warning(f"Narration too short for '{title}'. Using fallback.") narration_text = f"Let's explore {title}. {content[:200]}..." return narration_text except Exception as e: self.logger.error(f"Error generating TTS narration: {e}") return f"Now let's discuss {title}. {content[:150]}..." def _generate_content_slide_data(self, slide: Dict, audio_path: str, folder_path: str) -> Dict: audio_duration = self.get_audio_duration(audio_path) if audio_path else 0 cleaned_content, inline_code, code_lang = self.extract_inline_code(slide["content"]) bullets_data = format_bullet_points(cleaned_content) bullets = bullets_data["bullets"] bullets_with_highlights = [] if self.enable_highlighting and audio_path and audio_duration > 0: try: self.logger.info(f"Processing bullets for word highlighting for slide {slide['number']}...") bullets_with_highlights = process_all_bullets_for_highlighting( slide_title=slide["title"], bullets=[b for b in bullets if b], audio_path=audio_path, audio_duration=audio_duration, client=self.client, base_delay=2.8, bullet_spacing=1.7, ) except Exception as e: self.logger.warning(f"Word highlighting failed for slide {slide['number']}: {e}") animation_delays, _ = self._get_animation_delays(len(bullets)) formatted_code = format_code_with_pygments(inline_code, code_lang) if inline_code else "" template_data = { "main_title": slide["title"], "bullets_with_highlights": bullets_with_highlights, "enable_word_highlighting": self.enable_highlighting and bool(bullets_with_highlights), "bullet_point_1": bullets[0] if len(bullets) > 0 else "", "bullet_point_2": bullets[1] if len(bullets) > 1 else "", "bullet_point_3": bullets[2] if len(bullets) > 2 else "", "bullet_point_4": bullets[3] if len(bullets) > 3 else "", "bullet_point_5": bullets[4] if len(bullets) > 4 else "", "bullet_point_6": bullets[5] if len(bullets) > 5 else "", "has_code_snippet": bool(inline_code), "code_snippet_content": formatted_code, "pygments_css": get_pygments_css() if inline_code else "", "logo_path": f'file:///{os.path.join(folder_path, "corner_logo.png")}', "circle_large_color": "#dbe8ff", "circle_small_color": "#2f6fec", "bullet_color": "#2f6fec", "subtitle_text": "Key points:", "logo_delay": "0s", "bullet_highlight_delays": "[2800, 4500, 6200, 7900, 9600, 11300]", "code_snippet_label": f"{code_lang.title()} Example:" if inline_code else "", "code_snippet_delay": "0s", } template_data.update(animation_delays) return { "slide": slide, "template_name": "content_slide.html", "template_data": template_data, "html_content": None, "audio_path": audio_path, "audio_delay_ms": self.AUDIO_DELAY_MS, "fixed_duration_s": None, } def _generate_image_slide_data(self, slide: Dict, audio_path: str, folder_path: str) -> Dict: image_path = None if self.enable_images: image_path = os.path.join(folder_path, f"slide_{slide['number']}_gen.png") try: generate_infographic_img(f"Title: {slide['title']}\n\nContent: {slide['content']}", image_path) if not os.path.exists(image_path): image_path = None except Exception as e: self.logger.error(f"Image generation failed for slide {slide['number']}: {e}") image_path = None use_image_template = self.enable_images and image_path and os.path.exists(image_path) slide_content = slide["content"].replace("[IMAGE_PLACEHOLDER]", "").strip() audio_duration = self.get_audio_duration(audio_path) if audio_path else 0 cleaned_content, inline_code, code_lang = self.extract_inline_code(slide_content) bullets = format_bullet_points(cleaned_content)["bullets"] bullets_with_highlights = [] if self.enable_highlighting and audio_path and audio_duration > 0: try: bullets_with_highlights = process_all_bullets_for_highlighting( slide_title=slide["title"], bullets=[b for b in bullets if b], audio_path=audio_path, audio_duration=audio_duration, client=self.client, base_delay=2.8, bullet_spacing=1.7, ) except Exception as e: self.logger.warning(f"Word highlighting failed for slide {slide['number']}: {e}") animation_delays, _ = self._get_animation_delays(len(bullets)) template_data = { "main_title": slide["title"], "image_path": os.path.basename(image_path) if use_image_template else '', "image_alt": slide["title"], "logo_path": f'file:///{os.path.join(folder_path, "corner_logo.png")}', "circle_large_color": "#dbe8ff", "circle_small_color": "#2f6fec", "bullet_color": "#2f6fec", "subtitle_text": "Key insights:", "bullets_with_highlights": bullets_with_highlights, "enable_word_highlighting": self.enable_highlighting and bool(bullets_with_highlights), "bullet_point_1": bullets[0] if len(bullets) > 0 else "", "bullet_point_2": bullets[1] if len(bullets) > 1 else "", "bullet_point_3": bullets[2] if len(bullets) > 2 else "", "bullet_point_4": bullets[3] if len(bullets) > 3 else "", "bullet_point_5": bullets[4] if len(bullets) > 4 else "", "bullet_point_6": bullets[5] if len(bullets) > 5 else "", "logo_delay": "0s", "bullet_highlight_delays": "[3800, 5500, 7200, 8900, 10600, 12300]", "has_code_snippet": bool(inline_code), "code_snippet_content": format_code_with_pygments(inline_code, code_lang) if inline_code else "", "pygments_css": get_pygments_css() if inline_code else "", "code_snippet_label": f"{code_lang.title()} Example:" if inline_code else "", "code_snippet_delay": "0s", } template_data.update(animation_delays) return { "slide": slide, "template_name": "image_slide.html" if use_image_template else "content_slide.html", "template_data": template_data, "html_content": None, "audio_path": audio_path, "audio_delay_ms": self.AUDIO_DELAY_MS, "fixed_duration_s": None, } def _generate_visualization_slide_data(self, slide: Dict, audio_path: str, folder_path: str, presentation_topic: str) -> Dict: self.logger.info(f"Generating visualization slide data for: {slide['title']}") template_base_path = os.path.abspath("src/template/slide/visualization_base.html") if not os.path.exists(template_base_path): self.logger.error("MISSING TEMPLATE: 'src/template/slide/visualization_base.html' not found.") self.logger.warning("Falling back to a standard content slide for visualization.") return self._generate_content_slide_data(slide, audio_path, folder_path) with open(template_base_path, "r", encoding="utf-8") as f: template_html = f.read() visualization_topic = f"{presentation_topic}: {slide['title']}" complete_html = get_complete_html_page( topic=visualization_topic, template_html=template_html, logger=self.logger ) debug_path = os.path.join(folder_path, f"slide_{slide['number']}_viz_debug.html") with open(debug_path, "w", encoding="utf-8") as f: f.write(complete_html) default_viz_duration_s = 12.0 audio_delay_ms = self.AUDIO_DELAY_MS audio_duration_s = self.get_audio_duration(audio_path) if audio_path else 0 pre_audio_delay_s = audio_delay_ms / 1000.0 post_audio_delay_s = 2.0 required_audio_duration_s = pre_audio_delay_s + audio_duration_s + post_audio_delay_s final_duration_s = max(default_viz_duration_s, required_audio_duration_s) return { "slide": slide, "template_name": None, "template_data": None, "html_content": complete_html, "audio_path": audio_path, "audio_delay_ms": audio_delay_ms, "fixed_duration_s": final_duration_s, } def _process_slide_video(self, slide_index: int, slide_video_data: Dict, folder_path: str) -> Tuple[int, str]: slide = slide_video_data["slide"] self.logger.info(f"Creating video for Slide {slide['number']} (Index: {slide_index}): {slide['title']}") video_path = self._create_slide_video( folder_path, f"slide_{slide['number']}", slide_video_data.get("template_name"), slide_video_data.get("template_data"), slide_video_data.get("audio_path"), slide_video_data.get("audio_delay_ms", 0), fixed_duration_s=slide_video_data.get("fixed_duration_s"), html_content=slide_video_data.get("html_content"), ) return slide_index, video_path def generate_slides(self, state: VideoGenerationState) -> VideoGenerationState: if isinstance(state.target_language, list): state.target_language = [ (lang or "english").strip().lower() for lang in state.target_language ] else: state.target_language = ( state.target_language or "english" ).strip().lower() target_languages = state.target_language if isinstance(target_languages, list): self.logger.info(f"Multi-language mode detected: {target_languages}") all_video_urls = [] for lang in target_languages: self.logger.info(f"Processing language: {lang.upper()}") lang_state = VideoGenerationState( session_id=state.session_id, topic=state.topic, subtitle=state.subtitle, programming_language=state.programming_language, slide_colour=getattr(state, 'slide_colour', None), target_language=(lang or "english").strip().lower(), tts_gender=state.tts_gender, tts_voice=state.tts_voice, video_type=state.video_type, user_profile=state.user_profile, optional_params=getattr(state, 'optional_params', None), context_metadata=getattr(state, 'context_metadata', None), toggle_hinglish=getattr(state, 'toggle_hinglish', False) ) result = self._generate_slides_for_language(lang_state, skip_json_upload=True) if result.slide_video_path: video_url = list(result.slide_video_path.values())[0] normalized_lang = (lang or "english").strip().lower() all_video_urls.append({normalized_lang: video_url}) state.status = result.status state.error = result.error merged_videos = {} for video_dict in all_video_urls: merged_videos.update(video_dict) state.slide_videos = merged_videos if merged_videos else None state.slide_video_path = merged_videos if merged_videos else None state_json_key = f"video_states/{state.session_id}/{state.session_id}.json" self.s3_client.put_object( Bucket=self.bucket_name, Key=state_json_key, Body=json.dumps(state.model_dump(), indent=2), ContentType="application/json" ) self.logger.info(f"[JSON UPLOAD] Uploaded merged state JSON to s3://{self.bucket_name}/{state_json_key}") return state else: return self._generate_slides_for_language(state) def _generate_slides_for_language(self, state: VideoGenerationState, skip_json_upload: bool = False) -> VideoGenerationState: state.target_language = (state.target_language or "english").strip().lower() self.enable_highlighting = state.enable_highlighting if hasattr(state, 'enable_highlighting') else False self.enable_images = state.enable_images if hasattr(state, 'enable_images') else True self.enable_visualizations = state.enable_visualizations if hasattr(state, 'enable_visualizations') else True self.enable_maths = state.enable_maths if hasattr(state, 'enable_maths') else False self.logger.info(f"Starting slide generation for topic: {state.topic}") self.logger.info(f"Target language: {state.target_language or 'english'}") temp_dir = tempfile.mkdtemp() try: self.logger.info(f"Session initialized: session_id={state.session_id}, topic={state.topic}") safe_topic = sanitize_filename(state.topic) folder_path = os.path.join(temp_dir, "output", state.session_id, safe_topic) os.makedirs(folder_path, exist_ok=True) asset_path = get_assets_dir() shutil.copy(os.path.join(asset_path, "logo.svg"), folder_path) shutil.copy(os.path.join(asset_path, "corner_logo.png"), folder_path) # user_profile is already populated by the router node (via UserInfoRetriever). # For personalised_video, use it directly. For base_video, leave it empty. user_profile = state.user_profile or {} if state.video_type == "personalised_video" else {} use_cache = ( self.cached_presentation_content is not None and self.cached_presentation_content.get('session_id') == state.session_id ) if use_cache: self.logger.info(f"Using cached presentation content for session {state.session_id}") presentation_text = self.cached_presentation_content['presentation_text'] slides_data = self.cached_presentation_content['slides_data'] welcome_script = self.cached_presentation_content['welcome_script'] analogy_text = self.cached_presentation_content['analogy_text'] code_data = self.cached_presentation_content.get('code_data') else: self.logger.info(f"Generating new presentation content for session {state.session_id}") # No namespace or Pinecone calls here — profile already in state.user_profile presentation_text = self.generate_presentation_text( state.topic, state.programming_language, user_profile, ) presentation_text_path = os.path.join(folder_path, "presentation.txt") with open(presentation_text_path, "w", encoding="utf-8") as f: f.write(presentation_text) slides_data = self.parse_slides(presentation_text) if use_cache: presentation_text_path = os.path.join(folder_path, "presentation.txt") with open(presentation_text_path, "w", encoding="utf-8") as f: f.write(presentation_text) video_paths = [] # Welcome Slide welcome_video = None suppress_welcome = False if state.optional_params and isinstance(state.optional_params, dict): suppress_welcome = state.optional_params.get("suppress_welcome", False) if state.context_metadata and isinstance(state.context_metadata, dict): suppress_welcome = suppress_welcome or state.context_metadata.get("suppress_welcome", False) if not suppress_welcome: self.logger.info("Creating Welcome Slide...") if not use_cache: # user_name comes from state.user_profile — no Pinecone call needed user_name = (state.user_profile or {}).get("user_name") if user_name: welcome_script = ( f"Hello {user_name}! Welcome to Vidya. " f"Today we're diving deep into {state.topic}. " f"This is an amazing topic that will transform how you think. " f"Let's learn together and make it awesome!" ) self.logger.info(f"Personalized welcome for: {user_name}") else: welcome_script = ( f"Welcome to Vidya! " f"Today we're diving deep into {state.topic}. " f"This is an amazing topic that will transform how you think. " f"Let's learn together and make it awesome!" ) self.logger.info("No user name found in profile, using generic welcome") else: self.logger.info(f"Using cached welcome script for session {state.session_id}") welcome_audio = audio_fn_from_string( input_text=welcome_script, folder_path=folder_path, file_name_prefix="welcome_slide", target_language=state.target_language or "english", tts_gender=state.tts_gender or "male", tts_voice_name=state.tts_voice or "Puck", toggle_hinglish=state.toggle_hinglish or False, ) audio_duration = self.get_audio_duration(welcome_audio) if welcome_audio else 0 self.logger.info(f"Welcome audio duration: {audio_duration:.2f}s") welcome_data = { "main_title": state.topic, "subtitle_text": state.subtitle or "An Introduction", "logo_path": f'file:///{os.path.join(folder_path, "logo.svg")}', "circle_large_color": "#dbe8ff", "circle_small_color": "#2f6fec", "logo_animation_duration": 0, "welcome_content_delay": 0, "audio_start_delay": 0, "total_duration": 5, } """welcome_video = self._create_slide_video( folder_path, "welcome_slide", "welcome_slide.html", welcome_data, welcome_audio, audio_delay_ms=500, fixed_duration_s=None, add_post_delay=False )""" welcome_video = self._create_slide_video( folder_path, "welcome_slide", "welcome_slide.html", welcome_data, welcome_audio, audio_delay_ms=500, fixed_duration_s=None, add_post_delay=False ) if welcome_video: video_paths.append(welcome_video) # Generate content slide audio in parallel self.logger.info("Generating audio for content slides in parallel...") slide_audio_paths = [None] * len(slides_data) with ThreadPoolExecutor(max_workers=4) as executor: audio_tasks = [] for i, slide in enumerate(slides_data): previous_slide_title = slides_data[i - 1]["title"] if i > 0 else None previous_slide_summary = f"{slides_data[i - 1]['content'][:150]}..." if i > 0 else None task = executor.submit( self._generate_slide_audio, { "slide": slide, "index": i, "previous_slide_title": previous_slide_title, "previous_slide_summary": previous_slide_summary }, folder_path, state, i ) audio_tasks.append(task) for future in as_completed(audio_tasks): try: slide_index, audio_path = future.result() slide_audio_paths[slide_index] = audio_path except Exception as e: self.logger.error(f"Error generating audio for slide: {e}") # Prepare slide template data self.logger.info("Preparing slide template data...") slide_video_data_list = [] for i, slide in enumerate(slides_data): audio_path = slide_audio_paths[i] if slide["type"] == "visualization" and self.enable_visualizations: slide_video_data = self._generate_visualization_slide_data(slide, audio_path, folder_path, state.topic) elif slide['type'] == 'image': slide_video_data = self._generate_image_slide_data(slide, audio_path, folder_path) else: if slide["type"] == "visualization": self.logger.warning(f"Slide {slide['number']} was 'visualization' but feature is DISABLED.") slide_video_data = self._generate_content_slide_data(slide, audio_path, folder_path) slide_video_data_list.append(slide_video_data) # Generate slide videos in parallel self.logger.info("Generating slide videos in parallel...") content_video_paths = [None] * len(slides_data) with ThreadPoolExecutor(max_workers=3) as executor: video_tasks = [] for i, slide_video_data in enumerate(slide_video_data_list): task = executor.submit(self._process_slide_video, i, slide_video_data, folder_path) video_tasks.append(task) for future in as_completed(video_tasks): try: slide_index, video_path = future.result() content_video_paths[slide_index] = video_path except Exception as e: self.logger.error(f"Error generating video for slide: {e}", exc_info=True) for video_path in content_video_paths: if video_path: video_paths.append(video_path) # Code Slide code_data = None if self.is_programming_topic(state.topic, state.programming_language): self.logger.info("Creating Code Slide...") if not use_cache: code_data = generate_code_example(state.topic, state.programming_language, presentation_text) else: code_data = self.cached_presentation_content.get('code_data') self.logger.info(f"Using cached code data for session {state.session_id}") if code_data: formatted_code = format_code_with_pygments(code_data["code"], code_data["language"]) code_template_data = { "main_title": code_data["title"], "code_header_title": f"{code_data['language'].title()} Example", "code_content": formatted_code, "pygments_css": get_pygments_css(), "logo_path": f'file:///{os.path.join(folder_path, "corner_logo.png")}', "circle_large_color": "#dbe8ff", "circle_small_color": "#2f6fec", "code_title_color": "#2f6fec", "subtitle_text": "Implementation:", "logo_delay": "0s", "particle_color": "#dbe8ff", } code_narration = self._generate_tts_optimized_narration( content=code_data["explanation"], title=code_data["title"], topic=state.topic, state=state ) code_audio = audio_fn_from_string( input_text=code_narration, folder_path=folder_path, file_name_prefix="code_slide", target_language=state.target_language or "english", tts_gender=state.tts_gender or "male", tts_voice_name=state.tts_voice or "Puck", toggle_hinglish=state.toggle_hinglish or False, ) code_audio = audio_fn_from_string( input_text=code_narration, folder_path=folder_path, file_name_prefix="code_slide", target_language=state.target_language or "english", tts_gender=state.tts_gender or "male", tts_voice_name=state.tts_voice or "Puck", toggle_hinglish=state.toggle_hinglish or False, ) code_video = self._create_slide_video( folder_path, "code_slide", "code_slide.html", code_template_data, code_audio, audio_delay_ms=self.AUDIO_DELAY_MS ) if code_video: video_paths.append(code_video) else: self.logger.warning("Code generation failed, skipping code slide.") # Analogy Slide self.logger.info("Creating Analogy Slide...") analogy_text = None if not use_cache: analogy_completion = self._call_llm_with_fallback( messages=[ { "role": "system", "content": "You create simple, relatable analogies. Output ONLY the analogy content as 3-4 bullet points starting with '-'. No thinking tags, no explanations, no preamble." }, { "role": "user", "content": f"Generate a simple, real-world analogy for '{state.topic}' suitable for a presentation slide. Present using 3 or 4 clear bullet points starting with '- '. Do not use the word 'analogy'. Output ONLY the bullet points." } ] ) analogy_text = analogy_completion.choices[0].message.content or "" analogy_text = re.sub(r'.*?', '', analogy_text, flags=re.DOTALL).strip() if not analogy_text or len(analogy_text) < 20: self.logger.warning("Analogy generation returned empty or short response. Using fallback.") analogy_text = ( f"- {state.topic} is like learning a new skill - it takes practice and patience.\n" f"- Start with the basics and build up gradually.\n" f"- Once you understand the fundamentals, complex concepts become easier.\n" f"- Practice regularly to reinforce your understanding." ) else: analogy_text = self.cached_presentation_content['analogy_text'] self.logger.info(f"Using cached analogy text for session {state.session_id}") bullets = format_bullet_points(analogy_text)["bullets"] if len(bullets) < 3: bullets = [ f"{state.topic} is like learning a new skill - it takes practice", "Start with the basics and build up gradually", "Once you understand fundamentals, complex concepts become easier", "Practice regularly to reinforce your understanding" ] analogy_narration = self._generate_tts_optimized_narration( content=" ".join(bullets), title="A Real-World Analogy", topic=state.topic, state=state ) audio_path_analogy = audio_fn_from_string( input_text=analogy_narration, folder_path=folder_path, file_name_prefix="analogy_slide", target_language=state.target_language or "english", tts_gender=state.tts_gender or "male", tts_voice_name=state.tts_voice or "Puck", toggle_hinglish=state.toggle_hinglish or False, ) bullets_with_highlights = [] if self.enable_highlighting and audio_path_analogy: audio_duration_analogy = self.get_audio_duration(audio_path_analogy) try: bullets_with_highlights = process_all_bullets_for_highlighting( slide_title="A Real-World Analogy", bullets=bullets, audio_path=audio_path_analogy, audio_duration=audio_duration_analogy, client=self.client, base_delay=2.8, bullet_spacing=1.7, ) except Exception as e: self.logger.warning(f"Word highlighting failed for analogy slide: {e}") animation_delays, _ = self._get_animation_delays(len(bullets)) analogy_data = { "main_title": "A Real-World Analogy", "bullets_with_highlights": bullets_with_highlights, "enable_word_highlighting": self.enable_highlighting and bool(bullets_with_highlights), "bullet_point_1": bullets[0] if len(bullets) > 0 else "", "bullet_point_2": bullets[1] if len(bullets) > 1 else "", "bullet_point_3": bullets[2] if len(bullets) > 2 else "", "bullet_point_4": bullets[3] if len(bullets) > 3 else "", "bullet_point_5": bullets[4] if len(bullets) > 4 else "", "bullet_point_6": bullets[5] if len(bullets) > 5 else "", "has_code_snippet": False, "code_snippet_content": "", "pygments_css": "", "logo_path": f'file:///{os.path.join(folder_path, "corner_logo.png")}', "circle_large_color": "#dbe8ff", "circle_small_color": "#2f6fec", "bullet_color": "#2f6fec", "subtitle_text": "Key points:", "logo_delay": "0s", "bullet_highlight_delays": "[2800, 4500, 6200, 7900, 9600, 11300]", "code_snippet_label": "", "code_snippet_delay": "12s", "code_lang": "", } analogy_data.update(animation_delays) analogy_video = self._create_slide_video( folder_path, "analogy_slide", "content_slide.html", analogy_data, audio_path_analogy, audio_delay_ms=self.AUDIO_DELAY_MS ) if analogy_video: video_paths.append(analogy_video) if not use_cache: self.cached_presentation_content = { 'session_id': state.session_id, 'presentation_text': presentation_text, 'slides_data': slides_data, 'welcome_script': welcome_script if not suppress_welcome else None, 'analogy_text': analogy_text, 'code_data': code_data if self.is_programming_topic(state.topic, state.programming_language) else None } # Concatenate all videos final_video_path = os.path.join(folder_path, "slide_video.mp4") self.concatenate_videos(video_paths, final_video_path) s3_key = f"video_states/{state.session_id}/slide_video/{state.target_language}/slide_video.mp4" s3_url = self._upload_to_s3(final_video_path, s3_key) state.slide_videos = {state.target_language: s3_url} state.slide_video_path = {state.target_language: s3_url} state.status = "slide_video_generated" self.logger.info(f"Successfully generated {state.target_language} video: {s3_url}") except Exception as e: state.error = f"An unexpected error occurred in SlideCreationNode: {str(e)}" state.status = "slide_generation_failed" self.logger.error(f"Fatal error in slide generation pipeline: {e}", exc_info=True) raise finally: if not skip_json_upload: state_json_key = f"video_states/{state.session_id}/{state.session_id}.json" self.s3_client.put_object( Bucket=self.bucket_name, Key=state_json_key, Body=json.dumps(state.model_dump(), indent=2), ContentType="application/json" ) self.logger.info(f"[JSON UPLOAD] Uploaded state JSON to s3://{self.bucket_name}/{state_json_key}") if os.path.exists(temp_dir) and os.path.isdir(temp_dir): shutil.rmtree(temp_dir) self.logger.info(f"Cleaned up temporary directory: {temp_dir}") return state def process(state: VideoGenerationState) -> VideoGenerationState: slide_creator = SlideCreationNode() return slide_creator.generate_slides(state) def run(state: VideoGenerationState) -> VideoGenerationState: node = SlideCreationNode() return node.generate_slides(state)