mirror of
https://github.com/truewhile/MeBox.git
synced 2026-10-02 20:26:36 +08:00
Fix organize targets and cloud transfer controls
This commit is contained in:
@@ -109,7 +109,7 @@ func scanLibraryHandler(svc *service.Container) gin.HandlerFunc {
|
||||
})
|
||||
}
|
||||
_, _, _ = svc.Scan.StartCloudLibraryScan(id, false)
|
||||
finishHTTPTask(task, nil, "queued", "云盘扫描已加入后台队列", map[string]int64{"queued": 1})
|
||||
finishHTTPTask(task, nil, "queued", "云盘扫描已加入后台队列", map[string]int64{"queued": 1}, nil)
|
||||
c.JSON(http.StatusAccepted, gin.H{
|
||||
"library_id": id,
|
||||
"visited": 0,
|
||||
@@ -126,11 +126,11 @@ func scanLibraryHandler(svc *service.Container) gin.HandlerFunc {
|
||||
task := startScanHTTPTask(svc, "手动扫描入库", lib.Name, lib.Path)
|
||||
res, err := svc.Scan.ScanLibrary(c.Request.Context(), id)
|
||||
if err != nil {
|
||||
finishHTTPTask(task, err, "scan", "手动扫描入库失败", nil)
|
||||
finishHTTPTask(task, err, "scan", "手动扫描入库失败", nil, nil)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
finishHTTPTask(task, nil, "completed", "手动扫描入库结束", scanTaskMetrics(res))
|
||||
finishHTTPTask(task, nil, "completed", "手动扫描入库结束", scanTaskMetrics(res), nil)
|
||||
c.JSON(http.StatusOK, res)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -61,18 +61,18 @@ func organizeMediaHandler(svc *service.Container) gin.HandlerFunc {
|
||||
task := startOrganizeHTTPTask(svc, "手动整理媒体", opts)
|
||||
dst, err := svc.Organizer.OrganizeMediaWithOptions(c.Request.Context(), c.Param("id"), opts)
|
||||
if err != nil {
|
||||
finishHTTPTask(task, err, "organize", "手动整理媒体失败", nil)
|
||||
finishHTTPTask(task, err, "organize", "手动整理媒体失败", nil, nil)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
payload := gin.H{"path": dst}
|
||||
if organizeScanAfter(req.ScanAfter) && !req.DryRun && svc.Scan != nil {
|
||||
updateHTTPTask(task, "scan_scrape", "正在扫描入库并按设置刮削", nil)
|
||||
updateHTTPTask(task, "scan_scrape", "正在扫描入库并按设置刮削", nil, nil)
|
||||
scans, scrapes := scanAndScrapeAfterOrganize(c, svc, dst, strings.TrimSpace(req.LibraryID), req.ScrapeAfter)
|
||||
payload["scans"] = scans
|
||||
payload["scrapes"] = scrapes
|
||||
}
|
||||
finishHTTPTask(task, nil, "completed", "手动整理媒体结束", nil)
|
||||
finishHTTPTask(task, nil, "completed", "手动整理媒体结束", nil, nil)
|
||||
c.JSON(http.StatusOK, payload)
|
||||
}
|
||||
}
|
||||
@@ -85,15 +85,16 @@ func organizeLibraryHandler(svc *service.Container) gin.HandlerFunc {
|
||||
task := startOrganizeHTTPTask(svc, "手动整理媒体库", opts)
|
||||
res, err := svc.Organizer.OrganizeLibraryWithOptions(c.Request.Context(), c.Param("id"), opts)
|
||||
if err != nil {
|
||||
finishHTTPTask(task, err, "organize", "手动整理媒体库失败", nil)
|
||||
finishHTTPTask(task, err, "organize", "手动整理媒体库失败", nil, nil)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
details := service.OrganizeTaskDetails(res, 8)
|
||||
if organizeScanAfter(req.ScanAfter) && !req.DryRun && svc.Scan != nil {
|
||||
updateHTTPTask(task, "scan_scrape", "正在扫描入库并按设置刮削", nil)
|
||||
updateHTTPTask(task, "scan_scrape", "正在扫描入库并按设置刮削", service.OrganizeTaskMetrics(res), details)
|
||||
res.Scans, res.Scrapes = scanAndScrapeAfterOrganize(c, svc, res.DestPath, c.Param("id"), req.ScrapeAfter)
|
||||
}
|
||||
finishHTTPTask(task, nil, "completed", "手动整理媒体库结束", service.OrganizeTaskMetrics(res))
|
||||
finishHTTPTask(task, nil, "completed", "手动整理媒体库结束", service.OrganizeTaskMetrics(res), details)
|
||||
c.JSON(http.StatusOK, res)
|
||||
}
|
||||
}
|
||||
@@ -116,16 +117,17 @@ func organizeDirectoryHandler(svc *service.Container) gin.HandlerFunc {
|
||||
task := startOrganizeHTTPTask(svc, "手动整理入库", opts)
|
||||
res, err := svc.Organizer.OrganizeDirectory(c.Request.Context(), opts)
|
||||
if err != nil {
|
||||
finishHTTPTask(task, err, "organize", "手动整理入库失败", nil)
|
||||
finishHTTPTask(task, err, "organize", "手动整理入库失败", nil, nil)
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
updateHTTPTask(task, "organize", "手动整理完成,准备扫描入库", service.OrganizeTaskMetrics(res))
|
||||
details := service.OrganizeTaskDetails(res, 8)
|
||||
updateHTTPTask(task, "organize", "手动整理完成,准备扫描入库", service.OrganizeTaskMetrics(res), details)
|
||||
if organizeScanAfter(req.ScanAfter) && !req.DryRun && svc.Scan != nil {
|
||||
updateHTTPTask(task, "scan_scrape", "正在扫描入库并按设置刮削", service.OrganizeTaskMetrics(res))
|
||||
updateHTTPTask(task, "scan_scrape", "正在扫描入库并按设置刮削", service.OrganizeTaskMetrics(res), details)
|
||||
res.Scans, res.Scrapes = scanAndScrapeAfterOrganize(c, svc, res.DestPath, strings.TrimSpace(req.LibraryID), req.ScrapeAfter)
|
||||
}
|
||||
finishHTTPTask(task, nil, "completed", "手动整理入库结束", service.OrganizeTaskMetrics(res))
|
||||
finishHTTPTask(task, nil, "completed", "手动整理入库结束", service.OrganizeTaskMetrics(res), details)
|
||||
c.JSON(http.StatusOK, res)
|
||||
}
|
||||
}
|
||||
@@ -146,18 +148,18 @@ func startOrganizeHTTPTask(svc *service.Container, name string, opts service.Org
|
||||
})
|
||||
}
|
||||
|
||||
func updateHTTPTask(task *service.TaskHandle, stage, message string, metrics map[string]int64) {
|
||||
func updateHTTPTask(task *service.TaskHandle, stage, message string, metrics map[string]int64, details []string) {
|
||||
if task == nil {
|
||||
return
|
||||
}
|
||||
task.Update(service.TaskUpdate{Stage: stage, Message: message, Metrics: metrics})
|
||||
task.Update(service.TaskUpdate{Stage: stage, Message: message, Metrics: metrics, Details: details})
|
||||
}
|
||||
|
||||
func finishHTTPTask(task *service.TaskHandle, err error, stage, message string, metrics map[string]int64) {
|
||||
func finishHTTPTask(task *service.TaskHandle, err error, stage, message string, metrics map[string]int64, details []string) {
|
||||
if task == nil {
|
||||
return
|
||||
}
|
||||
task.Finish(err, service.TaskUpdate{Stage: stage, Message: message, Metrics: metrics})
|
||||
task.Finish(err, service.TaskUpdate{Stage: stage, Message: message, Metrics: metrics, Details: details})
|
||||
}
|
||||
|
||||
func scanAndScrapeAfterOrganize(c *gin.Context, svc *service.Container, destRoot, preferredLibraryID string, scrapeOverride *bool) ([]service.OrganizeScanSummary, []service.OrganizeScrapeSummary) {
|
||||
|
||||
@@ -83,7 +83,7 @@ func scrapeOneHandler(svc *service.Container) gin.HandlerFunc {
|
||||
}
|
||||
task := startScrapeHTTPTask(svc, "手动刮削媒体", m.Title, m.Path)
|
||||
if err := svc.Scraper.EnrichOne(c.Request.Context(), m); err != nil {
|
||||
finishHTTPTask(task, err, "scrape", "手动刮削媒体失败", nil)
|
||||
finishHTTPTask(task, err, "scrape", "手动刮削媒体失败", nil, nil)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
@@ -92,7 +92,7 @@ func scrapeOneHandler(svc *service.Container) gin.HandlerFunc {
|
||||
if refreshed != nil && refreshed.ScrapeStatus == "matched" {
|
||||
metrics["matched"] = 1
|
||||
}
|
||||
finishHTTPTask(task, nil, "completed", "手动刮削媒体结束", metrics)
|
||||
finishHTTPTask(task, nil, "completed", "手动刮削媒体结束", metrics, nil)
|
||||
c.JSON(http.StatusOK, refreshed)
|
||||
}
|
||||
}
|
||||
@@ -118,7 +118,7 @@ func scrapeLibraryHandler(svc *service.Container) gin.HandlerFunc {
|
||||
stage = "scrape"
|
||||
message = "手动刮削媒体库失败"
|
||||
}
|
||||
finishHTTPTask(task, err, stage, message, metrics)
|
||||
finishHTTPTask(task, err, stage, message, metrics, nil)
|
||||
}(libID, task)
|
||||
c.JSON(http.StatusAccepted, gin.H{"status": "scraping"})
|
||||
}
|
||||
|
||||
@@ -121,6 +121,10 @@ func schemaHandler(_ *service.Container) gin.HandlerFunc {
|
||||
{"key": "cloud.upload_recursive", "type": "toggle", "label": "递归扫描源目录"},
|
||||
{"key": "cloud.upload_sidecars", "type": "toggle", "label": "同步 NFO / 海报 / 字幕"},
|
||||
{"key": "cloud.upload_overwrite", "type": "toggle", "label": "覆盖远端同名文件"},
|
||||
{"key": "cloud.upload_transfer_mode", "type": "select", "label": "自动转存方式", "options": []gin.H{
|
||||
{"value": "copy", "label": "复制"},
|
||||
{"value": "move", "label": "移动"},
|
||||
}},
|
||||
{"key": "cloud.upload_interval_seconds", "type": "number", "label": "自动转存间隔秒数"},
|
||||
},
|
||||
},
|
||||
|
||||
@@ -1136,6 +1136,7 @@ func (d *DownloadService) onTorrentComplete(ctx context.Context, torrent QBitTor
|
||||
DestPath: res.DestPath,
|
||||
Message: "下载完成整理已完成,准备扫描入库",
|
||||
Metrics: OrganizeTaskMetrics(res),
|
||||
Details: OrganizeTaskDetails(res, 8),
|
||||
})
|
||||
}
|
||||
if d.scanner != nil && res != nil && strings.TrimSpace(res.DestPath) != "" && OrganizeResultNeedsVisibilitySync(res) {
|
||||
@@ -1144,6 +1145,7 @@ func (d *DownloadService) onTorrentComplete(ctx context.Context, torrent QBitTor
|
||||
Stage: "scan_scrape",
|
||||
Message: "正在扫描入库并按设置刮削",
|
||||
Metrics: OrganizeTaskMetrics(res),
|
||||
Details: OrganizeTaskDetails(res, 8),
|
||||
})
|
||||
}
|
||||
res.Scans, res.Scrapes = d.scanner.ScanAndScrapeLibrariesForPath(ctx, res.DestPath, "", OrganizeScrapeAfterEnabled(ctx, d.repo))
|
||||
@@ -1160,6 +1162,7 @@ func (d *DownloadService) onTorrentComplete(ctx context.Context, torrent QBitTor
|
||||
Stage: "completed",
|
||||
Message: "下载完成自动整理入库结束",
|
||||
Metrics: OrganizeTaskMetrics(res),
|
||||
Details: OrganizeTaskDetails(res, 8),
|
||||
})
|
||||
}
|
||||
d.markCompletedTorrentCatchupRecorded(context.Background(), torrent)
|
||||
|
||||
@@ -115,7 +115,18 @@ func (o *OrganizerService) OrganizeMediaWithOptions(ctx context.Context, mediaID
|
||||
if err != nil || lib == nil {
|
||||
return "", errors.New("library not found")
|
||||
}
|
||||
if _, ok := ParseCloudLibraryMount(lib.Path); ok {
|
||||
return "", errors.New("local organize cannot use cloud libraries directly; use external storage scan/mount for cloud media or enable cloud transfer to write to cloud")
|
||||
}
|
||||
baseRoot := o.resolveBaseRoot(ctx, lib, opts.DestPath)
|
||||
if _, ok := ParseCloudLibraryMount(baseRoot); ok {
|
||||
return "", errors.New("organize destination must be a local writable media directory; enable cloud transfer in external storage when writing to cloud")
|
||||
}
|
||||
if !opts.DryRun {
|
||||
if err := ensureOrganizeDestinationWritable(baseRoot); err != nil {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
mode := o.resolveTransferMode(ctx, opts.TransferMode)
|
||||
if isSeriesLibraryType(lib.Type) {
|
||||
if err := o.refreshEpisodeIdentity(m, lib); err != nil {
|
||||
@@ -284,6 +295,9 @@ func (o *OrganizerService) OrganizeLibraryWithOptions(ctx context.Context, libra
|
||||
if err != nil || lib == nil {
|
||||
return nil, errors.New("library not found")
|
||||
}
|
||||
if _, ok := ParseCloudLibraryMount(lib.Path); ok {
|
||||
return nil, errors.New("local organize cannot use cloud libraries directly; use external storage scan/mount for cloud media or enable cloud transfer to write to cloud")
|
||||
}
|
||||
var rows []model.Media
|
||||
if err := o.repo.DB.WithContext(ctx).
|
||||
Where("library_id = ? AND deleted_at IS NULL", libraryID).
|
||||
@@ -292,8 +306,19 @@ func (o *OrganizerService) OrganizeLibraryWithOptions(ctx context.Context, libra
|
||||
}
|
||||
// 源目录(待整理):仅整理位于该目录下的媒体;留空 = 整个媒体库。
|
||||
sourceRoot := o.resolveSourceRoot(ctx, lib, opts.SourcePath)
|
||||
if _, ok := ParseCloudLibraryMount(sourceRoot); ok {
|
||||
return nil, errors.New("organize source must be a local directory; cloud libraries should be managed from external storage scan/mount")
|
||||
}
|
||||
// 目的地目录:已位于该根下的文件视为已整理;受 dest_path 覆盖与设置影响。
|
||||
baseRoot := o.resolveBaseRoot(ctx, lib, opts.DestPath)
|
||||
if _, ok := ParseCloudLibraryMount(baseRoot); ok {
|
||||
return nil, errors.New("organize destination must be a local writable media directory; enable cloud transfer in external storage when writing to cloud")
|
||||
}
|
||||
if !opts.DryRun {
|
||||
if err := ensureOrganizeDestinationWritable(baseRoot); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
res := &OrganizeResult{SourcePath: sourceRoot, DestPath: baseRoot, DryRun: opts.DryRun}
|
||||
for i := range rows {
|
||||
// 不在源目录内的文件跳过(不属于本次「从源目录整理」的范围)。
|
||||
|
||||
@@ -123,10 +123,19 @@ func (o *OrganizerService) OrganizeDirectory(ctx context.Context, opts OrganizeO
|
||||
if statErr != nil {
|
||||
return nil, fmt.Errorf("source directory not accessible: %s", filepath.Clean(requestedSource))
|
||||
}
|
||||
dest := resolveMappedDestinationPath(o.defaultDestRoot(ctx, opts.DestPath))
|
||||
requestedDest := strings.TrimSpace(o.defaultDestRoot(ctx, opts.DestPath))
|
||||
if _, ok := ParseCloudLibraryMount(requestedDest); ok {
|
||||
return nil, errors.New("organize destination must be a local writable media directory; enable cloud transfer in external storage when writing to cloud")
|
||||
}
|
||||
dest := resolveMappedDestinationPath(requestedDest)
|
||||
if dest == "" || dest == "." {
|
||||
return nil, errors.New("destination path required")
|
||||
}
|
||||
if !opts.DryRun {
|
||||
if err := ensureOrganizeDestinationWritable(dest); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
mode := o.resolveTransferMode(ctx, opts.TransferMode)
|
||||
res := &OrganizeResult{SourcePath: source, DestPath: dest, DryRun: opts.DryRun}
|
||||
if !info.IsDir() {
|
||||
@@ -197,6 +206,32 @@ func (o *OrganizerService) OrganizeDirectory(ctx context.Context, opts OrganizeO
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func ensureOrganizeDestinationWritable(dest string) error {
|
||||
dest = strings.TrimSpace(dest)
|
||||
if dest == "" || dest == "." {
|
||||
return errors.New("destination path required")
|
||||
}
|
||||
if _, ok := ParseCloudLibraryMount(dest); ok {
|
||||
return errors.New("organize destination must be a local writable media directory; enable cloud transfer in external storage when writing to cloud")
|
||||
}
|
||||
if err := os.MkdirAll(dest, 0o755); err != nil { // #nosec G301 -- organized media directories must remain readable by NAS/player users.
|
||||
return fmt.Errorf("destination path is not a writable directory: %s: %w", dest, err)
|
||||
}
|
||||
probe, err := os.CreateTemp(dest, ".mediastation-write-test-*") // #nosec G304 -- dest is operator-configured organize root.
|
||||
if err != nil {
|
||||
return fmt.Errorf("destination path is not writable: %s: %w", dest, err)
|
||||
}
|
||||
name := probe.Name()
|
||||
if closeErr := probe.Close(); closeErr != nil {
|
||||
_ = os.Remove(name)
|
||||
return fmt.Errorf("destination path write probe failed: %s: %w", dest, closeErr)
|
||||
}
|
||||
if err := os.Remove(name); err != nil {
|
||||
return fmt.Errorf("destination path cleanup probe failed: %s: %w", dest, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type organizeDirectoryLayout struct {
|
||||
MediaType string
|
||||
Category string
|
||||
|
||||
@@ -394,6 +394,7 @@ func (s *SchedulerService) cloudUploadInput(ctx context.Context) CloudUploadInpu
|
||||
Recursive: parseBoolSetting(get(CloudUploadRecursiveKey), true),
|
||||
IncludeSidecars: parseBoolSetting(get(CloudUploadSidecarsKey), true),
|
||||
Overwrite: parseBoolSetting(get(CloudUploadOverwriteKey), false),
|
||||
TransferMode: get(CloudUploadTransferModeKey),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -569,6 +570,7 @@ func (s *SchedulerService) jobOrganizeSource(ctx context.Context) error {
|
||||
DestPath: res.DestPath,
|
||||
Message: "自动整理完成,准备扫描入库",
|
||||
Metrics: OrganizeTaskMetrics(res),
|
||||
Details: OrganizeTaskDetails(res, 8),
|
||||
})
|
||||
}
|
||||
if s.scanner != nil && res != nil && strings.TrimSpace(res.DestPath) != "" && OrganizeResultNeedsVisibilitySync(res) {
|
||||
@@ -577,6 +579,7 @@ func (s *SchedulerService) jobOrganizeSource(ctx context.Context) error {
|
||||
Stage: "scan_scrape",
|
||||
Message: "正在扫描入库并按设置刮削",
|
||||
Metrics: OrganizeTaskMetrics(res),
|
||||
Details: OrganizeTaskDetails(res, 8),
|
||||
})
|
||||
}
|
||||
res.Scans, res.Scrapes = s.scanner.ScanAndScrapeLibrariesForPath(ctx, res.DestPath, "", OrganizeScrapeAfterEnabled(ctx, s.repo))
|
||||
@@ -594,6 +597,7 @@ func (s *SchedulerService) jobOrganizeSource(ctx context.Context) error {
|
||||
Stage: "completed",
|
||||
Message: "自动整理入库结束",
|
||||
Metrics: OrganizeTaskMetrics(res),
|
||||
Details: OrganizeTaskDetails(res, 8),
|
||||
})
|
||||
}
|
||||
if s.log != nil && res != nil {
|
||||
|
||||
@@ -147,9 +147,10 @@ func (s *StorageConfigService) Save(ctx context.Context, in StorageInput) (*Stor
|
||||
return s.Get(ctx, in.Type)
|
||||
}
|
||||
|
||||
// Logout clears saved cloud login credentials and disables the storage backend.
|
||||
// It intentionally keeps non-secret connection hints such as server / WebDAV
|
||||
// URL / timeout so the admin can log in again without rebuilding the form.
|
||||
// Logout clears saved cloud login credentials, disables the storage backend,
|
||||
// and removes virtual cloud libraries/media for that provider. It intentionally
|
||||
// keeps non-secret connection hints such as server / WebDAV URL / timeout so
|
||||
// the admin can log in again without rebuilding the form.
|
||||
func (s *StorageConfigService) Logout(ctx context.Context, typ string) (*StorageView, error) {
|
||||
if !validStorageType(typ) {
|
||||
return nil, fmt.Errorf("unsupported storage type %q", typ)
|
||||
@@ -172,7 +173,53 @@ func (s *StorageConfigService) Logout(ctx context.Context, typ string) (*Storage
|
||||
cfg[k] = v
|
||||
}
|
||||
enabled := false
|
||||
return s.Save(ctx, StorageInput{Type: typ, Config: cfg, Enabled: &enabled})
|
||||
saved, err := s.Save(ctx, StorageInput{Type: typ, Config: cfg, Enabled: &enabled})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
purged, err := s.purgeCloudLibraries(ctx, typ)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s.log != nil {
|
||||
s.log.Info("storage logout cleared cloud libraries",
|
||||
zap.String("storage_type", typ),
|
||||
zap.Int("libraries_deleted", purged))
|
||||
}
|
||||
return saved, nil
|
||||
}
|
||||
|
||||
func (s *StorageConfigService) purgeCloudLibraries(ctx context.Context, storageType string) (int, error) {
|
||||
if s == nil || s.repo == nil || s.repo.Library == nil || s.repo.Media == nil {
|
||||
return 0, nil
|
||||
}
|
||||
libs, err := s.repo.Library.List(ctx)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("list libraries: %w", err)
|
||||
}
|
||||
var affectedLibs []string
|
||||
for _, lib := range libs {
|
||||
if mount, ok := ParseCloudLibraryMount(lib.Path); ok && mount.Provider == storageType {
|
||||
affectedLibs = append(affectedLibs, lib.ID)
|
||||
}
|
||||
}
|
||||
for _, libID := range affectedLibs {
|
||||
if err := s.repo.Media.PurgeByLibrary(ctx, libID); err != nil {
|
||||
if s.log != nil {
|
||||
s.log.Warn("purge media by library failed", zap.String("library_id", libID), zap.Error(err))
|
||||
}
|
||||
return len(affectedLibs), fmt.Errorf("purge media by library %s: %w", libID, err)
|
||||
}
|
||||
}
|
||||
for _, libID := range affectedLibs {
|
||||
if err := s.repo.Library.Delete(ctx, libID); err != nil {
|
||||
if s.log != nil {
|
||||
s.log.Warn("delete library failed", zap.String("library_id", libID), zap.Error(err))
|
||||
}
|
||||
return len(affectedLibs), fmt.Errorf("delete library %s: %w", libID, err)
|
||||
}
|
||||
}
|
||||
return len(affectedLibs), nil
|
||||
}
|
||||
|
||||
func isStorageLoginSecretKey(key string) bool {
|
||||
@@ -778,32 +825,9 @@ func (s *StorageConfigService) DeleteStorage(ctx context.Context, storageType st
|
||||
return fmt.Errorf("storage config not found: %s", storageType)
|
||||
}
|
||||
|
||||
// 查找使用该存储的云盘库
|
||||
libs, err := s.repo.Library.List(ctx)
|
||||
affectedLibs, err := s.purgeCloudLibraries(ctx, storageType)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list libraries: %w", err)
|
||||
}
|
||||
|
||||
var affectedLibs []string
|
||||
for _, lib := range libs {
|
||||
// 检查是否是云盘库且Provider匹配
|
||||
if mount, ok := ParseCloudLibraryMount(lib.Path); ok && mount.Provider == storageType {
|
||||
affectedLibs = append(affectedLibs, lib.ID)
|
||||
}
|
||||
}
|
||||
|
||||
// 删除关联媒体
|
||||
for _, libID := range affectedLibs {
|
||||
if err := s.repo.Media.PurgeByLibrary(ctx, libID); err != nil {
|
||||
s.log.Warn("purge media by library failed", zap.String("library_id", libID), zap.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
// 删除关联库
|
||||
for _, libID := range affectedLibs {
|
||||
if err := s.repo.Library.Delete(ctx, libID); err != nil {
|
||||
s.log.Warn("delete library failed", zap.String("library_id", libID), zap.Error(err))
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// 删除存储配置
|
||||
@@ -813,7 +837,7 @@ func (s *StorageConfigService) DeleteStorage(ctx context.Context, storageType st
|
||||
|
||||
s.log.Info("storage deleted",
|
||||
zap.String("storage_type", storageType),
|
||||
zap.Int("libraries_deleted", len(affectedLibs)))
|
||||
zap.Int("libraries_deleted", affectedLibs))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
package service
|
||||
|
||||
import "testing"
|
||||
import (
|
||||
"testing"
|
||||
|
||||
func TestStorageConfigLogoutClearsCloudCredentialsOnly(t *testing.T) {
|
||||
_, storage := newStorageUploadTestService(t)
|
||||
"github.com/ShukeBta/MediaStationGo/internal/model"
|
||||
)
|
||||
|
||||
func TestStorageConfigLogoutClearsCredentialsAndCloudLibraries(t *testing.T) {
|
||||
repos, storage := newStorageUploadTestService(t)
|
||||
enabled := true
|
||||
if _, err := storage.Save(t.Context(), StorageInput{
|
||||
Type: "openlist",
|
||||
@@ -20,6 +24,20 @@ func TestStorageConfigLogoutClearsCloudCredentialsOnly(t *testing.T) {
|
||||
}); err != nil {
|
||||
t.Fatalf("save storage: %v", err)
|
||||
}
|
||||
cloudLib := model.Library{Name: "OpenList", Path: BuildCloudLibraryPath("openlist", "/TV", "/TV"), Type: "tv", Enabled: true}
|
||||
if err := repos.Library.Create(t.Context(), &cloudLib); err != nil {
|
||||
t.Fatalf("create cloud library: %v", err)
|
||||
}
|
||||
localLib := model.Library{Name: "Local", Path: t.TempDir(), Type: "movie", Enabled: true}
|
||||
if err := repos.Library.Create(t.Context(), &localLib); err != nil {
|
||||
t.Fatalf("create local library: %v", err)
|
||||
}
|
||||
if err := repos.Media.Upsert(t.Context(), &model.Media{LibraryID: cloudLib.ID, Title: "Cloud", Path: "cloud://openlist/TV/Movie.mkv"}); err != nil {
|
||||
t.Fatalf("create cloud media: %v", err)
|
||||
}
|
||||
if err := repos.Media.Upsert(t.Context(), &model.Media{LibraryID: localLib.ID, Title: "Local", Path: localLib.Path + "/Movie.mkv"}); err != nil {
|
||||
t.Fatalf("create local media: %v", err)
|
||||
}
|
||||
|
||||
view, err := storage.Logout(t.Context(), "openlist")
|
||||
if err != nil {
|
||||
@@ -36,4 +54,21 @@ func TestStorageConfigLogoutClearsCloudCredentialsOnly(t *testing.T) {
|
||||
if view.Config["server"] != "http://openlist.test" || view.Config["url"] != "http://openlist.test/dav/" || view.Config["timeout_seconds"] != "120" {
|
||||
t.Fatalf("logout should keep non-secret connection hints, config = %#v", view.Config)
|
||||
}
|
||||
if got, err := repos.Library.FindByID(t.Context(), cloudLib.ID); err != nil {
|
||||
t.Fatalf("find cloud library: %v", err)
|
||||
} else if got != nil {
|
||||
t.Fatalf("cloud library should be removed after logout: %#v", got)
|
||||
}
|
||||
if got, err := repos.Library.FindByID(t.Context(), localLib.ID); err != nil {
|
||||
t.Fatalf("find local library: %v", err)
|
||||
} else if got == nil {
|
||||
t.Fatal("local library should remain after cloud logout")
|
||||
}
|
||||
var cloudMediaCount int64
|
||||
if err := repos.DB.Unscoped().Model(&model.Media{}).Where("path = ?", "cloud://openlist/TV/Movie.mkv").Count(&cloudMediaCount).Error; err != nil {
|
||||
t.Fatalf("count cloud media: %v", err)
|
||||
}
|
||||
if cloudMediaCount != 0 {
|
||||
t.Fatalf("cloud media should be purged after logout, count=%d", cloudMediaCount)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -23,6 +23,7 @@ const (
|
||||
CloudUploadRecursiveKey = "cloud.upload_recursive"
|
||||
CloudUploadSidecarsKey = "cloud.upload_sidecars"
|
||||
CloudUploadOverwriteKey = "cloud.upload_overwrite"
|
||||
CloudUploadTransferModeKey = "cloud.upload_transfer_mode"
|
||||
CloudUploadIntervalSecondsKey = "cloud.upload_interval_seconds"
|
||||
CloudUploadUnsupportedProvider = "本地文件直传目前支持 Alist / OpenList / WebDAV / CloudDrive2;115/夸克原生上传需要各自的分片上传私有接口,建议先用 CloudDrive2、OpenList 或 Alist 桥接后转存。"
|
||||
)
|
||||
@@ -34,12 +35,14 @@ type CloudUploadInput struct {
|
||||
Recursive bool `json:"recursive"`
|
||||
IncludeSidecars bool `json:"include_sidecars"`
|
||||
Overwrite bool `json:"overwrite"`
|
||||
TransferMode string `json:"transfer_mode"`
|
||||
}
|
||||
|
||||
type CloudUploadResult struct {
|
||||
SourcePath string `json:"source_path"`
|
||||
DestPath string `json:"dest_path"`
|
||||
Uploaded int `json:"uploaded"`
|
||||
Moved int `json:"moved,omitempty"`
|
||||
Skipped int `json:"skipped"`
|
||||
Bytes int64 `json:"bytes"`
|
||||
Errors []string `json:"errors,omitempty"`
|
||||
@@ -49,7 +52,7 @@ type CloudUploadResult struct {
|
||||
type CloudUploadResultItem struct {
|
||||
Source string `json:"source"`
|
||||
Target string `json:"target"`
|
||||
Action string `json:"action"` // upload / skip / error
|
||||
Action string `json:"action"` // upload / move / skip / error
|
||||
Size int64 `json:"size,omitempty"`
|
||||
Reason string `json:"reason,omitempty"`
|
||||
}
|
||||
@@ -65,10 +68,10 @@ var cloudUploadSidecarExtensions = map[string]struct{}{
|
||||
".srt": {}, ".ass": {}, ".ssa": {}, ".vtt": {}, ".sub": {}, ".idx": {},
|
||||
}
|
||||
|
||||
// UploadLocal copies local media files into an external storage backend. It is
|
||||
// intentionally conservative: it never deletes local files, skips existing
|
||||
// remote targets unless Overwrite is set, and only uploads video + common
|
||||
// sidecar metadata files.
|
||||
// UploadLocal copies or moves local media files into an external storage
|
||||
// backend. Cloud writes are intentionally gated by the per-storage
|
||||
// transfer_enabled switch so mounting/scanning a cloud provider does not
|
||||
// accidentally make it writable.
|
||||
func (s *StorageConfigService) UploadLocal(ctx context.Context, in CloudUploadInput) (*CloudUploadResult, error) {
|
||||
in.Type = strings.TrimSpace(in.Type)
|
||||
in.SourcePath = strings.TrimSpace(in.SourcePath)
|
||||
@@ -76,7 +79,22 @@ func (s *StorageConfigService) UploadLocal(ctx context.Context, in CloudUploadIn
|
||||
if in.SourcePath == "" {
|
||||
return nil, errors.New("source_path required")
|
||||
}
|
||||
uploader, err := s.uploader(ctx, in.Type)
|
||||
view, err := s.Get(ctx, in.Type)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if view == nil || !view.Enabled {
|
||||
return nil, fmt.Errorf("%s storage not configured", in.Type)
|
||||
}
|
||||
if !parseBoolSetting(strr(view.Config["transfer_enabled"]), false) {
|
||||
return nil, errors.New("cloud transfer is disabled for this storage; enable transfer in external storage settings before writing to cloud")
|
||||
}
|
||||
mode := resolveCloudUploadTransferMode(in.TransferMode, strr(view.Config["transfer_mode"]), s.settingValue(ctx, CloudUploadTransferModeKey))
|
||||
if mode != TransferCopy && mode != TransferMove {
|
||||
return nil, errors.New("transfer_mode must be copy or move")
|
||||
}
|
||||
in.TransferMode = string(mode)
|
||||
uploader, err := s.uploaderForView(in.Type, view)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -123,6 +141,13 @@ func (s *StorageConfigService) uploader(ctx context.Context, typ string) (storag
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if view == nil || !view.Enabled {
|
||||
return nil, fmt.Errorf("%s storage not configured", typ)
|
||||
}
|
||||
return s.uploaderForView(typ, view)
|
||||
}
|
||||
|
||||
func (s *StorageConfigService) uploaderForView(typ string, view *StorageView) (storageUploader, error) {
|
||||
if view == nil || !view.Enabled {
|
||||
return nil, fmt.Errorf("%s storage not configured", typ)
|
||||
}
|
||||
@@ -171,7 +196,38 @@ func (s *StorageConfigService) uploadOne(ctx context.Context, uploader storageUp
|
||||
}
|
||||
result.Uploaded++
|
||||
result.Bytes += size
|
||||
addUploadItem(result, CloudUploadResultItem{Source: localPath, Target: remotePath, Action: "upload", Size: size})
|
||||
action := "upload"
|
||||
mode := resolveCloudUploadTransferMode(in.TransferMode)
|
||||
if mode == TransferMove {
|
||||
if err := os.Remove(localPath); err != nil {
|
||||
addUploadError(result, localPath, remotePath, fmt.Errorf("uploaded but failed to remove local source: %w", err))
|
||||
addUploadItem(result, CloudUploadResultItem{Source: localPath, Target: remotePath, Action: "upload", Size: size, Reason: "source remove failed"})
|
||||
return
|
||||
}
|
||||
result.Moved++
|
||||
action = "move"
|
||||
}
|
||||
addUploadItem(result, CloudUploadResultItem{Source: localPath, Target: remotePath, Action: action, Size: size})
|
||||
}
|
||||
|
||||
func (s *StorageConfigService) settingValue(ctx context.Context, key string) string {
|
||||
if s == nil || s.repo == nil || s.repo.Setting == nil {
|
||||
return ""
|
||||
}
|
||||
v, _ := s.repo.Setting.Get(ctx, key)
|
||||
return strings.TrimSpace(v)
|
||||
}
|
||||
|
||||
func resolveCloudUploadTransferMode(values ...string) TransferMode {
|
||||
for _, value := range values {
|
||||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||||
case "move", "移动":
|
||||
return TransferMove
|
||||
case "copy", "复制":
|
||||
return TransferCopy
|
||||
}
|
||||
}
|
||||
return TransferCopy
|
||||
}
|
||||
|
||||
func eligibleCloudUploadFile(localPath string, includeSidecars bool) bool {
|
||||
|
||||
@@ -48,8 +48,9 @@ func TestStorageConfigUploadLocalToAlist(t *testing.T) {
|
||||
if _, err := storage.Save(t.Context(), StorageInput{
|
||||
Type: "alist",
|
||||
Config: map[string]any{
|
||||
"server": alist.URL,
|
||||
"token": "alist-token",
|
||||
"server": alist.URL,
|
||||
"token": "alist-token",
|
||||
"transfer_enabled": "true",
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -122,8 +123,9 @@ func TestStorageConfigUploadLocalToOpenListAPI(t *testing.T) {
|
||||
if _, err := storage.Save(t.Context(), StorageInput{
|
||||
Type: "openlist",
|
||||
Config: map[string]any{
|
||||
"server": openlist.URL,
|
||||
"token": "openlist-token",
|
||||
"server": openlist.URL,
|
||||
"token": "openlist-token",
|
||||
"transfer_enabled": "true",
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -195,9 +197,10 @@ func TestStorageConfigUploadLocalToOpenListAPIWithUsernamePassword(t *testing.T)
|
||||
if _, err := storage.Save(t.Context(), StorageInput{
|
||||
Type: "openlist",
|
||||
Config: map[string]any{
|
||||
"server": openlist.URL,
|
||||
"username": "alice",
|
||||
"password": "secret",
|
||||
"server": openlist.URL,
|
||||
"username": "alice",
|
||||
"password": "secret",
|
||||
"transfer_enabled": "true",
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -285,8 +288,9 @@ func TestSchedulerCloudUploadUsesConfiguredLocalSource(t *testing.T) {
|
||||
if _, err := storage.Save(t.Context(), StorageInput{
|
||||
Type: "alist",
|
||||
Config: map[string]any{
|
||||
"server": alist.URL,
|
||||
"token": "token",
|
||||
"server": alist.URL,
|
||||
"token": "token",
|
||||
"transfer_enabled": "true",
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -337,9 +341,10 @@ func TestStorageConfigUploadLocalToCloudDrive2(t *testing.T) {
|
||||
if _, err := storage.Save(t.Context(), StorageInput{
|
||||
Type: "clouddrive2",
|
||||
Config: map[string]any{
|
||||
"url": dav.URL + "/dav",
|
||||
"username": "user",
|
||||
"password": "pass",
|
||||
"url": dav.URL + "/dav",
|
||||
"username": "user",
|
||||
"password": "pass",
|
||||
"transfer_enabled": "true",
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -362,13 +367,91 @@ func TestStorageConfigUploadLocalToCloudDrive2(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestStorageConfigUploadLocalRequiresTransferEnabled(t *testing.T) {
|
||||
_, storage := newStorageUploadTestService(t)
|
||||
if _, err := storage.Save(t.Context(), StorageInput{
|
||||
Type: "alist",
|
||||
Config: map[string]any{
|
||||
"server": "http://alist.test",
|
||||
"token": "token",
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
source := t.TempDir()
|
||||
if err := os.WriteFile(filepath.Join(source, "Movie.mkv"), []byte("movie"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err := storage.UploadLocal(t.Context(), CloudUploadInput{
|
||||
Type: "alist",
|
||||
SourcePath: source,
|
||||
DestPath: "/MediaStationGo",
|
||||
Recursive: true,
|
||||
})
|
||||
if err == nil || !strings.Contains(err.Error(), "transfer is disabled") {
|
||||
t.Fatalf("upload error = %v, want transfer disabled", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStorageConfigUploadLocalMoveDeletesSourceAfterUpload(t *testing.T) {
|
||||
var uploaded []string
|
||||
alist := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/api/fs/mkdir":
|
||||
_, _ = w.Write([]byte(`{"code":200}`))
|
||||
case "/api/fs/get":
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
case "/api/fs/put":
|
||||
decoded, _ := url.PathUnescape(r.Header.Get("File-Path"))
|
||||
uploaded = append(uploaded, decoded)
|
||||
_, _ = w.Write([]byte(`{"code":200}`))
|
||||
default:
|
||||
t.Fatalf("unexpected alist path %s", r.URL.Path)
|
||||
}
|
||||
}))
|
||||
defer alist.Close()
|
||||
|
||||
_, storage := newStorageUploadTestService(t)
|
||||
if _, err := storage.Save(t.Context(), StorageInput{
|
||||
Type: "alist",
|
||||
Config: map[string]any{
|
||||
"server": alist.URL,
|
||||
"token": "token",
|
||||
"transfer_enabled": "true",
|
||||
"transfer_mode": "move",
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
source := t.TempDir()
|
||||
file := filepath.Join(source, "Movie.mkv")
|
||||
if err := os.WriteFile(file, []byte("movie"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
res, err := storage.UploadLocal(t.Context(), CloudUploadInput{
|
||||
Type: "alist",
|
||||
SourcePath: source,
|
||||
DestPath: "/MediaStationGo",
|
||||
Recursive: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("upload local move: %v", err)
|
||||
}
|
||||
if res.Uploaded != 1 || res.Moved != 1 || len(uploaded) != 1 {
|
||||
t.Fatalf("result = %+v uploaded=%#v", res, uploaded)
|
||||
}
|
||||
if _, err := os.Stat(file); !os.IsNotExist(err) {
|
||||
t.Fatalf("source should be removed after move upload, stat err=%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func newStorageUploadTestService(t *testing.T) (*repository.Container, *StorageConfigService) {
|
||||
t.Helper()
|
||||
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.AutoMigrate(&model.StorageConfig{}, &model.Setting{}); err != nil {
|
||||
if err := db.AutoMigrate(&model.StorageConfig{}, &model.Setting{}, &model.Library{}, &model.Media{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
repos := repository.New(db)
|
||||
|
||||
@@ -32,6 +32,7 @@ type BackgroundTask struct {
|
||||
DestPath string `json:"dest_path,omitempty"`
|
||||
Message string `json:"message,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Details []string `json:"details,omitempty"`
|
||||
Metrics map[string]int64 `json:"metrics,omitempty"`
|
||||
StartedAt time.Time `json:"started_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
@@ -43,6 +44,7 @@ type TaskUpdate struct {
|
||||
SourcePath string
|
||||
DestPath string
|
||||
Message string
|
||||
Details []string
|
||||
Metrics map[string]int64
|
||||
}
|
||||
|
||||
@@ -206,6 +208,9 @@ func applyTaskUpdate(task *BackgroundTask, update TaskUpdate) {
|
||||
if update.Message != "" {
|
||||
task.Message = update.Message
|
||||
}
|
||||
if update.Details != nil {
|
||||
task.Details = append([]string(nil), update.Details...)
|
||||
}
|
||||
if update.Metrics != nil {
|
||||
task.Metrics = cloneTaskMetrics(update.Metrics)
|
||||
}
|
||||
@@ -213,6 +218,9 @@ func applyTaskUpdate(task *BackgroundTask, update TaskUpdate) {
|
||||
|
||||
func cloneBackgroundTask(task BackgroundTask) BackgroundTask {
|
||||
task.Metrics = cloneTaskMetrics(task.Metrics)
|
||||
if task.Details != nil {
|
||||
task.Details = append([]string(nil), task.Details...)
|
||||
}
|
||||
if task.FinishedAt != nil {
|
||||
finishedAt := *task.FinishedAt
|
||||
task.FinishedAt = &finishedAt
|
||||
@@ -278,6 +286,43 @@ func OrganizeTaskMetrics(res *OrganizeResult) map[string]int64 {
|
||||
return metrics
|
||||
}
|
||||
|
||||
func OrganizeTaskDetails(res *OrganizeResult, limit int) []string {
|
||||
if res == nil || limit <= 0 {
|
||||
return nil
|
||||
}
|
||||
out := make([]string, 0, limit)
|
||||
for _, line := range res.Errors {
|
||||
line = strings.TrimSpace(line)
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
out = append(out, "错误: "+line)
|
||||
if len(out) >= limit {
|
||||
return out
|
||||
}
|
||||
}
|
||||
for _, item := range res.Items {
|
||||
if item.Action != "error" && item.Action != "skip" {
|
||||
continue
|
||||
}
|
||||
line := strings.TrimSpace(item.Source)
|
||||
if item.Reason != "" {
|
||||
line += ": " + strings.TrimSpace(item.Reason)
|
||||
}
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
out = append(out, item.Action+": "+line)
|
||||
if len(out) >= limit {
|
||||
return out
|
||||
}
|
||||
}
|
||||
if len(out) == 0 {
|
||||
return nil
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func OrganizeSkipReasonCounts(res *OrganizeResult) map[string]int {
|
||||
if res == nil || len(res.Items) == 0 {
|
||||
return nil
|
||||
|
||||
Reference in New Issue
Block a user