push all

.gitignore

@@ -0,0 +1,3 @@
+.env
+__pycache__/
+.cache

Dockerfile

@@ -0,0 +1,51 @@
+# Base Image
+FROM python:3.10-slim
+
+ENV DEBIAN_FRONTEND=noninteractive \
+    PYTHONUNBUFFERED=1 \
+    PYTHONDONTWRITEBYTECODE=1
+
+WORKDIR /code
+
+# System Dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    git \
+    curl \
+    libopenblas-dev \
+    libomp-dev \
+    libpq-dev \
+    libgl1 \
+    libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Hugging Face + model tools (extra, if needed)
+RUN pip install --no-cache-dir huggingface-hub sentencepiece accelerate fasttext
+
+# Hugging Face cache environment
+ENV HF_HOME=/models/huggingface \
+    TRANSFORMERS_CACHE=/models/huggingface \
+    HUGGINGFACE_HUB_CACHE=/models/huggingface \
+    HF_HUB_CACHE=/models/huggingface
+
+# Create cache dir and set permissions
+RUN mkdir -p /models/huggingface && chmod -R 777 /models/huggingface
+
+# Copy project files
+COPY . .
+
+# Expose FastAPI port (Hugging Face Spaces expects 7860)
+EXPOSE 7860
+
+# CPU-related env tuning
+ENV OMP_NUM_THREADS=4 \
+    MKL_NUM_THREADS=4 \
+    NUMEXPR_NUM_THREADS=4
+
+# Run FastAPI app with uvicorn
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860", "--workers", "1", "--timeout-keep-alive", "30"]
+
+

README.md

