Convert a file to markdown asynchronously. Args: file_path (str): Absolute path to the file to convert. output_format (str): Output format.
(self, file_path: str, output_format: str = "markdown", **kwargs)
| 62 | self.converter = MarkitdownConverter(timeout=self.timeout) |
| 63 | |
| 64 | async def __call__(self, file_path: str, output_format: str = "markdown", **kwargs) -> ToolResponse: |
| 65 | """ |
| 66 | Convert a file to markdown asynchronously. |
| 67 | |
| 68 | Args: |
| 69 | file_path (str): Absolute path to the file to convert. |
| 70 | output_format (str): Output format. |
| 71 | """ |
| 72 | try: |
| 73 | # Validate input |
| 74 | if not file_path.strip(): |
| 75 | return ToolResponse(success=False, message="Error: Empty file path provided") |
| 76 | |
| 77 | # Check if file exists |
| 78 | if not os.path.exists(file_path): |
| 79 | return ToolResponse(success=False, message=f"Error: File not found: {file_path}") |
| 80 | |
| 81 | # Check if it's a file (not directory) |
| 82 | if not os.path.isfile(file_path): |
| 83 | return ToolResponse(success=False, message=f"Error: Path is not a file: {file_path}") |
| 84 | |
| 85 | # Get file info |
| 86 | file_size = os.path.getsize(file_path) |
| 87 | file_name = os.path.basename(file_path) |
| 88 | file_ext = os.path.splitext(file_path)[1].lower() |
| 89 | |
| 90 | # Check file size (limit to 100MB for safety) |
| 91 | max_size = 100 * 1024 * 1024 # 100MB |
| 92 | if file_size > max_size: |
| 93 | return ToolResponse( |
| 94 | success=False, message=f"Error: File too large ({file_size / (1024*1024):.1f}MB). " |
| 95 | f"Maximum allowed size is {max_size / (1024*1024)}MB" |
| 96 | ) |
| 97 | |
| 98 | # Run conversion in thread pool to avoid blocking |
| 99 | loop = asyncio.get_event_loop() |
| 100 | result = await loop.run_in_executor( |
| 101 | None, |
| 102 | self._convert_file, |
| 103 | file_path, |
| 104 | output_format |
| 105 | ) |
| 106 | |
| 107 | if result is None: |
| 108 | return ToolResponse(success=False, message="Error: Conversion failed - unable to process the file") |
| 109 | |
| 110 | # Save to base_dir if specified |
| 111 | saved_path = None |
| 112 | if self.base_dir: |
| 113 | # Create base_dir if it doesn't exist |
| 114 | os.makedirs(self.base_dir, exist_ok=True) |
| 115 | |
| 116 | # Generate output filename (replace original extension with .md) |
| 117 | base_name = os.path.splitext(file_name)[0] |
| 118 | output_filename = f"{base_name}.md" |
| 119 | saved_path = os.path.join(self.base_dir, output_filename) |
| 120 | |
| 121 | # Save markdown content to file |