mirror of
https://github.com/OpenListTeam/OpenList.git
synced 2026-11-14 06:30:47 +08:00
* feat(func): media library
* feat(func): media library
* feat(func): media library
* feat(func): media library build test
* feat(func): media library build test
* feat(func): media library fix remote fetch error
* feat(func): media library fix remote fetch error
* feat(func): media library fix remote fetch error
* feat(func): media library support select
* feat(func): media library support select
* feat(func): media library support select
* feat(func): media library support select
* feat(func): media library support change api url
* feat(func): media library support change api url
* feat(func): media library support change api url
* feat(media): 并发刮削(默认 5)+ 单条落库
增强 TMDB 搜索:模糊匹配 + 多候选关键词
管理页面新增导入/导出媒体库(全部 / 按扫描路径)
缩略图路径默认值改为 imgs
* feat(media): 清空刮削、删除已失效
* fix(media): fix api proxy path
* fix(media): fix lrc file display & music id3
* fix(media): fix media scan sub path error
* feat(media): support cloud codec via ffmpeg
* feat(media): support cloud codec via ffmpeg
* feat(media): support cloud codec via ffmpeg
* feat(media): support cloud codec via ffmpeg
* feat(media): 提升刮削精准度,支持模糊刮削
* fix(media): clean up imports and improve FetchFileReader logic
* feat(media): 提升刮削精准度,支持模糊刮削
* feat(media): 刮削支持自定义刮削数量,避免太占性能
* feat(media): 刮削支持自定义刮削数量,避免太占性能
* feat(media): 刮削修复写入作品名问题
* fix(media): return 404 on missing item_id and fix unscraped item query
- Validate GetMediaItemByID result before spawning goroutine in
StartMediaScrape; return 404 when item_id > 0 but item is not found
- Replace `description` check with `plot` check in GetUnscrappedItems
since scrapers populate Plot, not Description, to avoid re-selecting
already-scraped items on every run
* Apply remaining changes
* feat(media): add master switch, app keys, STRM, Emby and OpenSubsonic APIs
* feat(media): allow any logged-in user to manage their own media app keys
* feat(media): allow signing arbitrary media path for local player
* fix(media): always return an array for media status types
* fix(media): persist per-type scraper config and apply it to thumbnails
* feat(media): add local thumbnail endpoint and cover proxy with fallback gateway
* fix(media): allow guest media access, per-file local thumbnails and optional cover proxy
* feat(media): thumbnail storage modes and STRM same-directory output
* feat(media): Google Books/OpenLibrary sources and remote cover download
* chore(media): rename transcode cache dir to ctmp and log thumbnail storage dir
* feat(media): progress and favorites API shared with Emby/Subsonic
* chore: ignore runtime thumbnail directory
* feat(media): add preferred scrape language and namespace protocol endpoints under @
* fix(media): harden anonymous emby images, pagination and cover proxy SSRF
* fix(emby): allow anonymous user discovery and login before auth
* chore: drop redundant media build workflow and protocol doc
* fix(media): ID3v2.4 syncsafe frame size and preserve Hidden flag on rescan
* fix(media): address security review (path traversal, guest bypass, cover XSS, auth rate limit, SSRF, quotas, migration)
* fix(media): 清理失效条目时只删除确认不存在的记录
此前 fs.Get 返回任何错误都视为失效并硬删除,驱动故障/超时/鉴权失败时会误删有效媒体记录。现在仅 errs.IsObjectNotFound 才删除,其余错误跳过并在响应中返回 skipped 计数。
* fix(transcode): 播放凭据与 Worker 回写凭据分离,并收紧 profile 与鉴权校验
- 新增 Job.PlayToken(json 标签为 -),master/playlist/切片只认播放凭据,播放器不再拿到可上传、可结束任务的 callback_token
- Worker 上传切片时校验 profile 必须属于任务声明的档位
- TCMaster 与 WorkerJobFinish 改用恒定时间比较,避免时序侧信道
* fix(transcode): 含冒号的本地文件名可转码,补基址缺失与重新激活配额
- ValidSourceInput 不再按冒号位置判协议:本地输入只要求以 / 开头或 Windows 盘符路径,concat:/subfile: 等前缀仍被拒且由 -protocol_whitelist 兜底(修复 /mnt/x/Star Wars: IV.mkv 被误判)
- trustedInternalBaseURL 返回空串时直接 503 并给出明确原因,不再拼出 /d/xxx 让 ffmpeg 报文件不存在
- 远程输入白名单收窄为 http,https,tcp,tls(去掉 file 与 crypto)
- Reactivate 复用同一套单用户配额,避免「先让任务闲置取消、再批量激活」绕过 SubmitLimited;配额不足返回 429
- OwnerID 改为配额键:普通用户 user:<id>,访客 guest:<clientIP>,避免单个匿名访客占满全部匿名配额;补充对应单测
* fix(media): 收紧缩略图解码内存、跳过超长条目可见化,并补 CI 单测
- MaxImagePixels 从 64MP 降到 32MP,宽高改 int64 相乘防 32 位溢出,解码阶段加 12 路全局信号量
- 封面解码失败退回原始字节时限制 512KB,避免十几 MB 的 base64 写进 cover 列(豆瓣/本地书籍三处)
- 扫描进度新增 Skipped 计数,入库被跳过的超长路径会在完成消息与日志中体现
- 数据库迁移失败日志补充 OPENLIST_MEDIA_MIGRATE_WIPE=1 的处置提示
- Subsonic 认证只对「确实带了凭据却校验失败」计数,裸探测不再刷新 IP 锁定
- goexif 从 indirect 调整为直接依赖(internal/media/scraper/image.go 直接 import)
- CI 新增 go test ./internal/db/... ./internal/transcode/... ./internal/media/...
* fix(transcode): 配额/队列满时仍放行已缓存切片,并同步 worker 协议白名单
- TCSegment/TCPlaylist: Reactivate 返回 ErrUserQuota/ErrQueueFull 时不再直接 429,改为标记 quotaBlocked,缓存命中的切片与 playlist 仍可读,只有缓存缺失才返回 429
- Reactivate: 补充全局排队上限 (MaxPendingJobs) 检查,避免批量激活顶爆队列,并补单测
- cmd/worker: 远程输入协议白名单与 master 同步为 http,https,tcp,tls(去掉 file/crypto)
- scraper: decodeSem 容量 12 -> 3,避免信号量大于默认刮削并发而形同失效
* feat(emby): 完善客户端兼容层并修复认证锁定,Subsonic 复用统一密钥校验
- Emby: 用户 ID 改用 GUID 并兼容数字/旧 ID;补齐 /Users/{id}/Authenticate、Sessions 与 WebSocket、首页分区、合集/播放列表、字幕流与歌词、Series/Season 结构
- Emby: 版本号返回客户端可接受的兼容版本,/Users/Public 返回可登录用户以渲染用户选择
- Subsonic: 支持 enc: 十六进制密码编码;app key 校验与 Emby 共用 LookUpAppKeyUser
- 限流: 认证被拒时不再刷新锁定过期时间,避免客户端持续轮询把 IP 永久锁死
* fix(media): 认证失败计数增加用户名维度,防止伪造 X-Forwarded-For 绕过锁定
- media_ratelimit: 新增 mediaAuthLockedFor / mediaAuthFailedFor / mediaAuthSucceededFor,IP 与用户名双维度计数;达到上限同样锁定 5 分钟且被拒时不续期
- emby: 按用户名登录与按 id 登录都检查用户名维度;失败计入两个维度,成功清除两个维度
- subsonic: subAuth 与 429 分支改用双维度判定;成功时用真实用户名清理计数(apiKey 认证时 u 参数为空)
- 仍保留主干 5 次/5 分钟的策略,代理信任范围(SetTrustedProxies)留给项目层面单独处理
* fix(media): 匿名封面按访客身份过滤,认证日志脱敏并收紧限流计数
- embyImage: 匿名封面不再走 loadEmbyLibraryAll(全库可见),改为 loadEmbyLibrary(guest):
hidden / base_path 之外 / meta 受保护的条目都不可见,访客被禁用时直接 401
- embyUnauthorized: 只记录 Path、UA 与是否带 token,避免 query 里的 Pw/Password/api_key
和 X-Emby-Authorization 里的设备令牌写进日志
- media_ratelimit: 失败计数 Set 时带 5min TTL,避免 IP/用户名 key 无限增长(用户名来自请求参数)
- auth.loginHash: 锁定期间不再续期,失败计数同样带 TTL,与媒体侧语义一致
- /System/Endpoint: 按 RemoteAddr 判断 IsLocal/IsInNetwork;/System/Info 不再返回真实监听端口
* feat(media): PDF 批注持久化
- 新增 MediaAnnotation 模型:按 (user_id, item_id, remote_id) 唯一,anchor 原样保存查看器批注对象,不重复建模页码/坐标
- GET/POST /fs/media/annotations:读取当前用户批注、按 remote_id 增量 upsert/delete(单条 512KB、单次 8MB、最多 500 条)
- 批注为用户私有数据:未登录返回空列表,访客写入 403,并校验条目可见性与访问权限
- 条目删除 / 清空类型 / 清空扫描路径 / 批量删除时,在同一事务内清理对应批注
* fix(media): 选集签名前校验最终文件权限;转码在 ffprobe 前预检配额
- GetMediaStreamURL / GetMediaStrm: 解析选集后对最终 rawPath 再执行 CanUserAccessPath,避免「目录可见但 meta 隐藏规则禁止其中某个文件」时仍能拿到有效签名(签名之后下载链路不会重跑隐藏规则)
- TranscodePlay: ProbeSource 之前调用 Scheduler.Capacity 预检配额,已满直接 429,不再先起最长 30s 的 ffprobe
- 新增探测并发闸门(全局 8 / 每用户 3),覆盖预检与 SubmitLimited 之间的并发窗口;用 defer 释放,探测 panic 也不会占死槽位
- 补 Capacity 单测
---------
Co-authored-by: j2rong4cn <j2rong@qq.com>
Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: pikachuren <pikachuren@example.com>
657 lines
22 KiB
Go
657 lines
22 KiB
Go
package handles
|
||
|
||
import (
|
||
"crypto/subtle"
|
||
"errors"
|
||
"fmt"
|
||
"net/http"
|
||
"net/url"
|
||
"os"
|
||
"path/filepath"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/OpenListTeam/OpenList/v4/internal/conf"
|
||
"github.com/OpenListTeam/OpenList/v4/internal/driver"
|
||
"github.com/OpenListTeam/OpenList/v4/internal/fs"
|
||
"github.com/OpenListTeam/OpenList/v4/internal/op"
|
||
"github.com/OpenListTeam/OpenList/v4/internal/setting"
|
||
"github.com/OpenListTeam/OpenList/v4/internal/sign"
|
||
"github.com/OpenListTeam/OpenList/v4/internal/transcode"
|
||
"github.com/OpenListTeam/OpenList/v4/server/common"
|
||
"github.com/gin-gonic/gin"
|
||
)
|
||
|
||
// 探测阶段的独立并发上限。
|
||
//
|
||
// 转码配额(每用户 3 个活跃任务)只约束「入队」,而配额判定发生在探测之后:
|
||
// 并发请求可能同时通过配额预检、各起一个最长 30s 的 ffprobe。这里给探测单独
|
||
// 设一道闸,保证这段窗口内的进程数有界。每用户上限取任务配额值,正常并发播放
|
||
// 不会被误伤(用户有 3 个任务时预检已经先拒绝了)。
|
||
const (
|
||
maxConcurrentProbes = 8
|
||
maxProbesPerUser = transcode.MaxActiveJobsPerUser
|
||
)
|
||
|
||
type probeLimiter struct {
|
||
mu sync.Mutex
|
||
active int
|
||
users map[string]int
|
||
}
|
||
|
||
var probeSlots = &probeLimiter{users: make(map[string]int)}
|
||
|
||
func (p *probeLimiter) acquire(owner string) bool {
|
||
p.mu.Lock()
|
||
defer p.mu.Unlock()
|
||
if p.active >= maxConcurrentProbes || p.users[owner] >= maxProbesPerUser {
|
||
return false
|
||
}
|
||
p.active++
|
||
p.users[owner]++
|
||
return true
|
||
}
|
||
|
||
func (p *probeLimiter) release(owner string) {
|
||
p.mu.Lock()
|
||
defer p.mu.Unlock()
|
||
if p.active > 0 {
|
||
p.active--
|
||
}
|
||
if n := p.users[owner]; n <= 1 {
|
||
delete(p.users, owner)
|
||
} else {
|
||
p.users[owner] = n - 1
|
||
}
|
||
}
|
||
|
||
// ============================================================
|
||
// 播放端:/api/transcode/play 发起或复用一个转码任务
|
||
// ============================================================
|
||
|
||
// TranscodePlay 入口:POST /api/fs/transcode body: {path}
|
||
//
|
||
// 行为:
|
||
// 1. 检查 transcode_enabled
|
||
// 2. fs.Get 获取文件并跑 Decide
|
||
// 3. 不需要转码 → 返回 {transcode:false, direct: true}
|
||
// 4. 需要转码 → 创建 Job 入队,返回 {transcode:true, job_id, master_url}
|
||
func TranscodePlay(c *gin.Context) {
|
||
if !transcode.IsEnabled() {
|
||
common.SuccessResp(c, gin.H{"transcode": false, "reason": "disabled"})
|
||
return
|
||
}
|
||
var req struct {
|
||
Path string `json:"path" binding:"required"`
|
||
}
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
common.ErrorResp(c, err, 400)
|
||
return
|
||
}
|
||
// 该路由挂在 Auth(true)(允许被禁用的访客通过)之下,必须在这里自行拒绝,
|
||
// 否则关闭访客后匿名请求仍能触发 ffprobe/ffmpeg 并占用转码缓存。
|
||
user := mediaUser(c)
|
||
if user == nil || (user.IsGuest() && user.Disabled) {
|
||
common.ErrorStrResp(c, "Guest user is disabled, login please", 401)
|
||
return
|
||
}
|
||
if !CanUserAccessPath(user, req.Path) {
|
||
common.ErrorStrResp(c, "path is not accessible", 403)
|
||
return
|
||
}
|
||
obj, err := fs.Get(c.Request.Context(), req.Path, &fs.GetArgs{NoLog: true})
|
||
if err != nil {
|
||
common.ErrorResp(c, err, 500)
|
||
return
|
||
}
|
||
if obj.IsDir() {
|
||
common.ErrorStrResp(c, "path is a directory", 400)
|
||
return
|
||
}
|
||
dec := transcode.Decide(transcode.DecideRequest{
|
||
FileName: obj.GetName(),
|
||
FileSize: obj.GetSize(),
|
||
})
|
||
if !dec.NeedTranscode {
|
||
common.SuccessResp(c, gin.H{
|
||
"transcode": false,
|
||
"reason": dec.Reason,
|
||
})
|
||
return
|
||
}
|
||
|
||
mgr := transcode.Default()
|
||
mgr.Start()
|
||
|
||
// apiURL 仅用于返回给浏览器的 master_url(需要跟随请求 Host 才能被客户端访问)。
|
||
apiURL := common.GetApiUrlFromRequest(c.Request)
|
||
|
||
// 配额键:普通用户按 ID,访客按 IP——访客共用一个账号,
|
||
// 否则一个匿名访客就能占满全部匿名转码配额。
|
||
ownerID := "user:" + strconv.FormatUint(uint64(user.ID), 10)
|
||
if user.IsGuest() {
|
||
ownerID = "guest:" + c.ClientIP()
|
||
}
|
||
|
||
// 同一用户重复请求同一文件:直接复用进行中的任务,避免重复 ffprobe 与重复转码。
|
||
if exist, ok := mgr.Scheduler.FindReusable(ownerID, req.Path); ok {
|
||
respondTranscodeJob(c, apiURL, exist)
|
||
return
|
||
}
|
||
|
||
// 【内存优化】优先尝试解析为本地文件路径,让 ffmpeg 直接读取本地文件,
|
||
// 完全绕过 HTTP /d 代理路径——避免 net.Downloader 给每个 range 请求分配
|
||
// 高达 MaxBufferLimit (默认系统内存 5%) 的大缓冲。这是降低 Go 进程内存
|
||
// 占用最有效的方法。如果不是本地驱动则退回到 HTTP 签名 URL。
|
||
var sourceURL string
|
||
if localFile, ok := resolveLocalFilePath(req.Path); ok {
|
||
// ffmpeg 接受本地路径作为输入,无需 file:// 前缀
|
||
sourceURL = localFile
|
||
fmt.Printf("[transcode] using local file path for ffmpeg: %s\n", localFile)
|
||
} else {
|
||
// 远程驱动:构造签名 HTTP URL。
|
||
// 这个 URL 会被服务端 ffmpeg 直接访问,绝不能使用 Host / X-Forwarded-Host:
|
||
// 未配置 site_url 时攻击者可借此让服务器去拉取任意地址(SSRF)。
|
||
base := trustedInternalBaseURL()
|
||
if base == "" {
|
||
// 仅 HTTPS 监听且未配置 site_url 时无可用基址;
|
||
// 若继续拼出 "/d/..." 会被当成绝对本地路径,ffmpeg 只会报文件不存在。
|
||
common.ErrorStrResp(
|
||
c,
|
||
"transcode needs site_url (or an HTTP port) to fetch the source",
|
||
503,
|
||
)
|
||
return
|
||
}
|
||
signedPath := sign.Sign(req.Path)
|
||
sourceURL = fmt.Sprintf("%s/d%s?sign=%s", base, encodePath(req.Path), signedPath)
|
||
}
|
||
if !transcode.ValidSourceInput(sourceURL) {
|
||
common.ErrorStrResp(c, "unsupported transcode source", 400)
|
||
return
|
||
}
|
||
|
||
// 创建 Job
|
||
job := transcode.NewJob()
|
||
job.Path = req.Path
|
||
job.OwnerID = ownerID
|
||
job.SourceURL = sourceURL
|
||
// 配额预检:ProbeSource 会起一个最长 30s 的 ffprobe 子进程,
|
||
// 不能等探测完才告诉用户「已有 3 个活跃任务」。最终判定仍由 SubmitLimited 原子完成。
|
||
if err := mgr.Scheduler.Capacity(ownerID); err != nil {
|
||
common.ErrorResp(c, err, 429)
|
||
return
|
||
}
|
||
if !probeSlots.acquire(ownerID) {
|
||
common.ErrorStrResp(c, "too many concurrent transcode requests", 429)
|
||
return
|
||
}
|
||
// 【关键】探测视频总时长(异步会让首响应变慢,但只有一次,必须等待)
|
||
// 总时长用于生成完整的 playlist,避免 HLS.js 时长显示错误和拖动跳跃问题
|
||
probe, perr := func() (transcode.SourceProbe, error) {
|
||
// 用 defer 释放:探测内部 panic 时也不能把槽位永久占住
|
||
defer probeSlots.release(ownerID)
|
||
return transcode.ProbeSource(sourceURL)
|
||
}()
|
||
if perr != nil {
|
||
// 探测失败不阻塞流程,但记录日志;后续 playlist 会缺少总时长信息
|
||
fmt.Printf("[transcode] probe source failed: %v\n", perr)
|
||
probe = transcode.SourceProbe{Size: obj.GetSize()}
|
||
} else if probe.Size == 0 {
|
||
probe.Size = obj.GetSize()
|
||
}
|
||
job.Probe = probe
|
||
profile := transcode.BuildProfile()
|
||
job.Profiles = []transcode.Profile{profile}
|
||
job.Output = transcode.BuildOutputSpec()
|
||
timeoutMin := setting.GetInt(conf.TranscodeJobTimeoutMin, 120)
|
||
job.Deadline = time.Now().Add(time.Duration(timeoutMin) * time.Minute).Unix()
|
||
submitted, _, err := mgr.Scheduler.SubmitLimited(job)
|
||
if err != nil {
|
||
// 队列已满 / 用户配额超限:返回 429,前端可降级为直链播放
|
||
common.ErrorResp(c, err, 429)
|
||
return
|
||
}
|
||
respondTranscodeJob(c, apiURL, submitted)
|
||
}
|
||
|
||
// respondTranscodeJob 返回转码任务的播放信息
|
||
func respondTranscodeJob(c *gin.Context, apiURL string, job *transcode.Job) {
|
||
profileName := ""
|
||
if len(job.Profiles) > 0 {
|
||
profileName = job.Profiles[0].Name
|
||
}
|
||
masterURL := fmt.Sprintf("%s/tc/%s/%s/master.m3u8", apiURL, job.ID, job.PlayToken)
|
||
common.SuccessResp(c, gin.H{
|
||
"transcode": true,
|
||
"job_id": job.ID,
|
||
"master_url": masterURL,
|
||
"profile": profileName,
|
||
})
|
||
}
|
||
|
||
func encodePath(p string) string {
|
||
// 与 down.go 保持一致:保留 '/' 不转义
|
||
parts := strings.Split(p, "/")
|
||
for i, s := range parts {
|
||
parts[i] = url.PathEscape(s)
|
||
}
|
||
return strings.Join(parts, "/")
|
||
}
|
||
|
||
// trustedInternalBaseURL 返回供服务端 ffmpeg 访问本站 /d 接口的可信基址,不读取任何请求头。
|
||
// 优先使用管理员配置的 site_url;未配置时回环到本机实际监听的 HTTP 端口。
|
||
// 两者都不可用(例如仅监听 Unix socket)时返回空串,调用方应拒绝转码。
|
||
func trustedInternalBaseURL() string {
|
||
if site := strings.TrimSuffix(conf.Conf.SiteURL, "/"); strings.HasPrefix(site, "http") {
|
||
return site
|
||
}
|
||
if p := conf.Conf.Scheme.HttpPort; p > 0 {
|
||
return fmt.Sprintf("http://127.0.0.1:%d", p)
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// resolveLocalFilePath 尝试把 OpenList 路径解析为宿主机文件系统的实际路径。
|
||
// 仅对本地驱动 (Local) 有效,其他驱动(云盘等)返回 false。
|
||
//
|
||
// 这是内存优化的关键路径:本地文件场景下让 ffmpeg 直接读文件,可以完全绕过
|
||
// HTTP /d 代理路径上的 net.Downloader 大块缓冲(默认每个 range 请求最高
|
||
// 占用 MaxBufferLimit = 系统内存 5%,多 chunk 并发 + 多 range 累计可达数 GB)。
|
||
func resolveLocalFilePath(rawPath string) (string, bool) {
|
||
storage, actualPath, err := op.GetStorageAndActualPath(rawPath)
|
||
if err != nil || storage == nil {
|
||
return "", false
|
||
}
|
||
if storage.Config().Name != "Local" {
|
||
return "", false
|
||
}
|
||
rooter, ok := storage.(driver.IRootPath)
|
||
if !ok {
|
||
return "", false
|
||
}
|
||
root := rooter.GetRootPath()
|
||
if root == "" {
|
||
return "", false
|
||
}
|
||
full := filepath.Join(root, actualPath)
|
||
// 只接受真实存在的文件,避免传给 ffmpeg 一个无效路径
|
||
fi, err := os.Stat(full)
|
||
if err != nil || fi.IsDir() {
|
||
return "", false
|
||
}
|
||
return full, true
|
||
}
|
||
|
||
// ============================================================
|
||
// 播放端:/tc/:job/:token/... Master/Variant playlist & 切片
|
||
// ============================================================
|
||
|
||
// TCMaster GET /tc/:job/:token/master.m3u8
|
||
func TCMaster(c *gin.Context) {
|
||
jobID := c.Param("job")
|
||
token := c.Param("token")
|
||
job, ok := transcode.Default().Scheduler.Get(jobID)
|
||
if !ok {
|
||
c.String(http.StatusNotFound, "job not found")
|
||
return
|
||
}
|
||
if !tokenMatches(job.PlayToken, token) {
|
||
c.String(http.StatusForbidden, "bad token")
|
||
return
|
||
}
|
||
if len(job.Profiles) == 0 {
|
||
c.String(http.StatusInternalServerError, "no profiles")
|
||
return
|
||
}
|
||
apiURL := common.GetApiUrlFromRequest(c.Request)
|
||
baseURL := fmt.Sprintf("%s/tc/%s/%s/", apiURL, jobID, token)
|
||
body := transcode.Default().Cache.MasterPlaylist(job.Profiles, baseURL)
|
||
job.Touch()
|
||
c.Header("Content-Type", "application/vnd.apple.mpegurl")
|
||
// Master playlist 内容在 job 生命周期内不变,设置较长缓存
|
||
c.Header("Cache-Control", "public, max-age=3600")
|
||
c.String(http.StatusOK, body)
|
||
}
|
||
|
||
// TCPlaylist GET /tc/:job/:token/:profile/playlist.m3u8
|
||
// 等待首切片就绪后返回;阻塞最多 30s
|
||
func TCPlaylist(c *gin.Context) {
|
||
jobID := c.Param("job")
|
||
token := c.Param("token")
|
||
profile := c.Param("profile")
|
||
mgr := transcode.Default()
|
||
job, ok := mgr.Scheduler.Get(jobID)
|
||
if !ok {
|
||
c.String(http.StatusNotFound, "job not found")
|
||
return
|
||
}
|
||
if !tokenMatches(job.PlayToken, token) {
|
||
c.String(http.StatusForbidden, "bad token")
|
||
return
|
||
}
|
||
if !transcode.ValidProfileName(profile) {
|
||
c.String(http.StatusBadRequest, "bad profile")
|
||
return
|
||
}
|
||
// 【关键】如果 job 之前因 idle 被 cancel,这里重新激活:复用 jobID/token 重新入队
|
||
// 让 worker 再次启动 ffmpeg 转码,避免播放器看到 404 后无法恢复。
|
||
// 配额/队列已满时不再起新的 ffmpeg,但 playlist 与已缓存的切片仍应可读:
|
||
// 这里只记下 quotaBlocked,交给切片请求在「缓存也没有」时才返回 429。
|
||
quotaBlocked := false
|
||
if st := job.GetStatus(); st == transcode.JobCancelled || st == transcode.JobFinished || st == transcode.JobFailed {
|
||
if _, rerr := mgr.Scheduler.Reactivate(jobID); rerr == nil {
|
||
fmt.Printf("[transcode] reactivate job %s on playlist request (was %s)\n", jobID, st)
|
||
} else if errors.Is(rerr, transcode.ErrUserQuota) || errors.Is(rerr, transcode.ErrQueueFull) {
|
||
quotaBlocked = true
|
||
fmt.Printf("[transcode] reactivate job %s on playlist request rejected (%v), serving from cache only\n", jobID, rerr)
|
||
}
|
||
}
|
||
job.Touch()
|
||
// 等首切片;配额已满时无法启动新的转码,没必要干等 30s
|
||
if !quotaBlocked {
|
||
select {
|
||
case <-job.WaitReady():
|
||
case <-time.After(30 * time.Second):
|
||
c.String(http.StatusGatewayTimeout, "timeout waiting first segment")
|
||
return
|
||
}
|
||
}
|
||
if job.Status == transcode.JobFailed {
|
||
c.String(http.StatusInternalServerError, "transcode failed: "+job.Error)
|
||
return
|
||
}
|
||
apiURL := common.GetApiUrlFromRequest(c.Request)
|
||
baseURL := fmt.Sprintf("%s/tc/%s/%s/%s/", apiURL, jobID, token, profile)
|
||
isFinished := job.Status == transcode.JobFinished
|
||
// 【关键】传入总时长,让 BuildPlaylist 能够生成完整切片列表(VOD 模式),
|
||
// 这样 HLS.js 就知道视频真实总时长,避免时长显示错误和拖动跳跃问题。
|
||
body := transcode.Default().Cache.BuildPlaylist(jobID, profile, job.Output.SegmentDuration,
|
||
isFinished, baseURL, job.Probe.Duration)
|
||
c.Header("Content-Type", "application/vnd.apple.mpegurl")
|
||
if isFinished || job.Probe.Duration > 0 {
|
||
// VOD 模式:playlist 完整不变,长缓存
|
||
c.Header("Cache-Control", "public, max-age=3600")
|
||
} else {
|
||
// 没有总时长时退回 EVENT 模式:短缓存
|
||
c.Header("Cache-Control", "public, max-age=2")
|
||
}
|
||
c.String(http.StatusOK, body)
|
||
}
|
||
|
||
// TCSegment GET /tc/:job/:token/:profile/seg-:seq.ts
|
||
// 切片就绪则直接返回;否则等待最多 60s
|
||
func TCSegment(c *gin.Context) {
|
||
jobID := c.Param("job")
|
||
token := c.Param("token")
|
||
profile := c.Param("profile")
|
||
segName := c.Param("seg")
|
||
mgr := transcode.Default()
|
||
job, ok := mgr.Scheduler.Get(jobID)
|
||
if !ok {
|
||
c.String(http.StatusNotFound, "job not found")
|
||
return
|
||
}
|
||
if !tokenMatches(job.PlayToken, token) {
|
||
c.String(http.StatusForbidden, "bad token")
|
||
return
|
||
}
|
||
if !transcode.ValidProfileName(profile) {
|
||
c.String(http.StatusBadRequest, "bad profile")
|
||
return
|
||
}
|
||
// 【关键】job 之前因 idle 被 cancel 时,复用同一 jobID/token 重新入队启动转码。
|
||
// 这样浏览器持有的旧 master.m3u8 + 切片 URL 仍然有效,暂停再继续也不会 404。
|
||
// 配额/队列已满时不重新起 ffmpeg,但已缓存的切片依然要能命中(回看已转完的视频)。
|
||
quotaBlocked := false
|
||
if st := job.GetStatus(); st == transcode.JobCancelled || st == transcode.JobFinished || st == transcode.JobFailed {
|
||
if _, rerr := mgr.Scheduler.Reactivate(jobID); rerr == nil {
|
||
fmt.Printf("[transcode] reactivate job %s on segment request (was %s)\n", jobID, st)
|
||
} else if errors.Is(rerr, transcode.ErrUserQuota) || errors.Is(rerr, transcode.ErrQueueFull) {
|
||
quotaBlocked = true
|
||
fmt.Printf("[transcode] reactivate job %s on segment request rejected (%v), serving from cache only\n", jobID, rerr)
|
||
}
|
||
}
|
||
// seg-N.ts
|
||
name := strings.TrimSuffix(strings.TrimPrefix(segName, "seg-"), ".ts")
|
||
seq, err := strconv.Atoi(name)
|
||
if err != nil {
|
||
c.String(http.StatusBadRequest, "bad segment name")
|
||
return
|
||
}
|
||
cache := mgr.Cache
|
||
// 【智能调度】触发对应 chunk 的 ffmpeg 启动(如已运行则只更新 LastAccess)
|
||
// 这是用户拖动进度条到任意位置时能快速响应的关键。
|
||
// 配额已满时不能触发新的 ffmpeg,否则「限制并发」就没有意义了。
|
||
if mgr.Chunks != nil && !quotaBlocked {
|
||
chunkIdx := mgr.Chunks.EnsureChunkRunningForSeg(jobID, profile, seq)
|
||
fmt.Printf("[tc-segment] req seq=%d chunk=%d job=%s\n", seq, chunkIdx, jobID)
|
||
}
|
||
info, err := cache.GetSegment(jobID, profile, seq)
|
||
if err != nil {
|
||
if quotaBlocked {
|
||
// 配额已满且该切片不在缓存里:只能拒绝,让前端降级为直链播放
|
||
c.String(http.StatusTooManyRequests, "too many active transcode jobs")
|
||
return
|
||
}
|
||
// 等待
|
||
waitStart := time.Now()
|
||
info, err = cache.WaitForSegment(jobID, profile, seq, 60*time.Second)
|
||
if err != nil {
|
||
fmt.Printf("[tc-segment] req seq=%d FAILED after %.1fs: %v\n", seq, time.Since(waitStart).Seconds(), err)
|
||
c.String(http.StatusNotFound, "segment not available: "+err.Error())
|
||
return
|
||
}
|
||
fmt.Printf("[tc-segment] req seq=%d ready after %.1fs\n", seq, time.Since(waitStart).Seconds())
|
||
}
|
||
job.Touch()
|
||
// 切片内容不可变,设置强缓存避免浏览器/HLS.js 反复请求同一切片
|
||
c.Header("Cache-Control", "public, max-age=86400, immutable")
|
||
// 【内存优化】流式发送切片文件,避免 gin c.File() 将整个切片读入内存
|
||
// 注意:不能使用 http.ServeFile,它会自动嗅探 Content-Type 并处理 Range 请求,
|
||
// 与 gin 已写入的响应头冲突,导致 HLS.js 收到异常数据触发 bufferAppendError。
|
||
f, err := os.Open(info.Path)
|
||
if err != nil {
|
||
c.String(http.StatusInternalServerError, "open segment: "+err.Error())
|
||
return
|
||
}
|
||
defer f.Close()
|
||
fi, err := f.Stat()
|
||
if err != nil {
|
||
c.String(http.StatusInternalServerError, "stat segment: "+err.Error())
|
||
return
|
||
}
|
||
c.DataFromReader(http.StatusOK, fi.Size(), "video/mp2t", f, nil)
|
||
}
|
||
|
||
// ============================================================
|
||
// Worker 协议端
|
||
// ============================================================
|
||
|
||
// WorkerRegister POST /api/transcode/worker/register
|
||
func WorkerRegister(c *gin.Context) {
|
||
if !transcode.IsEnabled() {
|
||
common.ErrorStrResp(c, "transcode disabled", 403)
|
||
return
|
||
}
|
||
if !checkWorkerSecret(c) {
|
||
common.ErrorStrResp(c, "unauthorized", 401)
|
||
return
|
||
}
|
||
var req transcode.RegisterRequest
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
common.ErrorResp(c, err, 400)
|
||
return
|
||
}
|
||
w := transcode.Default().Registry.Register(&req, false)
|
||
common.SuccessResp(c, transcode.RegisterResponse{
|
||
WorkerID: w.ID,
|
||
HeartbeatInterval: 10,
|
||
ClaimStrategy: "pull",
|
||
ProtocolVersion: transcode.ProtocolVersion,
|
||
})
|
||
}
|
||
|
||
// WorkerHeartbeat POST /api/transcode/worker/heartbeat
|
||
func WorkerHeartbeat(c *gin.Context) {
|
||
if !checkWorkerSecret(c) {
|
||
common.ErrorStrResp(c, "unauthorized", 401)
|
||
return
|
||
}
|
||
var req transcode.HeartbeatRequest
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
common.ErrorResp(c, err, 400)
|
||
return
|
||
}
|
||
mgr := transcode.Default()
|
||
if _, ok := mgr.Registry.Heartbeat(&req); !ok {
|
||
common.SuccessResp(c, transcode.HeartbeatResponse{OK: false, Kick: true})
|
||
return
|
||
}
|
||
cancellations := mgr.Scheduler.PopCancellations(req.WorkerID)
|
||
common.SuccessResp(c, transcode.HeartbeatResponse{OK: true, Cancel: cancellations})
|
||
}
|
||
|
||
// WorkerClaim POST /api/transcode/worker/claim
|
||
func WorkerClaim(c *gin.Context) {
|
||
if !checkWorkerSecret(c) {
|
||
common.ErrorStrResp(c, "unauthorized", 401)
|
||
return
|
||
}
|
||
var req transcode.ClaimRequest
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
common.ErrorResp(c, err, 400)
|
||
return
|
||
}
|
||
wait := time.Duration(req.Wait) * time.Second
|
||
if wait > 30*time.Second {
|
||
wait = 30 * time.Second
|
||
}
|
||
jobs := transcode.Default().Scheduler.Claim(req.WorkerID, req.Slots, wait)
|
||
common.SuccessResp(c, transcode.ClaimResponse{Jobs: jobs})
|
||
}
|
||
|
||
// WorkerSegmentUpload PUT /api/transcode/worker/segment?job=&profile=&seq=&duration=&final=
|
||
func WorkerSegmentUpload(c *gin.Context) {
|
||
jobID := c.Query("job")
|
||
profile := c.Query("profile")
|
||
seq, _ := strconv.Atoi(c.Query("seq"))
|
||
dur, _ := strconv.ParseFloat(c.Query("duration"), 64)
|
||
isFinal, _ := strconv.ParseBool(c.Query("final"))
|
||
job, ok := transcode.Default().Scheduler.Get(jobID)
|
||
if !ok {
|
||
common.ErrorStrResp(c, "job not found", 404)
|
||
return
|
||
}
|
||
// 校验 callback_token (Bearer),使用恒定时间比较避免时序侧信道
|
||
auth := c.GetHeader("Authorization")
|
||
expect := "Bearer " + job.CallbackToken
|
||
if subtle.ConstantTimeCompare([]byte(auth), []byte(expect)) != 1 {
|
||
common.ErrorStrResp(c, "unauthorized", 401)
|
||
return
|
||
}
|
||
// profile 会被拼进缓存目录路径,必须是白名单字符,防止 ../ 路径穿越
|
||
if !transcode.ValidProfileName(profile) {
|
||
common.ErrorStrResp(c, "invalid profile", 400)
|
||
return
|
||
}
|
||
// 只接受任务声明过的档位,避免持有 callback_token 的一方往任意档位目录写数据
|
||
if !jobHasProfile(job, profile) {
|
||
common.ErrorStrResp(c, "profile not declared for this job", 400)
|
||
return
|
||
}
|
||
defer c.Request.Body.Close()
|
||
_, err := transcode.Default().Cache.PutSegment(jobID, profile, seq, dur, c.Request.Body, isFinal)
|
||
if err != nil {
|
||
switch err {
|
||
case transcode.ErrInvalidProfile, transcode.ErrInvalidJobID, transcode.ErrInvalidSegment:
|
||
common.ErrorResp(c, err, 400)
|
||
case transcode.ErrSegmentTooLarge:
|
||
common.ErrorResp(c, err, 413)
|
||
default:
|
||
common.ErrorResp(c, err, 500)
|
||
}
|
||
return
|
||
}
|
||
if seq == 0 {
|
||
job.MarkReady()
|
||
}
|
||
c.Status(http.StatusNoContent)
|
||
}
|
||
|
||
// WorkerJobFinish POST /api/transcode/worker/job/finish
|
||
func WorkerJobFinish(c *gin.Context) {
|
||
var req transcode.FinishRequest
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
common.ErrorResp(c, err, 400)
|
||
return
|
||
}
|
||
job, ok := transcode.Default().Scheduler.Get(req.JobID)
|
||
if !ok {
|
||
common.ErrorStrResp(c, "job not found", 404)
|
||
return
|
||
}
|
||
auth := c.GetHeader("Authorization")
|
||
if !tokenMatches("Bearer "+job.CallbackToken, auth) {
|
||
common.ErrorStrResp(c, "unauthorized", 401)
|
||
return
|
||
}
|
||
if err := transcode.Default().Scheduler.Finish(&req); err != nil {
|
||
common.ErrorResp(c, err, 500)
|
||
return
|
||
}
|
||
common.SuccessResp(c)
|
||
}
|
||
|
||
// ListTranscodeWorkers GET /api/admin/transcode/workers (admin)
|
||
func ListTranscodeWorkers(c *gin.Context) {
|
||
common.SuccessResp(c, transcode.Default().Registry.List())
|
||
}
|
||
|
||
// ListTranscodeJobs GET /api/admin/transcode/jobs (admin)
|
||
func ListTranscodeJobs(c *gin.Context) {
|
||
common.SuccessResp(c, transcode.Default().Scheduler.ListJobs())
|
||
}
|
||
|
||
// CancelTranscodeJob POST /api/admin/transcode/cancel
|
||
func CancelTranscodeJob(c *gin.Context) {
|
||
var req struct {
|
||
JobID string `json:"job_id" binding:"required"`
|
||
}
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
common.ErrorResp(c, err, 400)
|
||
return
|
||
}
|
||
transcode.Default().Scheduler.Cancel(req.JobID)
|
||
common.SuccessResp(c)
|
||
}
|
||
|
||
// jobHasProfile 判断 profile 是否属于任务声明的输出档位
|
||
func jobHasProfile(job *transcode.Job, profile string) bool {
|
||
if job == nil {
|
||
return false
|
||
}
|
||
for _, p := range job.Profiles {
|
||
if p.Name == profile {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// tokenMatches 恒定时间比较凭据,避免按字节比较造成的时序侧信道
|
||
func tokenMatches(expected, got string) bool {
|
||
if expected == "" || got == "" {
|
||
return false
|
||
}
|
||
return subtle.ConstantTimeCompare([]byte(expected), []byte(got)) == 1
|
||
}
|
||
|
||
func checkWorkerSecret(c *gin.Context) bool {
|
||
auth := c.GetHeader("Authorization")
|
||
auth = strings.TrimPrefix(auth, "Bearer ")
|
||
return transcode.VerifyWorkerSecret(auth)
|
||
}
|