cottongin/NtR-soudcloud-fetcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add user resolution and likes fetching with 401 retry

Browse Source 
Made-with: Cursor
This commit is contained in:
cottongin
2026-03-12 01:36:28 -04:00
parent 49846f9d7e
commit f7dde6781a
2 changed files with 202 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
src/ntr_fetcher/soundcloud.py
View File

@@ -50,5 +50,103 @@ class SoundCloudClient:
def invalidate_client_id(self) -> None:
self._client_id = None
async def _api_get(self, url: str, params: dict | None = None) -> httpx.Response:
client_id = await self._extract_client_id()
params = dict(params or {})
params["client_id"] = client_id
for attempt in range(3):
resp = await self._http.get(url, params=params)
if resp.status_code == 401:
logger.warning("Got 401 from SoundCloud API, refreshing client_id (attempt %d)", attempt + 1)
self.invalidate_client_id()
client_id = await self._extract_client_id()
params["client_id"] = client_id
continue
resp.raise_for_status()
return resp
raise httpx.HTTPStatusError(
"Failed after 3 attempts (401)",
request=resp.request,
response=resp,
)
async def resolve_user(self, username: str) -> int:
resp = await self._api_get(
f"{API_BASE}/resolve",
params={"url": f"{SOUNDCLOUD_BASE}/{username}"},
)
return resp.json()["id"]
async def fetch_likes(
self,
user_id: int,
since: datetime,
until: datetime,
limit: int = 50,
) -> list[Track]:
cursor = _build_cursor(until, user_id)
collected: list[Track] = []
while True:
params: dict = {"limit": limit}
if cursor:
params["offset"] = cursor
resp = await self._api_get(f"{API_BASE}/users/{user_id}/likes", params=params)
data = resp.json()
collection = data.get("collection", [])
if not collection:
break
stop = False
for item in collection:
liked_at_str = item.get("created_at", "")
liked_at = datetime.fromisoformat(liked_at_str.replace("Z", "+00:00"))
if liked_at < since:
stop = True
break
if liked_at > until:
continue
track_data = item.get("track")
if track_data is None:
continue
user_data = track_data.get("user", {})
collected.append(
Track(
id=track_data["id"],
title=track_data["title"],
artist=user_data.get("username", "Unknown"),
permalink_url=track_data["permalink_url"],
artwork_url=track_data.get("artwork_url"),
duration_ms=track_data.get("full_duration", track_data.get("duration", 0)),
license=track_data.get("license", ""),
liked_at=liked_at,
raw_json=json.dumps(track_data),
)
)
if stop:
break
next_href = data.get("next_href")
if not next_href:
break
parsed = urlparse(next_href)
qs = parse_qs(parsed.query)
cursor = qs.get("offset", [None])[0]
if cursor is None:
break
collected.sort(key=lambda t: t.liked_at)
return collected
async def close(self) -> None:
await self._http.aclose()

105
tests/test_soundcloud.py
View File

@@ -1,7 +1,7 @@
import re
from datetime import datetime, timezone
import pytest
import httpx
from ntr_fetcher.soundcloud import SoundCloudClient
@@ -48,3 +48,106 @@ async def test_extract_client_id_bad_html(httpx_mock):
client = SoundCloudClient()
with pytest.raises(ValueError, match="client_id"):
await client._extract_client_id()
FAKE_RESOLVE_RESPONSE = {"id": 206979918, "kind": "user", "username": "NICKtheRAT"}
FAKE_LIKES_RESPONSE = {
"collection": [
{
"created_at": "2026-03-09T02:25:43Z",
"kind": "like",
"track": {
"id": 12345,
"title": "Test Track",
"permalink_url": "https://soundcloud.com/artist/test-track",
"duration": 180000,
"full_duration": 180000,
"genre": "Electronic",
"tag_list": "",
"created_at": "2026-03-01T00:00:00Z",
"description": "",
"artwork_url": "https://i1.sndcdn.com/artworks-abc-large.jpg",
"license": "cc-by",
"user": {
"id": 999,
"username": "TestArtist",
"permalink_url": "https://soundcloud.com/testartist",
},
"media": {"transcodings": []},
},
}
],
"next_href": None,
}
@pytest.mark.asyncio
async def test_resolve_user(httpx_mock):
httpx_mock.add_response(url="https://soundcloud.com", text=FAKE_HTML)
httpx_mock.add_response(
url=re.compile(r"https://api-v2\.soundcloud\.com/resolve.*"),
json=FAKE_RESOLVE_RESPONSE,
)
client = SoundCloudClient()
user_id = await client.resolve_user("nicktherat")
assert user_id == 206979918
@pytest.mark.asyncio
async def test_fetch_likes(httpx_mock):
httpx_mock.add_response(url="https://soundcloud.com", text=FAKE_HTML)
httpx_mock.add_response(
url=re.compile(r"https://api-v2\.soundcloud\.com/users/206979918/likes.*"),
json=FAKE_LIKES_RESPONSE,
)
client = SoundCloudClient()
tracks = await client.fetch_likes(
user_id=206979918,
since=datetime(2026, 3, 1, 0, 0, 0, tzinfo=timezone.utc),
until=datetime(2026, 3, 10, 0, 0, 0, tzinfo=timezone.utc),
)
assert len(tracks) == 1
assert tracks[0].title == "Test Track"
assert tracks[0].artist == "TestArtist"
assert tracks[0].id == 12345
@pytest.mark.asyncio
async def test_fetch_likes_filters_outside_range(httpx_mock):
httpx_mock.add_response(url="https://soundcloud.com", text=FAKE_HTML)
httpx_mock.add_response(
url=re.compile(r"https://api-v2\.soundcloud\.com/users/206979918/likes.*"),
json=FAKE_LIKES_RESPONSE,
)
client = SoundCloudClient()
tracks = await client.fetch_likes(
user_id=206979918,
since=datetime(2026, 3, 10, 0, 0, 0, tzinfo=timezone.utc),
until=datetime(2026, 3, 12, 0, 0, 0, tzinfo=timezone.utc),
)
assert len(tracks) == 0
@pytest.mark.asyncio
async def test_fetch_likes_retries_on_401(httpx_mock):
httpx_mock.add_response(url="https://soundcloud.com", text=FAKE_HTML)
httpx_mock.add_response(
url=re.compile(r"https://api-v2\.soundcloud\.com/users/206979918/likes.*"),
status_code=401,
)
httpx_mock.add_response(
url="https://soundcloud.com",
text=FAKE_HTML.replace("test_client_id_abc123", "new_client_id_456"),
)
httpx_mock.add_response(
url=re.compile(r"https://api-v2\.soundcloud\.com/users/206979918/likes.*"),
json=FAKE_LIKES_RESPONSE,
)
client = SoundCloudClient()
tracks = await client.fetch_likes(
user_id=206979918,
since=datetime(2026, 3, 1, 0, 0, 0, tzinfo=timezone.utc),
until=datetime(2026, 3, 10, 0, 0, 0, tzinfo=timezone.utc),
)
assert len(tracks) == 1