feat(travel-proxy): db.py — SQLite 스키마 + 쿼리 헬퍼

travel-proxy/app/db.py

@@ -0,0 +1,182 @@
+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),
+        )