Spaces:
Sleeping
Sleeping
import requests | |
from typing import Annotated | |
from langchain_core.tools import tool | |
from langgraph.prebuilt import InjectedState | |
from src.state import State | |
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" | |
def download_file( | |
task_id: str, | |
state: Annotated[State, InjectedState] | |
) -> str: | |
"""Download a file specified by using the task id.""" | |
file_name = state.file_name | |
task_id = state.task_id | |
if not file_name: | |
return "No file name in input, unable to download." | |
if not task_id: | |
return "No task id in input, unable to download." | |
base_url = DEFAULT_API_URL + "/files" | |
url = f"{base_url}/{task_id}" if task_id else None | |
if not url: | |
return "No URL provided." | |
try: | |
response = requests.get(url, stream=True) | |
response.raise_for_status() # Raise an error for bad responses | |
local_file_path = f"downloads/{file_name}" | |
with open(local_file_path, "wb") as f: | |
for chunk in response.iter_content(chunk_size=8192): | |
f.write(chunk) | |
return f"File downloaded successfully: {local_file_path}" | |
except requests.exceptions.RequestException as e: | |
return f"Error downloading file: {e}" |