push oauth fix

app/main.py

@@ -19,9 +19,10 @@ from .spotify_oauth import (
     build_authorize_url,
     default_redirect_uri,
     exchange_code_for_token,
+    exchange_code_for_token_pkce,
     generate_state,
     validate_redirect_uri,
- )
+)
 from .spotify_playlists_api import (
     add_items_to_playlist,
     create_playlist,
@@ -66,6 +67,12 @@ class RecommendationResponse(BaseModel):
     playlist_external_url: str | None = None
 
 
+class SpotifyMobileTokenRequest(BaseModel):
+    code: str
+    code_verifier: str
+    redirect_uri: str
+
+
 def _extract_bearer_token(authorization: str | None) -> str:
     if not authorization:
         raise HTTPException(
@@ -299,6 +306,23 @@ def spotify_me(authorization: str | None = Header(default=None)) -> dict:
         raise HTTPException(status_code=500, detail="Unexpected server error") from exc
 
 
+@app.post("/mobile/spotify/token")
+def spotify_mobile_token(body: SpotifyMobileTokenRequest) -> dict:
+    try:
+        return exchange_code_for_token_pkce(
+            code=body.code,
+            redirect_uri=body.redirect_uri,
+            code_verifier=body.code_verifier,
+        )
+    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/artists/id/{artist_id}")
 def spotify_get_artist(
     artist_id: str,

app/spotify_oauth.py

@@ -86,7 +86,6 @@ 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")
@@ -117,6 +116,43 @@ def exchange_code_for_token(
     return resp.json()
 
 
+def exchange_code_for_token_pkce(
+    *,
+    code: str,
+    redirect_uri: str,
+    code_verifier: str,
+) -> Dict[str, Any]:
+    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,
+        "code_verifier": code_verifier,
+    }
+
+    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).