"""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" # Bundled copies ship with the Space as a fallback when /files/{id} 404s. 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 # Optional: GAIA hub download when HF_TOKEN is present 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