feat: Initial backend setup for Music Analyser

- Created FastAPI backend structure.
- Implemented Spotify Recently Played ingestion logic.
- Set up SQLite database with SQLAlchemy models.
- Added AI Service using Google Gemini.
- Created helper scripts for auth and background worker.
- Added Dockerfile and GitHub Actions workflow.
This commit is contained in:
google-labs-jules[bot]
2025-12-24 17:26:01 +00:00
parent a458eb00db
commit a97997a17a
16 changed files with 579 additions and 2 deletions

View File

@@ -0,0 +1,70 @@
import os
import base64
import time
import httpx
from fastapi import HTTPException
SPOTIFY_TOKEN_URL = "https://accounts.spotify.com/api/token"
SPOTIFY_API_BASE = "https://api.spotify.com/v1"
class SpotifyClient:
def __init__(self, client_id: str, client_secret: str, refresh_token: str):
self.client_id = client_id
self.client_secret = client_secret
self.refresh_token = refresh_token
self.access_token = None
self.token_expires_at = 0
async def get_access_token(self):
"""Returns a valid access token, refreshing if necessary."""
if self.access_token and time.time() < self.token_expires_at:
return self.access_token
print("Refreshing Spotify Access Token...")
async with httpx.AsyncClient() as client:
auth_str = f"{self.client_id}:{self.client_secret}"
b64_auth = base64.b64encode(auth_str.encode()).decode()
response = await client.post(
SPOTIFY_TOKEN_URL,
data={
"grant_type": "refresh_token",
"refresh_token": self.refresh_token,
},
headers={"Authorization": f"Basic {b64_auth}"},
)
if response.status_code != 200:
print(f"Failed to refresh token: {response.text}")
raise Exception("Could not refresh Spotify token")
data = response.json()
self.access_token = data["access_token"]
# expires_in is usually 3600 seconds. buffer by 60s
self.token_expires_at = time.time() + data["expires_in"] - 60
return self.access_token
async def get_recently_played(self, limit=50):
token = await self.get_access_token()
async with httpx.AsyncClient() as client:
response = await client.get(
f"{SPOTIFY_API_BASE}/me/player/recently-played",
params={"limit": limit},
headers={"Authorization": f"Bearer {token}"},
)
if response.status_code != 200:
print(f"Error fetching recently played: {response.text}")
return []
return response.json().get("items", [])
async def get_track(self, track_id: str):
token = await self.get_access_token()
async with httpx.AsyncClient() as client:
response = await client.get(
f"{SPOTIFY_API_BASE}/tracks/{track_id}",
headers={"Authorization": f"Bearer {token}"},
)
if response.status_code != 200:
return None
return response.json()