97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
import os
|
|
import re
|
|
import mimetypes
|
|
from typing import Optional
|
|
from fastapi import APIRouter, HTTPException, Response
|
|
from fastapi.responses import FileResponse, RedirectResponse
|
|
import httpx
|
|
|
|
from ..runtime import get_runtime_settings
|
|
|
|
router = APIRouter(prefix="/images", tags=["images"])
|
|
|
|
_TMDB_BASE = "https://image.tmdb.org/t/p"
|
|
_ALLOWED_SIZES = {"w92", "w154", "w185", "w342", "w500", "w780", "original"}
|
|
|
|
|
|
def _safe_filename(path: str) -> str:
|
|
trimmed = path.strip("/")
|
|
trimmed = trimmed.replace("/", "_")
|
|
safe = re.sub(r"[^A-Za-z0-9_.-]", "_", trimmed)
|
|
return safe or "image"
|
|
|
|
def tmdb_cache_path(path: str, size: str) -> Optional[str]:
|
|
if not path or "://" in path or ".." in path:
|
|
return None
|
|
if not path.startswith("/"):
|
|
path = f"/{path}"
|
|
if size not in _ALLOWED_SIZES:
|
|
return None
|
|
cache_dir = os.path.join(os.getcwd(), "data", "artwork", "tmdb", size)
|
|
return os.path.join(cache_dir, _safe_filename(path))
|
|
|
|
|
|
def is_tmdb_cached(path: str, size: str) -> bool:
|
|
file_path = tmdb_cache_path(path, size)
|
|
return bool(file_path and os.path.exists(file_path))
|
|
|
|
|
|
async def cache_tmdb_image(path: str, size: str = "w342") -> bool:
|
|
if not path or "://" in path or ".." in path:
|
|
return False
|
|
|
|
runtime = get_runtime_settings()
|
|
cache_mode = (runtime.artwork_cache_mode or "remote").lower()
|
|
if cache_mode != "cache":
|
|
return False
|
|
|
|
file_path = tmdb_cache_path(path, size)
|
|
if not file_path:
|
|
return False
|
|
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
|
if os.path.exists(file_path):
|
|
return True
|
|
|
|
url = f"{_TMDB_BASE}/{size}{path}"
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
response = await client.get(url)
|
|
response.raise_for_status()
|
|
content = response.content
|
|
with open(file_path, "wb") as handle:
|
|
handle.write(content)
|
|
return True
|
|
|
|
|
|
@router.get("/tmdb")
|
|
async def tmdb_image(path: str, size: str = "w342"):
|
|
if not path or "://" in path or ".." in path:
|
|
raise HTTPException(status_code=400, detail="Invalid image path")
|
|
if not path.startswith("/"):
|
|
path = f"/{path}"
|
|
if size not in _ALLOWED_SIZES:
|
|
raise HTTPException(status_code=400, detail="Invalid size")
|
|
|
|
runtime = get_runtime_settings()
|
|
cache_mode = (runtime.artwork_cache_mode or "remote").lower()
|
|
url = f"{_TMDB_BASE}/{size}{path}"
|
|
if cache_mode != "cache":
|
|
return RedirectResponse(url=url)
|
|
|
|
file_path = tmdb_cache_path(path, size)
|
|
if not file_path:
|
|
raise HTTPException(status_code=400, detail="Invalid image path")
|
|
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
|
headers = {"Cache-Control": "public, max-age=86400"}
|
|
if os.path.exists(file_path):
|
|
media_type = mimetypes.guess_type(file_path)[0] or "image/jpeg"
|
|
return FileResponse(file_path, media_type=media_type, headers=headers)
|
|
|
|
try:
|
|
await cache_tmdb_image(path, size)
|
|
if os.path.exists(file_path):
|
|
media_type = mimetypes.guess_type(file_path)[0] or "image/jpeg"
|
|
return FileResponse(file_path, media_type=media_type, headers=headers)
|
|
raise HTTPException(status_code=502, detail="Image cache failed")
|
|
except httpx.HTTPError as exc:
|
|
raise HTTPException(status_code=502, detail=f"Image fetch failed: {exc}") from exc
|