from pathlib import Path import requests from config import CACHE_DIR, DEFAULT_API_URL, REQUEST_HEADERS from tools.types import SolverResult def local_attachment_candidates(task_id: str, file_name: str) -> list[Path]: return [ Path("data") / "attachments" / task_id / file_name, Path("data") / "attachments" / file_name, Path("resource") / file_name, Path(file_name), ] def remote_attachment_candidates(task_id: str, file_name: str) -> list[str]: return [ f"{DEFAULT_API_URL}/files/{task_id}", f"{DEFAULT_API_URL}/files/{file_name}", f"{DEFAULT_API_URL}/files/{task_id}/{file_name}", f"https://huggingface.co/datasets/asteriadyt/2023/resolve/main/validation/{file_name}", f"https://huggingface.co/spaces/jitendra217/Final_Assignment/resolve/main/resource/{file_name}", ] def download_task_file(task_id: str, file_name: str) -> tuple[Path | None, str]: """下载/定位课程附件,并缓存在 /tmp 下。""" if not task_id or not file_name: return None, "没有附件。" for local_path in local_attachment_candidates(task_id, file_name): if local_path.exists() and local_path.stat().st_size > 0: return local_path, f"使用本地附件:{local_path}" destination = CACHE_DIR / task_id / file_name if destination.exists() and destination.stat().st_size > 0: return destination, f"使用缓存附件:{destination}" destination.parent.mkdir(parents=True, exist_ok=True) errors = [] for url in remote_attachment_candidates(task_id, file_name): try: response = requests.get(url, headers=REQUEST_HEADERS, timeout=45) if response.status_code == 404: errors.append(f"{url} -> 404") continue response.raise_for_status() if response.content: destination.write_bytes(response.content) return destination, f"已下载附件:{url}" except Exception as exc: errors.append(f"{url} -> {exc}") return None, "附件下载失败:" + " | ".join(errors[:5])