Implement basic recommendation system

app/api/v1/api.py

@@ -1,7 +1,7 @@
 from fastapi import APIRouter
 
 # Import individual routers here
-from app.api.v1.endpoints import users, auth, songs, albums, artists, playlists, streaming, upload
+from app.api.v1.endpoints import users, auth, songs, albums, artists, playlists, streaming, upload, recommendations
 
 api_router = APIRouter()
 
@@ -13,4 +13,5 @@ api_router.include_router(albums.router, prefix="/albums", tags=["albums"])
 api_router.include_router(artists.router, prefix="/artists", tags=["artists"])
 api_router.include_router(playlists.router, prefix="/playlists", tags=["playlists"])
 api_router.include_router(streaming.router, prefix="/stream", tags=["streaming"])
-api_router.include_router(upload.router, prefix="/upload", tags=["upload"])
+api_router.include_router(upload.router, prefix="/upload", tags=["upload"])
+api_router.include_router(recommendations.router, prefix="/recommendations", tags=["recommendations"])

app/api/v1/endpoints/recommendations.py

@@ -0,0 +1,82 @@
+from typing import Any, List
+
+from fastapi import APIRouter, Depends, HTTPException, status, Query
+from sqlalchemy.orm import Session
+
+from app.api.deps import get_db, get_current_active_user
+from app.models.user import User
+from app.schemas.song import Song
+from app.schemas.artist import Artist
+from app.services.song import get_song
+from app.services.artist import get_artist
+from app.services.recommendation import (
+    get_similar_songs,
+    get_recommended_songs_for_user,
+    get_popular_songs,
+    get_artist_recommendations,
+)
+
+router = APIRouter()
+
+
+@router.get("/similar-songs/{song_id}", response_model=List[Song])
+def similar_songs(
+    song_id: int,
+    limit: int = Query(10, ge=1, le=50),
+    db: Session = Depends(get_db),
+) -> Any:
+    """
+    Get similar songs to the given song ID.
+    """
+    # Check if song exists
+    song = get_song(db, song_id=song_id)
+    if not song:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Song not found",
+        )
+    
+    return get_similar_songs(db, song_id=song_id, limit=limit)
+
+
+@router.get("/for-user", response_model=List[Song])
+def recommendations_for_user(
+    limit: int = Query(10, ge=1, le=50),
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_active_user),
+) -> Any:
+    """
+    Get personalized song recommendations for the current user.
+    """
+    return get_recommended_songs_for_user(db, user_id=current_user.id, limit=limit)
+
+
+@router.get("/popular", response_model=List[Song])
+def popular_songs(
+    limit: int = Query(10, ge=1, le=50),
+    db: Session = Depends(get_db),
+) -> Any:
+    """
+    Get popular songs based on how many playlists they appear in.
+    """
+    return get_popular_songs(db, limit=limit)
+
+
+@router.get("/similar-artists/{artist_id}", response_model=List[Artist])
+def similar_artists(
+    artist_id: int,
+    limit: int = Query(5, ge=1, le=20),
+    db: Session = Depends(get_db),
+) -> Any:
+    """
+    Get similar artists to the given artist ID.
+    """
+    # Check if artist exists
+    artist = get_artist(db, artist_id=artist_id)
+    if not artist:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Artist not found",
+        )
+    
+    return get_artist_recommendations(db, artist_id=artist_id, limit=limit)

app/services/recommendation.py

@@ -0,0 +1,103 @@
+from typing import List
+from sqlalchemy.orm import Session
+from sqlalchemy import func, desc
+
+from app.models.song import Song, song_playlist
+from app.models.playlist import Playlist
+from app.models.artist import Artist
+
+
+def get_similar_songs(db: Session, song_id: int, limit: int = 10) -> List[Song]:
+    """
+    Get similar songs based on the same artist and album.
+    """
+    # Get the current song
+    current_song = db.query(Song).filter(Song.id == song_id).first()
+    if not current_song:
+        return []
+    
+    # Find songs from the same artist and album (if applicable)
+    query = db.query(Song).filter(Song.id != song_id)
+    
+    if current_song.album_id:
+        # Prioritize songs from same album
+        query = query.filter(Song.album_id == current_song.album_id)
+    else:
+        # Otherwise, find songs from the same artist
+        query = query.filter(Song.artist_id == current_song.artist_id)
+    
+    return query.limit(limit).all()
+
+
+def get_recommended_songs_for_user(db: Session, user_id: int, limit: int = 10) -> List[Song]:
+    """
+    Get song recommendations for a user based on their playlists.
+    """
+    # First get all songs in user's playlists
+    user_playlist_songs = (
+        db.query(Song.id)
+        .join(song_playlist)
+        .join(Playlist)
+        .filter(Playlist.user_id == user_id)
+        .subquery()
+    )
+    
+    # Get artists that the user listens to
+    user_artists = (
+        db.query(Artist.id)
+        .join(Song, Song.artist_id == Artist.id)
+        .join(user_playlist_songs, user_playlist_songs.c.id == Song.id)
+        .distinct()
+        .subquery()
+    )
+    
+    # Recommend songs from the same artists that aren't in the user's playlists
+    recommended = (
+        db.query(Song)
+        .join(Artist, Song.artist_id == Artist.id)
+        .filter(
+            Song.id.notin_(user_playlist_songs),
+            Artist.id.in_(user_artists)
+        )
+        .order_by(func.random())
+        .limit(limit)
+        .all()
+    )
+    
+    return recommended
+
+
+def get_popular_songs(db: Session, limit: int = 10) -> List[Song]:
+    """
+    Get popular songs based on how many playlists they appear in.
+    """
+    # Count the number of playlists each song is in
+    song_counts = (
+        db.query(
+            Song,
+            func.count(song_playlist.c.playlist_id).label("playlist_count")
+        )
+        .join(song_playlist)
+        .group_by(Song.id)
+        .order_by(desc("playlist_count"))
+        .limit(limit)
+        .all()
+    )
+    
+    # Extract just the songs
+    return [song for song, _ in song_counts]
+
+
+def get_artist_recommendations(db: Session, artist_id: int, limit: int = 5) -> List[Artist]:
+    """
+    Get similar artists based on the given artist.
+    """
+    # This is a very simple implementation that just returns random artists
+    # In a real system, this would use more sophisticated algorithms
+    return (
+        db.query(Artist)
+        .filter(Artist.id != artist_id)
+        .order_by(func.random())
+        .limit(limit)
+        .all()
+    )