38 lines
958 B
Python
38 lines
958 B
Python
"""简单的文件下载器，负责将 Supabase Storage 签名 URL 暂存到临时目录。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from urllib.parse import urlparse
|
|
|
|
import httpx
|
|
|
|
|
|
class StorageFetcher:
|
|
def __init__(self) -> None:
|
|
self._client = httpx.Client(timeout=30.0)
|
|
|
|
def download(self, file_url: str) -> str:
|
|
response = self._client.get(file_url)
|
|
response.raise_for_status()
|
|
suffix = Path(urlparse(file_url).path).suffix or ".bin"
|
|
fd, path = tempfile.mkstemp(suffix=suffix)
|
|
try:
|
|
os.write(fd, response.content)
|
|
finally:
|
|
os.close(fd)
|
|
return path
|
|
|
|
def cleanup(self, path: Optional[str]) -> None:
|
|
if path and os.path.exists(path):
|
|
try:
|
|
os.remove(path)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
storage_fetcher = StorageFetcher()
|