# Travel-Proxy 성능 개선 구현 계획

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** travel-proxy의 os.scandir 기반 아키텍처를 SQLite 인덱스 DB로 전환하고, 앨범 커버 지정 + 썸네일 사전 생성을 지원한다.

**Architecture:** 기존 main.py의 스캔/캐시/썸네일 로직을 db.py(스키마+쿼리)와 indexer.py(동기화+썸네일)로 분리. main.py는 라우트만 담당. DB 경로는 `/data/thumbs/travel.db`.

**Tech Stack:** Python 3.12, FastAPI, SQLite (표준 라이브러리 sqlite3), Pillow

---

### 파일 구조

| 파일 | 역할 | 상태 |
|------|------|------|
| `travel-proxy/app/db.py` | SQLite 스키마 정의, 커넥션 헬퍼, 쿼리 함수 | 신규 |
| `travel-proxy/app/indexer.py` | 폴더 스캔 → DB 동기화 + 썸네일 일괄 생성 | 신규 |
| `travel-proxy/app/main.py` | FastAPI 라우트 (기존 수정 + 신규 추가) | 수정 |

---

### Task 1: db.py — SQLite 스키마 및 쿼리 헬퍼

**Files:**
- Create: `travel-proxy/app/db.py`

- [ ] **Step 1: db.py 파일 생성**

