mirror of
https://github.com/truewhile/MeBox.git
synced 2026-09-29 03:26:37 +08:00
truewhile
599 lines
19 KiB
Go
599 lines
19 KiB
Go
// STRM 元数据下载/上传队列 worker。
|
|
//
|
|
// 下载队列：远端网盘 → 本地输出目录（nfo/图片/字幕）；上传队列：本地 → 远端。
|
|
// 任务持久化在 DB，worker 轮询认领；失败按指数退避重试，超过上限标记 failed。
|
|
package service
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
"github.com/ShukeBta/MMTL/internal/model"
|
|
"github.com/ShukeBta/MMTL/internal/service/cloud"
|
|
"github.com/ShukeBta/MMTL/internal/service/cloud115"
|
|
)
|
|
|
|
const (
|
|
strmMaxTaskRetry = 3
|
|
)
|
|
|
|
// downloadWorker 下载队列 worker：认领 → 解析直链 → 下载 → 落盘。
|
|
//
|
|
// 采用「批量认领 + 全局并发限流」：一次认领数个任务，用 StrmService 上的全局信号量
|
|
// 限制整个进程「同时换直链+下载」的并发数（与 115 换链风控匹配，见 strmDownloadSemCap），
|
|
// 同时让下载充分并行。换链走全局令牌桶（QPS=3）兜底，下载走 CDN 不限速。
|
|
func (s *StrmService) downloadWorker(ctx context.Context) {
|
|
const claimBatch = 12 // 每次批量认领的任务数
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-s.stopCh:
|
|
return
|
|
default:
|
|
}
|
|
// 115 风控/限流熔断：冷却期间整体暂停，不给 WAF 续封机会
|
|
if left := s.wafCooldownLeft(); left > 0 {
|
|
s.log.Debug("下载队列冷却中", zap.Duration("remaining", left))
|
|
sleepContext(ctx, left)
|
|
continue
|
|
}
|
|
tasks, err := s.repo.StrmDownload.ClaimPendingDownload(ctx, claimBatch)
|
|
if err != nil {
|
|
s.log.Warn("claim strm download task failed", zap.Error(err))
|
|
sleepContext(ctx, 3*time.Second)
|
|
continue
|
|
}
|
|
if len(tasks) == 0 {
|
|
sleepContext(ctx, 2*time.Second)
|
|
continue
|
|
}
|
|
// 并发处理本批任务：每个任务先获取全局下载槽位，槽位内部执行换链+下载。
|
|
// 信号量与令牌桶双重限速，确保任意时刻并发换链请求不超过安全阈值。
|
|
var wg sync.WaitGroup
|
|
for i := range tasks {
|
|
wg.Add(1)
|
|
go func(i int) {
|
|
defer wg.Done()
|
|
if !s.acquireDownloadSlot(ctx) {
|
|
return
|
|
}
|
|
defer s.releaseDownloadSlot()
|
|
s.processDownloadTask(ctx, &tasks[i])
|
|
}(i)
|
|
}
|
|
wg.Wait()
|
|
}
|
|
}
|
|
|
|
func (s *StrmService) processDownloadTask(ctx context.Context, task *model.StrmDownloadTask) {
|
|
cleanPath := sanitizeLocalPath(task.LocalPath)
|
|
if cleanPath != "" && cleanPath != task.LocalPath {
|
|
task.LocalPath = cleanPath
|
|
_ = s.repo.StrmDownload.Update(context.Background(), task)
|
|
}
|
|
finish := func(status, message string) {
|
|
now := time.Now()
|
|
task.Status = status
|
|
task.Error = message
|
|
task.FinishedAt = &now
|
|
if err := s.repo.StrmDownload.Update(context.Background(), task); err != nil {
|
|
s.log.Warn("update strm download task failed", zap.Error(err))
|
|
}
|
|
}
|
|
acct, err := s.repo.StrmAccount.FindByID(ctx, task.AccountID)
|
|
if err != nil || acct == nil {
|
|
finish(model.StrmTaskFailed, "网盘账号不存在")
|
|
return
|
|
}
|
|
provider, err := s.providerFor(ctx, acct)
|
|
if err != nil {
|
|
s.downloadTaskFailWithRetry(task, err.Error())
|
|
return
|
|
}
|
|
link, err := provider.Resolve(ctx, task.RemoteRef)
|
|
if err != nil {
|
|
if is115Blocked(err) {
|
|
s.triggerWAFCooldown()
|
|
}
|
|
s.downloadTaskFailWithRetry(task, "解析下载地址失败："+err.Error())
|
|
return
|
|
}
|
|
if err := downloadToFile(ctx, link, task.LocalPath, s.http); err != nil {
|
|
// 直链失效（403/404/410 等）：清掉缓存让下一轮重新换取
|
|
if isHTTPDownloadFailure(err) && task.Provider == model.StrmProvider115 {
|
|
cloud115.ClearDownloadURLCache(task.RemoteRef)
|
|
}
|
|
s.downloadTaskFailWithRetry(task, "下载失败："+err.Error())
|
|
return
|
|
}
|
|
finish(model.StrmTaskDone, "")
|
|
}
|
|
|
|
// uploadWorker 上传队列 worker：认领 → WebDAV/OpenList 上传 → 收尾。
|
|
func (s *StrmService) uploadWorker(ctx context.Context) {
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-s.stopCh:
|
|
return
|
|
default:
|
|
}
|
|
tasks, err := s.repo.StrmUpload.ClaimPendingUpload(ctx, 1)
|
|
if err != nil {
|
|
s.log.Warn("claim strm upload task failed", zap.Error(err))
|
|
sleepContext(ctx, 3*time.Second)
|
|
continue
|
|
}
|
|
if len(tasks) == 0 {
|
|
sleepContext(ctx, 2*time.Second)
|
|
continue
|
|
}
|
|
for i := range tasks {
|
|
s.processUploadTask(ctx, &tasks[i])
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *StrmService) processUploadTask(ctx context.Context, task *model.StrmUploadTask) {
|
|
finish := func(status, message string) {
|
|
now := time.Now()
|
|
task.Status = status
|
|
task.Error = message
|
|
task.FinishedAt = &now
|
|
if err := s.repo.StrmUpload.Update(context.Background(), task); err != nil {
|
|
s.log.Warn("update strm upload task failed", zap.Error(err))
|
|
}
|
|
}
|
|
if task.Provider == model.StrmProvider115 {
|
|
finish(model.StrmTaskFailed, "115 网盘暂不支持元数据上传")
|
|
return
|
|
}
|
|
acct, err := s.repo.StrmAccount.FindByID(ctx, task.AccountID)
|
|
if err != nil || acct == nil {
|
|
finish(model.StrmTaskFailed, "网盘账号不存在")
|
|
return
|
|
}
|
|
cfg, err := s.strmAccountConfig(acct)
|
|
if err == nil && task.Provider == model.StrmProviderOpenList && cfg["token"] == "" && cfg["password"] == "" {
|
|
finish(model.StrmTaskFailed, "OpenList 账号需要配置 Token 或密码才能上传")
|
|
return
|
|
}
|
|
provider, err := s.providerFor(ctx, acct)
|
|
if err != nil {
|
|
s.uploadTaskFailWithRetry(task, err.Error())
|
|
return
|
|
}
|
|
putter, ok := provider.(interface {
|
|
PutFile(ctx context.Context, remotePath string, r io.Reader) error
|
|
})
|
|
if !ok {
|
|
finish(model.StrmTaskFailed, "该网盘不支持元数据上传")
|
|
return
|
|
}
|
|
f, err := os.Open(task.LocalPath)
|
|
if err != nil {
|
|
s.uploadTaskFailWithRetry(task, "打开本地文件失败："+err.Error())
|
|
return
|
|
}
|
|
if err := putter.PutFile(ctx, task.RemotePath, f); err != nil {
|
|
_ = f.Close()
|
|
s.uploadTaskFailWithRetry(task, "上传失败："+err.Error())
|
|
return
|
|
}
|
|
_ = f.Close()
|
|
finish(model.StrmTaskDone, "")
|
|
}
|
|
|
|
// downloadTaskFailWithRetry 下载失败任务按退避重试，超过上限标记 failed。
|
|
func (s *StrmService) downloadTaskFailWithRetry(task *model.StrmDownloadTask, message string) {
|
|
if !retryTask(&task.RetryCount, &task.Status, &task.Error, &task.NextTryAt, &task.FinishedAt, message) {
|
|
return
|
|
}
|
|
_ = s.repo.StrmDownload.Update(context.Background(), task)
|
|
}
|
|
|
|
// uploadTaskFailWithRetry 上传失败任务按退避重试，超过上限标记 failed。
|
|
func (s *StrmService) uploadTaskFailWithRetry(task *model.StrmUploadTask, message string) {
|
|
if !retryTask(&task.RetryCount, &task.Status, &task.Error, &task.NextTryAt, &task.FinishedAt, message) {
|
|
return
|
|
}
|
|
_ = s.repo.StrmUpload.Update(context.Background(), task)
|
|
}
|
|
|
|
// retryTask 失败状态机：重试次数不足则回 pending 并设置退避时间，否则 failed。
|
|
// 返回 false 表示无需再次落库（每次都会通过 Update 落库，因此恒返回 true）。
|
|
func retryTask(retryCount *int, status *string, errMsg *string, nextTryAt **time.Time, finishedAt **time.Time, message string) bool {
|
|
now := time.Now()
|
|
if *retryCount >= strmMaxTaskRetry {
|
|
*status = model.StrmTaskFailed
|
|
*errMsg = message
|
|
*finishedAt = &now
|
|
return true
|
|
}
|
|
*retryCount++
|
|
*status = model.StrmTaskPending
|
|
*errMsg = message
|
|
next := now.Add(time.Duration(*retryCount) * 30 * time.Second)
|
|
*nextTryAt = &next
|
|
*finishedAt = nil
|
|
return true
|
|
}
|
|
|
|
// downloadToFile 把直链内容下载到目标文件（临时文件 + 原子改名）。
|
|
func downloadToFile(ctx context.Context, link *cloud.DirectLink, target string, client *http.Client) error {
|
|
target = sanitizeLocalPath(target)
|
|
if link == nil || link.URL == "" {
|
|
return errors.New("空下载地址")
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, link.URL, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for k, v := range link.Headers {
|
|
req.Header.Set(k, v)
|
|
}
|
|
// 115 的 CDN 直链要求与换取链接时相同的浏览器 UA，否则返回 403；
|
|
// 其他网盘（WebDAV/OpenList）对该值不敏感，统一兜底设置。
|
|
if req.Header.Get("User-Agent") == "" {
|
|
req.Header.Set("User-Agent", cloud115.DefaultUA)
|
|
}
|
|
if client == nil {
|
|
client = http.DefaultClient
|
|
}
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
|
return fmt.Errorf("http %d", resp.StatusCode)
|
|
}
|
|
if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
|
|
return err
|
|
}
|
|
tmp := target + ".tmp"
|
|
out, err := os.Create(tmp)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, copyErr := io.Copy(out, resp.Body)
|
|
closeErr := out.Close()
|
|
if copyErr != nil {
|
|
_ = os.Remove(tmp)
|
|
return copyErr
|
|
}
|
|
if closeErr != nil {
|
|
_ = os.Remove(tmp)
|
|
return closeErr
|
|
}
|
|
return os.Rename(tmp, target)
|
|
}
|
|
|
|
// queueCleanupLoop 定期清理 7 天前的完成/失败/取消任务。
|
|
func (s *StrmService) queueCleanupLoop(ctx context.Context) {
|
|
ticker := time.NewTicker(6 * time.Hour)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-s.stopCh:
|
|
return
|
|
case <-ticker.C:
|
|
before := time.Now().AddDate(0, 0, -7)
|
|
if err := s.repo.StrmDownload.DeleteFinishedOlderThan(ctx, before); err != nil {
|
|
s.log.Warn("clean old strm download tasks failed", zap.Error(err))
|
|
}
|
|
if err := s.repo.StrmUpload.DeleteFinishedOlderThan(ctx, before); err != nil {
|
|
s.log.Warn("clean old strm upload tasks failed", zap.Error(err))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// ─── 队列查询与操作（handler 使用） ─────────────────────────────────────────────
|
|
|
|
// StrmQueueCounts 队列统计。
|
|
type StrmQueueCounts struct {
|
|
Pending int64 `json:"pending"`
|
|
Running int64 `json:"running"`
|
|
Done int64 `json:"done"`
|
|
Failed int64 `json:"failed"`
|
|
Canceled int64 `json:"canceled"`
|
|
}
|
|
|
|
// StrmQueueSnapshot 队列快照（统计 + 任务明细，分页）。
|
|
type StrmQueueSnapshot struct {
|
|
Counts StrmQueueCounts `json:"counts"`
|
|
Tasks []strmTaskView `json:"tasks"`
|
|
Total int64 `json:"total"` // 当前过滤条件下任务总数
|
|
Page int `json:"page"` // 当前页码（从 1 开始）
|
|
PageSize int `json:"page_size"` // 单页大小
|
|
}
|
|
|
|
// strmTaskView 队列任务统一视图（下载/上传共用）。
|
|
type strmTaskView struct {
|
|
ID string `json:"id"`
|
|
Kind string `json:"kind"` // download / upload
|
|
SyncPathID string `json:"sync_path_id"`
|
|
AccountID string `json:"account_id"`
|
|
Provider string `json:"provider"`
|
|
FileName string `json:"file_name"`
|
|
LocalPath string `json:"local_path"`
|
|
RemotePath string `json:"remote_path"`
|
|
Size int64 `json:"size"`
|
|
Status string `json:"status"`
|
|
Error string `json:"error"`
|
|
RetryCount int `json:"retry_count"`
|
|
CreatedAt string `json:"created_at"`
|
|
StartedAt *string `json:"started_at"`
|
|
FinishedAt *string `json:"finished_at"`
|
|
}
|
|
|
|
// DownloadQueueSnapshot 下载队列快照（分页）。
|
|
func (s *StrmService) DownloadQueueSnapshot(ctx context.Context, status string, page, pageSize int) (*StrmQueueSnapshot, error) {
|
|
page, pageSize = normalizeStrmTaskPage(page, pageSize)
|
|
tasks, total, err := s.repo.StrmDownload.List(ctx, status, page, pageSize)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
counts, err := s.repo.StrmDownload.CountByStatus(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
snap := &StrmQueueSnapshot{Counts: strmTaskCountsFrom(counts), Tasks: make([]strmTaskView, 0, len(tasks)), Total: total, Page: page, PageSize: pageSize}
|
|
for i := range tasks {
|
|
t := &tasks[i]
|
|
snap.Tasks = append(snap.Tasks, strmTaskView{
|
|
ID: t.ID,
|
|
Kind: "download",
|
|
SyncPathID: t.SyncPathID,
|
|
AccountID: t.AccountID,
|
|
Provider: t.Provider,
|
|
FileName: t.FileName,
|
|
LocalPath: t.LocalPath,
|
|
RemotePath: t.RemoteDir + "/" + t.FileName,
|
|
Size: t.Size,
|
|
Status: t.Status,
|
|
Error: t.Error,
|
|
RetryCount: t.RetryCount,
|
|
CreatedAt: t.CreatedAt.Local().Format(time.RFC3339),
|
|
StartedAt: timePtrString(t.StartedAt),
|
|
FinishedAt: timePtrString(t.FinishedAt),
|
|
})
|
|
}
|
|
return snap, nil
|
|
}
|
|
|
|
// UploadQueueSnapshot 上传队列快照（分页）。
|
|
func (s *StrmService) UploadQueueSnapshot(ctx context.Context, status string, page, pageSize int) (*StrmQueueSnapshot, error) {
|
|
page, pageSize = normalizeStrmTaskPage(page, pageSize)
|
|
tasks, total, err := s.repo.StrmUpload.List(ctx, status, page, pageSize)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
counts, err := s.repo.StrmUpload.CountByStatus(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
snap := &StrmQueueSnapshot{Counts: strmTaskCountsFrom(counts), Tasks: make([]strmTaskView, 0, len(tasks)), Total: total, Page: page, PageSize: pageSize}
|
|
for i := range tasks {
|
|
t := &tasks[i]
|
|
snap.Tasks = append(snap.Tasks, strmTaskView{
|
|
ID: t.ID,
|
|
Kind: "upload",
|
|
SyncPathID: t.SyncPathID,
|
|
AccountID: t.AccountID,
|
|
Provider: t.Provider,
|
|
FileName: t.FileName,
|
|
LocalPath: t.LocalPath,
|
|
RemotePath: t.RemotePath,
|
|
Size: t.Size,
|
|
Status: t.Status,
|
|
Error: t.Error,
|
|
RetryCount: t.RetryCount,
|
|
CreatedAt: t.CreatedAt.Local().Format(time.RFC3339),
|
|
StartedAt: timePtrString(t.StartedAt),
|
|
FinishedAt: timePtrString(t.FinishedAt),
|
|
})
|
|
}
|
|
return snap, nil
|
|
}
|
|
|
|
func strmTaskCountsFrom(m map[string]int64) StrmQueueCounts {
|
|
return StrmQueueCounts{
|
|
Pending: m[model.StrmTaskPending],
|
|
Running: m[model.StrmTaskRunning],
|
|
Done: m[model.StrmTaskDone],
|
|
Failed: m[model.StrmTaskFailed],
|
|
Canceled: m[model.StrmTaskCanceled],
|
|
}
|
|
}
|
|
|
|
// normalizeStrmTaskPage 钳制队列分页参数（与 repository 一致，保证回显正确）。
|
|
func normalizeStrmTaskPage(page, pageSize int) (int, int) {
|
|
if page < 1 {
|
|
page = 1
|
|
}
|
|
if pageSize <= 0 || pageSize > 200 {
|
|
pageSize = 50
|
|
}
|
|
return page, pageSize
|
|
}
|
|
|
|
func timePtrString(t *time.Time) *string {
|
|
if t == nil {
|
|
return nil
|
|
}
|
|
v := t.Local().Format(time.RFC3339)
|
|
return &v
|
|
}
|
|
|
|
// CancelDownloadTask 取消一个排队/进行中的下载任务。
|
|
func (s *StrmService) CancelDownloadTask(ctx context.Context, id string) error {
|
|
task, err := s.repo.StrmDownload.FindByID(ctx, id)
|
|
if err != nil || task == nil {
|
|
return errNotFoundOr(err, "下载任务不存在")
|
|
}
|
|
if task.Status != model.StrmTaskPending && task.Status != model.StrmTaskRunning {
|
|
return errors.New("任务已结束，无法取消")
|
|
}
|
|
now := time.Now()
|
|
task.Status = model.StrmTaskCanceled
|
|
task.Error = "已取消"
|
|
task.FinishedAt = &now
|
|
return s.repo.StrmDownload.Update(ctx, task)
|
|
}
|
|
|
|
// RetryDownloadTask 重试一个失败的下载任务。
|
|
func (s *StrmService) RetryDownloadTask(ctx context.Context, id string) error {
|
|
task, err := s.repo.StrmDownload.FindByID(ctx, id)
|
|
if err != nil || task == nil {
|
|
return errNotFoundOr(err, "下载任务不存在")
|
|
}
|
|
if task.Status != model.StrmTaskFailed && task.Status != model.StrmTaskCanceled {
|
|
return errors.New("只有失败/已取消的任务可以重试")
|
|
}
|
|
task.Status = model.StrmTaskPending
|
|
task.Error = ""
|
|
task.RetryCount = 0
|
|
task.NextTryAt = nil
|
|
task.FinishedAt = nil
|
|
return s.repo.StrmDownload.Update(ctx, task)
|
|
}
|
|
|
|
// CancelUploadTask 取消一个排队/进行中的上传任务。
|
|
func (s *StrmService) CancelUploadTask(ctx context.Context, id string) error {
|
|
task, err := s.repo.StrmUpload.FindByID(ctx, id)
|
|
if err != nil || task == nil {
|
|
return errNotFoundOr(err, "上传任务不存在")
|
|
}
|
|
if task.Status != model.StrmTaskPending && task.Status != model.StrmTaskRunning {
|
|
return errors.New("任务已结束，无法取消")
|
|
}
|
|
now := time.Now()
|
|
task.Status = model.StrmTaskCanceled
|
|
task.Error = "已取消"
|
|
task.FinishedAt = &now
|
|
return s.repo.StrmUpload.Update(ctx, task)
|
|
}
|
|
|
|
// RetryUploadTask 重试一个失败的上传任务。
|
|
func (s *StrmService) RetryUploadTask(ctx context.Context, id string) error {
|
|
task, err := s.repo.StrmUpload.FindByID(ctx, id)
|
|
if err != nil || task == nil {
|
|
return errNotFoundOr(err, "上传任务不存在")
|
|
}
|
|
if task.Status != model.StrmTaskFailed && task.Status != model.StrmTaskCanceled {
|
|
return errors.New("只有失败/已取消的任务可以重试")
|
|
}
|
|
task.Status = model.StrmTaskPending
|
|
task.Error = ""
|
|
task.RetryCount = 0
|
|
task.NextTryAt = nil
|
|
task.FinishedAt = nil
|
|
return s.repo.StrmUpload.Update(ctx, task)
|
|
}
|
|
|
|
// ─── 下载队列批量操作（handler 使用） ─────────────────────────────────────────
|
|
|
|
// ClearDoneDownloadTasks 清空全部已完成下载记录，返回删除数量。
|
|
func (s *StrmService) ClearDoneDownloadTasks(ctx context.Context) (int64, error) {
|
|
return s.repo.StrmDownload.ClearDone(ctx)
|
|
}
|
|
|
|
// ClearFinishedDownloadTasks 清空全部已完成与失败的下载记录，返回删除数量。
|
|
func (s *StrmService) ClearFinishedDownloadTasks(ctx context.Context) (int64, error) {
|
|
return s.repo.StrmDownload.ClearFinished(ctx)
|
|
}
|
|
|
|
// ClearCanceledDownloadTasks 清空全部已取消的下载记录，返回删除数量。
|
|
func (s *StrmService) ClearCanceledDownloadTasks(ctx context.Context) (int64, error) {
|
|
return s.repo.StrmDownload.ClearCanceled(ctx)
|
|
}
|
|
|
|
// ClearCanceledUploadTasks 清空全部已取消的上传记录，返回删除数量。
|
|
func (s *StrmService) ClearCanceledUploadTasks(ctx context.Context) (int64, error) {
|
|
return s.repo.StrmUpload.ClearCanceled(ctx)
|
|
}
|
|
|
|
// RetryAllFailedDownloadTasks 批量重试所有失败下载任务，返回重新入队数量。
|
|
func (s *StrmService) RetryAllFailedDownloadTasks(ctx context.Context) (int64, error) {
|
|
return s.repo.StrmDownload.RetryAllFailed(ctx)
|
|
}
|
|
|
|
// CancelPendingDownloadTasks 批量取消所有排队下载任务，返回取消数量。
|
|
func (s *StrmService) CancelPendingDownloadTasks(ctx context.Context) (int64, error) {
|
|
return s.repo.StrmDownload.CancelPending(ctx)
|
|
}
|
|
|
|
// CancelPendingUploadTasks 批量取消所有排队上传任务，返回取消数量。
|
|
func (s *StrmService) CancelPendingUploadTasks(ctx context.Context) (int64, error) {
|
|
return s.repo.StrmUpload.CancelPending(ctx)
|
|
}
|
|
|
|
func sleepContext(ctx context.Context, d time.Duration) {
|
|
select {
|
|
case <-ctx.Done():
|
|
case <-time.After(d):
|
|
}
|
|
}
|
|
|
|
// ─── 115 风控/限流熔断 ────────────────────────────────────────────────────────
|
|
|
|
// triggerWAFCooldown 检测到 115 风控/限流后触发全局冷却，冷却期间下载 worker 暂停。
|
|
// 冷却时间取最大值，避免连续触发时缩短等待。
|
|
func (s *StrmService) triggerWAFCooldown() {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
until := time.Now().Add(strmWAFCooldown)
|
|
if until.After(s.wafUntil) {
|
|
s.wafUntil = until
|
|
s.log.Warn("115 风控/限流，下载队列进入冷却", zap.Duration("cooldown", strmWAFCooldown))
|
|
}
|
|
}
|
|
|
|
// wafCooldownLeft 返回剩余冷却时间（0 表示无需冷却）。
|
|
func (s *StrmService) wafCooldownLeft() time.Duration {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.wafUntil.After(time.Now()) {
|
|
return time.Until(s.wafUntil)
|
|
}
|
|
return 0
|
|
}
|
|
|
|
// is115Blocked 判断错误是否来自 115 的风控/限流（WAF 405 拦截页或限流错误码）。
|
|
func is115Blocked(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
msg := strings.ToLower(err.Error())
|
|
return strings.Contains(msg, "115 接口返回 http 405") ||
|
|
strings.Contains(msg, "访问被阻断") ||
|
|
strings.Contains(msg, "request has been blocked") ||
|
|
strings.Contains(msg, "115 接口错误（770004") ||
|
|
strings.Contains(msg, "115 接口错误（406")
|
|
}
|
|
|
|
// isHTTPDownloadFailure 判断下载是否因 HTTP 状态码失败（直链失效需清缓存重取）。
|
|
func isHTTPDownloadFailure(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
msg := strings.ToLower(err.Error())
|
|
return strings.Contains(msg, "http 4") || strings.Contains(msg, "http 5")
|
|
}
|