Joey
71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
"""本地文件存储：落 data/media/，由 FastAPI /media 静态托管。"""
|
||
from __future__ import annotations
|
||
|
||
import mimetypes
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
|
||
from config import get_settings
|
||
|
||
|
||
def media_root() -> Path:
|
||
root = Path(get_settings().data_dir) / "media"
|
||
root.mkdir(parents=True, exist_ok=True)
|
||
return root
|
||
|
||
|
||
def public_url(key: str) -> str:
|
||
"""media key → 可访问 URL。"""
|
||
settings = get_settings()
|
||
return f"{settings.app_base_url.rstrip('/')}/media/{key}"
|
||
|
||
|
||
def _ext_from_url_or_type(hint: str, content_type: str = "") -> str:
|
||
if content_type:
|
||
ctype = content_type.split(";")[0].strip().lower()
|
||
mapping = {
|
||
"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp",
|
||
"image/gif": ".gif", "image/bmp": ".bmp", "video/mp4": ".mp4",
|
||
}
|
||
if ctype in mapping:
|
||
return mapping[ctype]
|
||
ext = mimetypes.guess_extension(hint.split("?")[0].lower()) or ".jpg"
|
||
return ".jpg" if ext == ".jpe" else ext
|
||
|
||
|
||
def write_bytes(data: bytes, key_prefix: str = "", ext: str = ".jpg") -> str:
|
||
"""写文件，返回 media key（相对 media 根的路径）。"""
|
||
key = f"{key_prefix + '/' if key_prefix else ''}{uuid.uuid4().hex}{ext}"
|
||
path = media_root() / key
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_bytes(data)
|
||
return key
|
||
|
||
|
||
async def download_bytes(url: str, referer: str | None = None, timeout: float = 60.0) -> tuple[bytes, str]:
|
||
"""下载远程字节。返回 (bytes, content_type)。"""
|
||
headers = {"Referer": referer} if referer else {}
|
||
headers.setdefault("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)")
|
||
async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, verify=False) as client:
|
||
resp = await client.get(url, headers=headers)
|
||
resp.raise_for_status()
|
||
ctype = (resp.headers.get("content-type") or "application/octet-stream").split(";")[0].strip()
|
||
return resp.content, ctype
|
||
|
||
|
||
async def save_from_url(url: str, key_prefix: str = "", referer: str | None = None) -> str:
|
||
data, ctype = await download_bytes(url, referer)
|
||
key = write_bytes(data, key_prefix, _ext_from_url_or_type(url, ctype))
|
||
return public_url(key)
|
||
|
||
|
||
def local_path(stored_url_or_key: str) -> Path | None:
|
||
"""stored_url（http.../media/xxx）或 key → 本地文件路径。"""
|
||
s = stored_url_or_key
|
||
if "/media/" in s:
|
||
s = s.split("/media/", 1)[1]
|
||
p = media_root() / s
|
||
return p if p.exists() else None
|