mirror of
https://github.com/truewhile/MeBox.git
synced 2026-09-28 11:16:37 +08:00
truewhile
694 lines
23 KiB
Go
694 lines
23 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 {
|
||
s.processUpload115(ctx, task)
|
||
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, "")
|
||
}
|
||
|
||
// processUpload115 115 元数据上传：task.RemotePath 存的是父目录 cid，FileName 为远端文件名。
|
||
func (s *StrmService) processUpload115(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))
|
||
}
|
||
}
|
||
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.uploadTaskFailWithRetry(task, err.Error())
|
||
return
|
||
}
|
||
named, ok := provider.(interface {
|
||
PutFileNamed(ctx context.Context, parentCID, fileName 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 := named.PutFileNamed(ctx, task.RemotePath, task.FileName, 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 使用） ─────────────────────────────────────────
|
||
|
||
// DeleteDownloadTask 删除一个下载任务记录。
|
||
func (s *StrmService) DeleteDownloadTask(ctx context.Context, id string) error {
|
||
return s.repo.StrmDownload.Delete(ctx, id)
|
||
}
|
||
|
||
// DeleteUploadTask 删除一个上传任务记录。
|
||
func (s *StrmService) DeleteUploadTask(ctx context.Context, id string) error {
|
||
return s.repo.StrmUpload.Delete(ctx, id)
|
||
}
|
||
|
||
// BatchActionDownloadTasks 对选中的下载任务执行批量操作（delete / retry / cancel）。
|
||
func (s *StrmService) BatchActionDownloadTasks(ctx context.Context, action string, ids []string) (int64, error) {
|
||
switch action {
|
||
case "delete":
|
||
return s.repo.StrmDownload.DeleteBatch(ctx, ids)
|
||
case "retry":
|
||
return s.repo.StrmDownload.RetryBatch(ctx, ids)
|
||
case "cancel":
|
||
return s.repo.StrmDownload.CancelBatch(ctx, ids)
|
||
default:
|
||
return 0, fmt.Errorf("不支持的批量操作: %s", action)
|
||
}
|
||
}
|
||
|
||
// BatchActionUploadTasks 对选中的上传任务执行批量操作（delete / retry / cancel）。
|
||
func (s *StrmService) BatchActionUploadTasks(ctx context.Context, action string, ids []string) (int64, error) {
|
||
switch action {
|
||
case "delete":
|
||
return s.repo.StrmUpload.DeleteBatch(ctx, ids)
|
||
case "retry":
|
||
return s.repo.StrmUpload.RetryBatch(ctx, ids)
|
||
case "cancel":
|
||
return s.repo.StrmUpload.CancelBatch(ctx, ids)
|
||
default:
|
||
return 0, fmt.Errorf("不支持的批量操作: %s", action)
|
||
}
|
||
}
|
||
|
||
// 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)
|
||
}
|
||
|
||
// ClearDoneUploadTasks 清空全部已完成上传记录，返回删除数量。
|
||
func (s *StrmService) ClearDoneUploadTasks(ctx context.Context) (int64, error) {
|
||
return s.repo.StrmUpload.ClearDone(ctx)
|
||
}
|
||
|
||
// ClearFinishedUploadTasks 清空全部已完成与失败的上传记录，返回删除数量。
|
||
func (s *StrmService) ClearFinishedUploadTasks(ctx context.Context) (int64, error) {
|
||
return s.repo.StrmUpload.ClearFinished(ctx)
|
||
}
|
||
|
||
// RetryAllFailedDownloadTasks 批量重试所有失败下载任务，返回重新入队数量。
|
||
func (s *StrmService) RetryAllFailedDownloadTasks(ctx context.Context) (int64, error) {
|
||
return s.repo.StrmDownload.RetryAllFailed(ctx)
|
||
}
|
||
|
||
// RetryAllFailedUploadTasks 批量重试所有失败上传任务，返回重新入队数量。
|
||
func (s *StrmService) RetryAllFailedUploadTasks(ctx context.Context) (int64, error) {
|
||
return s.repo.StrmUpload.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")
|
||
}
|