Add YouTube Music playlist import support

- Add ytmusicapi dependency for fetching public YouTube Music playlists
- Create youtube_music service with playlist fetching and track search
- Add /api/youtube-music/import and /api/youtube-music/search endpoints
- Add importYouTubePlaylist and searchYouTubeMusic API client functions
- Update Playlists page with YouTube Music import button and URL input modal
This commit is contained in:
root
2026-03-30 21:33:27 -05:00
parent cd88ed2983
commit 58c17498be
6 changed files with 339 additions and 18 deletions

View File

@@ -0,0 +1,110 @@
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import settings
from app.core.database import get_db
from app.core.security import get_current_user
from app.models.user import User
from app.models.playlist import Playlist
from app.models.track import Track
from app.services.youtube_music import get_playlist_tracks, search_track
from app.services.recommender import build_taste_profile
from app.schemas.playlist import PlaylistDetailResponse
router = APIRouter(prefix="/youtube-music", tags=["youtube-music"])
class ImportYouTubeRequest(BaseModel):
url: str
class SearchYouTubeRequest(BaseModel):
query: str
class YouTubeTrackResult(BaseModel):
title: str
artist: str
album: str | None = None
youtube_id: str | None = None
image_url: str | None = None
@router.post("/import", response_model=PlaylistDetailResponse)
async def import_youtube_playlist(
data: ImportYouTubeRequest,
user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
# Free tier limit
if not user.is_pro:
result = await db.execute(
select(Playlist).where(Playlist.user_id == user.id)
)
existing = list(result.scalars().all())
if len(existing) >= settings.FREE_MAX_PLAYLISTS:
raise HTTPException(
status_code=403,
detail="Free tier limited to 1 playlist. Upgrade to Pro for unlimited.",
)
# Fetch tracks from YouTube Music
try:
playlist_name, playlist_image, raw_tracks = get_playlist_tracks(data.url)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception:
raise HTTPException(status_code=400, detail="Failed to fetch playlist from YouTube Music. Make sure the URL is valid and the playlist is public.")
if not raw_tracks:
raise HTTPException(status_code=400, detail="Playlist is empty or could not be read.")
# Create playlist
playlist = Playlist(
user_id=user.id,
name=playlist_name,
platform_source="youtube_music",
external_id=data.url,
track_count=len(raw_tracks),
)
db.add(playlist)
await db.flush()
# Create tracks (no audio features available from YouTube Music)
tracks = []
for rt in raw_tracks:
track = Track(
playlist_id=playlist.id,
title=rt["title"],
artist=rt["artist"],
album=rt.get("album"),
image_url=rt.get("image_url"),
)
db.add(track)
tracks.append(track)
await db.flush()
# Build taste profile (without audio features, will be limited)
playlist.taste_profile = build_taste_profile(tracks)
playlist.tracks = tracks
return playlist
@router.post("/search", response_model=list[YouTubeTrackResult])
async def search_youtube_music(
data: SearchYouTubeRequest,
user: User = Depends(get_current_user),
):
if not data.query.strip():
raise HTTPException(status_code=400, detail="Query cannot be empty")
try:
results = search_track(data.query.strip())
except Exception:
raise HTTPException(status_code=500, detail="Failed to search YouTube Music")
return results