Repository: streamlink/streamlink
Pull request #3970 (2021-08-31 16:56:09)

plugin: metadata attributes
- add author, category and title attributes to base Plugin class
- make getters return attributes instead of None
- remove unnecessary duplicate attributes and getters from plugins
- keep getters in plugins which use custom logic

----

This change makes it much easier to set the plugin metadata by simply setting the attributes instead of overriding the `get_{author,category,title}` methods every single time with different return values.

The `author`, `category` and `title` attributes are already a well established pattern, so making them part of the Plugin base class is the right choice.

odnar-dev: just add my [pr](https://github.com/streamlink/streamlink/pull/3941) to this, i couldn't fix the tests anyway
bastimeyer: @odnar-dev 
I will take a look at this feature once I've finished some other stuff. Sorry for wasting your time with #3941. The changes should be simple though. Just a bit difficult when stuff is changing the whole time on the master branch, I understand.

Patch:
diff --git a/src/streamlink/plugin/api/validate.py b/src/streamlink/plugin/api/validate.py
index 21ac3824..d2734de1 100644
--- a/src/streamlink/plugin/api/validate.py
+++ b/src/streamlink/plugin/api/validate.py
@@ -24,11 +24,19 @@ from urllib.parse import urlparse
 from lxml.etree import Element, iselement
 
 from streamlink.exceptions import PluginError
+from streamlink.utils.parse import (
+    parse_html as _parse_html,
+    parse_json as _parse_json,
+    parse_qsd as _parse_qsd,
+    parse_xml as _parse_xml
+)
+
 
 __all__ = [
     "any", "all", "filter", "get", "getattr", "hasattr", "length", "optional",
     "transform", "text", "union", "union_get", "url", "startswith", "endswith", "contains",
     "xml_element", "xml_find", "xml_findall", "xml_findtext", "xml_xpath", "xml_xpath_string",
+    "parse_json", "parse_html", "parse_xml", "parse_qsd",
     "validate", "Schema", "SchemaContainer"
 ]
 
@@ -69,8 +77,10 @@ class SchemaContainer:
 class transform:
     """Applies function to value to transform it."""
 
-    def __init__(self, func):
+    def __init__(self, func, *args, **kwargs):
         self.func = func
+        self.args = args
+        self.kwargs = kwargs
 
 
 class optional:
@@ -103,6 +113,9 @@ class xml_element:
         self.attrib = attrib
 
 
+# ----
+
+
 def length(length):
     """Checks value for minimum length using len()."""
     def min_len(value):
@@ -315,6 +328,25 @@ def xml_xpath_string(xpath):
     return xml_xpath(f"string({xpath})")
 
 
+def parse_json(*args, **kwargs):
+    return transform(_parse_json, *args, **kwargs, exception=ValueError, schema=None)
+
+
+def parse_html(*args, **kwargs):
+    return transform(_parse_html, *args, **kwargs, exception=ValueError, schema=None)
+
+
+def parse_xml(*args, **kwargs):
+    return transform(_parse_xml, *args, **kwargs, exception=ValueError, schema=None)
+
+
+def parse_qsd(*args, **kwargs):
+    return transform(_parse_qsd, *args, **kwargs, exception=ValueError, schema=None)
+
+
+# ----
+
+
 @singledispatch
 def validate(schema, value):
     if callable(schema):
@@ -351,9 +383,9 @@ def validate_all(schemas, value):
 
 
 @validate.register(transform)
-def validate_transform(schema, value):
+def validate_transform(schema: transform, value):
     validate(callable, schema.func)
-    return schema.func(value)
+    return schema.func(value, *schema.args, **schema.kwargs)
 
 
 @validate.register(list)
diff --git a/src/streamlink/plugin/plugin.py b/src/streamlink/plugin/plugin.py
index 1b789ddc..c4869508 100644
--- a/src/streamlink/plugin/plugin.py
+++ b/src/streamlink/plugin/plugin.py
@@ -202,6 +202,11 @@ class Plugin:
     # a reference to the `re.Match` result of the first matching matcher
     match: Match
 
+    # plugin metadata attributes
+    author: Optional[str] = None
+    category: Optional[str] = None
+    title: Optional[str] = None
+
     cache = None
     logger = None
     module = "unknown"
@@ -425,14 +430,14 @@ class Plugin:
     def _get_streams(self):
         raise NotImplementedError
 
-    def get_title(self):
-        return None
+    def get_title(self) -> Optional[str]:
+        return self.title
 
-    def get_author(self):
-        return None
+    def get_author(self) -> Optional[str]:
+        return self.author
 
-    def get_category(self):
-        return None
+    def get_category(self) -> Optional[str]:
+        return self.category
 
     def save_cookies(self, cookie_filter=None, default_expires=60 * 60 * 24 * 7):
         """
diff --git a/src/streamlink/plugins/atresplayer.py b/src/streamlink/plugins/atresplayer.py
index d7becef6..70852b66 100644
--- a/src/streamlink/plugins/atresplayer.py
+++ b/src/streamlink/plugins/atresplayer.py
@@ -1,6 +1,5 @@
 import logging
 import re
-from functools import partial
 
 from streamlink.plugin import Plugin, pluginmatcher
 from streamlink.plugin.api import validate
@@ -22,7 +21,7 @@ class AtresPlayer(Plugin):
             validate.all(
                 validate.get(1),
                 validate.transform(parse_json),
-                validate.transform(partial(search_dict, key="href")),
+                validate.transform(search_dict, key="href"),
             )
         )
     )
@@ -31,7 +30,7 @@ class AtresPlayer(Plugin):
             None,
             validate.all(
                 validate.transform(parse_json),
-                validate.transform(partial(search_dict, key="urlVideo")),
+                validate.transform(search_dict, key="urlVideo"),
             )
         )
     )
diff --git a/src/streamlink/plugins/booyah.py b/src/streamlink/plugins/booyah.py
index 099977b4..2fc6f332 100644
--- a/src/streamlink/plugins/booyah.py
+++ b/src/streamlink/plugins/booyah.py
@@ -55,19 +55,6 @@ class Booyah(Plugin):
         },
     })
 
-    author = None
-    category = None
-    title = None
-
-    def get_author(self):
-        return self.author
-
-    def get_category(self):
-        return self.category
-
-    def get_title(self):
-        return self.title
-
     @classmethod
     def stream_weight(cls, stream):
         if stream == "source":
diff --git a/src/streamlink/plugins/deutschewelle.py b/src/streamlink/plugins/deutschewelle.py
index 805b8b19..9e6e2128 100644
--- a/src/streamlink/plugins/deutschewelle.py
+++ b/src/streamlink/plugins/deutschewelle.py
@@ -19,15 +19,6 @@ class DeutscheWelle(Plugin):
     DEFAULT_CHANNEL = "1"
     API_URL = "https://www.dw.com/playersources/v-{media_id}?hls=true"
 
-    author = None
-    title = None
-
-    def get_author(self):
-        return self.author
-
-    def get_title(self):
-        return self.title
-
     def _find_metadata(self, elem: Element):
         self.author = elem.xpath("string(.//input[@name='channel_name'][1]/@value)") or None
         self.title = elem.xpath("string(.//input[@name='media_title'][1]/@value)") or None
diff --git a/src/streamlink/plugins/dlive.py b/src/streamlink/plugins/dlive.py
index 104b6ba5..0d329a33 100644
--- a/src/streamlink/plugins/dlive.py
+++ b/src/streamlink/plugins/dlive.py
@@ -58,18 +58,10 @@ class DLive(Plugin):
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.author = None
-        self.title = None
 
         self.video = self.match.group("video")
         self.channel = self.match.group("channel")
 
-    def get_author(self):
-        return self.author
-
-    def get_title(self):
-        return self.title
-
     def _get_streams_video(self):
         log.debug("Getting video HLS streams for {0}".format(self.video))
         try:
diff --git a/src/streamlink/plugins/galatasaraytv.py b/src/streamlink/plugins/galatasaraytv.py
index eddd8a59..9573a7e1 100644
--- a/src/streamlink/plugins/galatasaraytv.py
+++ b/src/streamlink/plugins/galatasaraytv.py
@@ -13,8 +13,7 @@ log = logging.getLogger(__name__)
 class GalatasarayTV(Plugin):
     playervars_re = re.compile(r"sources\s*:\s*\[\s*\{\s*type\s*:\s*\"(.*?)\",\s*src\s*:\s*\"(.*?)\"", re.DOTALL)
 
-    def get_title(self):
-        return "Galatasaray TV"
+    title = "Galatasaray TV"
 
     def _get_streams(self):
         res = self.session.http.get(self.url)
diff --git a/src/streamlink/plugins/latina.py b/src/streamlink/plugins/latina.py
index 89d3020f..f4ec9eac 100644
--- a/src/streamlink/plugins/latina.py
+++ b/src/streamlink/plugins/latina.py
@@ -13,8 +13,7 @@ log = logging.getLogger(__name__)
     r"https?://(?:www\.)?latina\.pe/tvenvivo"
 ))
 class Latina(Plugin):
-    def get_title(self):
-        return "Latina"
+    title = "Latina"
 
     def _get_streams(self):
         self.session.http.headers.update({
diff --git a/src/streamlink/plugins/mjunoon.py b/src/streamlink/plugins/mjunoon.py
index ccd4c508..d6b2f8a6 100644
--- a/src/streamlink/plugins/mjunoon.py
+++ b/src/streamlink/plugins/mjunoon.py
@@ -61,19 +61,6 @@ class Mjunoon(Plugin):
         'aes-256-cbc': AES.MODE_CBC,
     }
 
-    author = None
-    category = None
-    title = None
-
-    def get_author(self):
-        return self.author
-
-    def get_category(self):
-        return self.category
-
-    def get_title(self):
-        return self.title
-
     def get_data(self):
         js_data = {}
         res = self.session.http.get(self.url)
diff --git a/src/streamlink/plugins/nbcnews.py b/src/streamlink/plugins/nbcnews.py
index 23d5ecd4..b4c2d72d 100644
--- a/src/streamlink/plugins/nbcnews.py
+++ b/src/streamlink/plugins/nbcnews.py
@@ -51,8 +51,7 @@ class NBCNews(Plugin):
         ))
     )
 
-    def get_title(self):
-        return 'NBC News Now'
+    title = "NBC News Now"
 
     def _get_streams(self):
         video_id = self.session.http.get(self.url, schema=self.json_data_schema)
diff --git a/src/streamlink/plugins/nimotv.py b/src/streamlink/plugins/nimotv.py
index a7681bbe..f448fb9b 100644
--- a/src/streamlink/plugins/nimotv.py
+++ b/src/streamlink/plugins/nimotv.py
@@ -16,10 +16,6 @@ class NimoTV(Plugin):
     data_url = 'https://m.nimo.tv/{0}'
     data_re = re.compile(r'<script>var G_roomBaseInfo = ({.*?});</script>')
 
-    author = None
-    category = None
-    title = None
-
     data_schema = validate.Schema(
         validate.transform(data_re.search),
         validate.any(None, validate.all(
@@ -47,15 +43,6 @@ class NimoTV(Plugin):
     _re_id = re.compile(br'id=([^|\\]+)')
     _re_tp = re.compile(br'tp=(\d+)')
 
-    def get_author(self):
-        return self.author
-
-    def get_category(self):
-        return self.category
-
-    def get_title(self):
-        return self.title
-
     def _get_streams(self):
         username = self.match.group('username')
         if not username:
diff --git a/src/streamlink/plugins/nos.py b/src/streamlink/plugins/nos.py
index 48181aff..a7c3cc53 100644
--- a/src/streamlink/plugins/nos.py
+++ b/src/streamlink/plugins/nos.py
@@ -15,15 +15,11 @@ log = logging.getLogger(__name__)
 ))
 class NOS(Plugin):
     _msg_live_offline = "This livestream is offline."
-    title = None
     vod_keys = {
         "pages/Collection/Video/Video": "item",
         "pages/Video/Video": "video",
     }
 
-    def get_title(self):
-        return self.title
-
     def _get_streams(self):
         res = self.session.http.get(self.url)
         for script in itertags(res.text, "script"):
diff --git a/src/streamlink/plugins/nrk.py b/src/streamlink/plugins/nrk.py
index 9d265843..7980e5c7 100644
--- a/src/streamlink/plugins/nrk.py
+++ b/src/streamlink/plugins/nrk.py
@@ -48,15 +48,6 @@ class NRK(Plugin):
         },
     ))
 
-    category = None
-    title = None
-
-    def get_category(self):
-        return self.category
-
-    def get_title(self):
-        return self.title
-
     def _get_streams(self):
         # Construct manifest URL for this program.
         program_type, program_id = self.match.groups()
diff --git a/src/streamlink/plugins/openrectv.py b/src/streamlink/plugins/openrectv.py
index f16ecbf5..a55eb86e 100644
--- a/src/streamlink/plugins/openrectv.py
+++ b/src/streamlink/plugins/openrectv.py
@@ -56,8 +56,6 @@ class OPENRECtv(Plugin):
 
     def __init__(self, url):
         super().__init__(url)
-        self.author = None
-        self.title = None
         self.video_id = None
 
     def login(self, email, password):
diff --git a/src/streamlink/plugins/picarto.py b/src/streamlink/plugins/picarto.py
index 8b776863..cdb4a34d 100644
--- a/src/streamlink/plugins/picarto.py
+++ b/src/streamlink/plugins/picarto.py
@@ -45,19 +45,6 @@ class Picarto(Plugin):
 
     HLS_URL = 'https://{origin}.picarto.tv/stream/hls/{file_name}/index.m3u8'
 
-    author = None
-    category = None
-    title = None
-
-    def get_author(self):
-        return self.author
-
-    def get_category(self):
-        return self.category
-
-    def get_title(self):
-        return self.title
-
     def get_live(self, username):
         res = self.session.http.get(f'https://ptvintern.picarto.tv/api/channel/detail/{username}')
         channel_data = self.session.http.json(res, schema=self.channel_schema)
diff --git a/src/streamlink/plugins/pluto.py b/src/streamlink/plugins/pluto.py
index 00a563b9..911f68c4 100644
--- a/src/streamlink/plugins/pluto.py
+++ b/src/streamlink/plugins/pluto.py
@@ -21,11 +21,6 @@ log = logging.getLogger(__name__)
     )
 ''', re.VERBOSE))
 class Pluto(Plugin):
-    title = None
-
-    def get_title(self):
-        return self.title
-
     def _schema_media(self, slug):
         return validate.Schema(
             [{
diff --git a/src/streamlink/plugins/reuters.py b/src/streamlink/plugins/reuters.py
index 1e963d4e..96b5120a 100644
--- a/src/streamlink/plugins/reuters.py
+++ b/src/streamlink/plugins/reuters.py
@@ -46,10 +46,6 @@ class Reuters(Plugin):
         )
     )
 
-    def __init__(self, url):
-        super().__init__(url)
-        self.title = None
-
     def get_title(self):
         if not self.title:
             self._get_data()
diff --git a/src/streamlink/plugins/rtve.py b/src/streamlink/plugins/rtve.py
index e30b7782..802c2185 100644
--- a/src/streamlink/plugins/rtve.py
+++ b/src/streamlink/plugins/rtve.py
@@ -1,7 +1,6 @@
 import base64
 import logging
 import re
-from functools import partial
 from urllib.parse import urlparse
 
 from Crypto.Cipher import Blowfish
@@ -57,7 +56,7 @@ class Rtve(Plugin):
     _re_idAsset = re.compile(r"\"idAsset\":\"(\d+)\"")
     secret_key = base64.b64decode("eWVMJmRhRDM=")
     cdn_schema = validate.Schema(
-        validate.transform(partial(parse_xml, invalid_char_entities=True)),
+        validate.transform(parse_xml, invalid_char_entities=True),
         validate.xml_findall(".//preset"),
         [
             validate.union({
diff --git a/src/streamlink/plugins/streann.py b/src/streamlink/plugins/streann.py
index f8da5699..243b56a8 100644
--- a/src/streamlink/plugins/streann.py
+++ b/src/streamlink/plugins/streann.py
@@ -58,10 +58,6 @@ class Streann(Plugin):
 
     _device_id = None
     _domain = None
-    title = None
-
-    def get_title(self):
-        return self.title
 
     @property
     def device_id(self):
diff --git a/src/streamlink/plugins/stv.py b/src/streamlink/plugins/stv.py
index 1430c394..7ed7f2ef 100644
--- a/src/streamlink/plugins/stv.py
+++ b/src/streamlink/plugins/stv.py
@@ -13,8 +13,6 @@ log = logging.getLogger(__name__)
 class STV(Plugin):
     API_URL = 'https://player.api.stv.tv/v1/streams/stv/'
 
-    title = None
-
     def get_title(self):
         if self.title is None:
             self._get_api_results()
diff --git a/src/streamlink/plugins/svtplay.py b/src/streamlink/plugins/svtplay.py
index 1eea9325..fd608663 100644
--- a/src/streamlink/plugins/svtplay.py
+++ b/src/streamlink/plugins/svtplay.py
@@ -16,10 +16,6 @@ log = logging.getLogger(__name__)
 class SVTPlay(Plugin):
     api_url = 'https://api.svt.se/videoplayer-api/video/{0}'
 
-    author = None
-    category = None
-    title = None
-
     latest_episode_url_re = re.compile(r'''
         data-rt="top-area-play-button"\s+href="(?P<url>[^"]+)"
     ''', re.VERBOSE)
@@ -43,18 +39,6 @@ class SVTPlay(Plugin):
         PluginArgument("mux-subtitles", is_global=True)
     )
 
-    def get_author(self):
-        if self.author is not None:
-            return self.author
-
-    def get_category(self):
-        if self.category is not None:
-            return self.category
-
-    def get_title(self):
-        if self.title is not None:
-            return self.title
-
     def _set_metadata(self, data, category):
         if 'programTitle' in data:
             self.author = data['programTitle']
diff --git a/src/streamlink/plugins/tv4play.py b/src/streamlink/plugins/tv4play.py
index d53fc39f..02c6a8da 100644
--- a/src/streamlink/plugins/tv4play.py
+++ b/src/streamlink/plugins/tv4play.py
@@ -19,7 +19,6 @@ log = logging.getLogger(__name__)
     /(?P<video_id>\d+)
 """, re.VERBOSE))
 class TV4Play(Plugin):
