diff --git a/internal/handler/strm.go b/internal/handler/strm.go index cf9e4dd..12477fb 100644 --- a/internal/handler/strm.go +++ b/internal/handler/strm.go @@ -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), } } diff --git a/internal/model/model.go b/internal/model/model.go index 812aec8..6deee22 100644 --- a/internal/model/model.go +++ b/internal/model/model.go @@ -54,7 +54,8 @@ func AllModels() []interface{} { &StrmAccount{}, &StrmSyncPath{}, &StrmSyncRecord{}, - &StrmDownloadTask{}, - &StrmUploadTask{}, - } + &StrmDownloadTask{}, + &StrmUploadTask{}, + &StrmDirCache{}, + } } diff --git a/internal/model/strm.go b/internal/model/strm.go index 709de50..e85a4c7 100644 --- a/internal/model/strm.go +++ b/internal/model/strm.go @@ -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"` // 相对根目录的路径 +} + diff --git a/internal/repository/repository.go b/internal/repository/repository.go index 99cb43e..332acfa 100644 --- a/internal/repository/repository.go +++ b/internal/repository/repository.go @@ -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}, } } diff --git a/internal/repository/strm_repository.go b/internal/repository/strm_repository.go index bccb039..6e25ad8 100644 --- a/internal/repository/strm_repository.go +++ b/internal/repository/strm_repository.go @@ -96,11 +96,12 @@ func (r *StrmSyncPathRepository) Update(ctx context.Context, p *model.StrmSyncPa "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, + "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(), @@ -122,6 +123,7 @@ func (r *StrmSyncRecordRepository) Create(ctx context.Context, rec *model.StrmSy 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{ + "sync_type": rec.SyncType, "status": rec.Status, "total": rec.Total, "new_strm": rec.NewStrm, @@ -436,3 +438,39 @@ func (r *StrmUploadTaskRepository) DeleteFinishedOlderThan(ctx context.Context, []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 { + 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 r.db.WithContext(ctx).Where("sync_path_id = ?", syncPathID).Delete(&model.StrmDirCache{}).Error +} + diff --git a/internal/service/cloud/cloud.go b/internal/service/cloud/cloud.go index 8c28b63..d334f99 100644 --- a/internal/service/cloud/cloud.go +++ b/internal/service/cloud/cloud.go @@ -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"` } diff --git a/internal/service/cloud/pan115_openapi.go b/internal/service/cloud/pan115_openapi.go index 3419358..603f0fd 100644 --- a/internal/service/cloud/pan115_openapi.go +++ b/internal/service/cloud/pan115_openapi.go @@ -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 } diff --git a/internal/service/cloud115/client.go b/internal/service/cloud115/client.go index bea72f4..27effed 100644 --- a/internal/service/cloud115/client.go +++ b/internal/service/cloud115/client.go @@ -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:"-"` // 原始响应体（外层附加字段用） } diff --git a/internal/service/cloud115/open.go b/internal/service/cloud115/open.go index 276f151..bbbd0b9 100644 --- a/internal/service/cloud115/open.go +++ b/internal/service/cloud115/open.go @@ -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 { diff --git a/internal/service/strm_service.go b/internal/service/strm_service.go index 05e19f1..f4fc9c8 100644 --- a/internal/service/strm_service.go +++ b/internal/service/strm_service.go @@ -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 表达式") } diff --git a/internal/service/strm_sync.go b/internal/service/strm_sync.go index a708708..7ca89d3 100644 --- a/internal/service/strm_sync.go +++ b/internal/service/strm_sync.go @@ -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,19 @@ 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 // 远端元数据大小（上传比对用） + 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 +68,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 +160,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 +207,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)) } @@ -208,8 +228,14 @@ func (st *strmSyncState) run() error { return fmt.Errorf("创建输出目录失败：%w", err) } 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 { @@ -374,6 +400,215 @@ func (st *strmSyncState) isMetaExt(ext string) bool { return false } +// walk115Flat 使用 115 开放平台扁平化分页批量拉取机制与目录拓扑缓存（参考 QMediaSync）。 +// 极大地降低 API 请求次数并支持毫秒级/秒级增量同步。 +func (st *strmSyncState) walk115Flat(open115 *cloud115.OpenClient) error { + 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 +626,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 +650,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 +675,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() @@ -576,29 +832,43 @@ 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 }) } diff --git a/internal/service/strm_sync_test.go b/internal/service/strm_sync_test.go index 7943bae..8abe99f 100644 --- a/internal/service/strm_sync_test.go +++ b/internal/service/strm_sync_test.go @@ -34,10 +34,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 +159,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 { diff --git a/web/src/api/strm.ts b/web/src/api/strm.ts index 896baa2..395f41d 100644 --- a/web/src/api/strm.ts +++ b/web/src/api/strm.ts @@ -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), diff --git a/web/src/pages/StrmDialogs.tsx b/web/src/pages/StrmDialogs.tsx index d165377..d2b80db 100644 --- a/web/src/pages/StrmDialogs.tsx +++ b/web/src/pages/StrmDialogs.tsx @@ -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({ set('exclude_name', e.target.value)} /> -
+
- + + + + set('cron', e.target.value)} />
diff --git a/web/src/pages/StrmManagePage.tsx b/web/src/pages/StrmManagePage.tsx index 357f707..c6768f0 100644 --- a/web/src/pages/StrmManagePage.tsx +++ b/web/src/pages/StrmManagePage.tsx @@ -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({ 取消 ) : ( - + <> + + + )}