feat(backend,frontend): copy liked songs into a shareable playlist

backend/src/api/app.py

@@ -5,11 +5,13 @@ from pathlib import Path
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse
+from starlette.middleware.sessions import SessionMiddleware
 
 from api import routers
 from utils.env import env
 from utils.logging import setup_logging
 
+SESSION_DAYS = 30
 IMMUTABLE_HEADERS = {"Cache-Control": "public, max-age=31536000, immutable"}
 NO_CACHE_HEADERS = {"Cache-Control": "no-cache"}
 
@@ -30,6 +32,13 @@ app.add_middleware(
     CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]
 )
 
+app.add_middleware(
+    SessionMiddleware,
+    secret_key=env.session.secret.get_secret_value(),
+    max_age=SESSION_DAYS * 86400,
+    https_only=env.public_url.startswith("https"),
+)
+
 app.include_router(routers.router)
 
 spa_dir = Path(env.api.static_dir).resolve()

backend/src/api/modules/copier.py

@@ -0,0 +1,152 @@
+import asyncio
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from typing import Any
+
+from utils.logging import logger
+
+from .spotify import Spotify
+
+PAGE = 50
+CHUNK = 100
+LIMIT = 10_000
+PARALLEL = 4
+WINDOW = 200
+ROWS = 30
+COLS = 60
+
+
+def moment(item: dict[str, Any]) -> float:
+    return datetime.fromisoformat(item["added_at"]).timestamp()
+
+
+@dataclass
+class Job:
+    total: int = 0
+    progress: int = 0
+    tracks: list[dict[str, Any]] = field(default_factory=list)
+    bins: list[list[int]] = field(
+        default_factory=lambda: [[0] * COLS for _ in range(ROWS)]
+    )
+    row: int = 0
+    newest: float = 0
+    oldest: float = 0
+    playlist: dict[str, str] | None = None
+    state: str = "running"
+    task: asyncio.Task[None] | None = None
+
+    def place(self, item: dict[str, Any]) -> None:
+        span = max(self.newest - self.oldest, 1)
+        share = min(max((self.newest - moment(item)) / span, 0), 1)
+        self.row, offset = divmod(min(int(share * ROWS * COLS), ROWS * COLS - 1), COLS)
+        self.bins[self.row][COLS - 1 - offset] += 1
+
+    def view(self, since: int = 0) -> dict[str, Any]:
+        since = max(since, len(self.tracks) - WINDOW)
+        return {
+            "state": self.state,
+            "total": self.total,
+            "progress": self.progress,
+            "added": len(self.tracks),
+            "since": since,
+            "tracks": self.tracks[since:],
+            "bins": self.bins,
+            "row": self.row,
+            "playlist": self.playlist if self.state == "done" else None,
+        }
+
+
+jobs: dict[str, Job] = {}
+
+
+def playlist_name() -> str:
+    today = datetime.now(UTC)
+    return f"Liked · {today:%b} {today.day}, {today.year}"
+
+
+def card(track: dict[str, Any]) -> dict[str, Any]:
+    images = track["album"]["images"]
+    return {
+        "title": track["name"],
+        "artists": ", ".join(artist["name"] for artist in track["artists"]),
+        "cover": images[-1]["url"] if images else None,
+        "url": track["external_urls"].get("spotify"),
+    }
+
+
+def start(user_id: str, access_token: str) -> Job:
+    job = jobs.get(user_id)
+    if job and job.state == "running":
+        return job
+    job = Job()
+    job.task = asyncio.create_task(run(job, access_token))
+    jobs[user_id] = job
+    return job
+
+
+async def run(job: Job, access_token: str) -> None:
+    try:
+        async with Spotify(access_token) as spotify:
+            await copy(job, spotify)
+    except Exception:
+        logger.exception("copy failed")
+        job.state = "error"
+    else:
+        job.state = "done"
+
+
+async def copy(job: Job, spotify: Spotify) -> None:
+    head = await spotify.call("GET", "/me/tracks", params={"limit": 1})
+    job.total = min(head["total"], LIMIT)
+    if not job.total:
+        return
+    tail = await spotify.call(
+        "GET", "/me/tracks", params={"limit": 1, "offset": job.total - 1}
+    )
+    job.newest = moment(head["items"][0])
+    job.oldest = moment((tail["items"] or head["items"])[0])
+    playlist = await spotify.call(
+        "POST", "/me/playlists", json={"name": playlist_name(), "public": True}
+    )
+    gate = asyncio.Semaphore(PARALLEL)
+
+    async def fetch(offset: int) -> list[dict[str, Any]]:
+        async with gate:
+            page = await spotify.call(
+                "GET", "/me/tracks", params={"limit": PAGE, "offset": offset}
+            )
+        return page["items"]
+
+    async def add(items: list[dict[str, Any]]) -> None:
+        playable = [
+            item
+            for item in items
+            if item["track"] and not item["track"].get("is_local")
+        ]
+        if playable:
+            await spotify.call(
+                "POST",
+                f"/playlists/{playlist['id']}/items",
+                json={"uris": [item["track"]["uri"] for item in playable]},
+            )
+        for item in playable:
+            job.place(item)
+            job.tracks.append(card(item["track"]))
+        job.progress += len(items)
+
+    async with asyncio.TaskGroup() as group:
+        pages = [
+            group.create_task(fetch(offset)) for offset in range(0, job.total, PAGE)
+        ]
+        buffer: list[dict[str, Any]] = []
+        for page in pages:
+            buffer += await page
+            while len(buffer) >= CHUNK:
+                await add(buffer[:CHUNK])
+                buffer = buffer[CHUNK:]
+        await add(buffer)
+
+    job.playlist = {
+        "name": playlist["name"],
+        "url": playlist["external_urls"]["spotify"],
+    }

