215 lines
8.5 KiB
Python
215 lines
8.5 KiB
Python
import os
|
|
import time
|
|
import shutil
|
|
from typing import List, Dict, Any, Set, Optional
|
|
from app.config import settings
|
|
from app.wave.collector import WaveCollector, _normalize
|
|
from app.wave.recommender import WaveRecommender
|
|
from app.wave.playlist import ensure_smart_playlist
|
|
from app.core.tagger import get_track_dest_path
|
|
from app.core.lyrics import fetch_and_save_lyrics
|
|
from app.providers.yandex import YandexMusicProvider
|
|
from app.providers.universal import UniversalProvider
|
|
|
|
class WaveRotator:
|
|
def __init__(
|
|
self,
|
|
target_capacity: int = 25,
|
|
expire_days: int = 7,
|
|
subpath: str = "AI Recommendations/Моя Волна"
|
|
):
|
|
self.target_capacity = target_capacity
|
|
self.expire_days = expire_days
|
|
self.subpath = subpath
|
|
self.wave_dir = os.path.join(settings.MUSIC_DIR, subpath)
|
|
os.makedirs(self.wave_dir, exist_ok=True)
|
|
|
|
self.collector = WaveCollector()
|
|
self.yandex_provider = YandexMusicProvider()
|
|
self.universal_provider = UniversalProvider()
|
|
self.recommender = WaveRecommender(self.yandex_provider)
|
|
|
|
def get_current_buffer(self) -> List[Dict[str, Any]]:
|
|
"""List all audio files currently in the Wave rotation buffer."""
|
|
items: List[Dict[str, Any]] = []
|
|
if not os.path.exists(self.wave_dir):
|
|
return items
|
|
|
|
now = time.time()
|
|
for f in os.listdir(self.wave_dir):
|
|
if not f.lower().endswith(".mp3"):
|
|
continue
|
|
fpath = os.path.join(self.wave_dir, f)
|
|
stat = os.stat(fpath)
|
|
size_mb = round(stat.st_size / (1024 * 1024), 2)
|
|
age_days = round((now - stat.st_mtime) / 86400, 1)
|
|
|
|
# Derive artist and title from filename
|
|
name_no_ext = os.path.splitext(f)[0]
|
|
artist = "Unknown"
|
|
title = name_no_ext
|
|
if " - " in name_no_ext:
|
|
parts = name_no_ext.split(" - ", 1)
|
|
artist = parts[0].strip()
|
|
title = parts[1].strip()
|
|
|
|
items.append({
|
|
"filename": f,
|
|
"path": fpath,
|
|
"artist": artist,
|
|
"title": title,
|
|
"size_mb": size_mb,
|
|
"age_days": age_days,
|
|
"mtime": stat.st_mtime
|
|
})
|
|
|
|
items.sort(key=lambda x: x["mtime"], reverse=True)
|
|
return items
|
|
|
|
def get_wave_track_keys(self) -> Set[str]:
|
|
"""Normalized keys ('artist|title') of songs currently in Wave."""
|
|
keys = set()
|
|
for item in self.get_current_buffer():
|
|
art = _normalize(item["artist"])
|
|
tit = _normalize(item["title"])
|
|
if art and tit:
|
|
keys.add(f"{art}|{tit}")
|
|
return keys
|
|
|
|
def promote_starred_tracks(self) -> List[Dict[str, str]]:
|
|
"""Promote Wave tracks that user has starred in Navidrome to permanent library."""
|
|
promoted: List[Dict[str, str]] = []
|
|
starred_in_wave = self.collector.get_starred_wave_tracks(self.subpath)
|
|
|
|
for item in starred_in_wave:
|
|
src_path = os.path.join(settings.MUSIC_DIR, item["path"])
|
|
if not os.path.exists(src_path):
|
|
continue
|
|
|
|
# Target location in permanent library: /music/Artist/Album/Title.mp3
|
|
dest_path = get_track_dest_path(
|
|
settings.MUSIC_DIR,
|
|
artist=item["artist"],
|
|
album=item["album"] or "Singles",
|
|
track_num=None,
|
|
title=item["title"],
|
|
ext="mp3"
|
|
)
|
|
|
|
try:
|
|
os.makedirs(os.path.dirname(dest_path), exist_ok=True)
|
|
shutil.move(src_path, dest_path)
|
|
|
|
# Move .lrc if exists
|
|
src_lrc = os.path.splitext(src_path)[0] + ".lrc"
|
|
if os.path.exists(src_lrc):
|
|
dest_lrc = os.path.splitext(dest_path)[0] + ".lrc"
|
|
shutil.move(src_lrc, dest_lrc)
|
|
|
|
print(f"[WaveRotator] ⭐ Promoted starred track: {item['artist']} - {item['title']} -> {dest_path}")
|
|
promoted.append({"artist": item["artist"], "title": item["title"], "dest": dest_path})
|
|
except Exception as e:
|
|
print(f"[WaveRotator] Failed to promote {item['title']}: {e}")
|
|
|
|
return promoted
|
|
|
|
def purge_expired_tracks(self) -> List[Dict[str, str]]:
|
|
"""Remove tracks older than expire_days that were not starred."""
|
|
purged: List[Dict[str, str]] = []
|
|
now = time.time()
|
|
max_age_seconds = self.expire_days * 86400
|
|
|
|
for item in self.get_current_buffer():
|
|
age_sec = now - item["mtime"]
|
|
if age_sec > max_age_seconds:
|
|
try:
|
|
os.remove(item["path"])
|
|
lrc_path = os.path.splitext(item["path"])[0] + ".lrc"
|
|
if os.path.exists(lrc_path):
|
|
os.remove(lrc_path)
|
|
print(f"[WaveRotator] 🗑️ Purged expired wave track: {item['artist']} - {item['title']} (age: {item['age_days']} days)")
|
|
purged.append({"artist": item["artist"], "title": item["title"]})
|
|
except Exception as e:
|
|
print(f"[WaveRotator] Failed to remove {item['filename']}: {e}")
|
|
|
|
return purged
|
|
|
|
async def fill_buffer(self) -> Dict[str, Any]:
|
|
"""Run rotation (promotion + purge) and download candidates up to target capacity."""
|
|
promoted = self.promote_starred_tracks()
|
|
purged = self.purge_expired_tracks()
|
|
|
|
current_items = self.get_current_buffer()
|
|
current_count = len(current_items)
|
|
needed = max(0, self.target_capacity - current_count)
|
|
|
|
added: List[Dict[str, str]] = []
|
|
|
|
if needed > 0:
|
|
print(f"[WaveRotator] Wave buffer at {current_count}/{self.target_capacity}. Sourcing {needed} new tracks...")
|
|
profile = self.collector.get_anchor_profile()
|
|
existing_keys = self.collector.get_existing_track_keys()
|
|
wave_keys = self.get_wave_track_keys()
|
|
|
|
candidates = self.recommender.get_candidate_tracks(
|
|
anchor_profile=profile,
|
|
existing_keys=existing_keys,
|
|
wave_keys=wave_keys,
|
|
target_count=needed
|
|
)
|
|
|
|
for cand in candidates:
|
|
filename = f"{cand.artist} - {cand.title}.mp3"
|
|
# Clean unsafe filesystem chars
|
|
filename = "".join(c for c in filename if c not in '<>:"/\\|?*')
|
|
dest_path = os.path.join(self.wave_dir, filename)
|
|
|
|
try:
|
|
success = await self.yandex_provider.download_track(cand, dest_path)
|
|
if not success:
|
|
success = await self.universal_provider.download_track(cand, dest_path)
|
|
|
|
if success:
|
|
# Fetch and save .lrc lyrics
|
|
try:
|
|
fetch_and_save_lyrics(dest_path, f"{cand.artist} - {cand.title}")
|
|
except Exception as e:
|
|
print(f"[WaveRotator] Warning: lyrics fetch error for {cand.title}: {e}")
|
|
|
|
added.append({"artist": cand.artist, "title": cand.title})
|
|
print(f"[WaveRotator] 🌊 Added to Wave: {cand.artist} - {cand.title}")
|
|
except Exception as e:
|
|
print(f"[WaveRotator] Error downloading candidate {cand.title}: {e}")
|
|
|
|
# Ensure Smart Playlist exists
|
|
ensure_smart_playlist()
|
|
|
|
return {
|
|
"target_capacity": self.target_capacity,
|
|
"total_in_buffer": len(self.get_current_buffer()),
|
|
"added_count": len(added),
|
|
"added_tracks": added,
|
|
"promoted_count": len(promoted),
|
|
"promoted_tracks": promoted,
|
|
"purged_count": len(purged),
|
|
"purged_tracks": purged
|
|
}
|
|
|
|
def get_status(self) -> Dict[str, Any]:
|
|
"""Get summary status of Smart Wave."""
|
|
buffer_tracks = self.get_current_buffer()
|
|
total_size_mb = round(sum(t["size_mb"] for t in buffer_tracks), 2)
|
|
profile = self.collector.get_anchor_profile()
|
|
anchor_artists = [a["artist"] for a in profile.get("top_artists", [])[:6]]
|
|
|
|
return {
|
|
"capacity": self.target_capacity,
|
|
"count": len(buffer_tracks),
|
|
"total_size_mb": total_size_mb,
|
|
"expire_days": self.expire_days,
|
|
"anchor_artists": anchor_artists,
|
|
"tracks": buffer_tracks[:15]
|
|
}
|
|
|
|
wave_rotator = WaveRotator()
|