0
# Utilities
1
2
Streamlink provides a comprehensive set of utility functions for URL manipulation, parsing, caching, and system integration. These utilities are commonly used in plugin development and stream processing.
3
4
## Capabilities
5
6
### URL Utilities
7
8
Functions for manipulating and working with URLs in plugin development.
9
10
```python { .api }
11
def absolute_url(base, url):
12
"""
13
Convert relative URL to absolute using base URL.
14
15
Parameters:
16
- base: Base URL for resolution
17
- url: Relative or absolute URL
18
19
Returns:
20
Absolute URL string
21
"""
22
23
def prepend_www(url):
24
"""
25
Add www prefix to URL if not present.
26
27
Parameters:
28
- url: URL to modify
29
30
Returns:
31
URL with www prefix added if needed
32
"""
33
34
def update_qsd(url, **params):
35
"""
36
Update URL query string parameters.
37
38
Parameters:
39
- url: URL to modify
40
- **params: Query parameters to add/update
41
42
Returns:
43
URL with updated query string
44
"""
45
46
def update_scheme(scheme, url, force=False):
47
"""
48
Update URL scheme (protocol).
49
50
Parameters:
51
- scheme: New scheme (e.g., 'https://')
52
- url: URL to modify
53
- force: Force scheme change even if URL already has scheme
54
55
Returns:
56
URL with updated scheme
57
"""
58
59
def url_concat(base, *urls):
60
"""
61
Concatenate URL segments properly handling slashes.
62
63
Parameters:
64
- base: Base URL
65
- *urls: URL segments to append
66
67
Returns:
68
Concatenated URL
69
"""
70
71
def url_equal(a, b):
72
"""
73
Compare URLs for equality ignoring insignificant differences.
74
75
Parameters:
76
- a: First URL
77
- b: Second URL
78
79
Returns:
80
True if URLs are equivalent
81
"""
82
```
83
84
### Parsing Utilities
85
86
Functions for parsing different data formats commonly encountered in streaming services.
87
88
```python { .api }
89
def parse_html(data, **kwargs):
90
"""
91
Parse HTML content using lxml.
92
93
Parameters:
94
- data: HTML string or bytes
95
- **kwargs: Additional lxml.html parsing options
96
97
Returns:
98
Parsed HTML element tree
99
"""
100
101
def parse_json(data, **kwargs):
102
"""
103
Parse JSON data with error handling.
104
105
Parameters:
106
- data: JSON string
107
- **kwargs: Additional json.loads() options
108
109
Returns:
110
Parsed JSON object (dict, list, etc.)
111
"""
112
113
def parse_qsd(data, **kwargs):
114
"""
115
Parse query string data into dictionary.
116
117
Parameters:
118
- data: Query string (can include leading '?')
119
- **kwargs: Additional parsing options
120
121
Returns:
122
Dictionary of query parameters
123
"""
124
125
def parse_xml(data, **kwargs):
126
"""
127
Parse XML content using lxml.
128
129
Parameters:
130
- data: XML string or bytes
131
- **kwargs: Additional lxml.etree parsing options
132
133
Returns:
134
Parsed XML element tree
135
"""
136
```
137
138
### Data Utilities
139
140
Functions for searching and manipulating data structures.
141
142
```python { .api }
143
def search_dict(obj, key):
144
"""
145
Recursively search nested dictionaries and lists for key.
146
147
Parameters:
148
- obj: Dictionary, list, or other object to search
149
- key: Key to search for
150
151
Returns:
152
First value found for key, or None if not found
153
"""
154
```
155
156
### Caching Utilities
157
158
LRU (Least Recently Used) cache implementation for efficient data caching.
159
160
```python { .api }
161
class LRUCache:
162
"""
163
Least Recently Used cache with maximum size limit.
164
165
Automatically evicts least recently used items when
166
maximum capacity is reached.
167
"""
168
169
def __init__(self, maxsize=128):
170
"""
171
Initialize LRU cache.
172
173
Parameters:
174
- maxsize: Maximum number of items to cache
175
"""
176
177
def get(self, key, default=None):
178
"""
179
Get cached value, updating access time.
180
181
Parameters:
182
- key: Cache key
183
- default: Default value if key not found
184
185
Returns:
186
Cached value or default
187
"""
188
189
def set(self, key, value):
190
"""
191
Set cached value.
192
193
Parameters:
194
- key: Cache key
195
- value: Value to cache
196
"""
197
198
def clear(self):
199
"""Clear all cached items"""
200
201
def size(self):
202
"""Get current cache size"""
203
```
204
205
### System Utilities
206
207
System integration utilities for cross-platform functionality.
208
209
```python { .api }
210
class NamedPipe:
211
"""
212
Cross-platform named pipe implementation for streaming data.
213
214
Provides a file-like interface for creating named pipes
215
that can be used to stream data to external applications.
216
"""
217
218
def __init__(self):
219
"""Initialize named pipe"""
220
221
def open(self, mode='wb'):
222
"""
223
Open named pipe for reading or writing.
224
225
Parameters:
226
- mode: File mode ('rb', 'wb', etc.)
227
228
Returns:
229
File-like object for pipe access
230
"""
231
232
@property
233
def path(self):
234
"""Path to the named pipe"""
235
236
def load_module(path):
237
"""
238
Dynamically load Python module from file path.
239
240
Parameters:
241
- path: Path to Python module file
242
243
Returns:
244
Loaded module object
245
"""
246
```
247
248
## Usage Examples
249
250
### URL Manipulation
251
252
```python
253
from streamlink.utils import absolute_url, update_qsd, url_concat
254
255
# Convert relative URLs to absolute
256
base = "https://example.com/videos/"
257
relative = "../api/stream.json"
258
absolute = absolute_url(base, relative)
259
# Result: "https://example.com/api/stream.json"
260
261
# Update query parameters
262
original_url = "https://api.example.com/stream?quality=720p"
263
updated_url = update_qsd(original_url, quality="1080p", format="hls")
264
# Result: "https://api.example.com/stream?quality=1080p&format=hls"
265
266
# Concatenate URL segments
267
api_base = "https://api.example.com"
268
endpoint = url_concat(api_base, "v1", "streams", "12345")
269
# Result: "https://api.example.com/v1/streams/12345"
270
```
271
272
### HTML Parsing in Plugins
273
274
```python
275
from streamlink.utils import parse_html
276
277
class ExamplePlugin(Plugin):
278
def _extract_stream_url(self):
279
# Fetch webpage
280
res = self.session.http.get(self.url)
281
282
# Parse HTML
283
doc = parse_html(res.text)
284
285
# Extract stream URL using XPath
286
video_elements = doc.xpath(".//video[@src]")
287
if video_elements:
288
return video_elements[0].get("src")
289
290
# Try alternative extraction
291
script_elements = doc.xpath(".//script[contains(text(), 'streamUrl')]")
292
for script in script_elements:
293
text = script.text
294
if "streamUrl" in text:
295
# Extract URL from JavaScript
296
start = text.find('"streamUrl":"') + 13
297
end = text.find('"', start)
298
return text[start:end]
299
300
return None
301
```
302
303
### JSON Data Processing
304
305
```python
306
from streamlink.utils import parse_json, search_dict
307
308
class APIPlugin(Plugin):
309
def _get_stream_data(self):
310
# Fetch JSON data from API
311
api_url = f"https://api.example.com/stream/{self.video_id}"
312
res = self.session.http.get(api_url)
313
314
# Parse JSON response
315
data = parse_json(res.text)
316
317
# Search nested data for stream URLs
318
hls_url = search_dict(data, "hls_url")
319
dash_url = search_dict(data, "dash_url")
320
321
return {
322
"hls": hls_url,
323
"dash": dash_url,
324
"title": search_dict(data, "title")
325
}
326
```
327
328
### Caching in Plugins
329
330
```python
331
from streamlink.utils import LRUCache
332
333
class CachedPlugin(Plugin):
334
# Class-level cache shared across instances
335
_cache = LRUCache(maxsize=50)
336
337
def _get_stream_metadata(self, video_id):
338
# Check cache first
339
cached_data = self._cache.get(video_id)
340
if cached_data:
341
return cached_data
342
343
# Fetch fresh data
344
api_url = f"https://api.example.com/video/{video_id}"
345
res = self.session.http.get(api_url)
346
metadata = parse_json(res.text)
347
348
# Cache the result
349
self._cache.set(video_id, metadata)
350
351
return metadata
352
```
353
354
### Named Pipe Usage
355
356
```python
357
from streamlink.utils import NamedPipe
358
import subprocess
359
360
# Create named pipe for streaming to external player
361
pipe = NamedPipe()
362
363
# Start external player reading from pipe
364
player_cmd = ["vlc", pipe.path]
365
player_process = subprocess.Popen(player_cmd)
366
367
# Stream data to pipe
368
stream = streams['best']
369
with pipe.open('wb') as pipe_fd, stream.open() as stream_fd:
370
while True:
371
data = stream_fd.read(8192)
372
if not data:
373
break
374
pipe_fd.write(data)
375
376
player_process.wait()
377
```
378
379
### Query String Parsing
380
381
```python
382
from streamlink.utils import parse_qsd
383
384
class RedirectPlugin(Plugin):
385
def _extract_real_url(self):
386
# Parse current URL query parameters
387
from urllib.parse import urlparse
388
parsed = urlparse(self.url)
389
390
# Extract query parameters
391
params = parse_qsd(parsed.query)
392
393
# Get real URL from redirect parameter
394
real_url = params.get('url') or params.get('redirect')
395
396
if real_url:
397
# Decode if URL-encoded
398
from urllib.parse import unquote
399
return unquote(real_url)
400
401
return None
402
```
403
404
### Dynamic Module Loading
405
406
```python
407
from streamlink.utils import load_module
408
import os
409
410
def load_custom_plugins(plugin_dir):
411
"""Load custom plugins from directory"""
412
plugins = []
413
414
for filename in os.listdir(plugin_dir):
415
if filename.endswith('.py') and not filename.startswith('_'):
416
plugin_path = os.path.join(plugin_dir, filename)
417
try:
418
module = load_module(plugin_path)
419
plugins.append(module)
420
except Exception as e:
421
print(f"Failed to load plugin {filename}: {e}")
422
423
return plugins
424
```
425
426
### Comprehensive URL Processing
427
428
```python
429
from streamlink.utils import *
430
431
class URLProcessor:
432
def process_stream_url(self, base_url, stream_path, **params):
433
# Convert to absolute URL
434
full_url = absolute_url(base_url, stream_path)
435
436
# Update with parameters
437
if params:
438
full_url = update_qsd(full_url, **params)
439
440
# Ensure HTTPS
441
full_url = update_scheme("https://", full_url)
442
443
# Add www if needed for certain domains
444
if "example.com" in full_url:
445
full_url = prepend_www(full_url)
446
447
return full_url
448
449
# Usage in plugin
450
processor = URLProcessor()
451
stream_url = processor.process_stream_url(
452
"https://example.com/videos/",
453
"stream.m3u8",
454
quality="720p",
455
token="abc123"
456
)
457
```
458
459
### Advanced Data Searching
460
461
```python
462
from streamlink.utils import search_dict
463
464
def extract_nested_streams(api_response):
465
"""Extract stream URLs from complex nested API response"""
466
467
# Search for various possible stream URL keys
468
stream_keys = ['stream_url', 'hls_url', 'manifest_url', 'source']
469
470
streams = {}
471
for key in stream_keys:
472
url = search_dict(api_response, key)
473
if url:
474
# Determine stream type from URL
475
if '.m3u8' in url:
476
streams['hls'] = url
477
elif '.mpd' in url:
478
streams['dash'] = url
479
else:
480
streams['http'] = url
481
482
# Also search for quality-specific streams
483
for quality in ['720p', '1080p', '480p']:
484
quality_url = search_dict(api_response, f"{quality}_url")
485
if quality_url:
486
streams[quality] = quality_url
487
488
return streams
489
```