```python
import os
import sqlite3
from typing import Any, Dict, List, Optional

DB_PATH = os.getenv("TRAVEL_DB_PATH", "/data/thumbs/travel.db")


def _conn() -> sqlite3.Connection:
    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    return conn


def init_db() -> None:
    with _conn() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS photos (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                album TEXT NOT NULL,
                filename TEXT NOT NULL,
                mtime REAL NOT NULL,
                has_thumb INTEGER DEFAULT 0,
                indexed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
                UNIQUE(album, filename)
            )
        """)
        conn.execute("CREATE INDEX IF NOT EXISTS idx_photos_album ON photos(album)")

        conn.execute("""
            CREATE TABLE IF NOT EXISTS album_covers (
                album TEXT PRIMARY KEY,
                filename TEXT NOT NULL,
                updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
            )
        """)


def get_photos_by_region(albums: List[str], page: int, size: int) -> Dict[str, Any]:
    """region에 속한 앨범들의 사진을 페이지네이션하여 반환."""
    if not albums:
        return {"items": [], "total": 0, "has_next": False, "matched_albums": []}

    placeholders = ",".join("?" for _ in albums)

    with _conn() as conn:
        # 앨범별 사진 수
        rows = conn.execute(
            f"SELECT album, COUNT(*) as cnt FROM photos WHERE album IN ({placeholders}) GROUP BY album",
            albums,
        ).fetchall()
        matched_albums = [{"album": r["album"], "count": r["cnt"]} for r in rows]

        # 전체 수
        total_row = conn.execute(
            f"SELECT COUNT(*) as cnt FROM photos WHERE album IN ({placeholders})",
            albums,
        ).fetchone()
        total = total_row["cnt"]

        # 페이지네이션
        offset = (page - 1) * size
        items = conn.execute(
            f"""SELECT album, filename, mtime FROM photos
                WHERE album IN ({placeholders})
                ORDER BY album, filename
                LIMIT ? OFFSET ?""",
            [*albums, size, offset],
        ).fetchall()

    return {
        "items": [dict(r) for r in items],
        "total": total,
        "has_next": (offset + size) < total,
        "matched_albums": matched_albums,
    }


def get_all_albums() -> List[Dict[str, Any]]:
    """전체 앨범 목록 + 사진 수 + 커버 정보."""
    with _conn() as conn:
        rows = conn.execute("""
            SELECT p.album, COUNT(*) as count,
                   COALESCE(c.filename, MIN(p.filename)) as cover_filename
            FROM photos p
            LEFT JOIN album_covers c ON p.album = c.album
            GROUP BY p.album
            ORDER BY p.album
        """).fetchall()
    return [dict(r) for r in rows]


def set_album_cover(album: str, filename: str) -> bool:
    """앨범 커버 지정. 해당 photo가 존재하면 True, 없으면 False."""
    with _conn() as conn:
        exists = conn.execute(
            "SELECT 1 FROM photos WHERE album = ? AND filename = ?",
            (album, filename),
        ).fetchone()
        if not exists:
            return False

        conn.execute(
            """INSERT INTO album_covers (album, filename, updated_at)
               VALUES (?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
               ON CONFLICT(album) DO UPDATE SET
                   filename = excluded.filename,
                   updated_at = excluded.updated_at""",
            (album, filename),
        )
    return True


def get_album_cover(album: str) -> Optional[str]:
    """앨범 커버 파일명 반환. 미지정 시 None."""
    with _conn() as conn:
        row = conn.execute(
            "SELECT filename FROM album_covers WHERE album = ?",
            (album,),
        ).fetchone()
    return row["filename"] if row else None


def upsert_photo(album: str, filename: str, mtime: float) -> str:
    """사진 upsert. 반환: 'added' | 'updated' | 'unchanged'."""
    with _conn() as conn:
        existing = conn.execute(
            "SELECT mtime, has_thumb FROM photos WHERE album = ? AND filename = ?",
            (album, filename),
        ).fetchone()

        if not existing:
            conn.execute(
                "INSERT INTO photos (album, filename, mtime, has_thumb) VALUES (?, ?, ?, 0)",
                (album, filename, mtime),
            )
            return "added"
        elif existing["mtime"] != mtime:
            conn.execute(
                "UPDATE photos SET mtime = ?, has_thumb = 0 WHERE album = ? AND filename = ?",
                (mtime, album, filename),
            )
            return "updated"
        return "unchanged"


def remove_missing_photos(album: str, existing_filenames: set) -> int:
    """폴더에 없는 사진을 DB에서 제거. 제거 수 반환."""
    with _conn() as conn:
        db_rows = conn.execute(
            "SELECT filename FROM photos WHERE album = ?", (album,)
        ).fetchall()
        db_filenames = {r["filename"] for r in db_rows}
        to_remove = db_filenames - existing_filenames

        if to_remove:
            placeholders = ",".join("?" for _ in to_remove)
            conn.execute(
                f"DELETE FROM photos WHERE album = ? AND filename IN ({placeholders})",
                [album, *to_remove],
            )
            # 삭제된 파일이 커버였으면 커버도 제거
            conn.execute(
                f"DELETE FROM album_covers WHERE album = ? AND filename IN ({placeholders})",
                [album, *to_remove],
            )
    return len(to_remove)


def get_photos_without_thumb() -> List[Dict[str, str]]:
    """썸네일 미생성 사진 목록."""
    with _conn() as conn:
        rows = conn.execute(
            "SELECT album, filename FROM photos WHERE has_thumb = 0"
        ).fetchall()
    return [dict(r) for r in rows]


def mark_thumb_done(album: str, filename: str) -> None:
    """썸네일 생성 완료 표시."""
    with _conn() as conn:
        conn.execute(
            "UPDATE photos SET has_thumb = 1 WHERE album = ? AND filename = ?",
            (album, filename),
        )
```

- [ ] **Step 2: 커밋**

```bash
git add travel-proxy/app/db.py
git commit -m "feat(travel-proxy): db.py — SQLite 스키마 + 쿼리 헬퍼"
```

---

### Task 2: indexer.py — 폴더 동기화 + 썸네일 일괄 생성

**Files:**
- Create: `travel-proxy/app/indexer.py`

- [ ] **Step 1: indexer.py 파일 생성**

기존 main.py의 `ensure_thumb` 로직(라인 105-144)과 `scan_album` 로직(라인 146-166)을 기반으로 작성. `IMAGE_EXT`, `THUMB_SIZE`, 경로 상수는 main.py에서 import.