-    title = None
     video_id = None
 
     api_url = "https://playback-api.b17g.net"
diff --git a/src/streamlink/plugins/tv8.py b/src/streamlink/plugins/tv8.py
index 69b7dabb..60b74aba 100644
--- a/src/streamlink/plugins/tv8.py
+++ b/src/streamlink/plugins/tv8.py
@@ -14,8 +14,7 @@ log = logging.getLogger(__name__)
 class TV8(Plugin):
     _re_hls = re.compile(r"""file\s*:\s*(["'])(?P<hls_url>https?://.*?\.m3u8.*?)\1""")
 
-    def get_title(self):
-        return 'TV8'
+    title = "TV8"
 
     def _get_streams(self):
         hls_url = self.session.http.get(self.url, schema=validate.Schema(
diff --git a/src/streamlink/plugins/twitch.py b/src/streamlink/plugins/twitch.py
index f8d97094..65cf2fa3 100644
--- a/src/streamlink/plugins/twitch.py
+++ b/src/streamlink/plugins/twitch.py
@@ -463,9 +463,6 @@ class Twitch(Plugin):
         self.video_id = None
         self.channel = None
         self.clip_name = None
-        self.title = None
-        self.author = None
-        self.category = None
 
         if self.subdomain == "player":
             # pop-out player
diff --git a/src/streamlink/plugins/youtube.py b/src/streamlink/plugins/youtube.py
index 6d6e9ef0..302d4b6c 100644
--- a/src/streamlink/plugins/youtube.py
+++ b/src/streamlink/plugins/youtube.py
@@ -83,16 +83,8 @@ class YouTube(Plugin):
         elif parsed.scheme != "https":
             self.url = urlunparse(parsed._replace(scheme="https"))
 
-        self.author = None
-        self.title = None
         self.session.http.headers.update({'User-Agent': useragents.CHROME})
 
-    def get_author(self):
-        return self.author
-
-    def get_title(self):
-        return self.title
-
     @classmethod
     def stream_weight(cls, stream):
         match_3d = re.match(r"(\w+)_3d", stream)
diff --git a/src/streamlink/plugins/zeenews.py b/src/streamlink/plugins/zeenews.py
index 3e812e8e..f12a53fd 100644
--- a/src/streamlink/plugins/zeenews.py
+++ b/src/streamlink/plugins/zeenews.py
@@ -14,8 +14,7 @@ class ZeeNews(Plugin):
     HLS_URL = 'https://z5ams.akamaized.net/zeenews/index.m3u8{0}'
     TOKEN_URL = 'https://useraction.zee5.com/token/live.php'
 
-    def get_title(self):
-        return 'Zee News'
+    title = 'Zee News'
 
     def _get_streams(self):
         res = self.session.http.get(self.TOKEN_URL)
diff --git a/src/streamlink/utils/__init__.py b/src/streamlink/utils/__init__.py
index f2ff849f..be690f2a 100644
--- a/src/streamlink/utils/__init__.py
+++ b/src/streamlink/utils/__init__.py
@@ -1,17 +1,14 @@
-import json
-import re
 import zlib
 from collections import OrderedDict
 from importlib.machinery import FileFinder, SOURCE_SUFFIXES, SourceFileLoader
 from importlib.util import module_from_spec
 from typing import Dict, Generic, Optional, TypeVar
-from urllib.parse import parse_qsl, urljoin, urlparse
-
-from lxml.etree import XML
+from urllib.parse import urljoin, urlparse
 
 from streamlink.exceptions import PluginError
 from streamlink.utils.lazy_formatter import LazyFormatter
 from streamlink.utils.named_pipe import NamedPipe
+from streamlink.utils.parse import parse_html, parse_json, parse_qsd, parse_xml
 from streamlink.utils.url import update_scheme, url_equal
 
 
@@ -61,74 +58,6 @@ def prepend_www(url):
         return url
 
 
-def parse_json(data, name="JSON", exception=PluginError, schema=None):
-    """Wrapper around json.loads.
-
-    Wraps errors in custom exception with a snippet of the data in the message.
-    """
-    try:
-        json_data = json.loads(data)
-    except ValueError as err:
-        snippet = repr(data)
-        if len(snippet) > 35:
-            snippet = snippet[:35] + " ..."
-        else:
-            snippet = data
-
-        raise exception("Unable to parse {0}: {1} ({2})".format(name, err, snippet))
-
-    if schema:
-        json_data = schema.validate(json_data, name=name, exception=exception)
-
-    return json_data
-
-
-def parse_xml(data, name="XML", ignore_ns=False, exception=PluginError, schema=None, invalid_char_entities=False):
-    """Wrapper around ElementTree.fromstring with some extras.
-
-    Provides these extra features:
-     - Handles incorrectly encoded XML
-     - Allows stripping namespace information
-     - Wraps errors in custom exception with a snippet of the data in the message
-    """
-    if isinstance(data, str):
-        data = bytes(data, "utf8")
-
-    if ignore_ns:
-        data = re.sub(br"[\t ]xmlns=\"(.+?)\"", b"", data)
-
-    if invalid_char_entities:
-        data = re.sub(br"&(?!(?:#(?:[0-9]+|[Xx][0-9A-Fa-f]+)|[A-Za-z0-9]+);)", b"&amp;", data)
-
-    try:
-        tree = XML(data)
-    except Exception as err:
-        snippet = repr(data)
-        if len(snippet) > 35:
-            snippet = snippet[:35] + " ..."
-
-        raise exception(f"Unable to parse {name}: {err} ({snippet})")
-
-    if schema:
-        tree = schema.validate(tree, name=name, exception=exception)
-
-    return tree
-
-
-def parse_qsd(data, name="query string", exception=PluginError, schema=None, **params):
-    """Parses a query string into a dict.
-
-    Unlike parse_qs and parse_qsl, duplicate keys are not preserved in
-    favor of a simpler return value.
-    """
-
-    value = dict(parse_qsl(data, **params))
-    if schema:
-        value = schema.validate(value, name=name, exception=exception)
-
-    return value
-
-
 def rtmpparse(url):
     parse = urlparse(url)
     netloc = "{hostname}:{port}".format(hostname=parse.hostname,
@@ -210,7 +139,15 @@ class LRUCache(Generic[TCacheKey, TCacheValue]):
             self.cache.popitem(last=False)
 
 
-__all__ = ["load_module", "swfdecompress", "update_scheme", "url_equal",
-           "verifyjson", "absolute_url", "parse_qsd", "parse_json",
-           "parse_xml", "rtmpparse", "prepend_www", "NamedPipe",
-           "escape_librtmp", "LRUCache", "LazyFormatter"]
+__all__ = [
+    "load_module",
+    "escape_librtmp", "rtmpparse", "swfdecompress",
+    "verifyjson",
+    "absolute_url", "prepend_www",
+    "search_dict",
+    "LRUCache",
+    "LazyFormatter",
+    "NamedPipe",
+    "parse_html", "parse_json", "parse_qsd", "parse_xml",
+    "update_scheme", "url_equal",
+]
diff --git a/src/streamlink/utils/parse.py b/src/streamlink/utils/parse.py
new file mode 100644
index 00000000..8165a2c3
--- /dev/null
+++ b/src/streamlink/utils/parse.py
@@ -0,0 +1,99 @@
+import json
+import re
+from urllib.parse import parse_qsl
+
+from lxml.etree import HTML, XML
+
+from streamlink.plugin import PluginError
+
+
+def _parse(parser, data, name, exception, schema, *args, **kwargs):
+    try:
+        parsed = parser(data, *args, **kwargs)
+    except Exception as err:
+        snippet = repr(data)
+        if len(snippet) > 35:
+            snippet = f"{snippet[:35]} ..."
+
+        raise exception(f"Unable to parse {name}: {err} ({snippet})")
+
+    if schema:
+        parsed = schema.validate(parsed, name=name, exception=exception)
+
+    return parsed
+
+
+def parse_json(
+    data,
+    name="JSON",
+    exception=PluginError,
+    schema=None,
+    *args, **kwargs
+):
+    """Wrapper around json.loads.
+
+    Provides these extra features:
+     - Wraps errors in custom exception with a snippet of the data in the message
+    """
+    return _parse(json.loads, data, name, exception, schema, *args, **kwargs)
+
+
+def parse_html(
+    data,
+    name="HTML",
+    exception=PluginError,
+    schema=None,
+    *args, **kwargs
+):
+    """Wrapper around lxml.etree.HTML with some extras.
+
+    Provides these extra features:
+     - Handles incorrectly encoded HTML
+     - Wraps errors in custom exception with a snippet of the data in the message
+    """
+    if isinstance(data, str):
+        data = bytes(data, "utf8")
+
+    return _parse(HTML, data, name, exception, schema, *args, **kwargs)
+
+
+def parse_xml(
+    data,
+    ignore_ns=False,
+    invalid_char_entities=False,
+    name="XML",
+    exception=PluginError,
+    schema=None,
+    *args, **kwargs
+):
+    """Wrapper around lxml.etree.XML with some extras.
+
+    Provides these extra features:
+     - Handles incorrectly encoded XML
+     - Allows stripping namespace information
+     - Wraps errors in custom exception with a snippet of the data in the message
+    """
+    if isinstance(data, str):
+        data = bytes(data, "utf8")
+    if ignore_ns:
+        data = re.sub(br"[\t ]xmlns=\"(.+?)\"", b"", data)
+    if invalid_char_entities:
+        data = re.sub(br"&(?!(?:#(?:[0-9]+|[Xx][0-9A-Fa-f]+)|[A-Za-z0-9]+);)", b"&amp;", data)
+
+    return _parse(XML, data, name, exception, schema, *args, **kwargs)
+
+
+def parse_qsd(
+    data,
+    name="query string",
+    exception=PluginError,
+    schema=None,
+    *args, **kwargs
+):
+    """Parses a query string into a dict.
+
+    Provides these extra features:
+     - Unlike parse_qs and parse_qsl, duplicate keys are not preserved in favor of a simpler return value
+     - Wraps errors in custom exception with a snippet of the data in the message
+    """
+    return _parse(lambda d: dict(parse_qsl(d, *args, **kwargs)), data, name, exception, schema)


Test patch:
diff --git a/tests/plugin/testplugin.py b/tests/plugin/testplugin.py
index f311246a..f030982b 100644
--- a/tests/plugin/testplugin.py
+++ b/tests/plugin/testplugin.py
@@ -35,14 +35,9 @@ class TestPlugin(Plugin):
         "a_option": "default"
     })
 
-    def get_title(self):
-        return "Test Title"
-
-    def get_author(self):
-        return "Tѥst Āuƭhǿr"
-
-    def get_category(self):
-        return None
+    author = "Tѥst Āuƭhǿr"
+    category = None
+    title = "Test Title"
 
     def _get_streams(self):
         if "empty" in self.url:
diff --git a/tests/test_api_validate.py b/tests/test_api_validate.py
index a8663f7a..19dcc691 100644
--- a/tests/test_api_validate.py
+++ b/tests/test_api_validate.py
@@ -4,9 +4,34 @@ import unittest
 from lxml.etree import Element
 
 from streamlink.plugin.api.validate import (
-    all, any, attr, endswith, filter, get, getattr, hasattr,
-    length, map, optional, startswith, text, transform, union, union_get, url,
-    validate, xml_element, xml_find, xml_findall, xml_findtext, xml_xpath, xml_xpath_string
+    all,
+    any,
+    attr,
+    endswith,
+    filter,
+    get,
+    getattr,
+    hasattr,
+    length,
+    map,
+    optional,
+    parse_html,
+    parse_json,
+    parse_qsd,
+    parse_xml,
+    startswith,
+    text,
+    transform,
+    union,
+    union_get,
+    url,
+    validate,
+    xml_element,
+    xml_find,
+    xml_findall,
+    xml_findtext,
+    xml_xpath,
+    xml_xpath_string,
 )
 
 
@@ -16,11 +41,8 @@ class TestPluginAPIValidate(unittest.TestCase):
 
         assert validate(int, 1) == 1
 
-        assert validate(transform(int), "1") == 1
-
         assert validate(text, "abc") == "abc"
         assert validate(text, "日本語") == "日本語"
-        assert validate(transform(text), 1) == "1"
 
         assert validate(list, ["a", 1]) == ["a", 1]
         assert validate(dict, {"a": 1}) == {"a": 1}
@@ -38,6 +60,23 @@ class TestPluginAPIValidate(unittest.TestCase):
 
         assert validate(any(int), 4) == 4
 
+    def test_transform(self):
+        assert validate(transform(int), "1") == 1
+        assert validate(transform(str), 1) == "1"
+        assert validate(
+            transform(
+                lambda value, *args, **kwargs: f"{value}{args}{kwargs}",
+                *("b", "c"),
+                **dict(d="d", e="e")
+            ),
+            "a"
+        ) == "a('b', 'c'){'d': 'd', 'e': 'e'}"
+
+        def no_args():
+            pass  # pragma: no cover
+
+        self.assertRaises(TypeError, validate, transform(no_args), "some value")
+
     def test_union(self):
         assert validate(union((get("foo"), get("bar"))),
                         {"foo": "alpha", "bar": "beta"}) == ("alpha", "beta")
@@ -238,3 +277,27 @@ class TestPluginAPIValidate(unittest.TestCase):
 
     def test_endswith(self):
         assert validate(endswith("åäö"), "xyzåäö")
+
+    def test_parse_json(self):
+        assert validate(parse_json(), '{"a": ["b", true, false, null, 1, 2.3]}') == {"a": ["b", True, False, None, 1, 2.3]}
+        with self.assertRaises(ValueError) as cm:
+            validate(parse_json(), "invalid")
+        assert str(cm.exception) == "Unable to parse JSON: Expecting value: line 1 column 1 (char 0) ('invalid')"
+
+    def test_parse_html(self):
+        assert validate(parse_html(), '<!DOCTYPE html><body>&quot;perfectly&quot;<a>valid<div>HTML').tag == "html"
+        with self.assertRaises(ValueError) as cm:
+            validate(parse_html(), None)
+        assert str(cm.exception) == "Unable to parse HTML: can only parse strings (None)"
+
+    def test_parse_xml(self):
+        assert validate(parse_xml(), '<?xml version="1.0" encoding="utf-8"?><root></root>').tag == "root"
+        with self.assertRaises(ValueError) as cm:
+            validate(parse_xml(), None)
+        assert str(cm.exception) == "Unable to parse XML: can only parse strings (None)"
+
+    def test_parse_qsd(self):
+        assert validate(parse_qsd(), 'foo=bar&foo=baz') == {"foo": "baz"}
+        with self.assertRaises(ValueError) as cm:
+            validate(parse_qsd(), 123)
+        assert str(cm.exception) == "Unable to parse query string: 'int' object has no attribute 'decode' (123)"
diff --git a/tests/test_cli_main.py b/tests/test_cli_main.py
index c43cac4c..9f763c96 100644
--- a/tests/test_cli_main.py
+++ b/tests/test_cli_main.py
@@ -346,7 +346,10 @@ class TestCLIMainHandleStream(unittest.TestCase):
         args.player_continuous_http = False
         mock_output_stream.return_value = True
 
-        plugin = Mock(FakePlugin(""), get_author=lambda: "AUTHOR", get_category=lambda: "CATEGORY", get_title=lambda: "TITLE")
+        plugin = FakePlugin("")
+        plugin.author = "AUTHOR"
+        plugin.category = "CATEGORY"
+        plugin.title = "TITLE"
         stream = Stream(session=Mock())
         streams = {"best": stream}