feat: 扫描整理支持中止;存放目录始终创建 KMP已整理 层

This commit is contained in:
2025-05-31 21:09:26 +08:00
parent 0aa9d44025
commit ea88f3e766
6 changed files with 157 additions and 32 deletions
+2 -1
View File
@@ -33,9 +33,10 @@ func NewRouter(st *store.Store) *gin.Engine {
apiGroup.POST("/media/list", mc.List)
apiGroup.POST("/media/stats", mc.Stats)
// 扫描整理(开始 / 查进度 / 实时进度)
// 扫描整理(开始 / 中止 / 查进度 / 实时进度)
sc := NewScanController(service.NewScanService(st))
apiGroup.POST("/scan/start", sc.Start)
apiGroup.POST("/scan/cancel", sc.Cancel)
apiGroup.GET("/scan/status", sc.Status)
apiGroup.GET("/scan/ws", sc.Ws)
+10
View File
@@ -47,3 +47,13 @@ func (c *ScanController) Start(ctx *gin.Context) {
func (c *ScanController) Status(ctx *gin.Context) {
Success(ctx, c.service.Status())
}
// Cancel 处理 POST /api/scan/cancel:中止正在运行的扫描整理任务。
// 成功后立刻返回当前进度快照,好让页面马上知道任务在收尾。
func (c *ScanController) Cancel(ctx *gin.Context) {
if err := c.service.Cancel(); err != nil {
Error(ctx, http.StatusBadRequest, err.Error())
return
}
Success(ctx, c.service.Status())
}
+24 -4
View File
@@ -11,6 +11,7 @@
package organize
import (
"context"
"fmt"
"os"
"path/filepath"
@@ -59,6 +60,7 @@ type Organizer struct {
ffmpegPath string // ffmpeg 程序路径;为空表示没有转正 EXIF 旋转图片的能力
store *store.Store // 数据库;为 nil 表示不记录媒体信息
mover *mover.Mover // 负责实际搬文件的家伙
ctx context.Context // 中止信号;ctx 被取消时探测/归类立即停下
buckets map[string][]string // 目标目录 → 里面该放哪些文件(只记源路径)
order []string // 目标目录的先后顺序(保证输出稳定、不来回跳)
onProgress func(phase string, done, total int)
@@ -66,13 +68,15 @@ type Organizer struct {
}
// New 创建整理器。store 可以为 nil——数据库打不开时,整理照常进行,只是不记入库信息。
func New(pickerDir, ffprobePath, ffmpegPath string, st *store.Store) *Organizer {
// ctx 可以为 nil——不提供中止能力,任务跑到完为止。
func New(pickerDir, ffprobePath, ffmpegPath string, st *store.Store, ctx context.Context) *Organizer {
return &Organizer{
pickerDir: pickerDir,
ffprobePath: ffprobePath,
ffmpegPath: ffmpegPath,
store: st,
mover: &mover.Mover{},
ctx: ctx,
buckets: make(map[string][]string),
}
}
@@ -104,7 +108,9 @@ func workerCount() int {
// 通俗理解:一条流水线有 N 个工人,每个工人从传送带上拿一个活干完,
// 干完把结果按编号放回架子上(results[i] 对应 items[i] 的结果)。
// onProgress 每干完一个活就报一次数(已完成几个),线程安全、只会涨不会退。
func runParallel[T any, R any](items []T, fn func(T) R, onProgress func(done int)) []R {
// ctx 被取消时工人不再干活(任务照常消费,避免发送方堵死),
// 被跳过的位置留零值,调用方看到 ctx.Err() 会自行停下。
func runParallel[T any, R any](ctx context.Context, items []T, fn func(T) R, onProgress func(done int)) []R {
results := make([]R, len(items))
if len(items) == 0 {
return results
@@ -121,6 +127,10 @@ func runParallel[T any, R any](items []T, fn func(T) R, onProgress func(done int
go func() {
defer wg.Done()
for i := range jobs {
// 用户点了中止:不再干活,但任务还要继续消费,免得主 goroutine 堵死在发送上
if ctx != nil && ctx.Err() != nil {
continue
}
results[i] = fn(items[i])
if onProgress != nil {
onProgress(int(atomic.AddInt32(&doneCount, 1)))
@@ -350,10 +360,15 @@ func (o *Organizer) ProcessImages(images []scanner.File) {
// 统计结果,最后打一行总结给用户看
success, fail, ignore, dup := 0, 0, 0, 0
total := len(images)
results := runParallel(images, o.probeImage, func(done int) {
results := runParallel(o.ctx, images, o.probeImage, func(done int) {
o.report(PhaseImages, done, total)
})
for i, r := range results {
// 用户点了中止:停下归类,已分堆的交给 runScan 决定要不要搬
if o.ctx != nil && o.ctx.Err() != nil {
fmt.Println("=== 扫描已中止,跳过剩余图片归类")
break
}
f := images[i]
switch {
case r.psd:
@@ -418,10 +433,15 @@ func (o *Organizer) probeMedia(f scanner.File) mediaProbeResult {
func (o *Organizer) ProcessMedia(media []scanner.File) {
video, audio, fail, ignore, dup := 0, 0, 0, 0, 0
total := len(media)
results := runParallel(media, o.probeMedia, func(done int) {
results := runParallel(o.ctx, media, o.probeMedia, func(done int) {
o.report(PhaseMedia, done, total)
})
for i, r := range results {
// 用户点了中止:停下归类,已分堆的交给 runScan 决定要不要搬
if o.ctx != nil && o.ctx.Err() != nil {
fmt.Println("=== 扫描已中止,跳过剩余媒体归类")
break
}
f := media[i]
// ---------- 体检失败 ----------
if r.err != nil {
+20 -5
View File
@@ -7,6 +7,7 @@
package scanner
import (
"context"
"fmt"
"os"
"path/filepath"
@@ -38,11 +39,12 @@ type Result struct {
// Options 扫描的参数。
type Options struct {
RootDir string // 从哪个目录开始扫
SkipDir string // 这个目录整个跳过(整理输出目录,不能把自己整理的成果又扫一遍)
SkipDirNames []string // 名字叫这些的目录跳过(比如任意位置的「KMP已整理」)
SkipNames []string // 名字叫这些的文件跳过(数据库文件、ffprobe 程序等)
OnProgress func(done, total int) // 每处理完一些文件就报一次进度
Ctx context.Context // 中止信号:ctx 被取消时扫描立即停下(可为 nil)
RootDir string // 从哪个目录开始扫
SkipDir string // 这个目录整个跳过(整理输出目录,不能把自己整理的成果又扫一遍)
SkipDirNames []string // 名字叫这些的目录跳过(比如任意位置的「KMP已整理」)
SkipNames []string // 名字叫这些的文件跳过(数据库文件、ffprobe 程序等)
OnProgress func(done, total int) // 每处理完一些文件就报一次进度
}
// Scan 从根目录开始递归扫描,找出所有文件并识别每个文件的 MIME 类型。
@@ -58,6 +60,10 @@ func Scan(opts Options) (*Result, error) {
walkCount := 0
lastReport := time.Now()
err := filepath.Walk(opts.RootDir, func(currentPath string, info os.FileInfo, err error) error {
// 用户点了中止:立刻停下来,不继续翻目录
if opts.Ctx != nil && opts.Ctx.Err() != nil {
return opts.Ctx.Err()
}
if err != nil {
// 某个文件夹打不开(没权限/坏了)不影响大局,警告一声跳过它继续扫
fmt.Printf("=== 扫描跳过(无法访问): %s, 错误: %v\n", currentPath, err)
@@ -126,6 +132,11 @@ func Scan(opts Options) (*Result, error) {
go func() {
defer wg.Done()
for i := range jobs {
// 用户点了中止:不再认类型,但还要继续把通道里的任务消费完,
// 否则发任务的主 goroutine 会一直堵在发送上
if opts.Ctx != nil && opts.Ctx.Err() != nil {
continue
}
res.Files[i].Mime = detectMime(res.Files[i].Path)
if opts.OnProgress != nil {
opts.OnProgress(int(atomic.AddInt32(&done, 1)), total)
@@ -138,6 +149,10 @@ func Scan(opts Options) (*Result, error) {
}
close(jobs)
wg.Wait()
// 中止时把"已取消"当错误带回去,好让上层识别成"已中止"而不是正常完成
if opts.Ctx != nil && opts.Ctx.Err() != nil {
return res, opts.Ctx.Err()
}
return res, nil
}
+66 -19
View File
@@ -11,6 +11,7 @@ package service
import (
"bufio"
"context"
"errors"
"fmt"
"os"
@@ -34,16 +35,18 @@ import (
// ScanState 表示一次扫描整理任务当前处于什么状态。
type ScanState string
// 一次任务的四种状态:
// - idle 空闲:目前没有任务在跑
// - running 运行中:任务正在后台干活
// - done 完成:顺利干完了
// - error 出错:干活过程中出了岔子
// 一次任务的五种状态:
// - idle 空闲:目前没有任务在跑
// - running 运行中:任务正在后台干活
// - done 完成:顺利干完了
// - error 出错:干活过程中出了岔子
// - cancelled 已中止:用户手动点了中止
const (
ScanIdle ScanState = "idle"
ScanRunning ScanState = "running"
ScanDone ScanState = "done"
ScanError ScanState = "error"
ScanIdle ScanState = "idle"
ScanRunning ScanState = "running"
ScanDone ScanState = "done"
ScanError ScanState = "error"
ScanCancelled ScanState = "cancelled"
)
// ScanStatus 是发给前端的"当前进度快照",页面轮询 / 实时推送用的都是它。
@@ -107,6 +110,10 @@ type ScanService struct {
destDir string // 整理后文件的存放目录
lines []string // 运行日志
// 中止用的"停止开关":Cancel 调用 cancel() 后,管道各环节检查 ctx 立即停下。
ctx context.Context
cancel context.CancelFunc
stage string // 当前阶段:scan/organize/move
done int // 当前阶段已完成数
total int // 当前阶段总数
@@ -180,6 +187,8 @@ func (s *ScanService) Start(rootDir, destDir string) error {
s.stage = stageScan
s.done = 0
s.total = 0
// 建一个可中止的上下文:页面点「中止扫描」时调用 cancel(),各环节就会停下
s.ctx, s.cancel = context.WithCancel(context.Background())
s.mu.Unlock()
// 把存放目录记到数据库里,下次打开页面自动填上,省得用户重敲
@@ -190,7 +199,21 @@ func (s *ScanService) Start(rootDir, destDir string) error {
}
// 开个后台线程干活,这个函数立刻返回
go s.run(rootDir, destDir)
go s.run(s.ctx, rootDir, destDir)
return nil
}
// Cancel 中止正在运行的扫描整理任务。
// 调用后扫描会在最近的检查点停下,已归类但还没搬走的文件会留在原处。
func (s *ScanService) Cancel() error {
s.mu.Lock()
defer s.mu.Unlock()
if s.state != ScanRunning {
return errors.New("当前没有正在运行的扫描任务")
}
if s.cancel != nil {
s.cancel()
}
return nil
}
@@ -297,7 +320,7 @@ func resolveFFmpegPath(configured string) string {
//
// 做法:把标准输出 os.Stdout 临时换成一根管道(pipe)的写端,
// 另起一个 goroutine 从管道的读端一行行读日志。干完活再把 Stdout 换回去。
func (s *ScanService) run(rootDir, destDir string) {
func (s *ScanService) run(ctx context.Context, rootDir, destDir string) {
// 创建一根管道,准备截取日志
r, w, err := os.Pipe()
if err != nil {
@@ -341,13 +364,18 @@ func (s *ScanService) run(rootDir, destDir string) {
}()
// 真正开始干活。期间程序里所有 fmt.Println/Printf 都会进上面的管道
scanErr := s.runScan(rootDir, destDir)
scanErr := s.runScan(ctx, rootDir, destDir)
// 干完了:把 Stdout 换回去,关上管道,等读日志的 goroutine 把剩下的读完
os.Stdout = oldStdout
_ = w.Close()
<-done
// 先看是不是用户主动中止:ctx 已被取消 → 报"已中止",而不是当异常
if ctx.Err() != nil {
s.finish(ScanCancelled, "扫描已中止")
return
}
if scanErr != nil {
s.finish(ScanError, scanErr.Error())
return
@@ -355,11 +383,17 @@ func (s *ScanService) run(rootDir, destDir string) {
s.finish(ScanDone, "扫描整理完成")
}
// finish 把任务状态改成最终状态(完成或出错),并立刻推给前端。
// finish 把任务状态改成最终状态(完成/出错/中止),并立刻推给前端。
// 顺便把中止用的上下文关掉、清空引用,避免留着占资源、也防止状态错乱。
func (s *ScanService) finish(state ScanState, msg string) {
s.mu.Lock()
s.state = state
s.msg = msg
if s.cancel != nil {
s.cancel()
s.cancel = nil
}
s.ctx = nil
s.mu.Unlock()
s.publish()
}
@@ -401,11 +435,12 @@ func sanitizeLine(line string) string {
// 2. 过滤:把"值得处理"的文件分成图片、视频/音频两拨
// 3. 准备解码器:把内嵌的 ffprobe 程序释放到当前目录(用它读视频/音频信息)
// 4. 整理:给每个文件探测信息 → 判断是否重复 → 分到分类目录 → 搬进去
func (s *ScanService) runScan(rootDir, destDir string) error {
// 确定整理后文件的存放目录:用户没填就默认放在扫描目录下的「KMP已整理」
pickerDir := destDir
if pickerDir == "" {
pickerDir = filepath.Join(rootDir, DefaultPickerDirName)
func (s *ScanService) runScan(ctx context.Context, rootDir, destDir string) error {
// 确定整理后文件的存放目录:统一都放进「KMP已整理」这一层——
// 用户填了存放目录就放在它下面(存放目录/KMP已整理),没填就放在扫描目录下(扫描目录/KMP已整理)。
pickerDir := filepath.Join(rootDir, DefaultPickerDirName)
if destDir != "" {
pickerDir = filepath.Join(destDir, DefaultPickerDirName)
}
fmt.Printf("=== 扫描根目录: %s \n", rootDir)
fmt.Printf("=== 整理输出目录(白名单): %s \n", pickerDir)
@@ -414,6 +449,7 @@ func (s *ScanService) runScan(rootDir, destDir string) error {
// 扫描时跳过这些目录/文件:整理输出目录本身、数据库文件、ffprobe 程序,
// 不然它们会被当成"媒体文件"被自己整理进去,那可就乱套了。
res, err := scanner.Scan(scanner.Options{
Ctx: ctx,
RootDir: rootDir,
SkipDir: pickerDir,
SkipDirNames: []string{DefaultPickerDirName},
@@ -421,6 +457,9 @@ func (s *ScanService) runScan(rootDir, destDir string) error {
OnProgress: func(done, total int) { s.setProgress(stageScan, done, total) },
})
if err != nil {
if ctx.Err() != nil {
return ctx.Err() // 中止的"扫描失败"不当异常,run 里会识别成已中止
}
return fmt.Errorf("扫描失败: %v", err)
}
@@ -453,7 +492,7 @@ func (s *ScanService) runScan(rootDir, destDir string) error {
fmt.Printf("=== 使用 ffmpeg 转正图片EXIF旋转: %s\n", ffmpegPath)
}
}
org := organize.New(pickerDir, ffprobePath, ffmpegPath, s.store)
org := organize.New(pickerDir, ffprobePath, ffmpegPath, s.store, ctx)
// 整个"整理"分两个阶段,界面上共用一条进度条:
// 归类阶段:图片、视频/音频分两批处理。进度 = 已处理文件数 / 全部待整理文件数
@@ -477,9 +516,17 @@ func (s *ScanService) runScan(rootDir, destDir string) error {
if len(res.Images) > 0 {
org.ProcessImages(res.Images)
}
if ctx.Err() != nil {
fmt.Println("=== 扫描已中止,跳过视频/音频整理与移动")
return ctx.Err()
}
if len(res.Media) > 0 && ffprobePath != "" {
org.ProcessMedia(res.Media)
}
if ctx.Err() != nil {
fmt.Println("=== 扫描已中止,跳过文件移动")
return ctx.Err()
}
// 移动:把文件搬进各自的分类目录,并记录去向
org.Flush()