@@ -7,4 +7,263 @@ sdk: docker
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+## Vicoka – Mood → Spotify Song Recommender (Python)
+
+Backend prototype that:
+- **Detects mood** from **text** (Hugging Face `transformers`) or **face image**
+- **Maps mood → music features**
+- **Fetches a recommended song** using the **Spotify Web API**
+- (Optionally) **logs interactions** in **PostgreSQL**
+
+### Spotify API (access token) — what you need to know
+
+There are two common auth flows:
+- **Client Credentials (app-only)**: best for server-side calls to public Spotify data (search, recommendations, track metadata).  
+  This is what our backend uses via `spotipy` (it requests/refreshes the 1‑hour access token automatically).
+- **Authorization Code (user login)**: required if you need private user data (profile email, private playlists, saving tracks, etc.).
+
+### Set Spotify credentials
+
+1. Create an app in the Spotify Developer Dashboard and copy:
+   - `Client ID`
+   - `Client Secret`
+2. Set environment variables (local or on Hugging Face Space “Secrets”):
+   - `SPOTIFY_CLIENT_ID`
+   - `SPOTIFY_CLIENT_SECRET`
+   - `SPOTIFY_REDIRECT_URI` (needed for user login / `/v1/me`)
+
+### Spotify Dashboard app setup (Redirect URIs)
+
+In the Spotify Developer Dashboard:
+- Create your app (name/description) and accept the terms.
+- Go to **Edit Settings**.
+- Add a value under **Redirect URIs** that matches your backend callback exactly.
+
+For this project the callback endpoint is:
+- `GET /auth/spotify/callback`
+
+Spotify redirect URI rules (enforced for newer apps):
+- **Do not use `localhost`** (Spotify blocks it). Use `127.0.0.1` or `[::1]` for local development.
+- **Use HTTPS** for non-loopback redirect URIs (e.g. Hugging Face Space URLs).
+
+So your Redirect URI should look like one of these examples:
+- **Local dev**: `http://127.0.0.1:8000/auth/spotify/callback`
+- **Local dev (IPv6)**: `http://[::1]:8000/auth/spotify/callback`
+- **Hugging Face Space**: `https://<your-space-subdomain>.hf.space/auth/spotify/callback`
+
+Important:
+- The Redirect URI must match **exactly** (scheme/host/port/path).
+- **Never expose your Client Secret** publicly; store it in Space **Secrets**.
+
+### (Optional) Request a token manually (curl)
+
+This is the same “Client Credentials” token request you pasted:
+
+```bash
+curl -X POST "https://accounts.spotify.com/api/token" \
+  -H "Content-Type: application/x-www-form-urlencoded" \
+  -d "grant_type=client_credentials&client_id=$SPOTIFY_CLIENT_ID&client_secret=$SPOTIFY_CLIENT_SECRET"
+```
+
+### Using `Authorization: Bearer <token>` (examples)
+
+- Call Spotify **directly** (track endpoint):
+
+```bash
+ACCESS_TOKEN="your_access_token_here"
+curl --request GET \
+  "https://api.spotify.com/v1/tracks/2TpxZ7JUBn3uw46aR7qd6V" \
+  --header "Authorization: Bearer $ACCESS_TOKEN"
+```
+
+- Call our backend **track proxy demo** (uses app-only token server-side):
+
+```bash
+curl "http://127.0.0.1:8000/spotify/tracks/2TpxZ7JUBn3uw46aR7qd6V"
+```
+
+- Call Spotify **current user profile** (`/v1/me`):
+  - **Requires Authorization Code flow user token** (Client Credentials will not work)
+
+```bash
+USER_ACCESS_TOKEN="your_user_access_token_here"
+curl "https://api.spotify.com/v1/me" \
+  --header "Authorization: Bearer $USER_ACCESS_TOKEN"
+```
+
+- Call our backend `/spotify/me` (passes your Bearer token through):
+
+```bash
+USER_ACCESS_TOKEN="your_user_access_token_here"
+curl "http://127.0.0.1:8000/spotify/me" \
+  --header "Authorization: Bearer $USER_ACCESS_TOKEN"
+```
+
+### Run locally
+
+```bash
+pip install -r requirements.txt
+export SPOTIFY_CLIENT_ID="..."
+export SPOTIFY_CLIENT_SECRET="..."
+export DATABASE_URL="postgresql+psycopg2://user:password@localhost:5432/vicoka"
+uvicorn app.main:app --reload
+```
+
+### Test the API
+
+- **Text mood → song**:
+
+```bash
+curl -X POST "http://127.0.0.1:8000/mood/text" \
+  -H "Content-Type: application/json" \
+  -d '{"text":"I feel calm and focused"}'
+```
+
+- **Face image → song**:
+
+```bash
+curl -X POST "http://127.0.0.1:8000/mood/face" \
+  -F "file=@/path/to/face.jpg"
+```
+
+### User login (Authorization Code flow) for private endpoints (like `/v1/me`)
+
+1. Visit (in browser) to login and consent:
+
+`GET /auth/spotify/login`
+
+Example:
+
+`http://127.0.0.1:8000/auth/spotify/login`
+
+2. After Spotify redirects back, your backend will exchange the `code` and return JSON:
+- `access_token` (use as `Authorization: Bearer ...`)
+- `refresh_token` (store securely; used to refresh access later)
+- `expires_in` (seconds)
+
+To request additional permissions, pass scopes:
+
+`/auth/spotify/login?scope=user-read-email%20user-read-private%20playlist-read-private%20playlist-modify-private`
+
+### Playlists (user token + scopes)
+
+Playlist endpoints require a **user access token** (Authorization Code flow) and the right **scopes**:
+- **Read playlists**:
+  - `playlist-read-private` (to include private playlists)
+  - `playlist-read-collaborative` (to include collaborative playlists)
+- **Create / modify playlists**:
+  - `playlist-modify-public` (if `public=true`)
+  - `playlist-modify-private` (if `public=false`)
+
+Backend endpoints added:
+- `GET /spotify/playlists` (current user’s playlists)
+- `GET /spotify/playlists/{playlist_id}` (playlist details)
+- `GET /spotify/playlists/{playlist_id}/items` (playlist items, paginated)
+- `POST /spotify/playlists` (create playlist)
+- `POST /spotify/playlists/{playlist_id}/items` (add tracks by URI)
+
+Examples (after you have `USER_ACCESS_TOKEN`):
+
+```bash
+curl "http://127.0.0.1:8000/spotify/playlists?limit=20&offset=0" \
+  --header "Authorization: Bearer $USER_ACCESS_TOKEN"
+```
+
+```bash
+curl -X POST "http://127.0.0.1:8000/spotify/playlists" \
+  --header "Authorization: Bearer $USER_ACCESS_TOKEN" \
+  -H "Content-Type: application/json" \
+  -d '{"user_id":"YOUR_SPOTIFY_USER_ID","name":"Vicoka – Happy Mix","public":false,"description":"Created by Vicoka"}'
+```
+
+```bash
+curl -X POST "http://127.0.0.1:8000/spotify/playlists/PLAYLIST_ID/items" \
+  --header "Authorization: Bearer $USER_ACCESS_TOKEN" \
+  -H "Content-Type: application/json" \
+  -d '{"uris":["spotify:track:2TpxZ7JUBn3uw46aR7qd6V"]}'
+```
+
+### Spotify URIs vs IDs vs URLs (quick guide)
+
+- **Spotify URI**: identifies a resource *and* its type  
+  Example: `spotify:track:6rqhFgbbKwnb9MLmUQDhG6`
+- **Spotify ID**: the base62 string at the end (often 22 chars)  
+  Example: `6rqhFgbbKwnb9MLmUQDhG6`
+- **Spotify URL**: web link that opens the client  
+  Example: `https://open.spotify.com/track/6rqhFgbbKwnb9MLmUQDhG6`
+
+In this backend:
+- `POST /spotify/playlists/{playlist_id}/items` accepts **track references** in any of these forms:
+  - `spotify:track:<id>` (URI)
+  - `https://open.spotify.com/track/<id>` (URL)
+  - `<id>` (raw 22-char ID)
+  and normalizes them into track URIs before calling Spotify.
+
+Debug helper:
+- `GET /spotify/parse?value=...` returns the parsed `{resource_type, id, uri, url}`.
+
+### Track relinking (market availability)
+
+Track availability varies by country. Spotify can “relink” a track to another catalog instance that **is playable** in a given market.
+
+How to use it:
+- Pass a `market` query parameter (ISO country code like `US`, `GB`, `IN`) to track/playlist-item endpoints.
+- For **user tokens**, you can also use `market=from_token` to use the user’s country.
+
+Supported in this backend:
+- `GET /spotify/tracks/{track_id}?market=US`
+- `GET /spotify/playlists/{playlist_id}/items?market=from_token`
+
+What changes in responses when `market` is supplied:
+- Spotify may include **`is_playable`**
+- If relinked, Spotify may include **`linked_from`** (the originally requested track)
+- If no relink is possible, you may see **`restrictions`** with reason `"market"`
+
+Important:
+- If you later do operations that must target the **original track**, use the ID/URI inside **`linked_from`** (Spotify can reject using the relinked root ID for some operations).
+
+### Notes for Hugging Face Spaces (Docker)
+
+- Your container must listen on **port 7860**.
+- Put `SPOTIFY_CLIENT_ID` / `SPOTIFY_CLIENT_SECRET` in the Space **Settings → Secrets** (don’t hardcode them).
+
+More info: [Spaces config reference](https://huggingface.co/docs/hub/spaces-config-reference)
+
+### Spotify Web API responses (status codes)
+
+Spotify endpoints can return common HTTP codes like:
+- **200/201/204**: success (204 can be “no content”)
+- **400**: bad request (often includes a JSON error message)
+- **401/403**: unauthorized/forbidden (token missing/invalid or insufficient permissions)
+- **404**: not found
+- **429**: rate limited (**Retry-After** header tells you how many seconds to wait)
+
+In this project, our `/spotify/*` demo endpoints **propagate Spotify’s status code** and include a structured JSON `detail` similar to Spotify’s own error object. If Spotify returns **429**, we also forward the **Retry-After** header.
+
+### Spotify quota modes (Development vs Extended)
+
+Spotify apps can run in two quota modes:
+- **Development mode**:
+  - Intended for building/testing and small demos
+  - Only a small number of Spotify users can use your app unless they’re **allowlisted**
+  - If a user is not allowlisted, Spotify may return **403 Forbidden** for that user’s token
+- **Extended quota mode**:
+  - For production-scale apps with higher limits and a broader audience
+
+Practical tips for this project:
+- **If your friend/classmate gets a 403** after logging in, you likely need to add them in the Spotify Developer Dashboard under **Users and Access / User Management** for your app (development mode allowlist).
+- **Premium requirement**: most **Web API** endpoints (profile, playlists, recommendations, search) do not require Premium, but **playback/streaming features** (player controls, actual streaming) typically require Premium and additional scopes.
+
+### Rate limits (429 Too Many Requests)
+
+If you exceed Spotify’s rate limit in a rolling window, Spotify responds with:
+- **429** and usually a **`Retry-After`** header (seconds to wait)
+
+This backend handles that by:
+- **Auto-retrying** Spotify requests a small number of times
+- **Sleeping for `Retry-After` seconds** when Spotify returns 429 (then retrying)
+- Also retrying some transient upstream errors (e.g. 502/503) with short exponential backoff
+
+App-side tips:
+- Prefer fewer calls (lazy load playlists/items)
+- Use batch endpoints when available (e.g. “Get Multiple …”)

System Overview.md

@@ -0,0 +1,43 @@
+## Vicoka – Mood-based Spotify Song Recommender
+
+Backend prototype for an AI agent that:
+- **Takes mood as input** via text or facial scan
+- **Infers emotion** using Hugging Face models
+- **Maps emotion → music** and fetches a suitable track from the **Spotify Web API**
+- (Optionally) **logs interactions** in PostgreSQL.
+
+### Tech stack
+- **Language**: Python
+- **Web framework**: FastAPI
+- **AI models**: Hugging Face `transformers` (text & image pipelines)
+- **Database**: PostgreSQL (via SQLAlchemy)
+- **Music**: Spotify Web API (`spotipy`)
+
+### Quick start
+1. Create and activate a virtualenv.
+2. Install dependencies:
+   ```bash
+   pip install -r requirements.txt
+   ```
+3. Set environment variables (or a `.env` file):
+   - `DATABASE_URL` – e.g. `postgresql+psycopg2://user:password@localhost:5432/vicoka`
+   - `SPOTIFY_CLIENT_ID`
+   - `SPOTIFY_CLIENT_SECRET`
+   - `SPOTIFY_REDIRECT_URI` (for OAuth, if you later build a full user-auth flow)
+4. Run the API:
+   ```bash
+   uvicorn app.main:app --reload
+   ```
+
+### High-level flow
+1. Client sends **text** or **image** to backend.
+2. Backend runs Hugging Face **text** or **image** emotion pipeline.
+3. Emotion label (e.g., `joy`, `sadness`, `anger`, `calm`) is mapped to Spotify **search parameters**.
+4. Backend queries Spotify and returns a **recommended track** (or playlist).
+
+### Next steps
+- Implement a simple frontend (web or mobile) to capture text and camera input.
+- Tune emotion → Spotify mapping based on user feedback.
+- Add authentication and per-user history in PostgreSQL.
+
+

app/config.py

@@ -0,0 +1,27 @@
+import os
+from functools import lru_cache
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+class Settings:
+    """Application settings loaded from environment."""
+
+    database_url: str = os.getenv(
+        "DATABASE_URL",
+        "postgresql+psycopg2://postgres:postgres@localhost:5432/vicoka",
+    )
+
+    spotify_client_id: str | None = os.getenv("SPOTIFY_CLIENT_ID")
+    spotify_client_secret: str | None = os.getenv("SPOTIFY_CLIENT_SECRET")
+    spotify_redirect_uri: str | None = os.getenv("SPOTIFY_REDIRECT_URI")
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()
+
+
+

app/db.py

@@ -0,0 +1,26 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, DeclarativeBase
+
+from .config import get_settings
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+settings = get_settings()
+engine = create_engine(settings.database_url, echo=False, future=True)
+SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
+
+
+def get_db():
+    from sqlalchemy.orm import Session
+
+    db: Session = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
+

app/main.py

@@ -0,0 +1,421 @@
+from io import BytesIO
+
+from fastapi import Depends, FastAPI, File, HTTPException, UploadFile, Header
+from fastapi.responses import RedirectResponse
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+from PIL import Image
+from sqlalchemy.orm import Session
+
+from .db import Base, engine, get_db
+from .models import MoodLog, RecommendationLog
+from .mood_text_model import analyze_text_emotion
+from .mood_face_model import analyze_face_emotion
+from .spotify_client import recommend_track_for_emotion, get_track_via_bearer
+from .spotify_user_api import get_current_user_profile
+from .spotify_http import SpotifyAPIError
+from .spotify_oauth import (
+    build_authorize_url,
+    default_redirect_uri,
+    exchange_code_for_token,
+    generate_state,
+    validate_redirect_uri,
+ )
+from .spotify_playlists_api import (
+    add_items_to_playlist,
+    create_playlist,
+    get_playlist,
+    get_playlist_items,
+    list_current_user_playlists,
+)
+from .spotify_ids import parse_spotify_ref, to_track_uri
+
+
+Base.metadata.create_all(bind=engine)
+
+app = FastAPI(title="Vicoka – Mood-based Spotify Recommender")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+class TextMoodRequest(BaseModel):
+    text: str
+
+
+class RecommendationResponse(BaseModel):
+    mood_label: str
+    mood_score: float
+    spotify_track_id: str | None
+    track_name: str | None
+    artists: list[str] | None
+    preview_url: str | None
+    external_url: str | None
+
+
+def _extract_bearer_token(authorization: str | None) -> str:
+    if not authorization:
+        raise HTTPException(
+            status_code=401,
+            detail="Missing Authorization header. Use: Authorization: Bearer <access_token>",
+        )
+    if not authorization.lower().startswith("bearer "):
+        raise HTTPException(
+            status_code=401,
+            detail="Invalid Authorization header format. Use: Bearer <access_token>",
+        )
+    return authorization.split(" ", 1)[1].strip()
+
+
+@app.post("/mood/text", response_model=RecommendationResponse)
+def mood_from_text(
+    body: TextMoodRequest, db: Session = Depends(get_db)
+) -> RecommendationResponse:
+    if not body.text.strip():
+        raise HTTPException(status_code=400, detail="Text cannot be empty")
+
+    label, score = analyze_text_emotion(body.text)
+    track = recommend_track_for_emotion(label, source="text")
+
+    mood_log = MoodLog(
+        source="text",
+        raw_input=body.text[:512],
+        emotion_label=label,
+        emotion_score=score,
+    )
+    db.add(mood_log)
+    db.flush()
+
+    rec_log = None
+    track_id = None
+    if track:
+        track_id = track["id"]
+        rec_log = RecommendationLog(
+            user_id=None,
+            mood_id=mood_log.id,
+            spotify_track_id=track_id,
+        )
+        db.add(rec_log)
+
+    db.commit()
+
+    return RecommendationResponse(
+        mood_label=label,
+        mood_score=score,
+        spotify_track_id=track_id,
+        track_name=track.get("name") if track else None,
+        artists=track.get("artists") if track else None,
+        preview_url=track.get("preview_url") if track else None,
+        external_url=track.get("external_url") if track else None,
+    )
+
+
+@app.post("/mood/face", response_model=RecommendationResponse)
+async def mood_from_face(
+    file: UploadFile = File(...), db: Session = Depends(get_db)
+) -> RecommendationResponse:
+    contents = await file.read()
+    try:
+        image = Image.open(BytesIO(contents)).convert("RGB")
+    except Exception as exc:
+        raise HTTPException(status_code=400, detail="Invalid image file") from exc
+
+    label, score = analyze_face_emotion(image)
+    track = recommend_track_for_emotion(label, source="face")
+
+    mood_log = MoodLog(
+        source="face",
+        raw_input=None,
+        emotion_label=label,
+        emotion_score=score,
+    )
+    db.add(mood_log)
+    db.flush()
+
+    rec_log = None
+    track_id = None
+    if track:
+        track_id = track["id"]
+        rec_log = RecommendationLog(
+            user_id=None,
+            mood_id=mood_log.id,
+            spotify_track_id=track_id,
+        )
+        db.add(rec_log)
+
+    db.commit()
+
+    return RecommendationResponse(
+        mood_label=label,
+        mood_score=score,
+        spotify_track_id=track_id,
+        track_name=track.get("name") if track else None,
+        artists=track.get("artists") if track else None,
+        preview_url=track.get("preview_url") if track else None,
+        external_url=track.get("external_url") if track else None,
+    )
+
+
+@app.get("/health")
+def health() -> dict:
+    return {"status": "ok"}
+
+
+@app.get("/spotify/tracks/{track_id}")
+def spotify_track(track_id: str, market: str | None = None) -> dict:
+    """
+    Demo endpoint: fetches track info using Spotify Web API with
+    Authorization: Bearer <token> (app-only Client Credentials token).
+    """
+    try:
+        return get_track_via_bearer(track_id, market=market)
+    except SpotifyAPIError as exc:
+        headers = {}
+        if exc.retry_after is not None:
+            headers["Retry-After"] = str(exc.retry_after)
+        raise HTTPException(status_code=exc.status_code, detail=exc.to_dict(), headers=headers) from exc
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail="Unexpected server error") from exc
+
+
+@app.get("/spotify/me")
+def spotify_me(authorization: str | None = Header(default=None)) -> dict:
+    """
+    Calls Spotify GET /v1/me.
+
+    IMPORTANT: This requires a USER access token (Authorization Code flow),
+    NOT a Client Credentials token.
+
+    Send header:
+      Authorization: Bearer <user_access_token>
+    """
+    token = _extract_bearer_token(authorization)
+    try:
+        return get_current_user_profile(token)
+    except SpotifyAPIError as exc:
+        headers = {}
+        if exc.retry_after is not None:
+            headers["Retry-After"] = str(exc.retry_after)
+        raise HTTPException(status_code=exc.status_code, detail=exc.to_dict(), headers=headers) from exc
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail="Unexpected server error") from exc
+
+
+@app.get("/spotify/playlists")
+def spotify_my_playlists(
+    authorization: str | None = Header(default=None),
+    limit: int = 20,
+    offset: int = 0,
+) -> dict:
+    """
+    List current user's playlists: GET /v1/me/playlists
+    Scopes:
+    - playlist-read-private (to include private playlists)
+    - playlist-read-collaborative (to include collaborative playlists)
+    """
+    token = _extract_bearer_token(authorization)
+    try:
+        return list_current_user_playlists(token, limit=limit, offset=offset)
+    except SpotifyAPIError as exc:
+        headers = {}
+        if exc.retry_after is not None:
+            headers["Retry-After"] = str(exc.retry_after)
+        raise HTTPException(status_code=exc.status_code, detail=exc.to_dict(), headers=headers) from exc
+
+
+@app.get("/spotify/playlists/{playlist_id}")
+def spotify_get_playlist(
+    playlist_id: str,
+    authorization: str | None = Header(default=None),
+) -> dict:
+    token = _extract_bearer_token(authorization)
+    try:
+        return get_playlist(token, playlist_id)
+    except SpotifyAPIError as exc:
+        headers = {}
+        if exc.retry_after is not None:
+            headers["Retry-After"] = str(exc.retry_after)
+        raise HTTPException(status_code=exc.status_code, detail=exc.to_dict(), headers=headers) from exc
+
+
+@app.get("/spotify/playlists/{playlist_id}/items")
+def spotify_get_playlist_items(
+    playlist_id: str,
+    authorization: str | None = Header(default=None),
+    limit: int = 50,
+    offset: int = 0,
+    additional_types: str = "track,episode",
+    market: str | None = None,
+) -> dict:
+    token = _extract_bearer_token(authorization)
+    try:
+        return get_playlist_items(
+            token,
+            playlist_id,
+            limit=limit,
+            offset=offset,
+            additional_types=additional_types,
+            market=market,
+        )
+    except SpotifyAPIError as exc:
+        headers = {}
+        if exc.retry_after is not None:
+            headers["Retry-After"] = str(exc.retry_after)
+        raise HTTPException(status_code=exc.status_code, detail=exc.to_dict(), headers=headers) from exc
+
+
+class CreatePlaylistRequest(BaseModel):
+    user_id: str
+    name: str
+    description: str | None = None
+    public: bool = True
+    collaborative: bool = False
+
+
+@app.post("/spotify/playlists")
+def spotify_create_playlist(
+    body: CreatePlaylistRequest,
+    authorization: str | None = Header(default=None),
+) -> dict:
+    """
+    Create a playlist for a user: POST /v1/users/{user_id}/playlists
+    Scopes:
+    - playlist-modify-public (if public=true)
+    - playlist-modify-private (if public=false)
+    """
+    token = _extract_bearer_token(authorization)
+    try:
+        return create_playlist(
+            token,
+            body.user_id,
+            name=body.name,
+            description=body.description,
+            public=body.public,
+            collaborative=body.collaborative,
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    except SpotifyAPIError as exc:
+        headers = {}
+        if exc.retry_after is not None:
+            headers["Retry-After"] = str(exc.retry_after)
+        raise HTTPException(status_code=exc.status_code, detail=exc.to_dict(), headers=headers) from exc
+
+
+class AddItemsRequest(BaseModel):
+    uris: list[str]
+    position: int | None = None
+
+
+@app.post("/spotify/playlists/{playlist_id}/items")
+def spotify_add_items(
+    playlist_id: str,
+    body: AddItemsRequest,
+    authorization: str | None = Header(default=None),
+) -> dict:
+    """
+    Add items to a playlist: POST /v1/playlists/{playlist_id}/tracks
+    Scopes:
+    - playlist-modify-public or playlist-modify-private (depending on playlist visibility)
+    """
+    token = _extract_bearer_token(authorization)
+    try:
+        uris = [to_track_uri(u) for u in body.uris]
+        return add_items_to_playlist(
+            token,
+            playlist_id,
+            uris=uris,
+            position=body.position,
+        )
+    except SpotifyAPIError as exc:
+        headers = {}
+        if exc.retry_after is not None:
+            headers["Retry-After"] = str(exc.retry_after)
+        raise HTTPException(status_code=exc.status_code, detail=exc.to_dict(), headers=headers) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@app.get("/spotify/parse")
+def spotify_parse(value: str) -> dict:
+    """
+    Debug helper: parse Spotify URI/URL/ID into {type, id, uri, url}.
+    """
+    ref = parse_spotify_ref(value)
+    return {
+        "resource_type": ref.resource_type,
+        "id": ref.id,
+        "uri": ref.uri,
+        "url": ref.url,
+    }
+
+
+@app.get("/auth/spotify/login")
+def spotify_login(
+    scope: str = "user-read-email user-read-private",
+    redirect_uri: str | None = None,
+) -> RedirectResponse:
+    """
+    Redirect user to Spotify consent screen (Authorization Code flow).
+
+    IMPORTANT:
+    - The redirect_uri MUST exactly match one of the Redirect URIs configured in
+      your Spotify Developer Dashboard app settings.
+    - If redirect_uri is omitted, we use SPOTIFY_REDIRECT_URI from env.
+    """
+    ru = redirect_uri or default_redirect_uri()
+    if not ru:
+        raise HTTPException(
+            status_code=500,
+            detail="Missing redirect URI. Set SPOTIFY_REDIRECT_URI or pass ?redirect_uri=",
+        )
+    try:
+        validate_redirect_uri(ru)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    state = generate_state()
+    url = build_authorize_url(redirect_uri=ru, scope=scope, state=state)
+    return RedirectResponse(url=url, status_code=302)
+
+
+@app.get("/auth/spotify/callback")
+def spotify_callback(
+    code: str | None = None,
+    state: str | None = None,
+    error: str | None = None,
+    redirect_uri: str | None = None,
+) -> dict:
+    """
+    Spotify redirects here after login. Exchange `code` for access token.
+
+    For production:
+    - Validate `state` (CSRF protection) using a server-side session store.
+    - Store refresh token securely (e.g., Postgres) for long-lived sessions.
+    """
+    if error:
+        raise HTTPException(status_code=400, detail={"error": error, "state": state})
+    if not code:
+        raise HTTPException(status_code=400, detail="Missing `code` query parameter")
+
+    ru = redirect_uri or default_redirect_uri()
+    if not ru:
+        raise HTTPException(
+            status_code=500,
+            detail="Missing redirect URI. Set SPOTIFY_REDIRECT_URI or pass ?redirect_uri=",
+        )
+    try:
+        validate_redirect_uri(ru)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    try:
+        token = exchange_code_for_token(code=code, redirect_uri=ru)
+        return token
+    except SpotifyAPIError as exc:
+        raise HTTPException(status_code=exc.status_code, detail=exc.to_dict()) from exc
+
+

app/models.py

@@ -0,0 +1,56 @@
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import String, Integer, DateTime, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from .db import Base
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    spotify_user_id: Mapped[Optional[str]] = mapped_column(String(128), unique=True)
+
+    moods: Mapped[list["MoodLog"]] = relationship(back_populates="user")
+    recommendations: Mapped[list["RecommendationLog"]] = relationship(
+        back_populates="user"
+    )
+
+
+class MoodLog(Base):
+    __tablename__ = "mood_logs"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    user_id: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"))
+    source: Mapped[str] = mapped_column(String(32))
+    raw_input: Mapped[Optional[str]] = mapped_column(String(512))
+    emotion_label: Mapped[str] = mapped_column(String(64))
+    emotion_score: Mapped[float] = mapped_column()
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=datetime.utcnow
+    )
+
+    user: Mapped[Optional[User]] = relationship(back_populates="moods")
+    recommendation: Mapped[Optional["RecommendationLog"]] = relationship(
+        back_populates="mood", uselist=False
+    )
+
+
+class RecommendationLog(Base):
+    __tablename__ = "recommendation_logs"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    user_id: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"))
+    mood_id: Mapped[int] = mapped_column(ForeignKey("mood_logs.id"))
+    spotify_track_id: Mapped[str] = mapped_column(String(64))
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=datetime.utcnow
+    )
+
+    user: Mapped[Optional[User]] = relationship(back_populates="recommendations")
+    mood: Mapped[MoodLog] = relationship(back_populates="recommendation")
+
+
+

