319 lines
7.1 KiB
Go
319 lines
7.1 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"log"
|
||
"net/http"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
func newApp() *app {
|
||
defaultCfg := defaultConfig()
|
||
defaultForm := formFromConfig(defaultCfg)
|
||
settingsPath := resolveSettingsPath()
|
||
|
||
savedForm, err := loadSettings(settingsPath)
|
||
if err != nil {
|
||
log.Printf("load settings failed: %v", err)
|
||
}
|
||
|
||
return &app{
|
||
cfg: defaultCfg,
|
||
form: mergeForm(defaultForm, savedForm),
|
||
settingsPath: settingsPath,
|
||
lastError: loadSettingsErrorMessage(err),
|
||
}
|
||
}
|
||
|
||
func (a *app) handleIndex(w http.ResponseWriter, r *http.Request) {
|
||
if r.URL.Path != "/" {
|
||
http.NotFound(w, r)
|
||
return
|
||
}
|
||
if r.Method != http.MethodGet {
|
||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||
return
|
||
}
|
||
|
||
a.renderPage(w, formValues{}, "", "")
|
||
}
|
||
|
||
func (a *app) handleStart(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||
return
|
||
}
|
||
|
||
form := formFromRequest(r)
|
||
cfg, err := parseConfig(form)
|
||
if err != nil {
|
||
a.renderPage(w, form, "", err.Error())
|
||
return
|
||
}
|
||
|
||
normalizedForm := formFromConfig(cfg)
|
||
saveErr := a.persistForm(normalizedForm)
|
||
|
||
if err := a.startWatcher(cfg); err != nil {
|
||
a.renderPage(w, normalizedForm, "", joinMessages(err.Error(), saveErrorMessage(saveErr)))
|
||
return
|
||
}
|
||
|
||
a.renderPage(w, normalizedForm, "监听任务已启动，配置已保存到本地", saveErrorMessage(saveErr))
|
||
}
|
||
|
||
func (a *app) handleStop(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||
return
|
||
}
|
||
|
||
message := "当前没有正在运行的监听任务"
|
||
if a.stopWatcher() {
|
||
message = "已发送停止请求"
|
||
}
|
||
|
||
a.renderPage(w, formValues{}, message, "")
|
||
}
|
||
|
||
func (a *app) persistForm(form formValues) error {
|
||
a.mu.Lock()
|
||
a.form = form
|
||
a.mu.Unlock()
|
||
|
||
if err := saveSettings(a.settingsPath, form); err != nil {
|
||
a.addLogf("save settings failed: %v", err)
|
||
return err
|
||
}
|
||
|
||
a.mu.Lock()
|
||
if strings.HasPrefix(a.lastError, "读取本地配置失败") || strings.HasPrefix(a.lastError, "配置保存失败") {
|
||
a.lastError = ""
|
||
}
|
||
a.mu.Unlock()
|
||
return nil
|
||
}
|
||
|
||
func (a *app) startWatcher(cfg config) error {
|
||
cfg, err := validateConfig(cfg)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
if cfg.archive.enabled {
|
||
if err := ensureExistingDir(cfg.archive.sourceDir); err != nil {
|
||
return fmt.Errorf("压缩包规则源路径错误: %w", err)
|
||
}
|
||
if err := ensureDir(cfg.archive.zipDir); err != nil {
|
||
return fmt.Errorf("压缩包规则 zip 路径错误: %w", err)
|
||
}
|
||
if err := ensureDir(cfg.archive.unzipDir); err != nil {
|
||
return fmt.Errorf("压缩包规则 unzip 路径错误: %w", err)
|
||
}
|
||
}
|
||
|
||
if cfg.generic.enabled {
|
||
if err := ensureExistingDir(cfg.generic.sourceDir); err != nil {
|
||
return fmt.Errorf("普通文件规则源路径错误: %w", err)
|
||
}
|
||
if err := ensureDir(cfg.generic.targetDir); err != nil {
|
||
return fmt.Errorf("普通文件规则目标路径错误: %w", err)
|
||
}
|
||
}
|
||
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
|
||
a.mu.Lock()
|
||
if a.cancel != nil {
|
||
a.cancel()
|
||
}
|
||
a.runID++
|
||
runID := a.runID
|
||
a.cancel = cancel
|
||
a.running = true
|
||
a.startedAt = time.Now()
|
||
a.lastError = ""
|
||
a.lastMessage = "监听任务运行中"
|
||
a.cfg = cfg
|
||
a.form = formFromConfig(cfg)
|
||
a.mu.Unlock()
|
||
|
||
if cfg.archive.enabled {
|
||
a.addLogf("archive watcher enabled: source=%q zip=%q unzip=%q", cfg.archive.sourceDir, cfg.archive.zipDir, cfg.archive.unzipDir)
|
||
}
|
||
if cfg.generic.enabled {
|
||
a.addLogf(
|
||
"generic watcher enabled: source=%q target=%q extensions=%s dedupe=%s",
|
||
cfg.generic.sourceDir,
|
||
cfg.generic.targetDir,
|
||
strings.Join(cfg.generic.extensions, ", "),
|
||
cfg.generic.dedupeMode,
|
||
)
|
||
}
|
||
a.addLogf("common settings: interval=%s stable_for=%s stable_scans=%d mode=%s recursive=%t",
|
||
cfg.interval, cfg.stableFor, cfg.stableScans, cfg.mode, cfg.recursive)
|
||
|
||
go a.runWatcher(ctx, runID, cfg)
|
||
return nil
|
||
}
|
||
|
||
func (a *app) stopWatcher() bool {
|
||
a.mu.RLock()
|
||
cancel := a.cancel
|
||
running := a.running
|
||
a.mu.RUnlock()
|
||
|
||
if cancel == nil || !running {
|
||
return false
|
||
}
|
||
|
||
a.addLogf("stop requested from web UI")
|
||
cancel()
|
||
return true
|
||
}
|
||
|
||
func (a *app) runWatcher(ctx context.Context, runID int64, cfg config) {
|
||
defer a.finishRun(runID)
|
||
|
||
archiveState := make(map[string]fileState)
|
||
genericState := make(map[string]fileState)
|
||
|
||
runScan := func() {
|
||
var errorsFound []string
|
||
|
||
if cfg.archive.enabled {
|
||
err := scanRule(
|
||
"压缩包规则",
|
||
cfg.archive.sourceDir,
|
||
cfg.recursive,
|
||
func(name string) bool {
|
||
return strings.HasSuffix(strings.ToLower(name), tarGzSuffix)
|
||
},
|
||
cfg,
|
||
archiveState,
|
||
func(sourcePath string) (string, bool, error) {
|
||
return processArchiveCandidate(cfg, sourcePath)
|
||
},
|
||
a.addLogf,
|
||
)
|
||
if err != nil {
|
||
errorsFound = append(errorsFound, "压缩包规则扫描失败: "+err.Error())
|
||
}
|
||
}
|
||
|
||
if cfg.generic.enabled {
|
||
err := scanRule(
|
||
"普通文件规则",
|
||
cfg.generic.sourceDir,
|
||
cfg.recursive,
|
||
func(name string) bool {
|
||
return matchesExtensions(name, cfg.generic.extensions)
|
||
},
|
||
cfg,
|
||
genericState,
|
||
func(sourcePath string) (string, bool, error) {
|
||
return processGenericCandidate(cfg, sourcePath)
|
||
},
|
||
a.addLogf,
|
||
)
|
||
if err != nil {
|
||
errorsFound = append(errorsFound, "普通文件规则扫描失败: "+err.Error())
|
||
}
|
||
}
|
||
|
||
if len(errorsFound) > 0 {
|
||
message := strings.Join(errorsFound, "；")
|
||
a.setLastError(message)
|
||
a.addLogf("%s", message)
|
||
return
|
||
}
|
||
a.setLastError("")
|
||
}
|
||
|
||
runScan()
|
||
|
||
ticker := time.NewTicker(cfg.interval)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
a.addLogf("watcher stopped")
|
||
return
|
||
case <-ticker.C:
|
||
runScan()
|
||
}
|
||
}
|
||
}
|
||
|
||
func (a *app) finishRun(runID int64) {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
|
||
if a.runID != runID {
|
||
return
|
||
}
|
||
|
||
a.running = false
|
||
a.cancel = nil
|
||
a.lastMessage = "监听任务已停止"
|
||
}
|
||
|
||
func (a *app) setLastError(message string) {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
a.lastError = message
|
||
}
|
||
|
||
func (a *app) addLogf(format string, args ...any) {
|
||
msg := fmt.Sprintf(format, args...)
|
||
log.Print(msg)
|
||
|
||
line := fmt.Sprintf("%s %s", time.Now().Format("2006-01-02 15:04:05"), msg)
|
||
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
|
||
a.logs = append(a.logs, line)
|
||
if len(a.logs) > maxLogLines {
|
||
a.logs = a.logs[len(a.logs)-maxLogLines:]
|
||
}
|
||
}
|
||
|
||
func (a *app) renderPage(w http.ResponseWriter, override formValues, submitMessage, submitError string) {
|
||
page := a.snapshot()
|
||
if hasFormInput(override) {
|
||
page.Form = override
|
||
}
|
||
page.SubmitMessage = submitMessage
|
||
page.SubmitError = submitError
|
||
|
||
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
||
if err := pageTemplate.Execute(w, page); err != nil {
|
||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||
}
|
||
}
|
||
|
||
func (a *app) snapshot() pageData {
|
||
a.mu.RLock()
|
||
defer a.mu.RUnlock()
|
||
|
||
logs := append([]string(nil), a.logs...)
|
||
startedAt := ""
|
||
if !a.startedAt.IsZero() {
|
||
startedAt = a.startedAt.Format("2006-01-02 15:04:05")
|
||
}
|
||
|
||
return pageData{
|
||
Running: a.running,
|
||
StartedAt: startedAt,
|
||
LastMessage: a.lastMessage,
|
||
LastError: a.lastError,
|
||
SettingsPath: a.settingsPath,
|
||
Form: a.form,
|
||
Logs: logs,
|
||
}
|
||
}
|