7abec6de7c
- Playlist Generator: describe a vibe, get a 15-30 song playlist, save or copy as text - Artist Deep Dive: click any artist name for influences, best album, hidden gems, similar artists - Music Timeline: visual decade breakdown of your taste with AI insight - Nav updates: Create Playlist, Timeline links
501 lines
16 KiB
Python
501 lines
16 KiB
Python
import hashlib
|
|
import json
|
|
from urllib.parse import quote_plus
|
|
|
|
import anthropic
|
|
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.models.recommendation import Recommendation
|
|
from app.schemas.recommendation import RecommendationRequest, RecommendationResponse, RecommendationItem
|
|
from app.services.recommender import generate_recommendations, build_taste_profile
|
|
|
|
router = APIRouter(prefix="/recommendations", tags=["recommendations"])
|
|
|
|
|
|
@router.post("/generate", response_model=RecommendationResponse)
|
|
async def generate(
|
|
data: RecommendationRequest,
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
if not data.playlist_id and not data.query:
|
|
raise HTTPException(status_code=400, detail="Provide a playlist_id or query")
|
|
|
|
recs, remaining = await generate_recommendations(
|
|
db, user, playlist_id=data.playlist_id, query=data.query, bandcamp_mode=data.bandcamp_mode,
|
|
mode=data.mode, adventurousness=data.adventurousness, exclude=data.exclude, count=data.count,
|
|
mood_energy=data.mood_energy, mood_valence=data.mood_valence,
|
|
)
|
|
|
|
if not recs and remaining == 0:
|
|
raise HTTPException(status_code=429, detail="Weekly recommendation limit reached. Upgrade to Premium for unlimited.")
|
|
|
|
return RecommendationResponse(
|
|
recommendations=[RecommendationItem.model_validate(r) for r in recs],
|
|
remaining_this_week=remaining,
|
|
)
|
|
|
|
|
|
@router.post("/surprise", response_model=RecommendationResponse)
|
|
async def surprise(
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
recs, remaining = await generate_recommendations(
|
|
db, user, query=None, mode="surprise", count=5
|
|
)
|
|
|
|
if not recs and remaining == 0:
|
|
raise HTTPException(status_code=429, detail="Weekly recommendation limit reached. Upgrade to Premium for unlimited.")
|
|
|
|
return RecommendationResponse(
|
|
recommendations=[RecommendationItem.model_validate(r) for r in recs],
|
|
remaining_this_week=remaining,
|
|
)
|
|
|
|
|
|
@router.get("/history", response_model=list[RecommendationItem])
|
|
async def history(
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(Recommendation)
|
|
.where(Recommendation.user_id == user.id)
|
|
.order_by(Recommendation.created_at.desc())
|
|
.limit(50)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.get("/saved", response_model=list[RecommendationItem])
|
|
async def saved(
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(Recommendation)
|
|
.where(Recommendation.user_id == user.id, Recommendation.saved == True)
|
|
.order_by(Recommendation.created_at.desc())
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
class AnalyzeRequest(BaseModel):
|
|
artist: str
|
|
title: str
|
|
|
|
|
|
class AnalyzeResponse(BaseModel):
|
|
analysis: str
|
|
qualities: list[str]
|
|
recommendations: list[RecommendationItem]
|
|
|
|
|
|
@router.post("/analyze", response_model=AnalyzeResponse)
|
|
async def analyze_song(
|
|
data: AnalyzeRequest,
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
prompt = f"""You are Vynl, a music analysis expert. The user wants to understand why they love this song:
|
|
|
|
Artist: {data.artist}
|
|
Title: {data.title}
|
|
|
|
Respond with a JSON object:
|
|
{{
|
|
"analysis": "A warm, insightful 3-4 sentence explanation of what makes this song special and why someone would be drawn to it. Reference specific sonic qualities, production choices, lyrical themes, and emotional resonance.",
|
|
"qualities": ["quality1", "quality2", ...],
|
|
"recommendations": [
|
|
{{"title": "...", "artist": "...", "album": "...", "reason": "...", "score": 0.9}}
|
|
]
|
|
}}
|
|
|
|
For "qualities", list 4-6 specific musical qualities (e.g., "warm analog production", "introspective lyrics about loss", "driving bass line with syncopated rhythm").
|
|
For "recommendations", suggest 5 songs that share these same qualities. Only suggest songs that actually exist.
|
|
Return ONLY the JSON object."""
|
|
|
|
client = anthropic.Anthropic(api_key=settings.ANTHROPIC_API_KEY)
|
|
message = client.messages.create(
|
|
model="claude-haiku-4-5-20251001",
|
|
max_tokens=2000,
|
|
messages=[{"role": "user", "content": prompt}],
|
|
)
|
|
|
|
response_text = message.content[0].text.strip()
|
|
if response_text.startswith("```"):
|
|
response_text = response_text.split("\n", 1)[1]
|
|
response_text = response_text.rsplit("```", 1)[0]
|
|
|
|
try:
|
|
parsed = json.loads(response_text)
|
|
except json.JSONDecodeError:
|
|
raise HTTPException(status_code=500, detail="Failed to parse AI response")
|
|
|
|
analysis = parsed.get("analysis", "")
|
|
qualities = parsed.get("qualities", [])
|
|
recs_data = parsed.get("recommendations", [])
|
|
|
|
recommendations = []
|
|
for rec in recs_data[:5]:
|
|
artist = rec.get("artist", "Unknown")
|
|
title = rec.get("title", "Unknown")
|
|
youtube_url = f"https://music.youtube.com/search?q={quote_plus(f'{artist} {title}')}"
|
|
|
|
r = Recommendation(
|
|
user_id=user.id,
|
|
title=title,
|
|
artist=artist,
|
|
album=rec.get("album"),
|
|
reason=rec.get("reason", ""),
|
|
score=rec.get("score"),
|
|
query=f"analyze: {data.artist} - {data.title}",
|
|
youtube_url=youtube_url,
|
|
)
|
|
db.add(r)
|
|
recommendations.append(r)
|
|
|
|
await db.flush()
|
|
|
|
return AnalyzeResponse(
|
|
analysis=analysis,
|
|
qualities=qualities,
|
|
recommendations=[RecommendationItem.model_validate(r) for r in recommendations],
|
|
)
|
|
|
|
|
|
class ArtistDeepDiveRequest(BaseModel):
|
|
artist: str
|
|
|
|
|
|
class ArtistDeepDiveResponse(BaseModel):
|
|
artist: str
|
|
summary: str
|
|
why_they_matter: str
|
|
influences: list[str]
|
|
influenced: list[str]
|
|
start_with: str
|
|
start_with_reason: str
|
|
deep_cut: str
|
|
similar_artists: list[str]
|
|
genres: list[str]
|
|
|
|
|
|
@router.post("/artist-dive", response_model=ArtistDeepDiveResponse)
|
|
async def artist_deep_dive(
|
|
data: ArtistDeepDiveRequest,
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
prompt = f"""You are Vynl, a music expert. Give a deep dive on this artist:
|
|
|
|
Artist: {data.artist}
|
|
|
|
Respond with a JSON object:
|
|
{{
|
|
"artist": "{data.artist}",
|
|
"summary": "2-3 sentences about who they are and their sound",
|
|
"why_they_matter": "Their cultural significance and impact on music",
|
|
"influences": ["artist1", "artist2", "artist3"],
|
|
"influenced": ["artist1", "artist2", "artist3"],
|
|
"start_with": "Album Name",
|
|
"start_with_reason": "Why this is the best entry point",
|
|
"deep_cut": "A hidden gem track title",
|
|
"similar_artists": ["artist1", "artist2", "artist3", "artist4", "artist5"],
|
|
"genres": ["genre1", "genre2"]
|
|
}}
|
|
Return ONLY the JSON object."""
|
|
|
|
client = anthropic.Anthropic(api_key=settings.ANTHROPIC_API_KEY)
|
|
message = client.messages.create(
|
|
model="claude-haiku-4-5-20251001",
|
|
max_tokens=2000,
|
|
messages=[{"role": "user", "content": prompt}],
|
|
)
|
|
|
|
response_text = message.content[0].text.strip()
|
|
if response_text.startswith("```"):
|
|
response_text = response_text.split("\n", 1)[1]
|
|
response_text = response_text.rsplit("```", 1)[0]
|
|
|
|
try:
|
|
parsed = json.loads(response_text)
|
|
except json.JSONDecodeError:
|
|
raise HTTPException(status_code=500, detail="Failed to parse AI response")
|
|
|
|
return ArtistDeepDiveResponse(
|
|
artist=parsed.get("artist", data.artist),
|
|
summary=parsed.get("summary", ""),
|
|
why_they_matter=parsed.get("why_they_matter", ""),
|
|
influences=parsed.get("influences", []),
|
|
influenced=parsed.get("influenced", []),
|
|
start_with=parsed.get("start_with", ""),
|
|
start_with_reason=parsed.get("start_with_reason", ""),
|
|
deep_cut=parsed.get("deep_cut", ""),
|
|
similar_artists=parsed.get("similar_artists", []),
|
|
genres=parsed.get("genres", []),
|
|
)
|
|
|
|
|
|
class GeneratePlaylistRequest(BaseModel):
|
|
theme: str
|
|
count: int = 25
|
|
save: bool = False
|
|
|
|
|
|
class PlaylistTrack(BaseModel):
|
|
title: str
|
|
artist: str
|
|
album: str | None = None
|
|
reason: str
|
|
youtube_url: str | None = None
|
|
|
|
|
|
class GeneratedPlaylistResponse(BaseModel):
|
|
name: str
|
|
description: str
|
|
tracks: list[PlaylistTrack]
|
|
playlist_id: int | None = None
|
|
|
|
|
|
@router.post("/generate-playlist", response_model=GeneratedPlaylistResponse)
|
|
async def generate_playlist(
|
|
data: GeneratePlaylistRequest,
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
if not data.theme.strip():
|
|
raise HTTPException(status_code=400, detail="Theme is required")
|
|
if data.count < 5 or data.count > 50:
|
|
raise HTTPException(status_code=400, detail="Count must be between 5 and 50")
|
|
|
|
# Build taste context from user's playlists
|
|
taste_context = ""
|
|
result = await db.execute(
|
|
select(Playlist).where(Playlist.user_id == user.id)
|
|
)
|
|
playlists = list(result.scalars().all())
|
|
all_tracks = []
|
|
for p in playlists:
|
|
track_result = await db.execute(select(Track).where(Track.playlist_id == p.id))
|
|
all_tracks.extend(track_result.scalars().all())
|
|
if all_tracks:
|
|
profile = build_taste_profile(all_tracks)
|
|
taste_context = f"\n\nThe user's taste profile from {len(all_tracks)} tracks:\n{json.dumps(profile, indent=2)}\n\nUse this to personalize the playlist to their taste while staying true to the theme."
|
|
|
|
prompt = f"""You are Vynl, a playlist curator. Create a cohesive playlist for this theme:
|
|
|
|
Theme: {data.theme}
|
|
{taste_context}
|
|
|
|
Generate a playlist of exactly {data.count} songs. The playlist should flow naturally — songs should be ordered for a great listening experience, not random.
|
|
|
|
Respond with a JSON object:
|
|
{{
|
|
"name": "A creative playlist name",
|
|
"description": "A 1-2 sentence description of the playlist vibe",
|
|
"tracks": [
|
|
{{"title": "...", "artist": "...", "album": "...", "reason": "Why this fits the playlist"}}
|
|
]
|
|
}}
|
|
|
|
Only recommend real songs that actually exist. Do not invent song titles.
|
|
Return ONLY the JSON object."""
|
|
|
|
client = anthropic.Anthropic(api_key=settings.ANTHROPIC_API_KEY)
|
|
message = client.messages.create(
|
|
model="claude-haiku-4-5-20251001",
|
|
max_tokens=4000,
|
|
messages=[{"role": "user", "content": prompt}],
|
|
)
|
|
|
|
response_text = message.content[0].text.strip()
|
|
if response_text.startswith("```"):
|
|
response_text = response_text.split("\n", 1)[1]
|
|
response_text = response_text.rsplit("```", 1)[0]
|
|
|
|
try:
|
|
parsed = json.loads(response_text)
|
|
except json.JSONDecodeError:
|
|
raise HTTPException(status_code=500, detail="Failed to parse AI response")
|
|
|
|
playlist_name = parsed.get("name", f"Playlist: {data.theme}")
|
|
description = parsed.get("description", "")
|
|
tracks_data = parsed.get("tracks", [])
|
|
|
|
# Add YouTube Music search links
|
|
tracks = []
|
|
for t in tracks_data:
|
|
artist = t.get("artist", "Unknown")
|
|
title = t.get("title", "Unknown")
|
|
yt_url = f"https://music.youtube.com/search?q={quote_plus(f'{artist} {title}')}"
|
|
tracks.append(PlaylistTrack(
|
|
title=title,
|
|
artist=artist,
|
|
album=t.get("album"),
|
|
reason=t.get("reason", ""),
|
|
youtube_url=yt_url,
|
|
))
|
|
|
|
# Optionally save as a playlist in the DB
|
|
playlist_id = None
|
|
if data.save:
|
|
new_playlist = Playlist(
|
|
user_id=user.id,
|
|
name=playlist_name,
|
|
platform_source="generated",
|
|
track_count=len(tracks),
|
|
)
|
|
db.add(new_playlist)
|
|
await db.flush()
|
|
|
|
for t in tracks:
|
|
track_record = Track(
|
|
playlist_id=new_playlist.id,
|
|
title=t.title,
|
|
artist=t.artist,
|
|
album=t.album,
|
|
)
|
|
db.add(track_record)
|
|
|
|
await db.flush()
|
|
playlist_id = new_playlist.id
|
|
|
|
return GeneratedPlaylistResponse(
|
|
name=playlist_name,
|
|
description=description,
|
|
tracks=tracks,
|
|
playlist_id=playlist_id,
|
|
)
|
|
|
|
|
|
@router.post("/{rec_id}/save")
|
|
async def save_recommendation(
|
|
rec_id: int,
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(Recommendation).where(Recommendation.id == rec_id, Recommendation.user_id == user.id)
|
|
)
|
|
rec = result.scalar_one_or_none()
|
|
if not rec:
|
|
raise HTTPException(status_code=404, detail="Recommendation not found")
|
|
rec.saved = not rec.saved
|
|
return {"saved": rec.saved}
|
|
|
|
|
|
@router.post("/{rec_id}/dislike")
|
|
async def dislike_recommendation(
|
|
rec_id: int,
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(Recommendation).where(Recommendation.id == rec_id, Recommendation.user_id == user.id)
|
|
)
|
|
rec = result.scalar_one_or_none()
|
|
if not rec:
|
|
raise HTTPException(status_code=404, detail="Recommendation not found")
|
|
rec.disliked = not rec.disliked
|
|
return {"disliked": rec.disliked}
|
|
|
|
|
|
@router.post("/{rec_id}/share")
|
|
async def share_recommendation(
|
|
rec_id: int,
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Generate a share link for a recommendation."""
|
|
result = await db.execute(
|
|
select(Recommendation).where(Recommendation.id == rec_id, Recommendation.user_id == user.id)
|
|
)
|
|
rec = result.scalar_one_or_none()
|
|
if not rec:
|
|
raise HTTPException(status_code=404, detail="Recommendation not found")
|
|
|
|
token = hashlib.sha256(f"{rec_id}:{settings.SECRET_KEY}".encode()).hexdigest()[:16]
|
|
return {"share_url": f"{settings.FRONTEND_URL}/shared/{rec_id}/{token}"}
|
|
|
|
|
|
@router.get("/shared/{rec_id}/{token}")
|
|
async def get_shared_recommendation(
|
|
rec_id: int,
|
|
token: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""View a shared recommendation (no auth required)."""
|
|
expected = hashlib.sha256(f"{rec_id}:{settings.SECRET_KEY}".encode()).hexdigest()[:16]
|
|
if token != expected:
|
|
raise HTTPException(status_code=404, detail="Invalid share link")
|
|
|
|
result = await db.execute(
|
|
select(Recommendation).where(Recommendation.id == rec_id)
|
|
)
|
|
rec = result.scalar_one_or_none()
|
|
if not rec:
|
|
raise HTTPException(status_code=404, detail="Recommendation not found")
|
|
|
|
return {
|
|
"title": rec.title,
|
|
"artist": rec.artist,
|
|
"album": rec.album,
|
|
"reason": rec.reason,
|
|
"youtube_url": rec.youtube_url,
|
|
"image_url": rec.image_url,
|
|
}
|
|
|
|
|
|
@router.post("/share-batch")
|
|
async def share_batch(
|
|
rec_ids: list[int],
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Generate a share link for multiple recommendations."""
|
|
ids_str = ",".join(str(i) for i in sorted(rec_ids))
|
|
token = hashlib.sha256(f"batch:{ids_str}:{settings.SECRET_KEY}".encode()).hexdigest()[:16]
|
|
return {"share_url": f"{settings.FRONTEND_URL}/shared/batch/{ids_str}/{token}"}
|
|
|
|
|
|
@router.get("/shared/batch/{ids_str}/{token}")
|
|
async def get_shared_batch(
|
|
ids_str: str,
|
|
token: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""View shared recommendations (no auth required)."""
|
|
expected = hashlib.sha256(f"batch:{ids_str}:{settings.SECRET_KEY}".encode()).hexdigest()[:16]
|
|
if token != expected:
|
|
raise HTTPException(status_code=404, detail="Invalid share link")
|
|
|
|
rec_ids = [int(i) for i in ids_str.split(",")]
|
|
result = await db.execute(
|
|
select(Recommendation).where(Recommendation.id.in_(rec_ids))
|
|
)
|
|
recs = result.scalars().all()
|
|
|
|
return {
|
|
"recommendations": [
|
|
{
|
|
"title": r.title,
|
|
"artist": r.artist,
|
|
"album": r.album,
|
|
"reason": r.reason,
|
|
"youtube_url": r.youtube_url,
|
|
"image_url": r.image_url,
|
|
}
|
|
for r in recs
|
|
]
|
|
}
|