File size: 2,087 Bytes
b64d2eb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | """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
|