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>
330 lines
11 KiB
Python
330 lines
11 KiB
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:
|
|
try:
|
|
mtime = entry.stat().st_mtime
|
|
except OSError as e:
|
|
logger.warning("Cannot stat %s: %s", entry.path, e)
|
|
continue
|
|
items.append({
|
|
"filename": entry.name,
|
|
"mtime": 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 _load_region_map_merged(region_map_path: Path, thumb_root: Path) -> dict:
|
|
"""
|
|
원본 region_map.json + RW 오버라이드 파일 병합.
|
|
원본은 RO 마운트이므로 신규 폴더는 thumb_root/region_map_extra.json에 저장.
|
|
"""
|
|
try:
|
|
with open(region_map_path, "r", encoding="utf-8") as f:
|
|
region_map = json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError) as e:
|
|
logger.error("Failed to load region_map: %s", e)
|
|
raise
|
|
|
|
# 오버라이드 파일 병합
|
|
extra = _load_extra(thumb_root)
|
|
if extra:
|
|
# _removes: 원본에서 제거할 항목 적용
|
|
removes = extra.get("_removes", {})
|
|
for region_id, remove_albums in removes.items():
|
|
if region_id in region_map:
|
|
existing = region_map[region_id]
|
|
if isinstance(existing, list):
|
|
region_map[region_id] = [a for a in existing if a not in remove_albums]
|
|
elif isinstance(existing, dict) and "albums" in existing:
|
|
existing["albums"] = [a for a in existing["albums"] if a not in remove_albums]
|
|
|
|
# 나머지 키 병합 (_removes, _regions_meta 제외)
|
|
for k, v in extra.items():
|
|
if k.startswith("_"):
|
|
continue
|
|
if k in region_map:
|
|
existing = region_map[k]
|
|
if isinstance(existing, list) and isinstance(v, list):
|
|
merged = list(dict.fromkeys(existing + v))
|
|
region_map[k] = merged
|
|
else:
|
|
region_map[k] = v
|
|
|
|
return region_map
|
|
|
|
|
|
def _extra_path(thumb_root: Path) -> Path:
|
|
return thumb_root / "region_map_extra.json"
|
|
|
|
|
|
def _load_extra(thumb_root: Path) -> dict:
|
|
"""RW 오버라이드 파일 로드."""
|
|
path = _extra_path(thumb_root)
|
|
if path.exists():
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, OSError):
|
|
pass
|
|
return {}
|
|
|
|
|
|
def _save_extra(thumb_root: Path, data: dict) -> None:
|
|
"""RW 오버라이드 파일 저장."""
|
|
path = _extra_path(thumb_root)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
|
|
|
|
def _save_extra_albums(thumb_root: Path, new_folders: List[str]) -> None:
|
|
"""신규 발견 폴더를 RW 영역의 오버라이드 파일에 저장."""
|
|
extra = _load_extra(thumb_root)
|
|
uncategorized = extra.get("미분류", [])
|
|
if isinstance(uncategorized, dict):
|
|
uncategorized = uncategorized.get("albums", [])
|
|
existing = set(uncategorized)
|
|
added = [f for f in new_folders if f not in existing]
|
|
if added:
|
|
uncategorized.extend(added)
|
|
extra["미분류"] = uncategorized
|
|
try:
|
|
_save_extra(thumb_root, extra)
|
|
logger.info("Saved %d new folders to region_map_extra.json → 미분류", len(added))
|
|
except OSError as e:
|
|
logger.warning("Cannot write region_map_extra: %s", e)
|
|
|
|
|
|
def move_album_region(
|
|
thumb_root: Path,
|
|
region_map_path: Path,
|
|
album: str,
|
|
new_region: str,
|
|
) -> dict:
|
|
"""
|
|
앨범의 지역을 변경. extra 파일에서 이전 지역에서 제거 + 새 지역에 추가.
|
|
원본 region_map.json의 매핑을 오버라이드 하기 위해
|
|
extra에 _remove 키로 원본에서 제거할 항목을 기록.
|
|
|
|
Returns: {"album": str, "old_region": str, "new_region": str}
|
|
"""
|
|
# 현재 병합 맵에서 앨범의 기존 지역 조회
|
|
merged = _load_region_map_merged(region_map_path, thumb_root)
|
|
old_region = None
|
|
for region_id, v in merged.items():
|
|
album_list = v if isinstance(v, list) else v.get("albums", [])
|
|
if album in album_list:
|
|
old_region = region_id
|
|
break
|
|
|
|
extra = _load_extra(thumb_root)
|
|
|
|
# 1. 기존 지역에서 제거
|
|
if old_region:
|
|
# extra에 해당 지역의 앨범 목록이 있으면 거기서 제거
|
|
if old_region in extra:
|
|
old_list = extra[old_region]
|
|
if isinstance(old_list, dict):
|
|
old_list = old_list.get("albums", [])
|
|
if album in old_list:
|
|
old_list.remove(album)
|
|
extra[old_region] = old_list
|
|
|
|
# 원본 region_map에 있는 경우 _overrides로 제거 기록
|
|
try:
|
|
with open(region_map_path, "r", encoding="utf-8") as f:
|
|
original = json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
original = {}
|
|
|
|
orig_list = original.get(old_region, [])
|
|
if isinstance(orig_list, dict):
|
|
orig_list = orig_list.get("albums", [])
|
|
if album in orig_list:
|
|
removes = extra.setdefault("_removes", {})
|
|
remove_list = removes.setdefault(old_region, [])
|
|
if album not in remove_list:
|
|
remove_list.append(album)
|
|
|
|
# 2. 새 지역에 추가
|
|
new_list = extra.get(new_region, [])
|
|
if isinstance(new_list, dict):
|
|
new_list = new_list.get("albums", [])
|
|
if album not in new_list:
|
|
new_list.append(album)
|
|
extra[new_region] = new_list
|
|
|
|
# _removes에서 새 지역의 해당 앨범은 제거 (되돌리기 대응)
|
|
if "_removes" in extra and new_region in extra["_removes"]:
|
|
if album in extra["_removes"][new_region]:
|
|
extra["_removes"][new_region].remove(album)
|
|
|
|
_save_extra(thumb_root, extra)
|
|
logger.info("Moved album '%s': %s → %s", album, old_region or "없음", new_region)
|
|
|
|
return {"album": album, "old_region": old_region or "미분류", "new_region": new_region}
|
|
|
|
|
|
def sync(
|
|
travel_root: Path,
|
|
thumb_root: Path,
|
|
region_map_path: Path,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
폴더 스캔 → DB 동기화 + 썸네일 일괄 생성.
|
|
|
|
Returns:
|
|
{"added": int, "removed": int, "thumbs_generated": int,
|
|
"discovered": int, "duration_sec": float}
|
|
"""
|
|
start = time.time()
|
|
|
|
# 1. region_map.json + 오버라이드 병합
|
|
region_map = _load_region_map_merged(region_map_path, thumb_root)
|
|
|
|
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"])
|
|
|
|
# 1-b. 파일시스템에서 신규 폴더 자동 탐색
|
|
discovered = 0
|
|
new_folders = []
|
|
try:
|
|
with os.scandir(travel_root) as entries:
|
|
for entry in entries:
|
|
if not entry.is_dir() or entry.name.startswith("_"):
|
|
continue
|
|
if entry.name in all_albums:
|
|
continue
|
|
# 이미지가 하나라도 있는 폴더만 추가
|
|
folder = travel_root / entry.name
|
|
has_images = any(
|
|
Path(f.name).suffix.lower() in IMAGE_EXT
|
|
for f in os.scandir(folder)
|
|
if f.is_file()
|
|
)
|
|
if has_images:
|
|
all_albums.add(entry.name)
|
|
new_folders.append(entry.name)
|
|
discovered += 1
|
|
logger.info("Discovered new album folder: %s", entry.name)
|
|
except OSError as e:
|
|
logger.warning("Failed to scan travel_root for new folders: %s", e)
|
|
|
|
# 신규 폴더를 RW 오버라이드 파일에 영구 저장
|
|
if new_folders:
|
|
_save_extra_albums(thumb_root, new_folders)
|
|
|
|
# 2. 각 앨범 폴더 스캔 → DB 배치 동기화
|
|
added = 0
|
|
updated = 0
|
|
removed = 0
|
|
sorted_albums = sorted(all_albums)
|
|
total_albums = len(sorted_albums)
|
|
|
|
logger.info("Syncing %d albums...", total_albums)
|
|
for i, album in enumerate(sorted_albums, 1):
|
|
folder = travel_root / album
|
|
items = _scan_folder(folder)
|
|
existing_filenames = {item["filename"] for item in items}
|
|
|
|
result = db.batch_sync_album(album, items, existing_filenames)
|
|
added += result["added"]
|
|
updated += result["updated"]
|
|
removed += result["removed"]
|
|
|
|
if result["added"] or result["removed"]:
|
|
logger.info(
|
|
"[%d/%d] %s: +%d/-%d (%d files)",
|
|
i, total_albums, album, result["added"], result["removed"], len(items),
|
|
)
|
|
|
|
logger.info("DB sync done: added=%d updated=%d removed=%d", added, updated, removed)
|
|
|
|
# 3. 썸네일 미생성 분 일괄 생성
|
|
no_thumb = db.get_photos_without_thumb()
|
|
thumbs_generated = 0
|
|
thumb_done_batch = []
|
|
|
|
if no_thumb:
|
|
logger.info("Generating %d thumbnails...", len(no_thumb))
|
|
|
|
for idx, photo in enumerate(no_thumb, 1):
|
|
src = travel_root / photo["album"] / photo["filename"]
|
|
dest = thumb_root / photo["album"] / photo["filename"]
|
|
if _generate_thumb(src, dest):
|
|
thumb_done_batch.append(photo)
|
|
thumbs_generated += 1
|
|
if idx % 100 == 0:
|
|
logger.info("Thumbnails: %d/%d done", idx, len(no_thumb))
|
|
|
|
db.batch_mark_thumbs_done(thumb_done_batch)
|
|
|
|
duration = round(time.time() - start, 2)
|
|
logger.info(
|
|
"Sync complete: added=%d updated=%d removed=%d thumbs=%d discovered=%d duration=%.2fs",
|
|
added, updated, removed, thumbs_generated, discovered, duration,
|
|
)
|
|
|
|
return {
|
|
"added": added,
|
|
"updated": updated,
|
|
"removed": removed,
|
|
"thumbs_generated": thumbs_generated,
|
|
"discovered": discovered,
|
|
"duration_sec": duration,
|
|
}
|