File size: 2,160 Bytes
6c511d6 | 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 | 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])
|