```python
import os
import time
import json
import logging
from pathlib import Path
from typing import Any, Dict, List, Set

from PIL import Image

from . import db

logger = logging.getLogger(__name__)

IMAGE_EXT = {".jpg", ".jpeg", ".png", ".webp"}
THUMB_SIZE = (480, 480)


def _scan_folder(folder: Path) -> List[Dict[str, Any]]:
    """폴더 내 이미지 파일 목록 수집 (os.scandir)."""
    if not folder.exists():
        return []
    items = []
    with os.scandir(folder) as entries:
        for entry in entries:
            if entry.is_file() and Path(entry.name).suffix.lower() in IMAGE_EXT:
                items.append({
                    "filename": entry.name,
                    "mtime": entry.stat().st_mtime,
                })
    return items


def _generate_thumb(src: Path, dest: Path) -> bool:
    """원본에서 480x480 썸네일 생성. 성공 시 True."""
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_name(dest.stem + ".tmp" + dest.suffix)
    try:
        with Image.open(src) as im:
            im.thumbnail(THUMB_SIZE)
            ext = dest.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(dest)
        return True
    except Exception as e:
        logger.warning("Thumb generation failed: %s → %s", src, e)
        try:
            if tmp.exists():
                tmp.unlink()
        except Exception:
            pass
        return False


def sync(
    travel_root: Path,
    thumb_root: Path,
    region_map_path: Path,
) -> Dict[str, Any]:
    """
    폴더 스캔 → DB 동기화 + 썸네일 일괄 생성.

    Returns:
        {"added": int, "removed": int, "thumbs_generated": int, "duration_sec": float}
    """
    start = time.time()

    # 1. region_map.json에서 전체 앨범 폴더 수집
    with open(region_map_path, "r", encoding="utf-8") as f:
        region_map = json.load(f)

    all_albums: Set[str] = set()
    for v in region_map.values():
        if isinstance(v, list):
            all_albums.update(v)
        elif isinstance(v, dict) and isinstance(v.get("albums"), list):
            all_albums.update(v["albums"])

    # 2. 각 앨범 폴더 스캔 → DB 동기화
    added = 0
    removed = 0

    for album in sorted(all_albums):
        folder = travel_root / album
        items = _scan_folder(folder)
        existing_filenames = set()

        for item in items:
            existing_filenames.add(item["filename"])
            result = db.upsert_photo(album, item["filename"], item["mtime"])
            if result == "added":
                added += 1

        removed += db.remove_missing_photos(album, existing_filenames)

    # 3. 썸네일 미생성 분 일괄 생성
    no_thumb = db.get_photos_without_thumb()
    thumbs_generated = 0

    for photo in no_thumb:
        src = travel_root / photo["album"] / photo["filename"]
        dest = thumb_root / photo["album"] / photo["filename"]
        if _generate_thumb(src, dest):
            db.mark_thumb_done(photo["album"], photo["filename"])
            thumbs_generated += 1

    duration = round(time.time() - start, 2)
    logger.info(
        "Sync complete: added=%d removed=%d thumbs=%d duration=%.2fs",
        added, removed, thumbs_generated, duration,
    )

    return {
        "added": added,
        "removed": removed,
        "thumbs_generated": thumbs_generated,
        "duration_sec": duration,
    }
```

- [ ] **Step 2: 커밋**

```bash
git add travel-proxy/app/indexer.py
git commit -m "feat(travel-proxy): indexer.py — 폴더 동기화 + 썸네일 일괄 생성"
```

---

### Task 3: main.py 리팩토링 — DB 기반 photos API + 캐시 제거

**Files:**
- Modify: `travel-proxy/app/main.py`

이 Task에서 main.py의 메모리 캐시, `scan_album()`, 기존 `photos()` 라우트를 DB 기반으로 교체한다.

- [ ] **Step 1: main.py를 DB 기반으로 재작성**

main.py 전체를 아래로 교체:

