764 lines
26 KiB
Go
764 lines
26 KiB
Go
// Package service 是"扫描整理"这个后台任务的管家。
|
||
//
|
||
// 你可以把它理解成一家快递分拣中心的调度室：
|
||
// - 页面上点「开始扫描」，调度室就把整批活接下来，开一个后台线程慢慢干；
|
||
// - 干到哪一步（扫描 / 归类 / 移动）、干了多少个、还要多久，调度室都记下来；
|
||
// - 前端页面通过 WebSocket 实时看到这些进度，也能随时问一句"现在啥情况"。
|
||
//
|
||
// 这个包不自己干活，真正的"扫描、整理、移动"分别由 scanner、organize、mover
|
||
// 三个包完成，这里只负责：发号施令、记录进度、把日志发给前端看。
|
||
package service
|
||
|
||
import (
|
||
"bufio"
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"os"
|
||
"os/exec"
|
||
"path/filepath"
|
||
"regexp"
|
||
"runtime/debug"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"kenaito-media-picker/internal/anime_nsfw"
|
||
"kenaito-media-picker/internal/config"
|
||
"kenaito-media-picker/internal/mover"
|
||
"kenaito-media-picker/internal/nsfw"
|
||
"kenaito-media-picker/internal/organize"
|
||
"kenaito-media-picker/internal/probe"
|
||
"kenaito-media-picker/internal/scanner"
|
||
"kenaito-media-picker/internal/store"
|
||
)
|
||
|
||
// ============================ 扫描状态 ============================
|
||
|
||
// ScanState 表示一次扫描整理任务当前处于什么状态。
|
||
type ScanState string
|
||
|
||
// 一次任务的五种状态：
|
||
// - idle 空闲：目前没有任务在跑
|
||
// - running 运行中：任务正在后台干活
|
||
// - done 完成：顺利干完了
|
||
// - error 出错：干活过程中出了岔子
|
||
// - cancelled 已中止：用户手动点了中止
|
||
const (
|
||
ScanIdle ScanState = "idle"
|
||
ScanRunning ScanState = "running"
|
||
ScanDone ScanState = "done"
|
||
ScanError ScanState = "error"
|
||
ScanCancelled ScanState = "cancelled"
|
||
)
|
||
|
||
// 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 表示算不出来）
|
||
}
|
||
|
||
// ============================ 应用级常量 ============================
|
||
|
||
// 应用级配置键。目前只存了两个：上次使用的存放目录、色情检测服务是否真的起来了。
|
||
const (
|
||
ConfigKeyScanDir = "scan.scanDir"
|
||
ConfigKeyDestDir = "scan.destDir"
|
||
)
|
||
|
||
// nsfwDetectorReadyKey 记录最近一次扫描时色情检测服务是否真的起来了（python+nudenet 可用）。
|
||
// 配置开了但检测器没起来（没装 python/模型下载失败等）也能区分出来，前端据此提示。
|
||
const nsfwDetectorReadyKey = "nsfw.detectorReady"
|
||
|
||
// setNsfwDetectorReady 把"色情检测服务是否真的起来了"记进数据库（AppConfig）。
|
||
func (s *ScanService) setNsfwDetectorReady(ready bool) {
|
||
if s.store == nil {
|
||
return
|
||
}
|
||
v := "0"
|
||
if ready {
|
||
v = "1"
|
||
}
|
||
_ = s.store.SaveConfig(nsfwDetectorReadyKey, v)
|
||
}
|
||
|
||
// DefaultPickerDirName 用户没指定存放目录时，默认在扫描目录下新建这个文件夹装整理好的文件。
|
||
const DefaultPickerDirName = "KMP已整理"
|
||
|
||
// 三个工作阶段的名字。前端根据这个名字显示"正在扫描/正在整理/正在移动"。
|
||
const (
|
||
stageScan = "scan"
|
||
stageOrganize = "organize"
|
||
stageMove = "move"
|
||
)
|
||
|
||
// publishInterval 进度推送的节流间隔。
|
||
// 扫描大量文件时进度变化非常频繁，如果每次都推给前端，WebSocket 会被刷爆。
|
||
// 这里限制为每 0.1 秒最多推一次（约每秒 10 次），既实时又不至于泛滥。
|
||
// 注意：任务结束（done/error）不受这个限制，必须立刻推出去，否则用户看不到结果。
|
||
const publishInterval = 100 * time.Millisecond
|
||
|
||
// 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 // 运行日志
|
||
|
||
// 中止用的"停止开关"：Cancel 调用 cancel() 后，管道各环节检查 ctx 立即停下。
|
||
ctx context.Context
|
||
cancel context.CancelFunc
|
||
|
||
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 {
|
||
s := &ScanService{store: st, hub: NewStatusHub(), state: ScanIdle}
|
||
// 从数据库恢复上次用过的目录，状态接口可以直接返回给前端回填
|
||
if st != nil {
|
||
s.rootDir = st.GetConfig(ConfigKeyScanDir)
|
||
s.destDir = st.GetConfig(ConfigKeyDestDir)
|
||
}
|
||
return s
|
||
}
|
||
|
||
// Subscribe 让调用方（通常是 WebSocket 处理器）订阅进度更新。
|
||
// 返回值第一个是"收进度"的管道，第二个是取消订阅的函数（取消后管道会被关闭）。
|
||
func (s *ScanService) Subscribe() (StatusListener, func()) {
|
||
return s.hub.Subscribe()
|
||
}
|
||
|
||
// publish 把当前进度推给所有订阅者。
|
||
// 运行中按 publishInterval 节流；任务结束时无条件推送，保证"干完了"这个消息一定送达。
|
||
func (s *ScanService) publish() {
|
||
s.mu.Lock()
|
||
if s.state == ScanRunning && time.Since(s.lastPublish) < publishInterval {
|
||
s.mu.Unlock()
|
||
return
|
||
}
|
||
s.lastPublish = time.Now()
|
||
s.mu.Unlock()
|
||
s.hub.Publish(s.Status())
|
||
}
|
||
|
||
// 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)
|
||
}
|
||
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 = stageScan
|
||
s.done = 0
|
||
s.total = 0
|
||
// 建一个可中止的上下文：页面点「中止扫描」时调用 cancel()，各环节就会停下
|
||
s.ctx, s.cancel = context.WithCancel(context.Background())
|
||
s.mu.Unlock()
|
||
|
||
// 把扫描目录和存放目录记到数据库里，下次打开页面自动填上，省得用户重敲
|
||
if s.store != nil {
|
||
if err := s.store.SaveConfig(ConfigKeyScanDir, rootDir); err != nil {
|
||
fmt.Printf("=== 保存扫描目录配置失败: %v\n", err)
|
||
}
|
||
if err := s.store.SaveConfig(ConfigKeyDestDir, destDir); err != nil {
|
||
fmt.Printf("=== 保存存放目录配置失败: %v\n", err)
|
||
}
|
||
}
|
||
|
||
// 开个后台线程干活，这个函数立刻返回
|
||
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
|
||
}
|
||
|
||
// Move 触发文件移动操作。
|
||
// 根据 scan_log 中记录的分类结果，将文件从源路径移动到目标目录。
|
||
// 只有在扫描完成（state == done）或 scan_log 有记录时才能调用。
|
||
func (s *ScanService) Move() error {
|
||
s.mu.Lock()
|
||
|
||
// 正在跑就不能再开一个，直接拒绝
|
||
if s.state == ScanRunning {
|
||
s.mu.Unlock()
|
||
return errors.New("扫描整理正在运行中，请等待完成")
|
||
}
|
||
|
||
// 必须先完成一次扫描，或者 scan_log 有记录，才有数据可移动
|
||
if s.state != ScanDone {
|
||
// 检查 scan_log 是否有记录
|
||
hasLog, err := s.store.HasScanLog()
|
||
if err != nil || !hasLog {
|
||
s.mu.Unlock()
|
||
return errors.New("请先完成扫描后再执行移动")
|
||
}
|
||
}
|
||
|
||
// 把状态改成"运行中"，清空上一次的残留信息
|
||
s.state = ScanRunning
|
||
s.msg = ""
|
||
s.started = time.Now()
|
||
s.lines = nil
|
||
s.stage = stageMove
|
||
s.done = 0
|
||
s.total = 0
|
||
s.ctx, s.cancel = context.WithCancel(context.Background())
|
||
s.mu.Unlock()
|
||
|
||
go s.runMove(s.ctx)
|
||
return nil
|
||
}
|
||
|
||
// runMove 是移动文件的后台任务。
|
||
// 它从数据库读取 scan_log 记录，构建移动任务列表，然后并行执行文件移动。
|
||
func (s *ScanService) runMove(ctx context.Context) {
|
||
// 创建日志管道（和 run() 一样，截取 fmt.Println 输出）
|
||
r, w, err := os.Pipe()
|
||
if err != nil {
|
||
s.finish(ScanError, "创建日志管道失败: "+err.Error())
|
||
return
|
||
}
|
||
oldStdout := os.Stdout
|
||
os.Stdout = w
|
||
|
||
done := make(chan struct{})
|
||
go func() {
|
||
defer close(done)
|
||
sc := bufio.NewScanner(r)
|
||
sc.Buffer(make([]byte, 1024*1024), 1024*1024)
|
||
for sc.Scan() {
|
||
line := sanitizeLine(sc.Text())
|
||
if line == "" {
|
||
continue
|
||
}
|
||
s.mu.Lock()
|
||
if len(s.lines) >= maxLogLines {
|
||
s.lines = append([]string{}, s.lines[len(s.lines)/2:]...)
|
||
}
|
||
s.lines = append(s.lines, line)
|
||
s.mu.Unlock()
|
||
}
|
||
}()
|
||
|
||
defer func() {
|
||
os.Stdout = oldStdout
|
||
_ = w.Close()
|
||
<-done
|
||
if rcv := recover(); rcv != nil {
|
||
s.finish(ScanError, fmt.Sprintf("移动文件内部错误: %v\n%s", rcv, debug.Stack()))
|
||
}
|
||
}()
|
||
|
||
// 从数据库读取 scan_log 记录
|
||
if s.store == nil {
|
||
s.finish(ScanError, "数据库不可用，无法执行移动")
|
||
return
|
||
}
|
||
|
||
logs, err := s.store.LoadScanLog()
|
||
if err != nil {
|
||
s.finish(ScanError, "加载续传记录失败: "+err.Error())
|
||
return
|
||
}
|
||
|
||
if len(logs) == 0 {
|
||
fmt.Println("=== 没有待移动的文件（scan_log 为空）")
|
||
s.finish(ScanDone, "没有待移动的文件")
|
||
return
|
||
}
|
||
|
||
// 构建移动任务列表
|
||
jobs := make([]mover.MoveJob, 0, len(logs))
|
||
for _, sl := range logs {
|
||
// 检查源文件是否还存在
|
||
if _, err := os.Stat(sl.SourcePath); err != nil {
|
||
fmt.Printf("=== 源文件不存在，跳过: %s\n", sl.SourcePath)
|
||
continue
|
||
}
|
||
jobs = append(jobs, mover.MoveJob{Source: sl.SourcePath, TargetDir: sl.TargetDir})
|
||
}
|
||
|
||
if len(jobs) == 0 {
|
||
fmt.Println("=== 所有源文件均已不存在，无需移动")
|
||
s.finish(ScanDone, "没有需要移动的文件（源文件已不存在）")
|
||
return
|
||
}
|
||
|
||
fmt.Printf("=== 开始移动文件，共 %d 个任务\n", len(jobs))
|
||
s.setProgress(stageMove, 0, len(jobs))
|
||
|
||
// 执行移动
|
||
mv := &mover.Mover{}
|
||
moved := mv.MoveAll(jobs, func(doneCount int) {
|
||
s.setProgress(stageMove, doneCount, len(jobs))
|
||
if ctx.Err() != nil {
|
||
return
|
||
}
|
||
})
|
||
|
||
if ctx.Err() != nil {
|
||
s.finish(ScanCancelled, "移动已中止")
|
||
return
|
||
}
|
||
|
||
// 更新 media_info 中的 file_path 为新路径
|
||
if len(moved) > 0 {
|
||
if err := s.store.UpdateMediaFilePaths(moved); err != nil {
|
||
fmt.Printf("=== 更新媒体文件路径失败: %v\n", err)
|
||
}
|
||
}
|
||
|
||
// 移动完成后清空续传记录
|
||
if err := s.store.ClearScanLog(); err != nil {
|
||
fmt.Printf("=== 清空续传记录失败: %v\n", err)
|
||
}
|
||
|
||
fmt.Println("=== 文件移动完成")
|
||
s.finish(ScanDone, fmt.Sprintf("文件移动完成，共移动 %d 个文件", len(jobs)))
|
||
}
|
||
|
||
// 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)
|
||
if progress > 100 {
|
||
progress = 100
|
||
}
|
||
}
|
||
|
||
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,
|
||
Done: s.done,
|
||
Total: s.total,
|
||
Eta: s.estimateEtaLocked(),
|
||
}
|
||
}
|
||
|
||
// estimateEtaLocked 估算"还剩多少秒"，规则是：记住这个阶段开始的时间，
|
||
// 看这段时间干了多少个文件，算出速度，再用剩下的数量除以速度。
|
||
// 返回 0 表示还估不出来（刚开跑、或者数量太小）。
|
||
// 注意：调用前必须已经持有锁。
|
||
func (s *ScanService) estimateEtaLocked() int64 {
|
||
if s.state != ScanRunning || s.total <= 0 || s.done <= 0 || s.done >= s.total {
|
||
return 0
|
||
}
|
||
if s.stageStart.IsZero() || s.done <= s.stageDone {
|
||
return 0
|
||
}
|
||
elapsedSec := time.Since(s.stageStart).Seconds()
|
||
if elapsedSec <= 0 {
|
||
return 0
|
||
}
|
||
rate := float64(s.done-s.stageDone) / elapsedSec // 每秒处理多少个文件
|
||
if rate <= 0 {
|
||
return 0
|
||
}
|
||
remain := float64(s.total-s.done) / rate
|
||
if remain <= 0 {
|
||
return 0
|
||
}
|
||
return int64(remain + 0.5)
|
||
}
|
||
|
||
// resolveRootDir 把用户填的目录整理成标准绝对路径；留空就取程序当前所在目录。
|
||
func resolveRootDir(dir string) string {
|
||
if strings.TrimSpace(dir) == "" {
|
||
wd, err := os.Getwd()
|
||
if err == nil {
|
||
return wd
|
||
}
|
||
return "."
|
||
}
|
||
abs, err := filepath.Abs(strings.TrimSpace(dir))
|
||
if err != nil {
|
||
return strings.TrimSpace(dir)
|
||
}
|
||
return abs
|
||
}
|
||
|
||
// resolveFFmpegPath 确定 ffmpeg 程序的路径：
|
||
// - 配置里指定了就用配置的（不存在时退回下一步）；
|
||
// - 否则在 PATH 里找 ffmpeg。
|
||
//
|
||
// 找不到返回空串（调用方会退化为"不转正、按显示方向归类"）。
|
||
func resolveFFmpegPath(configured string) string {
|
||
if configured != "" {
|
||
if _, err := os.Stat(configured); err == nil {
|
||
return configured
|
||
}
|
||
}
|
||
if p, err := exec.LookPath("ffmpeg"); err == nil {
|
||
return p
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// ============================ 后台任务本体 ============================
|
||
|
||
// run 是后台干活线程。除了调用 runScan 真正干活外，它还要做一件特殊的事：
|
||
// 把程序打印到屏幕上的日志"截胡"下来，存进 s.lines，好让前端能看到实时日志。
|
||
//
|
||
// 做法：把标准输出 os.Stdout 临时换成一根管道（pipe）的写端，
|
||
// 另起一个 goroutine 从管道的读端一行行读日志。干完活再把 Stdout 换回去。
|
||
func (s *ScanService) run(ctx context.Context, rootDir, destDir string) {
|
||
// 创建一根管道，准备截取日志
|
||
r, w, err := os.Pipe()
|
||
if err != nil {
|
||
s.finish(ScanError, "创建日志管道失败: "+err.Error())
|
||
return
|
||
}
|
||
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) // 允许很长的日志行
|
||
for sc.Scan() {
|
||
line := sanitizeLine(sc.Text())
|
||
if line == "" {
|
||
continue
|
||
}
|
||
s.mu.Lock()
|
||
if len(s.lines) >= maxLogLines {
|
||
// 日志太多就丢最旧一半，只留最近的部分
|
||
s.lines = append([]string{}, s.lines[len(s.lines)/2:]...)
|
||
}
|
||
s.lines = append(s.lines, line)
|
||
s.mu.Unlock()
|
||
}
|
||
}()
|
||
|
||
// 收尾必须恢复 stdout 并关闭管道，否则整个进程后续输出都会丢进这根无人读的管道里。
|
||
// 用 defer 保证即使 runScan 中途 panic 也不会留下这种状态；panic 还会被接住，
|
||
// 转成"出错"状态推给前端，而不是让整个程序崩溃。
|
||
defer func() {
|
||
os.Stdout = oldStdout
|
||
_ = w.Close()
|
||
<-done // 等读日志的 goroutine 把剩下的日志读完
|
||
if rcv := recover(); rcv != nil {
|
||
s.finish(ScanError, fmt.Sprintf("扫描整理内部错误: %v\n%s", rcv, debug.Stack()))
|
||
}
|
||
}()
|
||
|
||
// 真正开始干活。期间程序里所有 fmt.Println/Printf 都会进上面的管道
|
||
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
|
||
}
|
||
s.finish(ScanDone, "扫描整理完成")
|
||
}
|
||
|
||
// 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()
|
||
}
|
||
|
||
// 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
|
||
}
|
||
s.stage = stage
|
||
s.done = done
|
||
s.total = total
|
||
s.mu.Unlock()
|
||
s.publish()
|
||
}
|
||
|
||
// ansiRe 用来去掉日志里的 ANSI 颜色码（那些 [31m 之类的转义序列）。
|
||
var ansiRe = regexp.MustCompile(`\x1b\[[0-9;]*m`)
|
||
|
||
// 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 执行完整的整理流程，一共四步：
|
||
// 1. 扫描：把目录里的文件全部找出来，识别每个文件的类型
|
||
// 2. 过滤：把"值得处理"的文件分成图片、视频/音频两拨
|
||
// 3. 准备解码器：把内嵌的 ffprobe 程序释放到当前目录（用它读视频/音频信息）
|
||
// 4. 整理：给每个文件探测信息 → 判断是否重复 → 分到分类目录 → 搬进去
|
||
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)
|
||
|
||
// ---------- 第一步：扫描 ----------
|
||
// 扫描时跳过这些目录/文件：整理输出目录本身、数据库文件、ffprobe 程序，
|
||
// 不然它们会被当成"媒体文件"被自己整理进去，那可就乱套了。
|
||
res, err := scanner.Scan(scanner.Options{
|
||
Ctx: ctx,
|
||
RootDir: rootDir,
|
||
SkipDir: pickerDir,
|
||
SkipDirNames: []string{DefaultPickerDirName},
|
||
SkipNames: []string{store.DBFileName, probe.FFprobeExeName},
|
||
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)
|
||
}
|
||
|
||
// ---------- 第二步：过滤 ----------
|
||
// 把扫描到的文件分成"图片"和"视频/音频候选"两拨
|
||
res.Filter()
|
||
|
||
// ---------- 第三步：准备解码器 ----------
|
||
// 读取视频/音频信息需要 ffprobe。Windows 构建内嵌了一份，先释放到当前目录；
|
||
// Linux 构建没有内嵌，自动回退用系统 PATH 里的 ffprobe（如 /usr/bin/ffprobe）。
|
||
// 两者都没有就跳过视频/音频整理，图片照常处理。
|
||
ffprobePath := ""
|
||
if len(res.Images) > 0 || len(res.Media) > 0 {
|
||
if err := probe.ReleaseFFprobe(); err != nil {
|
||
fmt.Printf("=== 释放解码器失败, 视频/音频整理将跳过(图片照常处理): %v\n", err)
|
||
} else {
|
||
ffprobePath = probe.FFprobePath()
|
||
defer probe.RemoveFFprobe()
|
||
}
|
||
}
|
||
|
||
// ---------- 第四步：整理 ----------
|
||
// ffmpeg 用于把带 EXIF 旋转标记的图片转正（竖拍的图文件里横着存，显示才竖起来）。
|
||
// 找不到 ffmpeg 也没关系——图片照常处理，只是不转正、按显示方向归类。
|
||
ffmpegPath := ""
|
||
if config.GetAppConfig().RotateImages {
|
||
ffmpegPath = resolveFFmpegPath(config.GetAppConfig().FFmpegPath)
|
||
if ffmpegPath == "" {
|
||
fmt.Println("=== 未找到 ffmpeg, 图片EXIF旋转将不做物理转正(仍按显示方向归类)")
|
||
} else {
|
||
fmt.Printf("=== 使用 ffmpeg 转正图片EXIF旋转: %s\n", ffmpegPath)
|
||
}
|
||
}
|
||
|
||
// ---------- 第五步（可选）：色情识别 ----------
|
||
// 从数据库读取 NSFW 配置（支持运行时修改，立即生效）
|
||
nsfwType := s.store.GetConfig("nsfw.type") // "" / "human" / "anime"
|
||
pythonPath := config.GetAppConfig().NSFWPython
|
||
|
||
nsfwDet := (*nsfw.Detector)(nil)
|
||
s.setNsfwDetectorReady(false)
|
||
if nsfwType == "human" {
|
||
threshold := 0.5
|
||
if v := s.store.GetConfig("nsfw.threshold"); v != "" {
|
||
if t, err := strconv.ParseFloat(v, 64); err == nil && t > 0 && t < 1 {
|
||
threshold = t
|
||
}
|
||
}
|
||
if d, err := nsfw.Start(pythonPath, threshold); err != nil {
|
||
fmt.Printf("=== 色情识别未启用(图片照常处理): %v\n", err)
|
||
} else {
|
||
nsfwDet = d
|
||
defer nsfwDet.Stop()
|
||
s.setNsfwDetectorReady(true)
|
||
fmt.Println("=== 色情识别已启用(NudeNet)")
|
||
}
|
||
}
|
||
|
||
// ---------- 第六步（可选）：动漫色情识别 ----------
|
||
animeNsfwDet := (*anime_nsfw.Detector)(nil)
|
||
if nsfwType == "anime" {
|
||
threshold := 0.5
|
||
if v := s.store.GetConfig("anime.threshold"); v != "" {
|
||
if t, err := strconv.ParseFloat(v, 64); err == nil && t > 0 && t < 1 {
|
||
threshold = t
|
||
}
|
||
}
|
||
if d, err := anime_nsfw.Start(pythonPath, threshold); err != nil {
|
||
fmt.Printf("=== 动漫色情识别未启用(图片照常处理): %v\n", err)
|
||
} else {
|
||
animeNsfwDet = d
|
||
defer animeNsfwDet.Stop()
|
||
fmt.Println("=== 动漫色情识别已启用(WD Tagger v3)")
|
||
}
|
||
}
|
||
|
||
org := organize.New(pickerDir, ffprobePath, ffmpegPath, s.store, ctx)
|
||
org.SetNSFWDetector(nsfwDet)
|
||
org.SetAnimeNSFWDetector(animeNsfwDet)
|
||
org.SetRotateImages(config.GetAppConfig().RotateImages)
|
||
|
||
// 整个"整理"分两个阶段，界面上共用一条进度条：
|
||
// 归类阶段：图片、视频/音频分两批处理。进度 = 已处理文件数 / 全部待整理文件数
|
||
// 移动阶段：把所有文件搬进分类目录。进度 = 已移动数 / 待移动数
|
||
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)
|
||
}
|
||
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()
|
||
}
|
||
|
||
// 扫描完成后只写入媒体记录和续传记录，不移动文件。
|
||
// 文件移动由用户在页面上手动触发（POST /api/scan/move）。
|
||
org.FlushRecordsOnly()
|
||
|
||
// 把进度条顶满 100%，再打印"整理完成"
|
||
s.setProgress(stageMove, totalWork, totalWork)
|
||
fmt.Println("=== 扫描整理完成，请在页面点击「移动文件」按钮执行文件移动")
|
||
return nil
|
||
}
|