backend/src/api/modules/session.py

@@ -0,0 +1,30 @@
+from typing import Annotated
+
+import httpx
+from fastapi import Depends, HTTPException, Request, status
+
+from . import spotify
+
+
+async def access_token(request: Request) -> str:
+    token = request.session.get("token")
+    if not token:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED)
+    try:
+        token = await spotify.fresh_token(token)
+    except httpx.HTTPStatusError as error:
+        request.session.clear()
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED) from error
+    request.session["token"] = token
+    return token["access"]
+
+
+def user_id(request: Request) -> str:
+    user = request.session.get("user")
+    if not user:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED)
+    return user["id"]
+
+
+AccessToken = Annotated[str, Depends(access_token)]
+UserId = Annotated[str, Depends(user_id)]

backend/src/api/modules/spotify.py

@@ -0,0 +1,96 @@
+import asyncio
+import time
+from types import TracebackType
+from typing import Any, Self
+from urllib.parse import urlencode
+
+import httpx
+
+from utils.env import env
+
+ACCOUNTS = "https://accounts.spotify.com"
+API = "https://api.spotify.com/v1"
+SCOPES = "user-library-read playlist-modify-public playlist-modify-private"
+TOKEN_MARGIN = 600
+ATTEMPTS = 5
+MAX_WAIT = 30
+
+
+def redirect_uri() -> str:
+    return f"{env.public_url}/api/callback"
+
+
+def authorize_url(state: str, *, show_dialog: bool) -> str:
+    query = urlencode(
+        {
+            "client_id": env.spotify.client_id,
+            "response_type": "code",
+            "redirect_uri": redirect_uri(),
+            "scope": SCOPES,
+            "state": state,
+            "show_dialog": str(show_dialog).lower(),
+        }
+    )
+    return f"{ACCOUNTS}/authorize?{query}"
+
+
+async def request_token(data: dict[str, str]) -> dict[str, Any]:
+    auth = (env.spotify.client_id, env.spotify.client_secret.get_secret_value())
+    async with httpx.AsyncClient(timeout=30) as client:
+        response = await client.post(f"{ACCOUNTS}/api/token", data=data, auth=auth)
+    response.raise_for_status()
+    token = response.json()
+    return {
+        "access": token["access_token"],
+        "refresh": token.get("refresh_token") or data.get("refresh_token"),
+        "expires": time.time() + token["expires_in"],
+    }
+
+
+async def exchange_code(code: str) -> dict[str, Any]:
+    return await request_token(
+        {
+            "grant_type": "authorization_code",
+            "code": code,
+            "redirect_uri": redirect_uri(),
+        }
+    )
+
+
+async def fresh_token(token: dict[str, Any]) -> dict[str, Any]:
+    if token["expires"] - time.time() > TOKEN_MARGIN:
+        return token
+    return await request_token(
+        {"grant_type": "refresh_token", "refresh_token": token["refresh"]}
+    )
+
+
+class Spotify:
+    def __init__(self, access_token: str) -> None:
+        self.http = httpx.AsyncClient(
+            base_url=API,
+            headers={"Authorization": f"Bearer {access_token}"},
+            timeout=30,
+        )
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        await self.http.aclose()
+
+    async def call(self, method: str, path: str, **kwargs: Any) -> Any:  # noqa: ANN401
+        for attempt in range(ATTEMPTS):
+            response = await self.http.request(method, path, **kwargs)
+            retryable = response.status_code == httpx.codes.TOO_MANY_REQUESTS
+            if not retryable and not response.is_server_error:
+                break
+            wait = float(response.headers.get("retry-after", 2**attempt))
+            await asyncio.sleep(min(wait, MAX_WAIT))
+        response.raise_for_status()
+        return response.json()

