community-plugins/spotify-lyrics/spotify_lyrics_daemon.py

import os
import time
import json
import hashlib
import subprocess
import threading
import urllib.request
import syncedlyrics
from pathlib import Path

# Config
CACHE_DIR = Path.home() / ".cache" / "noctalia" / "lyrics"
CACHE_DIR.mkdir(parents=True, exist_ok=True)
CURRENT_STATE_FILE = CACHE_DIR / "current.json"

ART_CACHE_DIR = CACHE_DIR / "art"
ART_CACHE_DIR.mkdir(parents=True, exist_ok=True)

class SpotifyLyricsDaemon:
    def __init__(self):
        self.lyrics_cache = {}  # song_key -> list of lines
        self.fetching_keys = set()  # Tracks keys currently fetching in the background
        self.art_cache = {}     # art_url -> local file path
        self.art_fetching = set()  # URLs currently being downloaded

    def clean_filename(self, name):
        return "".join(c for c in name if c.isalnum() or c in (" ", "_", "-")).strip()

    def get_parsed_lyrics(self, title, artist):
        song_key = f"{artist} - {title}"
        if song_key in self.lyrics_cache:
            return self.lyrics_cache[song_key]
        
        # Check local disk cache first
        safe_name = self.clean_filename(song_key)
        lrc_file = CACHE_DIR / f"{safe_name}.lrc"
        
        if lrc_file.exists():
            parsed = self.load_lrc_file(lrc_file)
            self.lyrics_cache[song_key] = parsed
            return parsed
        
        # Fetch from syncedlyrics asynchronously to prevent daemon thread lag
        if song_key not in self.fetching_keys:
            self.fetching_keys.add(song_key)
            threading.Thread(
                target=self._async_fetch_lyrics, 
                args=(song_key, lrc_file), 
                daemon=True
            ).start()
        
        return []

    def _async_fetch_lyrics(self, song_key, lrc_file):
        try:
            print(f"[Daemon] Fetching lyrics in background for: {song_key}...")
            lrc_text = syncedlyrics.search(song_key, providers=["NetEase", "Lrclib"])
            if lrc_text:
                with open(lrc_file, "w", encoding="utf-8") as f:
                    f.write(lrc_text)
                
                parsed = self.parse_lrc_text(lrc_text)
                self.lyrics_cache[song_key] = parsed
                print(f"[Daemon] Fetch completed for: {song_key}")
            else:
                print(f"[Daemon] No lyrics found online for: {song_key}")
        except Exception as e:
            print(f"[Daemon] Error fetching lyrics for {song_key}: {e}")
        finally:
            self.fetching_keys.discard(song_key)

    def parse_lrc_text(self, lrc_text):
        parsed = []
        for line in lrc_text.splitlines():
            # Format: [mm:ss.xx] Text
            if line.startswith("[") and "]" in line:
                parts = line.split("]", 1)
                time_part = parts[0].replace("[", "").strip()
                text = parts[1].strip()
                
                try:
                    # mm:ss.xx or mm:ss
                    if "." in time_part:
                        min_sec, hund = time_part.split(".")
                        hund_val = int(hund) * 10 if len(hund) == 2 else int(hund)
                    else:
                        min_sec = time_part
                        hund_val = 0
                    
                    minutes, seconds = min_sec.split(":")
                    time_ms = ((int(minutes) * 60) + int(seconds)) * 1000 + hund_val
                    parsed.append({"time_ms": time_ms, "text": text})
                except Exception:
                    pass
        return parsed

    def load_lrc_file(self, lrc_file):
        try:
            with open(lrc_file, "r", encoding="utf-8") as f:
                return self.parse_lrc_text(f.read())
        except Exception as e:
            print(f"[Daemon] Error reading LRC file: {e}")
        return []

    def get_album_art_path(self, art_url):
        """Download album art from URL and return local cached file path."""
        if not art_url or art_url == "":
            return ""
        
        # Check in-memory cache
        if art_url in self.art_cache:
            path = self.art_cache[art_url]
            if os.path.exists(path):
                return path
        
        # Derive a stable filename from the URL hash
        url_hash = hashlib.md5(art_url.encode()).hexdigest()
        ext = ".jpg"  # Spotify art is always JPEG
        local_path = str(ART_CACHE_DIR / f"{url_hash}{ext}")
        
        # If already downloaded on disk, cache and return
        if os.path.exists(local_path):
            self.art_cache[art_url] = local_path
            return local_path
        
        # Download in background to avoid blocking the main loop
        if art_url not in self.art_fetching:
            self.art_fetching.add(art_url)
            threading.Thread(
                target=self._download_art,
                args=(art_url, local_path),
                daemon=True
            ).start()
        
        return ""  # Not yet available
    
    def _download_art(self, url, local_path):
        try:
            tmp_path = local_path + ".tmp"
            urllib.request.urlretrieve(url, tmp_path)
            os.replace(tmp_path, local_path)
            self.art_cache[url] = local_path
            print(f"[Daemon] Downloaded album art: {url[:60]}...")
        except Exception as e:
            print(f"[Daemon] Error downloading album art: {e}")
            # Clean up partial download
            try:
                os.remove(local_path + ".tmp")
            except OSError:
                pass
        finally:
            self.art_fetching.discard(url)

    def get_player_status(self):
        try:
            # Query active players
            players = subprocess.check_output(["playerctl", "-l"], stderr=subprocess.DEVNULL).decode("utf-8").strip().splitlines()
            if not players:
                return None
            
            # Prioritize Spotify
            player_name = "spotify" if "spotify" in players else players[0]
            
            # Query all metadata in ONE execution using custom delimiters to eliminate subprocess latency
            output = subprocess.check_output([
                "playerctl", "-p", player_name, "metadata", 
                "--format", "{{status}}|||{{position}}|||{{title}}|||{{artist}}|||{{mpris:artUrl}}"
            ], stderr=subprocess.DEVNULL).decode("utf-8").strip()
            
            parts = output.split("|||")
            if len(parts) >= 4:
                status, pos_us, title, artist = parts[0], parts[1], parts[2], parts[3]
                art_url = parts[4] if len(parts) >= 5 else ""
                
                # Position is in microseconds (us), convert to milliseconds (ms)
                position_ms = int(int(pos_us) / 1000)
                
                return {
                    "status": status,
                    "position_ms": position_ms,
                    "title": title,
                    "artist": artist,
                    "art_url": art_url
                }
        except Exception:
            pass
        return None

    def run(self):
        print("[Daemon] Starting Universal lyrics cache daemon...")
        
        while True:
            player = self.get_player_status()
            
            if not player or not player["title"]:
                # Write empty/inactive state
                empty_state = {"status": "Stopped"}
                tmp_file = CURRENT_STATE_FILE.with_suffix('.tmp')
                with open(tmp_file, "w", encoding="utf-8") as f:
                    json.dump(empty_state, f)
                tmp_file.replace(CURRENT_STATE_FILE)
                time.sleep(1.0)
                continue
            
            title = player["title"]
            artist = player["artist"]
            
            lyrics_lines = self.get_parsed_lyrics(title, artist)
            
            # Find active line
            active_idx = -1
            pos_ms = player["position_ms"]
            
            for i, line in enumerate(lyrics_lines):
                if pos_ms >= line["time_ms"]:
                    active_idx = i
                else:
                    break
            
            # Get surrounding lines
            prev_prev = lyrics_lines[active_idx - 2]["text"] if active_idx >= 2 else ""
            prev = lyrics_lines[active_idx - 1]["text"] if active_idx >= 1 else ""
            current = lyrics_lines[active_idx]["text"] if active_idx >= 0 else "..."
            next_line = lyrics_lines[active_idx + 1]["text"] if active_idx >= 0 and active_idx + 1 < len(lyrics_lines) else ""
            next_next = lyrics_lines[active_idx + 2]["text"] if active_idx >= 0 and active_idx + 2 < len(lyrics_lines) else ""
            
            # Resolve album art to a local file path
            art_path = self.get_album_art_path(player.get("art_url", ""))
            
            state = {
                "status": player["status"],
                "title": title,
                "artist": artist,
                "prev_prev": prev_prev,
                "prev": prev,
                "current": current,
                "next": next_line,
                "next_next": next_next,
                "art_path": art_path
            }
            
            # Save state
            tmp_file = CURRENT_STATE_FILE.with_suffix('.tmp')
            with open(tmp_file, "w", encoding="utf-8") as f:
                json.dump(state, f)
            tmp_file.replace(CURRENT_STATE_FILE)
            
            # Update more frequently if playing to maintain tight sync
            if player["status"] == "Playing":
                time.sleep(0.3)  # Reduce polling frequency to prevent massive OS subprocess leak
            else:
                time.sleep(1.0)

if __name__ == "__main__":
    daemon = SpotifyLyricsDaemon()
    daemon.run()