"Write Tags to File" in the Enhanced Library Manager

core/tag_writer.py

@@ -0,0 +1,487 @@
+"""
+Tag Writer — reads current file tags and writes DB metadata into audio file tags.
+Supports MP3 (ID3v2.4), FLAC, OGG Vorbis, and MP4/M4A.
+Reuses the same Mutagen patterns as _enhance_file_metadata in web_server.py.
+"""
+
+import os
+import logging
+import urllib.request
+from typing import Dict, Any, Optional, List, Tuple
+
+from mutagen import File as MutagenFile
+from mutagen.id3 import ID3, TIT2, TPE1, TALB, TDRC, TRCK, TCON, TPE2, TPOS, TXXX, APIC, TBPM
+from mutagen.flac import FLAC, Picture
+from mutagen.mp4 import MP4, MP4Cover, MP4FreeForm
+from mutagen.oggvorbis import OggVorbis
+from mutagen.apev2 import APEv2, APENoHeaderError
+
+logger = logging.getLogger("tag_writer")
+
+# Supported extensions
+SUPPORTED_EXTENSIONS = {'.mp3', '.flac', '.ogg', '.oga', '.opus', '.m4a', '.mp4'}
+
+
+def read_file_tags(file_path: str) -> Dict[str, Any]:
+    """
+    Read current tags from an audio file. Returns a dict of tag values
+    that can be compared against DB metadata.
+    """
+    result = {
+        'title': None,
+        'artist': None,
+        'album_artist': None,
+        'album': None,
+        'year': None,
+        'genre': None,
+        'track_number': None,
+        'disc_number': None,
+        'bpm': None,
+        'has_cover_art': False,
+        'format': None,
+        'error': None,
+    }
+
+    if not file_path or not os.path.exists(file_path):
+        result['error'] = 'File not found'
+        return result
+
+    ext = os.path.splitext(file_path)[1].lower()
+    if ext not in SUPPORTED_EXTENSIONS:
+        result['error'] = f'Unsupported format: {ext}'
+        return result
+
+    try:
+        audio = MutagenFile(file_path)
+        if audio is None:
+            result['error'] = 'Could not read file with Mutagen'
+            return result
+
+        result['format'] = ext.lstrip('.').upper()
+
+        if isinstance(audio.tags, ID3):
+            # MP3
+            result['title'] = _id3_text(audio.tags, 'TIT2')
+            result['artist'] = _id3_text(audio.tags, 'TPE1')
+            result['album_artist'] = _id3_text(audio.tags, 'TPE2')
+            result['album'] = _id3_text(audio.tags, 'TALB')
+            result['year'] = _id3_text(audio.tags, 'TDRC')
+            result['genre'] = _id3_text(audio.tags, 'TCON')
+            result['track_number'] = _parse_track_num(_id3_text(audio.tags, 'TRCK'))
+            result['disc_number'] = _parse_track_num(_id3_text(audio.tags, 'TPOS'))
+            bpm_text = _id3_text(audio.tags, 'TBPM')
+            if bpm_text:
+                try:
+                    result['bpm'] = float(bpm_text)
+                except (ValueError, TypeError):
+                    pass
+            result['has_cover_art'] = bool(audio.tags.getall('APIC'))
+
+        elif isinstance(audio, (FLAC, OggVorbis)):
+            # FLAC / OGG
+            result['title'] = _vorbis_first(audio, 'title')
+            result['artist'] = _vorbis_first(audio, 'artist')
+            result['album_artist'] = _vorbis_first(audio, 'albumartist')
+            result['album'] = _vorbis_first(audio, 'album')
+            result['year'] = _vorbis_first(audio, 'date')
+            result['genre'] = _vorbis_first(audio, 'genre')
+            result['track_number'] = _parse_track_num(_vorbis_first(audio, 'tracknumber'))
+            result['disc_number'] = _parse_track_num(_vorbis_first(audio, 'discnumber'))
+            bpm_val = _vorbis_first(audio, 'bpm')
+            if bpm_val:
+                try:
+                    result['bpm'] = float(bpm_val)
+                except (ValueError, TypeError):
+                    pass
+            if isinstance(audio, FLAC):
+                result['has_cover_art'] = bool(audio.pictures)
+            else:
+                # OGG doesn't have a standard picture field we can easily check
+                result['has_cover_art'] = False
+
+        elif isinstance(audio, MP4):
+            # MP4 / M4A
+            result['title'] = _mp4_first(audio, '\xa9nam')
+            result['artist'] = _mp4_first(audio, '\xa9ART')
+            result['album_artist'] = _mp4_first(audio, 'aART')
+            result['album'] = _mp4_first(audio, '\xa9alb')
+            result['year'] = _mp4_first(audio, '\xa9day')
+            result['genre'] = _mp4_first(audio, '\xa9gen')
+            trkn = audio.tags.get('trkn', []) if audio.tags else []
+            if trkn:
+                result['track_number'] = trkn[0][0] if isinstance(trkn[0], tuple) else None
+            disk = audio.tags.get('disk', []) if audio.tags else []
+            if disk:
+                result['disc_number'] = disk[0][0] if isinstance(disk[0], tuple) else None
+            result['has_cover_art'] = bool(audio.tags.get('covr', [])) if audio.tags else False
+
+    except Exception as e:
+        result['error'] = str(e)
+
+    return result
+
+
+def build_tag_diff(file_tags: Dict[str, Any], db_data: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """
+    Compare file tags against DB metadata. Returns a list of diffs:
+    [{ field, file_value, db_value, changed }]
+    """
+    fields = [
+        ('title', 'title', 'Title'),
+        ('artist', 'artist_name', 'Artist'),
+        ('album', 'album_title', 'Album'),
+        ('album_artist', 'artist_name', 'Album Artist'),
+        ('year', 'year', 'Year'),
+        ('genre', 'genres', 'Genre'),
+        ('track_number', 'track_number', 'Track #'),
+        ('disc_number', 'disc_number', 'Disc #'),
+        ('bpm', 'bpm', 'BPM'),
+    ]
+
+    diffs = []
+    for file_key, db_key, label in fields:
+        file_val = file_tags.get(file_key)
+        db_val = db_data.get(db_key)
+
+        # Normalize for comparison
+        file_str = _normalize_for_compare(file_val)
+        db_str = _normalize_for_compare(db_val)
+
+        # Special: genres can be a list in DB
+        if db_key == 'genres' and isinstance(db_val, list):
+            db_str = ', '.join(db_val) if db_val else ''
+            db_val = db_str if db_str else None
+
+        # Special: year — DB stores int, file stores string
+        if db_key == 'year' and db_val is not None:
+            db_str = str(db_val)
+            db_val = str(db_val)
+
+        # Only mark as changed if DB has a value AND it differs from file
+        # (writer skips fields where DB value is empty, so don't show them as diffs)
+        changed = bool(db_str) and file_str != db_str
+        diffs.append({
+            'field': label,
+            'file_key': file_key,
+            'file_value': str(file_val) if file_val is not None else '',
+            'db_value': str(db_val) if db_val is not None else '',
+            'changed': changed,
+        })
+
+    # Cover art — special row
+    diffs.append({
+        'field': 'Cover Art',
+        'file_key': 'cover_art',
+        'file_value': 'Embedded' if file_tags.get('has_cover_art') else 'None',
+        'db_value': 'Available' if db_data.get('thumb_url') else 'None',
+        'changed': not file_tags.get('has_cover_art') and bool(db_data.get('thumb_url')),
+    })
+
+    return diffs
+
+
+def download_cover_art(cover_url: str) -> Optional[Tuple[bytes, str]]:
+    """
+    Download cover art once. Returns (image_data, mime_type) or None on failure.
+    Call this once per album, then pass the result to write_tags_to_file for each track.
+    """
+    if not cover_url:
+        return None
+    try:
+        with urllib.request.urlopen(cover_url, timeout=15) as response:
+            image_data = response.read()
+            mime_type = response.info().get_content_type() or 'image/jpeg'
+        if image_data:
+            return (image_data, mime_type)
+    except Exception as e:
+        logger.error(f"Error downloading cover art from {cover_url}: {e}")
+    return None
+
+
+def write_tags_to_file(file_path: str, db_data: Dict[str, Any],
+                       embed_cover: bool = True,
+                       cover_url: Optional[str] = None,
+                       cover_data: Optional[Tuple[bytes, str]] = None) -> Dict[str, Any]:
+    """
+    Write DB metadata into audio file tags. Only writes fields that have DB values.
+    Returns { success, written_fields, error }
+
+    For cover art, pass either:
+    - cover_url: downloads art on-the-fly (fine for single track)
+    - cover_data: (image_bytes, mime_type) tuple from download_cover_art() (preferred for batch)
+    """
+    if not file_path or not os.path.exists(file_path):
+        return {'success': False, 'error': 'File not found'}
+
+    ext = os.path.splitext(file_path)[1].lower()
+    if ext not in SUPPORTED_EXTENSIONS:
+        return {'success': False, 'error': f'Unsupported format: {ext}'}
+
+    try:
+        audio = MutagenFile(file_path)
+        if audio is None:
+            return {'success': False, 'error': 'Could not open file with Mutagen'}
+
+        # Ensure tags exist
+        if audio.tags is None:
+            audio.add_tags()
+
+        written = []
+
+        # Build metadata dict from DB data
+        title = db_data.get('title')
+        artist = db_data.get('artist_name')
+        album = db_data.get('album_title')
+        album_artist = db_data.get('artist_name')  # Use artist name as album artist
+        year = db_data.get('year')
+        genres = db_data.get('genres')
+        track_num = db_data.get('track_number')
+        total_tracks = db_data.get('track_count')
+        disc_num = db_data.get('disc_number')
+        bpm = db_data.get('bpm')
+
+        # Genre: list → comma string
+        genre_str = None
+        if genres:
+            if isinstance(genres, list):
+                genre_str = ', '.join(genres) if genres else None
+            elif isinstance(genres, str):
+                genre_str = genres
+
+        if isinstance(audio.tags, ID3):
+            written = _write_id3(audio, title, artist, album_artist, album,
+                                 year, genre_str, track_num, total_tracks,
+                                 disc_num, bpm)
+        elif isinstance(audio, (FLAC, OggVorbis)):
+            written = _write_vorbis(audio, title, artist, album_artist, album,
+                                    year, genre_str, track_num, total_tracks,
+                                    disc_num, bpm)
+        elif isinstance(audio, MP4):
+            written = _write_mp4(audio, title, artist, album_artist, album,
+                                 year, genre_str, track_num, total_tracks,
+                                 disc_num, bpm)
+
+        # Embed cover art if requested
+        if embed_cover:
+            art_ok = False
+            if cover_data:
+                # Use pre-downloaded art (batch mode)
+                art_ok = _embed_cover_art_data(audio, cover_data[0], cover_data[1])
+            elif cover_url:
+                # Download on-the-fly (single track mode)
+                art_ok = _embed_cover_art(audio, cover_url)
+            if art_ok:
+                written.append('cover_art')
+
+        # Save
+        if isinstance(audio.tags, ID3):
+            audio.save(v1=0, v2_version=4)
+        elif isinstance(audio, FLAC):
+            audio.save(deleteid3=True)
+        else:
+            audio.save()
+
+        return {'success': True, 'written_fields': written}
+
+    except PermissionError:
+        return {'success': False, 'error': 'Permission denied — file may be in use'}
+    except Exception as e:
+        logger.error(f"Error writing tags to {file_path}: {e}")
+        return {'success': False, 'error': str(e)}
+
+
+# ── Format-specific writers ──
+
+def _write_id3(audio, title, artist, album_artist, album, year, genre,
+               track_num, total_tracks, disc_num, bpm) -> List[str]:
+    written = []
+    if title:
+        audio.tags.delall('TIT2')
+        audio.tags.add(TIT2(encoding=3, text=[title]))
+        written.append('title')
+    if artist:
+        audio.tags.delall('TPE1')
+        audio.tags.add(TPE1(encoding=3, text=[artist]))
+        written.append('artist')
+    if album_artist:
+        audio.tags.delall('TPE2')
+        audio.tags.add(TPE2(encoding=3, text=[album_artist]))
+        written.append('album_artist')
+    if album:
+        audio.tags.delall('TALB')
+        audio.tags.add(TALB(encoding=3, text=[album]))
+        written.append('album')
+    if year is not None:
+        audio.tags.delall('TDRC')
+        audio.tags.add(TDRC(encoding=3, text=[str(year)]))
+        written.append('year')
+    if genre:
+        audio.tags.delall('TCON')
+        audio.tags.add(TCON(encoding=3, text=[genre]))
+        written.append('genre')
+    if track_num is not None:
+        audio.tags.delall('TRCK')
+        trk_str = f"{track_num}/{total_tracks}" if total_tracks else str(track_num)
+        audio.tags.add(TRCK(encoding=3, text=[trk_str]))
+        written.append('track_number')
+    if disc_num is not None:
+        audio.tags.delall('TPOS')
+        audio.tags.add(TPOS(encoding=3, text=[str(disc_num)]))
+        written.append('disc_number')
+    if bpm is not None:
+        audio.tags.delall('TBPM')
+        audio.tags.add(TBPM(encoding=3, text=[str(int(bpm))]))
+        written.append('bpm')
+    return written
+
+
+def _write_vorbis(audio, title, artist, album_artist, album, year, genre,
+                  track_num, total_tracks, disc_num, bpm) -> List[str]:
+    written = []
+    if title:
+        audio['title'] = [title]
+        written.append('title')
+    if artist:
+        audio['artist'] = [artist]
+        written.append('artist')
+    if album_artist:
+        audio['albumartist'] = [album_artist]
+        written.append('album_artist')
+    if album:
+        audio['album'] = [album]
+        written.append('album')
+    if year is not None:
+        audio['date'] = [str(year)]
+        written.append('year')
+    if genre:
+        audio['genre'] = [genre]
+        written.append('genre')
+    if track_num is not None:
+        trk_str = f"{track_num}/{total_tracks}" if total_tracks else str(track_num)
+        audio['tracknumber'] = [trk_str]
+        written.append('track_number')
+    if disc_num is not None:
+        audio['discnumber'] = [str(disc_num)]
+        written.append('disc_number')
+    if bpm is not None:
+        audio['bpm'] = [str(int(bpm))]
+        written.append('bpm')
+    return written
+
+
+def _write_mp4(audio, title, artist, album_artist, album, year, genre,
+               track_num, total_tracks, disc_num, bpm) -> List[str]:
+    written = []
+    if title:
+        audio['\xa9nam'] = [title]
+        written.append('title')
+    if artist:
+        audio['\xa9ART'] = [artist]
+        written.append('artist')
+    if album_artist:
+        audio['aART'] = [album_artist]
+        written.append('album_artist')
+    if album:
+        audio['\xa9alb'] = [album]
+        written.append('album')
+    if year is not None:
+        audio['\xa9day'] = [str(year)]
+        written.append('year')
+    if genre:
+        audio['\xa9gen'] = [genre]
+        written.append('genre')
+    if track_num is not None:
+        total = total_tracks or 0
+        audio['trkn'] = [(track_num, total)]
+        written.append('track_number')
+    if disc_num is not None:
+        audio['disk'] = [(disc_num, 0)]
+        written.append('disc_number')
+    if bpm is not None:
+        audio['tmpo'] = [int(bpm)]
+        written.append('bpm')
+    return written
+
+
+def _embed_cover_art(audio, cover_url: str) -> bool:
+    """Download and embed cover art from URL (single-track convenience)."""
+    try:
+        result = download_cover_art(cover_url)
+        if not result:
+            return False
+        return _embed_cover_art_data(audio, result[0], result[1])
+    except Exception as e:
+        logger.error(f"Error embedding cover art: {e}")
+        return False
+
+
+def _embed_cover_art_data(audio, image_data: bytes, mime_type: str) -> bool:
+    """Embed pre-downloaded cover art bytes into an audio file object."""
+    try:
+        if not image_data:
+            return False
+
+        if isinstance(audio.tags, ID3):
+            audio.tags.delall('APIC')
+            audio.tags.add(APIC(encoding=3, mime=mime_type, type=3, desc='Cover', data=image_data))
+        elif isinstance(audio, FLAC):
+            audio.clear_pictures()
+            picture = Picture()
+            picture.data = image_data
+            picture.type = 3
+            picture.mime = mime_type
+            picture.width = 640
+            picture.height = 640
+            picture.depth = 24
+            audio.add_picture(picture)
+        elif isinstance(audio, MP4):
+            fmt = MP4Cover.FORMAT_JPEG if 'jpeg' in mime_type else MP4Cover.FORMAT_PNG
+            audio['covr'] = [MP4Cover(image_data, imageformat=fmt)]
+
+        return True
+    except Exception as e:
+        logger.error(f"Error embedding cover art data: {e}")
+        return False
+
+
+# ── Helpers ──
+
+def _id3_text(tags, frame_id: str) -> Optional[str]:
+    frames = tags.getall(frame_id)
+    if frames and frames[0].text:
+        return str(frames[0].text[0])
+    return None
+
+
+def _vorbis_first(audio, key: str) -> Optional[str]:
+    vals = audio.get(key, [])
+    return vals[0] if vals else None
+
+
+def _mp4_first(audio, key: str) -> Optional[str]:
+    vals = audio.tags.get(key, []) if audio.tags else []
+    return str(vals[0]) if vals else None
+
+
+def _parse_track_num(val) -> Optional[int]:
+    """Parse track number from '3/12' or '3' format."""
+    if val is None:
+        return None
+    try:
+        return int(str(val).split('/')[0])
+    except (ValueError, IndexError):
+        return None
+
+
+def _normalize_for_compare(val) -> str:
+    """Normalize a value for comparison."""
+    if val is None:
+        return ''
+    if isinstance(val, (list, tuple)):
+        return ', '.join(str(v) for v in val) if val else ''
+    # Normalize numeric types: 120.0 → '120', 120 → '120'
+    if isinstance(val, float):
+        return str(int(val)) if val == int(val) else str(val)
+    if isinstance(val, int):
+        return str(val)
+    return str(val).strip()

web_server.py

@@ -8815,6 +8815,353 @@ def batch_update_library_tracks():
     except Exception as e:
         return jsonify({"success": False, "error": str(e)}), 500
 
+# ── Write Tags to File endpoints ──
+
+@app.route('/api/library/track/<track_id>/tag-preview', methods=['GET'])
+def get_track_tag_preview(track_id):
+    """Read current file tags and compare against DB metadata for a single track."""
+    try:
+        from core.tag_writer import read_file_tags, build_tag_diff
+        database = get_database()
+
+        # Get track + album + artist data from DB
+        conn = database._get_connection()
+        cursor = conn.cursor()
+        cursor.execute("""
+            SELECT t.*, a.name as artist_name, al.title as album_title,
+                   al.year, al.genres as album_genres, al.track_count,
+                   al.thumb_url as album_thumb_url, a.thumb_url as artist_thumb_url
+            FROM tracks t
+            JOIN artists a ON t.artist_id = a.id
+            JOIN albums al ON t.album_id = al.id
+            WHERE t.id = ?
+        """, (str(track_id),))
+        row = cursor.fetchone()
+        if not row:
+            return jsonify({"success": False, "error": "Track not found"}), 404
+
+        track_data = dict(row)
+        file_path = track_data.get('file_path')
+
+        # Resolve path if needed
+        resolved_path = _resolve_library_file_path(file_path)
+        if not resolved_path:
+            return jsonify({"success": False, "error": "File not found on disk", "file_path": file_path}), 404
+
+        # Read current file tags
+        file_tags = read_file_tags(resolved_path)
+        if file_tags.get('error'):
+            return jsonify({"success": False, "error": file_tags['error']}), 400
+
+        # Parse album genres for diff
+        album_genres = []
+        if track_data.get('album_genres'):
+            try:
+                import json as _json
+                parsed = _json.loads(track_data['album_genres'])
+                album_genres = parsed if isinstance(parsed, list) else [str(parsed)]
+            except (ValueError, TypeError):
+                album_genres = [g.strip() for g in track_data['album_genres'].split(',') if g.strip()]
+
+        # Build DB metadata dict for comparison
+        db_data = {
+            'title': track_data.get('title'),
+            'artist_name': track_data.get('artist_name'),
+            'album_title': track_data.get('album_title'),
+            'year': track_data.get('year'),
+            'genres': album_genres,
+            'track_number': track_data.get('track_number'),
+            'disc_number': track_data.get('disc_number'),
+            'bpm': track_data.get('bpm'),
+            'track_count': track_data.get('track_count'),
+            'thumb_url': track_data.get('album_thumb_url') or track_data.get('artist_thumb_url'),
+        }
+
+        diff = build_tag_diff(file_tags, db_data)
+        has_changes = any(d['changed'] for d in diff)
+
+        return jsonify({
+            "success": True,
+            "file_path": resolved_path,
+            "file_tags": file_tags,
+            "db_data": db_data,
+            "diff": diff,
+            "has_changes": has_changes,
+        })
+
+    except Exception as e:
+        logger.error(f"Tag preview error for track {track_id}: {e}")
+        return jsonify({"success": False, "error": str(e)}), 500
+
+
+@app.route('/api/library/track/<track_id>/write-tags', methods=['POST'])
+def write_track_tags(track_id):
+    """Write DB metadata into the audio file tags for a single track."""
+    try:
+        from core.tag_writer import write_tags_to_file
+        database = get_database()
+        data = request.get_json() or {}
+        embed_cover = data.get('embed_cover', True)
+
+        # Get full track data
+        conn = database._get_connection()
+        cursor = conn.cursor()
+        cursor.execute("""
+            SELECT t.*, a.name as artist_name, al.title as album_title,
+                   al.year, al.genres as album_genres, al.track_count,
+                   al.thumb_url as album_thumb_url, a.thumb_url as artist_thumb_url
+            FROM tracks t
+            JOIN artists a ON t.artist_id = a.id
+            JOIN albums al ON t.album_id = al.id
+            WHERE t.id = ?
+        """, (str(track_id),))
+        row = cursor.fetchone()
+        if not row:
+            return jsonify({"success": False, "error": "Track not found"}), 404
+
+        track_data = dict(row)
+        file_path = track_data.get('file_path')
+        resolved_path = _resolve_library_file_path(file_path)
+        if not resolved_path:
+            return jsonify({"success": False, "error": "File not found on disk"}), 404
+
+        # Parse genres
+        album_genres = []
+        if track_data.get('album_genres'):
+            try:
+                import json as _json
+                parsed = _json.loads(track_data['album_genres'])
+                album_genres = parsed if isinstance(parsed, list) else [str(parsed)]
+            except (ValueError, TypeError):
+                album_genres = [g.strip() for g in track_data['album_genres'].split(',') if g.strip()]
+
+        # Build data for writer
+        db_data = {
+            'title': track_data.get('title'),
+            'artist_name': track_data.get('artist_name'),
+            'album_title': track_data.get('album_title'),
+            'year': track_data.get('year'),
+            'genres': album_genres,
+            'track_number': track_data.get('track_number'),
+            'disc_number': track_data.get('disc_number'),
+            'bpm': track_data.get('bpm'),
+            'track_count': track_data.get('track_count'),
+        }
+
+        # Resolve cover URL
+        cover_url = None
+        if embed_cover:
+            thumb = track_data.get('album_thumb_url') or track_data.get('artist_thumb_url')
+            if thumb and thumb.startswith('http'):
+                cover_url = thumb
+
+        # Use file lock for thread safety
+        file_lock = _get_file_lock(resolved_path)
+        with file_lock:
+            result = write_tags_to_file(resolved_path, db_data, embed_cover=embed_cover, cover_url=cover_url)
+
+        return jsonify(result)
+
+    except Exception as e:
+        logger.error(f"Write tags error for track {track_id}: {e}")
+        return jsonify({"success": False, "error": str(e)}), 500
+
+
+_write_tags_batch_state = {
+    'status': 'idle',  # idle | running | done
+    'total': 0,
+    'processed': 0,
+    'written': 0,
+    'failed': 0,
+    'current_track': '',
+    'errors': [],
+}
+_write_tags_batch_lock = threading.Lock()
+
+
+@app.route('/api/library/tracks/write-tags-batch', methods=['POST'])
+def write_tracks_tags_batch():
+    """Write DB metadata into audio file tags for multiple tracks (runs in background)."""
+    try:
+        with _write_tags_batch_lock:
+            if _write_tags_batch_state['status'] == 'running':
+                return jsonify({"success": False, "error": "A batch tag write is already in progress"}), 409
+
+        database = get_database()
+        data = request.get_json()
+        if not data or not data.get('track_ids'):
+            return jsonify({"success": False, "error": "track_ids required"}), 400
+
+        track_ids = data['track_ids']
+        embed_cover = data.get('embed_cover', True)
+
+        # Fetch all track data upfront (in the request thread, fast DB query)
+        conn = database._get_connection()
+        cursor = conn.cursor()
+        placeholders = ','.join('?' * len(track_ids))
+        cursor.execute(f"""
+            SELECT t.*, a.name as artist_name, al.title as album_title,
+                   al.year, al.genres as album_genres, al.track_count,
+                   al.thumb_url as album_thumb_url, a.thumb_url as artist_thumb_url
+            FROM tracks t
+            JOIN artists a ON t.artist_id = a.id
+            JOIN albums al ON t.album_id = al.id
+            WHERE t.id IN ({placeholders})
+        """, [str(tid) for tid in track_ids])
+
+        rows = [dict(r) for r in cursor.fetchall()]
+
+        # Initialize state
+        with _write_tags_batch_lock:
+            _write_tags_batch_state.update({
+                'status': 'running',
+                'total': len(track_ids),
+                'processed': 0,
+                'written': 0,
+                'failed': 0,
+                'current_track': '',
+                'errors': [],
+            })
+
+        # Count missing DB rows
+        found_ids = {str(r['id']) for r in rows}
+        missing = [tid for tid in track_ids if str(tid) not in found_ids]
+        if missing:
+            with _write_tags_batch_lock:
+                _write_tags_batch_state['failed'] += len(missing)
+                _write_tags_batch_state['processed'] += len(missing)
+                for tid in missing:
+                    _write_tags_batch_state['errors'].append({'track_id': tid, 'error': 'Track not found in database'})
+
+        # Run the actual writes in a background thread
+        def _run_batch():
+            try:
+                from core.tag_writer import write_tags_to_file, download_cover_art
+
+                # Pre-download cover art once per unique album URL
+                cover_cache = {}  # url → (bytes, mime) or None
+                if embed_cover:
+                    unique_urls = set()
+                    for td in rows:
+                        thumb = td.get('album_thumb_url') or td.get('artist_thumb_url')
+                        if thumb and thumb.startswith('http'):
+                            unique_urls.add(thumb)
+                    if unique_urls:
+                        with _write_tags_batch_lock:
+                            _write_tags_batch_state['current_track'] = f'Downloading cover art ({len(unique_urls)} album{"s" if len(unique_urls) != 1 else ""})...'
+                        for url in unique_urls:
+                            cover_cache[url] = download_cover_art(url)
+
+                for track_data in rows:
+                    file_path = track_data.get('file_path')
+                    resolved_path = _resolve_library_file_path(file_path)
+                    track_title = track_data.get('title', 'Unknown')
+
+                    with _write_tags_batch_lock:
+                        _write_tags_batch_state['current_track'] = track_title
+
+                    if not resolved_path:
+                        with _write_tags_batch_lock:
+                            _write_tags_batch_state['failed'] += 1
+                            _write_tags_batch_state['processed'] += 1
+                            _write_tags_batch_state['errors'].append({'track_id': track_data['id'], 'error': 'File not found'})
+                        continue
+
+                    # Parse genres
+                    album_genres = []
+                    if track_data.get('album_genres'):
+                        try:
+                            parsed = json.loads(track_data['album_genres'])
+                            album_genres = parsed if isinstance(parsed, list) else [str(parsed)]
+                        except (ValueError, TypeError):
+                            album_genres = [g.strip() for g in track_data['album_genres'].split(',') if g.strip()]
+
+                    db_data = {
+                        'title': track_data.get('title'),
+                        'artist_name': track_data.get('artist_name'),
+                        'album_title': track_data.get('album_title'),
+                        'year': track_data.get('year'),
+                        'genres': album_genres,
+                        'track_number': track_data.get('track_number'),
+                        'disc_number': track_data.get('disc_number'),
+                        'bpm': track_data.get('bpm'),
+                        'track_count': track_data.get('track_count'),
+                    }
+
+                    # Get pre-downloaded cover art for this track's album
+                    art_data = None
+                    if embed_cover:
+                        thumb = track_data.get('album_thumb_url') or track_data.get('artist_thumb_url')
+                        if thumb and thumb.startswith('http'):
+                            art_data = cover_cache.get(thumb)
+
+                    file_lock = _get_file_lock(resolved_path)
+                    with file_lock:
+                        write_result = write_tags_to_file(
+                            resolved_path, db_data,
+                            embed_cover=embed_cover,
+                            cover_data=art_data
+                        )
+
+                    with _write_tags_batch_lock:
+                        _write_tags_batch_state['processed'] += 1
+                        if write_result.get('success'):
+                            _write_tags_batch_state['written'] += 1
+                        else:
+                            _write_tags_batch_state['failed'] += 1
+                            _write_tags_batch_state['errors'].append({
+                                'track_id': track_data['id'],
+                                'error': write_result.get('error', 'Unknown')
+                            })
+
+            except Exception as e:
+                logger.error(f"Batch write tags background error: {e}")
+            finally:
+                with _write_tags_batch_lock:
+                    _write_tags_batch_state['status'] = 'done'
+                    _write_tags_batch_state['current_track'] = ''
+
+        thread = threading.Thread(target=_run_batch, daemon=True, name="WriteTagsBatch")
+        thread.start()
+
+        return jsonify({"success": True, "message": "Batch tag write started", "total": len(track_ids)})
+
+    except Exception as e:
+        logger.error(f"Batch write tags error: {e}")
+        with _write_tags_batch_lock:
+            _write_tags_batch_state['status'] = 'idle'
+        return jsonify({"success": False, "error": str(e)}), 500
+
+
+@app.route('/api/library/tracks/write-tags-batch/status', methods=['GET'])
+def get_write_tags_batch_status():
+    """Poll the status of a running batch tag write."""
+    with _write_tags_batch_lock:
+        state = dict(_write_tags_batch_state)
+        state['errors'] = list(_write_tags_batch_state['errors'])  # snapshot to avoid mutation during serialize
+        return jsonify(state)
+
+
+def _resolve_library_file_path(file_path):
+    """Resolve a library file path to an actual file on disk."""
+    if not file_path:
+        return None
+    if os.path.exists(file_path):
+        return file_path
+
+    # Try resolving server-side paths to local transfer path
+    transfer_dir = docker_resolve_path(config_manager.get('soulseek.transfer_path', './Transfer'))
+    path_parts = file_path.replace('\\', '/').split('/')
+
+    # Try progressively shorter path suffixes (skip index 0 to avoid drive letter issues)
+    for i in range(1, len(path_parts)):
+        candidate = os.path.join(transfer_dir, *path_parts[i:])
+        if os.path.exists(candidate):
+            return candidate
+
+    return None
+
+
 @app.route('/api/library/play', methods=['POST'])
 def library_play_track():
     """Start playing a track directly from the user's library (no download needed)."""

webui/index.html

@@ -2347,10 +2347,34 @@
                 </div>
                 <div class="enhanced-bulk-bar-actions">
                     <button class="enhanced-bulk-btn secondary" onclick="showBulkEditModal()">Edit Selected</button>
+                    <button class="enhanced-bulk-btn tag-write" onclick="batchWriteTagsSelected()">Write Tags</button>
                     <button class="enhanced-bulk-btn clear" onclick="clearTrackSelection()">Clear Selection</button>
                 </div>
             </div>
 
+            <!-- Tag Preview Modal -->
+            <div class="modal-overlay hidden" id="tag-preview-overlay">
+                <div class="enhanced-bulk-modal tag-preview-modal">
+                    <div class="enhanced-bulk-modal-header">
+                        <h3 id="tag-preview-title">Write Tags to File</h3>
+                        <button class="enhanced-bulk-modal-close" onclick="closeTagPreviewModal()">&times;</button>
+                    </div>
+                    <div class="enhanced-bulk-modal-body" id="tag-preview-body">
+                        <!-- Populated dynamically -->
+                    </div>
+                    <div class="enhanced-bulk-modal-footer">
+                        <label class="tag-preview-cover-label">
+                            <input type="checkbox" id="tag-preview-embed-cover" checked>
+                            Embed cover art
+                        </label>
+                        <div class="tag-preview-footer-actions">
+                            <button class="enhanced-bulk-btn secondary" onclick="closeTagPreviewModal()">Cancel</button>
+                            <button class="enhanced-bulk-btn primary" id="tag-preview-write-btn" onclick="executeWriteTags()">Write Tags</button>
+                        </div>
+                    </div>
+                </div>
+            </div>
+
             <!-- Discover Page -->
             <div class="page" id="discover-page">
                 <div class="discover-container">

webui/static/script.js

@@ -33717,6 +33717,14 @@ function renderExpandedAlbumHeader(album) {
     albumEnrichWrap.appendChild(albumEnrichMenu);
     enrichRow.appendChild(albumEnrichWrap);
 
+    // Write Tags button for entire album
+    const writeTagsBtn = document.createElement('button');
+    writeTagsBtn.className = 'enhanced-write-tags-album-btn';
+    writeTagsBtn.innerHTML = '✎ Write All Tags';
+    writeTagsBtn.title = 'Write DB metadata to file tags for all tracks in this album';
+    writeTagsBtn.onclick = (e) => { e.stopPropagation(); writeAlbumTags(album.id); };
+    enrichRow.appendChild(writeTagsBtn);
+
     // Delete album button
     const deleteAlbumBtn = document.createElement('button');
     deleteAlbumBtn.className = 'enhanced-delete-album-btn';
@@ -33822,6 +33830,7 @@ function renderTrackTable(album) {
         { label: 'File', cls: 'col-path' },
         { label: 'Match', cls: 'col-match' },
         { label: '', cls: 'col-queue' },
+        { label: '', cls: 'col-writetag' },
         { label: '', cls: 'col-delete' },
     ];
     columns.forEach(col => {
@@ -34022,6 +34031,19 @@ function renderTrackTable(album) {
         }
         tr.appendChild(queueTd);
 
+        // Write Tags button
+        const tagTd = document.createElement('td');
+        tagTd.className = 'col-writetag';
+        if (track.file_path) {
+            const tagBtn = document.createElement('button');
+            tagBtn.className = 'enhanced-write-tag-btn';
+            tagBtn.innerHTML = '✎';
+            tagBtn.title = 'Write tags to file';
+            tagBtn.onclick = (e) => { e.stopPropagation(); showTagPreview(track.id); };
+            tagTd.appendChild(tagBtn);
+        }
+        tr.appendChild(tagTd);
+
         // Delete button
         const delTd = document.createElement('td');
         delTd.className = 'col-delete';
@@ -34837,6 +34859,173 @@ document.addEventListener('click', (e) => {
     }
 });
 
+// ---- Write Tags to File ----
+
+let _tagPreviewTrackId = null;
+
+async function showTagPreview(trackId) {
+    _tagPreviewTrackId = trackId;
+    const overlay = document.getElementById('tag-preview-overlay');
+    const body = document.getElementById('tag-preview-body');
+    const title = document.getElementById('tag-preview-title');
+    if (!overlay || !body) return;
+
+    title.textContent = 'Write Tags to File';
+    body.innerHTML = '<div class="tag-preview-loading">Loading tag comparison...</div>';
+    overlay.classList.remove('hidden');
+
+    try {
+        const response = await fetch(`/api/library/track/${trackId}/tag-preview`);
+        const result = await response.json();
+        if (!result.success) {
+            body.innerHTML = `<div class="tag-preview-error">${escapeHtml(result.error)}</div>`;
+            return;
+        }
+
+        const diff = result.diff || [];
+        const hasChanges = result.has_changes;
+
+        let html = '<table class="tag-preview-table"><thead><tr>';
+        html += '<th>Field</th><th>Current File Tag</th><th></th><th>DB Value</th>';
+        html += '</tr></thead><tbody>';
+
+        diff.forEach(d => {
+            const rowClass = d.changed ? 'tag-diff-changed' : 'tag-diff-same';
+            const arrow = d.changed ? '<span class="tag-diff-arrow">&rarr;</span>' : '<span class="tag-diff-check">&#10003;</span>';
+            html += `<tr class="${rowClass}">`;
+            html += `<td class="tag-field-name">${d.field}</td>`;
+            html += `<td class="tag-file-value">${escapeHtml(d.file_value) || '<span class="tag-empty">empty</span>'}</td>`;
+            html += `<td class="tag-diff-indicator">${arrow}</td>`;
+            html += `<td class="tag-db-value">${escapeHtml(d.db_value) || '<span class="tag-empty">empty</span>'}</td>`;
+            html += '</tr>';
+        });
+
+        html += '</tbody></table>';
+
+        if (!hasChanges) {
+            html += '<div class="tag-preview-no-changes">File tags already match DB metadata</div>';
+        }
+
+        body.innerHTML = html;
+
+        const writeBtn = document.getElementById('tag-preview-write-btn');
+        if (writeBtn) {
+            writeBtn.disabled = !hasChanges && !document.getElementById('tag-preview-embed-cover')?.checked;
+        }
+
+    } catch (error) {
+        body.innerHTML = `<div class="tag-preview-error">Failed to load preview: ${escapeHtml(error.message)}</div>`;
+    }
+}
+
+function closeTagPreviewModal() {
+    const overlay = document.getElementById('tag-preview-overlay');
+    if (overlay) overlay.classList.add('hidden');
+    _tagPreviewTrackId = null;
+}
+
+async function executeWriteTags() {
+    if (!_tagPreviewTrackId) return;
+
+    const writeBtn = document.getElementById('tag-preview-write-btn');
+    if (writeBtn) {
+        writeBtn.disabled = true;
+        writeBtn.textContent = 'Writing...';
+    }
+
+    const embedCover = document.getElementById('tag-preview-embed-cover')?.checked ?? true;
+
+    try {
+        const response = await fetch(`/api/library/track/${_tagPreviewTrackId}/write-tags`, {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ embed_cover: embedCover })
+        });
+        const result = await response.json();
+        if (!result.success) throw new Error(result.error);
+
+        const fieldCount = (result.written_fields || []).length;
+        showToast(`Tags written successfully (${fieldCount} fields)`, 'success');
+        closeTagPreviewModal();
+
+    } catch (error) {
+        showToast(`Failed to write tags: ${error.message}`, 'error');
+    } finally {
+        if (writeBtn) {
+            writeBtn.disabled = false;
+            writeBtn.textContent = 'Write Tags';
+        }
+    }
+}
+
+async function writeAlbumTags(albumId) {
+    const album = findEnhancedAlbum(albumId);
+    if (!album) return;
+
+    const tracks = (album.tracks || []).filter(t => t.file_path);
+    if (tracks.length === 0) {
+        showToast('No tracks with files in this album', 'error');
+        return;
+    }
+
+    if (!confirm(`Write DB metadata to file tags for all ${tracks.length} tracks in "${album.title}"?`)) return;
+    await _startBatchWriteTags(tracks.map(t => t.id), true);
+}
+
+async function batchWriteTagsSelected() {
+    const trackIds = Array.from(artistDetailPageState.selectedTracks);
+    if (trackIds.length === 0) return;
+
+    if (!confirm(`Write DB metadata to file tags for ${trackIds.length} selected track(s)?`)) return;
+    await _startBatchWriteTags(trackIds, true);
+}
+
+async function _startBatchWriteTags(trackIds, embedCover) {
+    try {
+        const response = await fetch('/api/library/tracks/write-tags-batch', {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ track_ids: trackIds, embed_cover: embedCover })
+        });
+        const result = await response.json();
+        if (!result.success) throw new Error(result.error);
+
+        showToast(`Writing tags for ${trackIds.length} tracks...`, 'info');
+        _pollBatchWriteTagsStatus();
+
+    } catch (error) {
+        showToast(`Failed to start tag write: ${error.message}`, 'error');
+    }
+}
+
+let _batchWriteTagsPollTimer = null;
+
+function _pollBatchWriteTagsStatus() {
+    if (_batchWriteTagsPollTimer) clearTimeout(_batchWriteTagsPollTimer);
+
+    async function poll() {
+        try {
+            const response = await fetch('/api/library/tracks/write-tags-batch/status');
+            const state = await response.json();
+
+            if (state.status === 'running') {
+                const pct = state.total > 0 ? Math.round(state.processed / state.total * 100) : 0;
+                showToast(`Writing tags: ${state.processed}/${state.total} (${pct}%) — ${state.current_track}`, 'info');
+                _batchWriteTagsPollTimer = setTimeout(poll, 1000);
+            } else if (state.status === 'done') {
+                const msg = `Tags written: ${state.written} succeeded, ${state.failed} failed`;
+                showToast(msg, state.failed > 0 ? 'warning' : 'success');
+                _batchWriteTagsPollTimer = null;
+            }
+        } catch (error) {
+            console.error('Poll write-tags status failed:', error);
+            _batchWriteTagsPollTimer = null;
+        }
+    }
+
+    _batchWriteTagsPollTimer = setTimeout(poll, 800);
+}
+
 async function playLibraryTrack(track, albumTitle, artistName) {
     if (!track.file_path) {
         showToast('No file available for this track', 'error');

webui/static/style.css

@@ -31825,4 +31825,190 @@ textarea.enhanced-meta-field-input {
     color: rgba(29, 185, 84, 0.9);
     border-color: rgba(29, 185, 84, 0.4);
     transform: scale(1.1);
+}
+
+/* ── Write Tags to File ── */
+
+.enhanced-write-tag-btn {
+    width: 26px;
+    height: 26px;
+    border-radius: 6px;
+    border: 1px solid rgba(29, 185, 84, 0.15);
+    background: rgba(29, 185, 84, 0.04);
+    color: rgba(29, 185, 84, 0.5);
+    cursor: pointer;
+    font-size: 13px;
+    display: flex;
+    align-items: center;
+    justify-content: center;
+    transition: all 0.15s ease;
+    padding: 0;
+}
+.enhanced-write-tag-btn:hover {
+    background: rgba(29, 185, 84, 0.12);
+    color: rgba(29, 185, 84, 0.9);
+    border-color: rgba(29, 185, 84, 0.4);
+    transform: scale(1.1);
+}
+
+.enhanced-write-tags-album-btn {
+    padding: 6px 14px;
+    border-radius: 8px;
+    font-size: 12px;
+    font-weight: 500;
+    cursor: pointer;
+    background: rgba(29, 185, 84, 0.06);
+    border: 1px solid rgba(29, 185, 84, 0.15);
+    color: rgba(29, 185, 84, 0.6);
+    transition: all 0.15s ease;
+}
+.enhanced-write-tags-album-btn:hover {
+    background: rgba(29, 185, 84, 0.12);
+    color: rgba(29, 185, 84, 0.9);
+    border-color: rgba(29, 185, 84, 0.35);
+}
+
+.enhanced-bulk-btn.tag-write {
+    background: rgba(29, 185, 84, 0.08);
+    border-color: rgba(29, 185, 84, 0.2);
+    color: rgb(var(--accent-rgb));
+}
+.enhanced-bulk-btn.tag-write:hover {
+    background: rgba(29, 185, 84, 0.15);
+    border-color: rgba(29, 185, 84, 0.4);
+}
+
+.col-writetag {
+    width: 32px;
+    text-align: center;
+}
+
+/* Tag Preview Modal */
+.tag-preview-modal {
+    max-width: 700px;
+    width: 90vw;
+}
+
+.tag-preview-loading {
+    text-align: center;
+    padding: 30px;
+    color: rgba(255, 255, 255, 0.5);
+    font-size: 14px;
+}
+
+.tag-preview-error {
+    text-align: center;
+    padding: 20px;
+    color: rgba(255, 80, 80, 0.8);
+    font-size: 14px;
+}
+
+.tag-preview-table {
+    width: 100%;
+    border-collapse: collapse;
+    font-size: 13px;
+}
+
+.tag-preview-table th {
+    text-align: left;
+    padding: 8px 10px;
+    color: rgba(255, 255, 255, 0.4);
+    font-weight: 500;
+    font-size: 11px;
+    text-transform: uppercase;
+    letter-spacing: 0.5px;
+    border-bottom: 1px solid rgba(255, 255, 255, 0.06);
+}
+
+.tag-preview-table td {
+    padding: 7px 10px;
+    border-bottom: 1px solid rgba(255, 255, 255, 0.03);
+    color: rgba(255, 255, 255, 0.7);
+}
+
+.tag-field-name {
+    color: rgba(255, 255, 255, 0.5);
+    font-weight: 500;
+    white-space: nowrap;
+    width: 90px;
+}
+
+.tag-file-value {
+    max-width: 200px;
+    overflow: hidden;
+    text-overflow: ellipsis;
+    white-space: nowrap;
+}
+
+.tag-db-value {
+    max-width: 200px;
+    overflow: hidden;
+    text-overflow: ellipsis;
+    white-space: nowrap;
+}
+
+.tag-diff-indicator {
+    width: 30px;
+    text-align: center;
+}
+
+.tag-diff-arrow {
+    color: rgb(var(--accent-rgb));
+    font-size: 16px;
+    font-weight: bold;
+}
+
+.tag-diff-check {
+    color: rgba(255, 255, 255, 0.15);
+    font-size: 14px;
+}
+
+tr.tag-diff-changed .tag-file-value {
+    color: rgba(255, 80, 80, 0.6);
+    text-decoration: line-through;
+}
+
+tr.tag-diff-changed .tag-db-value {
+    color: rgb(var(--accent-rgb));
+    font-weight: 500;
+}
+
+tr.tag-diff-same {
+    opacity: 0.5;
+}
+
+.tag-empty {
+    color: rgba(255, 255, 255, 0.15);
+    font-style: italic;
+}
+
+.tag-preview-no-changes {
+    text-align: center;
+    padding: 16px;
+    color: rgba(29, 185, 84, 0.6);
+    font-size: 13px;
+    border-top: 1px solid rgba(255, 255, 255, 0.04);
+    margin-top: 8px;
+}
+
+.tag-preview-cover-label {
+    display: flex;
+    align-items: center;
+    gap: 8px;
+    color: rgba(255, 255, 255, 0.6);
+    font-size: 13px;
+    cursor: pointer;
+}
+
+.tag-preview-cover-label input[type="checkbox"] {
+    accent-color: rgb(var(--accent-rgb));
+}
+
+.tag-preview-footer-actions {
+    display: flex;
+    gap: 8px;
+}
+
+.tag-preview-modal .enhanced-bulk-modal-footer {
+    justify-content: space-between;
 }