app/mood_face_model.py

@@ -0,0 +1,62 @@
+from functools import lru_cache
+from typing import Literal
+
+import torch
+from PIL import Image
+from transformers import AutoImageProcessor, AutoModelForImageClassification
+
+
+FaceEmotionLabel = Literal[
+    "anger",
+    "disgust",
+    "fear",
+    "happiness",
+    "neutral",
+    "sadness",
+    "surprise",
+]
+
+
+MODEL_NAME = "dima806/facial_emotions_image_detection"
+
+
+@lru_cache
+def _get_face_model():
+    processor = AutoImageProcessor.from_pretrained(MODEL_NAME)
+    model = AutoModelForImageClassification.from_pretrained(MODEL_NAME)
+    model.eval()
+    return processor, model
+
+
+def analyze_face_emotion(image: Image.Image) -> tuple[FaceEmotionLabel, float]:
+    """Return (emotion_label, score) for the given face image."""
+    processor, model = _get_face_model()
+
+    inputs = processor(images=image, return_tensors="pt")
+
+    with torch.no_grad():
+        outputs = model(**inputs)
+        logits = outputs.logits
+        probs = torch.nn.functional.softmax(logits, dim=-1)[0]
+
+    id2label = model.config.id2label
+    best_id = int(torch.argmax(probs).item())
+    best_label = id2label[best_id]
+    best_score = float(probs[best_id].item())
+
+    label = best_label.lower()
+    known_labels: set[str] = {
+        "anger",
+        "disgust",
+        "fear",
+        "happiness",
+        "neutral",
+        "sadness",
+        "surprise",
+    }
+
+    norm_label: FaceEmotionLabel = (label if label in known_labels else "neutral")  # type: ignore[assignment]
+    return norm_label, best_score
+
+
+

