Compare commits

...

18 Commits

Author SHA1 Message Date
github-actions[bot] 1ea4724261 chore: bump version to 0.0.35 [skip ci] 2026-08-26 06:50:29 +00:00
truewhile 3d372f039e Merge branch 'main' of https://github.com/truewhile/MMTL 2026-08-26 14:50:10 +08:00
truewhile 0384017e98 6 2026-08-26 14:50:06 +08:00
github-actions[bot] 0332579d5f chore: bump version to 0.0.34 [skip ci] 2026-08-26 04:52:28 +00:00
truewhile 6aefe18caa 5 2026-08-26 12:52:10 +08:00
github-actions[bot] ef72fc8d83 chore: bump version to 0.0.33 [skip ci] 2026-08-26 04:12:13 +00:00
truewhile 4764c09572 4 2026-08-26 12:11:56 +08:00
github-actions[bot] 98ca766a37 chore: bump version to 0.0.32 [skip ci] 2026-08-26 03:38:14 +00:00
truewhile 13c9035b76 3 2026-08-26 11:37:58 +08:00
github-actions[bot] ad6d0ba21d chore: bump version to 0.0.31 [skip ci] 2026-08-26 03:19:09 +00:00
truewhile 431f7f088b 2 2026-08-26 11:18:53 +08:00
github-actions[bot] 3f13ed1113 chore: bump version to 0.0.30 [skip ci] 2026-08-26 01:51:54 +00:00
truewhile 9d359c40dd 1 2026-08-26 09:51:27 +08:00
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
40 changed files with 1561 additions and 403 deletions
+1 -1
View File
@@ -1 +1 @@
0.0.27
0.0.35
+3
View File
@@ -50,11 +50,14 @@ func registerAdminStrmRoutes(admin *gin.RouterGroup, svc *service.Container) {
admin.POST("/strm/downloads/:id/retry", retryStrmDownloadHandler(svc))
admin.POST("/strm/downloads/clear-done", clearDoneDownloadsHandler(svc))
admin.POST("/strm/downloads/clear-finished", clearFinishedDownloadsHandler(svc))
admin.POST("/strm/downloads/clear-canceled", clearCanceledDownloadsHandler(svc))
admin.POST("/strm/downloads/retry-failed", retryAllFailedDownloadsHandler(svc))
admin.POST("/strm/downloads/cancel-pending", cancelPendingDownloadsHandler(svc))
admin.GET("/strm/uploads", uploadQueueHandler(svc))
admin.POST("/strm/uploads/:id/cancel", cancelStrmUploadHandler(svc))
admin.POST("/strm/uploads/:id/retry", retryStrmUploadHandler(svc))
admin.POST("/strm/uploads/cancel-pending", cancelPendingUploadsHandler(svc))
admin.POST("/strm/uploads/clear-canceled", clearCanceledUploadsHandler(svc))
}
func registerAdminUserRoutes(admin *gin.RouterGroup, svc *service.Container) {
+45 -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
}
@@ -382,6 +392,28 @@ func clearFinishedDownloadsHandler(svc *service.Container) gin.HandlerFunc {
}
}
func clearCanceledDownloadsHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
n, err := svc.Strm.ClearCanceledDownloadTasks(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"deleted": n})
}
}
func clearCanceledUploadsHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
n, err := svc.Strm.ClearCanceledUploadTasks(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"deleted": n})
}
}
func retryAllFailedDownloadsHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
n, err := svc.Strm.RetryAllFailedDownloadTasks(c.Request.Context())
@@ -404,6 +436,17 @@ func cancelPendingDownloadsHandler(svc *service.Container) gin.HandlerFunc {
}
}
func cancelPendingUploadsHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
n, err := svc.Strm.CancelPendingUploadTasks(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"canceled": n})
}
}
// ─── 公开播放端点 ──────────────────────────────────────────────────────────────
// strmPlayHandler 处理 strm 文件指向的播放请求(Emby/Infuse 直接请求,无 JWT)。
@@ -458,6 +501,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),
}
}
+10 -4
View File
@@ -50,10 +50,16 @@ func TestStrmAdminRoutesAreRegistered(t *testing.T) {
"GET /api/admin/strm/downloads",
"POST /api/admin/strm/downloads/:id/cancel",
"POST /api/admin/strm/downloads/:id/retry",
"GET /api/admin/strm/uploads",
"POST /api/admin/strm/uploads/:id/cancel",
"POST /api/admin/strm/uploads/:id/retry",
"GET /api/strm/play/:provider/:file",
"POST /api/admin/strm/downloads/clear-finished",
"POST /api/admin/strm/downloads/clear-canceled",
"POST /api/admin/strm/downloads/retry-failed",
"POST /api/admin/strm/downloads/cancel-pending",
"GET /api/admin/strm/uploads",
"POST /api/admin/strm/uploads/:id/cancel",
"POST /api/admin/strm/uploads/:id/retry",
"POST /api/admin/strm/uploads/cancel-pending",
"POST /api/admin/strm/uploads/clear-canceled",
"GET /api/strm/play/:provider/:file",
} {
if !routes[want] {
t.Fatalf("%s route is not registered", want)
+1 -1
View File
@@ -163,7 +163,7 @@ func historyDeleteHandler(svc *service.Container) gin.HandlerFunc {
c.JSON(http.StatusBadRequest, gin.H{"error": "status must be completed or incomplete"})
return
}
res := q.Delete(&model.PlaybackHistory{})
res := q.Unscoped().Delete(&model.PlaybackHistory{})
if err := res.Error; err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
+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 -2
View File
@@ -62,9 +62,9 @@ func (r *ApiConfigRepository) Update(ctx context.Context, c *model.ApiConfig) er
}).Error
}
// Delete removes an API config.
// Delete 物理删除 API 配置。
func (r *ApiConfigRepository) Delete(ctx context.Context, provider string) error {
return r.db.WithContext(ctx).Where("provider = ?", provider).Delete(&model.ApiConfig{}).Error
return r.db.WithContext(ctx).Unscoped().Where("provider = ?", provider).Delete(&model.ApiConfig{}).Error
}
// UpdateTestResult 更新测试结果。
+1 -1
View File
@@ -23,7 +23,7 @@ func (r *FavoriteRepository) Toggle(ctx context.Context, userID, mediaID string)
if err != nil {
return false, err
}
return false, r.db.WithContext(ctx).Delete(&f).Error
return false, r.db.WithContext(ctx).Unscoped().Delete(&f).Error
}
// ListByUser returns all favourite media IDs for a user.
+3 -4
View File
@@ -79,10 +79,9 @@ func (r *LibraryRepository) FindByID(ctx context.Context, id string) (*model.Lib
return &l, nil
}
// Delete removes a library and (soft) cascades to its media via repository
// callers; we do not run CASCADE here to keep this method narrow.
// Delete 物理删除媒体库。
func (r *LibraryRepository) Delete(ctx context.Context, id string) error {
return r.db.WithContext(ctx).Delete(&model.Library{}, "id = ?", id).Error
return r.db.WithContext(ctx).Unscoped().Delete(&model.Library{}, "id = ?", id).Error
}
func (r *LibraryRepository) ListRoots(ctx context.Context, libraryID string) ([]model.LibraryRoot, error) {
@@ -149,7 +148,7 @@ func (r *LibraryRepository) DeleteRoot(ctx context.Context, libraryID, rootID st
if !r.hasLibraryRootsTable() {
return nil
}
return r.db.WithContext(ctx).Where("library_id = ?", libraryID).Delete(&model.LibraryRoot{}, "id = ?", rootID).Error
return r.db.WithContext(ctx).Unscoped().Where("library_id = ?", libraryID).Delete(&model.LibraryRoot{}, "id = ?", rootID).Error
}
func (r *LibraryRepository) hasLibraryRootsTable() bool {
+3 -3
View File
@@ -116,12 +116,12 @@ func (r *MediaRepository) ListByLibrariesFiltered(ctx context.Context, libraryID
// DeleteByLibrary purges all media tied to a library.
func (r *MediaRepository) DeleteByLibrary(ctx context.Context, libraryID string) error {
// FTS 行由 media 表上的触发器同步清理(软删/硬删都覆盖)。
return r.db.WithContext(ctx).Where("library_id = ?", libraryID).Delete(&model.Media{}).Error
// FTS 行由 media 表上的触发器同步清理(物理删除触发 FTS 清理)。
return r.db.WithContext(ctx).Unscoped().Where("library_id = ?", libraryID).Delete(&model.Media{}).Error
}
func (r *MediaRepository) DeleteByLibraryRoot(ctx context.Context, libraryID, rootID string) error {
return r.db.WithContext(ctx).
return r.db.WithContext(ctx).Unscoped().
Where("library_id = ? AND library_root_id = ?", libraryID, rootID).
Delete(&model.Media{}).Error
}
+2 -2
View File
@@ -51,9 +51,9 @@ func (r *PermissionRepository) Upsert(ctx context.Context, p *model.UserPermissi
})
}
// Delete removes a permission record.
// Delete 物理删除权限记录。
func (r *PermissionRepository) Delete(ctx context.Context, userID string) error {
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Where("user_id = ?", userID).Delete(&model.UserPermission{}).Error
return r.db.WithContext(ctx).Unscoped().Where("user_id = ?", userID).Delete(&model.UserPermission{}).Error
})
}
+2 -2
View File
@@ -59,9 +59,9 @@ func (r *PlayProfileRepository) Update(ctx context.Context, id string, patch map
Where("id = ?", id).Updates(patch).Error
}
// Delete soft-deletes a profile.
// Delete 物理删除播放档案。
func (r *PlayProfileRepository) Delete(ctx context.Context, id string) error {
return r.db.WithContext(ctx).Delete(&model.PlayProfile{}, "id = ?", id).Error
return r.db.WithContext(ctx).Unscoped().Delete(&model.PlayProfile{}, "id = ?", id).Error
}
// ClearDefaultsFor resets is_default for all of a user's profiles.
@@ -72,10 +72,10 @@ func (r *RefreshTokenRepository) RevokeOldestActiveByUserID(ctx context.Context,
})
}
// DeleteExpired removes all expired refresh tokens.
// DeleteExpired 物理清理所有过期的 refresh tokens。
func (r *RefreshTokenRepository) DeleteExpired(ctx context.Context) error {
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Where("expires_at < ?", time.Now()).Delete(&model.RefreshToken{}).Error
return r.db.WithContext(ctx).Unscoped().Where("expires_at < ?", time.Now()).Delete(&model.RefreshToken{}).Error
})
}
+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},
}
}
+2 -2
View File
@@ -29,9 +29,9 @@ func (r *SettingRepository) Set(ctx context.Context, key, value string) error {
return r.db.WithContext(ctx).Save(&s).Error
}
// Delete removes a setting key.
// Delete 物理删除设置键。
func (r *SettingRepository) Delete(ctx context.Context, key string) error {
return r.db.WithContext(ctx).Where("key = ?", key).Delete(&model.Setting{}).Error
return r.db.WithContext(ctx).Unscoped().Where("key = ?", key).Delete(&model.Setting{}).Error
}
// All returns every key/value pair (used by the admin UI).
+2 -2
View File
@@ -66,9 +66,9 @@ func (r *StorageConfigRepository) Upsert(ctx context.Context, c *model.StorageCo
}).Error
}
// Delete removes a storage config by ID.
// Delete 物理删除存储配置。
func (r *StorageConfigRepository) Delete(ctx context.Context, id string) error {
return r.db.WithContext(ctx).Where("id = ?", id).Delete(&model.StorageConfig{}).Error
return r.db.WithContext(ctx).Unscoped().Where("id = ?", id).Delete(&model.StorageConfig{}).Error
}
// FindByID returns a storage config by ID.
+346 -141
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).Unscoped().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).Unscoped().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,97 @@ 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).Unscoped().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).Unscoped().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).Unscoped().Where("status IN ?", []string{model.StrmTaskDone, model.StrmTaskFailed, model.StrmTaskCanceled}).
Delete(&model.StrmDownloadTask{})
count = res.RowsAffected
return res.Error
})
return count, err
}
// ClearCanceled 清空全部已取消下载任务。
func (r *StrmDownloadTaskRepository) ClearCanceled(ctx context.Context) (int64, error) {
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Unscoped().Where("status = ?", model.StrmTaskCanceled).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 批量取消所有排队中的任务。
// 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 IN ?", []string{model.StrmTaskPending, model.StrmTaskRunning}).
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 +374,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).Unscoped().Where("status IN ? AND finished_at < ?",
[]string{model.StrmTaskDone, model.StrmTaskFailed, model.StrmTaskCanceled}, before).
Delete(&model.StrmDownloadTask{}).Error
})
}
// ─── StrmUploadTask ────────────────────────────────────────────────────────────
@@ -311,7 +404,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 +487,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 +518,53 @@ 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).Unscoped().Where("id = ?", id).Delete(&model.StrmUploadTask{}).Error
})
}
// ClearCanceled 清空全部已取消上传任务。
func (r *StrmUploadTaskRepository) ClearCanceled(ctx context.Context) (int64, error) {
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Unscoped().Where("status = ?", model.StrmTaskCanceled).Delete(&model.StrmUploadTask{})
count = res.RowsAffected
return res.Error
})
return count, err
}
// CancelPending 批量取消所有排队中和进行中的任务。
func (r *StrmUploadTaskRepository) CancelPending(ctx context.Context) (int64, error) {
now := time.Now()
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Model(&model.StrmUploadTask{}).
Where("status IN ?", []string{model.StrmTaskPending, model.StrmTaskRunning}).
Updates(map[string]any{
"status": model.StrmTaskCanceled,
"error": "已批量取消",
"finished_at": now,
"updated_at": now,
})
count = res.RowsAffected
return res.Error
})
return count, err
}
// CountActive 统计某同步目录下目标仍在排队/进行的任务数(用于去重)。
@@ -431,8 +577,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).Unscoped().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).Unscoped().Where("sync_path_id = ?", syncPathID).Delete(&model.StrmDirCache{}).Error
})
}
+11 -20
View File
@@ -133,26 +133,17 @@ func (r *UserRepository) TouchLogin(ctx context.Context, id string) error {
})
}
// Delete removes a user (soft-delete via gorm.DeletedAt), releases the unique
// username, and drops Telegram bindings so future re-created users bind cleanly.
// Delete 物理删除用户并级联清理其关联记录。
func (r *UserRepository) Delete(ctx context.Context, id string) error {
return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var user model.User
if err := tx.Where("id = ?", id).First(&user).Error; err != nil {
return err
}
released := user.Username + "__deleted__" + time.Now().Format("20060102150405.000000000")
if len(released) > 64 {
sum := sha256.Sum256([]byte(user.ID + user.Username))
base := user.Username
if len(base) > 43 {
base = base[:43]
}
released = base + "__deleted__" + hex.EncodeToString(sum[:])[:10]
}
if err := tx.Model(&model.User{}).Where("id = ?", id).Update("username", released).Error; err != nil {
return err
}
return tx.Delete(&model.User{}, "id = ?", id).Error
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
_ = tx.Unscoped().Where("user_id = ?", id).Delete(&model.RefreshToken{})
_ = tx.Unscoped().Where("user_id = ?", id).Delete(&model.UserPermission{})
_ = tx.Unscoped().Where("user_id = ?", id).Delete(&model.PlayProfile{})
_ = tx.Unscoped().Where("user_id = ?", id).Delete(&model.PlaybackHistory{})
_ = tx.Unscoped().Where("user_id = ?", id).Delete(&model.Favorite{})
_ = tx.Unscoped().Where("user_id = ?", id).Delete(&model.UserDevice{})
return tx.Unscoped().Delete(&model.User{}, "id = ?", id).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:"-"` // 原始响应体(外层附加字段用)
}
+58 -3
View File
@@ -371,7 +371,62 @@ func TestThrottleCodeHandling(t *testing.T) {
if err == nil {
t.Fatal("expected throttle error")
}
if !tm.IsThrottled() {
t.Fatal("code 770004 should trigger throttle status")
if !tm.IsThrottled() {
t.Fatal("code 770004 should trigger throttle status")
}
}
}
func TestRemoteFileDetailRelativePath(t *testing.T) {
rootCID := "3238787832374488117" // 影视库
// 场景 1:115 目录 paths 中只有祖先目录链,不包含自身
d1 := &RemoteFileDetail{
FileId: "3251154147730910635",
FileName: "出包王女",
Paths: []struct {
FileId string `json:"file_id"`
Name string `json:"file_name"`
}{
{FileId: "0", Name: "根目录"},
{FileId: "3238787832374488117", Name: "影视库"},
{FileId: "3238787913223892116", Name: "动漫"},
},
}
if got := d1.RelativePath(rootCID); got != "动漫/出包王女" {
t.Errorf("d1.RelativePath = %q, want %q", got, "动漫/出包王女")
}
// 场景 2:祖先中间目录,自身在 paths 末尾
d2 := &RemoteFileDetail{
FileId: "3238787913223892116",
FileName: "动漫",
Paths: []struct {
FileId string `json:"file_id"`
Name string `json:"file_name"`
}{
{FileId: "0", Name: "根目录"},
{FileId: "3238787832374488117", Name: "影视库"},
{FileId: "3238787913223892116", Name: "动漫"},
},
}
if got := d2.RelativePath(rootCID); got != "动漫" {
t.Errorf("d2.RelativePath = %q, want %q", got, "动漫")
}
// 场景 3:根同步目录自身
d3 := &RemoteFileDetail{
FileId: rootCID,
FileName: "影视库",
Paths: []struct {
FileId string `json:"file_id"`
Name string `json:"file_name"`
}{
{FileId: "0", Name: "根目录"},
{FileId: rootCID, Name: "影视库"},
},
}
if got := d3.RelativePath(rootCID); got != "" {
t.Errorf("d3.RelativePath = %q, want %q", got, "")
}
}
+73
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,49 @@ type RemoteFileDetail struct {
} `json:"paths"`
}
// RelativePath 计算该目录相对于根同步目录(rootCID)的相对路径。
func (d *RemoteFileDetail) RelativePath(rootCID string) string {
if d == nil {
return ""
}
if rootCID == "" {
rootCID = "0"
}
if d.FileId == rootCID {
return ""
}
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
}
hasSelf := false
for i := start; i < len(d.Paths); i++ {
if d.Paths[i].FileId == d.FileId {
hasSelf = true
}
name := strings.TrimSpace(d.Paths[i].Name)
if name != "" {
segments = append(segments, name)
}
}
// 若 115 返回的 paths 祖先链未包含当前目录自身,则将其自身目录名 FileName 补在末尾
if !hasSelf && strings.TrimSpace(d.FileName) != "" && d.FileId != rootCID {
segments = append(segments, strings.TrimSpace(d.FileName))
}
return strings.Join(segments, "/")
}
// ─── 下载直链 ──────────────────────────────────────────────────────────────────
type downloadURLData struct {
+4 -4
View File
@@ -43,10 +43,10 @@ func (s *MediaService) DeleteLibrary(ctx context.Context, id string) error {
if err := tx.Unscoped().Where("library_id = ?", id).Delete(&model.Media{}).Error; err != nil {
return err
}
if err := hardDeleteLibraryRoots(ctx, tx, id); err != nil {
return err
}
return tx.Delete(&model.Library{}, "id = ?", id).Error
if err := hardDeleteLibraryRoots(ctx, tx, id); err != nil {
return err
}
return tx.Unscoped().Delete(&model.Library{}, "id = ?", id).Error
})
if err == nil {
s.invalidateMediaCache(ctx)
+2 -17
View File
@@ -10,25 +10,10 @@ import (
const maxRecycleBinRecords = 200
// SoftDelete moves a media row to the recycle bin (gorm soft delete).
// The on-disk file is kept; admins can purge it later.
// SoftDelete 物理删除媒体记录(统一硬删除以降低 SQLite 存储与索引压力)。
func (s *MediaService) SoftDelete(ctx context.Context, id string) error {
media, err := s.repo.Media.FindByID(ctx, id)
if err != nil {
return err
}
if media != nil && isCloudMediaPath(media.Path) {
err := s.repo.DB.WithContext(ctx).Unscoped().Where("id = ?", id).Delete(&model.Media{}).Error
if err == nil {
s.invalidateMediaCache(ctx)
}
return err
}
err = s.repo.DB.WithContext(ctx).Where("id = ?", id).Delete(&model.Media{}).Error
err := s.repo.DB.WithContext(ctx).Unscoped().Where("id = ?", id).Delete(&model.Media{}).Error
if err == nil {
if pruneErr := pruneRecycleBinRows(ctx, s.repo.DB, maxRecycleBinRecords); pruneErr != nil {
return pruneErr
}
s.invalidateMediaCache(ctx)
}
return err
@@ -229,9 +229,9 @@ func (o *OrganizerService) replaceVersions(ctx context.Context, src string, exis
o.log.Warn("organize replace remove existing failed",
zap.String("path", e), zap.Error(err))
}
if o.repo != nil && o.repo.DB != nil {
_ = o.repo.DB.WithContext(ctx).Where("path = ?", e).Delete(&model.Media{}).Error
}
if o.repo != nil && o.repo.DB != nil {
_ = o.repo.DB.WithContext(ctx).Unscoped().Where("path = ?", e).Delete(&model.Media{}).Error
}
}
// Move staged file + sidecars into the final path.
if err := os.Rename(stage, dst); err != nil {
@@ -97,7 +97,7 @@ func (o *OrganizerService) deleteMediaRowForPath(ctx context.Context, path strin
if o == nil || o.repo == nil || o.repo.DB == nil {
return
}
_ = o.repo.DB.WithContext(ctx).Where("path = ?", path).Delete(&model.Media{}).Error
_ = o.repo.DB.WithContext(ctx).Unscoped().Where("path = ?", path).Delete(&model.Media{}).Error
}
func (o *OrganizerService) mediaPathExists(ctx context.Context, path string) bool {
+5 -5
View File
@@ -196,18 +196,18 @@ func (p *PlaybackService) AddToPlaylist(ctx context.Context, playlistID, mediaID
return p.repo.DB.Create(item).Error
}
// RemoveFromPlaylist removes a media item from a playlist (idempotent).
// RemoveFromPlaylist 物理删除播放列表项(幂等)。
func (p *PlaybackService) RemoveFromPlaylist(ctx context.Context, playlistID, mediaID string) error {
return p.repo.DB.
return p.repo.DB.WithContext(ctx).Unscoped().
Where("playlist_id = ? AND media_id = ?", playlistID, mediaID).
Delete(&model.PlaylistItem{}).Error
}
// DeletePlaylist removes a playlist and all of its items.
// DeletePlaylist 物理删除播放列表及其全部条目。
func (p *PlaybackService) DeletePlaylist(ctx context.Context, playlistID string) error {
if err := p.repo.DB.Where("playlist_id = ?", playlistID).
if err := p.repo.DB.WithContext(ctx).Unscoped().Where("playlist_id = ?", playlistID).
Delete(&model.PlaylistItem{}).Error; err != nil {
return err
}
return p.repo.DB.Where("id = ?", playlistID).Delete(&model.Playlist{}).Error
return p.repo.DB.WithContext(ctx).Unscoped().Where("id = ?", playlistID).Delete(&model.Playlist{}).Error
}
+4 -5
View File
@@ -11,13 +11,12 @@ import (
"github.com/ShukeBta/MMTL/internal/model"
)
// RemovePath deletes the media row for a path that has disappeared from disk
// (incremental delete used by the watcher on Remove/Rename events).
// RemovePath 物理删除磁盘上已不存在的媒体记录。
func (s *ScannerService) RemovePath(ctx context.Context, path string) (int64, error) {
if _, err := os.Stat(path); err == nil {
return 0, nil // still exists; nothing to remove
}
res := s.repo.DB.WithContext(ctx).
res := s.repo.DB.WithContext(ctx).Unscoped().
Where("path = ?", path).
Delete(&model.Media{})
if res.Error == nil && res.RowsAffected > 0 {
@@ -55,7 +54,7 @@ func (s *ScannerService) pruneMissingMedia(ctx context.Context, libraryID string
}
stale = append(stale, row.ID)
}
return s.deleteMediaByIDs(ctx, stale, false)
return s.deleteMediaByIDs(ctx, stale, true)
}
func (s *ScannerService) pruneMissingMediaForRoot(ctx context.Context, libraryID, rootID, rootPath string, seen map[string]struct{}) (int64, error) {
@@ -92,7 +91,7 @@ func (s *ScannerService) pruneMissingMediaForRoot(ctx context.Context, libraryID
}
stale = append(stale, row.ID)
}
return s.deleteMediaByIDs(ctx, stale, false)
return s.deleteMediaByIDs(ctx, stale, true)
}
func pathBelongsToRoot(pathValue, rootPath string) bool {
+15
View File
@@ -501,6 +501,16 @@ func (s *StrmService) ClearFinishedDownloadTasks(ctx context.Context) (int64, er
return s.repo.StrmDownload.ClearFinished(ctx)
}
// ClearCanceledDownloadTasks 清空全部已取消的下载记录,返回删除数量。
func (s *StrmService) ClearCanceledDownloadTasks(ctx context.Context) (int64, error) {
return s.repo.StrmDownload.ClearCanceled(ctx)
}
// ClearCanceledUploadTasks 清空全部已取消的上传记录,返回删除数量。
func (s *StrmService) ClearCanceledUploadTasks(ctx context.Context) (int64, error) {
return s.repo.StrmUpload.ClearCanceled(ctx)
}
// RetryAllFailedDownloadTasks 批量重试所有失败下载任务,返回重新入队数量。
func (s *StrmService) RetryAllFailedDownloadTasks(ctx context.Context) (int64, error) {
return s.repo.StrmDownload.RetryAllFailed(ctx)
@@ -511,6 +521,11 @@ func (s *StrmService) CancelPendingDownloadTasks(ctx context.Context) (int64, er
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():
+25
View File
@@ -113,6 +113,7 @@ func NewStrmService(cfg *config.Config, log *zap.Logger, repos *repository.Conta
// Start 启动下载/上传队列 worker、定时同步巡检、115 token 刷新与队列清理。
func (s *StrmService) Start(ctx context.Context) {
s.sync115RelayKey(ctx)
s.recoverInterruptedSyncs(ctx)
downloadThreads := s.strmIntSetting(ctx, StrmSettingDownloadThreads, 3)
if downloadThreads < 1 {
downloadThreads = 1
@@ -141,6 +142,21 @@ func (s *StrmService) Start(ctx context.Context) {
zap.Int("upload_threads", uploadThreads))
}
// recoverInterruptedSyncs 在服务启动时自愈重置因服务重启遗留的 running 状态。
func (s *StrmService) recoverInterruptedSyncs(ctx context.Context) {
paths, err := s.repo.StrmSyncPath.List(ctx)
if err == nil {
for i := range paths {
p := &paths[i]
if p.LastSyncStatus == model.StrmSyncRecordRunning {
p.LastSyncStatus = model.StrmSyncRecordCanceled
p.LastSyncMessage = "服务重启,已重置同步状态"
_ = s.repo.StrmSyncPath.Update(ctx, p)
}
}
}
}
func (s *StrmService) Stop() {
s.stopOnce.Do(func() { close(s.stopCh) })
}
@@ -404,6 +420,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 +449,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 表达式")
}
+549 -112
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,26 @@ 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 // 远端元数据大小(上传比对用)
seenMetaTarget map[string]cloud.FileEntry
seenVideoTarget map[string]cloud.FileEntry
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 +75,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,
}
@@ -84,15 +106,27 @@ func (s *StrmService) StartSync(ctx context.Context, pathID string) error {
return nil
}
// CancelSync 取消正在进行的同步。
// CancelSync 取消正在进行的同步(若为僵尸运行状态则直接自愈重置)。
func (s *StrmService) CancelSync(ctx context.Context, pathID string) error {
s.mu.Lock()
defer s.mu.Unlock()
cancel, exists := s.running[pathID]
if !exists {
return errors.New("该目录当前没有进行中的同步")
if exists {
delete(s.running, pathID)
}
s.mu.Unlock()
if exists && cancel != nil {
cancel()
}
// 无论内存中是否活跃,确保同步目录状态正确重置为已取消
if p, err := s.repo.StrmSyncPath.FindByID(ctx, pathID); err == nil && p != nil {
if p.LastSyncStatus == model.StrmSyncRecordRunning {
p.LastSyncStatus = model.StrmSyncRecordCanceled
p.LastSyncMessage = "已取消"
_ = s.repo.StrmSyncPath.Update(ctx, p)
}
}
cancel()
return nil
}
@@ -140,14 +174,17 @@ func (s *StrmService) runSync(ctx context.Context, p *model.StrmSyncPath, rec *m
return
}
st := &strmSyncState{
s: s,
ctx: ctx,
p: p,
cfg: cfg,
rec: rec,
seenVideo: map[string]bool{},
seenMeta: map[string]bool{},
remoteMeta: map[string]int64{},
s: s,
ctx: ctx,
p: p,
cfg: cfg,
rec: rec,
syncType: rec.SyncType,
seenVideo: map[string]bool{},
seenMeta: map[string]bool{},
remoteMeta: map[string]int64{},
seenMetaTarget: map[string]cloud.FileEntry{},
seenVideoTarget: map[string]cloud.FileEntry{},
}
if p.Provider != model.StrmProviderLocal {
acct, err := s.repo.StrmAccount.FindByID(ctx, p.AccountID)
@@ -191,15 +228,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 +248,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 +303,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 +439,239 @@ 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 {
pathCounts := make(map[string]int, len(cached))
for _, item := range cached {
pathCounts[item.Path]++
}
for _, item := range cached {
// 若同一个 path 对应了多个不同 dir_id,说明包含历史层级塌陷的脏数据,不预加载,让后续步骤重新向 115 获取精确路径
if pathCounts[item.Path] > 1 {
continue
}
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)
}
st.updateSyncMessage(fmt.Sprintf("正在拉取远端文件列表 (共 %d 个文件)...", totalCount))
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 := 8
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 = 8
doneDirs atomic.Int64
totalDirs = len(pidList)
)
if len(pidList) < dirWorkers {
dirWorkers = len(pidList)
}
st.updateSyncMessage(fmt.Sprintf("正在解析目录树 (0/%d)...", totalDirs))
for i := 0; i < dirWorkers; i++ {
pwg.Add(1)
go func() {
defer pwg.Done()
for pid := range pidCh {
if ctx.Err() != nil {
return
}
if _, loaded := st.dirCache.Load(pid); loaded {
if n := doneDirs.Add(1); n%20 == 0 || n == int64(totalDirs) {
st.updateSyncMessage(fmt.Sprintf("正在解析目录树 (%d/%d)...", n, totalDirs))
}
continue
}
detail, err := open115.GetFsDetailByCid(ctx, pid)
if err != nil {
st.s.log.Warn("115: 获取目录详情失败", zap.String("pid", pid), zap.Error(err))
} else if detail != nil {
// 解析相对路径
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,
FileName: ancestor.Name,
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)
}
}
}
if n := doneDirs.Add(1); n%10 == 0 || n == int64(totalDirs) {
st.updateSyncMessage(fmt.Sprintf("正在解析目录树 (%d/%d)...", n, totalDirs))
}
}
}()
}
pwg.Wait()
}
st.updateSyncMessage(fmt.Sprintf("正在生成 STRM 与同步文件 (共 %d 个)...", len(allFiles)))
// 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 +689,30 @@ 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
}
st.mu.Lock()
if st.seenVideoTarget == nil {
st.seenVideoTarget = map[string]cloud.FileEntry{}
}
if _, exists := st.seenVideoTarget[target]; exists {
st.mu.Unlock()
st.touchProgress()
return
}
st.seenVideoTarget[target] = entry
st.mu.Unlock()
// 增量同步模式快速检查:本地 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 +725,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 +750,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 +813,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)
@@ -490,14 +851,40 @@ func (st *strmSyncState) handleMeta(entry cloud.FileEntry, rel, ext string) {
if err != nil {
return
}
st.mu.Lock()
if st.seenMetaTarget == nil {
st.seenMetaTarget = map[string]cloud.FileEntry{}
}
if _, exists := st.seenMetaTarget[target]; exists {
// 该本地目标路径在当前批次中已被处理(存在同名/重名冲突),直接忽略重复项,避免多份不同大小的文件在本地交替覆盖导致增量死循环
st.mu.Unlock()
st.touchProgress()
return
}
st.seenMetaTarget[target] = entry
st.mu.Unlock()
if info, err := os.Stat(target); err == nil && info.Size() == entry.Size {
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 +903,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 +966,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 +1169,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()
@@ -769,6 +1194,18 @@ func (st *strmSyncState) flushProgress() {
}
}
// updateSyncMessage 实时更新同步阶段提示信息,让前端界面清晰了解当前进度。
func (st *strmSyncState) updateSyncMessage(msg string) {
st.mu.Lock()
st.rec.Message = msg
st.p.LastSyncMessage = msg
rec := *st.rec
p := *st.p
st.mu.Unlock()
_ = st.s.repo.StrmSyncRecord.Update(st.ctx, &rec)
_ = st.s.repo.StrmSyncPath.Update(st.ctx, &p)
}
// ─── 定时同步巡检 ──────────────────────────────────────────────────────────────
func (s *StrmService) cronLoop(ctx context.Context) {
+194 -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,142 @@ 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)
}
}
// TestStrmDuplicateFileConflictResolution 测试远端存在多个同名不同大小文件时,本地确定性仲裁,避免增量死循环
func TestStrmDuplicateFileConflictResolution(t *testing.T) {
svc := testStrmService(t)
localDir := t.TempDir()
p := &model.StrmSyncPath{
Base: model.Base{ID: "dup-test-path"},
Provider: model.StrmProvider115,
RemotePath: "root",
LocalPath: localDir,
DownloadMeta: true,
}
st := &strmSyncState{
s: svc,
ctx: context.Background(),
p: p,
cfg: &strmPathConfig{DownloadMeta: true, MetaExt: []string{"nfo"}},
rec: &model.StrmSyncRecord{},
seenMeta: map[string]bool{},
remoteMeta: map[string]int64{},
seenMetaTarget: map[string]cloud.FileEntry{},
seenVideoTarget: map[string]cloud.FileEntry{},
}
// 模拟远端同目录下存在两个同名不同大小的 nfo 文件 (115 历史重复上传)
// entry1: 较早文件 (MTime: 1000, Size: 100)
entry1 := cloud.FileEntry{ID: "f1", Name: "test.nfo", Size: 100, MTime: 1000, PickCode: "p1"}
// entry2: 较新文件 (MTime: 2000, Size: 200)
entry2 := cloud.FileEntry{ID: "f2", Name: "test.nfo", Size: 200, MTime: 2000, PickCode: "p2"}
// 第一次全量处理:两者都在列表中
st.handleMeta(entry1, "test.nfo", ".nfo")
st.handleMeta(entry2, "test.nfo", ".nfo")
st.flushPendingDownloads()
// 验证仲裁结果:最终只产生 1 个下载任务,且使用的是首个匹配项 (Size 100/p1)
tasks, _, err := svc.repo.StrmDownload.List(context.Background(), "", 1, 10)
if err != nil {
t.Fatal(err)
}
if len(tasks) != 1 {
t.Fatalf("expected 1 download task after conflict resolution, got %d", len(tasks))
}
if tasks[0].Size != 100 || tasks[0].RemoteRef != "p1" {
t.Fatalf("expected task with size 100/p1, got size=%d ref=%s", tasks[0].Size, tasks[0].RemoteRef)
}
// 模拟该任务下载落盘完成
writeFile(t, filepath.Join(localDir, "test.nfo"), strings.Repeat("x", 100))
// 第二次增量同步:两者再次依次扫描
st2 := &strmSyncState{
s: svc,
ctx: context.Background(),
p: p,
cfg: &strmPathConfig{DownloadMeta: true, MetaExt: []string{"nfo"}},
rec: &model.StrmSyncRecord{},
seenMeta: map[string]bool{},
remoteMeta: map[string]int64{},
seenMetaTarget: map[string]cloud.FileEntry{},
seenVideoTarget: map[string]cloud.FileEntry{},
}
st2.handleMeta(entry1, "test.nfo", ".nfo")
st2.handleMeta(entry2, "test.nfo", ".nfo")
st2.flushPendingDownloads()
// 验证:不会新增任何下载任务,NewMeta 为 0,增量跳过
if st2.rec.NewMeta != 0 {
t.Fatalf("expected 0 new meta on incremental sync, got %d", st2.rec.NewMeta)
}
}
+25
View File
@@ -0,0 +1,25 @@
package main
import (
"context"
"fmt"
"github.com/ShukeBta/MMTL/internal/service"
"github.com/ShukeBta/MMTL/internal/service/cloud115"
"go.uber.org/zap"
)
func main() {
crypto := service.NewCryptoService("test-secret", zap.NewNop())
// Let's test with a mock RemoteFileDetail
d := &cloud115.RemoteFileDetail{
FileId: "3251154147730910635",
FileName: "出包王女",
Paths: []struct {
FileId string
Name string
}{
{FileId: "0", Name: "根目录"},
{FileId: "3238787832374488117", Name: "影视库"},
{FileId: "3238787913223892116", Name: "动漫"},
},
}
fmt.Println("RelativePath when rootCID is 3238787832374488117:", d.RelativePath("3238787832374488117"))
}
+11 -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),
@@ -142,6 +143,9 @@ export const strmAPI = {
clearFinishedDownloads: () =>
api.post<{ deleted: number }>('/admin/strm/downloads/clear-finished').then((r) => r.data),
clearCanceledDownloads: () =>
api.post<{ deleted: number }>('/admin/strm/downloads/clear-canceled').then((r) => r.data),
retryFailedDownloads: () =>
api.post<{ retried: number }>('/admin/strm/downloads/retry-failed').then((r) => r.data),
@@ -158,6 +162,12 @@ export const strmAPI = {
cancelUpload: (id: string) =>
api.post(`/admin/strm/uploads/${id}/cancel`).then((r) => r.data),
cancelPendingUploads: () =>
api.post<{ canceled: number }>('/admin/strm/uploads/cancel-pending').then((r) => r.data),
clearCanceledUploads: () =>
api.post<{ deleted: number }>('/admin/strm/uploads/clear-canceled').then((r) => r.data),
retryUpload: (id: string) =>
api.post(`/admin/strm/uploads/${id}/retry`).then((r) => r.data),
}
+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>
+63 -25
View File
@@ -96,7 +96,7 @@ function StrmQueuePanel({ kind }: { kind: 'download' | 'upload' }) {
}
}
const batchBtn = (title: string, cls: string, onClick: () => void) => (
const batchBtn = (title: string, icon: 'trash' | 'ban' | 'refresh', cls: string, onClick: () => void) => (
<button
type="button"
disabled={batchBusy}
@@ -105,33 +105,58 @@ function StrmQueuePanel({ kind }: { kind: 'download' | 'upload' }) {
'ml-auto inline-flex items-center gap-1 rounded-lg border px-2 py-1 text-xs font-semibold transition disabled:opacity-50 ' + cls
}
>
<Trash2 size={12} />
{icon === 'trash' && <Trash2 size={12} />}
{icon === 'ban' && <Ban size={12} />}
{icon === 'refresh' && <RefreshCw size={12} />}
{title}
</button>
)
const cancelAllPendingAction = () => {
const action = isDownload
? () => strmAPI.cancelPendingDownloads()
: () => strmAPI.cancelPendingUploads()
return runBatch(action, `确定取消所有排队中及进行中的${isDownload ? '下载' : '上传'}任务?`)
}
const batchActionByFilter = () => {
if (filter === 'pending' || filter === 'running')
return batchBtn(
'全部取消',
'ban',
'border-amber-200 text-amber-600 hover:bg-amber-50',
cancelAllPendingAction,
)
if (filter === 'canceled')
return batchBtn(
'清空已取消记录',
'trash',
'border-gray-200 text-rose-500 hover:bg-rose-50',
() =>
runBatch(
isDownload
? () => strmAPI.clearCanceledDownloads()
: () => strmAPI.clearCanceledUploads(),
`确定清空所有已取消的${isDownload ? '下载' : '上传'}记录?`,
),
)
if (!isDownload) return null
if (filter === 'done')
return batchBtn(
'清空成功记录',
'trash',
'border-gray-200 text-rose-500 hover:bg-rose-50',
() => runBatch(() => strmAPI.clearDoneDownloads(), '确定清空所有已完成下载记录?'),
)
if (filter === 'failed')
return batchBtn('批量重试', 'border-gray-200 text-brand-500 hover:bg-brand-50', () =>
return batchBtn('批量重试', 'refresh', 'border-gray-200 text-brand-500 hover:bg-brand-50', () =>
runBatch(() => strmAPI.retryFailedDownloads(), '确定重新入队所有失败下载任务?'),
)
if (filter === 'pending')
return batchBtn(
'批量取消',
'border-gray-200 text-amber-600 hover:bg-amber-50',
() => runBatch(() => strmAPI.cancelPendingDownloads(), '确定取消所有排队中的下载任务?'),
)
return null
}
const counts = snapshot?.counts
const activeTaskCount = (counts?.pending ?? 0) + (counts?.running ?? 0)
const tasks = snapshot?.tasks.filter((t) => filter === 'all' || t.status === filter) ?? []
const isDownload = kind === 'download'
const Icon = isDownload ? Download : Upload
@@ -152,23 +177,36 @@ function StrmQueuePanel({ kind }: { kind: 'download' | 'upload' }) {
: 'STRM 元数据上传情况(本地 → 远端网盘,3 秒自动刷新)'}
</p>
</div>
{isDownload && (
<button
type="button"
disabled={batchBusy}
onClick={() =>
runBatch(() => strmAPI.clearFinishedDownloads(), '确定清空所有失败和成功的下载记录?')
}
className="inline-flex items-center gap-1.5 rounded-xl border border-rose-200 px-3 py-2 text-sm font-semibold text-rose-500 transition hover:bg-rose-50 disabled:opacity-50"
>
<Trash2 size={14} />
清空失败与完成记录
<div className="ml-auto flex items-center gap-2">
{activeTaskCount > 0 && (
<button
type="button"
disabled={batchBusy}
onClick={cancelAllPendingAction}
className="inline-flex items-center gap-1.5 rounded-xl border border-amber-300 bg-amber-500/10 px-3 py-2 text-sm font-semibold text-amber-600 transition hover:bg-amber-500/20 disabled:opacity-50"
>
<Ban size={14} />
全部取消 ({activeTaskCount})
</button>
)}
{isDownload && (
<button
type="button"
disabled={batchBusy}
onClick={() =>
runBatch(() => strmAPI.clearFinishedDownloads(), '确定清空所有失败和成功的下载记录?')
}
className="inline-flex items-center gap-1.5 rounded-xl border border-rose-200 px-3 py-2 text-sm font-semibold text-rose-500 transition hover:bg-rose-50 disabled:opacity-50"
>
<Trash2 size={14} />
清空完成与失败
</button>
)}
<button type="button" onClick={refresh} className="inline-flex items-center gap-1.5 rounded-xl border border-gray-200 px-3 py-2 text-sm font-semibold text-ink-100 hover:bg-gray-50">
<RefreshCw size={14} />
刷新
</button>
)}
<button type="button" onClick={refresh} className="ml-auto inline-flex items-center gap-1.5 rounded-xl border border-gray-200 px-3 py-2 text-sm font-semibold text-ink-100 hover:bg-gray-50">
<RefreshCw size={14} />
刷新
</button>
</div>
</header>
<div className="flex flex-wrap items-center gap-2">
+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