backend/src/api/routers/api/__init__.py

@@ -1,6 +1,8 @@
 from fastapi import APIRouter
 
-from . import health
+from . import auth, copy, health
 
 router = APIRouter()
 router.include_router(health.router, prefix="/health")
+router.include_router(auth.router)
+router.include_router(copy.router, prefix="/copy")

backend/src/api/routers/api/auth.py

@@ -0,0 +1,61 @@
+import secrets
+from typing import Any
+
+import httpx
+from fastapi import APIRouter, Request, Response, status
+from fastapi.responses import RedirectResponse
+
+from api.modules import copier, spotify
+from api.modules.session import AccessToken, UserId
+
+router = APIRouter(tags=["auth"])
+
+
+@router.get("/login")
+async def login(request: Request, *, switch: bool = False) -> RedirectResponse:
+    state = secrets.token_urlsafe(16)
+    request.session["state"] = state
+    return RedirectResponse(spotify.authorize_url(state, show_dialog=switch))
+
+
+@router.get("/callback")
+async def callback(
+    request: Request, code: str | None = None, state: str | None = None
+) -> RedirectResponse:
+    expected = request.session.pop("state", None)
+    if not code or not state or state != expected:
+        return RedirectResponse("/")
+    token = await spotify.exchange_code(code)
+    try:
+        async with spotify.Spotify(token["access"]) as client:
+            me = await client.call("GET", "/me")
+    except httpx.HTTPStatusError as error:
+        if error.response.status_code != status.HTTP_403_FORBIDDEN:
+            raise
+        return RedirectResponse("/?denied")
+    images = me.get("images") or []
+    request.session["token"] = token
+    request.session["user"] = {
+        "id": me["id"],
+        "name": me.get("display_name") or me["id"],
+        "avatar": images[0]["url"] if images else None,
+    }
+    return RedirectResponse("/")
+
+
+@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
+async def logout(request: Request) -> Response:
+    request.session.clear()
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
+@router.get("/me")
+async def me(request: Request, token: AccessToken, uid: UserId) -> dict[str, Any]:
+    async with spotify.Spotify(token) as client:
+        liked = await client.call("GET", "/me/tracks", params={"limit": 1})
+    job = copier.jobs.get(uid)
+    return {
+        **request.session["user"],
+        "liked": liked["total"],
+        "job": job.view() if job else None,
+    }

backend/src/api/routers/api/copy.py

@@ -0,0 +1,21 @@
+from typing import Any
+
+from fastapi import APIRouter, HTTPException, status
+
+from api.modules import copier
+from api.modules.session import AccessToken, UserId
+
+router = APIRouter(tags=["copy"])
+
+
+@router.post("")
+async def start(token: AccessToken, uid: UserId) -> dict[str, Any]:
+    return copier.start(uid, token).view()
+
+
+@router.get("")
+async def progress(uid: UserId, since: int = 0) -> dict[str, Any]:
+    job = copier.jobs.get(uid)
+    if not job:
+        raise HTTPException(status.HTTP_404_NOT_FOUND)
+    return job.view(since)

backend/src/utils/env.py

@@ -1,4 +1,4 @@
-from pydantic import Field
+from pydantic import Field, SecretStr
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -22,9 +22,21 @@ class ApiSettings(Section):
     static_dir: str = "static"
 
 
+class SpotifySettings(Section):
+    client_id: str = ""
+    client_secret: SecretStr = SecretStr("")
+
+
+class SessionSettings(Section):
+    secret: SecretStr = SecretStr("dev")
+
+
 class Settings(BaseSettings):
     log: LogSettings = Field(default_factory=LogSettings)
     api: ApiSettings = Field(default_factory=ApiSettings)
+    spotify: SpotifySettings = Field(default_factory=SpotifySettings)
+    session: SessionSettings = Field(default_factory=SessionSettings)
+    public_url: str = "http://127.0.0.1:5173"
 
     model_config = SettingsConfigDict(
         case_sensitive=False,