app/mood_text_model.py

@@ -0,0 +1,47 @@
+from functools import lru_cache
+from typing import Literal
+
+from transformers import pipeline
+
+
+EmotionLabel = Literal[
+    "anger",
+    "disgust",
+    "fear",
+    "joy",
+    "neutral",
+    "sadness",
+    "surprise",
+]
+
+
+@lru_cache
+def _get_pipeline():
+    return pipeline(
+        "text-classification",
+        model="j-hartmann/emotion-english-distilroberta-base",
+        top_k=None,
+    )
+
+
+def analyze_text_emotion(text: str) -> tuple[EmotionLabel, float]:
+    """Return (emotion_label, score) for the given text."""
+    clf = _get_pipeline()
+    outputs = clf(text)[0]
+    best = max(outputs, key=lambda x: x["score"])
+    label = best["label"].lower()
+    score = float(best["score"])
+    known_labels: set[str] = {
+        "anger",
+        "disgust",
+        "fear",
+        "joy",
+        "neutral",
+        "sadness",
+        "surprise",
+    }
+    norm_label: EmotionLabel = (label if label in known_labels else "neutral")  # type: ignore[assignment]
+    return norm_label, score
+
+
+

app/spotify_client.py

@@ -0,0 +1,102 @@
+from typing import Any, Dict, Optional
+
+import spotipy
+from spotipy.oauth2 import SpotifyClientCredentials
+
+from .config import get_settings
+from .spotify_http import spotify_get
+
+
+settings = get_settings()
+
+
+def _get_spotify_client() -> spotipy.Spotify:
+    if not settings.spotify_client_id or not settings.spotify_client_secret:
+        raise RuntimeError(
+            "Spotify credentials not configured. "
+            "Set SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET."
+        )
+    auth_manager = SpotifyClientCredentials(
+        client_id=settings.spotify_client_id,
+        client_secret=settings.spotify_client_secret,
+    )
+    return spotipy.Spotify(auth_manager=auth_manager)
+
+
+def get_app_access_token() -> str:
+    """Get an app-only access token (Client Credentials flow)."""
+    sp = _get_spotify_client()
+    token_info = sp.auth_manager.get_access_token(as_dict=True)
+    return token_info["access_token"]
+
+
+def get_track_via_bearer(
+    track_id: str,
+    access_token: Optional[str] = None,
+    *,
+    market: Optional[str] = None,
+) -> Dict[str, Any]:
+    """Fetch a track using the raw Web API with a bearer token."""
+    token = access_token or get_app_access_token()
+    params = {"market": market} if market else None
+    data = spotify_get(f"https://api.spotify.com/v1/tracks/{track_id}", token, params=params)
+    return {
+        "id": data.get("id"),
+        "name": data.get("name"),
+        "artists": [a.get("name") for a in data.get("artists", [])],
+        "external_url": (data.get("external_urls") or {}).get("spotify"),
+        "preview_url": data.get("preview_url"),
+        "album": (data.get("album") or {}).get("name"),
+        "is_playable": data.get("is_playable"),
+        "linked_from": data.get("linked_from"),
+        "restrictions": data.get("restrictions"),
+    }
+
+
+def emotion_to_spotify_params(
+    emotion: str, source: str = "text"
+) -> Dict[str, Any]:
+    """Map emotion to Spotify search/recommendation parameters."""
+    emotion = emotion.lower()
+
+    params: Dict[str, Any] = {
+        "seed_genres": ["pop"],
+        "target_valence": 0.5,
+        "target_energy": 0.5,
+    }
+
+    if emotion in {"joy", "happy"}:
+        params.update(seed_genres=["pop", "dance", "party"], target_valence=0.9, target_energy=0.8)
+    elif emotion in {"sad", "sadness"}:
+        params.update(seed_genres=["acoustic", "sad"], target_valence=0.2, target_energy=0.4)
+    elif emotion in {"anger", "angry"}:
+        params.update(seed_genres=["rock", "metal"], target_valence=0.3, target_energy=0.9)
+    elif emotion in {"fear"}:
+        params.update(seed_genres=["ambient", "chill"], target_valence=0.3, target_energy=0.3)
+    elif emotion in {"surprise"}:
+        params.update(seed_genres=["electronic", "indie"], target_valence=0.7, target_energy=0.7)
+    elif emotion in {"neutral"}:
+        params.update(seed_genres=["pop", "indie"], target_valence=0.5, target_energy=0.5)
+    elif emotion in {"disgust"}:
+        params.update(seed_genres=["alternative"], target_valence=0.4, target_energy=0.6)
+    return params
+
+
+def recommend_track_for_emotion(emotion: str, source: str = "text") -> Dict[str, Any]:
+    sp = _get_spotify_client()
+    params = emotion_to_spotify_params(emotion, source=source)
+    recs = sp.recommendations(limit=1, **params)
+    tracks = recs.get("tracks", [])
+    if not tracks:
+        return {}
+    track = tracks[0]
+    return {
+        "id": track["id"],
+        "name": track["name"],
+        "artists": [a["name"] for a in track.get("artists", [])],
+        "preview_url": track.get("preview_url"),
+        "external_url": track.get("external_urls", {}).get("spotify"),
+    }
+
+
+

