import os import json from datetime import datetime import re from pptx import Presentation as PPTXPresentation from pydantic import BaseModel, Field class Tools: # Known template placeholders for validation. TEMPLATE_PLACEHOLDERS = { "TITLE", "Title", "Description or subtitle", "Agenda", "Section title 1", "Subheading 1", "Bullet 1 section 1", "Bullet 2 section 1", "Bullet 3 section 1", "Section title 2", "Bullet 1 section 2", "Bullet 2 section 2", "Bullet 3 section 2", "Section title 3", "Bullet 1 section 3", "Bullet 2 section 3", "Bullet 3 section 3", "Section title 4", "Bullet 1 section 4", "Bullet 2 section 4", "Bullet 3 section 4", "Bullet 4 section 4", "Bullet 5 section 4", "Bullet 6 section 4", "Charts and graphs", "Chart 1 title", "Chart 2 title", } class Valves(BaseModel): template_path: str = Field( default="/app/backend/data/template.pptx", description="Path to the PowerPoint template", ) output_dir: str = Field( default="/app/backend/data/", description="Directory where filled templates will be saved", ) validate_all_placeholders: bool = Field( default=False, description="Whether to validate that all placeholders have replacements", ) def __init__(self): """Initialize the tool and its configuration.""" self.valves = self.Valves() async def validate_template_exists(self, __event_emitter__=None) -> bool: """Validate that the template file exists.""" if not os.path.exists(self.valves.template_path): if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": f"Template not found at {self.valves.template_path}", "done": True, }, } ) print(f"Template not found at {self.valves.template_path}") return False return True async def validate_replacements(self, replacements: dict) -> tuple[bool, str]: """Validate that replacements is a dictionary (and optionally complete).""" if not isinstance(replacements, dict): return False, "Invalid input: Expected a dictionary." if not self.valves.validate_all_placeholders: return True, "" missing = self.TEMPLATE_PLACEHOLDERS - set(replacements.keys()) if missing: return False, f"Missing replacements for placeholders: {', '.join(missing)}" return True, "" async def fill_template( self, replacements: dict, filename_prefix: str, __event_emitter__=None ) -> str: """ Fill the PowerPoint template with the provided JSON replacements and save with a unique name. :param replacements: JSON dictionary with keys matching the template placeholders. :param filename_prefix: Prefix for the output filename. """ try: # Debug: print template path and output directory. print(f"Template path: {self.valves.template_path}") print(f"Output directory: {self.valves.output_dir}") # Validate that the template exists. if not await self.validate_template_exists(__event_emitter__): result = { "status": "error", "message": f"Template not found at {self.valves.template_path}", } print("Template not found error:", result) return json.dumps(result) # Validate the provided replacements. valid, message = await self.validate_replacements(replacements) if not valid: result = {"status": "error", "message": message} print("Validation error:", result) return json.dumps(result) # Ensure the replacements can be serialized as JSON. try: json.dumps(replacements, ensure_ascii=False) except Exception as e: result = { "status": "error", "message": f"Invalid JSON in replacements: {str(e)}", } print("JSON error:", result) return json.dumps(result) print("Loading template...") if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": "Loading template and applying replacements...", "done": False, }, } ) # Load the presentation template. prs = PPTXPresentation(self.valves.template_path) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_filename = f"{filename_prefix}_{timestamp}.pptx" output_path = os.path.join(self.valves.output_dir, output_filename) print(f"Will save to: {output_path}") def replace_text(text: str) -> str: if not isinstance(text, str): return text # Replace any placeholders wrapped in curly braces with the corresponding value. for key, value in replacements.items(): pattern = re.compile( r"\{\s*" + re.escape(key) + r"\s*\}", re.IGNORECASE ) text = pattern.sub(str(value), text) return text print("Processing slides...") for slide in prs.slides: for shape in slide.shapes: if hasattr(shape, "text_frame") and shape.text_frame: for paragraph in shape.text_frame.paragraphs: for run in paragraph.runs: original_text = run.text updated_text = replace_text(run.text) run.text = updated_text print( f"Replaced: '{original_text}' -> '{updated_text}'" ) os.makedirs(self.valves.output_dir, exist_ok=True) print(f"Saving to {output_path}...") prs.save(output_path) print(f"Save completed: {os.path.exists(output_path)}") if os.path.exists(output_path): result = { "status": "success", "filename": output_filename, "path": output_path, } print("Success:", result) if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": f"Presentation saved as {output_filename}", "done": True, }, } ) return f"Successfully created presentation: {output_filename}" else: result = { "status": "error", "message": f"File was not created at {output_path}", } print("File not created error:", result) return json.dumps(result, ensure_ascii=False) except Exception as e: print("Exception:", str(e)) if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": f"Error: {str(e)}", "done": True, }, } ) return f"Error filling template: {str(e)}" async def generate_and_create_presentation( self, generated_replacements: dict, filename_prefix: str, __event_emitter__=None ) -> str: """ Orchestrate the workflow by taking externally provided JSON replacements (from the LLM) and filling the template with that content. :param generated_replacements: JSON dictionary generated by the model. :param filename_prefix: Prefix for the output filename. """ try: print("Using externally generated content for presentation...") print("Received replacements:", generated_replacements) result = await self.fill_template( generated_replacements, filename_prefix, __event_emitter__ ) print("Presentation creation completed.") return result except Exception as e: return json.dumps( { "status": "error", "message": f"Error in generate_and_create_presentation: {str(e)}", } )