| 154 | |
| 155 | |
| 156 | class DocumentConverter: |
| 157 | _default_download_filename = "file" |
| 158 | |
| 159 | def __init__( |
| 160 | self, |
| 161 | allowed_formats: Optional[List[InputFormat]] = None, |
| 162 | format_options: Optional[Dict[InputFormat, FormatOption]] = None, |
| 163 | ): |
| 164 | self.allowed_formats = ( |
| 165 | allowed_formats if allowed_formats is not None else [e for e in InputFormat] |
| 166 | ) |
| 167 | self.format_to_options = { |
| 168 | format: ( |
| 169 | _get_default_option(format=format) |
| 170 | if (custom_option := (format_options or {}).get(format)) is None |
| 171 | else custom_option |
| 172 | ) |
| 173 | for format in self.allowed_formats |
| 174 | } |
| 175 | self.initialized_pipelines: Dict[Type[BasePipeline], BasePipeline] = {} |
| 176 | |
| 177 | def initialize_pipeline(self, format: InputFormat): |
| 178 | """Initialize the conversion pipeline for the selected format.""" |
| 179 | pipeline = self._get_pipeline(doc_format=format) |
| 180 | if pipeline is None: |
| 181 | raise ConversionError( |
| 182 | f"No pipeline could be initialized for format {format}" |
| 183 | ) |
| 184 | |
| 185 | @validate_call(config=ConfigDict(strict=True)) |
| 186 | def convert( |
| 187 | self, |
| 188 | source: Union[Path, str, DocumentStream], # TODO review naming |
| 189 | headers: Optional[Dict[str, str]] = None, |
| 190 | raises_on_error: bool = True, |
| 191 | max_num_pages: int = sys.maxsize, |
| 192 | max_file_size: int = sys.maxsize, |
| 193 | page_range: PageRange = DEFAULT_PAGE_RANGE, |
| 194 | ) -> ConversionResult: |
| 195 | all_res = self.convert_all( |
| 196 | source=[source], |
| 197 | raises_on_error=raises_on_error, |
| 198 | max_num_pages=max_num_pages, |
| 199 | max_file_size=max_file_size, |
| 200 | headers=headers, |
| 201 | page_range=page_range, |
| 202 | ) |
| 203 | return next(all_res) |
| 204 | |
| 205 | @validate_call(config=ConfigDict(strict=True)) |
| 206 | def convert_all( |
| 207 | self, |
| 208 | source: Iterable[Union[Path, str, DocumentStream]], # TODO review naming |
| 209 | headers: Optional[Dict[str, str]] = None, |
| 210 | raises_on_error: bool = True, # True: raises on first conversion error; False: does not raise on conv error |
| 211 | max_num_pages: int = sys.maxsize, |
| 212 | max_file_size: int = sys.maxsize, |
| 213 | page_range: PageRange = DEFAULT_PAGE_RANGE, |
no outgoing calls
no test coverage detected