| """Download or locate attachment files for a task.""" |
| from __future__ import annotations |
|
|
| import os |
| from pathlib import Path |
| from typing import Optional |
|
|
| import requests |
|
|
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" |
| |
| LOCAL_FILES_DIR = Path(__file__).resolve().parent.parent / "files" |
|
|
|
|
| def download_task_file( |
| task_id: str, |
| file_name: str = "", |
| api_url: str = DEFAULT_API_URL, |
| dest_dir: Optional[Path] = None, |
| ) -> Optional[Path]: |
| """Fetch attachment from scoring API; fall back to bundled files/.""" |
| dest_dir = Path(dest_dir or LOCAL_FILES_DIR) |
| dest_dir.mkdir(parents=True, exist_ok=True) |
|
|
| candidates = [] |
| if file_name: |
| candidates.append(dest_dir / file_name) |
| candidates.append(dest_dir / task_id) |
| for ext in (".png", ".mp3", ".py", ".xlsx", ".txt", ".pdf"): |
| candidates.append(dest_dir / f"{task_id}{ext}") |
|
|
| for path in candidates: |
| if path.is_file() and path.stat().st_size > 200: |
| return path |
|
|
| if not file_name: |
| return None |
|
|
| out = dest_dir / file_name |
| url = f"{api_url.rstrip('/')}/files/{task_id}" |
| try: |
| resp = requests.get(url, timeout=60) |
| if resp.status_code == 200 and len(resp.content) > 200: |
| out.write_bytes(resp.content) |
| return out |
| except requests.RequestException: |
| pass |
|
|
| |
| token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGINGFACEHUB_API_TOKEN") |
| if token and file_name: |
| try: |
| from huggingface_hub import hf_hub_download |
|
|
| remote = f"2023/validation/{file_name}" |
| cached = hf_hub_download( |
| repo_id="gaia-benchmark/GAIA", |
| filename=remote, |
| repo_type="dataset", |
| token=token, |
| ) |
| out.write_bytes(Path(cached).read_bytes()) |
| return out |
| except Exception: |
| pass |
| return out if out.is_file() else None |
|
|
