Spaces:
Sleeping
Sleeping
File size: 2,144 Bytes
7ace26a |
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 |
import os
import requests
from typing import Optional
import tempfile
# --- File Download Helper Function ---
def download_file(task_id: str, api_url: str) -> Optional[str]:
"""Download file associated with a task_id from the evaluation API"""
try:
file_url = f"{api_url}/files/{task_id}"
print(f"📁 Downloading file for task {task_id}")
response = requests.get(file_url, timeout=30)
response.raise_for_status()
# Get filename from headers or create one
content_disposition = response.headers.get('Content-Disposition', '')
if 'filename=' in content_disposition:
filename = content_disposition.split('filename=')[1].strip('"')
else:
content_type = response.headers.get('Content-Type', '')
if 'image' in content_type:
extension = '.jpg'
elif 'audio' in content_type:
extension = '.mp3'
elif 'video' in content_type:
extension = '.mp4'
else:
extension = '.txt'
filename = f"task_{task_id}_file{extension}"
# Save to temporary file
temp_dir = tempfile.gettempdir()
file_path = os.path.join(temp_dir, filename)
with open(file_path, 'wb') as f:
f.write(response.content)
print(f"✅ File downloaded: {file_path}")
return file_path
except Exception as e:
print(f"❌ Error downloading file for task {task_id}: {e}")
return None
def clean_answer_for_eval(answer: str) -> str:
"""Clean the answer string for evaluation purposes."""
#
if isinstance(answer, str):
answer = answer.strip()
prefixes_to_remove = [
"The answer is: ",
"Answer: ",
"The result is: ",
"Result: ",
"The final answer is: ",
]
for prefix in prefixes_to_remove:
if answer.startswith(prefix):
answer = answer[len(prefix):].strip()
break
return answer |