quickstream

commit

f8144c742a94389a1ff02c66951e80bb2eef2628

parent

55a628c8a8c4bf4ce6f64dc4af78e2be374715a5

Author

Tobias Bengfort <tobias.bengfort@posteo.de>

Date

2026-03-27 06:50

init

Diffstat

6 files changed, 168 insertions, 0 deletions

diff --git a/README.md b/README.md

@@ -0,0 +1,26 @@
   -1     1 # quickstream
   -1     2 
   -1     3 Find stream URIs for common streaming services.
   -1     4 
   -1     5 This project is similar in spirit to [yt-dlp](https://github.com/yt-dlp/yt-dlp), with a few major differences:
   -1     6 
   -1     7 -   focus on audio rather than video
   -1     8 -   focus on streaming rather than downloading
   -1     9 -   focus on finding a usable URI quickly rather than finding the best one
   -1    10 -   modern, clean code
   -1    11 -   much smaller set of supported sites (due to a much smaller community)
   -1    12 
   -1    13 ## usage
   -1    14 
   -1    15 ```
   -1    16 >>> import quickstream
   -1    17 >>> await quickstream.extract('http://youtube-dl.bandcamp.com/track/youtube-dl-test-song')
   -1    18 {
   -1    19     'id': 1812978515,
   -1    20     'url': 'https://youtube-dl.bandcamp.com/track/youtube-dl-test-song',
   -1    21     'artist': 'youtube-dl  "\'/\\ä↭',
   -1    22     'title': 'youtube-dl  "\'/\\ä↭ - youtube-dl test song "\'/\\ä↭',
   -1    23     'duration': 9.8485,
   -1    24     'stream': 'https://t4.bcbits.com/stream/de52650df97feb66af7cdb75ab0e20fa/mp3-128/1812978515?p=0&ts=1774681098&t=18489d2f73b58b6e5dbb97b30327531ae00776eb&token=1774681098_b56e7617d9e5f3c784567aa84a36ed7d077ccde8',
   -1    25 }
   -1    26 ```

diff --git a/pyproject.toml b/pyproject.toml

@@ -0,0 +1,23 @@
   -1     1 [build-system]
   -1     2 requires = ["setuptools"]
   -1     3 build-backend = "setuptools.build_meta"
   -1     4 
   -1     5 [project]
   -1     6 name = "quickstream"
   -1     7 version = "0.0.0"
   -1     8 description = "Find stream URIs for common streaming services."
   -1     9 readme = "README.md"
   -1    10 license = "MIT"
   -1    11 authors = [
   -1    12     {name = "Tobias Bengfort", email = "tobias.bengfort@posteo.de"}
   -1    13 ]
   -1    14 dependencies = [
   -1    15     "aiohttp",
   -1    16     "beautifulsoup4",
   -1    17 ]
   -1    18 
   -1    19 [project.urls]
   -1    20 Homepage = "https://github.com/xi/quickstream"
   -1    21 
   -1    22 [project.scripts]
   -1    23 quickstream = "quickstream:main"

diff --git a/quickstream/__init__.py b/quickstream/__init__.py

@@ -0,0 +1,41 @@
   -1     1 import asyncio
   -1     2 import json
   -1     3 import sys
   -1     4 
   -1     5 import aiohttp
   -1     6 from bs4 import BeautifulSoup
   -1     7 
   -1     8 from .base import registry
   -1     9 from .providers import bandcamp  # noqa
   -1    10 from .providers import soundcloud  # noqa
   -1    11 
   -1    12 
   -1    13 class Client:
   -1    14     def __init__(self, session):
   -1    15         self.session = session
   -1    16 
   -1    17     async def fetch(self, url, **kwargs):
   -1    18         r = await self.session.get(url, **kwargs)
   -1    19         return await r.read()
   -1    20 
   -1    21     async def fetch_html(self, url, **kwargs):
   -1    22         html = await self.fetch(url, **kwargs)
   -1    23         return BeautifulSoup(html, 'html.parser')
   -1    24 
   -1    25     async def fetch_json(self, url, **kwargs):
   -1    26         r = await self.session.get(url, **kwargs)
   -1    27         return await r.json()
   -1    28 
   -1    29 
   -1    30 async def extract(url):
   -1    31     for pattern, fn in registry:
   -1    32         m = pattern.match(url)
   -1    33         if not m:
   -1    34             continue
   -1    35         async with aiohttp.ClientSession(raise_for_status=True) as session:
   -1    36             return await fn(Client(session), url, *m.groups())
   -1    37 
   -1    38 
   -1    39 def main():
   -1    40     data = asyncio.run(extract(sys.argv[1]))
   -1    41     print(json.dumps(data, indent=2))

diff --git a/quickstream/base.py b/quickstream/base.py

@@ -0,0 +1,11 @@
   -1     1 import re
   -1     2 
   -1     3 registry = []
   -1     4 
   -1     5 
   -1     6 def provider(pattern):
   -1     7     def decorator(fn):
   -1     8         registry.append((re.compile(pattern), fn))
   -1     9         return fn
   -1    10 
   -1    11     return decorator

diff --git a/quickstream/providers/bandcamp.py b/quickstream/providers/bandcamp.py

@@ -0,0 +1,24 @@
   -1     1 import json
   -1     2 
   -1     3 from ..base import provider
   -1     4 
   -1     5 
   -1     6 def get_stream(trackinfo):
   -1     7     for _fmt, url in trackinfo['file'].items():
   -1     8         return url
   -1     9 
   -1    10 
   -1    11 @provider(r'https?://([^/]+)\.bandcamp\.com/track/([^/?#&]+)')
   -1    12 async def bandcamp_track(client, url, uploader, id):
   -1    13     soup = await client.fetch_html(url)
   -1    14     el = soup.select_one('[data-tralbum]')
   -1    15     tralbum = json.loads(el['data-tralbum'])
   -1    16     trackinfo = tralbum['trackinfo'][0]
   -1    17     return {
   -1    18         'id': trackinfo['id'],
   -1    19         'url': tralbum['url'],
   -1    20         'artist': trackinfo['artist'] or tralbum['artist'],
   -1    21         'title': trackinfo['title'],
   -1    22         'duration': trackinfo['duration'],
   -1    23         'stream': get_stream(trackinfo),
   -1    24     }

diff --git a/quickstream/providers/soundcloud.py b/quickstream/providers/soundcloud.py

@@ -0,0 +1,43 @@
   -1     1 import json
   -1     2 
   -1     3 from ..base import provider
   -1     4 
   -1     5 
   -1     6 def get_trackinfo(soup):
   -1     7     extra = {}
   -1     8     for el in soup.select('script'):
   -1     9         if el.text.startswith('window.__sc_hydration'):
   -1    10             hydration = json.loads(el.text[24:-1])
   -1    11             for item in hydration:
   -1    12                 if item['hydratable'] == 'apiClient':
   -1    13                     extra['_client_id'] = item['data']['id']
   -1    14                 if item['hydratable'] == 'sound':
   -1    15                     return {**item['data'], **extra}
   -1    16 
   -1    17 
   -1    18 async def get_streams(client, trackinfo):
   -1    19     for item in trackinfo['media']['transcodings']:
   -1    20         if item['snipped'] or '/preview/' in item['url']:
   -1    21             continue
   -1    22         if item['format']['protocol'] not in ['hls', 'progressive']:
   -1    23             continue
   -1    24 
   -1    25         streaminfo = await client.fetch_json(item['url'], params={
   -1    26             'client_id': trackinfo['_client_id'],
   -1    27             'track_authorization': trackinfo['track_authorization'],
   -1    28         })
   -1    29         return streaminfo['url']
   -1    30 
   -1    31 
   -1    32 @provider(r'https?://soundcloud.com/([^/]+)/([^/]+)')
   -1    33 async def soundcloud_track(client, url, uploader, id):
   -1    34     soup = await client.fetch_html(url)
   -1    35     trackinfo = get_trackinfo(soup)
   -1    36     return {
   -1    37         'id': trackinfo['id'],
   -1    38         'url': trackinfo['permalink_url'],
   -1    39         'title': trackinfo['title'],
   -1    40         'artist': trackinfo['publisher_metadata']['artist'],
   -1    41         'duration': trackinfo['duration'] / 1000,
   -1    42         'stream': await get_streams(client, trackinfo),
   -1    43     }