app/spotify_http.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict, Optional
+
+import random
+import time
+import requests
+
+
+@dataclass
+class SpotifyAPIError(Exception):
+    status_code: int
+    message: str
+    retry_after: Optional[int] = None
+    raw: Optional[Dict[str, Any]] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        payload: Dict[str, Any] = {
+            "status": self.status_code,
+            "message": self.message,
+        }
+        if self.retry_after is not None:
+            payload["retry_after"] = self.retry_after
+        if self.raw is not None:
+            payload["spotify_error"] = self.raw
+        return payload
+
+
+def _parse_spotify_error(resp: requests.Response) -> SpotifyAPIError:
+    retry_after = None
+    if resp.status_code == 429:
+        ra = resp.headers.get("Retry-After")
+        if ra and ra.isdigit():
+            retry_after = int(ra)
+
+    message = resp.reason or "Spotify API error"
+    raw: Optional[Dict[str, Any]] = None
+    try:
+        raw = resp.json()
+        if isinstance(raw, dict):
+            err = raw.get("error")
+            if isinstance(err, dict):
+                message = err.get("message") or message
+    except Exception:  # noqa: BLE001
+        raw = None
+
+    return SpotifyAPIError(
+        status_code=resp.status_code,
+        message=str(message),
+        retry_after=retry_after,
+        raw=raw,
+    )
+
+
+def spotify_get(
+    url: str,
+    access_token: str,
+    *,
+    params: Optional[Dict[str, Any]] = None,
+    timeout: int = 20,
+) -> Dict[str, Any]:
+    """
+    GET wrapper with basic retry/backoff for rate limits and transient upstream errors.
+
+    Retries:
+    - 429: waits Retry-After seconds (when present) then retries
+    - 502/503: exponential backoff retries (small capped)
+    """
+    max_retries = 2
+    attempt = 0
+    while True:
+        resp = requests.get(
+            url,
+            headers={"Authorization": f"Bearer {access_token}"},
+            params=params,
+            timeout=timeout,
+        )
+        if resp.status_code < 400:
+            if resp.status_code == 204 or not resp.content:
+                return {}
+            return resp.json()
+
+        err = _parse_spotify_error(resp)
+        retryable = err.status_code in {429, 502, 503}
+        if not retryable or attempt >= max_retries:
+            raise err
+
+        attempt += 1
+        if err.status_code == 429 and err.retry_after is not None:
+            sleep_s = err.retry_after
+        else:
+            sleep_s = min(8.0, (2.0 ** (attempt - 1))) + random.uniform(0.0, 0.25)
+        time.sleep(sleep_s)
+
+
+def spotify_post(
+    url: str,
+    access_token: str,
+    *,
+    params: Optional[Dict[str, Any]] = None,
+    json: Optional[Dict[str, Any]] = None,
+    data: Optional[Dict[str, Any]] = None,
+    timeout: int = 20,
+) -> Dict[str, Any]:
+    """
+    POST wrapper with basic retry/backoff for rate limits and transient upstream errors.
+    """
+    max_retries = 2
+    attempt = 0
+    while True:
+        resp = requests.post(
+            url,
+            headers={"Authorization": f"Bearer {access_token}"},
+            params=params,
+            json=json,
+            data=data,
+            timeout=timeout,
+        )
+        if resp.status_code < 400:
+            if resp.status_code == 204 or not resp.content:
+                return {}
+            return resp.json()
+
+        err = _parse_spotify_error(resp)
+        retryable = err.status_code in {429, 502, 503}
+        if not retryable or attempt >= max_retries:
+            raise err
+
+        attempt += 1
+        if err.status_code == 429 and err.retry_after is not None:
+            sleep_s = err.retry_after
+        else:
+            sleep_s = min(8.0, (2.0 ** (attempt - 1))) + random.uniform(0.0, 0.25)
+        time.sleep(sleep_s)
+
+
+

