Parse files with automatic path resolution
(params, **kwargs)
| 54 | |
| 55 | |
| 56 | async def file_parser(params, **kwargs): |
| 57 | """Parse files with automatic path resolution""" |
| 58 | urls = params.get('files', []) |
| 59 | if isinstance(urls, str): |
| 60 | urls = [urls] |
| 61 | |
| 62 | resolved_urls = [] |
| 63 | for url in urls: |
| 64 | if isinstance(url, list): |
| 65 | for sub_url in url: |
| 66 | if sub_url.startswith(("http://", "https://")): |
| 67 | resolved_urls.append(sub_url) |
| 68 | else: |
| 69 | abs_path = os.path.abspath(sub_url) |
| 70 | if os.path.exists(abs_path): |
| 71 | resolved_urls.append(abs_path) |
| 72 | else: |
| 73 | resolved_urls.append(sub_url) |
| 74 | else: |
| 75 | if url.startswith(("http://", "https://")): |
| 76 | resolved_urls.append(url) |
| 77 | else: |
| 78 | abs_path = os.path.abspath(url) |
| 79 | if os.path.exists(abs_path): |
| 80 | resolved_urls.append(abs_path) |
| 81 | else: |
| 82 | resolved_urls.append(url) |
| 83 | |
| 84 | results = [] |
| 85 | file_results = [] |
| 86 | for url in resolved_urls: |
| 87 | try: |
| 88 | result = SingleFileParser().call(json.dumps({'url': url}), **kwargs) |
| 89 | results.append(f"# File: {os.path.basename(url)}\n{result}") |
| 90 | file_results.append(result) |
| 91 | except Exception as e: |
| 92 | results.append(f"# Error processing {os.path.basename(url)}: {str(e)}") |
| 93 | if count_tokens(json.dumps(results)) < DEFAULT_MAX_INPUT_TOKENS: |
| 94 | return results |
| 95 | else: |
| 96 | return compress(file_results) |
| 97 | |
| 98 | # @register_tool("file_parser") |
| 99 | class FileParser(BaseTool): |
no test coverage detected