87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
import typing
|
|
from urllib import parse
|
|
from urllib.parse import urlencode
|
|
|
|
def encode_streamlink_server_url(
|
|
username: str,
|
|
password: str,
|
|
base_url: str,
|
|
stream_url: str,
|
|
endpoint: typing.Optional[str] = None,
|
|
agent: typing.Optional[str] = None,
|
|
proxy_url: typing.Optional[str] = None,
|
|
request_headers: typing.Optional[dict] = None,
|
|
) -> str:
|
|
"""
|
|
Constructs a Streamlink server URL with authentication and query parameters.
|
|
|
|
Creates a properly encoded URL for accessing Streamlink server endpoints with
|
|
embedded credentials and streaming parameters. All parameters are URL-encoded
|
|
and appended as query string parameters.
|
|
|
|
Args:
|
|
username: Authentication username for Streamlink server
|
|
password: Authentication password for Streamlink server
|
|
base_url: Base URL of the Streamlink server (e.g., 'http://localhost:6090')
|
|
stream_url: Target stream URL to be proxied through Streamlink
|
|
endpoint: Optional endpoint path to append to base URL (e.g., '/stream')
|
|
agent: User-Agent header value for the streaming request
|
|
proxy_url: Proxy server URL to use for the stream connection
|
|
request_headers: Dictionary of HTTP headers to forward as query parameters
|
|
|
|
Returns:
|
|
str: Fully constructed URL with encoded query parameters in the format:
|
|
{base_url}[/{endpoint}]?username=X&password=X&url=X&[params...]
|
|
|
|
Example:
|
|
>>> encode_streamlink_server_url(
|
|
... "user", "pass", "http://localhost:6090", "https://example.com/stream.m3u8",
|
|
... endpoint="stream", agent="MyApp/1.0", proxy_url="http://proxy:3128",
|
|
... request_headers={"Referer": "https://example.com"}
|
|
... )
|
|
'http://localhost:6090/stream?username=user&password=pass&url=https%3A%2F%2Fexample.com%2Fstream.m3u8&agent=MyApp%2F1.0&proxy=http%3A%2F%2Fproxy%3A3128&Referer=https%3A%2F%2Fexample.com'
|
|
"""
|
|
# Construct the base URL
|
|
if endpoint is None:
|
|
url = base_url
|
|
else:
|
|
url = parse.urljoin(base_url, endpoint)
|
|
|
|
# Ensure url doesn't end with a slash for consistent handling
|
|
if url.endswith("/"):
|
|
url = url[:-1]
|
|
|
|
# Prepare query parameters
|
|
query_params = {}
|
|
|
|
# Username
|
|
if username is not None:
|
|
query_params["username"] = username
|
|
|
|
# Password
|
|
if password is not None:
|
|
query_params["password"] = password
|
|
|
|
# Stream URL
|
|
if stream_url is not None:
|
|
query_params["url"] = stream_url
|
|
|
|
# Agent
|
|
if agent is not None:
|
|
query_params["agent"] = agent
|
|
|
|
# Proxy URL
|
|
if proxy_url is not None:
|
|
query_params["proxy"] = proxy_url
|
|
|
|
# Add headers if provided
|
|
if request_headers:
|
|
query_params.update(
|
|
{key: value for key, value in request_headers.items()}
|
|
)
|
|
|
|
if query_params:
|
|
return f"{url}?{urlencode(query_params)}"
|
|
|
|
return url
|