feat(music-lab): 파일 해시 기반 라이브러리 동기화 — rename 시 태그 보존

- music_library에 file_hash(MD5) 컬럼 추가
- _sync_library_with_disk를 3단계로 변경:
  1. 파일명 매칭 (빠른 경로)
  2. 해시 비교로 rename 감지 → 기존 레코드 업데이트 (태그 보존)
  3. 나머지 → 삭제/추가
- 파일명 변경 시 audio_url 업데이트 → 다운로드도 새 이름 적용

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

music-lab/app/db.py

@@ -72,6 +72,7 @@ def init_db() -> None:
         for col, default in [
             ("provider", "'local'"), ("lyrics", "''"),
             ("image_url", "''"), ("suno_id", "''"),
+            ("file_hash", "''"),
         ]:
             try:
                 conn.execute(f"ALTER TABLE music_library ADD COLUMN {col} TEXT NOT NULL DEFAULT {default}")
@@ -159,6 +160,7 @@ def _track_row_to_dict(r) -> Dict[str, Any]:
         "lyrics":       r["lyrics"] if "lyrics" in keys else "",
         "image_url":    r["image_url"] if "image_url" in keys else "",
         "suno_id":      r["suno_id"] if "suno_id" in keys else "",
+        "file_hash":    r["file_hash"] if "file_hash" in keys else "",
         "created_at":   r["created_at"],
     }
 
@@ -176,8 +178,8 @@ def add_track(data: Dict[str, Any]) -> Dict[str, Any]:
             INSERT INTO music_library
                 (title, genre, moods, instruments, duration_sec, bpm, key, scale,
                  prompt, audio_url, file_path, task_id, tags,
-                 provider, lyrics, image_url, suno_id)
+                 provider, lyrics, image_url, suno_id, file_hash)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             """,
             (
                 data.get("title", ""),
@@ -197,6 +199,7 @@ def add_track(data: Dict[str, Any]) -> Dict[str, Any]:
                 data.get("lyrics", ""),
                 data.get("image_url", ""),
                 data.get("suno_id", ""),
+                data.get("file_hash", ""),
             ),
         )
         row = conn.execute("SELECT * FROM music_library WHERE rowid = last_insert_rowid()").fetchone()
@@ -227,6 +230,24 @@ def update_track_duration(track_id: int, duration_sec: int) -> None:
         )
 
 
+def update_track_file_info(track_id: int, title: str, audio_url: str, file_path: str) -> None:
+    """파일 rename 시 파일 관련 정보만 업데이트 (태그 등 메타데이터 보존)."""
+    with _conn() as conn:
+        conn.execute(
+            "UPDATE music_library SET title=?, audio_url=?, file_path=? WHERE id=?",
+            (title, audio_url, file_path, track_id),
+        )
+
+
+def update_track_hash(track_id: int, file_hash: str) -> None:
+    """트랙의 file_hash를 업데이트."""
+    with _conn() as conn:
+        conn.execute(
+            "UPDATE music_library SET file_hash=? WHERE id=?",
+            (file_hash, track_id),
+        )
+
+
 def get_track_file_path(track_id: int) -> Optional[str]:
     with _conn() as conn:
         row = conn.execute("SELECT file_path FROM music_library WHERE id = ?", (track_id,)).fetchone()

music-lab/app/main.py

@@ -9,7 +9,7 @@ from .db import (
     init_db,
     create_task, get_task,
     get_all_tracks, add_track, delete_track, get_track_file_path, get_track_by_task_id,
-    update_track_duration,
+    update_track_duration, update_track_file_info, update_track_hash,
     get_all_lyrics, add_lyrics, update_lyrics, delete_lyrics,
 )
 from .local_provider import run_local_generation
@@ -203,10 +203,25 @@ def list_library():
     return {"tracks": get_all_tracks()}
 
 
+def _calc_file_hash(file_path: str) -> str:
+    """MD5 해시 계산 (파일 동일성 체크용)."""
+    import hashlib
+    h = hashlib.md5()
+    try:
+        with open(file_path, "rb") as f:
+            for chunk in iter(lambda: f.read(8192), b""):
+                h.update(chunk)
+        return h.hexdigest()
+    except OSError:
+        return ""
+
+
 def _sync_library_with_disk():
-    """파일시스템의 .mp3 파일과 DB를 동기화.
+    """파일시스템의 .mp3 파일과 DB를 동기화 (해시 기반 rename 감지).
-    - 디스크에 없는 트랙 → DB에서 삭제
+
-    - DB에 없는 .mp3 파일 → 새 트랙으로 추가
+    1단계: 파일명 매칭 (빠른 경로)
+    2단계: 미매칭 파일/레코드를 해시로 비교 → rename 감지 → 메타데이터 보존 업데이트
+    3단계: 나머지 → 삭제/추가
     """
     tracks = get_all_tracks()
     media_base = os.getenv("MUSIC_MEDIA_BASE", "/media/music")
