gahusb
1c255152d7
- PUT /api/travel/albums/{album}/region: 앨범의 지역 변경 (extra 파일 기반)
- PUT /api/travel/regions/{region_id}: 커스텀 지역 이름/좌표 수정 (Phase 2 준비)
- _load_extra/_save_extra 헬퍼 분리, _removes 키로 원본 오버라이드 지원
- regions API: 모든 커스텀 지역 동적 병합 + Point geometry 지원
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
307 lines
9.0 KiB
Python
307 lines
9.0 KiB
Python
import os
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any, List
|
|
|
|
from fastapi import FastAPI, HTTPException, Query
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel
|
|
from PIL import Image
|
|
|
|
from .db import init_db, get_photos_by_region, get_all_albums, set_album_cover, mark_thumb_done
|
|
from .indexer import sync, _load_region_map_merged, move_album_region
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
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)
|
|
|
|
IMAGE_EXT = {".jpg", ".jpeg", ".png", ".webp"}
|
|
|
|
# -----------------------------
|
|
# DB init
|
|
# -----------------------------
|
|
init_db()
|
|
|
|
# -----------------------------
|
|
# Helpers
|
|
# -----------------------------
|
|
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 _load_region_map_merged(REGION_MAP_PATH, THUMB_ROOT)
|
|
|
|
|
|
def load_regions_geojson() -> dict:
|
|
return _read_json(REGIONS_GEOJSON_PATH)
|
|
|
|
|
|
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 _ensure_thumb_fallback(src: Path, album: str) -> Path:
|
|
"""온디맨드 썸네일 폴백 (sync 누락 분 대응)."""
|
|
out = THUMB_ROOT / album / src.name
|
|
if out.exists():
|
|
return out
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = out.with_name(out.stem + ".tmp" + out.suffix)
|
|
try:
|
|
with Image.open(src) as im:
|
|
im.thumbnail(THUMB_SIZE)
|
|
ext = out.suffix.lower()
|
|
if ext in (".jpg", ".jpeg"):
|
|
fmt = "JPEG"
|
|
elif ext == ".png":
|
|
fmt = "PNG"
|
|
elif ext == ".webp":
|
|
fmt = "WEBP"
|
|
else:
|
|
fmt = (im.format or "").upper() or "JPEG"
|
|
im.save(tmp, format=fmt, quality=85, optimize=True)
|
|
tmp.replace(out)
|
|
mark_thumb_done(album, src.name)
|
|
return out
|
|
finally:
|
|
try:
|
|
if tmp.exists():
|
|
tmp.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# -----------------------------
|
|
# Models
|
|
# -----------------------------
|
|
class CoverRequest(BaseModel):
|
|
filename: str
|
|
|
|
|
|
class RegionRequest(BaseModel):
|
|
region: str
|
|
|
|
|
|
class RegionMetaRequest(BaseModel):
|
|
name: str | None = None
|
|
coordinates: list[float] | None = None # [lng, lat]
|
|
|
|
|
|
# -----------------------------
|
|
# Routes
|
|
# -----------------------------
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "healthy", "service": "travel-proxy"}
|
|
|
|
|
|
@app.get("/api/travel/regions")
|
|
def regions():
|
|
geojson = load_regions_geojson()
|
|
region_map = load_region_map()
|
|
|
|
# GeoJSON에 이미 있는 지역 ID 수집
|
|
existing_ids = {
|
|
f.get("properties", {}).get("id")
|
|
for f in geojson.get("features", [])
|
|
}
|
|
|
|
# region_map에 있지만 GeoJSON에 없는 커스텀 지역을 동적 추가
|
|
# _regions_meta에 좌표가 있으면 Point geometry로, 없으면 null
|
|
from .indexer import _load_extra
|
|
extra = _load_extra(THUMB_ROOT)
|
|
regions_meta = extra.get("_regions_meta", {})
|
|
|
|
for region_id in region_map:
|
|
if region_id in existing_ids:
|
|
continue
|
|
meta = regions_meta.get(region_id, {})
|
|
coords = meta.get("coordinates") # [lng, lat] or None
|
|
geometry = (
|
|
{"type": "Point", "coordinates": coords}
|
|
if coords and len(coords) == 2
|
|
else None
|
|
)
|
|
geojson.setdefault("features", []).append({
|
|
"type": "Feature",
|
|
"properties": {
|
|
"id": region_id,
|
|
"name": meta.get("name", region_id),
|
|
"custom": True,
|
|
},
|
|
"geometry": geometry,
|
|
})
|
|
|
|
return 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),
|
|
):
|
|
region_map = load_region_map()
|
|
albums = _get_albums_for_region(region, region_map)
|
|
result = get_photos_by_region(albums, page, size)
|
|
|
|
# URL 조합 (DB에는 경로를 저장하지 않음)
|
|
items = []
|
|
for row in result["items"]:
|
|
items.append({
|
|
"album": row["album"],
|
|
"file": row["filename"],
|
|
"url": f"{MEDIA_BASE}/{row['album']}/{row['filename']}",
|
|
"thumb": f"{MEDIA_BASE}/.thumb/{row['album']}/{row['filename']}",
|
|
"mtime": row["mtime"],
|
|
})
|
|
|
|
return {
|
|
"region": region,
|
|
"page": page,
|
|
"size": size,
|
|
"total": result["total"],
|
|
"has_next": result["has_next"],
|
|
"items": items,
|
|
"matched_albums": result["matched_albums"],
|
|
}
|
|
|
|
|
|
@app.post("/api/travel/sync")
|
|
def sync_endpoint():
|
|
result = sync(
|
|
travel_root=ROOT,
|
|
thumb_root=THUMB_ROOT,
|
|
region_map_path=REGION_MAP_PATH,
|
|
)
|
|
return result
|
|
|
|
|
|
@app.get("/api/travel/albums")
|
|
def albums_list():
|
|
rows = get_all_albums()
|
|
region_map = load_region_map()
|
|
|
|
# album → region 역인덱스
|
|
album_to_region = {}
|
|
for region_id, v in region_map.items():
|
|
album_list = v if isinstance(v, list) else v.get("albums", [])
|
|
for a in album_list:
|
|
album_to_region[a] = region_id
|
|
|
|
# region → name 매핑 (GeoJSON)
|
|
region_names = {}
|
|
try:
|
|
geojson = load_regions_geojson()
|
|
for feat in geojson.get("features", []):
|
|
props = feat.get("properties", {})
|
|
rid = props.get("id")
|
|
if rid:
|
|
region_names[rid] = props.get("name", rid)
|
|
except Exception:
|
|
pass
|
|
|
|
result = []
|
|
for r in rows:
|
|
cover = r["cover_filename"]
|
|
region_id = album_to_region.get(r["album"], "미분류")
|
|
result.append({
|
|
"album": r["album"],
|
|
"count": r["count"],
|
|
"region": region_id,
|
|
"regionName": region_names.get(region_id, region_id),
|
|
"cover_url": f"{MEDIA_BASE}/{r['album']}/{cover}",
|
|
"cover_thumb": f"{MEDIA_BASE}/.thumb/{r['album']}/{cover}",
|
|
})
|
|
return result
|
|
|
|
|
|
@app.put("/api/travel/regions/{region_id}")
|
|
def update_region_meta(region_id: str, body: RegionMetaRequest):
|
|
"""커스텀 지역의 이름/좌표 수정 (Phase 2: 지도 핀 표시용)."""
|
|
from .indexer import _load_extra, _save_extra
|
|
extra = _load_extra(THUMB_ROOT)
|
|
meta = extra.setdefault("_regions_meta", {})
|
|
entry = meta.setdefault(region_id, {})
|
|
if body.name is not None:
|
|
entry["name"] = body.name
|
|
if body.coordinates is not None:
|
|
if len(body.coordinates) != 2:
|
|
raise HTTPException(400, "coordinates must be [lng, lat]")
|
|
entry["coordinates"] = body.coordinates
|
|
_save_extra(THUMB_ROOT, extra)
|
|
return {"region_id": region_id, **entry}
|
|
|
|
|
|
@app.put("/api/travel/albums/{album}/region")
|
|
def set_region(album: str, body: RegionRequest):
|
|
new_region = body.region.strip()
|
|
if not new_region:
|
|
raise HTTPException(400, "Region name cannot be empty")
|
|
result = move_album_region(THUMB_ROOT, REGION_MAP_PATH, album, new_region)
|
|
return result
|
|
|
|
|
|
@app.put("/api/travel/albums/{album}/cover")
|
|
def set_cover(album: str, body: CoverRequest):
|
|
ok = set_album_cover(album, body.filename)
|
|
if not ok:
|
|
raise HTTPException(404, f"Photo not found: {album}/{body.filename}")
|
|
return {
|
|
"album": album,
|
|
"filename": body.filename,
|
|
"cover_url": f"{MEDIA_BASE}/{album}/{body.filename}",
|
|
"cover_thumb": f"{MEDIA_BASE}/.thumb/{album}/{body.filename}",
|
|
}
|
|
|
|
|
|
@app.get("/media/travel/.thumb/{album}/{filename}")
|
|
def get_thumb(album: str, filename: str):
|
|
if ".." in album or ".." in filename:
|
|
raise HTTPException(400, "Invalid path")
|
|
src = (ROOT / album / filename).resolve()
|
|
if not str(src).startswith(str(ROOT)):
|
|
raise HTTPException(403, "Access denied")
|
|
if not src.exists() or not src.is_file():
|
|
raise HTTPException(404, "Source not found")
|
|
p = _ensure_thumb_fallback(src, album)
|
|
if not p.exists() or not p.is_file():
|
|
raise HTTPException(404, "Thumbnail not found")
|
|
return FileResponse(str(p))
|
|
|
|
|
|
@app.get("/api/version")
|
|
def version():
|
|
return {"version": os.getenv("APP_VERSION", "dev")}
|