Spaces:
Runtime error
Runtime error
| from typing import Optional, List, Dict, Any, Union | |
| from pydantic import BaseModel | |
| import os | |
| import re | |
| from src.tools.assets_utils import sanitize_filename | |
| class CodeSegment(BaseModel): | |
| code_snippet: str | |
| explanation: str | |
| audio_path: Optional[str] = None | |
| audio_duration: Optional[float] = None | |
| start_char_idx: Optional[int] = None | |
| end_char_idx: Optional[int] = None | |
| class VideoGenerationState(BaseModel): | |
| # User-provided inputs | |
| topic: Optional[str] = None | |
| course_name: Optional[str] = None | |
| subtitle: Optional[str] = None | |
| programming_language: Optional[str] = None | |
| slide_colour: Optional[str] = "blue" | |
| # Allow target_language to be either a single language string or list of languages | |
| target_language: Optional[Union[str, List[str]]] = None | |
| tts_gender: Optional[str] = "male" | |
| visualization_plan: Optional[Dict[str, Any]] = None | |
| tts_voice: Optional[str] = "kore" | |
| toggle_hinglish: Optional[bool] = False | |
| video_type: Optional[str] = "base_video" # or "personalised_video" | |
| user_code_input: Optional[str] = None | |
| # Memory & Context | |
| session_id: Optional[str] = None | |
| user_profile: Optional[Dict[str, Any]] = None # pulled from DB or vector store | |
| # Intermediate content | |
| presentation_txt: Optional[str] = None | |
| slides: Optional[List[str]] = None | |
| rendered_slides: Optional[List[str]] = None # images or paths | |
| slide_contents: Optional[List[Dict[str, Any]]] = None | |
| code_blocks: Optional[List[str]] = None | |
| segmented_code_data: Optional[List[CodeSegment]] = None | |
| # Video paths - ALL path fields must be Dict[str, str] keyed by language | |
| web_output_screenshot_path: Optional[str] = None | |
| rendered_web_videos: Optional[List[str]] = None | |
| slide_videos: Optional[Dict[str, str]] = None | |
| visualization_videos: Optional[List[str]] = None # Added this field | |
| slide_video_path: Optional[Dict[str, str]] = None | |
| final_avatar_video: Optional[Dict[str, str]] = None | |
| web_code_file_paths: Optional[List[str]] = None | |
| final_avatar_video_path: Optional[Dict[str, str]] = None | |
| code_video_path: Optional[Dict[str, str]] = None | |
| ppt_viz_path: Optional[str] = None | |
| final_video_path: Optional[Dict[str, str]] = None | |
| master_audio_path: Optional[str] = None | |
| # Multi-language & audio flags | |
| generate_multi_language: Optional[bool] = False | |
| languages: Optional[List[str]] = None | |
| upload_audio_to_s3: Optional[bool] = False | |
| multi_audio_mode: Optional[str] = "per_language_videos" # or 'single_video_multi_audio_tracks' | |
| audio_s3_urls: Optional[Dict[str, str]] = None | |
| temp_frame_dir: Optional[str] = None | |
| actual_execution_output: Optional[str] = None | |
| upload_link: Optional[Union[str, Dict[str, Optional[str]]]] = None | |
| # Error handling | |
| error: Optional[str] = None | |
| subtitle_content: Optional[Union[str, Dict[str, Optional[str]]]] = None | |
| # Additional fields for input node | |
| http_headers: Optional[Dict[str, Any]] = None | |
| optional_params: Optional[Dict[str, Any]] = None | |
| context_metadata: Optional[Dict[str, Any]] = None | |
| memory_retrieval_status: Optional[str] = None | |
| fallback_indicator: Optional[bool] = None | |
| status: str = "pending" | |
| output_type: str = "terminal" | |
| # visualisation options | |
| enable_maths: Optional[bool] = False | |
| enable_highlighting: Optional[bool] = False | |
| enable_images: Optional[bool] = True | |
| enable_visualizations: Optional[bool] = True | |
| def get_base_dirs(self): | |
| """Get base directories for outputs.""" | |
| # Create topic-specific directory (sanitize topic for filesystem safety) | |
| current_dir = os.path.dirname(os.path.abspath(__file__)) | |
| safe_topic = sanitize_filename(self.topic) | |
| base_dir = os.path.join(current_dir, "output", safe_topic, "visualisation") | |
| os.makedirs(base_dir, exist_ok=True) | |
| return { | |
| "base_dir": base_dir, | |
| "web_code_dir": os.path.join(base_dir, "web_code"), | |
| "media_dir": os.path.join(base_dir, "media"), | |
| "audio_dir": os.path.join(base_dir, "audio"), | |
| "oration_dir": os.path.join(base_dir, "oration"), | |
| "combined_dir": os.path.join(base_dir, "combined"), | |
| "temp_dir": os.path.join(base_dir, "temp") | |
| } | |
| def _safe_title(self, title): | |
| """Convert title to safe filename.""" | |
| safe = re.sub(r'[^\w\s-]', '', str(title)).strip().replace(' ', '_').lower() | |
| return re.sub(r'_+', '_', safe) | |
| def set_error(self, message: str): | |
| """Set error message and update status.""" | |
| self.error = message | |
| self.status = 'failed' |