Compare commits

...

19 Commits

Author SHA1 Message Date
github-actions[bot] 0e7dbd6215 chore: bump version to 0.0.29 [skip ci] 2026-08-26 00:43:54 +00:00
truewhile 7fd8de91cb Merge branch 'main' of https://github.com/truewhile/MMTL 2026-08-26 08:43:40 +08:00
truewhile 5f323eb2ce 优化
yo 优化
2026-08-26 08:43:36 +08:00
github-actions[bot] c0ac8bf11a chore: bump version to 0.0.28 [skip ci] 2026-08-25 16:26:45 +00:00
truewhile b676733af7 优化strm同步
优化strm同步
2026-08-26 00:26:28 +08:00
github-actions[bot] 7a2027a3a7 chore: bump version to 0.0.27 [skip ci] 2026-08-25 15:19:28 +00:00
truewhile 9ffb74adce 优化
优化
2026-08-25 23:19:07 +08:00
github-actions[bot] 13faff7078 chore: bump version to 0.0.26 [skip ci] 2026-08-25 14:55:08 +00:00
truewhile a8a4e88d86 Merge branch 'main' of https://github.com/truewhile/MMTL 2026-08-25 22:54:50 +08:00
truewhile 8fa5db88ff 优化
优化
2026-08-25 22:54:46 +08:00
github-actions[bot] 3c325f81c8 chore: bump version to 0.0.25 [skip ci] 2026-08-25 14:12:30 +00:00
truewhile 585434010c 优化
优化
2026-08-25 22:12:05 +08:00
github-actions[bot] eb1a705cae chore: bump version to 0.0.24 [skip ci] 2026-08-25 11:41:17 +00:00
truewhile 10770b2b77 b u g
b u g
2026-08-25 19:41:02 +08:00
github-actions[bot] b61c51e064 chore: bump version to 0.0.23 [skip ci] 2026-08-25 11:19:05 +00:00
truewhile 019ecbec7b 优化
优化
2026-08-25 19:18:49 +08:00
github-actions[bot] 6a96c5640e chore: bump version to 0.0.22 [skip ci] 2026-08-25 11:06:23 +00:00
truewhile 4025e92cb4 Merge branch 'main' of https://github.com/truewhile/MMTL 2026-08-25 19:06:02 +08:00
truewhile 2355419ef9 优化
优化
2026-08-25 19:05:59 +08:00
32 changed files with 1503 additions and 403 deletions
+1 -1
View File
@@ -1 +1 @@
0.0.21
0.0.29
+2 -2
View File
@@ -32,7 +32,7 @@ func listLibrariesHandler(svc *service.Container) gin.HandlerFunc {
return
}
role, _ := c.Get(middleware.CtxUserRole)
includeHidden := role == "admin" && (c.Query("include_hidden") == "1" || c.Query("all") == "1")
includeHidden := role == "admin" && (c.Query("include_hidden") == "1" || c.Query("include_hidden") == "true" || c.Query("all") == "1")
if !includeHidden {
libs = service.FilterDisplayCloudLibraries(c.Request.Context(), svc.Repo, libs)
visibility := mediaVisibilityForRequest(c, svc)
@@ -60,7 +60,7 @@ func getLibraryHandler(svc *service.Container) gin.HandlerFunc {
return
}
role, _ := c.Get(middleware.CtxUserRole)
includeHidden := role == "admin" && (c.Query("include_hidden") == "1" || c.Query("all") == "1")
includeHidden := role == "admin" && (c.Query("include_hidden") == "1" || c.Query("include_hidden") == "true" || c.Query("all") == "1")
if !includeHidden {
libs := service.FilterDisplayCloudLibraries(c.Request.Context(), svc.Repo, []model.Library{*lib})
if len(libs) == 0 || !service.LibraryVisibleForUser(c.Request.Context(), svc.Repo, libs[0], mediaVisibilityForRequest(c, svc)) {
+12 -1
View File
@@ -178,6 +178,7 @@ type strmSyncPathReq struct {
DeleteDir *bool `json:"delete_dir"`
Cron string `json:"cron"`
EnableCron *bool `json:"enable_cron"`
SyncMode string `json:"sync_mode"`
Enabled *bool `json:"enabled"`
}
@@ -261,7 +262,16 @@ func deleteStrmSyncPathHandler(svc *service.Container) gin.HandlerFunc {
func startStrmSyncHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
if err := svc.Strm.StartSync(c.Request.Context(), c.Param("id")); err != nil {
mode := c.Query("mode")
if mode == "" {
var body struct {
Mode string `json:"mode"`
}
if err := c.ShouldBindJSON(&body); err == nil && body.Mode != "" {
mode = body.Mode
}
}
if err := svc.Strm.StartSync(c.Request.Context(), c.Param("id"), mode); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
@@ -458,6 +468,7 @@ func strmSyncPathFromReq(req strmSyncPathReq) *model.StrmSyncPath {
DeleteDir: boolValue(req.DeleteDir, false),
Cron: strings.TrimSpace(req.Cron),
EnableCron: boolValue(req.EnableCron, false),
SyncMode: strings.TrimSpace(req.SyncMode),
Enabled: boolValue(req.Enabled, true),
}
}
+4 -3
View File
@@ -54,7 +54,8 @@ func AllModels() []interface{} {
&StrmAccount{},
&StrmSyncPath{},
&StrmSyncRecord{},
&StrmDownloadTask{},
&StrmUploadTask{},
}
&StrmDownloadTask{},
&StrmUploadTask{},
&StrmDirCache{},
}
}
+17
View File
@@ -48,12 +48,19 @@ type StrmSyncPath struct {
DeleteDir bool `json:"delete_dir"` // 清理多余文件时删除空目录
Cron string `gorm:"size:128" json:"cron"` // 5 段 cron 表达式(可选)
EnableCron bool `json:"enable_cron"` // 是否按 Cron 定时同步
SyncMode string `gorm:"size:32;default:'incremental'" json:"sync_mode"` // 默认同步模式:incremental / full
Enabled bool `gorm:"default:true" json:"enabled"`
LastSyncAt *time.Time `json:"last_sync_at"`
LastSyncStatus string `gorm:"size:16" json:"last_sync_status"` // idle/running/ok/error/canceled
LastSyncMessage string `gorm:"size:1024" json:"last_sync_message"`
}
// STRM 同步类型。
const (
StrmSyncTypeIncremental = "incremental"
StrmSyncTypeFull = "full"
)
// StrmSyncRecord 是一次同步执行的记录。
const (
StrmSyncRecordPending = "pending"
@@ -66,6 +73,7 @@ const (
type StrmSyncRecord struct {
Base
SyncPathID string `gorm:"size:36;index" json:"sync_path_id"`
SyncType string `gorm:"size:32;default:'incremental'" json:"sync_type"` // incremental / full
Status string `gorm:"size:16;index" json:"status"`
Total int64 `json:"total"` // 远端发现的文件总数
NewStrm int64 `json:"new_strm"` // 本次新建/更新的 strm 数
@@ -123,3 +131,12 @@ type StrmUploadTask struct {
StartedAt *time.Time `json:"started_at"`
FinishedAt *time.Time `json:"finished_at"`
}
// StrmDirCache 缓存远端网盘目录 ID 与相对路径映射(支持 115 增量同步秒级寻址)。
type StrmDirCache struct {
Base
SyncPathID string `gorm:"size:36;index:idx_strm_dir_cache,priority:1" json:"sync_path_id"`
DirID string `gorm:"size:128;index:idx_strm_dir_cache,priority:2" json:"dir_id"`
Path string `gorm:"size:1024" json:"path"` // 相对根目录的路径
}
+2
View File
@@ -31,6 +31,7 @@ type Container struct {
StrmSyncRecord *StrmSyncRecordRepository
StrmDownload *StrmDownloadTaskRepository
StrmUpload *StrmUploadTaskRepository
StrmDirCache *StrmDirCacheRepository
}
// New 将每个 repository 连接到单个 *gorm.DB。
@@ -58,5 +59,6 @@ func New(db *gorm.DB) *Container {
StrmSyncRecord: &StrmSyncRecordRepository{db: db},
StrmDownload: &StrmDownloadTaskRepository{db: db},
StrmUpload: &StrmUploadTaskRepository{db: db},
StrmDirCache: &StrmDirCacheRepository{db: db},
}
}
+304 -140
View File
@@ -3,6 +3,7 @@ package repository
import (
"context"
"errors"
"sync"
"time"
"gorm.io/gorm"
@@ -10,13 +11,17 @@ import (
"github.com/ShukeBta/MMTL/internal/model"
)
var strmClaimMu sync.Mutex
// ─── StrmAccount ───────────────────────────────────────────────────────────────
// StrmAccountRepository persists model.StrmAccount.
type StrmAccountRepository struct{ db *gorm.DB }
func (r *StrmAccountRepository) Create(ctx context.Context, a *model.StrmAccount) error {
return r.db.WithContext(ctx).Create(a).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Create(a).Error
})
}
func (r *StrmAccountRepository) FindByID(ctx context.Context, id string) (*model.StrmAccount, error) {
@@ -38,20 +43,24 @@ func (r *StrmAccountRepository) List(ctx context.Context) ([]model.StrmAccount,
}
func (r *StrmAccountRepository) Update(ctx context.Context, a *model.StrmAccount) error {
return r.db.WithContext(ctx).Model(&model.StrmAccount{}).Where("id = ?", a.ID).Updates(map[string]any{
"name": a.Name,
"provider": a.Provider,
"config": a.Config,
"enabled": a.Enabled,
"last_test_at": a.LastTestAt,
"last_test_result": a.LastTestResult,
"last_test_ok": a.LastTestOK,
"updated_at": time.Now(),
}).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Model(&model.StrmAccount{}).Where("id = ?", a.ID).Updates(map[string]any{
"name": a.Name,
"provider": a.Provider,
"config": a.Config,
"enabled": a.Enabled,
"last_test_at": a.LastTestAt,
"last_test_result": a.LastTestResult,
"last_test_ok": a.LastTestOK,
"updated_at": time.Now(),
}).Error
})
}
func (r *StrmAccountRepository) Delete(ctx context.Context, id string) error {
return r.db.WithContext(ctx).Where("id = ?", id).Delete(&model.StrmAccount{}).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Where("id = ?", id).Delete(&model.StrmAccount{}).Error
})
}
// ─── StrmSyncPath ──────────────────────────────────────────────────────────────
@@ -60,7 +69,9 @@ func (r *StrmAccountRepository) Delete(ctx context.Context, id string) error {
type StrmSyncPathRepository struct{ db *gorm.DB }
func (r *StrmSyncPathRepository) Create(ctx context.Context, p *model.StrmSyncPath) error {
return r.db.WithContext(ctx).Create(p).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Create(p).Error
})
}
func (r *StrmSyncPathRepository) FindByID(ctx context.Context, id string) (*model.StrmSyncPath, error) {
@@ -82,33 +93,38 @@ func (r *StrmSyncPathRepository) List(ctx context.Context) ([]model.StrmSyncPath
}
func (r *StrmSyncPathRepository) Update(ctx context.Context, p *model.StrmSyncPath) error {
return r.db.WithContext(ctx).Model(&model.StrmSyncPath{}).Where("id = ?", p.ID).Updates(map[string]any{
"name": p.Name,
"account_id": p.AccountID,
"provider": p.Provider,
"remote_path": p.RemotePath,
"local_path": p.LocalPath,
"strm_base_url": p.StrmBaseURL,
"video_ext": p.VideoExt,
"meta_ext": p.MetaExt,
"exclude_name": p.ExcludeName,
"min_video_size_mb": p.MinVideoSizeMB,
"add_path": p.AddPath,
"download_meta": p.DownloadMeta,
"upload_meta": p.UploadMeta,
"delete_dir": p.DeleteDir,
"cron": p.Cron,
"enable_cron": p.EnableCron,
"enabled": p.Enabled,
"last_sync_at": p.LastSyncAt,
"last_sync_status": p.LastSyncStatus,
"last_sync_message": p.LastSyncMessage,
"updated_at": time.Now(),
}).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Model(&model.StrmSyncPath{}).Where("id = ?", p.ID).Updates(map[string]any{
"name": p.Name,
"account_id": p.AccountID,
"provider": p.Provider,
"remote_path": p.RemotePath,
"local_path": p.LocalPath,
"strm_base_url": p.StrmBaseURL,
"video_ext": p.VideoExt,
"meta_ext": p.MetaExt,
"exclude_name": p.ExcludeName,
"min_video_size_mb": p.MinVideoSizeMB,
"add_path": p.AddPath,
"download_meta": p.DownloadMeta,
"upload_meta": p.UploadMeta,
"delete_dir": p.DeleteDir,
"cron": p.Cron,
"enable_cron": p.EnableCron,
"sync_mode": p.SyncMode,
"enabled": p.Enabled,
"last_sync_at": p.LastSyncAt,
"last_sync_status": p.LastSyncStatus,
"last_sync_message": p.LastSyncMessage,
"updated_at": time.Now(),
}).Error
})
}
func (r *StrmSyncPathRepository) Delete(ctx context.Context, id string) error {
return r.db.WithContext(ctx).Where("id = ?", id).Delete(&model.StrmSyncPath{}).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Where("id = ?", id).Delete(&model.StrmSyncPath{}).Error
})
}
// ─── StrmSyncRecord ────────────────────────────────────────────────────────────
@@ -117,23 +133,28 @@ func (r *StrmSyncPathRepository) Delete(ctx context.Context, id string) error {
type StrmSyncRecordRepository struct{ db *gorm.DB }
func (r *StrmSyncRecordRepository) Create(ctx context.Context, rec *model.StrmSyncRecord) error {
return r.db.WithContext(ctx).Create(rec).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Create(rec).Error
})
}
func (r *StrmSyncRecordRepository) Update(ctx context.Context, rec *model.StrmSyncRecord) error {
return r.db.WithContext(ctx).Model(&model.StrmSyncRecord{}).Where("id = ?", rec.ID).Updates(map[string]any{
"status": rec.Status,
"total": rec.Total,
"new_strm": rec.NewStrm,
"new_meta": rec.NewMeta,
"uploaded": rec.Uploaded,
"pruned": rec.Pruned,
"skipped": rec.Skipped,
"message": rec.Message,
"started_at": rec.StartedAt,
"finished_at": rec.FinishedAt,
"updated_at": time.Now(),
}).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Model(&model.StrmSyncRecord{}).Where("id = ?", rec.ID).Updates(map[string]any{
"sync_type": rec.SyncType,
"status": rec.Status,
"total": rec.Total,
"new_strm": rec.NewStrm,
"new_meta": rec.NewMeta,
"uploaded": rec.Uploaded,
"pruned": rec.Pruned,
"skipped": rec.Skipped,
"message": rec.Message,
"started_at": rec.StartedAt,
"finished_at": rec.FinishedAt,
"updated_at": time.Now(),
}).Error
})
}
func (r *StrmSyncRecordRepository) List(ctx context.Context, syncPathID string, limit int) ([]model.StrmSyncRecord, error) {
@@ -155,7 +176,21 @@ func (r *StrmSyncRecordRepository) List(ctx context.Context, syncPathID string,
type StrmDownloadTaskRepository struct{ db *gorm.DB }
func (r *StrmDownloadTaskRepository) Create(ctx context.Context, t *model.StrmDownloadTask) error {
return r.db.WithContext(ctx).Create(t).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Create(t).Error
})
}
func (r *StrmDownloadTaskRepository) CreateInBatches(ctx context.Context, tasks []*model.StrmDownloadTask, batchSize int) error {
if len(tasks) == 0 {
return nil
}
if batchSize <= 0 {
batchSize = 100
}
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).CreateInBatches(tasks, batchSize).Error
})
}
func (r *StrmDownloadTaskRepository) FindByID(ctx context.Context, id string) (*model.StrmDownloadTask, error) {
@@ -205,24 +240,29 @@ func (r *StrmDownloadTaskRepository) CountByStatus(ctx context.Context) (map[str
// ClaimPendingDownload picks the oldest pending task and marks it running.
// Returns (nil, nil) when the queue is empty.
func (r *StrmDownloadTaskRepository) ClaimPendingDownload(ctx context.Context, limit int) ([]model.StrmDownloadTask, error) {
strmClaimMu.Lock()
defer strmClaimMu.Unlock()
var rows []model.StrmDownloadTask
err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Where("status = ? AND (next_try_at IS NULL OR next_try_at <= ?)", model.StrmTaskPending, time.Now()).
Order("created_at asc").Limit(limit).Find(&rows).Error; err != nil {
return err
}
if len(rows) == 0 {
return nil
}
ids := make([]string, 0, len(rows))
now := time.Now()
for i := range rows {
ids = append(ids, rows[i].ID)
rows[i].Status = model.StrmTaskRunning
rows[i].StartedAt = &now
}
return tx.Model(&model.StrmDownloadTask{}).Where("id IN ?", ids).
Updates(map[string]any{"status": model.StrmTaskRunning, "started_at": now}).Error
err := withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Where("status = ? AND (next_try_at IS NULL OR next_try_at <= ?)", model.StrmTaskPending, time.Now()).
Order("created_at asc").Limit(limit).Find(&rows).Error; err != nil {
return err
}
if len(rows) == 0 {
return nil
}
ids := make([]string, 0, len(rows))
now := time.Now()
for i := range rows {
ids = append(ids, rows[i].ID)
rows[i].Status = model.StrmTaskRunning
rows[i].StartedAt = &now
}
return tx.Model(&model.StrmDownloadTask{}).Where("id IN ?", ids).
Updates(map[string]any{"status": model.StrmTaskRunning, "started_at": now}).Error
})
})
if err != nil {
return nil, err
@@ -231,62 +271,86 @@ func (r *StrmDownloadTaskRepository) ClaimPendingDownload(ctx context.Context, l
}
func (r *StrmDownloadTaskRepository) Update(ctx context.Context, t *model.StrmDownloadTask) error {
return r.db.WithContext(ctx).Model(&model.StrmDownloadTask{}).Where("id = ?", t.ID).Updates(map[string]any{
"status": t.Status,
"error": t.Error,
"retry_count": t.RetryCount,
"next_try_at": t.NextTryAt,
"started_at": t.StartedAt,
"finished_at": t.FinishedAt,
"updated_at": time.Now(),
}).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Model(&model.StrmDownloadTask{}).Where("id = ?", t.ID).Updates(map[string]any{
"status": t.Status,
"error": t.Error,
"retry_count": t.RetryCount,
"next_try_at": t.NextTryAt,
"started_at": t.StartedAt,
"finished_at": t.FinishedAt,
"updated_at": time.Now(),
}).Error
})
}
func (r *StrmDownloadTaskRepository) Delete(ctx context.Context, id string) error {
return r.db.WithContext(ctx).Where("id = ?", id).Delete(&model.StrmDownloadTask{}).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Where("id = ?", id).Delete(&model.StrmDownloadTask{}).Error
})
}
// ClearDone 清空全部已完成下载任务。
func (r *StrmDownloadTaskRepository) ClearDone(ctx context.Context) (int64, error) {
res := r.db.WithContext(ctx).Where("status = ?", model.StrmTaskDone).Delete(&model.StrmDownloadTask{})
return res.RowsAffected, res.Error
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Where("status = ?", model.StrmTaskDone).Delete(&model.StrmDownloadTask{})
count = res.RowsAffected
return res.Error
})
return count, err
}
// ClearFinished 清空全部已完成与失败下载任务。
func (r *StrmDownloadTaskRepository) ClearFinished(ctx context.Context) (int64, error) {
res := r.db.WithContext(ctx).Where("status IN ?", []string{model.StrmTaskDone, model.StrmTaskFailed}).
Delete(&model.StrmDownloadTask{})
return res.RowsAffected, res.Error
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Where("status IN ?", []string{model.StrmTaskDone, model.StrmTaskFailed}).
Delete(&model.StrmDownloadTask{})
count = res.RowsAffected
return res.Error
})
return count, err
}
// RetryAllFailed 把所有失败任务重置回待处理,清空错误与重试计数。
func (r *StrmDownloadTaskRepository) RetryAllFailed(ctx context.Context) (int64, error) {
res := r.db.WithContext(ctx).Model(&model.StrmDownloadTask{}).
Where("status = ?", model.StrmTaskFailed).
Updates(map[string]any{
"status": model.StrmTaskPending,
"error": "",
"retry_count": 0,
"next_try_at": nil,
"started_at": nil,
"finished_at": nil,
"updated_at": time.Now(),
})
return res.RowsAffected, res.Error
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Model(&model.StrmDownloadTask{}).
Where("status = ?", model.StrmTaskFailed).
Updates(map[string]any{
"status": model.StrmTaskPending,
"error": "",
"retry_count": 0,
"next_try_at": nil,
"started_at": nil,
"finished_at": nil,
"updated_at": time.Now(),
})
count = res.RowsAffected
return res.Error
})
return count, err
}
// CancelPending 批量取消所有排队中的任务。
func (r *StrmDownloadTaskRepository) CancelPending(ctx context.Context) (int64, error) {
now := time.Now()
res := r.db.WithContext(ctx).Model(&model.StrmDownloadTask{}).
Where("status = ?", model.StrmTaskPending).
Updates(map[string]any{
"status": model.StrmTaskCanceled,
"error": "已批量取消",
"finished_at": now,
"updated_at": now,
})
return res.RowsAffected, res.Error
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Model(&model.StrmDownloadTask{}).
Where("status = ?", model.StrmTaskPending).
Updates(map[string]any{
"status": model.StrmTaskCanceled,
"error": "已批量取消",
"finished_at": now,
"updated_at": now,
})
count = res.RowsAffected
return res.Error
})
return count, err
}
// CountActive 统计某同步目录下目标仍在排队/进行的任务数(用于去重)。
@@ -299,10 +363,28 @@ func (r *StrmDownloadTaskRepository) CountActive(ctx context.Context, syncPathID
return count
}
// GetActiveLocalPathMap 一次性获取某同步目录下正在排队或执行中的 local_path 集合,供同步时 O(1) 内存去重。
func (r *StrmDownloadTaskRepository) GetActiveLocalPathMap(ctx context.Context, syncPathID string) (map[string]bool, error) {
var paths []string
err := r.db.WithContext(ctx).Model(&model.StrmDownloadTask{}).
Where("sync_path_id = ? AND status IN ?", syncPathID, []string{model.StrmTaskPending, model.StrmTaskRunning}).
Pluck("local_path", &paths).Error
if err != nil {
return nil, err
}
out := make(map[string]bool, len(paths))
for _, p := range paths {
out[p] = true
}
return out, nil
}
func (r *StrmDownloadTaskRepository) DeleteFinishedOlderThan(ctx context.Context, before time.Time) error {
return r.db.WithContext(ctx).Where("status IN ? AND finished_at < ?",
[]string{model.StrmTaskDone, model.StrmTaskFailed, model.StrmTaskCanceled}, before).
Delete(&model.StrmDownloadTask{}).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Where("status IN ? AND finished_at < ?",
[]string{model.StrmTaskDone, model.StrmTaskFailed, model.StrmTaskCanceled}, before).
Delete(&model.StrmDownloadTask{}).Error
})
}
// ─── StrmUploadTask ────────────────────────────────────────────────────────────
@@ -311,7 +393,21 @@ func (r *StrmDownloadTaskRepository) DeleteFinishedOlderThan(ctx context.Context
type StrmUploadTaskRepository struct{ db *gorm.DB }
func (r *StrmUploadTaskRepository) Create(ctx context.Context, t *model.StrmUploadTask) error {
return r.db.WithContext(ctx).Create(t).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Create(t).Error
})
}
func (r *StrmUploadTaskRepository) CreateInBatches(ctx context.Context, tasks []*model.StrmUploadTask, batchSize int) error {
if len(tasks) == 0 {
return nil
}
if batchSize <= 0 {
batchSize = 100
}
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).CreateInBatches(tasks, batchSize).Error
})
}
func (r *StrmUploadTaskRepository) FindByID(ctx context.Context, id string) (*model.StrmUploadTask, error) {
@@ -380,24 +476,29 @@ func (r *StrmUploadTaskRepository) CountByStatus(ctx context.Context) (map[strin
// ClaimPendingUpload picks the oldest pending task and marks it running.
// Returns (nil, nil) when the queue is empty.
func (r *StrmUploadTaskRepository) ClaimPendingUpload(ctx context.Context, limit int) ([]model.StrmUploadTask, error) {
strmClaimMu.Lock()
defer strmClaimMu.Unlock()
var rows []model.StrmUploadTask
err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Where("status = ? AND (next_try_at IS NULL OR next_try_at <= ?)", model.StrmTaskPending, time.Now()).
Order("created_at asc").Limit(limit).Find(&rows).Error; err != nil {
return err
}
if len(rows) == 0 {
return nil
}
ids := make([]string, 0, len(rows))
now := time.Now()
for i := range rows {
ids = append(ids, rows[i].ID)
rows[i].Status = model.StrmTaskRunning
rows[i].StartedAt = &now
}
return tx.Model(&model.StrmUploadTask{}).Where("id IN ?", ids).
Updates(map[string]any{"status": model.StrmTaskRunning, "started_at": now}).Error
err := withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Where("status = ? AND (next_try_at IS NULL OR next_try_at <= ?)", model.StrmTaskPending, time.Now()).
Order("created_at asc").Limit(limit).Find(&rows).Error; err != nil {
return err
}
if len(rows) == 0 {
return nil
}
ids := make([]string, 0, len(rows))
now := time.Now()
for i := range rows {
ids = append(ids, rows[i].ID)
rows[i].Status = model.StrmTaskRunning
rows[i].StartedAt = &now
}
return tx.Model(&model.StrmUploadTask{}).Where("id IN ?", ids).
Updates(map[string]any{"status": model.StrmTaskRunning, "started_at": now}).Error
})
})
if err != nil {
return nil, err
@@ -406,19 +507,23 @@ func (r *StrmUploadTaskRepository) ClaimPendingUpload(ctx context.Context, limit
}
func (r *StrmUploadTaskRepository) Update(ctx context.Context, t *model.StrmUploadTask) error {
return r.db.WithContext(ctx).Model(&model.StrmUploadTask{}).Where("id = ?", t.ID).Updates(map[string]any{
"status": t.Status,
"error": t.Error,
"retry_count": t.RetryCount,
"next_try_at": t.NextTryAt,
"started_at": t.StartedAt,
"finished_at": t.FinishedAt,
"updated_at": time.Now(),
}).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Model(&model.StrmUploadTask{}).Where("id = ?", t.ID).Updates(map[string]any{
"status": t.Status,
"error": t.Error,
"retry_count": t.RetryCount,
"next_try_at": t.NextTryAt,
"started_at": t.StartedAt,
"finished_at": t.FinishedAt,
"updated_at": time.Now(),
}).Error
})
}
func (r *StrmUploadTaskRepository) Delete(ctx context.Context, id string) error {
return r.db.WithContext(ctx).Where("id = ?", id).Delete(&model.StrmUploadTask{}).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Where("id = ?", id).Delete(&model.StrmUploadTask{}).Error
})
}
// CountActive 统计某同步目录下目标仍在排队/进行的任务数(用于去重)。
@@ -431,8 +536,67 @@ func (r *StrmUploadTaskRepository) CountActive(ctx context.Context, syncPathID,
return count
}
func (r *StrmUploadTaskRepository) DeleteFinishedOlderThan(ctx context.Context, before time.Time) error {
return r.db.WithContext(ctx).Where("status IN ? AND finished_at < ?",
[]string{model.StrmTaskDone, model.StrmTaskFailed, model.StrmTaskCanceled}, before).
Delete(&model.StrmUploadTask{}).Error
// GetActiveLocalPathMap 一次性获取某同步目录下正在排队或执行中的 local_path 集合,供同步时 O(1) 内存去重。
func (r *StrmUploadTaskRepository) GetActiveLocalPathMap(ctx context.Context, syncPathID string) (map[string]bool, error) {
var paths []string
err := r.db.WithContext(ctx).Model(&model.StrmUploadTask{}).
Where("sync_path_id = ? AND status IN ?", syncPathID, []string{model.StrmTaskPending, model.StrmTaskRunning}).
Pluck("local_path", &paths).Error
if err != nil {
return nil, err
}
out := make(map[string]bool, len(paths))
for _, p := range paths {
out[p] = true
}
return out, nil
}
func (r *StrmUploadTaskRepository) DeleteFinishedOlderThan(ctx context.Context, before time.Time) error {
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Where("status IN ? AND finished_at < ?",
[]string{model.StrmTaskDone, model.StrmTaskFailed, model.StrmTaskCanceled}, before).
Delete(&model.StrmUploadTask{}).Error
})
}
// ─── StrmDirCache ─────────────────────────────────────────────────────────────
// StrmDirCacheRepository persists model.StrmDirCache.
type StrmDirCacheRepository struct{ db *gorm.DB }
func (r *StrmDirCacheRepository) ListBySyncPathID(ctx context.Context, syncPathID string) ([]model.StrmDirCache, error) {
var rows []model.StrmDirCache
err := r.db.WithContext(ctx).Where("sync_path_id = ?", syncPathID).Find(&rows).Error
return rows, err
}
func (r *StrmDirCacheRepository) Set(ctx context.Context, syncPathID, dirID, path string) error {
return withSQLiteBusyRetry(ctx, func() error {
var row model.StrmDirCache
err := r.db.WithContext(ctx).Where("sync_path_id = ? AND dir_id = ?", syncPathID, dirID).First(&row).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
row = model.StrmDirCache{
SyncPathID: syncPathID,
DirID: dirID,
Path: path,
}
return r.db.WithContext(ctx).Create(&row).Error
}
if err != nil {
return err
}
return r.db.WithContext(ctx).Model(&model.StrmDirCache{}).Where("id = ?", row.ID).Updates(map[string]any{
"path": path,
"updated_at": time.Now(),
}).Error
})
}
func (r *StrmDirCacheRepository) DeleteBySyncPathID(ctx context.Context, syncPathID string) error {
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Where("sync_path_id = ?", syncPathID).Delete(&model.StrmDirCache{}).Error
})
}
+5 -4
View File
@@ -37,10 +37,11 @@ var ErrUnsupported = errors.New("unsupported cloud provider")
// FileEntry is one item in a cloud directory listing.
type FileEntry struct {
ID string `json:"id"` // provider-native file id
Name string `json:"name"`
IsDir bool `json:"is_dir"`
Size int64 `json:"size"`
ID string `json:"id"` // provider-native file id
Name string `json:"name"`
IsDir bool `json:"is_dir"`
Size int64 `json:"size"`
MTime int64 `json:"mtime,omitempty"`
// PickCode is 115-specific; other providers use ID directly.
PickCode string `json:"pick_code,omitempty"`
}
+16 -9
View File
@@ -22,6 +22,12 @@ import (
"github.com/ShukeBta/MMTL/internal/service/cloud115"
)
// OpenAPI115Provider 暴露 115 开放平台驱动接口。
type OpenAPI115Provider interface {
Provider
OpenClient() *cloud115.OpenClient
}
// openAPI115Provider 实现 Provider 接口:List 列目录、Resolve 用 pickcode
// 换下载直链(302 offload,无需代理)、Ping 探测根目录。
type openAPI115Provider struct {
@@ -55,15 +61,16 @@ func (p *openAPI115Provider) List(ctx context.Context, dirID string) ([]FileEntr
if err != nil {
return nil, err
}
for _, f := range files {
out = append(out, FileEntry{
ID: f.FileId,
Name: f.FileName,
IsDir: f.Category == cloud115.TypeDir,
Size: f.FileSize,
PickCode: f.PickCode,
})
}
for _, f := range files {
out = append(out, FileEntry{
ID: f.FileId,
Name: f.FileName,
IsDir: f.Category == cloud115.TypeDir,
Size: f.FileSize,
MTime: f.Utime,
PickCode: f.PickCode,
})
}
if len(files) < pageSize {
break
}
+1
View File
@@ -90,6 +90,7 @@ type RespBase struct {
Errno int `json:"errno"`
Message string `json:"message"`
Error string `json:"error"`
Count int64 `json:"count"`
Data json.RawMessage `json:"data"`
Raw json.RawMessage `json:"-"` // 原始响应体(外层附加字段用)
}
+62
View File
@@ -89,6 +89,36 @@ func (c *OpenClient) GetFsList(ctx context.Context, cid string, offset, limit in
return files, strings.Join(pathStr, "/"), nil
}
// GetFsListFlat 递归扁平化列出 cid 下的所有文件(跨越所有子目录,不包含文件夹节点),并返回文件列表与该树下的总文件数。
// 类似于 QMediaSync 的 115 扁平化批量拉取机制,极大地降低多层级子目录下的 API 请求次数。
func (c *OpenClient) GetFsListFlat(ctx context.Context, cid string, offset, limit int) ([]RemoteFile, int64, error) {
if cid == "" {
cid = "0"
}
if limit <= 0 {
limit = 1150
}
params := map[string]string{
"cid": cid,
"limit": fmt.Sprint(limit),
"offset": fmt.Sprint(offset),
"cur": "0",
"show_dir": "0",
}
resp, err := c.doAuthJSON(ctx, "GET", ProAPIBase+"/open/ufile/files", params, 2)
if err != nil {
return nil, 0, err
}
if !resp.State {
return nil, 0, NewOpenAPIResponseError(resp.Code, resp.Errno, resp.Message, resp.Error, "115 接口调用失败")
}
files, err := openList[RemoteFile](resp.Data)
if err != nil {
return nil, 0, fmt.Errorf("115: 解析文件列表失败:%w", err)
}
return files, resp.Count, nil
}
// GetFsDetailByCid 查询文件(夹)详情。
func (c *OpenClient) GetFsDetailByCid(ctx context.Context, fileId string) (*RemoteFileDetail, error) {
params := map[string]string{"file_id": fileId}
@@ -113,6 +143,38 @@ type RemoteFileDetail struct {
} `json:"paths"`
}
// RelativePath 计算该目录相对于根同步目录(rootCID)的相对路径。
func (d *RemoteFileDetail) RelativePath(rootCID string) string {
if d == nil || len(d.Paths) == 0 {
return ""
}
if rootCID == "" {
rootCID = "0"
}
rootIdx := -1
for i, p := range d.Paths {
if p.FileId == rootCID {
rootIdx = i
break
}
}
var segments []string
start := 0
if rootIdx >= 0 {
start = rootIdx + 1
} else if len(d.Paths) > 0 && (d.Paths[0].FileId == "0" || d.Paths[0].FileId == "") {
start = 1
}
for i := start; i < len(d.Paths); i++ {
name := strings.TrimSpace(d.Paths[i].Name)
if name != "" {
segments = append(segments, name)
}
}
return strings.Join(segments, "/")
}
// ─── 下载直链 ──────────────────────────────────────────────────────────────────
type downloadURLData struct {
+65 -14
View File
@@ -278,18 +278,69 @@ func TestDanmakuSameBase(t *testing.T) {
require.False(t, sameDanmakuBase("", "https://api.dandanplay.net"))
}
// fetchCommentWithFallback:配置源与官方同源时不重复请求;
// 全失败时带出最后一跳错误。
func TestDanmakuFetchCommentWithFallback(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusInternalServerError)
}))
t.Cleanup(srv.Close)
// fetchCommentWithFallback:配置源与官方同源时不重复请求;
// 全失败时带出最后一跳错误。
func TestDanmakuFetchCommentWithFallback(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusInternalServerError)
}))
t.Cleanup(srv.Close)
svc := newDanmakuTestService(t)
ctx := context.Background()
raw, st, err := svc.fetchCommentWithFallback(ctx, srv.URL, srv.URL, "25484")
require.Error(t, err)
require.Empty(t, raw)
require.Equal(t, "auto", st)
}
svc := newDanmakuTestService(t)
ctx := context.Background()
raw, st, err := svc.fetchCommentWithFallback(ctx, srv.URL, srv.URL, "25484")
require.Error(t, err)
require.Empty(t, raw)
require.Equal(t, "auto", st)
}
// 视频即便能命中 Hash 自动识别,当用户传入手动搜索关键词时应跳过 Hash 匹配,走关键词搜索。
func TestDanmakuFetchHashMatchSkippedOnManualKeyword(t *testing.T) {
videoPath, _ := writeDanmakuTestVideo(t, "测试动画.第01话.mkv")
// 官方服务同时提供 match 和 search:
// match 会返回 episodeId=25484(动画A)
// search 会根据关键词返回 episodeId=99999(动画B)
mux := http.NewServeMux()
var matchCalled bool
mux.HandleFunc("/api/v2/match", func(w http.ResponseWriter, r *http.Request) {
matchCalled = true
w.Header().Set("Content-Type", "application/json")
fmt.Fprint(w, `{"success":true,"isMatched":true,"matches":[{"episodeId":25484,"animeId":1001,"animeTitle":"自动识别动画A","episodeTitle":"第1话"}]}`)
})
mux.HandleFunc("/api/v2/search/episodes", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
fmt.Fprint(w, `{"hasMore":false,"animes":[{"animeId":2002,"animeTitle":"手动搜索动画B","episodes":[{"episodeId":99999,"episodeTitle":"第1话"}]}]}`)
})
mux.HandleFunc("/api/v2/comment/25484", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/xml")
fmt.Fprint(w, `<?xml version="1.0"?><i><d p="0.5,1,16777215,user1">自动识别弹幕</d></i>`)
})
mux.HandleFunc("/api/v2/comment/99999", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/xml")
fmt.Fprint(w, `<?xml version="1.0"?><i><d p="0.5,1,16777215,user2">手动搜索弹幕</d></i>`)
})
official := httptest.NewServer(mux)
t.Cleanup(official.Close)
overrideDanmakuOfficialBase(t, official.URL)
svc := newDanmakuTestService(t)
ctx := context.Background()
seedDanmakuVideoMedia(t, svc, "mManual", "自动识别动画A", videoPath, 32000, 1)
// 1) 默认自动识别:命中 Hash 识别
resAuto, err := svc.Fetch(ctx, "mManual", "", "")
require.NoError(t, err)
require.True(t, matchCalled)
require.Equal(t, "hash", resAuto.MatchMode)
require.Equal(t, int64(25484), resAuto.EpisodeID)
require.Contains(t, resAuto.Raw, "自动识别弹幕")
// 2) 用户传入手动搜索关键词:跳过 Hash 识别,命中搜索结果动画B
resManual, err := svc.Fetch(ctx, "mManual", "手动搜索动画B", "")
require.NoError(t, err)
require.Equal(t, "search", resManual.MatchMode)
require.Equal(t, int64(99999), resManual.EpisodeID)
require.Equal(t, "手动搜索动画B", resManual.AnimeTitle)
require.Contains(t, resManual.Raw, "手动搜索弹幕")
}
+2 -2
View File
@@ -219,8 +219,8 @@ func (s *DanmakuService) Fetch(ctx context.Context, mediaID, keyword, episodeID
target := ""
// 1) hash 识别:始终走官方 /api/v2/match。
if media != nil && media.Path != "" {
// 1) hash 识别:始终走官方 /api/v2/match(keyword 手动覆盖时跳过,直接走第 3 层)。
if target == "" && !manualKeyword && media != nil && media.Path != "" {
if hash, ok := s.mediaHash(ctx, media); ok {
fileSize := media.SizeBytes
if strings.EqualFold(filepath.Ext(media.Path), ".strm") {
+1
View File
@@ -111,6 +111,7 @@ const (
var (
embySeasonDirRE = regexp.MustCompile(`(?i)^(season[\s._-]*\d+|s\d+|specials?|sp|ova|oad|extra|extras|第\s*[0-9一二三四五六七八九十百零两]+\s*季|特别篇|特別篇|番外|特典)$`)
embySeasonSuffixRE = regexp.MustCompile(`(?i)(?:[\s._-]+(?:season[\s._-]*\d+|s\d+|第\s*[0-9一二三四五六七八九十百零两]+\s*季|specials?|sp|ova|oad|extra|extras|特别篇|特別篇|番外|特典)|\s*第\s*[0-9一二三四五六七八九十百零两]+\s*季)\s*$`)
embyYearSuffixRE = regexp.MustCompile(`\s*[\((\[]\d{4}[\))\]]\s*$`)
embyEpisodeTitleRE = regexp.MustCompile(`(?i)\s*[-_ ]*s\d{1,2}e\d{1,3}.*$`)
)
@@ -334,3 +334,105 @@ func TestEmbyCloudAnimeUsesSeriesNameFromChineseSeasonFolder(t *testing.T) {
t.Fatalf("cloud anime should be grouped as one series named 剑来, got %#v", items)
}
}
func TestEmbySeriesGroupingWithPrefixedSeasonFolders(t *testing.T) {
svc := newTestEmbyService(t)
lib := model.Library{Name: "动漫", Path: `/media/动漫`, Type: "anime", Enabled: true}
if err := svc.repo.Library.Create(t.Context(), &lib); err != nil {
t.Fatalf("create library: %v", err)
}
for season := 1; season <= 5; season++ {
for ep := 1; ep <= 3; ep++ {
media := model.Media{
Base: model.Base{ID: fmt.Sprintf("shokugeki-s%02de%02d", season, ep)},
LibraryID: lib.ID,
Title: "食戟之灵",
OriginalName: "食戟のソーマ",
ScrapeStatus: "matched",
TMDbID: 62273,
BangumiID: 116461,
Path: fmt.Sprintf(`/media/动漫/食戟之灵/食戟之灵 S%02d/食戟之灵 S%02dE%02d.strm`, season, season, ep),
SeasonNum: season,
EpisodeNum: ep,
}
if err := svc.repo.DB.Create(&media).Error; err != nil {
t.Fatalf("create media: %v", err)
}
}
}
root, err := svc.Items(t.Context(), ItemsParams{ParentID: lib.ID, Limit: 50})
if err != nil {
t.Fatalf("library items: %v", err)
}
rootItems := root["Items"].([]map[string]any)
if len(rootItems) != 1 {
t.Fatalf("expected 1 series card for 食戟之灵 across 5 seasons, got %d cards: %#v", len(rootItems), rootItems)
}
if rootItems[0]["Name"] != "食戟之灵" || rootItems[0]["Type"] != "Series" {
t.Fatalf("unexpected series item: %#v", rootItems[0])
}
seriesID := rootItems[0]["Id"].(string)
seasons, err := svc.Items(t.Context(), ItemsParams{ParentID: seriesID, Limit: 50})
if err != nil {
t.Fatalf("series seasons: %v", err)
}
seasonItems := seasons["Items"].([]map[string]any)
if len(seasonItems) != 5 {
t.Fatalf("expected 5 seasons, got %d: %#v", len(seasonItems), seasonItems)
}
for i, s := range seasonItems {
wantSeasonNum := i + 1
if s["Type"] != "Season" || s["IndexNumber"] != wantSeasonNum {
t.Errorf("season [%d] = %#v, want IndexNumber=%d", i, s, wantSeasonNum)
}
}
counts, err := svc.ItemCounts(t.Context(), "user-1")
if err != nil {
t.Fatalf("item counts: %v", err)
}
if counts["SeriesCount"] != 1 || counts["EpisodeCount"] != int64(15) {
t.Fatalf("counts = %#v, want 1 series and 15 episodes", counts)
}
}
func TestInferSeriesNameFromPath(t *testing.T) {
tests := []struct {
path string
want string
}{
{
path: `/media/动漫/食戟之灵/食戟之灵 S01/食戟之灵 S01E01.strm`,
want: "食戟之灵",
},
{
path: `/media/动漫/食戟之灵/食戟之灵 S05/食戟之灵 S05E12.strm`,
want: "食戟之灵",
},
{
path: `/media/动漫/食戟之灵/Season 02/01.mkv`,
want: "食戟之灵",
},
{
path: `/media/动漫/进击的巨人 第2季/01.mkv`,
want: "进击的巨人",
},
{
path: `cloud://openlist/国漫/剑来/第二季/04.mkv`,
want: "剑来",
},
{
path: `/media/tv/间谍过家家 (2022)/Specials/S00E01.mkv`,
want: "间谍过家家",
},
}
for _, tc := range tests {
got := inferSeriesNameFromPath(tc.path)
if got != tc.want {
t.Errorf("inferSeriesNameFromPath(%q) = %q, want %q", tc.path, got, tc.want)
}
}
}
+30 -1
View File
@@ -31,6 +31,14 @@ func (e *EmbyService) seriesNameForMedia(m *model.Media) string {
return series.Title
}
}
if strings.EqualFold(strings.TrimSpace(m.ScrapeStatus), "matched") && strings.TrimSpace(m.Title) != "" {
name := strings.TrimSpace(m.Title)
name = embyEpisodeTitleRE.ReplaceAllString(name, "")
name = embyYearSuffixRE.ReplaceAllString(name, "")
if name != "" {
return name
}
}
if name := inferSeriesNameFromPath(m.Path); name != "" {
return name
}
@@ -53,14 +61,35 @@ func inferSeriesNameFromPath(path string) string {
if embySeasonDirRE.MatchString(base) {
dir = filepath.Dir(dir)
base = filepath.Base(dir)
} else if stripped := strings.TrimSpace(embySeasonSuffixRE.ReplaceAllString(base, "")); stripped != "" && stripped != base {
parentDir := filepath.Dir(dir)
parentBase := filepath.Base(parentDir)
if parentBase != "." && parentBase != string(filepath.Separator) && !isEmbyGenericContainer(parentBase) {
dir = parentDir
base = parentBase
} else {
base = stripped
}
}
base = strings.TrimSpace(embyYearSuffixRE.ReplaceAllString(base, ""))
if base == "." || base == string(filepath.Separator) {
if base == "." || base == string(filepath.Separator) || isEmbyGenericContainer(base) {
return ""
}
return base
}
func isEmbyGenericContainer(name string) bool {
switch strings.ToLower(strings.TrimSpace(name)) {
case "movie", "movies", "film", "films", "tv", "series", "show", "shows", "anime", "animation", "variety",
"电视剧", "剧集", "连续剧", "短剧", "国产剧", "国剧", "欧美剧", "美剧", "英剧", "日韩剧", "日剧", "韩剧", "港剧", "台剧", "港台剧",
"综艺", "纪录片", "儿童", "动漫", "番剧", "国漫", "日番", "韩漫", "美漫", "欧美动漫", "欧美动画", "其他动漫", "电影", "成人", "未分类",
"media", "downloads", "download", "videos", "video", "share", "shares":
return true
default:
return false
}
}
func stableEmbyID(prefix string, parts ...string) string {
h := sha256.New()
for _, part := range parts {
+11 -4
View File
@@ -22,11 +22,18 @@ func TestNormalizeCloudPlayTarget(t *testing.T) {
if parsed.IsAbs() || parsed.Host != "" {
t.Fatalf("normalized target should be relative, got %q", got)
}
if parsed.Query().Get("ref") != ref {
t.Fatalf("ref round-trip failed: %q", parsed.Query().Get("ref"))
}
if parsed.Query().Get("ref") != ref {
t.Fatalf("ref round-trip failed: %q", parsed.Query().Get("ref"))
}
// 非云盘播放 URL 保持原样(WebDAV/直链等)。
strmStale := "http://bwg.linkmy.fun:1314/api/strm/play/cloud115/video.mkv?acct=abc&pickcode=123"
gotStrm := normalizeCloudPlayTarget(strmStale)
wantStrm := "/api/strm/play/cloud115/video.mkv?acct=abc&pickcode=123"
if gotStrm != wantStrm {
t.Fatalf("normalizeCloudPlayTarget(strm) = %q, want %q", gotStrm, wantStrm)
}
// 非云盘播放 URL 保持原样(WebDAV/直链等)。
passthrough := "https://dav.example.com/media/file.mkv"
if got := normalizeCloudPlayTarget(passthrough); got != passthrough {
t.Fatalf("non-cloud target should pass through, got %q", got)
+14 -3
View File
@@ -15,11 +15,22 @@ import (
// /api/cloud/play 路径,由 absoluteInternalRedirect 基于「当前请求」补全
// host,从而对历史脏数据免疫。
func normalizeCloudPlayTarget(raw string) string {
typ, ref, ok := parseCloudMediaPlaybackURL(raw)
if !ok {
raw = strings.TrimSpace(raw)
if raw == "" {
return raw
}
return BuildRelativeCloudPlayURL(typ, ref)
if typ, ref, ok := parseCloudMediaPlaybackURL(raw); ok {
return BuildRelativeCloudPlayURL(typ, ref)
}
if u, err := url.Parse(raw); err == nil {
path := strings.ToLower(u.Path)
if strings.HasPrefix(path, "/api/strm/play/") || strings.HasPrefix(path, "/api/cloud/play/") || strings.HasPrefix(path, "/api/stream/") {
u.Scheme = ""
u.Host = ""
return u.String()
}
}
return raw
}
// BuildRelativeCloudPlayURL 构造相对的云盘播放 API 路径。
+9
View File
@@ -404,6 +404,9 @@ func (s *StrmService) CreateSyncPath(ctx context.Context, p *model.StrmSyncPath)
if strings.TrimSpace(p.Name) == "" {
p.Name = "同步目录 " + time.Now().Format("01-02 15:04")
}
if p.SyncMode == "" {
p.SyncMode = model.StrmSyncTypeIncremental
}
if p.EnableCron && strings.TrimSpace(p.Cron) == "" {
return nil, errors.New("启用定时同步需要填写 cron 表达式")
}
@@ -430,6 +433,12 @@ func (s *StrmService) UpdateSyncPath(ctx context.Context, id string, p *model.St
p.LastSyncAt = existing.LastSyncAt
p.LastSyncStatus = existing.LastSyncStatus
p.LastSyncMessage = existing.LastSyncMessage
if p.SyncMode == "" {
p.SyncMode = existing.SyncMode
if p.SyncMode == "" {
p.SyncMode = model.StrmSyncTypeIncremental
}
}
if p.EnableCron && strings.TrimSpace(p.Cron) == "" {
return nil, errors.New("启用定时同步需要填写 cron 表达式")
}
+459 -99
View File
@@ -20,6 +20,7 @@ import (
"github.com/ShukeBta/MMTL/internal/model"
"github.com/ShukeBta/MMTL/internal/service/cloud"
"github.com/ShukeBta/MMTL/internal/service/cloud115"
)
// strmSyncState 是一次同步执行的上下文。
@@ -31,16 +32,24 @@ type strmSyncState struct {
provider cloud.Provider // local 提供方为 nil
cfg *strmPathConfig
rec *model.StrmSyncRecord
syncType string
mu sync.Mutex
processed int // 已处理文件计数(用于定期落库进度)
seenVideo map[string]bool // "v:"+去掉扩展名的相对路径 → 远端存在该视频
seenMeta map[string]bool // "m:"+相对路径 → 远端存在该元数据
remoteMeta map[string]int64 // 远端元数据大小(上传比对用)
mu sync.Mutex
processed int // 已处理文件计数(用于定期落库进度)
lastProgressFlush time.Time // 上次进度落库时间
seenVideo map[string]bool // "v:"+去掉扩展名的相对路径 → 远端存在该视频
seenMeta map[string]bool // "m:"+相对路径 → 远端存在该元数据
remoteMeta map[string]int64 // 远端元数据大小(上传比对用)
activeDownloadPaths map[string]bool // 本地已在排队/进行的下载任务路径(内存去重)
activeUploadPaths map[string]bool // 本地已在排队/进行的上传任务路径(内存去重)
pendingDownloads []*model.StrmDownloadTask
pendingUploads []*model.StrmUploadTask
dirCache sync.Map // dirID (string) -> relativePath (string)
}
// StartSync 启动一次同步(异步执行,同一目录同时只允许一个任务)。
func (s *StrmService) StartSync(ctx context.Context, pathID string) error {
// syncType 支持 "incremental"(默认增量)和 "full"(全量同步)。
func (s *StrmService) StartSync(ctx context.Context, pathID string, syncType ...string) error {
p, err := s.repo.StrmSyncPath.FindByID(ctx, pathID)
if err != nil || p == nil {
return errNotFoundOr(err, "同步目录不存在")
@@ -64,9 +73,20 @@ func (s *StrmService) StartSync(ctx context.Context, pathID string) error {
s.running[pathID] = cancel
s.mu.Unlock()
mode := model.StrmSyncTypeIncremental
if len(syncType) > 0 && syncType[0] != "" {
mode = syncType[0]
} else if p.SyncMode != "" {
mode = p.SyncMode
}
if mode != model.StrmSyncTypeFull {
mode = model.StrmSyncTypeIncremental
}
now := time.Now()
rec := &model.StrmSyncRecord{
SyncPathID: pathID,
SyncType: mode,
Status: model.StrmSyncRecordRunning,
StartedAt: &now,
}
@@ -145,6 +165,7 @@ func (s *StrmService) runSync(ctx context.Context, p *model.StrmSyncPath, rec *m
p: p,
cfg: cfg,
rec: rec,
syncType: rec.SyncType,
seenVideo: map[string]bool{},
seenMeta: map[string]bool{},
remoteMeta: map[string]int64{},
@@ -191,15 +212,19 @@ func (s *StrmService) finishSync(p *model.StrmSyncPath, rec *model.StrmSyncRecor
p.LastSyncStatus = status
p.LastSyncMessage = message
if status != model.StrmSyncRecordFailed && message == "" {
p.LastSyncMessage = fmt.Sprintf("完成:新增/更新 %d 个 strm,下载 %d 个元数据,清理 %d 个文件",
rec.NewStrm, rec.NewMeta, rec.Pruned)
syncTypeLabel := "增量"
if rec.SyncType == model.StrmSyncTypeFull {
syncTypeLabel = "全量"
}
p.LastSyncMessage = fmt.Sprintf("[%s] 完成:新增/更新 %d 个 strm,跳过 %d 个,下载 %d 个元数据,清理 %d 个文件",
syncTypeLabel, rec.NewStrm, rec.Skipped, rec.NewMeta, rec.Pruned)
}
if err := s.repo.StrmSyncPath.Update(context.Background(), p); err != nil {
s.log.Warn("update strm sync path failed", zap.Error(err))
}
s.log.Info("strm sync finished",
zap.String("path_id", p.ID), zap.String("status", status),
zap.Int64("new_strm", rec.NewStrm), zap.Int64("new_meta", rec.NewMeta),
zap.String("path_id", p.ID), zap.String("sync_type", rec.SyncType), zap.String("status", status),
zap.Int64("new_strm", rec.NewStrm), zap.Int64("skipped", rec.Skipped), zap.Int64("new_meta", rec.NewMeta),
zap.Int64("pruned", rec.Pruned), zap.String("message", message))
}
@@ -207,20 +232,43 @@ func (st *strmSyncState) run() error {
if err := ensureLocalDir(st.p.LocalPath); err != nil {
return fmt.Errorf("创建输出目录失败:%w", err)
}
if st.cfg.DownloadMeta {
if active, err := st.s.repo.StrmDownload.GetActiveLocalPathMap(st.ctx, st.p.ID); err == nil {
st.activeDownloadPaths = active
} else {
st.activeDownloadPaths = map[string]bool{}
}
}
if st.cfg.UploadMeta {
if active, err := st.s.repo.StrmUpload.GetActiveLocalPathMap(st.ctx, st.p.ID); err == nil {
st.activeUploadPaths = active
} else {
st.activeUploadPaths = map[string]bool{}
}
}
if st.provider != nil {
if err := st.walkRemote(); err != nil {
return err
if open115, ok := st.provider.(cloud.OpenAPI115Provider); ok && st.p.Provider == model.StrmProvider115 {
if err := st.walk115Flat(open115.OpenClient()); err != nil {
return err
}
} else {
if err := st.walkRemote(); err != nil {
return err
}
}
} else {
if err := st.walkLocalSource(); err != nil {
return err
}
}
st.flushPendingDownloads()
st.flushProgress()
if st.cfg.UploadMeta && st.provider != nil && st.p.Provider != model.StrmProvider115 {
if err := st.scanLocalMetaForUpload(); err != nil {
return err
}
st.flushPendingUploads()
}
if err := st.pruneLocal(); err != nil {
return err
@@ -239,6 +287,7 @@ const strmScanWorkers = 8
// 多个 worker 并行执行 List(受全局 115 令牌桶限流约束),子目录动态
// 入队;任一目录失败则取消其余 worker 并返回错误(与旧串行版语义一致)。
func (st *strmSyncState) walkRemote() error {
defer st.flushPendingDownloads()
root := strings.TrimSpace(st.p.RemotePath)
if root == "" {
root = "/"
@@ -374,6 +423,216 @@ func (st *strmSyncState) isMetaExt(ext string) bool {
return false
}
// walk115Flat 使用 115 开放平台扁平化分页批量拉取机制与目录拓扑缓存(参考 QMediaSync)。
// 极大地降低 API 请求次数并支持毫秒级/秒级增量同步。
func (st *strmSyncState) walk115Flat(open115 *cloud115.OpenClient) error {
defer st.flushPendingDownloads()
ctx := st.ctx
rootCID := strings.TrimSpace(st.p.RemotePath)
if rootCID == "" {
rootCID = "0"
}
// 1. 目录拓扑缓存处理
st.dirCache.Store(rootCID, "")
if st.syncType == model.StrmSyncTypeFull {
// 全量同步:清空本路径的历史目录缓存
if err := st.s.repo.StrmDirCache.DeleteBySyncPathID(ctx, st.p.ID); err != nil {
st.s.log.Warn("delete strm dir cache failed", zap.Error(err))
}
} else {
// 增量同步:预加载历史目录缓存
cached, err := st.s.repo.StrmDirCache.ListBySyncPathID(ctx, st.p.ID)
if err == nil {
for _, item := range cached {
st.dirCache.Store(item.DirID, item.Path)
}
}
}
// 2. 探测文件总数
const pageSize = 1150
firstBatch, totalCount, err := open115.GetFsListFlat(ctx, rootCID, 0, pageSize)
if err != nil {
return fmt.Errorf("115: 获取文件列表失败:%w", err)
}
allFiles := make([]cloud115.RemoteFile, 0, totalCount)
allFiles = append(allFiles, firstBatch...)
// 3. 并发分页拉取剩余文件
if totalCount > int64(len(firstBatch)) {
totalPages := int((totalCount + pageSize - 1) / pageSize)
type pageTask struct {
offset int
}
pageTasks := make([]pageTask, 0, totalPages-1)
for page := 1; page < totalPages; page++ {
pageTasks = append(pageTasks, pageTask{offset: page * pageSize})
}
var (
filesMu sync.Mutex
wg sync.WaitGroup
taskCh = make(chan pageTask, len(pageTasks))
errMu sync.Mutex
fetchErr error
)
for _, t := range pageTasks {
taskCh <- t
}
close(taskCh)
workers := 4
if len(pageTasks) < workers {
workers = len(pageTasks)
}
for i := 0; i < workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for t := range taskCh {
if ctx.Err() != nil {
return
}
files, _, err := open115.GetFsListFlat(ctx, rootCID, t.offset, pageSize)
if err != nil {
errMu.Lock()
if fetchErr == nil {
fetchErr = err
}
errMu.Unlock()
return
}
filesMu.Lock()
allFiles = append(allFiles, files...)
filesMu.Unlock()
}
}()
}
wg.Wait()
if fetchErr != nil {
return fmt.Errorf("115: 分页拉取失败:%w", fetchErr)
}
}
if ctx.Err() != nil {
return ctx.Err()
}
// 4. 收集所有未在缓存中的父目录 ID (file.Pid)
missingPids := make(map[string]struct{})
for _, f := range allFiles {
pid := f.Pid
if pid == "" || pid == rootCID {
continue
}
if _, ok := st.dirCache.Load(pid); !ok {
missingPids[pid] = struct{}{}
}
}
// 并发补全未知目录详情与祖先链
if len(missingPids) > 0 {
pidList := make([]string, 0, len(missingPids))
for pid := range missingPids {
pidList = append(pidList, pid)
}
pidCh := make(chan string, len(pidList))
for _, pid := range pidList {
pidCh <- pid
}
close(pidCh)
var (
pwg sync.WaitGroup
dirWorkers = 4
)
if len(pidList) < dirWorkers {
dirWorkers = len(pidList)
}
for i := 0; i < dirWorkers; i++ {
pwg.Add(1)
go func() {
defer pwg.Done()
for pid := range pidCh {
if ctx.Err() != nil {
return
}
detail, err := open115.GetFsDetailByCid(ctx, pid)
if err != nil {
st.s.log.Warn("115: 获取目录详情失败", zap.String("pid", pid), zap.Error(err))
continue
}
if detail == nil {
continue
}
// 解析相对路径
relPath := detail.RelativePath(rootCID)
st.dirCache.Store(pid, relPath)
_ = st.s.repo.StrmDirCache.Set(ctx, st.p.ID, pid, relPath)
// 顺便解析并缓存 detail.Paths 中包含的中间各层级目录
for _, ancestor := range detail.Paths {
if ancestor.FileId == "0" || ancestor.FileId == rootCID {
continue
}
if _, loaded := st.dirCache.Load(ancestor.FileId); !loaded {
subDetail := &cloud115.RemoteFileDetail{
FileId: ancestor.FileId,
Paths: nil,
}
for _, p := range detail.Paths {
subDetail.Paths = append(subDetail.Paths, p)
if p.FileId == ancestor.FileId {
break
}
}
ancestorRel := subDetail.RelativePath(rootCID)
st.dirCache.Store(ancestor.FileId, ancestorRel)
_ = st.s.repo.StrmDirCache.Set(ctx, st.p.ID, ancestor.FileId, ancestorRel)
}
}
}
}()
}
pwg.Wait()
}
// 5. 分类处理所有文件
for _, f := range allFiles {
if ctx.Err() != nil {
return ctx.Err()
}
cleanName := cleanEntryName(f.FileName, false)
var rel string
if f.Pid == "" || f.Pid == rootCID {
rel = cleanName
} else {
if parentVal, ok := st.dirCache.Load(f.Pid); ok && parentVal.(string) != "" {
rel = parentVal.(string) + "/" + cleanName
} else {
rel = cleanName
}
}
entry := cloud.FileEntry{
ID: f.FileId,
Name: f.FileName,
IsDir: false,
Size: f.FileSize,
MTime: f.Utime,
PickCode: f.PickCode,
}
st.processRemoteFile(entry, rel)
}
return nil
}
// handleVideo 生成/更新 .strm 文件。
func (st *strmSyncState) handleVideo(entry cloud.FileEntry, rel, ext string) {
relSansExt := rel[:len(rel)-len(ext)]
@@ -391,6 +650,18 @@ func (st *strmSyncState) handleVideo(entry cloud.FileEntry, rel, ext string) {
st.s.log.Warn("strm target path out of root", zap.String("rel", targetRel), zap.Error(err))
return
}
// 增量同步模式快速检查:本地 strm 文件存在、非空且修改时间与远端 mtime 一致,直接跳过无需读磁盘
if st.syncType == model.StrmSyncTypeIncremental && entry.MTime > 0 {
if info, err := os.Stat(target); err == nil && info.Size() > 0 && info.ModTime().Unix() == entry.MTime {
st.mu.Lock()
st.rec.Skipped++
st.mu.Unlock()
st.touchProgress()
return
}
}
content, err := st.strmContent(entry, rel, ext)
if err != nil {
// 并发 worker 下 rec.Message 无锁写会有数据竞争,这里仅记录日志;
@@ -403,6 +674,11 @@ func (st *strmSyncState) handleVideo(entry cloud.FileEntry, rel, ext string) {
existing = string(data)
}
if existing == content {
// 对齐本地 strm 修改时间为远端 mtime,便于后续秒级比对
if entry.MTime > 0 {
mTime := time.Unix(entry.MTime, 0)
_ = os.Chtimes(target, mTime, mTime)
}
st.mu.Lock()
st.rec.Skipped++
st.mu.Unlock()
@@ -423,6 +699,10 @@ func (st *strmSyncState) handleVideo(entry cloud.FileEntry, rel, ext string) {
st.s.log.Warn("rename strm failed", zap.String("file", target), zap.Error(err))
return
}
if entry.MTime > 0 {
mTime := time.Unix(entry.MTime, 0)
_ = os.Chtimes(target, mTime, mTime)
}
st.mu.Lock()
st.rec.NewStrm++
st.mu.Unlock()
@@ -482,6 +762,36 @@ func (st *strmSyncState) recordRemoteMeta(entry cloud.FileEntry, rel string) {
st.mu.Unlock()
}
func (st *strmSyncState) flushPendingDownloads() {
st.mu.Lock()
if len(st.pendingDownloads) == 0 {
st.mu.Unlock()
return
}
batch := st.pendingDownloads
st.pendingDownloads = nil
st.mu.Unlock()
if err := st.s.repo.StrmDownload.CreateInBatches(st.ctx, batch, 100); err != nil {
st.s.log.Warn("batch enqueue strm download tasks failed", zap.Error(err))
}
}
func (st *strmSyncState) flushPendingUploads() {
st.mu.Lock()
if len(st.pendingUploads) == 0 {
st.mu.Unlock()
return
}
batch := st.pendingUploads
st.pendingUploads = nil
st.mu.Unlock()
if err := st.s.repo.StrmUpload.CreateInBatches(st.ctx, batch, 100); err != nil {
st.s.log.Warn("batch enqueue strm upload tasks failed", zap.Error(err))
}
}
// handleMeta 元数据入下载队列(本地已存在且大小一致则跳过)。
func (st *strmSyncState) handleMeta(entry cloud.FileEntry, rel, ext string) {
st.recordRemoteMeta(entry, rel)
@@ -494,10 +804,22 @@ func (st *strmSyncState) handleMeta(entry cloud.FileEntry, rel, ext string) {
st.touchProgress()
return
}
if st.taskExists("download", st.p.ID, target) {
st.mu.Lock()
if st.activeDownloadPaths == nil {
if active, err := st.s.repo.StrmDownload.GetActiveLocalPathMap(st.ctx, st.p.ID); err == nil {
st.activeDownloadPaths = active
} else {
st.activeDownloadPaths = map[string]bool{}
}
}
if st.activeDownloadPaths[target] {
st.mu.Unlock()
st.touchProgress()
return
}
st.activeDownloadPaths[target] = true
st.mu.Unlock()
task := &model.StrmDownloadTask{
SyncPathID: st.p.ID,
AccountID: st.p.AccountID,
@@ -516,13 +838,16 @@ func (st *strmSyncState) handleMeta(entry cloud.FileEntry, rel, ext string) {
if st.p.Provider != model.StrmProvider115 {
task.RemoteRef = entry.ID
}
if err := st.s.repo.StrmDownload.Create(st.ctx, task); err != nil {
st.s.log.Warn("enqueue strm download task failed", zap.Error(err))
return
}
st.mu.Lock()
st.pendingDownloads = append(st.pendingDownloads, task)
shouldFlush := len(st.pendingDownloads) >= 100
st.rec.NewMeta++
st.mu.Unlock()
if shouldFlush {
st.flushPendingDownloads()
}
st.touchProgress()
}
@@ -576,93 +901,124 @@ func (st *strmSyncState) walkLocalSource() error {
if err != nil {
return nil
}
target, err := joinLocalRel(st.p.LocalPath, relSansExt+".strm")
if err != nil {
return nil
}
if data, err := os.ReadFile(target); err == nil && string(data) == content {
target, err := joinLocalRel(st.p.LocalPath, relSansExt+".strm")
if err != nil {
return nil
}
mTime := info.ModTime()
if st.syncType == model.StrmSyncTypeIncremental {
if tInfo, err := os.Stat(target); err == nil && tInfo.Size() > 0 && tInfo.ModTime().Unix() == mTime.Unix() {
st.mu.Lock()
st.rec.Skipped++
st.mu.Unlock()
st.touchProgress()
return nil
}
}
if data, err := os.ReadFile(target); err == nil && string(data) == content {
_ = os.Chtimes(target, mTime, mTime)
st.mu.Lock()
st.rec.Skipped++
st.mu.Unlock()
st.touchProgress()
return nil
}
if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
return nil
}
tmp := target + ".tmp"
if err := os.WriteFile(tmp, []byte(content), 0o644); err == nil {
_ = os.Rename(tmp, target)
_ = os.Chtimes(target, mTime, mTime)
} else {
_ = os.Remove(tmp)
}
st.mu.Lock()
st.rec.Skipped++
st.rec.NewStrm++
st.mu.Unlock()
st.touchProgress()
return nil
}
if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
return nil
}
tmp := target + ".tmp"
if err := os.WriteFile(tmp, []byte(content), 0o644); err == nil {
_ = os.Rename(tmp, target)
} else {
_ = os.Remove(tmp)
}
st.mu.Lock()
st.rec.NewStrm++
st.mu.Unlock()
return nil
})
}
// scanLocalMetaForUpload 扫描本地元数据,与远端比对后入上传队列。
func (st *strmSyncState) scanLocalMetaForUpload() error {
localRoot := filepath.Clean(st.p.LocalPath)
return filepath.WalkDir(localRoot, func(path string, d os.DirEntry, err error) error {
if err != nil {
// scanLocalMetaForUpload 扫描本地元数据,与远端比对后入上传队列。
func (st *strmSyncState) scanLocalMetaForUpload() error {
defer st.flushPendingUploads()
if st.activeUploadPaths == nil {
if active, err := st.s.repo.StrmUpload.GetActiveLocalPathMap(st.ctx, st.p.ID); err == nil {
st.activeUploadPaths = active
} else {
st.activeUploadPaths = map[string]bool{}
}
}
localRoot := filepath.Clean(st.p.LocalPath)
return filepath.WalkDir(localRoot, func(path string, d os.DirEntry, err error) error {
if err != nil {
return nil
}
if path == localRoot {
return nil
}
select {
case <-st.ctx.Done():
return st.ctx.Err()
default:
}
if d.IsDir() {
return nil
}
rel, err := filepath.Rel(localRoot, path)
if err != nil {
return nil
}
rel = filepath.ToSlash(rel)
ext := strings.ToLower(filepath.Ext(rel))
if !st.isMetaExt(ext) {
return nil
}
info, err := d.Info()
if err != nil {
return nil
}
st.mu.Lock()
_, exists := st.remoteMeta["m:"+rel]
st.mu.Unlock()
if exists {
// 网盘端已存在该元数据文件,跳过上传
return nil
}
st.mu.Lock()
if st.activeUploadPaths != nil && st.activeUploadPaths[path] {
st.mu.Unlock()
return nil
}
if st.activeUploadPaths != nil {
st.activeUploadPaths[path] = true
}
st.mu.Unlock()
task := &model.StrmUploadTask{
SyncPathID: st.p.ID,
AccountID: st.p.AccountID,
Provider: st.p.Provider,
FileName: filepath.Base(rel),
LocalPath: path,
RemotePath: st.remoteUploadPath(rel),
Size: info.Size(),
Status: model.StrmTaskPending,
}
st.mu.Lock()
st.pendingUploads = append(st.pendingUploads, task)
shouldFlush := len(st.pendingUploads) >= 100
st.rec.Uploaded++
st.mu.Unlock()
if shouldFlush {
st.flushPendingUploads()
}
return nil
}
if path == localRoot {
return nil
}
select {
case <-st.ctx.Done():
return st.ctx.Err()
default:
}
if d.IsDir() {
return nil
}
rel, err := filepath.Rel(localRoot, path)
if err != nil {
return nil
}
rel = filepath.ToSlash(rel)
ext := strings.ToLower(filepath.Ext(rel))
if !st.isMetaExt(ext) {
return nil
}
info, err := d.Info()
if err != nil {
return nil
}
st.mu.Lock()
_, exists := st.remoteMeta["m:"+rel]
st.mu.Unlock()
if exists {
// 网盘端已存在该元数据文件,跳过上传
return nil
}
if st.taskExists("upload", st.p.ID, path) {
return nil
}
task := &model.StrmUploadTask{
SyncPathID: st.p.ID,
AccountID: st.p.AccountID,
Provider: st.p.Provider,
FileName: filepath.Base(rel),
LocalPath: path,
RemotePath: st.remoteUploadPath(rel),
Size: info.Size(),
Status: model.StrmTaskPending,
}
if err := st.s.repo.StrmUpload.Create(st.ctx, task); err != nil {
st.s.log.Warn("enqueue strm upload task failed", zap.Error(err))
return nil
}
st.mu.Lock()
st.rec.Uploaded++
st.mu.Unlock()
return nil
})
}
})
}
// remoteUploadPath 远端元数据目标路径 = 同步目录远端根 + 相对路径。
func (st *strmSyncState) remoteUploadPath(rel string) string {
@@ -748,12 +1104,16 @@ func (st *strmSyncState) pruneLocal() error {
return nil
}
// touchProgress 每处理若干个文件落库一次进度。
// touchProgress 进度计数并限流防抖落库(避免高频写 SQLite 导致锁竞争)。
func (st *strmSyncState) touchProgress() {
st.mu.Lock()
st.rec.Total++
st.processed++
flush := st.processed%100 == 0
now := time.Now()
flush := st.processed%100 == 0 || (st.processed%20 == 0 && now.Sub(st.lastProgressFlush) >= 2*time.Second)
if flush {
st.lastProgressFlush = now
}
st.mu.Unlock()
if flush {
st.flushProgress()
+120 -7
View File
@@ -6,6 +6,7 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
@@ -34,10 +35,10 @@ func testStrmService(t *testing.T) *StrmService {
sqlDB.SetMaxOpenConns(4)
t.Cleanup(func() { _ = sqlDB.Close() })
}
if err := db.AutoMigrate(&model.StrmAccount{}, &model.StrmSyncPath{}, &model.StrmSyncRecord{},
&model.StrmDownloadTask{}, &model.StrmUploadTask{}, &model.Setting{}); err != nil {
t.Fatal(err)
}
if err := db.AutoMigrate(&model.StrmAccount{}, &model.StrmSyncPath{}, &model.StrmSyncRecord{},
&model.StrmDownloadTask{}, &model.StrmUploadTask{}, &model.StrmDirCache{}, &model.Setting{}); err != nil {
t.Fatal(err)
}
repos := repository.New(db)
ctx := context.Background()
if err := repos.Setting.Set(ctx, StrmSettingBaseURL, "http://test.local:8096"); err != nil {
@@ -159,6 +160,57 @@ func TestLocalStrmSync(t *testing.T) {
}
}
// TestStrmFullAndIncrementalSync 测试增量同步与全量同步模式切换及记录
func TestStrmFullAndIncrementalSync(t *testing.T) {
svc := testStrmService(t)
src := t.TempDir()
out := t.TempDir()
writeFile(t, filepath.Join(src, "电影", "星际穿越.mkv"), "fake-video-data")
p := syncPathRecord(t, svc, model.StrmProviderLocal, src, out, true)
// 1. 默认触发增量同步
if err := svc.StartSync(context.Background(), p.ID, model.StrmSyncTypeIncremental); err != nil {
t.Fatal(err)
}
record := waitSyncDone(t, svc, p.ID, 10*time.Second)
if record.Status != model.StrmSyncRecordDone {
t.Fatalf("sync status = %s, message = %s", record.Status, record.Message)
}
if record.SyncType != model.StrmSyncTypeIncremental {
t.Fatalf("expected sync_type = incremental, got %s", record.SyncType)
}
if record.NewStrm != 1 {
t.Fatalf("expected 1 new strm, got %d", record.NewStrm)
}
// 2. 再次执行增量同步,应当跳过
if err := svc.StartSync(context.Background(), p.ID, model.StrmSyncTypeIncremental); err != nil {
t.Fatal(err)
}
record = waitSyncDone(t, svc, p.ID, 10*time.Second)
if record.SyncType != model.StrmSyncTypeIncremental {
t.Fatalf("expected sync_type = incremental, got %s", record.SyncType)
}
if record.Skipped != 1 {
t.Fatalf("expected 1 skipped, got %d", record.Skipped)
}
// 3. 执行全量同步
if err := svc.StartSync(context.Background(), p.ID, model.StrmSyncTypeFull); err != nil {
t.Fatal(err)
}
record = waitSyncDone(t, svc, p.ID, 10*time.Second)
if record.SyncType != model.StrmSyncTypeFull {
t.Fatalf("expected sync_type = full, got %s", record.SyncType)
}
if record.Status != model.StrmSyncRecordDone {
t.Fatalf("full sync failed: status = %s, message = %s", record.Status, record.Message)
}
}
// TestStrmCronMatches cron 表达式匹配。
func TestStrmCronMatches(t *testing.T) {
cases := []struct {
@@ -442,7 +494,68 @@ func TestWalkRemoteConcurrent(t *testing.T) {
if walkErr != nil {
t.Fatal(walkErr)
}
if strmCount != 5 {
t.Errorf("生成的 .strm 数量 = %d,期望 5", strmCount)
if strmCount != 5 {
t.Errorf("生成的 .strm 数量 = %d,期望 5", strmCount)
}
}
}
// TestStrmBatchEnqueueAndConcurrentClaim 测试大规模批量入库及多协程并发认领无死锁
func TestStrmBatchEnqueueAndConcurrentClaim(t *testing.T) {
svc := testStrmService(t)
ctx := context.Background()
// 1. 批量插入 200 个下载任务
tasks := make([]*model.StrmDownloadTask, 0, 200)
for i := 0; i < 200; i++ {
tasks = append(tasks, &model.StrmDownloadTask{
SyncPathID: "test-sync-path",
AccountID: "test-acct",
Provider: model.StrmProvider115,
FileName: filepath.Base(string(rune('a'+i%26))) + ".nfo",
LocalPath: filepath.Join(t.TempDir(), string(rune('a'+i%26)), "test.nfo"),
Status: model.StrmTaskPending,
})
}
if err := svc.repo.StrmDownload.CreateInBatches(ctx, tasks, 50); err != nil {
t.Fatalf("CreateInBatches failed: %v", err)
}
// 2. 验证 ActiveLocalPathMap
activeMap, err := svc.repo.StrmDownload.GetActiveLocalPathMap(ctx, "test-sync-path")
if err != nil {
t.Fatalf("GetActiveLocalPathMap failed: %v", err)
}
if len(activeMap) == 0 {
t.Fatal("expected active local path map to have entries")
}
// 3. 模拟 6 个 worker 并发 ClaimPendingDownload
claimedCount := 0
var claimMu sync.Mutex
var wg sync.WaitGroup
for w := 0; w < 6; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for {
batch, err := svc.repo.StrmDownload.ClaimPendingDownload(ctx, 10)
if err != nil {
t.Errorf("concurrent ClaimPendingDownload failed: %v", err)
return
}
if len(batch) == 0 {
return
}
claimMu.Lock()
claimedCount += len(batch)
claimMu.Unlock()
}
}()
}
wg.Wait()
if claimedCount != 200 {
t.Fatalf("expected all 200 tasks claimed, got %d", claimedCount)
}
}
+13 -5
View File
@@ -123,12 +123,20 @@ func LibraryVisibleForUser(ctx context.Context, repo *repository.Container, lib
return false
}
if repo != nil && repo.DB != nil {
var count int64
var totalCount int64
_ = repo.DB.WithContext(ctx).Model(&model.Media{}).
Where("library_id = ? AND nsfw = ?", lib.ID, true).
Count(&count).Error
if count > 0 {
return false
Where("library_id = ?", lib.ID).
Count(&totalCount).Error
if totalCount > 0 {
var nsfwCount int64
_ = repo.DB.WithContext(ctx).Model(&model.Media{}).
Where("library_id = ? AND nsfw = ?", lib.ID, true).
Count(&nsfwCount).Error
// 仅当整库媒体全部为成人内容(纯成人库)时才隐藏整库;
// 含有普通内容的混合媒体库保持库本身可见,具体 NSFW 条目在媒体列表内过滤。
if nsfwCount == totalCount {
return false
}
}
}
return true
+2 -1
View File
@@ -107,7 +107,8 @@ export const strmAPI = {
deletePath: (id: string) => api.delete(`/admin/strm/paths/${id}`).then((r) => r.data),
startSync: (id: string) => api.post(`/admin/strm/paths/${id}/sync`).then((r) => r.data),
startSync: (id: string, mode: 'incremental' | 'full' = 'incremental') =>
api.post(`/admin/strm/paths/${id}/sync`, null, { params: { mode } }).then((r) => r.data),
cancelSync: (id: string) => api.post(`/admin/strm/paths/${id}/cancel`).then((r) => r.data),
+3
View File
@@ -128,6 +128,9 @@ export function DanmakuStage({
const loadDanmaku = async () => {
let loadedInfo: DanmakuLoadedInfo | null = null
comments = []
nextIndex = 0
manager.clear()
try {
const res = await danmakuAPI.fetch(media.id, {
kw: search ?? undefined,
+123 -32
View File
@@ -16,11 +16,7 @@ import type { SubtitleTrack } from '../api/subtitles'
// PlayerControls — custom bottom control bar replacing the native <video
// controls> (which cannot host custom buttons). The danmaku toggle sits right
// next to the volume control. The bar auto-hides while playing and reappears
// on mouse movement; it stays visible while paused.
//
// Native keyboard shortcuts (space / arrows) still work because they are
// element-level defaults on <video>. Subtitles from <track> elements keep
// rendering; the CC button opens a track picker (关闭 / 各轨道).
// on mouse movement; it stays visible while paused or when hovering/interacting.
function formatTime(s: number): string {
if (!Number.isFinite(s) || s < 0) s = 0
@@ -63,9 +59,32 @@ export function PlayerControls({
const [fullscreen, setFullscreen] = useState(false)
const [pip, setPip] = useState(false)
const [uiVisible, setUiVisible] = useState(true)
const [controlsHovered, setControlsHovered] = useState(false)
const [isScrubbing, setIsScrubbing] = useState(false)
const [scrubValue, setScrubValue] = useState<number | null>(null)
const [subtitleMenuOpen, setSubtitleMenuOpen] = useState(false)
const subtitleMenuRef = useRef<HTMLDivElement | null>(null)
const hideTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null)
const controlsHoveredRef = useRef(false)
const isScrubbingRef = useRef(false)
const subtitleMenuOpenRef = useRef(false)
const danmakuOpenRef = useRef(false)
useEffect(() => {
controlsHoveredRef.current = controlsHovered
}, [controlsHovered])
useEffect(() => {
isScrubbingRef.current = isScrubbing
}, [isScrubbing])
useEffect(() => {
subtitleMenuOpenRef.current = subtitleMenuOpen
}, [subtitleMenuOpen])
useEffect(() => {
danmakuOpenRef.current = danmakuOpen
}, [danmakuOpen])
// 点击控制栏外部时关闭字幕菜单
useEffect(() => {
@@ -79,42 +98,74 @@ export function PlayerControls({
return () => document.removeEventListener('mousedown', onDocClick)
}, [subtitleMenuOpen])
// 播放时 3 秒无操作自动隐藏控制栏;暂停时保持显示。监听挂在视频容器上,
// 控制栏隐藏(pointer-events-none)后移动鼠标仍能重新唤起。
// 播放时 3 秒无操作自动隐藏控制栏;暂停/悬停/拖动进度条/打开菜单时保持显示。
// 监听挂在整个播放器舞台容器(data-player-stage)上,避免光标移到控制栏时因离开视频画面而误触发 mouseleave。
useEffect(() => {
const el = video()
if (!el) return
const parent = el.parentElement
const onMove = () => {
setUiVisible(true)
const stage = container()
if (!el || !stage) return
const resetTimer = () => {
if (hideTimerRef.current) clearTimeout(hideTimerRef.current)
if (!el.paused) {
hideTimerRef.current = setTimeout(() => setUiVisible(false), 3000)
if (
!el.paused &&
!controlsHoveredRef.current &&
!isScrubbingRef.current &&
!subtitleMenuOpenRef.current &&
!danmakuOpenRef.current
) {
hideTimerRef.current = setTimeout(() => {
if (
!controlsHoveredRef.current &&
!isScrubbingRef.current &&
!subtitleMenuOpenRef.current &&
!danmakuOpenRef.current
) {
setUiVisible(false)
}
}, 3000)
}
}
const onLeave = () => {
if (el.paused) return
const onMove = () => {
setUiVisible(true)
resetTimer()
}
const onLeave = (e: MouseEvent) => {
// 仅当光标真正移出 stage 容器时才处理
if (e.relatedTarget && stage.contains(e.relatedTarget as Node)) {
return
}
if (el.paused || controlsHoveredRef.current || isScrubbingRef.current) return
if (hideTimerRef.current) clearTimeout(hideTimerRef.current)
setUiVisible(false)
}
const syncPlay = () => {
setPlaying(!el.paused)
onMove()
}
const syncTime = () => setCurrentTime(el.currentTime)
const syncTime = () => {
if (!isScrubbingRef.current) {
setCurrentTime(el.currentTime)
}
}
const syncMeta = () => {
setDuration(el.duration || 0)
setCurrentTime(el.currentTime)
if (!isScrubbingRef.current) {
setCurrentTime(el.currentTime)
}
}
const syncVolume = () => {
setVolume(el.volume)
setMuted(el.muted)
}
const syncFullscreen = () => setFullscreen(document.fullscreenElement === parent)
const syncFullscreen = () => setFullscreen(Boolean(document.fullscreenElement))
const syncPip = () => setPip(document.pictureInPictureElement === el)
parent?.addEventListener('mousemove', onMove)
parent?.addEventListener('mouseleave', onLeave)
stage.addEventListener('mousemove', onMove)
stage.addEventListener('mouseleave', onLeave)
el.addEventListener('play', syncPlay)
el.addEventListener('playing', syncPlay)
@@ -131,8 +182,8 @@ export function PlayerControls({
syncVolume()
syncFullscreen()
return () => {
parent?.removeEventListener('mousemove', onMove)
parent?.removeEventListener('mouseleave', onLeave)
stage.removeEventListener('mousemove', onMove)
stage.removeEventListener('mouseleave', onLeave)
el.removeEventListener('play', syncPlay)
el.removeEventListener('playing', syncPlay)
el.removeEventListener('pause', syncPlay)
@@ -148,6 +199,20 @@ export function PlayerControls({
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [videoRef])
// 当悬停或菜单状态改变时,更新控制栏计时器
useEffect(() => {
if (controlsHovered || isScrubbing || subtitleMenuOpen || danmakuOpen) {
setUiVisible(true)
if (hideTimerRef.current) clearTimeout(hideTimerRef.current)
} else {
const el = video()
if (el && !el.paused) {
if (hideTimerRef.current) clearTimeout(hideTimerRef.current)
hideTimerRef.current = setTimeout(() => setUiVisible(false), 3000)
}
}
}, [controlsHovered, isScrubbing, subtitleMenuOpen, danmakuOpen])
const togglePlay = () => {
const el = video()
if (!el) return
@@ -155,11 +220,29 @@ export function PlayerControls({
else el.pause()
}
const seek = (v: number) => {
const el = video()
if (!el) return
el.currentTime = v
const handleSeekChange = (v: number) => {
setScrubValue(v)
setCurrentTime(v)
const el = video()
if (el && !isScrubbing) {
el.currentTime = v
}
}
const handleSeekStart = () => {
setIsScrubbing(true)
setUiVisible(true)
if (hideTimerRef.current) clearTimeout(hideTimerRef.current)
}
const handleSeekEnd = (v: number) => {
const el = video()
if (el) {
el.currentTime = v
setCurrentTime(v)
}
setIsScrubbing(false)
setScrubValue(null)
}
const changeVolume = (v: number) => {
@@ -202,14 +285,18 @@ export function PlayerControls({
'pictureInPictureEnabled' in document &&
document.pictureInPictureEnabled
const displayTime = isScrubbing && scrubValue !== null ? scrubValue : currentTime
return (
<div
className={`pointer-events-auto absolute inset-x-0 bottom-0 z-20 px-3 pb-2 pt-12 transition-opacity duration-300 ${
className={`pointer-events-auto absolute inset-x-0 bottom-0 z-20 bg-gradient-to-t from-black/80 via-black/40 to-transparent px-3 pb-3 pt-14 transition-opacity duration-300 ${
uiVisible ? 'opacity-100' : 'pointer-events-none opacity-0'
}`}
onMouseEnter={() => setControlsHovered(true)}
onMouseLeave={() => setControlsHovered(false)}
onClick={(e) => e.stopPropagation()}
>
<div className="flex items-center gap-2 text-white">
<div className="flex items-center gap-2.5 text-white">
<button
onClick={togglePlay}
className="rounded-full p-1.5 transition hover:bg-white/15"
@@ -223,14 +310,18 @@ export function PlayerControls({
min={0}
max={duration || 0}
step={0.1}
value={currentTime}
onChange={(e) => seek(Number(e.target.value))}
className="min-w-0 flex-1 accent-rose-500"
value={displayTime}
onMouseDown={handleSeekStart}
onTouchStart={handleSeekStart}
onChange={(e) => handleSeekChange(Number(e.target.value))}
onMouseUp={(e) => handleSeekEnd(Number((e.target as HTMLInputElement).value))}
onTouchEnd={(e) => handleSeekEnd(Number((e.target as HTMLInputElement).value))}
className="min-w-0 flex-1 cursor-pointer accent-rose-500"
aria-label="播放进度"
/>
<span className="shrink-0 font-mono text-xs tabular-nums text-white/85">
{formatTime(currentTime)} / {formatTime(duration)}
{formatTime(displayTime)} / {formatTime(duration)}
</span>
{pipSupported && (
+36 -30
View File
@@ -1,7 +1,8 @@
import { useEffect, useMemo, useState } from 'react'
import { useCallback, useEffect, useMemo, useState } from 'react'
import { libraryAPI } from '../api/library'
import { toolsAPI } from '../api/tools'
import { openManageLibrariesDialog } from '../components/manageLibrariesDialog'
import {
LibrariesContent,
LibrariesEmptyState,
@@ -16,6 +17,32 @@ export function LibrariesPage() {
const [repairEpisodeArtwork, setRepairEpisodeArtwork] = useState(false)
const [repairMsg, setRepairMsg] = useState('')
const loadLibraries = useCallback(async () => {
setLoading(true)
try {
const libs = await libraryAPI.list()
const rows = await Promise.all(libs.map(async (library) => {
try {
if (isSeriesLibraryType(library.type)) {
const [seriesPage, mediaPage] = await Promise.all([
libraryAPI.listSeries(library.id, 1, 10),
libraryAPI.listMedia(library.id, 1, 1, { groupVersions: false }),
])
return { library, items: [], total: mediaPage.total, cards: seriesPage.items ?? [] } satisfies LibraryPreview
}
const page = await libraryAPI.listMedia(library.id, 1, 160, { groupVersions: false })
const cards = latestLibraryCards(page.items)
return { library, items: page.items, total: page.total, cards } satisfies LibraryPreview
} catch {
return { library, items: [], total: 0, cards: [] } satisfies LibraryPreview
}
}))
setPreviews(rows)
} finally {
setLoading(false)
}
}, [])
async function handleRepairRescrape() {
if (repairing) return
setRepairing(true)
@@ -30,36 +57,14 @@ export function LibrariesPage() {
}
}
const handleManageLibraries = async () => {
await openManageLibrariesDialog()
await loadLibraries()
}
useEffect(() => {
let cancelled = false
async function load() {
setLoading(true)
try {
const libs = await libraryAPI.list()
const rows = await Promise.all(libs.map(async (library) => {
try {
if (isSeriesLibraryType(library.type)) {
const [seriesPage, mediaPage] = await Promise.all([
libraryAPI.listSeries(library.id, 1, 10),
libraryAPI.listMedia(library.id, 1, 1, { groupVersions: false }),
])
return { library, items: [], total: mediaPage.total, cards: seriesPage.items ?? [] } satisfies LibraryPreview
}
const page = await libraryAPI.listMedia(library.id, 1, 160, { groupVersions: false })
const cards = latestLibraryCards(page.items)
return { library, items: page.items, total: page.total, cards } satisfies LibraryPreview
} catch {
return { library, items: [], total: 0, cards: [] } satisfies LibraryPreview
}
}))
if (!cancelled) setPreviews(rows)
} finally {
if (!cancelled) setLoading(false)
}
}
load()
return () => { cancelled = true }
}, [])
loadLibraries().catch(() => undefined)
}, [loadLibraries])
const total = useMemo(() => previews.reduce((sum, preview) => sum + preview.total, 0), [previews])
@@ -77,6 +82,7 @@ export function LibrariesPage() {
repairing={repairing}
onRepairEpisodeArtworkChange={setRepairEpisodeArtwork}
onRepairRescrape={handleRepairRescrape}
onManageLibraries={handleManageLibraries}
/>
{previews.length === 0 ? (
+3 -2
View File
@@ -6,7 +6,6 @@ import { ArrowRight, Film, FolderOpen, Library as LibraryIcon, Music, PlayCircle
import { imageURL } from '../api/client'
import { EpisodeArtworkToggle } from '../components/EpisodeArtworkToggle'
import { MediaCard } from '../components/MediaCard'
import { openManageLibrariesDialog } from '../components/manageLibrariesDialog'
import { seriesCardLink } from '../utils/groupSeries'
import { libraryDisplayPath } from './libraryDisplayModel'
import { libraryArtworkItems, type LibraryPreview } from './librariesPageModel'
@@ -37,6 +36,7 @@ export function LibrariesHeader({
repairing,
onRepairEpisodeArtworkChange,
onRepairRescrape,
onManageLibraries,
}: {
previewCount: number
total: number
@@ -45,6 +45,7 @@ export function LibrariesHeader({
repairing: boolean
onRepairEpisodeArtworkChange: (value: boolean) => void
onRepairRescrape: () => void
onManageLibraries: () => void
}) {
return (
<div className="flex flex-wrap items-end justify-between gap-4">
@@ -72,7 +73,7 @@ export function LibrariesHeader({
<RefreshCw size={14} className={repairing ? 'animate-spin' : ''} />
{repairing ? '正在启动…' : '全库修复+重刮'}
</button>
<button type="button" onClick={() => openManageLibrariesDialog()} className="btn-outline">
<button type="button" onClick={onManageLibraries} className="btn-outline">
管理媒体库
</button>
</div>
+11
View File
@@ -168,6 +168,17 @@ export function PlayerPage() {
setDanmakuInfo(null)
}, [])
// 切换视频时重置媒体与弹幕状态,确保新视频自动重新识别并加载弹幕
useEffect(() => {
setMedia(null)
setDanmakuEpisodeId(null)
setDanmakuCandidates([])
setDanmakuSearch(null)
setDanmakuSelectedSource('')
setDanmakuInfo(null)
setDanmakuSearching(true)
}, [id])
// Load metadata and pick a default mode.
useEffect(() => {
if (!id) return
+25 -27
View File
@@ -115,12 +115,6 @@ export function PlayerVideoStage({
return
}
const timers = new Set<ReturnType<typeof setTimeout>>()
const delay = (fn: () => void, ms: number) => {
const id = setTimeout(fn, ms)
timers.add(id)
}
const updateCue = () => {
const trackEls = Array.from(video.querySelectorAll('track'))
const selectedEl = trackEls[subtitleIndex]
@@ -130,16 +124,24 @@ export function PlayerVideoStage({
return
}
// 优先从浏览器 activeCues 中取当前文本;若浏览器在 hidden 模式下延迟触发 cuechange,
// 则从 tt.cues 中根据 video.currentTime 实时匹配当前字幕,确保初次加载无感立即可见。
const texts: string[] = []
if (tt.activeCues && tt.activeCues.length > 0) {
const texts: string[] = []
for (let i = 0; i < tt.activeCues.length; i++) {
const cue = tt.activeCues[i] as VTTCue
if (cue && cue.text) texts.push(cue.text)
}
setActiveCueText(texts.join('\n'))
} else {
setActiveCueText('')
} else if (tt.cues && tt.cues.length > 0) {
const cur = video.currentTime
for (let i = 0; i < tt.cues.length; i++) {
const cue = tt.cues[i] as VTTCue
if (cue && cur >= cue.startTime && cur <= cue.endTime && cue.text) {
texts.push(cue.text)
}
}
}
setActiveCueText(texts.join('\n'))
}
const apply = () => {
@@ -162,18 +164,8 @@ export function PlayerVideoStage({
tt.addEventListener('cuechange', updateCue)
}
// 竞态兜底重载检测
delay(() => {
const curTt = selected.track
if (curTt && curTt.mode === 'hidden' && (!curTt.cues || curTt.cues.length === 0)) {
const src = selected.getAttribute('src')
if (src) {
selected.setAttribute('src', '')
selected.setAttribute('src', src)
curTt.mode = 'hidden'
}
}
}, 1200)
selected.removeEventListener('load', updateCue)
selected.addEventListener('load', updateCue)
updateCue()
}
@@ -183,20 +175,24 @@ export function PlayerVideoStage({
video.addEventListener('timeupdate', updateCue)
video.addEventListener('seeking', updateCue)
video.addEventListener('seeked', updateCue)
video.addEventListener('playing', updateCue)
return () => {
video.removeEventListener('loadedmetadata', apply)
video.removeEventListener('timeupdate', updateCue)
video.removeEventListener('seeking', updateCue)
video.removeEventListener('seeked', updateCue)
timers.forEach(clearTimeout)
video.removeEventListener('playing', updateCue)
const trackEls = Array.from(video.querySelectorAll('track'))
const selected = trackEls[subtitleIndex]
if (selected?.track) {
selected.track.removeEventListener('cuechange', updateCue)
if (selected) {
selected.removeEventListener('load', updateCue)
if (selected.track) {
selected.track.removeEventListener('cuechange', updateCue)
}
}
}
}, [subtitleIndex, subs, videoRef])
}, [subtitleIndex, subs, videoRef, media])
// 根据视频画面宽高比与舞台宽高比,确定视频在哪个轴向撑满 100%
const isWiderThanStage =
@@ -238,17 +234,19 @@ export function PlayerVideoStage({
className="h-full w-full object-contain bg-black"
onError={onVideoError}
>
{subs.map((track) => (
{subs.map((track, index) => (
<track
key={track.path}
kind="subtitles"
src={subtitlesAPI.url(media.id, track.path)}
srcLang={track.lang}
label={track.label || track.lang}
default={index === subtitleIndex}
/>
))}
</video>
<DanmakuStage
key={media.id}
media={media}
videoRef={videoRef}
enabled={danmakuEnabled}
+9 -2
View File
@@ -674,6 +674,7 @@ export function StrmSyncPathDialog({
delete_dir: existing?.delete_dir ?? false,
cron: existing?.cron ?? '',
enable_cron: existing?.enable_cron ?? false,
sync_mode: existing?.sync_mode ?? 'incremental',
enabled: existing?.enabled ?? true,
}))
const [saving, setSaving] = useState(false)
@@ -831,7 +832,7 @@ export function StrmSyncPathDialog({
<input className={inputCls} value={form.exclude_name ?? ''} placeholder="sample,trailer" onChange={(e) => set('exclude_name', e.target.value)} />
</Field>
</div>
<div className="grid gap-3 md:grid-cols-2">
<div className="grid gap-3 md:grid-cols-3">
<Field label="STRM 链接 path 参数">
<select className={inputCls} value={form.add_path ?? 1} onChange={(e) => set('add_path', Number(e.target.value))}>
<option value={1}>完整远端路径</option>
@@ -839,7 +840,13 @@ export function StrmSyncPathDialog({
<option value={3}>不带 path</option>
</select>
</Field>
<Field label="定时同步 Cron" hint="5 段表达式,如 0 */6 * * *(每 6 小时)">
<Field label="默认同步模式" hint="定时触发或快速同步时的策略">
<select className={inputCls} value={form.sync_mode ?? 'incremental'} onChange={(e) => set('sync_mode', e.target.value as 'incremental' | 'full')}>
<option value="incremental">增量同步(快速)</option>
<option value="full">全量同步(全量校验)</option>
</select>
</Field>
<Field label="定时同步 Cron" hint="5 段表达式,如 0 */6 * * *">
<input className={inputCls} value={form.cron ?? ''} placeholder="0 */6 * * *" onChange={(e) => set('cron', e.target.value)} />
</Field>
</div>
+36 -14
View File
@@ -96,11 +96,11 @@ export function StrmManagePage() {
return () => clearInterval(timer)
}, [paths, refresh])
const startSync = async (path: StrmSyncPath) => {
const startSync = async (path: StrmSyncPath, mode: 'incremental' | 'full' = 'incremental') => {
setActingPath(path.id)
try {
await strmAPI.startSync(path.id)
toast.success(`已开始同步「${path.name}」`)
await strmAPI.startSync(path.id, mode)
toast.success(`已开始${mode === 'full' ? '全量' : '增量'}同步「${path.name}」`)
await refresh()
} catch (err) {
toast.error(apiErrorMessage(err))
@@ -344,7 +344,7 @@ function SyncPathSection({
onAdd: () => void
onEdit: (path: StrmSyncPath) => void
onDelete: (path: StrmSyncPath) => void
onStart: (path: StrmSyncPath) => void
onStart: (path: StrmSyncPath, mode?: 'incremental' | 'full') => void
onCancel: (path: StrmSyncPath) => void
}) {
return (
@@ -401,15 +401,28 @@ function SyncPathSection({
取消
</button>
) : (
<button
type="button"
disabled={actingPath === path.id || !path.enabled}
onClick={() => onStart(path)}
className={`${iconButtonCls} disabled:opacity-40`}
>
{actingPath === path.id ? <Loader2 size={14} className="animate-spin" /> : <Play size={14} />}
立即同步
</button>
<>
<button
type="button"
disabled={actingPath === path.id || !path.enabled}
onClick={() => onStart(path, 'incremental')}
className={`${iconButtonCls} text-brand-600 font-medium disabled:opacity-40`}
title="增量同步:基于目录缓存快速同步新增与更新文件"
>
{actingPath === path.id ? <Loader2 size={14} className="animate-spin" /> : <Play size={14} />}
增量同步
</button>
<button
type="button"
disabled={actingPath === path.id || !path.enabled}
onClick={() => onStart(path, 'full')}
className={`${iconButtonCls} text-sand-600 disabled:opacity-40`}
title="全量同步:重置目录缓存并全量比对所有文件"
>
<RefreshCw size={14} />
全量同步
</button>
</>
)}
<button type="button" onClick={() => onEdit(path)} className={`${iconButtonCls}`}>
<Pencil size={14} />
@@ -456,9 +469,11 @@ function RecordSection({ records }: { records: StrmSyncRecord[] }) {
<thead className="border-b border-gray-200 text-xs uppercase tracking-wider text-sand-500">
<tr>
<th className="px-3 py-2">时间</th>
<th className="px-3 py-2">类型</th>
<th className="px-3 py-2">状态</th>
<th className="px-3 py-2 text-right">扫描文件</th>
<th className="px-3 py-2 text-right">新增 strm</th>
<th className="px-3 py-2 text-right">新增/更新</th>
<th className="px-3 py-2 text-right">跳过</th>
<th className="px-3 py-2 text-right">下载元数据</th>
<th className="px-3 py-2 text-right">清理</th>
<th className="px-3 py-2">说明</th>
@@ -467,11 +482,17 @@ function RecordSection({ records }: { records: StrmSyncRecord[] }) {
<tbody>
{records.map((record) => {
const meta = RECORD_STATUS_META[record.status] ?? RECORD_STATUS_META.pending
const isFull = record.sync_type === 'full'
return (
<tr key={record.id} className="border-t border-gray-100">
<td className="whitespace-nowrap px-3 py-2 text-xs text-ink-50">
{formatTime(record.started_at ?? record.created_at)}
</td>
<td className="px-3 py-2">
<span className={`rounded-full px-2 py-0.5 text-[11px] font-medium ${isFull ? 'bg-amber-50 text-amber-600 border border-amber-200' : 'bg-brand-50 text-brand-600 border border-brand-200'}`}>
{isFull ? '全量' : '增量'}
</span>
</td>
<td className="px-3 py-2">
<span className={'rounded-full px-2 py-0.5 text-[11px] font-semibold ' + meta.cls}>
{meta.label}
@@ -479,6 +500,7 @@ function RecordSection({ records }: { records: StrmSyncRecord[] }) {
</td>
<td className="px-3 py-2 text-right">{record.total}</td>
<td className="px-3 py-2 text-right text-brand-500">{record.new_strm}</td>
<td className="px-3 py-2 text-right text-gray-500">{record.skipped}</td>
<td className="px-3 py-2 text-right">{record.new_meta}</td>
<td className="px-3 py-2 text-right">{record.pruned}</td>
<td className="max-w-[260px] truncate px-3 py-2 text-xs text-sand-500">{record.message}</td>
+3
View File
@@ -49,6 +49,7 @@ export interface StrmSyncPath {
delete_dir: boolean
cron: string
enable_cron: boolean
sync_mode?: 'incremental' | 'full'
enabled: boolean
created_at: string
last_sync_at?: string | null
@@ -75,12 +76,14 @@ export interface StrmSyncPathInput {
delete_dir?: boolean
cron?: string
enable_cron?: boolean
sync_mode?: 'incremental' | 'full'
enabled?: boolean
}
export interface StrmSyncRecord {
id: string
sync_path_id: string
sync_type?: 'incremental' | 'full'
status: 'pending' | 'running' | 'done' | 'failed' | 'canceled'
total: number
new_strm: number