@@ -218,31 +233,83 @@ def _sync_library_with_disk():
             if f.lower().endswith(".mp3"):
                 disk_files.add(f)
     except OSError:
-        return  # 디렉토리 접근 불가 시 동기화 스킵
+        return
 
-    # DB 트랙의 파일명 매핑
+    # ── 1단계: 파일명 매칭 ──────────────────────────────────────
-    db_filenames = {}  # filename → track
+    db_by_filename = {}  # filename → track
     for t in tracks:
         if t.get("audio_url"):
             fname = t["audio_url"].split("/")[-1]
-            db_filenames[fname] = t
+            db_by_filename[fname] = t
 
-    # DB에는 있지만 디스크에 없는 → 삭제
+    matched_disk = set()
-    for fname, t in db_filenames.items():
+    matched_db_ids = set()
-        if fname not in disk_files:
+
+    for f in disk_files:
+        if f in db_by_filename:
+            matched_disk.add(f)
+            track = db_by_filename[f]
+            matched_db_ids.add(track["id"])
+            # 기존 트랙에 file_hash 없으면 채우기
+            if not track.get("file_hash"):
+                file_hash = _calc_file_hash(os.path.join(MUSIC_DATA_DIR, f))
+                if file_hash:
+                    update_track_hash(track["id"], file_hash)
+
+    unmatched_disk = disk_files - matched_disk
+    unmatched_db = [t for t in tracks if t["id"] not in matched_db_ids]
+
+    # ── 2단계: 해시 기반 rename 감지 ────────────────────────────
+    if unmatched_disk and unmatched_db:
+        # DB 미매칭 레코드의 해시 맵
+        db_hash_map = {}  # hash → track
+        for t in unmatched_db:
+            h = t.get("file_hash", "")
+            if h:
+                db_hash_map[h] = t
+
+        resolved_disk = set()
+        resolved_db_ids = set()
+
+        for f in unmatched_disk:
+            file_path = os.path.join(MUSIC_DATA_DIR, f)
+            file_hash = _calc_file_hash(file_path)
+            if not file_hash:
+                continue
+
+            if file_hash in db_hash_map:
+                # rename 감지 — 기존 레코드 업데이트 (태그·메타데이터 보존)
+                track = db_hash_map[file_hash]
+                new_title = os.path.splitext(f)[0].replace("-", " ").replace("_", " ")
+                update_track_file_info(
+                    track["id"],
+                    title=new_title,
+                    audio_url=f"{media_base}/{f}",
+                    file_path=file_path,
+                )
+                resolved_disk.add(f)
+                resolved_db_ids.add(track["id"])
+
+        unmatched_disk -= resolved_disk
+        unmatched_db = [t for t in unmatched_db if t["id"] not in resolved_db_ids]
+
+    # ── 3단계: 나머지 처리 ──────────────────────────────────────
+    # DB에만 남은 레코드 → 파일 삭제됨 → DB 삭제
+    for t in unmatched_db:
         delete_track(t["id"])
 
-    # 디스크에는 있지만 DB에 없는 → 추가 (duration 자동 추출)
+    # 디스크에만 남은 파일 → 신규 → DB 추가 (해시 포함)
-    for f in disk_files:
+    for f in unmatched_disk:
-        if f not in db_filenames:
         file_path = os.path.join(MUSIC_DATA_DIR, f)
         title = os.path.splitext(f)[0].replace("-", " ").replace("_", " ")
+        file_hash = _calc_file_hash(file_path)
         add_track({
             "title": title,
             "audio_url": f"{media_base}/{f}",
             "file_path": file_path,
             "provider": "suno",
             "duration_sec": _get_mp3_duration(file_path),
+            "file_hash": file_hash,
         })