app/spotify_ids.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Literal, Optional
+from urllib.parse import urlparse
+
+
+SpotifyResourceType = Literal["track", "album", "artist", "playlist", "user", "unknown"]
+_BASE62_ID_RE = re.compile(r"^[A-Za-z0-9]{22}$")
+_URI_RE = re.compile(r"^spotify:(?P<type>track|album|artist|playlist|user):(?P<id>[A-Za-z0-9]{1,})$")
+
+
+@dataclass(frozen=True)
+class SpotifyRef:
+    resource_type: SpotifyResourceType
+    id: Optional[str] = None
+
+    @property
+    def uri(self) -> Optional[str]:
+        if self.id and self.resource_type in {"track", "album", "artist", "playlist", "user"}:
+            return f"spotify:{self.resource_type}:{self.id}"
+        return None
+
+    @property
+    def url(self) -> Optional[str]:
+        if self.id and self.resource_type in {"track", "album", "artist", "playlist"}:
+            return f"https://open.spotify.com/{self.resource_type}/{self.id}"
+        if self.id and self.resource_type == "user":
+            return f"https://open.spotify.com/user/{self.id}"
+        return None
+
+
+def parse_spotify_ref(value: str) -> SpotifyRef:
+    """Parse a Spotify reference in various forms."""
+    v = (value or "").strip()
+    if not v:
+        return SpotifyRef(resource_type="unknown", id=None)
+
+    # spotify:<type>:<id>
+    m = _URI_RE.match(v)
+    if m:
+        return SpotifyRef(resource_type=m.group("type"), id=m.group("id"))  # type: ignore[arg-type]
+
+    # https://open.spotify.com/<type>/<id>
+    try:
+        parsed = urlparse(v)
+        host = (parsed.hostname or "").lower()
+        if host in {"open.spotify.com", "play.spotify.com"}:
+            parts = [p for p in (parsed.path or "").split("/") if p]
+            if len(parts) >= 2:
+                rtype = parts[0]
+                rid = parts[1]
+                if rtype in {"track", "album", "artist", "playlist", "user"}:
+                    return SpotifyRef(resource_type=rtype, id=rid)  # type: ignore[arg-type]
+    except Exception:  # noqa: BLE001
+        pass
+
+    if _BASE62_ID_RE.match(v):
+        return SpotifyRef(resource_type="unknown", id=v)
+
+    return SpotifyRef(resource_type="unknown", id=None)
+
+
+def to_track_uri(value: str) -> str:
+    """Normalize a track reference (URI/URL/ID) into a Spotify track URI."""
+    ref = parse_spotify_ref(value)
+    if ref.resource_type == "track" and ref.id:
+        return f"spotify:track:{ref.id}"
+    if ref.resource_type == "unknown" and ref.id:
+        return f"spotify:track:{ref.id}"
+    raise ValueError("Invalid track reference. Provide a track URI, URL, or 22-char track ID.")
+
+

