Streamlink API 开发指南用 Python 提取、校验与消费流媒体数据【免费下载链接】streamlinkStreamlink is a CLI utility which pipes video streams from various services into a video player项目地址: https://gitcode.com/gh_mirrors/st/streamlinkStreamlink 不只提供命令行工具其底层还暴露了一套完整的 Python API——正是这套 API 支撑着 CLI 的每一次流提取。本文以 docs/api_guide.rst 组织的 Quickstart 与 Validation schemas 两篇指南为骨架系统讲解如何在自己的应用中通过streamlink.streams()/StreamlinkSession 提取流、如何打开流并读取数据以及如何使用声明式验证模式Validation schemas从 HTML、JSON、HTTP 响应中可靠地抽取流 URL 与元数据。读完本文你将能够独立编写一个基于 Streamlink 的流提取脚本并按照插件开发的规范实现健壮的流解析逻辑。说明docs/api_guide.rst本身是一个目录索引toctree其正文分别指向 docs/api_guide/quickstart.rst 与 docs/api_guide/validate.rst本文即围绕这两篇文档展开并补充仓库源码中的实现证据。一、快速上手提取流Quickstart1.1 最简单的流提取调用Streamlink 的 API 由 CLI 自身所使用同时也开放给希望在自有应用中复用流数据的开发者。最简用法是直接调用模块级的streamlink.streams()函数 import streamlink streams streamlink.streams(hls://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/bipbop_4x3_variant.m3u8)这一调用会尝试根据 URL 找到匹配的插件并调用插件从 URL 中提取流。其背后逻辑非常直接在 src/streamlink/api.py 中streams()只是创建了一个空白的Streamlink会话然后把 URL 转交给Streamlink.streams()def streams(url: str, **params): session Streamlink() return session.streams(url, **params)streamlink.streams()适合简单场景如果需要更精细的控制例如设置会话选项、传入插件选项则应使用 Session 对象手动获取流见本文第二节。返回值为一个dictkey 是流名称通常为清晰度/码率名value 是Stream子类的实例Stream基类定义见 src/streamlink/stream/stream.py streams {41k: HLSStream [hls, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/gear0/prog_index.m3u8, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/bipbop_4x3_variant.m3u8], 230k: HLSStream [hls, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/gear1/prog_index.m3u8, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/bipbop_4x3_variant.m3u8], 650k: HLSStream [hls, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/gear2/prog_index.m3u8, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/bipbop_4x3_variant.m3u8], 990k: HLSStream [hls, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/gear3/prog_index.m3u8, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/bipbop_4x3_variant.m3u8], 1900k: HLSStream [hls, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/gear4/prog_index.m3u8, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/bipbop_4x3_variant.m3u8], worst: HLSStream [hls, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/gear0/prog_index.m3u8, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/bipbop_4x3_variant.m3u8], best: HLSStream [hls, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/gear4/prog_index.m3u8, https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/bipbop_4x3_variant.m3u8]}注意两个细节返回的 dict 通常包含best与worst两个同义词分别指向最高/最低质量流该排序与去重逻辑实现在 src/streamlink/plugin/plugin.py 的Plugin.streams()中含stream_types排序、_alt后缀去重、非法名称过滤等。如果 URL 没有匹配到任何插件将抛出NoPluginError如果提取流的过程中发生错误将抛出PluginError。两者的定义见 src/streamlink/exceptions.py。1.2 打开流并读取数据拿到Stream对象后调用其open()方法即可得到一个类文件对象file-like object支持.read(size)与.close() fd streams[best].open() data fd.read(1024) fd.close()Stream.open()的契约在 src/streamlink/stream/stream.py 中定义成功时返回用于读取流数据的StreamIO文件对象失败时抛出StreamError。也就是说你可以像操作普通文件一样把流数据管道化到播放器或写入磁盘。1.3 检查流的内部参数Stream对象暴露了可供检视的属性各子类的可用属性参见 docs/api/stream.rst 中Stream子类的文档。例如HLSStream继承自HTTPStream含有一个url属性可直接拿到流地址 streams[best].url https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_4x3/gear4/prog_index.m3u8这在需要把解析结果交给第三方播放器或转码链路时非常实用。二、Session 对象可配置、可复用的流提取入口模块级streamlink.streams()每次都会创建全新的会话。当需要多次提取流、或要设置各种全局选项时应改用StreamlinkSession 对象——它会复用已加载的插件与 HTTP 连接配置效率更高。2.1 创建 Session 与设置选项 from streamlink import Streamlink session Streamlink({optional-session-option: 123})构造参数是一组可选的初始选项Mapping或Options实例。创建后既可以用便捷方法设置选项也可以直接操作session.options session.set_option(stream-timeout, 30) session.options.set(stream-timeout, 30)两种写法等价set_option()在 src/streamlink/session/session.py 中就是self.options.set()的便捷封装get_option()同理封装了self.options.get()。会话内部结构在 src/streamlink/session/session.py 中一目了然session.httpStreamlink 的requests.Session子类HTTPSession插件与流实现发起的全部 HTTP 请求都经由它session.optionsStreamlinkOptions实例StreamlinkOptions的定义见 src/streamlink/session/options.pysession.pluginsStreamlinkPlugins实例负责内置/外部插件的加载与 URL 匹配。2.2 常用会话选项一览下表整理了StreamlinkOptions中与日常开发关系最密切的选项类型、默认值、说明均取自 src/streamlink/session/options.py 的类文档与__init__默认值key类型默认值说明http-timeoutfloat20.0所有 HTTP/HTTPS 请求的通用超时http-proxystr \| NoneNone所有 HTTP/HTTPS 请求的代理地址http-cookiesdict \| str{}附加到每次请求的 Cookie如foobar;bazquxhttp-headersdict \| str{}附加到每次请求的请求头http-query-paramsdict \| str{}附加到每次请求的查询参数如foobarbazquxhttp-ssl-verifyboolTrue是否校验 TLS/SSL 证书http-trust-envboolTrue是否信任环境变量HTTP_PROXY等与~/.netrc认证localestr系统区域RFC 1766 格式的区域设置如en_US、es_ESinterfacestr \| NoneNone网络接口名或地址ringbuffer-sizeint1677721616 MiB多数流类型使用的内部环形缓冲区大小stream-timeoutfloat60.0从流中读取数据的超时stream-segment-attemptsint3分段流的分段下载尝试次数stream-segment-threadsint1并行下载分段的线程池大小stream-segment-timeoutfloat10.0分段的连接与读取超时stream-segmented-durationfloat0.0限制分段流播放时长四舍五入到最近分段0.0表示不限制hls-live-edgeint3HLS 直播流开始读取时距直播位置的段数hls-live-restartboolFalse是否跳到 HLS 直播流开头或尽可能靠前hls-playlist-reload-attemptsint3HLS 播放列表重载的最大尝试次数hls-audio-selectlist[str][]多音轨时按语言码或名称选择音源*表示全部dash-manifest-reload-attemptsint3DASH 清单重载的最大尝试次数ffmpeg-ffmpegstr \| NoneNone覆盖ffmpeg二进制路径默认从PATH查找mux-subtitlesboolFalse让支持的插件把字幕合入输出流webbrowserboolTrue启用/禁用 Streamlink 的 webbrowser API此外还有几组 HLS 专用选项如hls-start-offset、hls-segment-stream-data、hls-segment-ignore-names、hls-segment-key-uri、hls-playlist-reload-time与 FFmpeg 相关选项ffmpeg-loglevel、ffmpeg-fout、ffmpeg-video-transcode等完整清单可直接查阅 src/streamlink/session/options.py。从源码可见http-*选项大多直接映射到HTTPSession的属性如http-headers映射到session.http.headers见_OPTIONS_HTTP_ATTRS映射表src/streamlink/session/options.py。2.3 通过 Session 获取流获取流有两条路径路径一自动匹配插件 streams session.streams(URL)Streamlink.streams()的实现在 src/streamlink/session/session.py先调用resolve_url()解析出插件名、插件类与最终 URL再用会话与 URL 实例化插件并调用plugin.streams()。路径二手动解析并实例化插件Streamlink.streams()不允许传入插件选项plugin options。插件选项与会话选项是两回事——它们依赖具体插件因此在解析出匹配插件之前无法设置。如果需要认证数据或要改变插件行为必须手动解析插件 plugin_name, plugin_class, resolved_url session.resolve_url(URL) plugin plugin_class(session, resolved_url, options{plugin-option: 123}) streams plugin.streams()resolve_url()src/streamlink/session/session.py的行为要点若 URL 未指定协议默认补全为https://先用已加载插件对 URL 做正则匹配未匹配到时若follow_redirectTrue默认会尝试跟随 HTTP 重定向后再解析结果按(plugin_name, plugin_class, resolved_url)元组返回并且结果会被functools.lru_cache缓存容量 128重复解析同一 URL 不会重复开销匹配失败抛出NoPluginError。若只想做一次不跟随重定向的解析可使用resolve_url_no_redirect()。路径三直接导入插件类插件类还可以直接从streamlink.plugins包中对应模块导入通过模块的__plugin__属性此时输入 URL 必须命中该插件的 URL matchers from streamlink.plugins.twitch import __plugin__ as Twitch plugin Twitch(session, https://twitch.tv/CHANNEL, options{disable-ads: True, low-latency: True}) streams plugin.streams()插件的可用选项由模块内的pluginargument类装饰器声明见 src/streamlink/plugin/plugin.py 附近的定义例如上面Twitch插件的disable-ads、low-latency即来自 Twitch 插件自身的pluginargument声明。三、Validation schemas声明式数据校验与提取Streamlink 提供了一套用于声明式校验与提取数据的 APIstreamlink.plugin.api.validate模块其全部内容只是从streamlink.validate包重导出见 src/streamlink/plugin/api/validate.py。对插件实现者而言这是从网站与 Web API 中查找流 URL、流元数据等信息的利器。相比程序化地逐段写 if/else 校验 手工异常处理声明式 schema 的优势在于规则易读、易于组合、校验失败时给出有意义的错误信息。注意内部实现在streamlink.validate包公共接口在streamlink.plugin.api.validate二者在 src/streamlink/validate/init.py 中完成别名映射如all→AllSchema、any→AnySchema、get→GetItemSchema等。3.1 核心机制Schema 与 validateSchema类src/streamlink/validate/_validate.py是整个校验接口的外层封装它是AllSchema的子类额外实现了Schema.validate()方法class Schema(AllSchema): def validate(self, value: Any, name: str result, exception: type[Exception] PluginError) - Any: try: return validate(self, value) except ValidationError as err: raise exception(fUnable to validate {name}: {err}) from None也就是说Schema.validate()默认把ValidationError包装成PluginError抛出。这个传入schema关键字参数的接口被多种 Streamlink API 使用例如HTTPSession的请求方法见下文 3.5 节以及streamlink.utils.parse中的各解析函数。底层validate(schema, value)是一个基于functools.singledispatch的分发函数src/streamlink/validate/_validate.pyschema 对象的类型决定了校验规则默认分支做等值校验schema ! value即抛错其余类型则注册了各自的实现schema 类型校验规则返回普通对象123、123、None…等值校验原输入type如str、int输入是否为该类型的实例原输入list/tuple/set/frozenset输入类型匹配且每个元素能命中任意一个子 schema新的同类型序列dict输入为 dict逐个校验键值对可含optional键新的 dict可调用对象以输入为参数调用结果必须为真值原输入re.Pattern对输入执行search()re.Match或None各类 Schema 容器见下文各容器定义完整实现可查阅 src/streamlink/validate/_validate.py 与 docs/api/validate.rst后者是自动生成的 API 参考。3.2 简单 schema 与校验错误 from streamlink.plugin.api import validate schema_one validate.Schema(123) schema_two validate.Schema(123) schema_three validate.Schema(int, 123.0) schema_one.validate(123) 123 schema_two.validate(123) 123 schema_three.validate(123) 123Schema本身是AllSchema容器传入多个 schema 对象时会按顺序逐一校验每个子校验的输出作为下一个子校验的输入对应 src/streamlink/validate/_validate.py 的_validate_allschema。因此schema_one只含一个等值校验123schema_two只含一个等值校验123schema_three依次校验int类型校验与123.0等值校验。当输入为123时int校验通过并返回123再与123.0比较因123 123.0成立整体返回123。再看校验失败的情形 schema_one.validate(123) streamlink.exceptions.PluginError: Unable to validate result: ValidationError(equality): 123 does not equal 123 schema_three.validate(123.0) streamlink.exceptions.PluginError: Unable to validate result: ValidationError(type): Type of 123.0 should be int, but is float第一个例子123不等于123底层抛出ValidationError定义见 src/streamlink/validate/_exception.pySchema.validate()捕获后包装成带详细信息的PluginError。第二个例子尽管后续子 schema123.0是float理论上等值比较123.0 123.0能通过但第一个子 schemaint就失败了——123.0不是int实例类型校验实现在 src/streamlink/validate/_validate.py因此整个 schema 容器立即失败。3.3 从 JSON 数据中提取值下面这个 schema 演示了完整的解析 JSON → 校验结构 → 取值流水线 json_schema validate.Schema( ... str, ... validate.parse_json(), ... { ... status: validate.any(None, int), ... }, ... validate.get(status), ... ) json_schema.validate({status:null}) None json_schema.validate({status:123}) 123 json_schema.validate(Not JSON) streamlink.exceptions.PluginError: Unable to validate result: ValidationError: Unable to parse JSON: Expecting value: line 1 column 1 (char 0) (Not JSON) json_schema.validate({status:unknown}) streamlink.exceptions.PluginError: Unable to validate result: ValidationError(dict): Unable to validate value of key status Context(AnySchema): ValidationError(equality): unknown does not equal None ValidationError(type): Type of unknown should be int, but is str逐段拆解这个 schema 的四个子校验对应AllSchema的链式语义str只接受字符串输入validate.parse_json()这是一个TransformSchema转换 schema把字符串解析为 JSON 对象。parse_json在 src/streamlink/validate/_validators.py 中实现本质是把streamlink.utils.parse.parse_json()包成 transform并把解析异常转换为ValidationError{ status: validate.any(None, int) }dict 校验实现见 src/streamlink/validate/_validate.py。要求输入必须是 dict且必须包含status键该键的值用validate.any(None, int)校验——any是AnySchema容器src/streamlink/validate/_schemas.py至少一个子 schema 通过即可输出取第一个通过者。即status的值必须是NoneJSON 的null或intvalidate.get(status)GetItemSchemasrc/streamlink/validate/_schemas.py对任何实现__getitem__()的对象dict 等按下标取值并返回默认值defaultNone还支持元组形式的递归取值。当 JSON 数据非法或结构不符合 schema 时会生成如上所示的嵌套错误栈最外层标明失败的 schema 类型dict、AnySchema内层给出具体原因等值失败、类型失败便于定位问题。需要同时取多个值时还可以使用validate.unionUnionSchema或validate.union_getUnionGetSchema它们在同一输入上并行执行多个校验并把结果聚合为序列/dict。3.4 实战从 HTML 中查找流 URL设想一个网站播放器把流 URL 以 JSON 形式嵌在某个未知 HTML 元素的data-player属性里。用正则提取需要同时考虑 HTML 语法、属性内的 HTML 实体编码quot;等且 JSON 结构随时可能变化非常脆弱。更好的方案是解析 HTML → XPath 查询属性值 → 解析 JSON → 校验流 URL。同时当用户输入的 URL 对应的页面里没有播放器时不应抛出校验错误校验错误只应代表意外数据/真实错误而非直播未开播/页面不可访问因此 schema 应能优雅地返回空结果让插件返回空流列表CLI 正常退出。借助 schema这一切可以完全声明式地完成 schema validate.Schema( ... validate.parse_html(), ... validate.xml_xpath_string(.//*[data-player][1]/data-player), ... validate.none_or_all( ... validate.parse_json(), ... { ... validate.optional(url): validate.url( ... pathvalidate.endswith(.m3u8), ... ), ... }, ... validate.get(url), ... ), ... ) schema.validate( ... !doctype html ... section classno-video-player/section ... ) None schema.validate( ... !doctype html ... section ... classvideo-player ... >if schema: res schema.validate(res.text, nameresponse text, exceptionPluginError)下面是一个完整的最小插件实现复用了 3.4 节的 schema通过session.http.get(..., schema...)一步完成请求页面 校验提取 HLS URLimport re from streamlink.plugin import Plugin, pluginmatcher from streamlink.plugin.api import validate from streamlink.stream.hls import HLSStream pluginmatcher(re.compile(rhttps://example\.tld/)) class ExamplePlugin(Plugin): def _get_streams(): hls_url self.session.http.get(self.url, schemavalidate.Schema( validate.parse_html(), validate.xml_xpath_string(.//*[data-player][1]/data-player), validate.none_or_all( validate.parse_json(), { validate.optional(url): validate.url( pathvalidate.endswith(.m3u8), ), }, validate.get(url), ), )) if not hls_url: return None return HLSStream.parse_variant_playlist(self.session, hls_url) __plugin__ ExamplePlugin要点说明pluginmatcher(re.compile(...))定义于 src/streamlink/plugin/plugin.py声明插件匹配的 URL 正则Plugin基类中session、url等属性由框架注入插件必须实现_get_streams()src/streamlink/plugin/plugin.py返回流名 → 流对象的可迭代对/mapping或NoneHLSStream.parse_variant_playlist(session, hls_url)解析 HLS 变体播放列表并生成各档位的HLSStream__plugin__属性导出插件类供session.resolve_url()/streamlink.streams()发现。从 tests/session/test_session.py 等测试中可以看到session.streams()、resolve_url()在真实会话下的行为被测试覆盖可作为深入理解 API 契约的参考。四、更多验证工具速查除了上文用到的parse_json、parse_html、xml_xpath_string、none_or_all、optional、url、endswith、get、any、all之外streamlink.plugin.api.validate还提供以下常用构件完整定义见 src/streamlink/validate/_schemas.py 与 src/streamlink/validate/_validators.pySchema 容器类名称语义validate.all(*schemas)全部子 schema 依次通过输出逐级传递validate.any(*schemas)至少一个子 schema 通过返回第一个通过者的输出validate.none_or_all(*schemas)None输入直接放行否则等同allvalidate.transform(func, *args, **kwargs)把输入交给func转换并返回其输出validate.list(*schemas)输入必须是 list 且长度与 schema 数一致逐项校验validate.union(seq_or_dict)同一输入上并行执行多个校验并聚合结果validate.union_get(*keys)便捷版同一输入上执行多个get并聚合成元组validate.attr({...})校验对象属性返回输入对象的副本validate.get(key, defaultNone, strictFalse)按下标/键取值支持元组递归validate.regex(pattern, methodsearch)正则必须匹配method可选search/findall等validate.xml_element(tag..., text..., attrib...)校验 XML 元素的 tag/text/attrib/tail校验器函数返回可用的 schema 对象名称语义validate.url(**attributes)URL 校验可校验scheme/netloc/path/query等 urlparse 字段validate.parse_json()/parse_html()/parse_xml()/parse_qsd()调用对应streamlink.utils.parse解析函数validate.xml_xpath(...)/xml_xpath_string(...)/xml_find(...)/xml_findall(...)/xml_findtext(...)lxml 树查询与取值validate.startswith(...)/endswith(...)/contains(...)/length(n, opge)字符串/容器断言validate.getattr(attr, default)/hasattr(attr)对象属性取值/断言validate.filter(func)/map(func)对序列或 dict 逐项过滤/映射validate.nextjs_inline_rsc()解析 Next.js 内联 React Server ComponentRSCFlight 数据其中validate()函数本身通常不直接调用而是通过Schema.validate()或上述接受schema关键字的 API如session.http.get(..., schema...)间接触发。五、小结至此你已经掌握 Streamlink 开发侧的两大支柱流提取从streamlink.streams()的一行式调用到StreamlinkSession 的选项配置与resolve_url()/手动实例化插件的精细控制再到通过Stream.open()消费流数据、通过Stream.url等属性检视流参数声明式数据校验用Schema组合all/any/none_or_all/transform等容器与parse_json/parse_html/xml_xpath_string/url等校验器从 HTML、JSON、HTTP 响应中可靠提取流 URL 与元数据并让HTTPSession.request(..., schema...)在请求完成后自动完成校验。两者结合即可编写出健壮的流媒体提取程序Session 负责全局配置与插件解析Validation schemas 负责把不可信的网页/接口数据收敛为类型安全的结构化结果。若需进一步查阅 API 细节可继续阅读 docs/api/session.rst、docs/api/plugin.rst、docs/api/validate.rst、docs/api/stream.rst 与 docs/api/exceptions.rst。【免费下载链接】streamlinkStreamlink is a CLI utility which pipes video streams from various services into a video player项目地址: https://gitcode.com/gh_mirrors/st/streamlink创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考