新增录制时自动解析实际分辨率显示在webui;添加直播流 url 和 header 信息到 webui;修复统计界面的部分bug

This commit is contained in:
kira1928 committed 2026-02-20 04:15:10 +09:00
1 parent 6999f7e23c
commit 3d73e8d70a
27 files changed
+2698 -130

No files matched your search

+11 -11
View File
@@ -3,8 +3,9 @@
help:
@echo "Available commands:"
@echo " make build - Build release version"
@echo " Optional: VERSION, PLATFORM, ARCH, NO_TELEMETRY=1"
@echo " make dev - Build development version (with debug info)"
@echo " make dev-version VERSION=x.x.x - Build dev with custom version"
@echo " Optional: VERSION, PLATFORM, ARCH, NO_TELEMETRY=1"
@echo " make dev-incremental - Incremental dev build (only rebuild if sources changed)"
@echo " make test - Run unit tests"
@echo " make test-e2e - Run E2E tests with Playwright"
@@ -22,20 +23,19 @@ build: bililive
.PHONY: build
bililive:
@go run build.go release
@go run build.go release $(if $(VERSION),--version $(VERSION),) $(if $(PLATFORM),--platform $(PLATFORM),) $(if $(ARCH),--arch $(ARCH),) $(if $(NO_TELEMETRY),--no-telemetry,)
# 开发版构建(支持可选参数)
# 用法:
# make dev - 普通开发构建
# make dev VERSION=v0.8.0-rc.1 - 指定版本号
# make dev PLATFORM=linux ARCH=amd64 - 交叉编译
# make dev PLATFORM=linux ARCH=amd64 VERSION=v0.8.0-rc.2.1 - 交叉编译 + 指定版本号
# make dev NO_TELEMETRY=1 - 禁用启动统计上报
.PHONY: dev
dev:
@go run build.go dev
@go run build.go dev $(if $(VERSION),--version $(VERSION),) $(if $(PLATFORM),--platform $(PLATFORM),) $(if $(ARCH),--arch $(ARCH),) $(if $(NO_TELEMETRY),--no-telemetry,)
# 自定义版本号编译(用于本地测试升级功能)
# 用法: make dev-version VERSION=v0.8.0-rc.1
.PHONY: dev-version
dev-version:
ifndef VERSION
$(error VERSION is required. Usage: make dev-version VERSION=v0.8.0-rc.1)
endif
@go run build.go dev --version $(VERSION)
# 收集所有 Go 源文件作为依赖(使用通配符,兼容 Windows)
GO_SOURCES := $(wildcard src/**/*.go src/**/**/*.go src/**/**/**/*.go)
+2 -1
View File
@@ -8,6 +8,7 @@ require (
github.com/Microsoft/go-winio v0.6.2
github.com/alecthomas/kingpin v2.2.7-0.20180312062423-a39589180ebd+incompatible
github.com/bluele/gcache v0.0.0-20190518031135-bc40bd653833
github.com/bluenviron/mediacommon/v2 v2.7.2
github.com/getsentry/sentry-go v0.31.1
github.com/go-delve/delve v1.26.0
github.com/golang-migrate/migrate/v4 v4.19.1
@@ -20,7 +21,7 @@ require (
github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b
github.com/shirou/gopsutil/v3 v3.24.5
github.com/sirupsen/logrus v1.9.3
github.com/stretchr/testify v1.10.0
github.com/stretchr/testify v1.11.1
github.com/tidwall/gjson v1.9.3
go.uber.org/mock v0.5.2
golang.org/x/net v0.48.0
+4 -2
View File
@@ -28,6 +28,8 @@ github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM
github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ=
github.com/bluele/gcache v0.0.0-20190518031135-bc40bd653833 h1:yCfXxYaelOyqnia8F/Yng47qhmfC9nKTRIbYRrRueq4=
github.com/bluele/gcache v0.0.0-20190518031135-bc40bd653833/go.mod h1:8c4/i2VlovMO2gBnHGQPN5EJw+H0lx1u/5p+cgsXtCk=
github.com/bluenviron/mediacommon/v2 v2.7.2 h1:BA7X7YOKq4VC6Ygm5FIg2D2IYTt/Ud6T43anXkHbDE8=
github.com/bluenviron/mediacommon/v2 v2.7.2/go.mod h1:5V15TiOfeaNVmZPVuOqAwqQSWyvMV86/dijDKu5q9Zs=
github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
@@ -236,8 +238,8 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
github.com/tidwall/gjson v1.9.3 h1:hqzS9wAHMO+KVBBkLxYdkEeeFHuqr95GfClRLKlgK0E=
github.com/tidwall/gjson v1.9.3/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
+5
View File
@@ -53,6 +53,11 @@ func GetBuildFlags(isDev bool) BuildFlags {
data["SentryDSN"] = sentryDSN
}
// 禁用遥测统计(用于本地测试,避免污染生产统计数据)
if os.Getenv("NO_TELEMETRY") == "1" {
ldFlagsTmpl += " -X main.DisableTelemetry=1"
}
t := template.Must(template.New("ldFlags").Parse(ldFlagsTmpl))
t.Execute(&buf, data)
+30 -3
View File
@@ -21,6 +21,9 @@ const sourceAgentFile = "AGENTS.md"
// 全局变量,用于存储命令行参数
var customVersion string
var customPlatform string
var customArch string
var customNoTelemetry bool
func RunCmd() int {
app := kingpin.New("Build tool", "bililive-go Build tool.")
@@ -28,10 +31,20 @@ func RunCmd() int {
// dev 命令支持 --version 参数
devCmd := app.Command("dev", "Build for development.")
devCmd.Flag("version", "自定义版本号(用于测试升级功能)").StringVar(&customVersion)
devCmd.Flag("platform", "目标平台(如 linux、darwin、windows)").StringVar(&customPlatform)
devCmd.Flag("arch", "目标架构(如 amd64、arm64)").StringVar(&customArch)
devCmd.Flag("no-telemetry", "禁用启动统计上报").BoolVar(&customNoTelemetry)
devCmd.Action(devBuild)
app.Command("dev-incremental", "增量构建:只在源码变化时重新编译(用于调试)").Action(devIncrementalBuild)
app.Command("release", "Build for release.").Action(releaseBuild)
// release 命令也支持自定义参数(用于本地编译测试版本)
releaseCmd := app.Command("release", "Build for release.")
releaseCmd.Flag("version", "自定义版本号").StringVar(&customVersion)
releaseCmd.Flag("platform", "目标平台(如 linux、darwin、windows)").StringVar(&customPlatform)
releaseCmd.Flag("arch", "目标架构(如 amd64、arm64)").StringVar(&customArch)
releaseCmd.Flag("no-telemetry", "禁用启动统计上报").BoolVar(&customNoTelemetry)
releaseCmd.Action(releaseBuild)
app.Command("release-docker", "Build for release docker.").Action(releaseDocker)
app.Command("test", "Run tests.").Action(goTest)
app.Command("generate", "go generate ./...").Action(goGenerate)
@@ -44,11 +57,24 @@ func RunCmd() int {
return 0
}
func devBuild(c *kingpin.ParseContext) error {
// 如果指定了自定义版本号,设置环境变量供 GetBuildFlags 使用
// applyCustomFlags 将命令行参数设置为环境变量供构建函数使用
func applyCustomFlags() {
if customVersion != "" {
os.Setenv("APP_VERSION", customVersion)
}
if customPlatform != "" {
os.Setenv("PLATFORM", customPlatform)
}
if customArch != "" {
os.Setenv("ARCH", customArch)
}
if customNoTelemetry {
os.Setenv("NO_TELEMETRY", "1")
}
}
func devBuild(c *kingpin.ParseContext) error {
applyCustomFlags()
BuildGoBinary(true)
return nil
}
@@ -59,6 +85,7 @@ func devIncrementalBuild(c *kingpin.ParseContext) error {
}
func releaseBuild(c *kingpin.ParseContext) error {
applyCustomFlags()
BuildGoBinary(false)
return nil
}
+17
View File
@@ -152,12 +152,29 @@ var defaultTaskQueue = TaskQueue{
MaxConcurrent: 3,
}
// ProxyEntry 单个代理配置项
type ProxyEntry struct {
Enable bool `yaml:"enable" json:"enable"`
URL string `yaml:"url" json:"url"`
}
// Proxy 代理配置
// 支持通用代理、信息获取代理和下载代理三个层级
// 优先级:专用代理(如果设置)> 通用代理 > 系统环境变量
type Proxy struct {
// Enable 是否启用配置的代理(false 时使用系统环境变量 HTTP_PROXY 等)
Enable bool `yaml:"enable" json:"enable"`
// URL 代理地址,支持 http://host:port 或 socks5://host:port
URL string `yaml:"url" json:"url"`
// InfoProxy 信息获取专用代理(覆盖通用设置)
// 用于获取直播间信息、平台 API 请求等
// 注意:通过 bililive-tools 间接获取信息的平台(如抖音)暂不受此代理设置影响
InfoProxy *ProxyEntry `yaml:"info_proxy,omitempty" json:"info_proxy,omitempty"`
// DownloadProxy 下载专用代理(覆盖通用设置)
// 用于下载直播流数据
DownloadProxy *ProxyEntry `yaml:"download_proxy,omitempty" json:"download_proxy,omitempty"`
}
var defaultProxy = Proxy{
+13 -3
View File
@@ -79,14 +79,24 @@ func DecorateConfigNode(node *yaml.Node) {
proxyNode := findNode(root, "proxy")
if proxyNode != nil {
setFieldComment(proxyNode, "enable",
`# 是否启用配置的代理
`# 通用代理开关
# false: 使用系统环境变量 (HTTP_PROXY, HTTPS_PROXY, ALL_PROXY)
# true: 使用下方配置的代理地址`, "")
setFieldComment(proxyNode, "url",
`# 代理地址,支持以下格式:
`# 通用代理地址,支持以下格式:
# HTTP 代理: http://host:port 或 http://user:pass@host:port
# SOCKS5 代理: socks5://host:port 或 socks5://user:pass@host:port
# 示例: socks5://127.0.0.1:1080 (翻墙软件常用端口)`, "")
# 示例: socks5://127.0.0.1:1080 (翻墙软件常用端口)
# 此地址同时用于信息获取和下载,除非下方单独配置了专用代理`, "")
setFieldComment(proxyNode, "info_proxy",
`# 信息获取专用代理(可选,覆盖通用代理设置)
# 仅用于获取直播间信息、平台 API 请求等
# 注意:通过 bililive-tools 间接获取信息的平台(如抖音)暂不受此代理设置影响
# 如果只想为信息获取使用代理(例如解决临时 IP 封禁),可以只配置此项`, "")
setFieldComment(proxyNode, "download_proxy",
`# 下载专用代理(可选,覆盖通用代理设置)
# 仅用于下载直播流数据
# 如果不想让下载流量走代理,可以将此项的 enable 设为 false`, "")
}
// Feature 功能配置注释
+10
View File
@@ -0,0 +1,10 @@
package configs
// 功能特征标志
// 用于控制正在开发中的功能是否启用
// 开发者可以在本地修改这些值来启用/禁用功能
// EnableProxyConfig 控制是否启用代理配置功能
// false: 隐藏前端代理配置 UI,后端代理函数回退到环境变量
// true: 启用完整的代理配置(通用代理 + 信息获取代理 + 下载代理)
const EnableProxyConfig = false
+6
View File
@@ -0,0 +1,6 @@
//go:build dev
package consts
// IsDevBuild 标记当前是否为开发编译版本
const IsDevBuild = true
+6
View File
@@ -0,0 +1,6 @@
//go:build !dev
package consts
// IsDevBuild 标记当前是否为开发编译版本
const IsDevBuild = false
+4
View File
@@ -59,6 +59,8 @@ type Info struct {
Initializing bool
CustomLiveId string
AudioOnly bool
// 最近一次 API 请求的错误信息(用于前端显示错误提示)
LastError string
// 可用流列表(最近一次获取的)
AvailableStreams []*AvailableStreamInfo
// 可用流更新时间
@@ -86,6 +88,7 @@ func (i *Info) MarshalJSON() ([]byte, error) {
LastStartTimeUnix int64 `json:"last_start_time_unix,omitempty"`
AudioOnly bool `json:"audio_only"`
NickName string `json:"nick_name"`
LastError string `json:"last_error,omitempty"`
AvailableStreams []*AvailableStreamInfo `json:"available_streams,omitempty"`
AvailableStreamsUpdatedAt int64 `json:"available_streams_updated_at,omitempty"`
}{
@@ -100,6 +103,7 @@ func (i *Info) MarshalJSON() ([]byte, error) {
Initializing: i.Initializing,
AudioOnly: i.AudioOnly,
NickName: i.Live.GetOptions().NickName,
LastError: i.LastError,
AvailableStreams: i.AvailableStreams,
AvailableStreamsUpdatedAt: i.AvailableStreamsUpdatedAt,
}
+5 -1
View File
@@ -315,11 +315,15 @@ func (w *WrappedLive) GetInfo() (*Info, error) {
if err != nil {
if info, err2 := w.cache.Get(w); err2 == nil {
info.(*Info).RoomName = err.Error()
// 将错误信息存到 LastError 而非 RoomName
// 避免错误文本出现在录制文件名中
info.(*Info).LastError = err.Error()
}
return nil, err
}
if w.cache != nil {
// 成功获取信息,清除之前的错误
i.LastError = ""
w.cache.Set(w, i)
}
+32 -2
View File
@@ -63,8 +63,12 @@ func (c *MemoryCollector) run() {
// RecorderManagerProvider 录制器管理器提供者
type RecorderManagerProvider func() recorders.Manager
// LauncherPIDProvider 启动器 PID 提供者
type LauncherPIDProvider func() int
var (
recorderManagerProvider RecorderManagerProvider
launcherPIDProvider LauncherPIDProvider
)
// SetRecorderManagerProvider 设置录制器管理器提供者
@@ -72,6 +76,11 @@ func SetRecorderManagerProvider(provider RecorderManagerProvider) {
recorderManagerProvider = provider
}
// SetLauncherPIDProvider 设置启动器 PID 提供者
func SetLauncherPIDProvider(provider LauncherPIDProvider) {
launcherPIDProvider = provider
}
// collect 执行一次内存数据采集
func (c *MemoryCollector) collect() {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@@ -152,7 +161,22 @@ func (c *MemoryCollector) collect() {
}
}
// 4. 收集容器内存(仅 Linux 容器环境)
// 4. 收集 launcher 进程内存(如果由启动器管理)
if launcherPIDProvider != nil {
if launcherPID := launcherPIDProvider(); launcherPID > 0 {
if procMem, err := memstats.GetProcessMemory(launcherPID); err == nil {
stats = append(stats, &MemoryStat{
Timestamp: timestamp,
Category: MemoryCategoryLauncher,
RSS: procMem.RSS,
VMS: procMem.VMS,
})
totalRSS += procMem.RSS
}
}
}
// 5. 收集容器内存(仅 Linux 容器环境)
if memstats.IsInContainer() {
containerMem, err := memstats.GetContainerMemory()
if err == nil && containerMem != nil {
@@ -161,10 +185,16 @@ func (c *MemoryCollector) collect() {
Category: MemoryCategoryContainer,
RSS: containerMem.Used,
})
// 在容器环境中,使用 cgroup 报告的内存作为“总内存”
// cgroup memory.current 包含了容器内所有进程的实际内存使用
// 比简单累加各进程 RSS 更准确(避免共享内存重复计算)
if containerMem.Used > 0 {
totalRSS = containerMem.Used
}
}
}
// 5. 记录总内存
// 6. 记录总内存
stats = append(stats, &MemoryStat{
Timestamp: timestamp,
Category: MemoryCategoryTotal,
+2
View File
@@ -40,6 +40,8 @@ const (
MemoryCategoryKlive = "klive"
// MemoryCategoryRecorder BililiveRecorder 子进程
MemoryCategoryRecorder = "bililive-recorder"
// MemoryCategoryLauncher 启动器进程
MemoryCategoryLauncher = "launcher"
// MemoryCategoryContainer 容器内存
MemoryCategoryContainer = "container"
// MemoryCategoryTotal 总内存(self + 所有子进程)
+53 -10
View File
@@ -191,6 +191,7 @@ type Runner struct {
mainPID int
startupOK bool
startupCh chan struct{} // startup_success 收到时关闭此 channel
processDone chan struct{} // 子进程退出时关闭此 channel
verbose bool
}
@@ -277,6 +278,28 @@ func (r *Runner) Run(ctx context.Context, args []string) error {
// 启动成功确认,立即进入等待阶段
startupTimer.Stop()
r.log("收到启动确认,进入运行等待")
case <-r.processDone:
// 子进程在发送 startup_success 之前就退出了(如 Fatal 错误)
startupTimer.Stop()
if !r.startupOK {
r.log("主程序在启动确认前退出")
// 增加失败计数
r.state.FailureCount++
r.state.Save(r.statePath)
// 如果失败次数超过限制,尝试回滚
if r.state.FailureCount >= r.state.MaxRetries && r.state.BackupBinaryPath != "" {
r.log("启动失败次数过多(%d/%d),尝试回滚...", r.state.FailureCount, r.state.MaxRetries)
if err := r.rollback(); err != nil {
r.log("回滚失败: %v", err)
}
} else {
r.log("启动失败 %d/%d 次,等待后重试...", r.state.FailureCount, r.state.MaxRetries)
time.Sleep(2 * time.Second) // 短暂等待后重试,避免快速循环
}
continue
}
case <-startupTimer.C:
if !r.startupOK {
r.log("主程序启动超时")
@@ -288,7 +311,7 @@ func (r *Runner) Run(ctx context.Context, args []string) error {
// 如果失败次数超过限制,尝试回滚
if r.state.FailureCount >= r.state.MaxRetries && r.state.BackupBinaryPath != "" {
r.log("启动失败次数过多,尝试回滚...")
r.log("启动失败次数过多(%d/%d),尝试回滚...", r.state.FailureCount, r.state.MaxRetries)
if err := r.rollback(); err != nil {
r.log("回滚失败: %v", err)
}
@@ -358,6 +381,16 @@ func (r *Runner) startMainProgram(ctx context.Context, args []string) error {
r.mainPID = r.mainProcess.Process.Pid
r.startupOK = false
// 启动一个 goroutine 监听子进程退出
// 这样在 Run() 的 select 中可以立即检测到子进程崩溃
r.processDone = make(chan struct{})
go func() {
defer bilisentry.Recover()
r.mainProcess.Wait()
close(r.processDone)
}()
r.log("主程序已启动,PID: %d", r.mainPID)
return nil
@@ -376,16 +409,21 @@ func (r *Runner) stopMainProgram() {
})
r.server.Broadcast(msg)
// 等待进程退出
done := make(chan struct{})
go func() {
defer bilisentry.Recover()
r.mainProcess.Wait()
close(done)
}()
// 复用 processDone channel(由 startMainProgram 的 goroutine 统一管理 Wait())
// 避免多次调用 Wait() 导致 panic
waitCh := r.processDone
if waitCh == nil {
// 兜底:如果 processDone 不存在(不应该发生),直接 Wait
waitCh = make(chan struct{})
go func() {
defer bilisentry.Recover()
r.mainProcess.Wait()
close(waitCh)
}()
}
select {
case <-done:
case <-waitCh:
r.log("主程序已正常退出")
case <-time.After(35 * time.Second):
r.log("主程序未响应,强制终止")
@@ -398,7 +436,12 @@ func (r *Runner) waitForMainProgram() {
if r.mainProcess == nil {
return
}
r.mainProcess.Wait()
// 等待 processDone channel 关闭(由 startMainProgram 中的 goroutine 负责)
if r.processDone != nil {
<-r.processDone
} else {
r.mainProcess.Wait()
}
r.log("主程序已退出")
}
+23
View File
@@ -137,6 +137,18 @@ func (p *Parser) Status() (map[string]interface{}, error) {
return <-p.statusResp, nil
}
// ParseLiveStream 启动 FFmpeg 进程录制直播流。
//
// ⚠️ 已知问题(负负得正):
// 本函数的 ctx 参数实际上无法取消函数执行——核心阻塞点 cmd.Wait()(第 272 行)不监听 ctx.Done(),
// FFmpeg 进程的终止完全依赖外部调用 Stop() 方法。
//
// 然而这个"bug"意外地保护了录制行为:调用链中的 ctx 可能源自 HTTP handler 的 request context
// (handler → AddRecorder → recorder.Start → run → tryRecord → ParseLiveStream),
// 若 ParseLiveStream 正确响应 ctx 取消,录制会在 HTTP 请求结束后被意外终止。
//
// 正确修复需要同时解决整条 context 传播链路(将 request context 替换为应用级 context),
// 影响范围广,暂不在此处理。当前录制的停止完全由 recorder.Close() → parser.Stop() 控制。
func (p *Parser) ParseLiveStream(ctx context.Context, streamUrlInfo *live.StreamUrlInfo, live live.Live, file string) (err error) {
url := streamUrlInfo.Url
ffmpegPath, err := utils.GetFFmpegPathForLive(ctx, live)
@@ -269,6 +281,8 @@ func (p *Parser) ParseLiveStream(ctx context.Context, streamUrlInfo *live.Stream
}
bilisentry.Go(p.scheduler)
// 注意:cmd.Wait() 不监听 ctx.Done(),见函数顶部注释。
// 停止 FFmpeg 的唯一途径是通过 Stop() 方法。
err = p.cmd.Wait()
// 停止 FLV 代理
@@ -318,6 +332,15 @@ func (p *Parser) Stop() (err error) {
if _, err = p.cmdStdIn.Write([]byte("q")); err != nil {
err = fmt.Errorf("error sending stop command to ffmpeg: %v", err)
}
// 启动强制退出 goroutine:如果 FFmpeg 3 秒内未响应 "q" 命令退出
// (例如正在等待 HLS m3u8 网络超时),则强制杀掉进程
cmd := p.cmd
go func() {
time.Sleep(3 * time.Second)
if cmd.ProcessState == nil {
_ = cmd.Process.Kill()
}
}()
} else if p.cmdStdIn == nil {
err = fmt.Errorf("p.cmdStdIn == nil")
} else if p.cmd.Process == nil {
+122 -37
View File
@@ -1,4 +1,6 @@
// Package proxy 提供代理配置和管理功能
// 支持通用代理、信息获取代理和下载代理三个层级
// 优先级:专用代理(如果设置)> 通用代理 > 系统环境变量
package proxy
import (
@@ -13,12 +15,39 @@ import (
"golang.org/x/net/proxy"
)
// GetProxyURL 获取当前生效的代理 URL
// 优先级:配置文件 > 环境变量 (ALL_PROXY > HTTPS_PROXY > HTTP_PROXY)
func GetProxyURL() string {
// proxyScope 代理用途
type proxyScope int
const (
// scopeGeneral 通用代理
scopeGeneral proxyScope = iota
// scopeInfo 信息获取代理(获取直播间信息、平台 API 请求等)
scopeInfo
// scopeDownload 下载代理(下载直播流数据)
scopeDownload
)
// getProxyURLForScope 获取指定用途的代理 URL
// 优先级:专用代理(如果设置)> 通用代理 > 环境变量
func getProxyURLForScope(scope proxyScope) string {
cfg := configs.GetCurrentConfig()
if cfg != nil && cfg.Proxy.Enable && cfg.Proxy.URL != "" {
return cfg.Proxy.URL
if cfg != nil && configs.EnableProxyConfig {
// 先检查专用代理
switch scope {
case scopeInfo:
if cfg.Proxy.InfoProxy != nil && cfg.Proxy.InfoProxy.Enable && cfg.Proxy.InfoProxy.URL != "" {
return cfg.Proxy.InfoProxy.URL
}
case scopeDownload:
if cfg.Proxy.DownloadProxy != nil && cfg.Proxy.DownloadProxy.Enable && cfg.Proxy.DownloadProxy.URL != "" {
return cfg.Proxy.DownloadProxy.URL
}
}
// 回退到通用代理
if cfg.Proxy.Enable && cfg.Proxy.URL != "" {
return cfg.Proxy.URL
}
}
// 按优先级检查环境变量
@@ -31,32 +60,35 @@ func GetProxyURL() string {
return ""
}
// GetProxyFunc 返回 HTTP 代理函数
// 用于 http.Transport.Proxy 字段
func GetProxyFunc() func(*http.Request) (*url.URL, error) {
proxyURL := GetProxyURL()
if proxyURL == "" {
return nil
}
// SOCKS5 代理不能用于 HTTP Proxy 字段,需要通过 DialContext 处理
if strings.HasPrefix(proxyURL, "socks5://") || strings.HasPrefix(proxyURL, "socks5h://") {
return nil
}
// 解析代理 URL
parsedURL, err := url.Parse(proxyURL)
if err != nil {
return nil
}
return http.ProxyURL(parsedURL)
// GetProxyURL 获取通用代理 URL
// 优先级:配置文件 > 环境变量 (ALL_PROXY > HTTPS_PROXY > HTTP_PROXY)
func GetProxyURL() string {
return getProxyURLForScope(scopeGeneral)
}
// IsSocks5Proxy 检查当前代理是否为 SOCKS5 代理
// GetInfoProxyURL 获取信息获取代理 URL
// 用于获取直播间信息、平台 API 请求等
// 优先级:InfoProxy > 通用 Proxy > 环境变量
func GetInfoProxyURL() string {
return getProxyURLForScope(scopeInfo)
}
// GetDownloadProxyURL 获取下载代理 URL
// 用于下载直播流数据
// 优先级:DownloadProxy > 通用 Proxy > 环境变量
func GetDownloadProxyURL() string {
return getProxyURLForScope(scopeDownload)
}
// GetProxyFunc 返回 HTTP 代理函数(通用代理)
// 用于 http.Transport.Proxy 字段
func GetProxyFunc() func(*http.Request) (*url.URL, error) {
return getProxyFuncForURL(GetProxyURL())
}
// IsSocks5Proxy 检查通用代理是否为 SOCKS5 代理
func IsSocks5Proxy() bool {
proxyURL := GetProxyURL()
return strings.HasPrefix(proxyURL, "socks5://") || strings.HasPrefix(proxyURL, "socks5h://")
return isSocks5(GetProxyURL())
}
// CreateSocks5DialContext 为 SOCKS5 代理创建 DialContext 函数
@@ -69,7 +101,7 @@ func CreateSocks5DialContext(proxyURL string) func(ctx context.Context, network,
return nil, err
}
// 构建 SOCKS5 地址
// 构建 SOCKS5 认证信息
var auth *proxy.Auth
if parsedURL.User != nil {
auth = &proxy.Auth{
@@ -91,28 +123,81 @@ func CreateSocks5DialContext(proxyURL string) func(ctx context.Context, network,
}
}
// ApplyProxyToTransport 将代理设置应用到 http.Transport
// ApplyProxyToTransport 将通用代理设置应用到 http.Transport
func ApplyProxyToTransport(transport *http.Transport) {
proxyURL := GetProxyURL()
applyProxyURLToTransport(transport, GetProxyURL())
}
// ApplyInfoProxyToTransport 将信息获取代理设置应用到 http.Transport
// 用于获取直播间信息、平台 API 请求等
func ApplyInfoProxyToTransport(transport *http.Transport) {
applyProxyURLToTransport(transport, GetInfoProxyURL())
}
// ApplyDownloadProxyToTransport 将下载代理设置应用到 http.Transport
// 用于下载直播流数据
func ApplyDownloadProxyToTransport(transport *http.Transport) {
applyProxyURLToTransport(transport, GetDownloadProxyURL())
}
// GetProxyEnvVars 返回用于子进程的通用代理环境变量
// 可用于传递给外部工具(如 ffmpeg、dotnet 等)
func GetProxyEnvVars() []string {
return getProxyEnvVarsForURL(GetProxyURL())
}
// GetDownloadProxyEnvVars 返回用于子进程的下载代理环境变量
// 可用于传递给下载相关的外部工具
func GetDownloadProxyEnvVars() []string {
return getProxyEnvVarsForURL(GetDownloadProxyURL())
}
// ==================== 内部辅助函数 ====================
// isSocks5 检查代理 URL 是否为 SOCKS5 代理
func isSocks5(proxyURL string) bool {
return strings.HasPrefix(proxyURL, "socks5://") || strings.HasPrefix(proxyURL, "socks5h://")
}
// getProxyFuncForURL 为指定的代理 URL 返回 HTTP 代理函数
func getProxyFuncForURL(proxyURL string) func(*http.Request) (*url.URL, error) {
if proxyURL == "" {
return nil
}
// SOCKS5 代理不能用于 HTTP Proxy 字段,需要通过 DialContext 处理
if isSocks5(proxyURL) {
return nil
}
// 解析代理 URL
parsedURL, err := url.Parse(proxyURL)
if err != nil {
return nil
}
return http.ProxyURL(parsedURL)
}
// applyProxyURLToTransport 将指定的代理 URL 应用到 http.Transport
func applyProxyURLToTransport(transport *http.Transport, proxyURL string) {
if proxyURL == "" {
// 使用系统默认代理(从环境变量)
transport.Proxy = http.ProxyFromEnvironment
return
}
if strings.HasPrefix(proxyURL, "socks5://") || strings.HasPrefix(proxyURL, "socks5h://") {
if isSocks5(proxyURL) {
// SOCKS5 代理需要通过 DialContext 处理
transport.DialContext = CreateSocks5DialContext(proxyURL)
} else {
// HTTP/HTTPS 代理
transport.Proxy = GetProxyFunc()
transport.Proxy = getProxyFuncForURL(proxyURL)
}
}
// GetProxyEnvVars 返回用于子进程的代理环境变量
// 可用于传递给外部工具(如 ffmpeg、dotnet 等)
func GetProxyEnvVars() []string {
proxyURL := GetProxyURL()
// getProxyEnvVarsForURL 为指定的代理 URL 返回环境变量列表
func getProxyEnvVarsForURL(proxyURL string) []string {
if proxyURL == "" {
return nil
}
+601
View File
@@ -0,0 +1,601 @@
package streamprobe
import (
"encoding/binary"
"errors"
"fmt"
"io"
"math"
"github.com/bluenviron/mediacommon/v2/pkg/codecs/h264"
"github.com/bluenviron/mediacommon/v2/pkg/codecs/h265"
)
// FLV 常量
const (
// FLV tag types
flvTagAudio uint8 = 8
flvTagVideo uint8 = 9
flvTagScript uint8 = 18
// FLV header size
flvHeaderSize = 9
// Video CodecID (lower 4 bits of first video tag byte)
codecAVC uint8 = 7 // H.264
codecHEVC uint8 = 12 // H.265 (非标准 FLV 扩展)
// AVC packet type
avcSeqHeader uint8 = 0
avcNALU uint8 = 1
// Audio CodecID (upper 4 bits of first audio tag byte)
audioCodecAAC uint8 = 10
audioCodecMP3 uint8 = 2
audioCodecOPUS uint8 = 13 // 非标准扩展
// FLV Enhanced Header
// 参考: https://github.com/nickharvey/m3u/wiki/Enhanced-FLV
frameTypeCommandFrame uint8 = 5 // Enhanced header 命令帧标志
// Enhanced FLV PacketType
enhancedPacketTypeSequenceStart uint8 = 0
// Enhanced FLV FourCC
fourCCAV1 = "av01"
fourCCVP9 = "vp09"
fourCCHEVC = "hvc1"
)
var (
// ErrNotFLV 表示数据不是有效的 FLV 格式
ErrNotFLV = errors.New("不是有效的 FLV 格式")
// ErrTruncated 表示 FLV 数据被截断
ErrTruncated = errors.New("FLV 数据被截断")
)
// parseFLVHeader 解析 FLV 文件头(9 字节)
// 返回头部的原始字节数据
func parseFLVHeader(r io.Reader) ([]byte, error) {
header := make([]byte, flvHeaderSize)
if _, err := io.ReadFull(r, header); err != nil {
return nil, fmt.Errorf("读取 FLV 头失败: %w", err)
}
// 验证 FLV 签名 "FLV"
if header[0] != 'F' || header[1] != 'L' || header[2] != 'V' {
return nil, ErrNotFLV
}
return header, nil
}
// flvTagHeader 表示一个 FLV tag 的头部信息
type flvTagHeader struct {
PreviousTagSize uint32
TagType uint8
DataSize uint32
Timestamp uint32
}
// readFlvTagHeader 从 reader 中读取一个 FLV tag header( PreviousTagSize(4) + TagType(1) + DataSize(3) + Timestamp(3) + TimestampExtended(1) + StreamId(3) = 15 bytes)
func readFlvTagHeader(r io.Reader) (*flvTagHeader, []byte, error) {
buf := make([]byte, 15)
if _, err := io.ReadFull(r, buf); err != nil {
return nil, nil, err
}
tag := &flvTagHeader{
PreviousTagSize: binary.BigEndian.Uint32(buf[0:4]),
TagType: buf[4],
DataSize: uint32(buf[5])<<16 | uint32(buf[6])<<8 | uint32(buf[7]),
Timestamp: uint32(buf[8])<<16 | uint32(buf[9])<<8 | uint32(buf[10]) | uint32(buf[11])<<24,
}
return tag, buf, nil
}
// parseFLVStreamInfo 从 FLV 流的前几个 tag 中解析流信息
// 最多读取 maxTags 个 tag,避免在遇到问题时无限读取
func parseFLVStreamInfo(r io.Reader, maxTags int) (*StreamHeaderInfo, []byte, error) {
info := &StreamHeaderInfo{}
// 解析 FLV 头
headerData, err := parseFLVHeader(r)
if err != nil {
return nil, nil, err
}
// 收集所有已读取的数据(用于后续转发)
buffered := make([]byte, 0, 64*1024) // 预分配 64KB
buffered = append(buffered, headerData...)
// 读取 tag,直到获取到足够的信息或达到上限
gotVideo := false
gotAudio := false
gotScript := false
for i := 0; i < maxTags; i++ {
// 读取 tag header
tagHeader, tagHeaderData, err := readFlvTagHeader(r)
if err != nil {
if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
break
}
return nil, nil, fmt.Errorf("读取 FLV tag header 失败: %w", err)
}
buffered = append(buffered, tagHeaderData...)
// 读取 tag body
tagData := make([]byte, tagHeader.DataSize)
if _, err := io.ReadFull(r, tagData); err != nil {
if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
break
}
return nil, nil, fmt.Errorf("读取 FLV tag body 失败: %w", err)
}
buffered = append(buffered, tagData...)
// 根据 tag 类型解析
switch tagHeader.TagType {
case flvTagScript:
if !gotScript {
parseScriptTag(tagData, info)
gotScript = true
}
case flvTagVideo:
if !gotVideo {
parseVideoTag(tagData, info)
if info.ParsedFromSPS || info.Unsupported {
gotVideo = true
}
}
case flvTagAudio:
if !gotAudio {
parseAudioTag(tagData, info)
gotAudio = true
}
}
// 已获取所有需要的信息
if (gotVideo || info.Unsupported) && gotAudio && gotScript {
break
}
}
return info, buffered, nil
}
// parseVideoTag 解析视频 tag,提取编码格式和 SPS 信息
func parseVideoTag(data []byte, info *StreamHeaderInfo) {
if len(data) < 2 {
return
}
frameType := (data[0] >> 4) & 0x0F
codecID := data[0] & 0x0F
// 检查是否为 Enhanced FLV header
// Enhanced FLV: FrameType 的高 4 位第 8 位为标志位
// 如果 frameType 的最高位(bit 7)被设置,说明是 Enhanced FLV
isEnhanced := (data[0] & 0x80) != 0
if isEnhanced {
parseEnhancedVideoTag(data, info)
return
}
// 标准 FLV 视频 tag
switch codecID {
case codecAVC:
info.VideoCodec = "h264"
if len(data) >= 5 {
// AVCPacketType
avcPacketType := data[1]
if avcPacketType == avcSeqHeader {
// AVC Sequence Header - 包含 SPS/PPS
parseAVCDecoderConfig(data[5:], info)
}
}
case codecHEVC:
info.VideoCodec = "h265"
if len(data) >= 5 {
avcPacketType := data[1]
if avcPacketType == avcSeqHeader {
// HEVC Decoder Configuration Record
parseHEVCDecoderConfig(data[5:], info)
}
}
default:
// 未知 or 命令帧
if frameType != frameTypeCommandFrame {
info.VideoCodec = "unknown"
info.Unsupported = true
info.UnsupportedMsg = fmt.Sprintf("未知的视频编码格式 (CodecID: %d)", codecID)
}
}
}
// parseEnhancedVideoTag 解析 Enhanced FLV 视频 tag
// Enhanced FLV 使用 FourCC 标识编码格式
func parseEnhancedVideoTag(data []byte, info *StreamHeaderInfo) {
if len(data) < 5 {
return
}
// Enhanced header 格式:
// byte 0: FrameType(4) + PacketType(4)
// byte 1-4: FourCC (4 bytes)
packetType := data[0] & 0x0F
fourCC := string(data[1:5])
switch fourCC {
case fourCCHEVC:
info.VideoCodec = "h265"
if packetType == enhancedPacketTypeSequenceStart && len(data) > 5 {
parseHEVCDecoderConfig(data[5:], info)
}
case fourCCAV1:
info.VideoCodec = "av1"
info.Unsupported = true
info.UnsupportedMsg = "AV1 编码格式暂不支持 SPS 深度解析,无法获取精确分辨率"
case fourCCVP9:
info.VideoCodec = "vp9"
info.Unsupported = true
info.UnsupportedMsg = "VP9 编码格式暂不支持深度解析,无法获取精确分辨率"
default:
info.VideoCodec = "unknown"
info.Unsupported = true
info.UnsupportedMsg = fmt.Sprintf("未知的 Enhanced FLV 编码格式 (FourCC: %s)", fourCC)
}
}
// parseAVCDecoderConfig 解析 AVC Decoder Configuration Record
// 从中提取 SPS 并解析分辨率
func parseAVCDecoderConfig(data []byte, info *StreamHeaderInfo) {
if len(data) < 8 {
return
}
// AVCDecoderConfigurationRecord 格式:
// configurationVersion(1) + AVCProfileIndication(1) + ...
// numOfSequenceParameterSets(1) & 0x1F = SPS 数量
// sequenceParameterSetLength(2) + spsNALUnit(N)
// 跳过前 5 个字节到 numOfSequenceParameterSets
numSPS := int(data[5] & 0x1F)
if numSPS == 0 {
return
}
offset := 6
if offset+2 > len(data) {
return
}
spsLen := int(binary.BigEndian.Uint16(data[offset : offset+2]))
offset += 2
if offset+spsLen > len(data) || spsLen == 0 {
return
}
spsData := data[offset : offset+spsLen]
// 使用 mediacommon 解析 SPS
var sps h264.SPS
if err := sps.Unmarshal(spsData); err != nil {
// SPS 解析失败,不影响录制
return
}
info.Width = sps.Width()
info.Height = sps.Height()
info.ParsedFromSPS = true
// 尝试获取帧率
if fps := sps.FPS(); fps > 0 && fps < 300 {
info.FrameRate = fps
}
}
// parseHEVCDecoderConfig 解析 HEVC Decoder Configuration Record
// 从中提取 SPS 并解析分辨率
func parseHEVCDecoderConfig(data []byte, info *StreamHeaderInfo) {
if len(data) < 23 {
return
}
// HEVCDecoderConfigurationRecord 格式:
// 参考 ISO 14496-15 Section 8.3.3.1.2
// 前 22 字节是固定配置
// byte 22: numOfArrays
numArrays := int(data[22])
offset := 23
for i := 0; i < numArrays; i++ {
if offset+3 > len(data) {
break
}
naluType := data[offset] & 0x3F
numNalus := int(binary.BigEndian.Uint16(data[offset+1 : offset+3]))
offset += 3
for j := 0; j < numNalus; j++ {
if offset+2 > len(data) {
return
}
naluLen := int(binary.BigEndian.Uint16(data[offset : offset+2]))
offset += 2
if offset+naluLen > len(data) {
return
}
naluData := data[offset : offset+naluLen]
offset += naluLen
// NAL type 33 = SPS
if naluType == 33 && naluLen > 0 {
var sps h265.SPS
if err := sps.Unmarshal(naluData); err != nil {
continue
}
info.Width = sps.Width()
info.Height = sps.Height()
info.ParsedFromSPS = true
if fps := sps.FPS(); fps > 0 && fps < 300 {
info.FrameRate = fps
}
return
}
}
}
}
// parseAudioTag 解析音频 tag,提取编码格式信息
func parseAudioTag(data []byte, info *StreamHeaderInfo) {
if len(data) < 1 {
return
}
soundFormat := (data[0] >> 4) & 0x0F
switch soundFormat {
case audioCodecAAC:
info.AudioCodec = "aac"
case audioCodecMP3:
info.AudioCodec = "mp3"
case audioCodecOPUS:
info.AudioCodec = "opus"
default:
info.AudioCodec = fmt.Sprintf("audio_%d", soundFormat)
}
}
// parseScriptTag 解析 Script (onMetaData) tag
// 从中提取 width, height, framerate, videodatarate, audiodatarate 等
func parseScriptTag(data []byte, info *StreamHeaderInfo) {
// AMF0 格式解析 onMetaData
// Script tag data 通常包含:
// 1. AMF0 string "onMetaData"
// 2. AMF0 ECMA array 或 object 包含元数据
metadata := parseAMF0(data)
if metadata == nil {
return
}
info.RawMetaData = metadata
// 提取视频信息
if width, ok := getNumberFromMeta(metadata, "width"); ok && width > 0 {
if info.Width == 0 { // 不覆盖 SPS 解析的值
info.Width = int(width)
info.ParsedFromMeta = true
}
}
if height, ok := getNumberFromMeta(metadata, "height"); ok && height > 0 {
if info.Height == 0 { // 不覆盖 SPS 解析的值
info.Height = int(height)
info.ParsedFromMeta = true
}
}
if framerate, ok := getNumberFromMeta(metadata, "framerate"); ok && framerate > 0 {
if info.FrameRate == 0 { // 不覆盖 SPS 解析的值
info.FrameRate = framerate
}
}
if videodatarate, ok := getNumberFromMeta(metadata, "videodatarate"); ok && videodatarate > 0 {
info.VideoBitrate = int(videodatarate)
}
if audiodatarate, ok := getNumberFromMeta(metadata, "audiodatarate"); ok && audiodatarate > 0 {
info.AudioBitrate = int(audiodatarate)
}
// 编码器信息
if videoCodecStr, ok := getStringFromMeta(metadata, "videocodecid"); ok {
if info.VideoCodec == "" {
info.VideoCodec = normalizeCodecName(videoCodecStr)
}
}
if audioCodecStr, ok := getStringFromMeta(metadata, "audiocodecid"); ok {
if info.AudioCodec == "" {
info.AudioCodec = normalizeCodecName(audioCodecStr)
}
}
}
// getNumberFromMeta 从 metadata 中获取数值类型的字段
func getNumberFromMeta(meta map[string]interface{}, key string) (float64, bool) {
v, ok := meta[key]
if !ok {
return 0, false
}
switch n := v.(type) {
case float64:
return n, true
case int:
return float64(n), true
case int64:
return float64(n), true
default:
return 0, false
}
}
// getStringFromMeta 从 metadata 中获取字符串类型的字段
func getStringFromMeta(meta map[string]interface{}, key string) (string, bool) {
v, ok := meta[key]
if !ok {
return "", false
}
s, ok := v.(string)
return s, ok
}
// normalizeCodecName 标准化编码名称
func normalizeCodecName(name string) string {
switch name {
case "avc1", "AVC", "7":
return "h264"
case "hvc1", "hev1", "HEVC", "12":
return "h265"
case "mp4a", "AAC", "10":
return "aac"
default:
return name
}
}
// ==================== 简化的 AMF0 解析器 ====================
// parseAMF0 解析 AMF0 格式的 script tag data
// 只提取 onMetaData 中的键值对
func parseAMF0(data []byte) map[string]interface{} {
if len(data) < 3 {
return nil
}
offset := 0
// 跳过第一个 AMF0 值(通常是 "onMetaData" 字符串)
if data[offset] == 0x02 { // AMF0 string type
offset++
if offset+2 > len(data) {
return nil
}
strLen := int(binary.BigEndian.Uint16(data[offset : offset+2]))
offset += 2 + strLen
} else {
return nil
}
if offset >= len(data) {
return nil
}
// 解析第二个 AMF0 值(ECMA array 或 object)
return readAMF0Value(data, &offset)
}
// readAMF0Value 读取一个 AMF0 值
func readAMF0Value(data []byte, offset *int) map[string]interface{} {
if *offset >= len(data) {
return nil
}
amfType := data[*offset]
*offset++
switch amfType {
case 0x03: // Object
return readAMF0Object(data, offset)
case 0x08: // ECMA Array
if *offset+4 > len(data) {
return nil
}
*offset += 4 // 跳过 array count
return readAMF0Object(data, offset)
default:
return nil
}
}
// readAMF0Object 读取 AMF0 object/ECMA array 的键值对
func readAMF0Object(data []byte, offset *int) map[string]interface{} {
result := make(map[string]interface{})
for *offset+2 < len(data) {
// 读取 key(AMF0 string without type marker)
keyLen := int(binary.BigEndian.Uint16(data[*offset : *offset+2]))
*offset += 2
if keyLen == 0 {
// 检查 object end marker (0x000009)
if *offset < len(data) && data[*offset] == 0x09 {
*offset++
}
break
}
if *offset+keyLen > len(data) {
break
}
key := string(data[*offset : *offset+keyLen])
*offset += keyLen
if *offset >= len(data) {
break
}
// 读取 value
valueType := data[*offset]
*offset++
switch valueType {
case 0x00: // Number (float64)
if *offset+8 > len(data) {
return result
}
bits := binary.BigEndian.Uint64(data[*offset : *offset+8])
result[key] = math.Float64frombits(bits)
*offset += 8
case 0x01: // Boolean
if *offset >= len(data) {
return result
}
result[key] = data[*offset] != 0
*offset++
case 0x02: // String
if *offset+2 > len(data) {
return result
}
strLen := int(binary.BigEndian.Uint16(data[*offset : *offset+2]))
*offset += 2
if *offset+strLen > len(data) {
return result
}
result[key] = string(data[*offset : *offset+strLen])
*offset += strLen
case 0x05: // Null
result[key] = nil
case 0x06: // Undefined
result[key] = nil
default:
// 遇到未知类型,停止解析避免越界
return result
}
}
return result
}
+852
View File
@@ -0,0 +1,852 @@
package streamprobe
import (
"bytes"
"context"
"encoding/binary"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/url"
"strings"
"time"
"github.com/bluenviron/mediacommon/v2/pkg/codecs/h264"
"github.com/bluenviron/mediacommon/v2/pkg/codecs/h265"
"github.com/bililive-go/bililive-go/src/pkg/livelogger"
"github.com/bililive-go/bililive-go/src/pkg/proxy"
)
// TS (MPEG-2 Transport Stream) 常量
const (
tsPacketSize = 188
tsSyncByte = 0x47
tsMaxReadBytes = 64 * 1024 // 最多读取 64KB 的 TS 数据
// PES stream type
streamTypeH264 = 0x1B
streamTypeH265 = 0x24
streamTypeAAC = 0x0F
streamTypeAC3 = 0x81
streamTypeMPEG = 0x03
// H.264 NAL Unit Types
nalTypeSPS = 7
nalTypeIDR = 5
nalTypeNonIDR = 1
// H.265 NAL Unit Types
hevcNalTypeSPS = 33
hevcNalTypeVPS = 32
hevcNalTypePPS = 34
)
// probeHLSStreamInfo 探测 HLS 流的实际信息
// 支持两种 HLS 格式:
// - TS 格式:解析 m3u8 → 下载 TS 分段头部 → 解析 TS packets → SPS
// - fMP4 格式(m4s):解析 m3u8 的 #EXT-X-MAP → 下载 init 段 → 解析 moov box → SPS
func probeHLSStreamInfo(
ctx context.Context,
m3u8Body io.Reader,
m3u8URL *url.URL,
headers map[string]string,
logger *livelogger.LiveLogger,
) (*StreamHeaderInfo, error) {
// 1. 读取 m3u8 内容
m3u8Content, err := io.ReadAll(m3u8Body)
if err != nil {
return nil, fmt.Errorf("读取 m3u8 失败: %w", err)
}
content := string(m3u8Content)
// 2. 检测是 fMP4 格式还是 TS 格式
// fMP4 格式的 m3u8 包含 #EXT-X-MAP 标签指向 init 段
initURL, err := parseEXTXMap(content, m3u8URL)
if err == nil && initURL != "" {
// fMP4 格式:下载 init 段并解析 moov box
return probeFMP4Init(ctx, initURL, headers, logger)
}
// TS 格式:走原有逻辑
segmentURL, err := parseFirstSegmentURL(content, m3u8URL)
if err != nil {
return nil, fmt.Errorf("解析 m3u8 分段失败: %w", err)
}
if logger != nil {
logger.Infof("HLS 探测: 解析到第一个 TS 分段 URL: %s", segmentURL)
}
// 下载 TS 分段头部
tsData, err := downloadSegmentHeader(ctx, segmentURL, headers, tsMaxReadBytes)
if err != nil {
return nil, fmt.Errorf("下载 TS 分段头部失败: %w", err)
}
if logger != nil {
logger.Infof("HLS 探测: 已下载 TS 分段头部 %d 字节", len(tsData))
}
info, err := parseTSData(tsData)
if err != nil {
return nil, fmt.Errorf("解析 TS 数据失败: %w", err)
}
return info, nil
}
// parseEXTXMap 从 m3u8 内容中解析 #EXT-X-MAP 标签的 URI
// 返回 init 段的完整 URL。如果没有 #EXT-X-MAP 标签,返回空字符串。
func parseEXTXMap(content string, baseURL *url.URL) (string, error) {
for _, line := range strings.Split(content, "\n") {
line = strings.TrimSpace(line)
if !strings.HasPrefix(line, "#EXT-X-MAP:") {
continue
}
// 解析 URI="..." 属性
attrs := line[len("#EXT-X-MAP:"):]
uri := ""
for _, attr := range strings.Split(attrs, ",") {
attr = strings.TrimSpace(attr)
if strings.HasPrefix(attr, "URI=") {
uri = strings.Trim(attr[4:], `"`)
break
}
}
if uri == "" {
return "", errors.New("#EXT-X-MAP 缺少 URI 属性")
}
// 处理相对 URL
if !strings.HasPrefix(uri, "http://") && !strings.HasPrefix(uri, "https://") {
parsed, err := baseURL.Parse(uri)
if err != nil {
return "", fmt.Errorf("解析 init 段 URL 失败: %w", err)
}
uri = parsed.String()
}
return uri, nil
}
return "", nil
}
// probeFMP4Init 下载 fMP4 init 段并解析编解码器信息
func probeFMP4Init(ctx context.Context, initURL string, headers map[string]string, logger *livelogger.LiveLogger) (*StreamHeaderInfo, error) {
if logger != nil {
logger.Infof("HLS 探测 (fMP4): 下载 init 段: %s", initURL)
}
// init 段通常很小(几百字节到几 KB),完整下载即可
initData, err := downloadSegmentHeader(ctx, initURL, headers, 128*1024)
if err != nil {
return nil, fmt.Errorf("下载 fMP4 init 段失败: %w", err)
}
if logger != nil {
logger.Infof("HLS 探测 (fMP4): 已下载 init 段 %d 字节", len(initData))
}
info, err := parseFMP4InitSegment(initData)
if err != nil {
return nil, fmt.Errorf("解析 fMP4 init 段失败: %w", err)
}
return info, nil
}
// parseFMP4InitSegment 解析 fMP4 init 段(包含 moov box)
// 路径:moov → trak → mdia → minf → stbl → stsd → avc1/hev1/hvc1/mp4a
func parseFMP4InitSegment(data []byte) (*StreamHeaderInfo, error) {
if len(data) < 8 {
return nil, errors.New("init 段数据太短")
}
info := &StreamHeaderInfo{}
// 找到 moov box
moovData := findBox(data, "moov")
if moovData == nil {
return nil, errors.New("未找到 moov box")
}
// 遍历所有 trak box
offset := 0
for offset < len(moovData) {
boxSize, boxType, headerLen := readBoxHeader(moovData[offset:])
if boxSize == 0 || headerLen == 0 {
break
}
if int(boxSize) > len(moovData)-offset {
break
}
if boxType == "trak" {
parseTrakBox(moovData[offset+headerLen:offset+int(boxSize)], info)
}
offset += int(boxSize)
}
if info.VideoCodec == "" && info.AudioCodec == "" {
return nil, errors.New("init 段中未找到任何编解码器信息")
}
return info, nil
}
// parseTrakBox 解析 trak → mdia → minf → stbl → stsd 获取编解码器信息
func parseTrakBox(data []byte, info *StreamHeaderInfo) {
mdiaData := findBox(data, "mdia")
if mdiaData == nil {
return
}
minfData := findBox(mdiaData, "minf")
if minfData == nil {
return
}
stblData := findBox(minfData, "stbl")
if stblData == nil {
return
}
stsdData := findBox(stblData, "stsd")
if stsdData == nil {
return
}
// stsd box 内容:4 字节版本/标志 + 4 字节 entry_count + entries...
if len(stsdData) < 8 {
return
}
// 跳过版本/标志和 entry_count
entryData := stsdData[8:]
if len(entryData) < 8 {
return
}
// 读取第一个 entry
entrySize, entryType, _ := readBoxHeader(entryData)
if entrySize == 0 {
return
}
switch entryType {
case "avc1", "avc3":
info.VideoCodec = "h264"
// avc1 box: 78 字节固定头部后面跟子 box(包括 avcC)
if int(entrySize) <= len(entryData) {
parseAvcCBox(entryData[86:int(entrySize)], info)
}
case "hev1", "hvc1":
info.VideoCodec = "h265"
// hev1/hvc1 box: 78 字节固定头部后面跟子 box(包括 hvcC)
if int(entrySize) <= len(entryData) {
parseHvcCBox(entryData[86:int(entrySize)], info)
}
case "mp4a":
info.AudioCodec = "aac"
case "Opus", "opus":
info.AudioCodec = "opus"
case "ac-3":
info.AudioCodec = "ac3"
case "ec-3":
info.AudioCodec = "eac3"
}
}
// parseAvcCBox 在数据中找到 avcC box 并解析 SPS
func parseAvcCBox(data []byte, info *StreamHeaderInfo) {
avcCData := findBox(data, "avcC")
if len(avcCData) < 8 {
return
}
// avcC 格式:
// 1 byte configurationVersion
// 1 byte AVCProfileIndication
// 1 byte profile_compatibility
// 1 byte AVCLevelIndication
// 1 byte lengthSizeMinusOne (低 2 位)
// 1 byte numOfSPS (低 5 位)
// 2 bytes spsLength
// N bytes spsNALU
if len(avcCData) < 7 {
return
}
numSPS := int(avcCData[5] & 0x1F)
if numSPS == 0 {
return
}
offset := 6
if offset+2 > len(avcCData) {
return
}
spsLen := int(binary.BigEndian.Uint16(avcCData[offset : offset+2]))
offset += 2
if offset+spsLen > len(avcCData) || spsLen == 0 {
return
}
spsData := avcCData[offset : offset+spsLen]
var sps h264.SPS
if err := sps.Unmarshal(spsData); err == nil {
info.Width = sps.Width()
info.Height = sps.Height()
if fps := sps.FPS(); fps > 0 && fps < 300 {
info.FrameRate = fps
}
info.ParsedFromSPS = true
}
}
// parseHvcCBox 在数据中找到 hvcC box 并解析 SPS
func parseHvcCBox(data []byte, info *StreamHeaderInfo) {
hvcCData := findBox(data, "hvcC")
if len(hvcCData) < 23 {
return
}
// hvcC 格式比较复杂,需要遍历 array of arrays
// offset 22: numOfArrays
numArrays := int(hvcCData[22])
offset := 23
for i := 0; i < numArrays && offset < len(hvcCData); i++ {
if offset+3 > len(hvcCData) {
break
}
nalType := hvcCData[offset] & 0x3F
numNALUs := int(binary.BigEndian.Uint16(hvcCData[offset+1 : offset+3]))
offset += 3
for j := 0; j < numNALUs && offset+2 <= len(hvcCData); j++ {
nalLen := int(binary.BigEndian.Uint16(hvcCData[offset : offset+2]))
offset += 2
if offset+nalLen > len(hvcCData) {
break
}
// SPS NAL type = 33
if nalType == hevcNalTypeSPS {
var sps h265.SPS
if err := sps.Unmarshal(hvcCData[offset : offset+nalLen]); err == nil {
info.Width = sps.Width()
info.Height = sps.Height()
if fps := sps.FPS(); fps > 0 && fps < 300 {
info.FrameRate = fps
}
info.ParsedFromSPS = true
}
return
}
offset += nalLen
}
}
}
// findBox 在 MP4 数据中查找指定类型的 box,返回 box 内容(不含 header)
func findBox(data []byte, boxType string) []byte {
offset := 0
for offset+8 <= len(data) {
size, typ, headerLen := readBoxHeader(data[offset:])
if size == 0 || headerLen == 0 {
break
}
if int(size) > len(data)-offset {
break
}
if typ == boxType {
return data[offset+headerLen : offset+int(size)]
}
offset += int(size)
}
return nil
}
// readBoxHeader 读取 MP4 box 头部,返回 (size, type, headerLength)
func readBoxHeader(data []byte) (uint64, string, int) {
if len(data) < 8 {
return 0, "", 0
}
size := uint64(binary.BigEndian.Uint32(data[0:4]))
boxType := string(data[4:8])
if size == 1 {
// 64 位扩展 size
if len(data) < 16 {
return 0, "", 0
}
size = binary.BigEndian.Uint64(data[8:16])
return size, boxType, 16
}
if size == 0 {
// box 延伸到文件末尾
size = uint64(len(data))
}
return size, boxType, 8
}
// parseFirstSegmentURL 从 m3u8 内容中解析第一个媒体分段的 URL(TS 或 m4s)
func parseFirstSegmentURL(content string, baseURL *url.URL) (string, error) {
lines := strings.Split(content, "\n")
for _, line := range lines {
line = strings.TrimSpace(line)
if line == "" || strings.HasPrefix(line, "#") {
continue
}
// 第一个非注释、非空行就是分段 URL
segmentURL := line
// 处理相对 URL
if !strings.HasPrefix(segmentURL, "http://") && !strings.HasPrefix(segmentURL, "https://") {
parsed, err := baseURL.Parse(segmentURL)
if err != nil {
return "", fmt.Errorf("解析相对 URL 失败: %w", err)
}
segmentURL = parsed.String()
}
return segmentURL, nil
}
return "", errors.New("m3u8 中未找到 TS 分段")
}
// downloadSegmentHeader 下载 TS 分段的头部数据
func downloadSegmentHeader(ctx context.Context, segmentURL string, headers map[string]string, maxBytes int) ([]byte, error) {
transport := &http.Transport{
DialContext: (&net.Dialer{
Timeout: 10 * time.Second,
KeepAlive: 30 * time.Second,
}).DialContext,
}
proxy.ApplyDownloadProxyToTransport(transport)
client := &http.Client{
Transport: transport,
Timeout: 15 * time.Second,
}
ctxTimeout, cancel := context.WithTimeout(ctx, 15*time.Second)
defer cancel()
req, err := http.NewRequestWithContext(ctxTimeout, http.MethodGet, segmentURL, nil)
if err != nil {
return nil, err
}
// 使用 Range 请求只下载头部
req.Header.Set("Range", fmt.Sprintf("bytes=0-%d", maxBytes-1))
for k, v := range headers {
req.Header.Set(k, v)
}
if req.Header.Get("User-Agent") == "" {
req.Header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")
}
resp, err := client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
// 200 或 206 都可以
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
return nil, fmt.Errorf("HTTP %d", resp.StatusCode)
}
// 读取限定字节数
data, err := io.ReadAll(io.LimitReader(resp.Body, int64(maxBytes)))
if err != nil {
return nil, err
}
return data, nil
}
// parseTSData 解析 TS 数据,提取视频/音频信息
func parseTSData(data []byte) (*StreamHeaderInfo, error) {
if len(data) < tsPacketSize {
return nil, errors.New("TS 数据太短")
}
info := &StreamHeaderInfo{}
// 1. 找到 sync byte 对齐
offset := findTSSyncOffset(data)
if offset < 0 {
return nil, errors.New("未找到 TS sync byte")
}
// 2. 解析 PAT 找到 PMT PID
pmtPID := -1
// 3. 解析 PMT 找到视频/音频 PID 和类型
videoPID := -1
videoType := 0
audioPID := -1
audioType := 0
// 4. 收集视频 PES 中的 NAL units
var videoPESData []byte
// 第一轮:找 PAT 和 PMT
for pos := offset; pos+tsPacketSize <= len(data); pos += tsPacketSize {
pkt := data[pos : pos+tsPacketSize]
if pkt[0] != tsSyncByte {
break
}
pid := (int(pkt[1]&0x1F) << 8) | int(pkt[2])
payloadStart := pkt[1]&0x40 != 0
hasPayload := pkt[3]&0x10 != 0
hasAdaptation := pkt[3]&0x20 != 0
if !hasPayload {
continue
}
// 计算 payload 起始偏移
payloadOffset := 4
if hasAdaptation {
adaptLen := int(pkt[4])
payloadOffset = 5 + adaptLen
if payloadOffset >= tsPacketSize {
continue
}
}
payload := pkt[payloadOffset:]
// PAT (PID 0)
if pid == 0 && payloadStart && pmtPID < 0 {
if len(payload) > 0 {
// 跳过 pointer field
pointerField := int(payload[0])
tableStart := 1 + pointerField
if tableStart+8 < len(payload) {
table := payload[tableStart:]
// PAT table_id should be 0x00
if table[0] == 0x00 {
sectionLength := (int(table[1]&0x0F) << 8) | int(table[2])
if sectionLength > 5 {
// 跳到 program entries (offset 8)
progOffset := 8
if tableStart+progOffset+4 <= len(payload) {
entries := table[progOffset:]
if len(entries) >= 4 {
pmtPID = (int(entries[2]&0x1F) << 8) | int(entries[3])
}
}
}
}
}
}
}
// PMT
if pid == pmtPID && payloadStart && videoPID < 0 {
if len(payload) > 0 {
pointerField := int(payload[0])
tableStart := 1 + pointerField
if tableStart+12 < len(payload) {
table := payload[tableStart:]
if table[0] == 0x02 { // PMT table_id
sectionLength := (int(table[1]&0x0F) << 8) | int(table[2])
progInfoLen := (int(table[10]&0x0F) << 8) | int(table[11])
streamOffset := 12 + progInfoLen
// 最大长度不超过 section
maxOffset := 3 + sectionLength - 4 // 减去 CRC
if maxOffset > len(table) {
maxOffset = len(table)
}
for streamOffset+5 <= maxOffset {
sType := int(table[streamOffset])
sPID := (int(table[streamOffset+1]&0x1F) << 8) | int(table[streamOffset+2])
esInfoLen := (int(table[streamOffset+3]&0x0F) << 8) | int(table[streamOffset+4])
switch sType {
case streamTypeH264:
videoPID = sPID
videoType = streamTypeH264
info.VideoCodec = "H.264"
case streamTypeH265:
videoPID = sPID
videoType = streamTypeH265
info.VideoCodec = "H.265"
case streamTypeAAC:
audioPID = sPID
audioType = streamTypeAAC
info.AudioCodec = "AAC"
case streamTypeAC3:
audioPID = sPID
audioType = streamTypeAC3
info.AudioCodec = "AC-3"
case streamTypeMPEG:
if audioPID < 0 {
audioPID = sPID
audioType = streamTypeMPEG
info.AudioCodec = "MP3"
}
}
streamOffset += 5 + esInfoLen
}
}
}
}
}
}
// 如果没找到视频 PID,尝试标记不支持
if videoPID < 0 {
if pmtPID < 0 {
return nil, errors.New("未找到 PAT/PMT,TS 数据可能不完整")
}
info.Unsupported = true
info.UnsupportedMsg = "TS 中未发现支持的视频编码"
return info, nil
}
// 第二轮:提取视频 PES 数据
for pos := offset; pos+tsPacketSize <= len(data); pos += tsPacketSize {
pkt := data[pos : pos+tsPacketSize]
if pkt[0] != tsSyncByte {
break
}
pid := (int(pkt[1]&0x1F) << 8) | int(pkt[2])
hasPayload := pkt[3]&0x10 != 0
hasAdaptation := pkt[3]&0x20 != 0
if pid != videoPID || !hasPayload {
continue
}
payloadOffset := 4
if hasAdaptation {
adaptLen := int(pkt[4])
payloadOffset = 5 + adaptLen
if payloadOffset >= tsPacketSize {
continue
}
}
payload := pkt[payloadOffset:]
// 检查 PES header
payloadStart := pkt[1]&0x40 != 0
if payloadStart && len(payload) >= 9 {
// PES start code: 00 00 01
if payload[0] == 0 && payload[1] == 0 && payload[2] == 1 {
// 跳过 PES header
pesHeaderLen := int(payload[8])
pesPayloadStart := 9 + pesHeaderLen
if pesPayloadStart < len(payload) {
videoPESData = append(videoPESData, payload[pesPayloadStart:]...)
}
continue
}
}
// 非 PES 起始的 payload,直接追加
videoPESData = append(videoPESData, payload...)
}
// 5. 从 PES 数据中提取 NAL units 并解析 SPS
if len(videoPESData) > 0 {
switch videoType {
case streamTypeH264:
parseH264NALUnits(videoPESData, info)
case streamTypeH265:
parseH265NALUnits(videoPESData, info)
}
}
// 如果有 video codec 但没解析到分辨率,标记可能需要更多数据
if info.Width == 0 && info.Height == 0 && !info.Unsupported {
if info.VideoCodec != "" {
info.Unsupported = true
info.UnsupportedMsg = "在有限的 TS 数据中未找到 SPS,无法获取分辨率"
}
}
_ = audioPID
_ = audioType
return info, nil
}
// findTSSyncOffset 在数据中查找 TS sync byte 对齐位置
func findTSSyncOffset(data []byte) int {
for i := 0; i < len(data)-tsPacketSize*3; i++ {
if data[i] == tsSyncByte &&
data[i+tsPacketSize] == tsSyncByte &&
data[i+tsPacketSize*2] == tsSyncByte {
return i
}
}
return -1
}
// parseH264NALUnits 从 Annex B 字节流中提取 H.264 NAL units
func parseH264NALUnits(data []byte, info *StreamHeaderInfo) {
nals := extractNALUnits(data)
for _, nal := range nals {
if len(nal) == 0 {
continue
}
nalType := nal[0] & 0x1F
if nalType == nalTypeSPS {
// 使用 mediacommon 解析 SPS
var sps h264.SPS
if err := sps.Unmarshal(nal); err == nil {
info.Width = sps.Width()
info.Height = sps.Height()
if fps := sps.FPS(); fps > 0 && fps < 300 {
info.FrameRate = fps
}
}
return
}
}
}
// parseH265NALUnits 从 Annex B 字节流中提取 H.265 NAL units
func parseH265NALUnits(data []byte, info *StreamHeaderInfo) {
nals := extractNALUnits(data)
for _, nal := range nals {
if len(nal) < 2 {
continue
}
nalType := (nal[0] >> 1) & 0x3F
if nalType == hevcNalTypeSPS {
var sps h265.SPS
if err := sps.Unmarshal(nal); err == nil {
info.Width = sps.Width()
info.Height = sps.Height()
if fps := sps.FPS(); fps > 0 && fps < 300 {
info.FrameRate = fps
}
}
return
}
}
}
// extractNALUnits 从 Annex B 字节流提取 NAL units
// Annex B 格式使用 00 00 01 或 00 00 00 01 作为起始码
func extractNALUnits(data []byte) [][]byte {
var nals [][]byte
startCode3 := []byte{0, 0, 1}
startCode4 := []byte{0, 0, 0, 1}
// 找到所有起始码位置
var positions []int
for i := 0; i < len(data)-3; i++ {
if bytes.Equal(data[i:i+3], startCode3) {
// 检查是否是 4 字节起始码
if i > 0 && data[i-1] == 0 {
continue // 已经被 4 字节起始码处理
}
positions = append(positions, i+3)
} else if i < len(data)-4 && bytes.Equal(data[i:i+4], startCode4) {
positions = append(positions, i+4)
}
}
// 提取每个 NAL unit
for i, start := range positions {
var end int
if i+1 < len(positions) {
// 下一个起始码前面可能有 00 00 00 01 或 00 00 01
end = positions[i+1]
// 回退到起始码之前
for end > start && data[end-1] == 0 {
end--
}
// 处理可能的 3 字节起始码
if end >= 3 && bytes.Equal(data[end-3:end], startCode3) {
end -= 3
} else if end >= 4 && bytes.Equal(data[end-4:end], startCode4) {
end -= 4
}
} else {
end = len(data)
}
if end > start {
nals = append(nals, data[start:end])
}
}
return nals
}
// ProbeHLS 独立的 HLS 流探测函数(不经过 StreamProbe 代理)
// 流程:下载 m3u8 → 解析第一个 TS 分段 URL → 下载 TS 头部 → 解析 SPS
func ProbeHLS(ctx context.Context, m3u8URL *url.URL, headers map[string]string, logger *livelogger.LiveLogger) (*StreamHeaderInfo, error) {
if logger != nil {
logger.Debugf("HLS 探测开始: URL=%s, Headers=%v", m3u8URL.String(), headers)
}
// 1. 下载 m3u8 内容
transport := &http.Transport{
DialContext: (&net.Dialer{
Timeout: 10 * time.Second,
KeepAlive: 30 * time.Second,
}).DialContext,
}
proxy.ApplyDownloadProxyToTransport(transport)
client := &http.Client{
Transport: transport,
Timeout: 15 * time.Second,
}
ctxTimeout, cancel := context.WithTimeout(ctx, 15*time.Second)
defer cancel()
req, err := http.NewRequestWithContext(ctxTimeout, http.MethodGet, m3u8URL.String(), nil)
if err != nil {
return nil, err
}
for k, v := range headers {
req.Header.Set(k, v)
}
if req.Header.Get("User-Agent") == "" {
req.Header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")
}
resp, err := client.Do(req)
if err != nil {
return nil, fmt.Errorf("下载 m3u8 失败: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("m3u8 返回 HTTP %d", resp.StatusCode)
}
// 2. 解析并探测
return probeHLSStreamInfo(ctxTimeout, resp.Body, m3u8URL, headers, logger)
}
+52
View File
@@ -0,0 +1,52 @@
// Package streamprobe 提供统一的直播流探测代理功能
// 在转发直播流数据给下载器的同时,解析流头信息(分辨率、编码、帧率、码率等)
// 替代旧的 flvproxy 包,同时支持 SPS/PPS 变化分段检测
package streamprobe
import "fmt"
// StreamHeaderInfo 包含从直播流头部解析出的实际信息
type StreamHeaderInfo struct {
// 视频信息
VideoCodec string `json:"video_codec"` // "h264", "h265", "av1", "unknown"
Width int `json:"width"` // 从 SPS 解析的实际宽度(像素)
Height int `json:"height"` // 从 SPS 解析的实际高度(像素)
FrameRate float64 `json:"frame_rate"` // 帧率(来自 SPS 或 onMetaData)
VideoBitrate int `json:"video_bitrate"` // 视频码率 (kbps, 来自 onMetaData)
// 音频信息
AudioCodec string `json:"audio_codec"` // "aac", "mp3", "opus", "unknown"
AudioBitrate int `json:"audio_bitrate"` // 音频码率 (kbps)
// 解析来源和状态
ParsedFromSPS bool `json:"parsed_from_sps"` // 分辨率是否从 SPS 解析(最可靠)
ParsedFromMeta bool `json:"parsed_from_meta"` // 分辨率是否从 onMetaData 解析
// 不支持的编码格式信息
Unsupported bool `json:"unsupported"` // 编码格式是否不支持深度解析
UnsupportedMsg string `json:"unsupported_msg"` // 不支持时的说明信息
// 原始数据(供 debug 使用)
RawMetaData map[string]interface{} `json:"raw_meta_data,omitempty"`
}
// Resolution 返回格式化的分辨率字符串,如 "1920x1080"
// 如果宽高都为 0,返回空字符串
func (info *StreamHeaderInfo) Resolution() string {
if info.Width > 0 && info.Height > 0 {
return fmt.Sprintf("%dx%d", info.Width, info.Height)
}
return ""
}
// ProbeStatus 返回探测状态字符串
// 用于前端展示:"success" | "unsupported" | "pending"
func (info *StreamHeaderInfo) ProbeStatus() string {
if info.Unsupported {
return "unsupported"
}
if info.ParsedFromSPS || info.ParsedFromMeta {
return "success"
}
return "pending"
}
+407
View File
@@ -0,0 +1,407 @@
package streamprobe
import (
"context"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/url"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/bililive-go/bililive-go/src/pkg/livelogger"
"github.com/bililive-go/bililive-go/src/pkg/proxy"
)
const (
// maxProbeTags 探测阶段最多读取的 tag 数量
maxProbeTags = 30
)
// Config StreamProbe 配置
type Config struct {
// UpstreamURL 上游直播流 URL
UpstreamURL *url.URL
// Headers 下载用 HTTP headers(如 Cookie、Referer 等)
Headers map[string]string
// OnProbed 探测完成回调
// 在成功解析到流信息后调用
OnProbed func(info *StreamHeaderInfo)
// OnProbeError 探测错误回调(不影响流转发)
// 仅用于日志记录,不应阻止录制
OnProbeError func(err error, msg string)
// Logger 日志记录器
Logger *livelogger.LiveLogger
}
// StreamProbe 直播流探测代理
// 连接上游直播流,解析头部信息,然后在本地提供代理服务
type StreamProbe struct {
config Config
// 本地代理服务器
listener net.Listener
localURL *url.URL
server *http.Server
serverErr chan error
// 上游连接
upstreamResp *http.Response
upstreamBody io.ReadCloser
// 缓冲数据:探测阶段已读取但需要转发给下载器的数据
probedData []byte
probedDataMu sync.Mutex
// 探测结果
headerInfo atomic.Pointer[StreamHeaderInfo]
// 生命周期
ctx context.Context
cancel context.CancelFunc
wg sync.WaitGroup
// 连接状态
connected atomic.Bool
clientReady chan struct{} // 客户端(下载器)已连接
}
// New 创建一个新的 StreamProbe
func New(cfg Config) *StreamProbe {
return &StreamProbe{
config: cfg,
serverErr: make(chan error, 1),
clientReady: make(chan struct{}),
}
}
// Start 启动探测代理
// 1. 连接上游流并解析头部信息
// 2. 在本地随机端口启动 HTTP 代理
// 3. 通过回调返回解析到的流信息
func (p *StreamProbe) Start(ctx context.Context) error {
p.ctx, p.cancel = context.WithCancel(ctx)
// 1. 连接上游流
if err := p.connectUpstream(); err != nil {
p.cancel()
return fmt.Errorf("连接上游流失败: %w", err)
}
// 2. 探测流头信息
p.probeStreamHeader()
// 3. 启动本地代理
if err := p.startLocalServer(); err != nil {
p.cleanup()
return fmt.Errorf("启动本地代理失败: %w", err)
}
return nil
}
// Stop 停止探测代理
func (p *StreamProbe) Stop() {
if p.cancel != nil {
p.cancel()
}
p.cleanup()
}
// LocalURL 返回本地代理的 URL
// 下载器应使用此 URL 作为输入
func (p *StreamProbe) LocalURL() *url.URL {
return p.localURL
}
// GetHeaderInfo 返回已解析的流头信息
func (p *StreamProbe) GetHeaderInfo() *StreamHeaderInfo {
return p.headerInfo.Load()
}
// connectUpstream 连接上游直播流
func (p *StreamProbe) connectUpstream() error {
// 创建带下载代理的 HTTP 客户端
transport := &http.Transport{
DialContext: (&net.Dialer{
Timeout: 10 * time.Second,
KeepAlive: 30 * time.Second,
}).DialContext,
}
proxy.ApplyDownloadProxyToTransport(transport)
client := &http.Client{
Transport: transport,
Timeout: 0, // 不设超时,直播流是持久连接
}
req, err := http.NewRequestWithContext(p.ctx, http.MethodGet, p.config.UpstreamURL.String(), nil)
if err != nil {
return err
}
// 设置下载 headers
for k, v := range p.config.Headers {
req.Header.Set(k, v)
}
if req.Header.Get("User-Agent") == "" {
req.Header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")
}
resp, err := client.Do(req)
if err != nil {
return err
}
if resp.StatusCode != http.StatusOK {
resp.Body.Close()
return fmt.Errorf("上游返回 HTTP %d", resp.StatusCode)
}
p.upstreamResp = resp
p.upstreamBody = resp.Body
return nil
}
// probeStreamHeader 探测 FLV 流头信息
func (p *StreamProbe) probeStreamHeader() {
// 判断是否为 FLV 流
if p.isFLVStream() {
info, bufferedData, err := parseFLVStreamInfo(p.upstreamBody, maxProbeTags)
if err != nil {
p.notifyError(err, "FLV 流头解析失败")
// 如果是 ErrNotFLV,说明不是 FLV 格式,尝试作为原始数据通过
if errors.Is(err, ErrNotFLV) {
// 保留已读取的数据
p.probedData = bufferedData
}
return
}
p.probedData = bufferedData
p.headerInfo.Store(info)
p.notifyProbed(info)
} else if p.isHLSStream() {
// HLS 流:探测第一个分段的信息
info := p.probeHLSStream()
if info != nil {
p.headerInfo.Store(info)
p.notifyProbed(info)
}
// HLS 不需要代理转发,但为了架构统一仍然提供代理
// 代理直接转发上游响应
} else {
p.notifyError(nil, "不支持的流格式,跳过头部解析")
}
}
// probeHLSStream 探测 HLS 流的信息(通过解析第一个 TS 分段头部)
// 目前为简化实现,只设置基本信息
func (p *StreamProbe) probeHLSStream() *StreamHeaderInfo {
// TODO: 实现 HLS 流探测
// 1. 下载 m3u8
// 2. 获取第一个 TS 分段 URL
// 3. 下载 TS 分段头部(前几个 TS packet)
// 4. 解析 PAT/PMT -> PES -> SPS
info := &StreamHeaderInfo{
Unsupported: true,
UnsupportedMsg: "HLS 流探测功能开发中",
}
return info
}
// isFLVStream 判断上游是否为 FLV 流
func (p *StreamProbe) isFLVStream() bool {
urlPath := strings.ToLower(p.config.UpstreamURL.Path)
if strings.Contains(urlPath, ".flv") {
return true
}
// 检查 Content-Type
if p.upstreamResp != nil {
ct := strings.ToLower(p.upstreamResp.Header.Get("Content-Type"))
if strings.Contains(ct, "flv") || strings.Contains(ct, "x-flv") {
return true
}
}
return false
}
// isHLSStream 判断上游是否为 HLS 流
func (p *StreamProbe) isHLSStream() bool {
urlPath := strings.ToLower(p.config.UpstreamURL.Path)
if strings.Contains(urlPath, ".m3u8") || strings.HasSuffix(urlPath, ".m3u") {
return true
}
if p.upstreamResp != nil {
ct := strings.ToLower(p.upstreamResp.Header.Get("Content-Type"))
if strings.Contains(ct, "mpegurl") || strings.Contains(ct, "x-mpegurl") {
return true
}
}
return false
}
// startLocalServer 在本地随机端口启动 HTTP 代理
func (p *StreamProbe) startLocalServer() error {
var err error
p.listener, err = net.Listen("tcp", "127.0.0.1:0")
if err != nil {
return err
}
addr := p.listener.Addr().(*net.TCPAddr)
p.localURL, _ = url.Parse(fmt.Sprintf("http://127.0.0.1:%d/stream", addr.Port))
mux := http.NewServeMux()
mux.HandleFunc("/stream", p.handleStreamRequest)
p.server = &http.Server{
Handler: mux,
ReadTimeout: 5 * time.Second,
}
p.wg.Add(1)
go func() {
defer p.wg.Done()
if err := p.server.Serve(p.listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
p.serverErr <- err
}
}()
return nil
}
// handleStreamRequest 处理下载器的 HTTP 请求
// 将探测阶段缓冲的数据 + 后续上游数据一起返回
func (p *StreamProbe) handleStreamRequest(w http.ResponseWriter, r *http.Request) {
p.connected.Store(true)
// 通知客户端已连接
select {
case <-p.clientReady:
default:
close(p.clientReady)
}
// 设置响应头
if p.upstreamResp != nil {
for k, vs := range p.upstreamResp.Header {
for _, v := range vs {
w.Header().Add(k, v)
}
}
}
w.Header().Set("Connection", "close")
// 设置 Flusher 以实时推送数据
flusher, hasFlusher := w.(http.Flusher)
// 1. 先发送探测阶段缓冲的数据
p.probedDataMu.Lock()
buffered := p.probedData
p.probedDataMu.Unlock()
if len(buffered) > 0 {
if _, err := w.Write(buffered); err != nil {
return
}
if hasFlusher {
flusher.Flush()
}
}
// 2. 转发上游后续数据
if p.upstreamBody == nil {
return
}
buf := make([]byte, 32*1024) // 32KB 缓冲区
for {
select {
case <-p.ctx.Done():
return
case <-r.Context().Done():
return
default:
}
n, err := p.upstreamBody.Read(buf)
if n > 0 {
if _, writeErr := w.Write(buf[:n]); writeErr != nil {
return
}
if hasFlusher {
flusher.Flush()
}
}
if err != nil {
if !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) {
p.notifyError(err, "上游流读取中断")
}
return
}
}
}
// cleanup 清理资源
func (p *StreamProbe) cleanup() {
if p.server != nil {
shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
p.server.Shutdown(shutdownCtx)
}
if p.upstreamBody != nil {
p.upstreamBody.Close()
}
p.wg.Wait()
}
// notifyProbed 通知探测完成
func (p *StreamProbe) notifyProbed(info *StreamHeaderInfo) {
if p.config.OnProbed != nil {
p.config.OnProbed(info)
}
}
// notifyError 通知探测错误
func (p *StreamProbe) notifyError(err error, msg string) {
if p.config.OnProbeError != nil {
p.config.OnProbeError(err, msg)
}
}
// IsStreamFLV 判断给定的 URL 是否看起来像 FLV 流
// 这是一个工具函数,供外部判断是否应该使用 StreamProbe
func IsStreamFLV(u *url.URL) bool {
return strings.Contains(strings.ToLower(u.Path), ".flv")
}
// IsStreamHLS 判断给定的 URL 是否看起来像 HLS 流
func IsStreamHLS(u *url.URL) bool {
path := strings.ToLower(u.Path)
return strings.Contains(path, ".m3u8") || strings.HasSuffix(path, ".m3u")
}
// WaitForClient 等待下载器连接到代理
// 返回 true 表示客户端已连接,false 表示超时或 context 取消
func (p *StreamProbe) WaitForClient(timeout time.Duration) bool {
timer := time.NewTimer(timeout)
defer timer.Stop()
select {
case <-p.clientReady:
return true
case <-timer.C:
return false
case <-p.ctx.Done():
return false
}
}
+4 -4
View File
@@ -253,8 +253,8 @@ func CreateDefaultClient() *http.Client {
transport.DialContext = dialer.DialContext
transport.DialTLSContext = createTLSDialer(dialer, false, "")
// 应用代理设置
proxy.ApplyProxyToTransport(transport)
// 应用信息获取代理(这些客户端主要用于获取直播间信息等 API 请求)
proxy.ApplyInfoProxyToTransport(transport)
return &http.Client{Transport: transport}
}
@@ -283,8 +283,8 @@ func CreateConnCounterClient() (*http.Client, error) {
// Use "tls:" prefix to distinguish from plain connections
transport.DialTLSContext = createTLSDialer(dialer, true, "tls:")
// 应用代理设置
proxy.ApplyProxyToTransport(transport)
// 应用信息获取代理(这些客户端主要用于获取直播间信息等 API 请求)
proxy.ApplyInfoProxyToTransport(transport)
return &http.Client{Transport: transport}, nil
}
+183 -10
View File
@@ -31,6 +31,7 @@ import (
"github.com/bililive-go/bililive-go/src/pkg/parser/ffmpeg"
"github.com/bililive-go/bililive-go/src/pkg/parser/native/flv"
bilisentry "github.com/bililive-go/bililive-go/src/pkg/sentry"
"github.com/bililive-go/bililive-go/src/pkg/streamprobe"
"github.com/bililive-go/bililive-go/src/pkg/utils"
)
@@ -200,8 +201,15 @@ type recorder struct {
currentFileLock sync.RWMutex
currentFilePath string
// 当前录制的流信息
// 当前录制的流信息(来自平台 API)
currentStreamInfo *live.AvailableStreamInfo
// 当前录制使用的原始流 URL 和 Headers(供调试和前端展示)
currentStreamURL string
currentStreamHeaders map[string]string
// 实际流头部信息(来自 StreamProbe 探测)
actualStreamInfo atomic.Pointer[streamprobe.StreamHeaderInfo]
}
func NewRecorder(ctx context.Context, live live.Live) (Recorder, error) {
@@ -244,7 +252,12 @@ func (r *recorder) tryRecord(ctx context.Context) {
}
if err != nil || len(streamInfos) == 0 {
r.getLogger().WithError(err).Warn("failed to get stream url, will retry after 5s...")
time.Sleep(5 * time.Second)
// 使用可中断的等待,确保 Ctrl+C 能立即响应
select {
case <-ctx.Done():
case <-r.stop:
case <-time.After(5 * time.Second):
}
return
}
@@ -276,6 +289,12 @@ func (r *recorder) tryRecord(ctx context.Context) {
url := streamInfo.Url
// 保存原始流 URL 和 Headers(供前端调试展示)
r.currentFileLock.Lock()
r.currentStreamURL = url.String()
r.currentStreamHeaders = streamInfo.HeadersForDownloader
r.currentFileLock.Unlock()
if strings.Contains(url.Path, "m3u8") {
fileName = fileName[:len(fileName)-4] + ".ts"
}
@@ -300,6 +319,90 @@ func (r *recorder) tryRecord(ctx context.Context) {
parserCfg["use_flv_proxy"] = "true"
}
// StreamProbe 探测:仅对 FLV 流使用代理探测
// HLS 是分段 HTTP 请求协议,无法通过单一 HTTP 代理转发
isFLV := streamprobe.IsStreamFLV(url)
if isFLV {
// FLV 流:启动探测代理
probeConfig := streamprobe.Config{
UpstreamURL: url,
Headers: streamInfo.HeadersForDownloader,
OnProbed: func(info *streamprobe.StreamHeaderInfo) {
r.actualStreamInfo.Store(info)
r.getLogger().Infof("流探测完成: 编码=%s, 分辨率=%s, 帧率=%.1f, 状态=%s",
info.VideoCodec, info.Resolution(), info.FrameRate, info.ProbeStatus())
},
OnProbeError: func(err error, msg string) {
if err != nil {
r.getLogger().Warnf("流探测警告: %s: %v", msg, err)
} else {
r.getLogger().Warnf("流探测警告: %s", msg)
}
// 探测出错时也设置状态,避免永远显示"探测中"
r.actualStreamInfo.CompareAndSwap(nil, &streamprobe.StreamHeaderInfo{
Unsupported: true,
UnsupportedMsg: fmt.Sprintf("流探测失败: %s", msg),
})
},
Logger: r.getLogger(),
}
probe := streamprobe.New(probeConfig)
if probeErr := probe.Start(ctx); probeErr != nil {
// 探测代理启动失败不应影响录制,回退到直连上游
r.getLogger().WithError(probeErr).Warn("流探测代理启动失败,将直接连接上游")
r.actualStreamInfo.Store(&streamprobe.StreamHeaderInfo{
Unsupported: true,
UnsupportedMsg: fmt.Sprintf("流探测代理启动失败: %v", probeErr),
})
} else {
// 代理启动成功,用代理 URL 替换原始 URL
defer probe.Stop()
streamInfo = &live.StreamUrlInfo{
Url: probe.LocalURL(),
HeadersForDownloader: nil, // 本地代理不需要 headers
Format: streamInfo.Format,
Quality: streamInfo.Quality,
Description: streamInfo.Description,
Codec: streamInfo.Codec,
Width: streamInfo.Width,
Height: streamInfo.Height,
Bitrate: streamInfo.Bitrate,
Vbitrate: streamInfo.Vbitrate,
FrameRate: streamInfo.FrameRate,
}
url = probe.LocalURL()
}
} else if streamprobe.IsStreamHLS(url) {
// HLS 流:不使用代理,异步探测第一个 TS 分段的头部信息
// 使用 tryRecord 的 ctx,当录制结束/重试时自动取消探测
go func(probeCtx context.Context) {
hlsInfo, probeErr := streamprobe.ProbeHLS(probeCtx, url, streamInfo.HeadersForDownloader, r.getLogger())
if probeErr != nil {
// context 取消不算真正的错误,不需要打印
if probeCtx.Err() != nil {
return
}
r.getLogger().Warnf("HLS 流探测失败: %v", probeErr)
// 探测失败也设置一个状态,避免永远 pending
r.actualStreamInfo.Store(&streamprobe.StreamHeaderInfo{
Unsupported: true,
UnsupportedMsg: fmt.Sprintf("HLS 探测失败: %v", probeErr),
})
return
}
r.actualStreamInfo.Store(hlsInfo)
r.getLogger().Infof("HLS 流探测完成: 编码=%s, 分辨率=%s, 帧率=%.1f",
hlsInfo.VideoCodec, hlsInfo.Resolution(), hlsInfo.FrameRate)
}(ctx)
} else {
// 其他格式:标记为不支持
r.actualStreamInfo.Store(&streamprobe.StreamHeaderInfo{
Unsupported: true,
UnsupportedMsg: "该流格式暂不支持头部数据探测",
})
}
p, err := newParser(url, downloaderType, parserCfg, r.getLogger())
if err != nil {
r.getLogger().WithError(err).Error("failed to init parse")
@@ -497,12 +600,32 @@ func (r *recorder) selectPreferredStream(streamInfos []*live.StreamUrlInfo) (ret
}
func (r *recorder) run(ctx context.Context) {
const minRetryInterval = 5 * time.Second
for {
select {
case <-r.stop:
return
default:
r.tryRecord(ctx)
// 每次 tryRecord 使用独立的子 context
// tryRecord 返回时 cancel 会停止所有异步操作(如 HLS 探测 goroutine)
tryCtx, tryCancel := context.WithCancel(ctx)
start := time.Now()
r.tryRecord(tryCtx)
tryCancel()
// 确保两次 tryRecord 之间至少间隔 minRetryInterval
// 防止快速失败(如 FFmpeg 秒退 404)导致紧密循环
if elapsed := time.Since(start); elapsed < minRetryInterval {
delay := minRetryInterval - elapsed
select {
case <-r.stop:
return
case <-ctx.Done():
return
case <-time.After(delay):
}
}
}
}
}
@@ -572,13 +695,18 @@ func (r *recorder) getCurrentFilePath() string {
}
func (r *recorder) GetStatus() (map[string]interface{}, error) {
var status map[string]interface{}
// 尝试从 parser 获取状态(FFmpeg 进度等)
// 如果 parser 还没创建或不支持 StatusParser,使用空 map 继续
// (流信息 stream_quality/stream_format 等仍然需要返回给前端)
statusP, ok := r.getParser().(parser.StatusParser)
if !ok {
return nil, ErrParserNotSupportStatus
}
status, err := statusP.Status()
if err != nil {
return nil, err
if ok {
var err error
status, err = statusP.Status()
if err != nil {
status = nil
}
}
if status == nil {
status = make(map[string]interface{})
@@ -620,6 +748,48 @@ func (r *recorder) GetStatus() (map[string]interface{}, error) {
status["stream_codec"] = streamInfo.Codec
}
// 添加实际流头部信息(来自 StreamProbe 探测)
if actualInfo := r.actualStreamInfo.Load(); actualInfo != nil {
status["probe_status"] = actualInfo.ProbeStatus()
if actualInfo.Resolution() != "" {
status["actual_resolution"] = actualInfo.Resolution()
}
if actualInfo.VideoCodec != "" {
status["actual_video_codec"] = actualInfo.VideoCodec
}
if actualInfo.AudioCodec != "" {
status["actual_audio_codec"] = actualInfo.AudioCodec
}
if actualInfo.VideoBitrate > 0 {
status["actual_video_bitrate"] = fmt.Sprintf("%d", actualInfo.VideoBitrate)
}
if actualInfo.FrameRate > 0 {
status["actual_frame_rate"] = fmt.Sprintf("%.1f", actualInfo.FrameRate)
}
if actualInfo.Unsupported {
status["probe_message"] = actualInfo.UnsupportedMsg
}
// 判断实际分辨率与平台声称的是否一致
if actualInfo.Width > 0 && actualInfo.Height > 0 && streamInfo != nil {
resolutionMatch := (streamInfo.Width == 0 && streamInfo.Height == 0) || // 平台未提供分辨率,视为一致
(actualInfo.Width == streamInfo.Width && actualInfo.Height == streamInfo.Height)
status["resolution_match"] = resolutionMatch
}
} else {
status["probe_status"] = "pending"
}
// 添加原始流 URL 和 Headers(供前端调试展示)
r.currentFileLock.RLock()
if r.currentStreamURL != "" {
status["stream_url"] = r.currentStreamURL
}
if len(r.currentStreamHeaders) > 0 {
status["stream_headers"] = r.currentStreamHeaders
}
r.currentFileLock.RUnlock()
return status, nil
}
@@ -822,7 +992,10 @@ func (r *recorder) saveAvailableStreamsToDatabase(ctx context.Context, streams [
})
}
saveCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
// 使用 context.Background() 而非传入的 ctx,避免因录制 context 被取消
// 导致数据库写入失败(context canceled)。数据库保存是独立的短暂操作,
// 不应受录制生命周期影响。
saveCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := saver.SaveAvailableStreamsAny(saveCtx, string(r.Live.GetLiveId()), data); err != nil {
+13 -15
View File
@@ -90,25 +90,23 @@ func GetDownloaderAvailability() DownloaderAvailability {
NativeAvailable: true, // 内置解析器永远可用
}
api := tools.Get()
if api == nil {
return result
}
// 检查 FFmpeg
ffmpeg, err := api.GetTool("ffmpeg")
if err == nil && ffmpeg.DoesToolExist() {
// 检查 FFmpeg —— 复用 utils.GetFFmpegPath 保持与录制器实际使用的查找逻辑一致
// (配置文件指定路径 → remotetools → 系统 PATH)
if ffmpegPath, err := utils.GetFFmpegPath(context.Background()); err == nil {
result.FFmpegAvailable = true
result.FFmpegPath = ffmpeg.GetToolPath()
result.FFmpegPath = ffmpegPath
}
// 检查 BililiveRecorder CLI
dotnet, err := api.GetTool("dotnet")
if err == nil && dotnet.DoesToolExist() {
recorder, err := api.GetTool("bililive-recorder-cli")
if err == nil && recorder.DoesToolExist() {
result.BililiveRecorderAvailable = true
result.BililiveRecorderPath = recorder.GetToolPath()
api := tools.Get()
if api != nil {
dotnet, err := api.GetTool("dotnet")
if err == nil && dotnet.DoesToolExist() {
recorder, err := api.GetTool("bililive-recorder-cli")
if err == nil && recorder.DoesToolExist() {
result.BililiveRecorderAvailable = true
result.BililiveRecorderPath = recorder.GetToolPath()
}
}
}
+69 -21
View File
@@ -27,6 +27,10 @@ import CloudUploadSettings from './CloudUploadSettings';
const api = new API();
const { TextArea } = Input;
// 功能开关:代理配置(开发中,设为 false 隐藏 UI)
// 与后端 configs.EnableProxyConfig 对应
const ENABLE_PROXY_CONFIG = false;
const { Panel } = Collapse;
// 配置项类型定义
@@ -687,27 +691,71 @@ const GlobalSettings: React.FC<{
</ConfigField>
</Card>
{/* 代理设置 */}
<Card title="代理设置" size="small" style={{ marginBottom: 16 }}>
<ConfigField
label="启用代理配置"
description="关闭时使用系统环境变量 (HTTP_PROXY, HTTPS_PROXY, ALL_PROXY)"
valueDisplay={config.proxy?.enable ? '已启用' : '使用系统环境变量'}
>
<Form.Item name={['proxy', 'enable']} valuePropName="checked" noStyle>
<Switch />
</Form.Item>
</ConfigField>
<ConfigField
label="代理地址"
description="支持 HTTP 代理 (http://host:port) 和 SOCKS5 代理 (socks5://host:port)"
valueDisplay={config.proxy?.url || '(未设置)'}
>
<Form.Item name={['proxy', 'url']} noStyle>
<Input placeholder="例如: socks5://127.0.0.1:1080 或 http://127.0.0.1:7890" style={{ width: 400 }} />
</Form.Item>
</ConfigField>
</Card>
{/* 代理设置(功能开关控制,开发中) */}
{ENABLE_PROXY_CONFIG && (
<Card title="代理设置" size="small" style={{ marginBottom: 16 }}>
<ConfigField
label="通用代理"
description="关闭时使用系统环境变量 (HTTP_PROXY, HTTPS_PROXY, ALL_PROXY)"
valueDisplay={config.proxy?.enable ? '已启用' : '使用系统环境变量'}
>
<Form.Item name={['proxy', 'enable']} valuePropName="checked" noStyle>
<Switch />
</Form.Item>
</ConfigField>
<ConfigField
label="通用代理地址"
description="同时用于信息获取和下载,除非下方单独配置了专用代理"
valueDisplay={config.proxy?.url || '(未设置)'}
>
<Form.Item name={['proxy', 'url']} noStyle>
<Input placeholder="例如: socks5://127.0.0.1:1080 或 http://127.0.0.1:7890" style={{ width: 400 }} />
</Form.Item>
</ConfigField>
<Divider style={{ margin: '12px 0', fontSize: 12 }}>专用代理(可选覆盖)</Divider>
<ConfigField
label="信息获取代理"
description="用于获取直播间信息、平台 API 请求等。启用后覆盖通用代理。"
>
<Form.Item name={['proxy', 'info_proxy', 'enable']} valuePropName="checked" noStyle>
<Switch />
</Form.Item>
</ConfigField>
<ConfigField
label="信息获取代理地址"
>
<Form.Item name={['proxy', 'info_proxy', 'url']} noStyle>
<Input placeholder="留空则使用通用代理" style={{ width: 400 }} />
</Form.Item>
</ConfigField>
<ConfigField
label="下载代理"
description="用于下载直播流数据。启用后覆盖通用代理。"
>
<Form.Item name={['proxy', 'download_proxy', 'enable']} valuePropName="checked" noStyle>
<Switch />
</Form.Item>
</ConfigField>
<ConfigField
label="下载代理地址"
>
<Form.Item name={['proxy', 'download_proxy', 'url']} noStyle>
<Input placeholder="留空则使用通用代理" style={{ width: 400 }} />
</Form.Item>
</ConfigField>
<Alert
message="代理限制说明"
description="通过 bililive-tools 间接获取信息的平台(如抖音)暂不受代理设置影响。对于这些平台,需要在操作系统层面配置代理。"
type="info"
showIcon
style={{ marginTop: 12 }}
/>
</Card>
)}
{/* 自动更新设置 */}
<Card title="自动更新设置" size="small" style={{ marginBottom: 16 }} id="global-update">
@@ -40,6 +40,7 @@ const CATEGORY_CONFIG: Record<string, { name: string; color: string }> = {
'bililive-tools': { name: 'bililive-tools', color: '#722ed1' },
klive: { name: 'klive', color: '#eb2f96' },
'bililive-recorder': { name: '录播姬', color: '#fa8c16' },
launcher: { name: '启动器', color: '#2f54eb' },
container: { name: '容器总内存', color: '#fa541c' },
total: { name: '总内存', color: '#13c2c2' },
other: { name: '其他', color: '#8c8c8c' },
+171 -10
View File
@@ -92,13 +92,14 @@ const StreamListWithFilter: React.FC<StreamListWithFilterProps> = ({
// 渲染流列表项
const renderStreamItem = (stream: any, index: number) => {
let isCurrentStream = detail.recording && detail.recorder_status?.stream_quality === stream.quality
&& detail.recorder_status?.stream_format === stream.format;
if (isCurrentStream && detail.recorder_status?.stream_attributes_for_stream_select) {
isCurrentStream = Object.entries(detail.recorder_status.stream_attributes_for_stream_select).every(([k, v]) => {
return stream.attributes_for_stream_select[k] === v;
});
// 判断是否为当前录制使用的流
// 仅通过 attributes_for_stream_select 全量比较(包含 format_name、codec、quality 等所有区分字段)
let isCurrentStream = false;
if (detail.recording && detail.recorder_status?.stream_attributes_for_stream_select && stream.attributes_for_stream_select) {
const recorderAttrs = detail.recorder_status.stream_attributes_for_stream_select;
const streamAttrs = stream.attributes_for_stream_select;
isCurrentStream = Object.keys(recorderAttrs).length === Object.keys(streamAttrs).length
&& Object.entries(recorderAttrs).every(([k, v]) => streamAttrs[k] === v);
}
const handleSwitchStream = async () => {
@@ -326,6 +327,7 @@ interface CookieItemData {
interface Room {
roomName: string;
url: string;
lastError?: string;
}
class LiveList extends React.Component<Props, IState> {
@@ -463,7 +465,16 @@ class LiveList extends React.Component<Props, IState> {
title: '直播间名称',
dataIndex: 'room',
key: 'room',
render: (room: Room) => <a href={room.url} rel="noopener noreferrer" target="_blank" onClick={(e) => e.stopPropagation()}>{room.roomName}</a>
render: (room: Room) => (
<span>
<a href={room.url} rel="noopener noreferrer" target="_blank" onClick={(e) => e.stopPropagation()}>{room.roomName}</a>
{room.lastError && (
<Tooltip title={room.lastError}>
<ExclamationCircleOutlined style={{ color: '#ff4d4f', marginLeft: 6, fontSize: 14 }} />
</Tooltip>
)}
</span>
)
},
{
title: '直播平台',
@@ -483,7 +494,16 @@ class LiveList extends React.Component<Props, IState> {
title: '主播名称',
dataIndex: 'name',
key: 'name',
render: (name: string, data: ItemData) => <a href={data.room.url} rel="noopener noreferrer" target="_blank" onClick={(e) => e.stopPropagation()}>{name}</a>
render: (name: string, data: ItemData) => (
<span>
<a href={data.room.url} rel="noopener noreferrer" target="_blank" onClick={(e) => e.stopPropagation()}>{name}</a>
{data.room.lastError && (
<Tooltip title={data.room.lastError}>
<ExclamationCircleOutlined style={{ color: '#ff4d4f', marginLeft: 6, fontSize: 14 }} />
</Tooltip>
)}
</span>
)
},
this.runStatus,
this.runAction
@@ -879,7 +899,8 @@ class LiveList extends React.Component<Props, IState> {
name: item.nick_name || item.host_name,
room: {
roomName: item.room_name,
url: item.live_url
url: item.live_url,
lastError: item.last_error
},
address: item.platform_cn_name,
tags,
@@ -1445,6 +1466,54 @@ class LiveList extends React.Component<Props, IState> {
</Space>
</div>
)}
{/* 实际分辨率信息(来自 StreamProbe 探测) */}
{detail.recording && detail.recorder_status?.probe_status && (
<div style={configRowStyle}>
<span style={configLabelStyle}>实际分辨率</span>
<Space size="small">
{detail.recorder_status.probe_status === 'success' && (
<>
{detail.recorder_status.actual_resolution && (
<Tag color={detail.recorder_status.resolution_match === false ? 'warning' : 'success'}>
{detail.recorder_status.actual_resolution}
{detail.recorder_status.resolution_match === false ? ' ⚠️' : ' ✓'}
</Tag>
)}
{detail.recorder_status.actual_video_codec && (
<Tag>{detail.recorder_status.actual_video_codec.toUpperCase()}</Tag>
)}
{detail.recorder_status.actual_video_bitrate && (
<Tag>{detail.recorder_status.actual_video_bitrate} kbps</Tag>
)}
{detail.recorder_status.actual_frame_rate && (
<Tag>{detail.recorder_status.actual_frame_rate}fps</Tag>
)}
{detail.recorder_status.resolution_match === false && detail.recorder_status.stream_resolution && (
<span style={{ color: '#faad14', fontSize: '12px' }}>
与声称的 {detail.recorder_status.stream_resolution} 不符
</span>
)}
</>
)}
{detail.recorder_status.probe_status === 'unsupported' && (
<Tag color="default">
{detail.recorder_status.actual_video_codec ?
`${detail.recorder_status.actual_video_codec.toUpperCase()} - 无法解析` :
'无法解析'
}
</Tag>
)}
{detail.recorder_status.probe_status === 'pending' && (
<Tag>探测中...</Tag>
)}
{detail.recorder_status.probe_message && (
<span style={{ color: '#999', fontSize: '12px' }}>
{detail.recorder_status.probe_message}
</span>
)}
</Space>
</div>
)}
{detail.recording && detail.recorder_status?.bitrate && (
<div style={configRowStyle}>
<span style={configLabelStyle}>下载速度</span>
@@ -1475,6 +1544,98 @@ class LiveList extends React.Component<Props, IState> {
</Tooltip>
</div>
)}
{/* 录制流调试信息(可折叠) */}
{detail.recording && detail.recorder_status?.stream_url && (
<details style={{ padding: '4px 12px', margin: '4px 0' }}>
<summary style={{
cursor: 'pointer',
color: '#1890ff',
fontSize: '12px',
userSelect: 'none',
outline: 'none',
padding: '4px 0',
}}>
📡 查看录制流 URL 和 Headers
</summary>
<div style={{
marginTop: 8,
padding: '8px 12px',
background: '#f5f5f5',
borderRadius: 6,
fontSize: '12px',
lineHeight: '1.6',
wordBreak: 'break-all',
}}>
<div style={{ marginBottom: 8 }}>
<strong>流 URL:</strong>
<div style={{
fontFamily: 'monospace',
background: '#fff',
padding: '6px 8px',
borderRadius: 4,
border: '1px solid #e8e8e8',
marginTop: 4,
whiteSpace: 'pre-wrap',
}}>
{detail.recorder_status.stream_url}
</div>
</div>
{detail.recorder_status.stream_headers && Object.keys(detail.recorder_status.stream_headers).length > 0 && (
<div style={{ marginBottom: 8 }}>
<strong>Headers:</strong>
<div style={{
fontFamily: 'monospace',
background: '#fff',
padding: '6px 8px',
borderRadius: 4,
border: '1px solid #e8e8e8',
marginTop: 4,
}}>
{Object.entries(detail.recorder_status.stream_headers as Record<string, string>).map(
([k, v]) => (
<div key={k}>
<span style={{ color: '#1890ff' }}>{k}</span>: {v}
</div>
)
)}
</div>
</div>
)}
<Space size="small" style={{ marginTop: 4 }}>
<Button
size="small"
onClick={() => {
navigator.clipboard.writeText(detail.recorder_status.stream_url)
.then(() => message.success('URL 已复制'))
.catch(() => message.error('复制失败'));
}}
>
📋 复制 URL
</Button>
<Button
size="small"
type="primary"
ghost
onClick={() => {
const url = detail.recorder_status.stream_url;
const headers = detail.recorder_status.stream_headers as Record<string, string> | undefined;
let curlCmd = `curl '${url}'`;
if (headers) {
for (const [k, v] of Object.entries(headers)) {
curlCmd += ` \\\n -H '${k}: ${v}'`;
}
}
navigator.clipboard.writeText(curlCmd)
.then(() => message.success('curl 命令已复制'))
.catch(() => message.error('复制失败'));
}}
>
🔧 复制为 curl
</Button>
</Space>
</div>
</details>
)}
<div style={configRowStyle}>
<span style={configLabelStyle}>开播时间</span>
<span>{detail.live_start_time || (detail.status ? '获取中...' : '未开播')}</span>