Spaces:
Sleeping
Sleeping
| """ | |
| File handling tools for the GAIA Agent. | |
| Includes file reading and downloading from GAIA API. | |
| """ | |
| import os | |
| import tempfile | |
| import requests | |
| import pandas as pd | |
| from langchain_core.tools import tool | |
| def read_file(file_path: str) -> str: | |
| """Read the contents of a file. | |
| Supports text files, code files, CSV, Excel, etc. | |
| Args: | |
| file_path: Path to the file to read | |
| """ | |
| try: | |
| if file_path.endswith(('.xlsx', '.xls')): | |
| df = pd.read_excel(file_path) | |
| return f"Excel file with {len(df)} rows, {len(df.columns)} columns.\n\nColumns: {list(df.columns)}\n\nData:\n{df.to_string()}" | |
| elif file_path.endswith('.csv'): | |
| df = pd.read_csv(file_path) | |
| return f"CSV file with {len(df)} rows, {len(df.columns)} columns.\n\nColumns: {list(df.columns)}\n\nData:\n{df.to_string()}" | |
| elif file_path.endswith('.json'): | |
| import json | |
| with open(file_path, 'r', encoding='utf-8') as f: | |
| data = json.load(f) | |
| return f"JSON file contents:\n{json.dumps(data, indent=2)[:10000]}" | |
| else: | |
| with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: | |
| content = f.read() | |
| return content[:10000] | |
| except Exception as e: | |
| return f"Error reading file: {str(e)}" | |
| def download_file(task_id: str, file_name: str = "") -> str: | |
| """Download a file associated with a GAIA task. | |
| Returns the local file path where the file was saved. | |
| Args: | |
| task_id: The GAIA task ID | |
| file_name: Expected filename (helps determine file type) | |
| """ | |
| try: | |
| api_base = "https://agents-course-unit4-scoring.hf.space" | |
| url = f"{api_base}/files/{task_id}" | |
| response = requests.get(url, timeout=60) | |
| if response.status_code == 200: | |
| content_disp = response.headers.get('content-disposition', '') | |
| if 'filename=' in content_disp: | |
| filename = content_disp.split('filename=')[1].strip('"\'') | |
| elif file_name: | |
| filename = file_name | |
| else: | |
| filename = f"{task_id}_file" | |
| # Use a portable temp directory (works on Windows + Linux) | |
| file_path = os.path.join(tempfile.gettempdir(), filename) | |
| with open(file_path, 'wb') as f: | |
| f.write(response.content) | |
| return f"File downloaded to: {file_path}" | |
| else: | |
| return f"Download failed: HTTP {response.status_code}" | |
| except Exception as e: | |
| return f"Download error: {str(e)}" | |