```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, get_album_cover
from .indexer import sync

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 _read_json(REGION_MAP_PATH)


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)
        return out
    finally:
        try:
            if tmp.exists():
                tmp.unlink()
        except Exception:
            pass


# -----------------------------
# Models
# -----------------------------
class CoverRequest(BaseModel):
    filename: str


# -----------------------------
# Routes
# -----------------------------
@app.get("/health")
def health():
    return {"status": "healthy", "service": "travel-proxy"}


@app.get("/api/travel/regions")
def regions():
    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),
):
    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()
    result = []
    for r in rows:
        cover = r["cover_filename"]
        result.append({
            "album": r["album"],
            "count": r["count"],
            "cover_url": f"{MEDIA_BASE}/{r['album']}/{cover}",
            "cover_thumb": f"{MEDIA_BASE}/.thumb/{r['album']}/{cover}",
        })
    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")}
```

- [ ] **Step 2: 커밋**

```bash
git add travel-proxy/app/main.py
git commit -m "refactor(travel-proxy): main.py DB 기반 전환 — 메모리 캐시 제거 + 신규 API"
```

---

### Task 4: 통합 검증

**Files:**
- 없음 (기존 파일 검증만)

- [ ] **Step 1: import 구조 확인**

travel-proxy/app/ 디렉토리에 `__init__.py`가 필요한지 확인. FastAPI uvicorn 실행 명령이 `app.main:app`이므로 패키지 import가 동작하려면 `__init__.py`가 필요.

```bash
ls travel-proxy/app/
```

`__init__.py`가 없으면 생성:

```python
# travel-proxy/app/__init__.py
```

- [ ] **Step 2: Dockerfile 확인**

현재 Dockerfile의 `COPY app /app/app` 라인이 db.py, indexer.py를 포함하는지 확인. 디렉토리 단위 복사이므로 추가 파일은 자동 포함됨. 변경 불필요.

- [ ] **Step 3: docker-compose.yml 환경변수 확인**

`TRAVEL_DB_PATH` 환경변수를 docker-compose.yml에 추가:

```yaml
# docker-compose.yml의 travel-proxy 서비스 environment에 추가
- TRAVEL_DB_PATH=${TRAVEL_DB_PATH:-/data/thumbs/travel.db}
```

- [ ] **Step 4: photos 응답 호환성 검증**

기존 응답 필드와 비교:
- `region` ✓
- `page`, `size` ✓
- `total`, `has_next` ✓
- `items[].album`, `items[].file`, `items[].url`, `items[].thumb`, `items[].mtime` ✓
- `matched_albums` — 기존에는 `photos()` 응답에 없었으나 캐시 데이터에 포함. DB 버전은 항상 포함.

- [ ] **Step 5: 커밋 (변경 있을 시)**

```bash
git add travel-proxy/app/__init__.py docker-compose.yml
git commit -m "chore(travel-proxy): __init__.py + TRAVEL_DB_PATH 환경변수 추가"
```

---

### Task 5: CLAUDE.md 업데이트

**Files:**
- Modify: `CLAUDE.md`

- [ ] **Step 1: travel-proxy 섹션에 DB 정보 추가**

CLAUDE.md의 travel-proxy 섹션에 아래 내용 추가:

- DB: `/data/thumbs/travel.db` (photos, album_covers 테이블)
- 파일 구조에 `db.py`, `indexer.py` 추가

API 목록 테이블에 신규 API 3개 추가:

| 메서드 | 경로 | 설명 |
|--------|------|------|
| POST | `/api/travel/sync` | 폴더 스캔 → DB 동기화 + 썸네일 생성 |
| GET | `/api/travel/albums` | 앨범 목록 + 사진 수 + 커버 정보 |
| PUT | `/api/travel/albums/{album}/cover` | 앨범 커버 지정 |

`POST /api/travel/reload` 제거 표기.

- [ ] **Step 2: 커밋**

```bash
git add CLAUDE.md
git commit -m "docs: CLAUDE.md travel-proxy DB·API 업데이트"
```