| 462 | |
| 463 | # @register_tool('file_parser') |
| 464 | class SingleFileParser(BaseTool): |
| 465 | name="file_parser" |
| 466 | description = f"File parsing tool, supports parsing data in {'/'.join(PARSER_SUPPORTED_FILE_TYPES)} formats, and returns the parsed markdown format data." |
| 467 | parameters = [{ |
| 468 | 'name': 'url', |
| 469 | 'type': 'string', |
| 470 | 'description': 'The full path of the file to be parsed, which can be a local path or a downloadable http(s) link.', |
| 471 | 'required': True |
| 472 | }] |
| 473 | |
| 474 | def __init__(self, cfg: Optional[Dict] = None): |
| 475 | super().__init__(cfg) |
| 476 | self.data_root = self.cfg.get('path', os.path.join(DEFAULT_WORKSPACE, 'tools', self.name)) |
| 477 | self.db = Storage({'storage_root_path': self.data_root}) |
| 478 | self.structured_doc = self.cfg.get('structured_doc', True) |
| 479 | |
| 480 | |
| 481 | self.parsers = { |
| 482 | 'pdf': parse_pdf, |
| 483 | 'docx': parse_word, |
| 484 | 'doc': parse_word, |
| 485 | 'pptx': parse_ppt, |
| 486 | 'txt': parse_txt, |
| 487 | 'jsonl': parse_txt, |
| 488 | 'jsonld': parse_txt, |
| 489 | 'pdb': parse_txt, |
| 490 | 'py': parse_txt, |
| 491 | 'html': parse_html, |
| 492 | 'xml': parse_xml, |
| 493 | 'csv': lambda p: parse_tabular_file(p, sep=','), |
| 494 | 'tsv': lambda p: parse_tabular_file(p, sep='\t'), |
| 495 | 'xlsx': parse_tabular_file, |
| 496 | 'xls': parse_tabular_file, |
| 497 | 'zip': self.parse_zip |
| 498 | } |
| 499 | |
| 500 | def call(self, params: Union[str, dict], **kwargs) -> Union[str, list]: |
| 501 | params = self._verify_json_format_args(params) |
| 502 | file_path = self._prepare_file(params['url']) |
| 503 | try: |
| 504 | cached = self.db.get(f'{hash_sha256(file_path)}_ori') |
| 505 | return self._flatten_result(json.loads(cached)) |
| 506 | except KeyNotExistsError: |
| 507 | return self._flatten_result(self._process_new_file(file_path)) |
| 508 | |
| 509 | def _prepare_file(self, path: str) -> str: |
| 510 | if is_http_url(path): |
| 511 | download_dir = os.path.join(self.data_root, hash_sha256(path)) |
| 512 | os.makedirs(download_dir, exist_ok=True) |
| 513 | return save_url_to_local_work_dir(path, download_dir) |
| 514 | return sanitize_chrome_file_path(path) |
| 515 | |
| 516 | def _process_new_file(self, file_path: str) -> Union[str, list]: |
| 517 | file_type = get_file_type(file_path) |
| 518 | idp_types = ['pdf', 'docx', 'pptx', 'xlsx', 'jpg', 'png', 'mp3'] |
| 519 | logger.info(f'Start parsing {file_path}...') |
| 520 | logger.info(f'File type {file_type}...') |
| 521 | logger.info(f"structured_doc {self.cfg.get('structured_doc')}...") |