mirror of
https://github.com/bnair123/MusicAnalyser.git
synced 2026-02-25 11:46:07 +00:00
fa28b98c1a
- Add reccobeats_id column to Track model for API mapping - Fix ReccoBeats batch size limit (max 40 IDs per request) - Extract spotify_id from href field in ReccoBeats responses - Fix OpenAI API: remove unsupported temperature param, increase max_completion_tokens to 4000 - Add playlist/user management methods to spotify_client for future auto-playlist feature
137 lines
4.8 KiB
Python
137 lines
4.8 KiB
Python
import httpx
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
RECCOBEATS_BASE_URL = "https://api.reccobeats.com/v1"
|
|
|
|
|
|
class ReccoBeatsClient:
|
|
def __init__(self):
|
|
self.timeout = 30.0
|
|
|
|
async def get_tracks(self, spotify_ids: List[str]) -> List[Dict[str, Any]]:
|
|
if not spotify_ids:
|
|
return []
|
|
|
|
ids_param = ",".join(spotify_ids)
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
try:
|
|
response = await client.get(
|
|
f"{RECCOBEATS_BASE_URL}/track", params={"ids": ids_param}
|
|
)
|
|
if response.status_code != 200:
|
|
print(f"ReccoBeats /track returned status {response.status_code}")
|
|
return []
|
|
|
|
content = response.json().get("content", [])
|
|
|
|
for item in content:
|
|
href = item.get("href", "")
|
|
if "spotify.com/track/" in href:
|
|
item["spotify_id"] = href.split("/track/")[-1].split("?")[0]
|
|
|
|
return content
|
|
except Exception as e:
|
|
print(f"ReccoBeats /track error: {e}")
|
|
return []
|
|
|
|
async def get_audio_features(self, spotify_ids: List[str]) -> List[Dict[str, Any]]:
|
|
"""Fetch audio features for tracks. Batches in chunks of 40 (API limit)."""
|
|
if not spotify_ids:
|
|
return []
|
|
|
|
all_results = []
|
|
batch_size = 40 # ReccoBeats API returns 400 for 50+ IDs
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
for i in range(0, len(spotify_ids), batch_size):
|
|
batch = spotify_ids[i : i + batch_size]
|
|
ids_param = ",".join(batch)
|
|
|
|
try:
|
|
response = await client.get(
|
|
f"{RECCOBEATS_BASE_URL}/audio-features",
|
|
params={"ids": ids_param},
|
|
)
|
|
if response.status_code != 200:
|
|
print(
|
|
f"ReccoBeats /audio-features returned status {response.status_code}"
|
|
)
|
|
continue
|
|
|
|
content = response.json().get("content", [])
|
|
|
|
for item in content:
|
|
href = item.get("href", "")
|
|
if "spotify.com/track/" in href:
|
|
item["spotify_id"] = href.split("/track/")[-1].split("?")[0]
|
|
|
|
all_results.extend(content)
|
|
except Exception as e:
|
|
print(f"ReccoBeats /audio-features error: {e}")
|
|
|
|
return all_results
|
|
|
|
async def get_recommendations(
|
|
self,
|
|
seed_ids: List[str],
|
|
size: int = 20,
|
|
negative_seeds: Optional[List[str]] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
if not seed_ids:
|
|
return []
|
|
|
|
if len(seed_ids) > 5:
|
|
seed_ids = seed_ids[:5]
|
|
|
|
params = {"seeds": ",".join(seed_ids), "size": size}
|
|
|
|
if negative_seeds:
|
|
if len(negative_seeds) > 5:
|
|
negative_seeds = negative_seeds[:5]
|
|
params["negativeSeeds"] = ",".join(negative_seeds)
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
try:
|
|
response = await client.get(
|
|
f"{RECCOBEATS_BASE_URL}/track/recommendation", params=params
|
|
)
|
|
if response.status_code != 200:
|
|
print(
|
|
f"ReccoBeats /recommendation returned status {response.status_code}"
|
|
)
|
|
return []
|
|
|
|
content = response.json().get("content", [])
|
|
|
|
for item in content:
|
|
href = item.get("href", "")
|
|
if "spotify.com/track/" in href:
|
|
item["spotify_id"] = href.split("/track/")[-1].split("?")[0]
|
|
|
|
return content
|
|
except Exception as e:
|
|
print(f"ReccoBeats /recommendation error: {e}")
|
|
return []
|
|
|
|
async def get_audio_features_by_reccobeats_ids(
|
|
self, reccobeats_ids: List[str]
|
|
) -> List[Dict[str, Any]]:
|
|
if not reccobeats_ids:
|
|
return []
|
|
|
|
results = []
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
for rb_id in reccobeats_ids:
|
|
try:
|
|
response = await client.get(
|
|
f"{RECCOBEATS_BASE_URL}/track/{rb_id}/audio-features"
|
|
)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
data["reccobeats_id"] = rb_id
|
|
results.append(data)
|
|
except Exception as e:
|
|
print(f"ReccoBeats audio-features for {rb_id} error: {e}")
|
|
|
|
return results
|