mirror of
https://github.com/wahyd4/you-music.git
synced 2026-08-08 20:59:47 +10:00
296 lines
10 KiB
Python
296 lines
10 KiB
Python
from fastapi import APIRouter, Depends
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, func
|
|
from typing import List, Optional
|
|
import aiohttp
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
|
|
from app.db.session import get_db
|
|
from app.models.models import Music
|
|
from app.schemas.schemas import Music as MusicSchema
|
|
from app.core.config import settings
|
|
from pydantic import BaseModel
|
|
|
|
router = APIRouter()
|
|
|
|
# Cache directory for artist info
|
|
CACHE_DIR = os.path.join(settings.BASE_DIR, "data", "cache", "artists")
|
|
IMAGE_CACHE_DIR = os.path.join(settings.BASE_DIR, "data", "cache", "artist_images")
|
|
CACHE_DURATION_DAYS = 30 # Cache for 30 days
|
|
|
|
|
|
class Artist(BaseModel):
|
|
name: str
|
|
song_count: int
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class ArtistInfo(BaseModel):
|
|
name: str
|
|
image: Optional[str] = None
|
|
bio: Optional[str] = None
|
|
listeners: Optional[int] = None
|
|
playcount: Optional[int] = None
|
|
|
|
|
|
def get_cache_path(artist_name: str) -> str:
|
|
"""Get cache file path for an artist"""
|
|
os.makedirs(CACHE_DIR, exist_ok=True)
|
|
# Use a safe filename
|
|
safe_name = "".join(c if c.isalnum() or c in (' ', '-', '_') else '_' for c in artist_name)
|
|
return os.path.join(CACHE_DIR, f"{safe_name}.json")
|
|
|
|
|
|
def get_cached_artist_info(artist_name: str) -> Optional[ArtistInfo]:
|
|
"""Get cached artist info if it exists and is not expired"""
|
|
cache_path = get_cache_path(artist_name)
|
|
|
|
if not os.path.exists(cache_path):
|
|
return None
|
|
|
|
try:
|
|
# Check if cache is expired
|
|
file_time = datetime.fromtimestamp(os.path.getmtime(cache_path))
|
|
if datetime.now() - file_time > timedelta(days=CACHE_DURATION_DAYS):
|
|
# Cache expired, remove it
|
|
os.remove(cache_path)
|
|
return None
|
|
|
|
# Read and return cached data
|
|
with open(cache_path, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
return ArtistInfo(**data)
|
|
except Exception as e:
|
|
print(f"Error reading cache: {e}")
|
|
return None
|
|
|
|
|
|
def save_artist_info_to_cache(artist_name: str, info: ArtistInfo):
|
|
"""Save artist info to cache"""
|
|
cache_path = get_cache_path(artist_name)
|
|
|
|
try:
|
|
with open(cache_path, 'w', encoding='utf-8') as f:
|
|
json.dump(info.model_dump(), f, ensure_ascii=False, indent=2)
|
|
except Exception as e:
|
|
print(f"Error writing cache: {e}")
|
|
|
|
|
|
async def download_and_cache_artist_image(artist_name: str, image_url: str) -> Optional[str]:
|
|
"""Download artist image and cache it locally"""
|
|
os.makedirs(IMAGE_CACHE_DIR, exist_ok=True)
|
|
|
|
# Create safe filename
|
|
safe_name = "".join(c if c.isalnum() or c in (' ', '-', '_') else '_' for c in artist_name)
|
|
|
|
# Determine file extension from URL
|
|
ext = '.jpg'
|
|
if '.png' in image_url.lower():
|
|
ext = '.png'
|
|
elif '.webp' in image_url.lower():
|
|
ext = '.webp'
|
|
|
|
cache_path = os.path.join(IMAGE_CACHE_DIR, f"{safe_name}{ext}")
|
|
|
|
# Return cached path if exists and not expired
|
|
if os.path.exists(cache_path):
|
|
file_time = datetime.fromtimestamp(os.path.getmtime(cache_path))
|
|
if datetime.now() - file_time <= timedelta(days=CACHE_DURATION_DAYS):
|
|
return f"cache/artist_images/{safe_name}{ext}"
|
|
|
|
# Download image
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(image_url) as response:
|
|
if response.status == 200:
|
|
image_data = await response.read()
|
|
with open(cache_path, 'wb') as f:
|
|
f.write(image_data)
|
|
return f"cache/artist_images/{safe_name}{ext}"
|
|
except Exception as e:
|
|
print(f"Error downloading artist image: {e}")
|
|
|
|
return None
|
|
|
|
|
|
async def get_artist_info_from_apis(artist_name: str) -> Optional[ArtistInfo]:
|
|
"""Fetch artist info from Deezer API (primary) and MusicBrainz (fallback)"""
|
|
try:
|
|
# Try Deezer API first (no key needed, has good artist images)
|
|
deezer_url = "https://api.deezer.com/search/artist"
|
|
params = {"q": artist_name, "limit": 1}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(deezer_url, params=params) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
|
|
if data.get("data") and len(data["data"]) > 0:
|
|
artist = data["data"][0]
|
|
|
|
# Deezer provides different image sizes
|
|
image_url = artist.get("picture_xl") or artist.get("picture_big") or artist.get("picture_medium")
|
|
|
|
return ArtistInfo(
|
|
name=artist.get("name", artist_name),
|
|
image=image_url,
|
|
bio=None,
|
|
listeners=artist.get("nb_fan"), # Number of fans
|
|
playcount=None
|
|
)
|
|
|
|
# Fallback to MusicBrainz if Deezer fails
|
|
search_url = "https://musicbrainz.org/ws/2/artist/"
|
|
params = {
|
|
"query": artist_name,
|
|
"fmt": "json",
|
|
"limit": 1
|
|
}
|
|
|
|
async with session.get(search_url, params=params) as response:
|
|
if response.status != 200:
|
|
return None
|
|
|
|
data = await response.json()
|
|
|
|
if not data.get("artists"):
|
|
return None
|
|
|
|
artist = data["artists"][0]
|
|
artist_real_name = artist.get("name", artist_name)
|
|
|
|
# Get disambiguation and type as "bio"
|
|
bio_parts = []
|
|
if artist.get("type"):
|
|
bio_parts.append(f"Type: {artist['type']}")
|
|
if artist.get("disambiguation"):
|
|
bio_parts.append(artist["disambiguation"])
|
|
if artist.get("country"):
|
|
bio_parts.append(f"Country: {artist['country']}")
|
|
|
|
bio = " • ".join(bio_parts) if bio_parts else None
|
|
|
|
return ArtistInfo(
|
|
name=artist_real_name,
|
|
image=None, # MusicBrainz doesn't have images
|
|
bio=bio,
|
|
listeners=None,
|
|
playcount=None
|
|
)
|
|
|
|
return None
|
|
except Exception as e:
|
|
print(f"Error fetching artist data: {e}")
|
|
return None
|
|
|
|
|
|
@router.get("/", response_model=List[Artist])
|
|
async def get_artists(db: AsyncSession = Depends(get_db)):
|
|
"""Get all artists with song counts"""
|
|
result = await db.execute(
|
|
select(
|
|
Music.artist,
|
|
func.count(Music.id).label('song_count')
|
|
)
|
|
.where(Music.artist.isnot(None))
|
|
.where(Music.artist != "")
|
|
.where(Music.artist != "Unknown")
|
|
.group_by(Music.artist)
|
|
.order_by(func.count(Music.id).desc())
|
|
)
|
|
|
|
artists = []
|
|
for row in result:
|
|
artists.append(Artist(name=row[0], song_count=row[1]))
|
|
|
|
return artists
|
|
|
|
|
|
@router.get("/info", response_model=ArtistInfo)
|
|
async def get_artist_info(artist_name: str, db: AsyncSession = Depends(get_db)):
|
|
"""Get artist information from Deezer and MusicBrainz (with 30-day cache)
|
|
|
|
Artist Avatar Priority:
|
|
1. Cache (if exists and not expired)
|
|
2. Online API (Deezer/MusicBrainz)
|
|
3. Song thumbnail fallback (if artist has songs with thumbnails)
|
|
4. None (frontend shows default icon)
|
|
"""
|
|
# 1. Try to get from cache first
|
|
cached_info = get_cached_artist_info(artist_name)
|
|
if cached_info:
|
|
return cached_info
|
|
|
|
# 2. Fetch from online APIs
|
|
info = await get_artist_info_from_apis(artist_name)
|
|
if not info:
|
|
# Return minimal info if APIs fail
|
|
info = ArtistInfo(name=artist_name)
|
|
|
|
# Download and cache the image if available from API
|
|
if info.image and info.image.startswith('http'):
|
|
cached_image_path = await download_and_cache_artist_image(artist_name, info.image)
|
|
if cached_image_path:
|
|
# Update info with local cached image path
|
|
info.image = cached_image_path
|
|
|
|
# 3. If no image found from API, try to use artist's song thumbnail as fallback
|
|
if not info.image:
|
|
result = await db.execute(
|
|
select(Music)
|
|
.where(Music.artist == artist_name)
|
|
.where(Music.thumbnail.isnot(None))
|
|
.where(Music.thumbnail != "")
|
|
.limit(1)
|
|
)
|
|
song_with_thumbnail = result.scalar_one_or_none()
|
|
if song_with_thumbnail and song_with_thumbnail.thumbnail:
|
|
info.image = song_with_thumbnail.thumbnail
|
|
|
|
# 4. Save to cache (even if no image - will show default icon on frontend)
|
|
save_artist_info_to_cache(artist_name, info)
|
|
|
|
return info
|
|
|
|
|
|
@router.delete("/cache", response_model=dict)
|
|
async def clear_artist_cache(artist_name: str):
|
|
"""Clear cached artist info (forces refresh on next request)"""
|
|
cache_path = get_cache_path(artist_name)
|
|
|
|
if os.path.exists(cache_path):
|
|
os.remove(cache_path)
|
|
return {"message": f"Cache cleared for {artist_name}"}
|
|
|
|
return {"message": "No cache found"}
|
|
|
|
|
|
@router.delete("/cache/all")
|
|
async def clear_all_artist_cache():
|
|
"""Clear all cached artist info"""
|
|
if os.path.exists(CACHE_DIR):
|
|
count = 0
|
|
for file in Path(CACHE_DIR).glob("*.json"):
|
|
file.unlink()
|
|
count += 1
|
|
return {"message": f"Cleared {count} cached artists"}
|
|
|
|
return {"message": "No cache found"}
|
|
|
|
|
|
@router.get("/{artist_name}", response_model=List[MusicSchema])
|
|
async def get_artist_songs(artist_name: str, db: AsyncSession = Depends(get_db)):
|
|
"""Get all songs by a specific artist"""
|
|
result = await db.execute(
|
|
select(Music)
|
|
.where(Music.artist == artist_name)
|
|
.order_by(Music.created_at.desc())
|
|
)
|
|
songs = result.scalars().all()
|
|
return songs
|