(self, config: dict, context: ExecutionContext)
| 239 | return "pdf_to_word" |
| 240 | |
| 241 | async def execute(self, config: dict, context: ExecutionContext) -> ModuleResult: |
| 242 | pdf_path = context.resolve_value(config.get('pdfPath', '')) |
| 243 | output_dir = context.resolve_value(config.get('outputDir', '')) |
| 244 | page_range = context.resolve_value(config.get('pageRange', '')) |
| 245 | result_variable = config.get('resultVariable', '') |
| 246 | |
| 247 | if not pdf_path: |
| 248 | return ModuleResult(success=False, error="PDF文件路径不能为空") |
| 249 | if not os.path.exists(pdf_path): |
| 250 | return ModuleResult(success=False, error=f"PDF文件不存在: {pdf_path}") |
| 251 | |
| 252 | if not output_dir: |
| 253 | output_dir = os.path.dirname(pdf_path) |
| 254 | os.makedirs(output_dir, exist_ok=True) |
| 255 | |
| 256 | try: |
| 257 | loop = asyncio.get_event_loop() |
| 258 | result = await loop.run_in_executor( |
| 259 | None, self._convert, pdf_path, output_dir, page_range |
| 260 | ) |
| 261 | |
| 262 | if result_variable: |
| 263 | context.set_variable(result_variable, result) |
| 264 | |
| 265 | return ModuleResult(success=True, message=f"已将PDF转换为Word文档", data=result) |
| 266 | except Exception as e: |
| 267 | import traceback |
| 268 | traceback.print_exc() |
| 269 | return ModuleResult(success=False, error=f"PDF转Word失败: {str(e)}") |
| 270 | |
| 271 | def _convert(self, pdf_path: str, output_dir: str, page_range: str) -> dict: |
| 272 | """将PDF转换为Word文档 - 使用pdf2docx实现高保真转换""" |
nothing calls this directly
no test coverage detected