app/spotify_oauth.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import base64
+import ipaddress
+import secrets
+from urllib.parse import urlparse
+from typing import Any, Dict, Optional
+
+import requests
+
+from .config import get_settings
+from .spotify_http import SpotifyAPIError
+
+
+SPOTIFY_AUTHORIZE_URL = "https://accounts.spotify.com/authorize"
+SPOTIFY_TOKEN_URL = "https://accounts.spotify.com/api/token"
+
+def validate_redirect_uri(redirect_uri: str) -> None:
+    """
+    Enforce Spotify redirect URI requirements (2025+):
+    - HTTPS required unless loopback IP literal is used
+    - 'localhost' is NOT allowed
+    - Loopback IP literals allowed: 127.0.0.1 or [::1] (HTTP permitted)
+    """
+    parsed = urlparse(redirect_uri)
+    if parsed.scheme not in {"http", "https"}:
+        raise ValueError("Redirect URI must start with http:// or https://")
+    if not parsed.netloc:
+        raise ValueError("Redirect URI must include host (and optional port)")
+
+    host = parsed.hostname or ""
+    if host.lower() == "localhost":
+        raise ValueError("Redirect URI host 'localhost' is not allowed. Use 127.0.0.1 or [::1].")
+
+    is_loopback_ip = False
+    try:
+        ip = ipaddress.ip_address(host)
+        is_loopback_ip = ip.is_loopback
+    except ValueError:
+        is_loopback_ip = False
+
+    if parsed.scheme != "https" and not is_loopback_ip:
+        raise ValueError("Redirect URI must use HTTPS unless using a loopback IP literal (127.0.0.1 or [::1]).")
+
+
+def _basic_auth_header(client_id: str, client_secret: str) -> str:
+    token = base64.b64encode(f"{client_id}:{client_secret}".encode("utf-8")).decode(
+        "utf-8"
+    )
+    return f"Basic {token}"
+
+
+def generate_state() -> str:
+    return secrets.token_urlsafe(24)
+
+
+def build_authorize_url(
+    *,
+    redirect_uri: str,
+    scope: str,
+    state: str,
+    show_dialog: bool = True,
+) -> str:
+    settings = get_settings()
+    if not settings.spotify_client_id:
+        raise RuntimeError("Missing SPOTIFY_CLIENT_ID")
+
+    validate_redirect_uri(redirect_uri)
+
+    params = {
+        "client_id": settings.spotify_client_id,
+        "response_type": "code",
+        "redirect_uri": redirect_uri,
+        "scope": scope,
+        "state": state,
+        "show_dialog": "true" if show_dialog else "false",
+    }
+    req = requests.Request("GET", SPOTIFY_AUTHORIZE_URL, params=params).prepare()
+    if not req.url:
+        raise RuntimeError("Failed to build authorize URL")
+    return req.url
+
+
+def exchange_code_for_token(
+    *,
+    code: str,
+    redirect_uri: str,
+) -> Dict[str, Any]:
+    """Exchange authorization code for access token and refresh token."""
+    settings = get_settings()
+    if not settings.spotify_client_id or not settings.spotify_client_secret:
+        raise RuntimeError("Missing SPOTIFY_CLIENT_ID / SPOTIFY_CLIENT_SECRET")
+
+    headers = {
+        "Authorization": _basic_auth_header(
+            settings.spotify_client_id, settings.spotify_client_secret
+        ),
+        "Content-Type": "application/x-www-form-urlencoded",
+    }
+    data = {
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": redirect_uri,
+    }
+
+    resp = requests.post(SPOTIFY_TOKEN_URL, headers=headers, data=data, timeout=20)
+    if resp.status_code >= 400:
+        try:
+            raw = resp.json()
+        except Exception:  # noqa: BLE001
+            raw = {"error": resp.text}
+        raise SpotifyAPIError(
+            status_code=resp.status_code,
+            message=str(raw.get("error_description") or raw.get("error") or "auth_error"),
+            raw=raw if isinstance(raw, dict) else None,
+        )
+    return resp.json()
+
+
+def refresh_access_token(*, refresh_token: str) -> Dict[str, Any]:
+    """
+    Refresh an expired access token using a refresh token (Authorization Code flow).
+    """
+    settings = get_settings()
+    if not settings.spotify_client_id or not settings.spotify_client_secret:
+        raise RuntimeError("Missing SPOTIFY_CLIENT_ID / SPOTIFY_CLIENT_SECRET")
+
+    headers = {
+        "Authorization": _basic_auth_header(
+            settings.spotify_client_id, settings.spotify_client_secret
+        ),
+        "Content-Type": "application/x-www-form-urlencoded",
+    }
+    data = {
+        "grant_type": "refresh_token",
+        "refresh_token": refresh_token,
+    }
+
+    resp = requests.post(SPOTIFY_TOKEN_URL, headers=headers, data=data, timeout=20)
+    if resp.status_code >= 400:
+        try:
+            raw = resp.json()
+        except Exception:  # noqa: BLE001
+            raw = {"error": resp.text}
+        raise SpotifyAPIError(
+            status_code=resp.status_code,
+            message=str(raw.get("error_description") or raw.get("error") or "auth_error"),
+            raw=raw if isinstance(raw, dict) else None,
+        )
+    return resp.json()
+
+
+def default_redirect_uri() -> Optional[str]:
+    """
+    Preferred redirect URI is configured in SPOTIFY_REDIRECT_URI.
+    """
+    settings = get_settings()
+    return settings.spotify_redirect_uri
+
+

