247 lines
7.1 KiB
Python
247 lines
7.1 KiB
Python
import os
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Dict, Any, List
|
|
|
|
from fastapi import FastAPI, HTTPException, Query
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.responses import FileResponse
|
|
from PIL import Image
|
|
|
|
app = FastAPI()
|
|
|
|
# -----------------------------
|
|
# Env / Paths
|
|
# -----------------------------
|
|
ROOT = Path(os.getenv("TRAVEL_ROOT", "/data/travel")).resolve()
|
|
MEDIA_BASE = os.getenv("TRAVEL_MEDIA_BASE", "/media/travel")
|
|
|
|
META_DIR = ROOT / "_meta"
|
|
REGION_MAP_PATH = META_DIR / "region_map.json"
|
|
REGIONS_GEOJSON_PATH = META_DIR / "regions.geojson"
|
|
|
|
THUMB_ROOT = Path(os.getenv("TRAVEL_THUMB_ROOT", "/data/thumbs")).resolve()
|
|
THUMB_SIZE = (480, 480)
|
|
|
|
THUMB_ROOT.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 썸네일 정적 서빙
|
|
app.mount(
|
|
f"{MEDIA_BASE}/.thumb",
|
|
StaticFiles(directory=THUMB_ROOT),
|
|
name="travel-thumbs",
|
|
)
|
|
|
|
IMAGE_EXT = {".jpg", ".jpeg", ".png", ".webp"}
|
|
|
|
# -----------------------------
|
|
# Cache
|
|
# -----------------------------
|
|
CACHE: Dict[str, Dict[str, Any]] = {}
|
|
CACHE_TTL = int(os.getenv("TRAVEL_CACHE_TTL", "300"))
|
|
META_MTIME_CACHE: Dict[str, float] = {}
|
|
|
|
# -----------------------------
|
|
# Helpers
|
|
# -----------------------------
|
|
def _file_mtime(p: Path) -> float:
|
|
try:
|
|
return p.stat().st_mtime
|
|
except FileNotFoundError:
|
|
return 0.0
|
|
|
|
|
|
def _read_json(path: Path) -> Any:
|
|
if not path.exists():
|
|
raise HTTPException(500, f"Missing required file: {path}")
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def load_region_map() -> dict:
|
|
return _read_json(REGION_MAP_PATH)
|
|
|
|
|
|
def load_regions_geojson() -> dict:
|
|
return _read_json(REGIONS_GEOJSON_PATH)
|
|
|
|
|
|
def _meta_changed_invalidate_cache():
|
|
cur = _file_mtime(REGION_MAP_PATH) + _file_mtime(REGIONS_GEOJSON_PATH)
|
|
if META_MTIME_CACHE.get("meta") != cur:
|
|
CACHE.clear()
|
|
META_MTIME_CACHE["meta"] = cur
|
|
|
|
|
|
def _get_albums_for_region(region: str, region_map: dict) -> List[str]:
|
|
if region not in region_map:
|
|
raise HTTPException(400, "Unknown region")
|
|
|
|
v = region_map[region]
|
|
if isinstance(v, list):
|
|
return v
|
|
if isinstance(v, dict) and isinstance(v.get("albums"), list):
|
|
return v["albums"]
|
|
|
|
raise HTTPException(500, "Invalid region_map format")
|
|
|
|
|
|
def _thumb_path_for(src: Path, album: str) -> Path:
|
|
"""
|
|
썸네일 저장 위치 결정.
|
|
- TRAVEL_THUMB_ROOT가 있으면: THUMB_ROOT/album/<filename>
|
|
- 없으면: 원본 폴더 album/.thumb/<filename>
|
|
"""
|
|
if THUMB_ROOT:
|
|
base = THUMB_ROOT / album
|
|
base.mkdir(parents=True, exist_ok=True)
|
|
return base / src.name
|
|
|
|
thumb_dir = src.parent / ".thumb"
|
|
thumb_dir.mkdir(exist_ok=True)
|
|
return thumb_dir / src.name
|
|
|
|
def ensure_thumb(src: Path, album: str) -> Path:
|
|
out = _thumb_path_for(src, album) # THUMB_ROOT/album/<filename> or album/.thumb/<filename>
|
|
|
|
if out.exists():
|
|
return out
|
|
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# ✅ 확장자 유지: IMG_3281.tmp.JPG (끝이 .JPG로 끝나게)
|
|
tmp = out.with_name(out.stem + ".tmp" + out.suffix)
|
|
|
|
try:
|
|
with Image.open(src) as im:
|
|
im.thumbnail(THUMB_SIZE)
|
|
|
|
# ✅ 확장자 기준으로 포맷 명시 (대문자 .JPG도 대응)
|
|
ext = out.suffix.lower()
|
|
if ext in (".jpg", ".jpeg"):
|
|
fmt = "JPEG"
|
|
elif ext == ".png":
|
|
fmt = "PNG"
|
|
elif ext == ".webp":
|
|
fmt = "WEBP"
|
|
else:
|
|
# 혹시 모를 경우: Pillow가 읽은 포맷을 사용
|
|
fmt = (im.format or "").upper() or "JPEG"
|
|
|
|
im.save(tmp, format=fmt, quality=85, optimize=True)
|
|
|
|
# ✅ 생성 완료 후 교체
|
|
tmp.replace(out)
|
|
return out
|
|
|
|
finally:
|
|
# 실패 시 tmp 찌꺼기 정리
|
|
try:
|
|
if tmp.exists():
|
|
tmp.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
def scan_album(album: str) -> List[Dict[str, Any]]:
|
|
album_dir = ROOT / album
|
|
if not album_dir.exists():
|
|
return []
|
|
|
|
items = []
|
|
# os.scandir이 iterdir보다 빠름 (파일 속성 한 번에 가져옴)
|
|
with os.scandir(album_dir) as entries:
|
|
for entry in entries:
|
|
if entry.is_file() and Path(entry.name).suffix.lower() in IMAGE_EXT:
|
|
# ⚡ 성능 핵심: 여기서 썸네일 생성(ensure_thumb) 절대 하지 않음!
|
|
# 파일 존재 여부만 확인하고 바로 리턴
|
|
items.append({
|
|
"album": album,
|
|
"file": entry.name,
|
|
"url": f"{MEDIA_BASE}/{album}/{entry.name}",
|
|
"thumb": f"{MEDIA_BASE}/.thumb/{album}/{entry.name}",
|
|
# 정렬을 위해 수정시간/이름 필요하면 여기서 저장
|
|
"mtime": entry.stat().st_mtime
|
|
})
|
|
return items
|
|
|
|
# -----------------------------
|
|
# Routes
|
|
# -----------------------------
|
|
@app.get("/api/travel/regions")
|
|
def regions():
|
|
_meta_changed_invalidate_cache()
|
|
return load_regions_geojson()
|
|
|
|
|
|
@app.get("/api/travel/photos")
|
|
def photos(
|
|
region: str = Query(...),
|
|
page: int = Query(1, ge=1),
|
|
size: int = Query(20, ge=1, le=100),
|
|
):
|
|
_meta_changed_invalidate_cache()
|
|
|
|
# 1. 캐시 키에 페이지 정보 포함하지 않음 (전체 리스트 캐싱 후 슬라이싱 전략)
|
|
# 왜냐하면 파일 스캔 비용이 크지, 메모리 슬라이싱 비용은 작기 때문.
|
|
now = time.time()
|
|
cached = CACHE.get(region)
|
|
|
|
if not cached or now - cached["ts"] >= CACHE_TTL:
|
|
region_map = load_region_map()
|
|
albums = _get_albums_for_region(region, region_map)
|
|
|
|
all_items = []
|
|
matched = []
|
|
|
|
for album in albums:
|
|
items = scan_album(album)
|
|
matched.append({"album": album, "count": len(items)})
|
|
all_items.extend(items)
|
|
|
|
# 정렬: 앨범명 > 파일명 (또는 찍은 날짜)
|
|
all_items.sort(key=lambda x: (x["album"], x["file"]))
|
|
|
|
cached_data = {
|
|
"region": region,
|
|
"matched_albums": matched,
|
|
"items": all_items,
|
|
"total": len(all_items),
|
|
}
|
|
CACHE[region] = {"ts": now, "data": cached_data}
|
|
else:
|
|
cached_data = cached["data"]
|
|
|
|
# 2. 페이지네이션 슬라이싱
|
|
all_items = cached_data["items"]
|
|
total = len(all_items)
|
|
start = (page - 1) * size
|
|
end = start + size
|
|
|
|
paged_items = all_items[start:end]
|
|
|
|
return {
|
|
"region": region,
|
|
"page": page,
|
|
"size": size,
|
|
"total": total,
|
|
"has_next": end < total,
|
|
"items": paged_items,
|
|
}
|
|
|
|
|
|
@app.get("/media/travel/.thumb/{album}/{filename}")
|
|
def get_thumb(album: str, filename: str):
|
|
src = (ROOT / album / filename).resolve()
|
|
if not p.exists() or not p.is_file():
|
|
raise HTTPException(404, "Thumbnail not found")
|
|
|
|
# src로부터 thumb 생성/확인 (원본 확장자 유지)
|
|
p = ensure_thumb(src, album)
|
|
return FileResponse(str(p))
|
|
|
|
@app.get("/api/version")
|
|
def version():
|
|
import os
|
|
return {"version": os.getenv("APP_VERSION", "dev")}
|