refactor: 全面重构代码并加白话注释,移动全局并行,去掉去向/重复记录
- 重构后端与前端,全程大白话注释(不懂程序也能看懂) - 移动阶段改为全局并行(MoveAll,工人数=CPU核心数) - 去掉记录每个文件搬去哪了(删除去向回填,解决写库卡顿) - 去掉重复文件标记与统计:重复判定改为 MD5+路径都相同才算,重复则原地忽略不移动;MD5 相同但路径不同仍照常搬 - 媒体库去掉归类目录列,看板去掉重复/归类分布统计
This commit is contained in:
+19
-9
@@ -1,24 +1,31 @@
|
||||
// 状态订阅发布:将 ScanService 的状态变更实时推送给订阅者(WebSocket 等)。
|
||||
// 状态订阅/发布中心:把 ScanService 的状态变更实时"广播"给所有订阅者。
|
||||
//
|
||||
// 大白话理解:这是个广播站。后台干活时进度一变,就通过广播站喊一嗓子;
|
||||
// 谁订阅了(比如 WebSocket 连接),谁就收到。没有人听的时候也不影响干活。
|
||||
//
|
||||
// 还有个贴心小设计:每个订阅者面前的信箱最多存一条消息。
|
||||
// 如果进度变得太快、接收方来不及处理,就只留最新一条——反正看进度的人
|
||||
// 关心的是"现在到哪了",而不是"中间每一刻到哪了"。
|
||||
package service
|
||||
|
||||
import "sync"
|
||||
|
||||
// StatusListener 状态订阅通道(缓冲 1,保底容纳一条未消费状态)
|
||||
// StatusListener 订阅者的"信箱"(一个缓冲 1 的管道)。
|
||||
type StatusListener chan ScanStatus
|
||||
|
||||
// StatusHub 管理状态订阅者;发布时若消费者来不及消费则丢弃旧消息,
|
||||
// 保证订阅者拿到的始终是最新状态。
|
||||
// StatusHub 广播站:管理所有订阅者,负责把状态推给每个人。
|
||||
type StatusHub struct {
|
||||
mu sync.RWMutex
|
||||
subs map[StatusListener]struct{}
|
||||
mu sync.RWMutex // 保护订阅者名单
|
||||
subs map[StatusListener]struct{} // 当前所有订阅者
|
||||
}
|
||||
|
||||
// NewStatusHub 创建状态订阅中心
|
||||
// NewStatusHub 建一个广播站。
|
||||
func NewStatusHub() *StatusHub {
|
||||
return &StatusHub{subs: make(map[StatusListener]struct{})}
|
||||
}
|
||||
|
||||
// Subscribe 注册订阅,返回频道与取消函数。取消后频道被关闭,写方需停止发送。
|
||||
// Subscribe 让调用方订阅进度。返回它的信箱,和一个"退订"函数。
|
||||
// 退订后信箱会被关闭(写方要自己注意别再往里面塞)。
|
||||
func (h *StatusHub) Subscribe() (StatusListener, func()) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
@@ -34,7 +41,9 @@ func (h *StatusHub) Subscribe() (StatusListener, func()) {
|
||||
}
|
||||
}
|
||||
|
||||
// Publish 向所有订阅者推送最新状态(非阻塞;缓冲满时覆盖为最新状态)
|
||||
// Publish 把最新状态推给所有订阅者。
|
||||
// 谁的信箱满了(来不及消费),就把里面的旧消息挤掉、换上最新的一条。
|
||||
// 整个过程不阻塞,广播站永远不会因为某个订阅者太慢而卡住。
|
||||
func (h *StatusHub) Publish(st ScanStatus) {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
@@ -42,6 +51,7 @@ func (h *StatusHub) Publish(st ScanStatus) {
|
||||
select {
|
||||
case ch <- st:
|
||||
default:
|
||||
// 信箱满了:清掉旧消息,放最新的进去
|
||||
select {
|
||||
case <-ch:
|
||||
default:
|
||||
|
||||
+184
-91
@@ -1,4 +1,12 @@
|
||||
// Package service 提供 Web 服务端的后台任务编排(媒体库扫描整理)。
|
||||
// Package service 是"扫描整理"这个后台任务的管家。
|
||||
//
|
||||
// 你可以把它理解成一家快递分拣中心的调度室:
|
||||
// - 页面上点「开始扫描」,调度室就把整批活接下来,开一个后台线程慢慢干;
|
||||
// - 干到哪一步(扫描 / 归类 / 移动)、干了多少个、还要多久,调度室都记下来;
|
||||
// - 前端页面通过 WebSocket 实时看到这些进度,也能随时问一句"现在啥情况"。
|
||||
//
|
||||
// 这个包不自己干活,真正的"扫描、整理、移动"分别由 scanner、organize、mover
|
||||
// 三个包完成,这里只负责:发号施令、记录进度、把日志发给前端看。
|
||||
package service
|
||||
|
||||
import (
|
||||
@@ -18,9 +26,16 @@ import (
|
||||
"kenaito-media-picker/internal/store"
|
||||
)
|
||||
|
||||
// ScanState 扫描状态
|
||||
// ============================ 扫描状态 ============================
|
||||
|
||||
// ScanState 表示一次扫描整理任务当前处于什么状态。
|
||||
type ScanState string
|
||||
|
||||
// 一次任务的四种状态:
|
||||
// - idle 空闲:目前没有任务在跑
|
||||
// - running 运行中:任务正在后台干活
|
||||
// - done 完成:顺利干完了
|
||||
// - error 出错:干活过程中出了岔子
|
||||
const (
|
||||
ScanIdle ScanState = "idle"
|
||||
ScanRunning ScanState = "running"
|
||||
@@ -28,66 +43,93 @@ const (
|
||||
ScanError ScanState = "error"
|
||||
)
|
||||
|
||||
// ScanStatus 扫描进度与结果(供前端轮询)
|
||||
// ScanStatus 是发给前端的"当前进度快照",页面轮询 / 实时推送用的都是它。
|
||||
type ScanStatus struct {
|
||||
State ScanState `json:"state"` // idle/running/done/error
|
||||
Message string `json:"message"` // 简要说明
|
||||
Started string `json:"started"` // 开始时间
|
||||
Elapsed int64 `json:"elapsed"` // 已运行秒数
|
||||
RootDir string `json:"rootDir"` // 本次扫描的根目录
|
||||
DestDir string `json:"destDir"` // 上次使用的存放目录
|
||||
Lines []string `json:"lines"` // 运行日志
|
||||
Stage string `json:"stage"` // 当前阶段:scan/organize/move
|
||||
StageText string `json:"stageText"` // 阶段中文说明
|
||||
Progress int `json:"progress"` // 总体进度百分比 0-100
|
||||
Done int `json:"done"` // 已处理数
|
||||
Total int `json:"total"` // 总数
|
||||
Eta int64 `json:"eta"` // 预计剩余秒数(按当前阶段速率估算,0 表示无法估算)
|
||||
State ScanState `json:"state"` // 任务状态:idle/running/done/error
|
||||
Message string `json:"message"` // 一句话说明,比如出错原因
|
||||
Started string `json:"started"` // 任务开始时间
|
||||
Elapsed int64 `json:"elapsed"` // 已经运行了多少秒
|
||||
RootDir string `json:"rootDir"` // 本次扫描的是哪个目录
|
||||
DestDir string `json:"destDir"` // 整理后的文件放到哪个目录(上次用过的,下次自动填上)
|
||||
Lines []string `json:"lines"` // 运行日志,逐行显示给用户看
|
||||
Stage string `json:"stage"` // 当前干到哪一步:scan/organize/move
|
||||
StageText string `json:"stageText"` // 当前步骤的中文名
|
||||
Progress int `json:"progress"` // 总进度百分比 0~100
|
||||
Done int `json:"done"` // 这一步已经处理了多少个文件
|
||||
Total int `json:"total"` // 这一步一共要处理多少个文件
|
||||
Eta int64 `json:"eta"` // 预计还要多少秒(0 表示算不出来)
|
||||
}
|
||||
|
||||
// 应用级配置键
|
||||
const (
|
||||
ConfigKeyDestDir = "scan.destDir"
|
||||
)
|
||||
// ============================ 应用级常量 ============================
|
||||
|
||||
// DefaultPickerDirName 默认存放目录名(未指定 destDir 时创建于扫描目录下)
|
||||
// 应用级配置键。目前只存了一个:上次使用的存放目录,方便下次打开页面自动回填。
|
||||
const ConfigKeyDestDir = "scan.destDir"
|
||||
|
||||
// DefaultPickerDirName 用户没指定存放目录时,默认在扫描目录下新建这个文件夹装整理好的文件。
|
||||
const DefaultPickerDirName = "KMP已整理"
|
||||
|
||||
// ScanService 管理媒体库扫描整理任务(同一时间只允许一次)
|
||||
type ScanService struct {
|
||||
store *store.Store
|
||||
hub *StatusHub
|
||||
mu sync.Mutex
|
||||
state ScanState
|
||||
msg string
|
||||
started time.Time
|
||||
rootDir string
|
||||
destDir string
|
||||
lines []string
|
||||
stage string
|
||||
done int
|
||||
total int
|
||||
stageStart time.Time
|
||||
stageDone int
|
||||
lastPublish time.Time
|
||||
}
|
||||
// 三个工作阶段的名字。前端根据这个名字显示"正在扫描/正在整理/正在移动"。
|
||||
const (
|
||||
stageScan = "scan"
|
||||
stageOrganize = "organize"
|
||||
stageMove = "move"
|
||||
)
|
||||
|
||||
// publishInterval 运行中状态推送的节流间隔;非运行状态(done/error)不受限,保证结束状态必达
|
||||
// publishInterval 进度推送的节流间隔。
|
||||
// 扫描大量文件时进度变化非常频繁,如果每次都推给前端,WebSocket 会被刷爆。
|
||||
// 这里限制为每 0.1 秒最多推一次(约每秒 10 次),既实时又不至于泛滥。
|
||||
// 注意:任务结束(done/error)不受这个限制,必须立刻推出去,否则用户看不到结果。
|
||||
const publishInterval = 100 * time.Millisecond
|
||||
|
||||
// NewScanService 创建扫描服务;store 可为 nil(数据库不可用时仅不记录媒体信息)
|
||||
// maxLogLines 日志最多保留的行数。日志太长就丢掉最旧的一半,防止占用太多内存、拖慢页面。
|
||||
const maxLogLines = 500
|
||||
|
||||
// ============================ 扫描服务 ============================
|
||||
|
||||
// ScanService 是扫描整理任务的管家,负责:
|
||||
// - 开始任务(同一时间只允许一个任务在跑)
|
||||
// - 记录任务进度、当前阶段、运行日志
|
||||
// - 通过 StatusHub 把进度实时推给订阅者(前端 WebSocket)
|
||||
//
|
||||
// 用锁(mu)保护这些共享数据,因为后台干活线程和前端查询线程会同时读写。
|
||||
type ScanService struct {
|
||||
store *store.Store // 数据库;为 nil 表示数据库打不开,此时扫描照跑,只是不记录媒体信息
|
||||
hub *StatusHub // 订阅中心:把进度推给所有订阅者
|
||||
|
||||
mu sync.Mutex // 下面这些字段都会被多线程读写,所以用这把锁保护
|
||||
state ScanState // 当前任务状态
|
||||
msg string // 附加说明(一般存出错原因)
|
||||
started time.Time // 任务开始时间
|
||||
rootDir string // 本次扫描的根目录
|
||||
destDir string // 整理后文件的存放目录
|
||||
lines []string // 运行日志
|
||||
|
||||
stage string // 当前阶段:scan/organize/move
|
||||
done int // 当前阶段已完成数
|
||||
total int // 当前阶段总数
|
||||
|
||||
// 估算剩余时间用:记住当前阶段开始的时间点和当时已完成的数,
|
||||
// 这样就能算出"每秒干几个",进而估算"还剩多少秒"。
|
||||
stageStart time.Time
|
||||
stageDone int
|
||||
|
||||
lastPublish time.Time // 上次推送进度的时间,用于节流
|
||||
}
|
||||
|
||||
// NewScanService 创建扫描服务。
|
||||
// 注意 store 可以传 nil——数据库打不开时服务照常能用,只是不记录媒体信息。
|
||||
func NewScanService(st *store.Store) *ScanService {
|
||||
return &ScanService{store: st, hub: NewStatusHub(), state: ScanIdle}
|
||||
}
|
||||
|
||||
// Subscribe 订阅扫描状态变更(供 WebSocket 等实时推送);取消后返回的频道被关闭
|
||||
// Subscribe 让调用方(通常是 WebSocket 处理器)订阅进度更新。
|
||||
// 返回值第一个是"收进度"的管道,第二个是取消订阅的函数(取消后管道会被关闭)。
|
||||
func (s *ScanService) Subscribe() (StatusListener, func()) {
|
||||
return s.hub.Subscribe()
|
||||
}
|
||||
|
||||
// publish 将当前状态推送给所有订阅者。
|
||||
// 运行中按 publishInterval 节流(约 10 次/秒),避免海量文件时 WebSocket 消息泛滥;
|
||||
// 非运行状态(done/error)无条件推送,确保结束状态不丢失。
|
||||
// publish 把当前进度推给所有订阅者。
|
||||
// 运行中按 publishInterval 节流;任务结束时无条件推送,保证"干完了"这个消息一定送达。
|
||||
func (s *ScanService) publish() {
|
||||
s.mu.Lock()
|
||||
if s.state == ScanRunning && time.Since(s.lastPublish) < publishInterval {
|
||||
@@ -99,15 +141,24 @@ func (s *ScanService) publish() {
|
||||
s.hub.Publish(s.Status())
|
||||
}
|
||||
|
||||
// Start 启动一次扫描整理;已在运行时返回错误。rootDir 为空时默认扫描当前运行目录,
|
||||
// destDir 为空时默认输出到扫描目录下的「KMP已整理」文件夹。
|
||||
// Start 开始一次扫描整理任务。
|
||||
// - rootDir:要扫描哪个目录。留空就用程序当前所在的目录。
|
||||
// - destDir:整理好的文件放哪。留空就默认放到扫描目录下的「KMP已整理」。
|
||||
//
|
||||
// 任务在后台 goroutine 里跑,这个函数只负责"接下活"就立即返回,不阻塞页面。
|
||||
func (s *ScanService) Start(rootDir, destDir string) error {
|
||||
s.mu.Lock()
|
||||
|
||||
// 正在跑就不能再开一个,直接拒绝
|
||||
if s.state == ScanRunning {
|
||||
s.mu.Unlock()
|
||||
return errors.New("扫描整理正在运行中,请等待完成")
|
||||
}
|
||||
|
||||
// 把目录变成标准绝对路径;留空就取当前工作目录
|
||||
rootDir = resolveRootDir(rootDir)
|
||||
|
||||
// 先确认要扫描的目录真的存在,不然白忙活
|
||||
if st, err := os.Stat(rootDir); err != nil || !st.IsDir() {
|
||||
s.mu.Unlock()
|
||||
return fmt.Errorf("扫描目录不存在或不可访问: %s", rootDir)
|
||||
@@ -115,36 +166,43 @@ func (s *ScanService) Start(rootDir, destDir string) error {
|
||||
if destDir != "" {
|
||||
destDir = resolveRootDir(destDir)
|
||||
}
|
||||
|
||||
// 把状态改成"运行中",清空上一次的残留信息
|
||||
s.state = ScanRunning
|
||||
s.msg = ""
|
||||
s.started = time.Now()
|
||||
s.rootDir = rootDir
|
||||
s.destDir = destDir
|
||||
s.lines = nil
|
||||
s.stage = "scan"
|
||||
s.stage = stageScan
|
||||
s.done = 0
|
||||
s.total = 0
|
||||
s.mu.Unlock()
|
||||
|
||||
// 持久化存放目录,下次打开页面自动回填
|
||||
// 把存放目录记到数据库里,下次打开页面自动填上,省得用户重敲
|
||||
if s.store != nil {
|
||||
if err := s.store.SaveConfig(ConfigKeyDestDir, destDir); err != nil {
|
||||
fmt.Printf("=== 保存存放目录配置失败: %v\n", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 开个后台线程干活,这个函数立刻返回
|
||||
go s.run(rootDir, destDir)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Status 返回当前扫描状态
|
||||
// Status 生成一份"当前进度快照"返回给前端。
|
||||
func (s *ScanService) Status() ScanStatus {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
// 算已经跑了多少秒
|
||||
elapsed := int64(0)
|
||||
if !s.started.IsZero() {
|
||||
elapsed = int64(time.Since(s.started).Seconds())
|
||||
}
|
||||
|
||||
// 算百分比:已完成 / 总数 × 100,最多 100
|
||||
progress := 0
|
||||
if s.total > 0 {
|
||||
progress = int(float64(s.done) / float64(s.total) * 100)
|
||||
@@ -152,13 +210,15 @@ func (s *ScanService) Status() ScanStatus {
|
||||
progress = 100
|
||||
}
|
||||
}
|
||||
st := ScanStatus{
|
||||
State: s.state,
|
||||
Message: s.msg,
|
||||
Started: s.started.Format("2006-01-02 15:04:05"),
|
||||
Elapsed: elapsed,
|
||||
RootDir: s.rootDir,
|
||||
DestDir: s.destDir,
|
||||
|
||||
return ScanStatus{
|
||||
State: s.state,
|
||||
Message: s.msg,
|
||||
Started: s.started.Format("2006-01-02 15:04:05"),
|
||||
Elapsed: elapsed,
|
||||
RootDir: s.rootDir,
|
||||
DestDir: s.destDir,
|
||||
// 复制一份日志再返回,避免前端拿到后修改到我们内部的数据
|
||||
Lines: append([]string{}, s.lines...),
|
||||
Stage: s.stage,
|
||||
Progress: progress,
|
||||
@@ -166,10 +226,12 @@ func (s *ScanService) Status() ScanStatus {
|
||||
Total: s.total,
|
||||
Eta: s.estimateEtaLocked(),
|
||||
}
|
||||
return st
|
||||
}
|
||||
|
||||
// estimateEtaLocked 按当前阶段速率估算剩余秒数;需持有锁。返回 0 表示无法估算。
|
||||
// estimateEtaLocked 估算"还剩多少秒",规则是:记住这个阶段开始的时间,
|
||||
// 看这段时间干了多少个文件,算出速度,再用剩下的数量除以速度。
|
||||
// 返回 0 表示还估不出来(刚开跑、或者数量太小)。
|
||||
// 注意:调用前必须已经持有锁。
|
||||
func (s *ScanService) estimateEtaLocked() int64 {
|
||||
if s.state != ScanRunning || s.total <= 0 || s.done <= 0 || s.done >= s.total {
|
||||
return 0
|
||||
@@ -181,7 +243,7 @@ func (s *ScanService) estimateEtaLocked() int64 {
|
||||
if elapsedSec <= 0 {
|
||||
return 0
|
||||
}
|
||||
rate := float64(s.done-s.stageDone) / elapsedSec // 文件/秒
|
||||
rate := float64(s.done-s.stageDone) / elapsedSec // 每秒处理多少个文件
|
||||
if rate <= 0 {
|
||||
return 0
|
||||
}
|
||||
@@ -192,7 +254,7 @@ func (s *ScanService) estimateEtaLocked() int64 {
|
||||
return int64(remain + 0.5)
|
||||
}
|
||||
|
||||
// resolveRootDir 解析扫描根目录:空值回退当前工作目录,并做绝对路径归一化
|
||||
// resolveRootDir 把用户填的目录整理成标准绝对路径;留空就取程序当前所在目录。
|
||||
func resolveRootDir(dir string) string {
|
||||
if strings.TrimSpace(dir) == "" {
|
||||
wd, err := os.Getwd()
|
||||
@@ -208,11 +270,15 @@ func resolveRootDir(dir string) string {
|
||||
return abs
|
||||
}
|
||||
|
||||
// maxLogLines 日志保留的最大行数,超出后丢弃最旧行,避免前端渲染卡顿
|
||||
const maxLogLines = 500
|
||||
// ============================ 后台任务本体 ============================
|
||||
|
||||
// run 后台执行:捕获 stdout 日志流式写入 lines,结束后更新状态
|
||||
// run 是后台干活线程。除了调用 runScan 真正干活外,它还要做一件特殊的事:
|
||||
// 把程序打印到屏幕上的日志"截胡"下来,存进 s.lines,好让前端能看到实时日志。
|
||||
//
|
||||
// 做法:把标准输出 os.Stdout 临时换成一根管道(pipe)的写端,
|
||||
// 另起一个 goroutine 从管道的读端一行行读日志。干完活再把 Stdout 换回去。
|
||||
func (s *ScanService) run(rootDir, destDir string) {
|
||||
// 创建一根管道,准备截取日志
|
||||
r, w, err := os.Pipe()
|
||||
if err != nil {
|
||||
s.finish(ScanError, "创建日志管道失败: "+err.Error())
|
||||
@@ -221,11 +287,12 @@ func (s *ScanService) run(rootDir, destDir string) {
|
||||
oldStdout := os.Stdout
|
||||
os.Stdout = w
|
||||
|
||||
// 专门读日志的 goroutine:读到一行就存进 s.lines
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
sc := bufio.NewScanner(r)
|
||||
sc.Buffer(make([]byte, 1024*1024), 1024*1024)
|
||||
sc.Buffer(make([]byte, 1024*1024), 1024*1024) // 允许很长的日志行
|
||||
for sc.Scan() {
|
||||
line := sanitizeLine(sc.Text())
|
||||
if line == "" {
|
||||
@@ -233,7 +300,7 @@ func (s *ScanService) run(rootDir, destDir string) {
|
||||
}
|
||||
s.mu.Lock()
|
||||
if len(s.lines) >= maxLogLines {
|
||||
// 丢弃最旧一半,保留最近日志,控制内存与前端渲染规模
|
||||
// 日志太多就丢最旧一半,只留最近的部分
|
||||
s.lines = append([]string{}, s.lines[len(s.lines)/2:]...)
|
||||
}
|
||||
s.lines = append(s.lines, line)
|
||||
@@ -241,8 +308,10 @@ func (s *ScanService) run(rootDir, destDir string) {
|
||||
}
|
||||
}()
|
||||
|
||||
// 真正开始干活。期间程序里所有 fmt.Println/Printf 都会进上面的管道
|
||||
scanErr := s.runScan(rootDir, destDir)
|
||||
|
||||
// 干完了:把 Stdout 换回去,关上管道,等读日志的 goroutine 把剩下的读完
|
||||
os.Stdout = oldStdout
|
||||
_ = w.Close()
|
||||
<-done
|
||||
@@ -254,7 +323,7 @@ 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
|
||||
@@ -263,15 +332,16 @@ func (s *ScanService) finish(state ScanState, msg string) {
|
||||
s.publish()
|
||||
}
|
||||
|
||||
// setProgress 更新阶段与进度(线程安全,供各阶段回调调用)。
|
||||
// 阶段切换或进度回退(新阶段从头计数)时重置阶段起点,用于估算剩余时间。
|
||||
// setProgress 更新当前阶段与进度,并推给前端。
|
||||
// 它还负责记录"阶段开始点",用于估算剩余时间:
|
||||
// - 阶段换了(比如从归类切到移动)→ 重新记起点
|
||||
// - 阶段没换但进度倒退了(说明开始新一轮计数)→ 也重新记起点
|
||||
func (s *ScanService) setProgress(stage string, done, total int) {
|
||||
s.mu.Lock()
|
||||
if stage != s.stage {
|
||||
s.stageStart = time.Now()
|
||||
s.stageDone = done
|
||||
} else if done < s.stageDone {
|
||||
// 阶段未变但进度回退(如开始新一轮计数),重置起点
|
||||
s.stageStart = time.Now()
|
||||
s.stageDone = done
|
||||
}
|
||||
@@ -282,18 +352,25 @@ func (s *ScanService) setProgress(stage string, done, total int) {
|
||||
s.publish()
|
||||
}
|
||||
|
||||
// ansiRe 匹配 ANSI 颜色转义序列
|
||||
// ansiRe 用来去掉日志里的 ANSI 颜色码(那些 [31m 之类的转义序列)。
|
||||
var ansiRe = regexp.MustCompile(`\x1b\[[0-9;]*m`)
|
||||
|
||||
// sanitizeLine 清理日志行:去除 ANSI 颜色码,进度条按 \r 分段只保留最后一段
|
||||
// sanitizeLine 清理一行日志:
|
||||
// - 去掉颜色码
|
||||
// - 有些进度条用 \r(回车)原地刷新,这里按 \r 切开只留最后一段
|
||||
func sanitizeLine(line string) string {
|
||||
line = ansiRe.ReplaceAllString(line, "")
|
||||
segs := strings.Split(line, "\r")
|
||||
return strings.TrimSpace(segs[len(segs)-1])
|
||||
}
|
||||
|
||||
// runScan 执行完整整理流程:扫描 → 过滤 → 释放解码器 → 归类 → 移动
|
||||
// runScan 执行完整的整理流程,一共四步:
|
||||
// 1. 扫描:把目录里的文件全部找出来,识别每个文件的类型
|
||||
// 2. 过滤:把"值得处理"的文件分成图片、视频/音频两拨
|
||||
// 3. 准备解码器:把内嵌的 ffprobe 程序释放到当前目录(用它读视频/音频信息)
|
||||
// 4. 整理:给每个文件探测信息 → 判断是否重复 → 分到分类目录 → 搬进去
|
||||
func (s *ScanService) runScan(rootDir, destDir string) error {
|
||||
// 确定整理后文件的存放目录:用户没填就默认放在扫描目录下的「KMP已整理」
|
||||
pickerDir := destDir
|
||||
if pickerDir == "" {
|
||||
pickerDir = filepath.Join(rootDir, DefaultPickerDirName)
|
||||
@@ -301,19 +378,27 @@ func (s *ScanService) runScan(rootDir, destDir string) error {
|
||||
fmt.Printf("=== 扫描根目录: %s \n", rootDir)
|
||||
fmt.Printf("=== 整理输出目录(白名单): %s \n", pickerDir)
|
||||
|
||||
// ---------- 第一步:扫描 ----------
|
||||
// 扫描时跳过这些目录/文件:整理输出目录本身、数据库文件、ffprobe 程序,
|
||||
// 不然它们会被当成"媒体文件"被自己整理进去,那可就乱套了。
|
||||
res, err := scanner.Scan(scanner.Options{
|
||||
RootDir: rootDir,
|
||||
SkipDir: pickerDir,
|
||||
SkipDirNames: []string{DefaultPickerDirName},
|
||||
SkipNames: []string{store.DBFileName, probe.FFprobeExeName},
|
||||
OnProgress: func(done, total int) { s.setProgress("scan", done, total) },
|
||||
OnProgress: func(done, total int) { s.setProgress(stageScan, done, total) },
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("扫描失败: %v", err)
|
||||
}
|
||||
|
||||
// ---------- 第二步:过滤 ----------
|
||||
// 把扫描到的文件分成"图片"和"视频/音频候选"两拨
|
||||
res.Filter()
|
||||
|
||||
// 释放内嵌 ffprobe;失败只跳过媒体整理,图片照常处理
|
||||
// ---------- 第三步:准备解码器 ----------
|
||||
// 读取视频/音频信息需要 ffprobe。程序里内嵌了一份,用之前先释放到当前目录。
|
||||
// 释放失败就跳过视频/音频整理,图片照常处理。
|
||||
ffprobePath := ""
|
||||
if (len(res.Images) > 0 || len(res.Media) > 0) && probe.HasEmbeddedFFprobe() {
|
||||
if err := probe.ReleaseFFprobe(); err != nil {
|
||||
@@ -324,32 +409,40 @@ func (s *ScanService) runScan(rootDir, destDir string) error {
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- 第四步:整理 ----------
|
||||
org := organize.New(pickerDir, ffprobePath, s.store)
|
||||
totalWork := len(res.Images) + len(res.Media)
|
||||
doneWork := 0
|
||||
currentStage := "organize"
|
||||
org.SetProgress(func(done, total int) {
|
||||
// organize 阶段:doneWork 累计图片/媒体已处理数,加上本阶段内 done;
|
||||
// move 阶段:Flush 上报的 done 已是累计移动数,直接使用,避免叠加导致溢出。
|
||||
if currentStage == "move" {
|
||||
s.setProgress("move", done, total)
|
||||
} else {
|
||||
s.setProgress(currentStage, doneWork+done, totalWork)
|
||||
|
||||
// 整个"整理"分两个阶段,界面上共用一条进度条:
|
||||
// 归类阶段:图片、视频/音频分两批处理。进度 = 已处理文件数 / 全部待整理文件数
|
||||
// 移动阶段:把所有文件搬进分类目录。进度 = 已移动数 / 待移动数
|
||||
imageCount := len(res.Images)
|
||||
totalWork := imageCount + len(res.Media)
|
||||
org.SetProgress(func(phase string, done, total int) {
|
||||
switch phase {
|
||||
case organize.PhaseImages:
|
||||
// 图片阶段:媒体还没开始,全局进度 = 本阶段进度
|
||||
s.setProgress(stageOrganize, done, totalWork)
|
||||
case organize.PhaseMedia:
|
||||
// 媒体阶段:全局进度 = 图片数量 + 本阶段进度
|
||||
s.setProgress(stageOrganize, imageCount+done, totalWork)
|
||||
case organize.PhaseMove:
|
||||
// 移动阶段:进度单独统计(已移动数 / 待移动数)
|
||||
s.setProgress(stageMove, done, total)
|
||||
}
|
||||
})
|
||||
|
||||
if len(res.Images) > 0 {
|
||||
org.ProcessImages(res.Images)
|
||||
doneWork += len(res.Images)
|
||||
}
|
||||
if len(res.Media) > 0 && ffprobePath != "" {
|
||||
org.ProcessMedia(res.Media)
|
||||
doneWork += len(res.Media)
|
||||
}
|
||||
// 移动阶段:进度为累计已移动文件数
|
||||
currentStage = "move"
|
||||
|
||||
// 移动:把文件搬进各自的分类目录,并记录去向
|
||||
org.Flush()
|
||||
currentStage = "done"
|
||||
s.setProgress("move", totalWork, totalWork)
|
||||
|
||||
// 把进度条顶满 100%,再打印"整理完成"
|
||||
s.setProgress(stageMove, totalWork, totalWork)
|
||||
fmt.Println("=== 整理完成")
|
||||
return nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user