app/spotify_playlists_api.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from .spotify_http import spotify_get, spotify_post
+
+
+def list_current_user_playlists(
+    access_token: str, *, limit: int = 20, offset: int = 0
+) -> Dict[str, Any]:
+    return spotify_get(
+        "https://api.spotify.com/v1/me/playlists",
+        access_token,
+        params={"limit": limit, "offset": offset},
+    )
+
+
+def get_playlist(access_token: str, playlist_id: str) -> Dict[str, Any]:
+    return spotify_get(f"https://api.spotify.com/v1/playlists/{playlist_id}", access_token)
+
+
+def get_playlist_items(
+    access_token: str,
+    playlist_id: str,
+    *,
+    limit: int = 50,
+    offset: int = 0,
+    additional_types: str = "track,episode",
+    market: Optional[str] = None,
+) -> Dict[str, Any]:
+    params: Dict[str, Any] = {
+        "limit": limit,
+        "offset": offset,
+        "additional_types": additional_types,
+    }
+    if market:
+        params["market"] = market
+    return spotify_get(
+        f"https://api.spotify.com/v1/playlists/{playlist_id}/tracks",
+        access_token,
+        params=params,
+    )
+
+
+def create_playlist(
+    access_token: str,
+    user_id: str,
+    *,
+    name: str,
+    description: Optional[str] = None,
+    public: bool = True,
+    collaborative: bool = False,
+) -> Dict[str, Any]:
+    if collaborative and public:
+        raise ValueError("Spotify rule: collaborative playlists cannot be public. Set public=false.")
+
+    payload: Dict[str, Any] = {
+        "name": name,
+        "public": public,
+        "collaborative": collaborative,
+    }
+    if description is not None:
+        payload["description"] = description
+
+    return spotify_post(
+        f"https://api.spotify.com/v1/users/{user_id}/playlists",
+        access_token,
+        json=payload,
+    )
+
+
+def add_items_to_playlist(
+    access_token: str,
+    playlist_id: str,
+    *,
+    uris: List[str],
+    position: Optional[int] = None,
+) -> Dict[str, Any]:
+    payload: Dict[str, Any] = {"uris": uris}
+    if position is not None:
+        payload["position"] = position
+    return spotify_post(
+        f"https://api.spotify.com/v1/playlists/{playlist_id}/tracks",
+        access_token,
+        json=payload,
+    )
+
+

app/spotify_user_api.py

@@ -0,0 +1,14 @@
+from typing import Any, Dict
+
+from .spotify_http import spotify_get
+
+
+def get_current_user_profile(access_token: str) -> Dict[str, Any]:
+    """
+    Call Spotify Web API GET /v1/me using a *user* access token.
+    This requires Authorization Code flow (Client Credentials will NOT work).
+    """
+    return spotify_get("https://api.spotify.com/v1/me", access_token)
+
+
+

requirements.txt

@@ -0,0 +1,12 @@
+fastapi
+uvicorn[standard]
+SQLAlchemy
+psycopg2-binary
+python-dotenv
+requests
+spotipy
+transformers
+torch
+Pillow
+opencv-python
+numpy