from typing import Dict, Any, Optional import logging import os from pathlib import Path from modules.methods.methodBase import MethodBase, AuthSource, MethodResult from modules.models.userConnection import UserConnection from modules.models.account import Account from modules.protocols.msGraphProtocol import MSGraphProtocol logger = logging.getLogger(__name__) class MethodPowerpoint(MethodBase): """Powerpoint method implementation for PowerPoint operations""" def __init__(self): super().__init__() self.name = "powerpoint" self.description = "Handle PowerPoint operations like reading, writing, and converting presentations" self.authSource = AuthSource.MICROSOFT # PowerPoint operations need Microsoft auth @property def actions(self) -> Dict[str, Dict[str, Any]]: """Available actions and their parameters""" return { "read": { "description": "Read PowerPoint presentation content", "retryMax": 2, "timeout": 30, "parameters": { "path": {"type": "string", "required": True}, "format": {"type": "string", "required": False}, "includeNotes": {"type": "boolean", "required": False} } }, "write": { "description": "Write content to PowerPoint presentation", "retryMax": 2, "timeout": 60, "parameters": { "path": {"type": "string", "required": True}, "content": {"type": "object", "required": True}, "template": {"type": "string", "required": False} } }, "convert": { "description": "Convert PowerPoint presentation between formats", "retryMax": 2, "timeout": 60, "parameters": { "sourcePath": {"type": "string", "required": True}, "targetPath": {"type": "string", "required": True}, "sourceFormat": {"type": "string", "required": False}, "targetFormat": {"type": "string", "required": False} } }, "createPresentation": { "description": "Create a new PowerPoint presentation", "retryMax": 2, "timeout": 60, "parameters": { "title": {"type": "string", "required": True}, "template": {"type": "string", "required": False} } }, "addSlide": { "description": "Add a new slide to presentation", "retryMax": 2, "timeout": 60, "parameters": { "presentationId": {"type": "string", "required": True}, "layout": {"type": "string", "required": False}, "title": {"type": "string", "required": False} } }, "addContent": { "description": "Add content to a slide", "retryMax": 2, "timeout": 60, "parameters": { "presentationId": {"type": "string", "required": True}, "slideId": {"type": "string", "required": True}, "contentType": {"type": "string", "required": True}, "content": {"type": "object", "required": True}, "position": {"type": "object", "required": False} } } } async def execute(self, action: str, parameters: Dict[str, Any], authData: Optional[Dict[str, Any]] = None) -> MethodResult: """Execute PowerPoint method""" try: # Validate parameters if not await self.validateParameters(action, parameters): return self._createResult( success=False, data={"error": f"Invalid parameters for {action}"} ) # Get UserConnection from auth_data if not authData or "userConnection" not in authData: return self._createResult( success=False, data={"error": "UserConnection required for PowerPoint operations"} ) userConnection: UserConnection = authData["userConnection"] # Execute action if action == "createPresentation": return await self._createPresentation(parameters, userConnection) elif action == "addSlide": return await self._addSlide(parameters, userConnection) elif action == "addContent": return await self._addContent(parameters, userConnection) else: return self._createResult( success=False, data={"error": f"Unknown action: {action}"} ) except Exception as e: logger.error(f"Error executing PowerPoint {action}: {e}") return self._createResult( success=False, data={"error": str(e)} ) async def _read_presentation(self, parameters: Dict[str, Any], authData: Dict[str, Any]) -> MethodResult: """Read PowerPoint presentation content""" try: path = Path(parameters["path"]) if not path.exists(): return self._createResult( success=False, data={"error": f"File not found: {path}"} ) # Determine format if not specified format = parameters.get("format") if not format: format = path.suffix[1:] if path.suffix else "pptx" # TODO: Implement PowerPoint reading using Microsoft Graph API # This is a placeholder implementation return self._createResult( success=True, data={ "path": str(path), "format": format, "slides": [ { "number": 1, "title": "Example Slide", "content": "Example content", "notes": "Example notes" if parameters.get("includeNotes", False) else None } ] } ) except Exception as e: logger.error(f"Error reading presentation: {e}") return self._createResult( success=False, data={"error": f"Read failed: {str(e)}"} ) async def _write_presentation(self, parameters: Dict[str, Any], authData: Dict[str, Any]) -> MethodResult: """Write content to PowerPoint presentation""" try: path = Path(parameters["path"]) # Create directory if it doesn't exist path.parent.mkdir(parents=True, exist_ok=True) # Determine format if not specified format = parameters.get("format") if not format: format = path.suffix[1:] if path.suffix else "pptx" # TODO: Implement PowerPoint writing using Microsoft Graph API # This is a placeholder implementation return self._createResult( success=True, data={ "path": str(path), "format": format, "slides": len(parameters["content"].get("slides", [])) } ) except Exception as e: logger.error(f"Error writing presentation: {e}") return self._createResult( success=False, data={"error": f"Write failed: {str(e)}"} ) async def _convert_presentation(self, parameters: Dict[str, Any], authData: Dict[str, Any]) -> MethodResult: """Convert PowerPoint presentation between formats""" try: source_path = Path(parameters["sourcePath"]) target_path = Path(parameters["targetPath"]) if not source_path.exists(): return self._createResult( success=False, data={"error": f"Source file not found: {source_path}"} ) # Determine formats if not specified source_format = parameters.get("sourceFormat") if not source_format: source_format = source_path.suffix[1:] if source_path.suffix else "pptx" target_format = parameters.get("targetFormat") if not target_format: target_format = target_path.suffix[1:] if target_path.suffix else "pptx" # TODO: Implement PowerPoint conversion using Microsoft Graph API # This is a placeholder implementation return self._createResult( success=True, data={ "sourcePath": str(source_path), "targetPath": str(target_path), "sourceFormat": source_format, "targetFormat": target_format } ) except Exception as e: logger.error(f"Error converting presentation: {e}") return self._createResult( success=False, data={"error": f"Conversion failed: {str(e)}"} ) async def _createPresentation(self, parameters: Dict[str, Any], userConnection: UserConnection) -> MethodResult: """Create a new PowerPoint presentation""" try: title = parameters["title"] template = parameters.get("template") # Create PowerPoint account account = Account( credentials=(userConnection.authToken, userConnection.refreshToken), protocol=MSGraphProtocol() ) # Get drive drive = account.drive() # Create presentation if template: # Copy template templateFile = drive.get_item_by_path(template) newFile = templateFile.copy(f"{title}.pptx") else: # Create blank presentation newFile = drive.create_file( name=f"{title}.pptx", content_type="application/vnd.openxmlformats-officedocument.presentationml.presentation" ) return self._createResult( success=True, data={ "id": newFile.object_id, "name": newFile.name, "webUrl": newFile.web_url } ) except Exception as e: logger.error(f"Error creating PowerPoint presentation: {e}") return self._createResult( success=False, data={"error": f"Create failed: {str(e)}"} ) async def _addSlide(self, parameters: Dict[str, Any], userConnection: UserConnection) -> MethodResult: """Add a new slide to presentation""" try: presentationId = parameters["presentationId"] layout = parameters.get("layout", "title") title = parameters.get("title") # Create PowerPoint account account = Account( credentials=(userConnection.authToken, userConnection.refreshToken), protocol=MSGraphProtocol() ) # Get drive drive = account.drive() # Get presentation presentation = drive.get_item_by_id(presentationId) # Add slide slide = presentation.add_slide(layout=layout) if title: slide.title = title return self._createResult( success=True, data={ "slideId": slide.object_id, "layout": layout, "title": title } ) except Exception as e: logger.error(f"Error adding PowerPoint slide: {e}") return self._createResult( success=False, data={"error": f"Add slide failed: {str(e)}"} ) async def _addContent(self, parameters: Dict[str, Any], userConnection: UserConnection) -> MethodResult: """Add content to a slide""" try: presentationId = parameters["presentationId"] slideId = parameters["slideId"] contentType = parameters["contentType"] content = parameters["content"] position = parameters.get("position", {"x": 0, "y": 0}) # Create PowerPoint account account = Account( credentials=(userConnection.authToken, userConnection.refreshToken), protocol=MSGraphProtocol() ) # Get drive drive = account.drive() # Get presentation and slide presentation = drive.get_item_by_id(presentationId) slide = presentation.get_slide(slideId) # Add content based on type if contentType == "text": shape = slide.add_text_box( text=content, left=position["x"], top=position["y"] ) elif contentType == "image": shape = slide.add_picture( image_path=content, left=position["x"], top=position["y"] ) elif contentType == "table": shape = slide.add_table( rows=content["rows"], cols=content["cols"], left=position["x"], top=position["y"] ) else: raise ValueError(f"Unsupported content type: {contentType}") return self._createResult( success=True, data={ "shapeId": shape.object_id, "contentType": contentType, "position": position } ) except Exception as e: logger.error(f"Error adding PowerPoint content: {e}") return self._createResult( success=False, data={"error": f"Add content failed: {str(e)}"} )