diff --git a/Makefile b/Makefile index 3cfbf21..d606703 100644 --- a/Makefile +++ b/Makefile @@ -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) diff --git a/go.mod b/go.mod index 4f7f15c..ac0b573 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 57211b0..6d0dff8 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/src/cmd/build/build_go_binary.go b/src/cmd/build/build_go_binary.go index 2ba7d9e..7236797 100644 --- a/src/cmd/build/build_go_binary.go +++ b/src/cmd/build/build_go_binary.go @@ -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) diff --git a/src/cmd/build/cmd.go b/src/cmd/build/cmd.go index faf47a5..9b0d979 100644 --- a/src/cmd/build/cmd.go +++ b/src/cmd/build/cmd.go @@ -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 } diff --git a/src/configs/config.go b/src/configs/config.go index 2e4c076..fb8000f 100644 --- a/src/configs/config.go +++ b/src/configs/config.go @@ -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{ diff --git a/src/configs/config_comments.go b/src/configs/config_comments.go index 0fd510d..3cf2d70 100644 --- a/src/configs/config_comments.go +++ b/src/configs/config_comments.go @@ -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 功能配置注释 diff --git a/src/configs/feature_flags.go b/src/configs/feature_flags.go new file mode 100644 index 0000000..e7c189f --- /dev/null +++ b/src/configs/feature_flags.go @@ -0,0 +1,10 @@ +package configs + +// 功能特征标志 +// 用于控制正在开发中的功能是否启用 +// 开发者可以在本地修改这些值来启用/禁用功能 + +// EnableProxyConfig 控制是否启用代理配置功能 +// false: 隐藏前端代理配置 UI，后端代理函数回退到环境变量 +// true: 启用完整的代理配置（通用代理 + 信息获取代理 + 下载代理） +const EnableProxyConfig = false diff --git a/src/consts/build_dev.go b/src/consts/build_dev.go new file mode 100644 index 0000000..ac0d8a4 --- /dev/null +++ b/src/consts/build_dev.go @@ -0,0 +1,6 @@ +//go:build dev + +package consts + +// IsDevBuild 标记当前是否为开发编译版本 +const IsDevBuild = true diff --git a/src/consts/build_release.go b/src/consts/build_release.go new file mode 100644 index 0000000..6ba6dd8 --- /dev/null +++ b/src/consts/build_release.go @@ -0,0 +1,6 @@ +//go:build !dev + +package consts + +// IsDevBuild 标记当前是否为开发编译版本 +const IsDevBuild = false diff --git a/src/live/info.go b/src/live/info.go index c1063c4..5276dd7 100644 --- a/src/live/info.go +++ b/src/live/info.go @@ -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, } diff --git a/src/live/lives.go b/src/live/lives.go index 9dbd813..9454236 100755 --- a/src/live/lives.go +++ b/src/live/lives.go @@ -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) } diff --git a/src/pkg/iostats/memory_collector.go b/src/pkg/iostats/memory_collector.go index 10309c2..05843e4 100644 --- a/src/pkg/iostats/memory_collector.go +++ b/src/pkg/iostats/memory_collector.go @@ -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, diff --git a/src/pkg/iostats/memory_types.go b/src/pkg/iostats/memory_types.go index 5ab271d..3b7fa55 100644 --- a/src/pkg/iostats/memory_types.go +++ b/src/pkg/iostats/memory_types.go @@ -40,6 +40,8 @@ const ( MemoryCategoryKlive = "klive" // MemoryCategoryRecorder BililiveRecorder 子进程 MemoryCategoryRecorder = "bililive-recorder" + // MemoryCategoryLauncher 启动器进程 + MemoryCategoryLauncher = "launcher" // MemoryCategoryContainer 容器内存 MemoryCategoryContainer = "container" // MemoryCategoryTotal 总内存（self + 所有子进程） diff --git a/src/pkg/launcher/launcher.go b/src/pkg/launcher/launcher.go index 35a58d2..9277537 100644 --- a/src/pkg/launcher/launcher.go +++ b/src/pkg/launcher/launcher.go @@ -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("主程序已退出") } diff --git a/src/pkg/parser/ffmpeg/ffmpeg.go b/src/pkg/parser/ffmpeg/ffmpeg.go index 0fe9f3c..1a117b1 100644 --- a/src/pkg/parser/ffmpeg/ffmpeg.go +++ b/src/pkg/parser/ffmpeg/ffmpeg.go @@ -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 { diff --git a/src/pkg/proxy/proxy.go b/src/pkg/proxy/proxy.go index 464f43c..9e8e247 100644 --- a/src/pkg/proxy/proxy.go +++ b/src/pkg/proxy/proxy.go @@ -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 } diff --git a/src/pkg/streamprobe/flv_parser.go b/src/pkg/streamprobe/flv_parser.go new file mode 100644 index 0000000..1071ce4 --- /dev/null +++ b/src/pkg/streamprobe/flv_parser.go @@ -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 +} diff --git a/src/pkg/streamprobe/hls_parser.go b/src/pkg/streamprobe/hls_parser.go new file mode 100644 index 0000000..244593c --- /dev/null +++ b/src/pkg/streamprobe/hls_parser.go @@ -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) +} diff --git a/src/pkg/streamprobe/info.go b/src/pkg/streamprobe/info.go new file mode 100644 index 0000000..7da16e7 --- /dev/null +++ b/src/pkg/streamprobe/info.go @@ -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" +} diff --git a/src/pkg/streamprobe/probe.go b/src/pkg/streamprobe/probe.go new file mode 100644 index 0000000..e841fa2 --- /dev/null +++ b/src/pkg/streamprobe/probe.go @@ -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 + } +} diff --git a/src/pkg/utils/conn_counter.go b/src/pkg/utils/conn_counter.go index da3dc0f..fcad298 100644 --- a/src/pkg/utils/conn_counter.go +++ b/src/pkg/utils/conn_counter.go @@ -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 } diff --git a/src/recorders/recorder.go b/src/recorders/recorder.go index aa3af53..1beaf5c 100644 --- a/src/recorders/recorder.go +++ b/src/recorders/recorder.go @@ -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 { diff --git a/src/tools/tools.go b/src/tools/tools.go index 9fb5f44..9088264 100644 --- a/src/tools/tools.go +++ b/src/tools/tools.go @@ -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() + } } } diff --git a/src/webapp/src/component/config-info/index.tsx b/src/webapp/src/component/config-info/index.tsx index 812cc66..7343129 100644 --- a/src/webapp/src/component/config-info/index.tsx +++ b/src/webapp/src/component/config-info/index.tsx @@ -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<{ - {/* 代理设置 */} - - - - - - - - - - - - + {/* 代理设置（功能开关控制，开发中） */} + {ENABLE_PROXY_CONFIG && ( + + + + + + + + + + + + + 专用代理（可选覆盖） + + + + + + + + + + + + + + + + + + + + + + + + + + )} {/* 自动更新设置 */} diff --git a/src/webapp/src/component/io-stats/MemoryHistoryChart.tsx b/src/webapp/src/component/io-stats/MemoryHistoryChart.tsx index 0852ac8..04915e5 100644 --- a/src/webapp/src/component/io-stats/MemoryHistoryChart.tsx +++ b/src/webapp/src/component/io-stats/MemoryHistoryChart.tsx @@ -40,6 +40,7 @@ const CATEGORY_CONFIG: Record = { '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' }, diff --git a/src/webapp/src/component/live-list/index.tsx b/src/webapp/src/component/live-list/index.tsx index 3fe0dd6..81c51c1 100644 --- a/src/webapp/src/component/live-list/index.tsx +++ b/src/webapp/src/component/live-list/index.tsx @@ -92,13 +92,14 @@ const StreamListWithFilter: React.FC = ({ // 渲染流列表项 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 { @@ -463,7 +465,16 @@ class LiveList extends React.Component { title: '直播间名称', dataIndex: 'room', key: 'room', - render: (room: Room) => e.stopPropagation()}>{room.roomName} + render: (room: Room) => ( + + e.stopPropagation()}>{room.roomName} + {room.lastError && ( + + + + )} + + ) }, { title: '直播平台', @@ -483,7 +494,16 @@ class LiveList extends React.Component { title: '主播名称', dataIndex: 'name', key: 'name', - render: (name: string, data: ItemData) => e.stopPropagation()}>{name} + render: (name: string, data: ItemData) => ( + + e.stopPropagation()}>{name} + {data.room.lastError && ( + + + + )} + + ) }, this.runStatus, this.runAction @@ -879,7 +899,8 @@ class LiveList extends React.Component { 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 { )} + {/* 实际分辨率信息（来自 StreamProbe 探测） */} + {detail.recording && detail.recorder_status?.probe_status && ( +
+ 实际分辨率 + + {detail.recorder_status.probe_status === 'success' && ( + <> + {detail.recorder_status.actual_resolution && ( + + {detail.recorder_status.actual_resolution} + {detail.recorder_status.resolution_match === false ? ' ⚠️' : ' ✓'} + + )} + {detail.recorder_status.actual_video_codec && ( + {detail.recorder_status.actual_video_codec.toUpperCase()} + )} + {detail.recorder_status.actual_video_bitrate && ( + {detail.recorder_status.actual_video_bitrate} kbps + )} + {detail.recorder_status.actual_frame_rate && ( + {detail.recorder_status.actual_frame_rate}fps + )} + {detail.recorder_status.resolution_match === false && detail.recorder_status.stream_resolution && ( + + 与声称的 {detail.recorder_status.stream_resolution} 不符 + + )} + + )} + {detail.recorder_status.probe_status === 'unsupported' && ( + + {detail.recorder_status.actual_video_codec ? + `${detail.recorder_status.actual_video_codec.toUpperCase()} - 无法解析` : + '无法解析' + } + + )} + {detail.recorder_status.probe_status === 'pending' && ( + 探测中... + )} + {detail.recorder_status.probe_message && ( + + {detail.recorder_status.probe_message} + + )} + +
+ )} {detail.recording && detail.recorder_status?.bitrate && (
下载速度 @@ -1475,6 +1544,98 @@ class LiveList extends React.Component {
)} + {/* 录制流调试信息（可折叠） */} + {detail.recording && detail.recorder_status?.stream_url && ( +
+ + 📡 查看录制流 URL 和 Headers + +
+
+ 流 URL： +
+ {detail.recorder_status.stream_url} +
+
+ {detail.recorder_status.stream_headers && Object.keys(detail.recorder_status.stream_headers).length > 0 && ( +
+ Headers： +
+ {Object.entries(detail.recorder_status.stream_headers as Record).map( + ([k, v]) => ( +
+ {k}: {v} +
+ ) + )} +
+
+ )} + + + + +
+
+ )}
开播时间 {detail.live_start_time || (detail.status ? '获取中...' : '未开播')}