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