From 285064d14456c492b8391fc383e0f8b836069019 Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Wed, 16 Sep 2026 06:41:45 +0800 Subject: [PATCH] refactor: modularize downloader architecture --- am_client.go | 61 +++++ app.go | 32 +++ config.go | 89 ++++++++ downloader.go | 170 ++++++++++++++ handlers.go | 110 +++++++++ main.go | 569 +--------------------------------------------- models.go | 54 +++++ remote_storage.go | 41 ++++ settings.go | 85 +++++++ util.go | 37 +++ webdav.go | 76 +++++++ 11 files changed, 766 insertions(+), 558 deletions(-) create mode 100644 am_client.go create mode 100644 app.go create mode 100644 config.go create mode 100644 downloader.go create mode 100644 handlers.go create mode 100644 models.go create mode 100644 remote_storage.go create mode 100644 settings.go create mode 100644 util.go create mode 100644 webdav.go diff --git a/am_client.go b/am_client.go new file mode 100644 index 0000000..999540b --- /dev/null +++ b/am_client.go @@ -0,0 +1,61 @@ +package main + +import ( + "encoding/json" + "errors" + "net/http" + "net/url" + "strconv" + "strings" +) + +type AMAPIClient struct { + BaseURL string + HTTP *http.Client +} + +func (c *AMAPIClient) Search(keyword, source string, page, limit int) (*http.Response, error) { + u, err := url.Parse(strings.TrimRight(c.BaseURL, "/") + "/search") + if err != nil { + return nil, err + } + q := u.Query() + q.Set("keyword", keyword) + q.Set("source", source) + q.Set("page", strconv.Itoa(page)) + q.Set("limit", strconv.Itoa(limit)) + u.RawQuery = q.Encode() + return c.HTTP.Get(u.String()) +} + +func (c *AMAPIClient) Resolve(source, rid, quality string) (playInfo, error) { + u, err := url.Parse(strings.TrimRight(c.BaseURL, "/") + "/song/play-url") + if err != nil { + return playInfo{}, err + } + q := u.Query() + q.Set("source", source) + q.Set("rid", rid) + q.Set("level", quality) + u.RawQuery = q.Encode() + resp, err := c.HTTP.Get(u.String()) + if err != nil { + return playInfo{}, err + } + defer resp.Body.Close() + var x struct { + Code int `json:"code"` + Data playInfo `json:"data"` + Message string `json:"message"` + } + if err := json.NewDecoder(resp.Body).Decode(&x); err != nil { + return playInfo{}, err + } + if x.Code != 0 || x.Data.URL == "" { + if x.Message == "" { + x.Message = "music API returned no playable URL" + } + return playInfo{}, errors.New(x.Message) + } + return x.Data, nil +} diff --git a/app.go b/app.go new file mode 100644 index 0000000..5e681ae --- /dev/null +++ b/app.go @@ -0,0 +1,32 @@ +package main + +import ( + "net/http" + "sync" + + "gorm.io/gorm" +) + +type App struct { + db *gorm.DB + cfg Config + queue chan uint + wg sync.WaitGroup + httpClient *http.Client + am AMClient + storage RemoteStorage +} + +// AMClient isolates the music API from the task engine. +type AMClient interface { + Search(keyword, source string, page, limit int) (*http.Response, error) + Resolve(source, rid, quality string) (playInfo, error) +} + +// RemoteStorage isolates WebDAV and leaves room for S3/local backends later. +type RemoteStorage interface { + Upload(localPath, remotePath string) error + ExistsSameSize(remotePath string, size int64) (bool, error) + Remove(remotePath string) error + MkdirAll(remoteDir string) error +} diff --git a/config.go b/config.go new file mode 100644 index 0000000..26bed6d --- /dev/null +++ b/config.go @@ -0,0 +1,89 @@ +package main + +import ( + "os" + "path/filepath" + "strconv" + "strings" + + "gorm.io/gorm" +) + +func (a *App) loadSettings() { + var ss []Setting + a.db.Find(&ss) + for _, s := range ss { + switch s.Key { + case "am_api": + a.cfg.AMAPI = s.Value + case "download_dir": + a.cfg.DownloadDir = s.Value + case "webdav_url": + a.cfg.WebDAVURL = s.Value + case "webdav_user": + a.cfg.WebDAVUser = s.Value + case "webdav_pass": + a.cfg.WebDAVPass = s.Value + case "webdav_path": + a.cfg.WebDAVPath = s.Value + case "quality": + a.cfg.DefaultQuality = s.Value + case "webdav_version_policy": + a.cfg.WebDAVVersionPolicy = s.Value + case "local_file_policy": + a.cfg.LocalFilePolicy = s.Value + } + } +} + +func (a *App) saveSetting(k, v string) { + var s Setting + if a.db.Where("key = ?", k).First(&s).Error != nil { + a.db.Create(&Setting{Key: k, Value: v}) + } else { + s.Value = v + a.db.Save(&s) + } +} + +func loadConfig(db *gorm.DB) Config { + cfg := Config{ + AMAPI: getenv("AM_API_URL", "http://127.0.0.1:9530/api"), + DownloadDir: getenv("DOWNLOAD_DIR", "/downloads"), + DefaultQuality: getenv("DEFAULT_QUALITY", "standard"), + WorkerCount: atoi(getenv("WORKERS", "2"), 2), + } + _ = db + return cfg +} + +func normalizeConfig(cfg *Config) { + if cfg.WorkerCount < 1 { + cfg.WorkerCount = 1 + } + if cfg.WebDAVVersionPolicy != "keep" && cfg.WebDAVVersionPolicy != "replace_lower" { + cfg.WebDAVVersionPolicy = "keep" + } + if cfg.LocalFilePolicy != "keep" && cfg.LocalFilePolicy != "delete_after_upload" { + cfg.LocalFilePolicy = "keep" + } + cfg.WebDAVPath = strings.Trim(cfg.WebDAVPath, "/") + _ = os.MkdirAll(cfg.DownloadDir, 0755) +} + +func getenv(k, d string) string { + if v := os.Getenv(k); v != "" { + return v + } + return d +} + +func atoi(s string, d int) int { + v, e := strconv.Atoi(s) + if e != nil || v < 1 { + return d + } + return v +} + +func dbDir(path string) string { return filepath.Dir(path) } diff --git a/downloader.go b/downloader.go new file mode 100644 index 0000000..820eb03 --- /dev/null +++ b/downloader.go @@ -0,0 +1,170 @@ +package main + +import ( + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "strings" +) + +func (a *App) startWorkers() { + for i := 0; i < a.cfg.WorkerCount; i++ { + a.wg.Add(1) + go func() { + defer a.wg.Done() + for id := range a.queue { + a.download(id) + } + }() + } +} + +func (a *App) download(id uint) { + var t Task + if a.db.First(&t, id).Error != nil || t.Status != "pending" { + return + } + a.db.Model(&t).Updates(map[string]any{"status": "resolving", "error": "", "progress": 0}) + info, err := a.am.Resolve(t.Source, t.RID, t.Quality) + if err != nil { + a.fail(&t, err) + return + } + + ext := normalizedFormat(info.Format) + if ext == "" { + ext = normalizedFormat(filepath.Ext(info.URL)) + } + if ext == "" { + ext = "mp3" + } + filename := safe(t.Artist) + " - " + safe(t.Title) + "." + ext + local := filepath.Join(a.cfg.DownloadDir, filename) + t.Format, t.ResolvedQuality, t.ActualQuality, t.LocalPath = ext, info.Level, "", local + + // Reuse only the path recorded on this task. Never borrow another task's quality. + if t.LocalPath != "" { + if fi, statErr := os.Stat(t.LocalPath); statErr == nil && fi.Size() > 0 { + a.finishExisting(&t, fi.Size()) + return + } + } + a.downloadFile(&t, info) +} + +func (a *App) finishExisting(t *Task, size int64) { + actual, err := detectAudioFormat(t.LocalPath) + if err == nil { + actual = normalizedFormat(actual) + if actual != t.Format { + if renamed, renameErr := renameToActualFormat(t.LocalPath, actual); renameErr == nil { + t.LocalPath = renamed + } else { + a.fail(t, renameErr) + return + } + } + t.Format = actual + t.ActualQuality = actualQuality(t.Quality, t.ResolvedQuality, actual) + } + t.TotalBytes, t.DownloadedBytes, t.Progress, t.Status = size, size, 100, "done" + a.db.Save(t) + a.upload(t) +} + +func (a *App) downloadFile(t *Task, info playInfo) { + a.db.Model(t).Updates(map[string]any{"status": "downloading", "format": t.Format, "local_path": t.LocalPath}) + req, err := http.NewRequest("GET", info.URL, nil) + if err != nil { + a.fail(t, err) + return + } + for k, v := range info.Headers { + req.Header.Set(k, v) + } + resp, err := a.httpClient.Do(req) + if err != nil { + a.fail(t, err) + return + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + a.fail(t, fmt.Errorf("media http %d", resp.StatusCode)) + return + } + t.Status, t.TotalBytes, t.DownloadedBytes, t.Progress = "downloading", resp.ContentLength, 0, 0 + a.db.Save(t) + + tmp := t.LocalPath + ".part" + f, err := os.Create(tmp) + if err != nil { + a.fail(t, err) + return + } + n, err := a.copyWithProgress(f, resp.Body, t) + closeErr := f.Close() + if err != nil { + a.fail(t, err) + return + } + if closeErr != nil { + a.fail(t, closeErr) + return + } + if t.TotalBytes > 0 && n != t.TotalBytes { + a.fail(t, fmt.Errorf("incomplete download: got %d of %d bytes", n, t.TotalBytes)) + return + } + + actual, err := detectAudioFormat(tmp) + if err != nil { + a.fail(t, fmt.Errorf("detect audio format: %w", err)) + _ = os.Remove(tmp) + return + } + actual = normalizedFormat(actual) + finalPath := strings.TrimSuffix(t.LocalPath, filepath.Ext(t.LocalPath)) + "." + actual + if err = os.Rename(tmp, finalPath); err != nil { + a.fail(t, err) + _ = os.Remove(tmp) + return + } + t.Format, t.LocalPath, t.ActualQuality, t.DownloadedBytes = actual, finalPath, actualQuality(t.Quality, t.ResolvedQuality, actual), n + if t.TotalBytes > 0 { + t.Progress = 100 + } + a.db.Save(t) + a.upload(t) +} + +func (a *App) copyWithProgress(dst *os.File, src io.Reader, t *Task) (int64, error) { + var n, last int64 + buf := make([]byte, 256*1024) + for { + nr, er := src.Read(buf) + if nr > 0 { + wn, ew := dst.Write(buf[:nr]) + n += int64(wn) + if ew != nil { + return n, ew + } + if n-last >= 1024*1024 { + progress := 0.0 + if t.TotalBytes > 0 { + progress = float64(n) * 100 / float64(t.TotalBytes) + } + t.DownloadedBytes, t.Progress = n, progress + a.db.Model(t).Updates(map[string]any{"downloaded_bytes": n, "progress": progress}) + last = n + } + } + if er == io.EOF { + return n, nil + } + if er != nil { + return n, er + } + } +} diff --git a/handlers.go b/handlers.go new file mode 100644 index 0000000..13239b0 --- /dev/null +++ b/handlers.go @@ -0,0 +1,110 @@ +package main + +import ( + "io" + "strconv" + "strings" + + "github.com/gin-gonic/gin" +) + +func (a *App) search(c *gin.Context) { + keyword := strings.TrimSpace(c.Query("keyword")) + if keyword == "" { + c.JSON(400, gin.H{"error": "keyword required"}) + return + } + source := c.DefaultQuery("source", "kw") + page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) + if page < 1 { + page = 1 + } + limit, _ := strconv.Atoi(c.DefaultQuery("limit", "20")) + if limit < 1 { + limit = 20 + } + resp, err := a.am.Search(keyword, source, page, limit) + if err != nil { + c.JSON(502, gin.H{"error": err.Error()}) + return + } + defer resp.Body.Close() + c.Status(resp.StatusCode) + _, _ = io.Copy(c.Writer, resp.Body) +} + +func (a *App) listTasks(c *gin.Context) { + var ts []Task + if err := a.db.Order("id desc").Limit(100).Find(&ts).Error; err != nil { + c.JSON(500, gin.H{"error": err.Error()}) + return + } + c.JSON(200, gin.H{"data": ts}) +} + +func (a *App) addTask(c *gin.Context) { + var in SongRequest + if c.BindJSON(&in) != nil || in.RID == "" || in.Title == "" { + c.JSON(400, gin.H{"error": "source/rid/title required"}) + return + } + if in.Source == "" { + in.Source = "kw" + } + if in.Quality == "" { + in.Quality = a.cfg.DefaultQuality + } + var existing Task + err := a.db.Where("source = ? AND r_id = ? AND quality = ?", in.Source, in.RID, in.Quality).Order("id desc").First(&existing).Error + if err == nil { + if existing.Status != "error" { + c.JSON(200, existing) + return + } + a.db.Model(&existing).Updates(map[string]any{"title": in.Title, "artist": in.Artist, "album": in.Album, "pic": in.Pic, "status": "pending", "error": ""}) + a.queue <- existing.ID + c.JSON(202, existing) + return + } + t := Task{Source: in.Source, RID: in.RID, Title: in.Title, Artist: in.Artist, Album: in.Album, Pic: in.Pic, Quality: in.Quality, Status: "pending"} + if err := a.db.Create(&t).Error; err != nil { + if a.db.Where("source = ? AND r_id = ? AND quality = ?", in.Source, in.RID, in.Quality).Order("id desc").First(&existing).Error == nil { + c.JSON(200, existing) + return + } + c.JSON(500, gin.H{"error": err.Error()}) + return + } + a.queue <- t.ID + c.JSON(202, t) +} + +func (a *App) retryTask(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.Status(400) + return + } + var t Task + if a.db.First(&t, id).Error != nil { + c.Status(404) + return + } + a.db.Model(&t).Updates(map[string]any{"status": "pending", "error": "", "progress": 0, "downloaded_bytes": 0}) + t.Status, t.Error, t.Progress, t.DownloadedBytes = "pending", "", 0, 0 + a.queue <- t.ID + c.JSON(202, t) +} + +func (a *App) deleteTask(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil { + c.Status(400) + return + } + if err := a.db.Delete(&Task{}, id).Error; err != nil { + c.JSON(500, gin.H{"error": err.Error()}) + return + } + c.Status(204) +} diff --git a/main.go b/main.go index c2f552f..325cc69 100644 --- a/main.go +++ b/main.go @@ -1,98 +1,33 @@ package main import ( - "encoding/json" - "errors" - "fmt" - "io" "net/http" - "net/url" "os" "path/filepath" - "regexp" - "strconv" - "strings" - "sync" - "time" "github.com/gin-gonic/gin" "github.com/glebarez/sqlite" - "github.com/studio-b12/gowebdav" "gorm.io/gorm" ) -type Task struct { - ID uint `gorm:"primaryKey" json:"id"` - Source string `gorm:"uniqueIndex:idx_tasks_source_rid_quality" json:"source"` - RID string `gorm:"uniqueIndex:idx_tasks_source_rid_quality" json:"rid"` - Title string `json:"title"` - Artist string `json:"artist"` - Album string `json:"album"` - Pic string `json:"pic"` - Quality string `gorm:"uniqueIndex:idx_tasks_source_rid_quality" json:"quality"` - ResolvedQuality string `json:"resolvedQuality,omitempty"` - ActualQuality string `json:"actualQuality,omitempty"` - Format string `json:"format"` - Status string `json:"status"` - Error string `json:"error,omitempty"` - LocalPath string `json:"localPath,omitempty"` - RemotePath string `json:"remotePath,omitempty"` - TotalBytes int64 `json:"totalBytes"` - DownloadedBytes int64 `json:"downloadedBytes"` - Progress float64 `json:"progress"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` -} - -type Setting struct { - ID uint `gorm:"primaryKey"` - Key string `gorm:"uniqueIndex"` - Value string -} -type SongRequest struct { - Source string `json:"source"` - RID string `json:"rid"` - Title string `json:"title"` - Artist string `json:"artist"` - Album string `json:"album"` - Pic string `json:"pic"` - Quality string `json:"quality"` -} -type Config struct { - AMAPI, DownloadDir, WebDAVURL, WebDAVUser, WebDAVPass, WebDAVPath, DefaultQuality, WebDAVVersionPolicy, LocalFilePolicy string - WorkerCount int -} -type App struct { - db *gorm.DB - cfg Config - queue chan uint - wg sync.WaitGroup - httpClient *http.Client -} - -type playInfo struct { - URL string `json:"url"` - Level string `json:"level"` - Format string `json:"format"` - Headers map[string]string `json:"headers"` -} - func main() { dbPath := getenv("DB_PATH", "/data/downloader.db") _ = os.MkdirAll(filepath.Dir(dbPath), 0755) db, err := gorm.Open(sqlite.Open(dbPath), &gorm.Config{}) must(err) must(db.AutoMigrate(&Task{}, &Setting{})) - a := &App{db: db, cfg: Config{AMAPI: getenv("AM_API_URL", "http://127.0.0.1:9530/api"), DownloadDir: getenv("DOWNLOAD_DIR", "/downloads"), DefaultQuality: getenv("DEFAULT_QUALITY", "standard"), WorkerCount: atoi(getenv("WORKERS", "2"), 2)}, queue: make(chan uint, 100), httpClient: &http.Client{Timeout: 2 * time.Minute}} - _ = os.MkdirAll(a.cfg.DownloadDir, 0755) + + client := &http.Client{Timeout: 2 * 60 * 1e9} + cfg := loadConfig(db) + a := &App{db: db, cfg: cfg, queue: make(chan uint, 100), httpClient: client} a.loadSettings() - if a.cfg.WebDAVVersionPolicy == "" { - a.cfg.WebDAVVersionPolicy = "keep" - } - if a.cfg.LocalFilePolicy == "" { - a.cfg.LocalFilePolicy = "keep" + normalizeConfig(&a.cfg) + a.am = &AMAPIClient{BaseURL: a.cfg.AMAPI, HTTP: client} + if a.cfg.WebDAVURL != "" { + a.storage = NewWebDAVStorage(a.cfg) } a.startWorkers() + r := gin.Default() r.StaticFile("/", "/app/index.html") api := r.Group("/api") @@ -107,490 +42,8 @@ func main() { must(r.Run(":" + getenv("PORT", "9540"))) } -func (a *App) loadSettings() { - var ss []Setting - a.db.Find(&ss) - for _, s := range ss { - switch s.Key { - case "am_api": - a.cfg.AMAPI = s.Value - case "download_dir": - a.cfg.DownloadDir = s.Value - case "webdav_url": - a.cfg.WebDAVURL = s.Value - case "webdav_user": - a.cfg.WebDAVUser = s.Value - case "webdav_pass": - a.cfg.WebDAVPass = s.Value - case "webdav_path": - a.cfg.WebDAVPath = s.Value - case "quality": - a.cfg.DefaultQuality = s.Value - case "webdav_version_policy": - a.cfg.WebDAVVersionPolicy = s.Value - case "local_file_policy": - a.cfg.LocalFilePolicy = s.Value - } - } -} -func (a *App) save(k, v string) { - var s Setting - if a.db.Where("key = ?", k).First(&s).Error != nil { - a.db.Create(&Setting{Key: k, Value: v}) - } else { - s.Value = v - a.db.Save(&s) - } -} -func (a *App) startWorkers() { - for i := 0; i < a.cfg.WorkerCount; i++ { - a.wg.Add(1) - go func() { - defer a.wg.Done() - for id := range a.queue { - a.download(id) - } - }() - } -} -func (a *App) addTask(c *gin.Context) { - var in SongRequest - if c.BindJSON(&in) != nil || in.RID == "" || in.Title == "" { - c.JSON(400, gin.H{"error": "source/rid/title required"}) - return - } - if in.Source == "" { - in.Source = "kw" - } - if in.Quality == "" { - in.Quality = a.cfg.DefaultQuality - } - var existing Task - err := a.db.Where("source = ? AND r_id = ? AND quality = ?", in.Source, in.RID, in.Quality).Order("id desc").First(&existing).Error - if err == nil { - if existing.Status != "error" { - c.JSON(200, existing) - return - } - a.db.Model(&existing).Updates(map[string]any{"title": in.Title, "artist": in.Artist, "album": in.Album, "pic": in.Pic, "status": "pending", "error": ""}) - a.queue <- existing.ID - c.JSON(202, existing) - return - } - t := Task{Source: in.Source, RID: in.RID, Title: in.Title, Artist: in.Artist, Album: in.Album, Pic: in.Pic, Quality: in.Quality, Status: "pending"} - if err := a.db.Create(&t).Error; err != nil { - // A concurrent request may have inserted the same task; return that task. - if a.db.Where("source = ? AND r_id = ? AND quality = ?", in.Source, in.RID, in.Quality).Order("id desc").First(&existing).Error == nil { - c.JSON(200, existing) - return - } - c.JSON(500, gin.H{"error": err.Error()}) - return - } - a.queue <- t.ID - c.JSON(202, t) -} -func (a *App) search(c *gin.Context) { - keyword := strings.TrimSpace(c.Query("keyword")) - source := c.DefaultQuery("source", "kw") - if keyword == "" { - c.JSON(400, gin.H{"error": "keyword required"}) - return - } - u, _ := url.Parse(strings.TrimRight(a.cfg.AMAPI, "/") + "/search") - q := u.Query() - q.Set("keyword", keyword) - q.Set("source", source) - q.Set("page", c.DefaultQuery("page", "1")) - q.Set("limit", c.DefaultQuery("limit", "20")) - u.RawQuery = q.Encode() - resp, err := a.httpClient.Get(u.String()) +func must(err error) { if err != nil { - c.JSON(502, gin.H{"error": err.Error()}) - return - } - defer resp.Body.Close() - c.Status(resp.StatusCode) - _, _ = io.Copy(c.Writer, resp.Body) -} - -func (a *App) listTasks(c *gin.Context) { - var ts []Task - a.db.Order("id desc").Limit(100).Find(&ts) - c.JSON(200, gin.H{"data": ts}) -} -func (a *App) retryTask(c *gin.Context) { - id, _ := strconv.ParseUint(c.Param("id"), 10, 64) - var t Task - if a.db.First(&t, id).Error != nil { - c.Status(404) - return - } - a.db.Model(&t).Updates(map[string]any{"status": "pending", "error": "", "progress": 0, "downloaded_bytes": 0}) - t.Status = "pending" - t.Error = "" - a.queue <- t.ID - c.JSON(202, t) -} -func (a *App) deleteTask(c *gin.Context) { - id, _ := strconv.ParseUint(c.Param("id"), 10, 64) - a.db.Delete(&Task{}, id) - c.Status(204) -} - -func (a *App) download(id uint) { - var t Task - if a.db.First(&t, id).Error != nil { - return - } - // A task may be queued twice; only pending tasks are eligible to start. - if t.Status != "pending" { - return - } - a.db.Model(&t).Updates(map[string]any{"status": "resolving", "error": "", "progress": 0}) - info, err := a.resolve(t) - if err != nil { - a.fail(&t, err) - return - } - priorLocal := t.LocalPath - ext := normalizedFormat(info.Format) - if ext == "" { - ext = normalizedFormat(filepath.Ext(info.URL)) - } - if ext == "" { - ext = "mp3" - } - filename := safe(t.Artist) + " - " + safe(t.Title) + "." + ext - local := filepath.Join(a.cfg.DownloadDir, filename) - t.Format = ext - t.ResolvedQuality = info.Level - t.ActualQuality = "" - t.LocalPath = local - if priorLocal != "" { - if fi, statErr := os.Stat(priorLocal); statErr == nil && fi.Size() > 0 { - local = priorLocal - t.LocalPath = local - } - } - if fi, statErr := os.Stat(local); statErr == nil && fi.Size() > 0 { - actual, detectErr := detectAudioFormat(local) - if detectErr == nil { - actual = normalizedFormat(actual) - if actual != t.Format { - newLocal, renameErr := renameToActualFormat(local, actual) - if renameErr != nil { - a.fail(&t, renameErr) - return - } - local = newLocal - t.LocalPath = local - } - t.Format = actual - t.ActualQuality = actualQuality(t.Quality, t.ResolvedQuality, actual) - } - t.TotalBytes = fi.Size() - t.DownloadedBytes = fi.Size() - t.Progress = 100 - a.db.Save(&t) - a.upload(&t) - return - } - a.db.Model(&t).Updates(map[string]any{"status": "downloading", "format": ext, "local_path": local}) - req, err := http.NewRequest("GET", info.URL, nil) - if err != nil { - a.fail(&t, err) - return - } - for k, v := range info.Headers { - req.Header.Set(k, v) - } - resp, err := a.httpClient.Do(req) - if err != nil { - a.fail(&t, err) - return - } - defer resp.Body.Close() - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - a.fail(&t, fmt.Errorf("media http %d", resp.StatusCode)) - return - } - t.Status = "downloading" - t.TotalBytes = resp.ContentLength - t.DownloadedBytes = 0 - t.Progress = 0 - a.db.Save(&t) - tmp := local + ".part" - f, err := os.Create(tmp) - if err != nil { - a.fail(&t, err) - return - } - var n int64 - buf := make([]byte, 256*1024) - lastReport := int64(0) - for { - nr, er := resp.Body.Read(buf) - if nr > 0 { - wn, ew := f.Write(buf[:nr]) - n += int64(wn) - if ew != nil { - err = ew - break - } - if n-lastReport >= 1024*1024 { - progress := 0.0 - if resp.ContentLength > 0 { - progress = float64(n) * 100 / float64(resp.ContentLength) - } - a.db.Model(&t).Updates(map[string]any{"downloaded_bytes": n, "total_bytes": resp.ContentLength, "progress": progress}) - lastReport = n - } - } - if er == io.EOF { - break - } - if er != nil { - err = er - break - } - } - cerr := f.Close() - if err != nil { - a.fail(&t, err) - return - } - if cerr != nil { - a.fail(&t, cerr) - return - } - if t.TotalBytes > 0 && n != t.TotalBytes { - a.fail(&t, fmt.Errorf("incomplete download: got %d of %d bytes", n, t.TotalBytes)) - return - } - actual, detectErr := detectAudioFormat(tmp) - if detectErr != nil { - a.fail(&t, fmt.Errorf("detect audio format: %w", detectErr)) - _ = os.Remove(tmp) - return - } - actual = normalizedFormat(actual) - local = strings.TrimSuffix(local, filepath.Ext(local)) + "." + actual - if err = os.Rename(tmp, local); err != nil { - a.fail(&t, err) - _ = os.Remove(tmp) - return - } - t.Format = actual - t.LocalPath = local - t.ActualQuality = actualQuality(t.Quality, t.ResolvedQuality, actual) - t.DownloadedBytes = n - if t.TotalBytes > 0 { - t.Progress = 100 - } - a.db.Save(&t) - a.upload(&t) -} - -func (a *App) resolve(t Task) (playInfo, error) { - u, _ := url.Parse(strings.TrimRight(a.cfg.AMAPI, "/") + "/song/play-url") - q := u.Query() - q.Set("source", t.Source) - q.Set("rid", t.RID) - q.Set("level", t.Quality) - u.RawQuery = q.Encode() - resp, err := a.httpClient.Get(u.String()) - if err != nil { - return playInfo{}, err - } - defer resp.Body.Close() - var x struct { - Code int `json:"code"` - Data playInfo `json:"data"` - Message string `json:"message"` - } - if err = json.NewDecoder(resp.Body).Decode(&x); err != nil { - return playInfo{}, err - } - if x.Code != 0 || x.Data.URL == "" { - return playInfo{}, errors.New(x.Message) - } - return x.Data, nil -} - -func (a *App) upload(t *Task) { - if a.cfg.WebDAVURL == "" { - t.Status = "done" - a.db.Save(t) - return - } - a.db.Model(t).Updates(map[string]any{"status": "uploading"}) - remote := strings.TrimRight(a.cfg.WebDAVPath, "/") + "/" + safe(t.Artist) + "/" + safe(t.Album) + "/" + safe(t.Title) + "." + safe(t.Format) - t.RemotePath = remote - remoteDir := filepath.ToSlash(filepath.Dir(remote)) - client := gowebdav.NewClient(a.cfg.WebDAVURL, a.cfg.WebDAVUser, a.cfg.WebDAVPass) - client.SetTimeout(10 * time.Minute) - if err := client.MkdirAll(remoteDir, 0755); err != nil { - a.fail(t, err) - return - } - if fi, err := os.Stat(t.LocalPath); err == nil { - if ri, err := client.Stat(remote); err == nil && ri.Size() == fi.Size() { - t.Status = "done" - t.Progress = 100 - a.db.Save(t) - a.cleanupLocalFile(t) - return - } - } - f, err := os.Open(t.LocalPath) - if err != nil { - a.fail(t, err) - return - } - defer f.Close() - if err = client.WriteStream(remote, f, 0644); err != nil { - a.fail(t, err) - return - } - if a.cfg.WebDAVVersionPolicy == "replace_lower" { - a.removeLowerQualityVersions(client, remote, t) - } - t.Status = "done" - t.Progress = 100 - a.db.Save(t) - a.cleanupLocalFile(t) -} -func (a *App) cleanupLocalFile(t *Task) { - if a.cfg.LocalFilePolicy != "delete_after_upload" || t.LocalPath == "" { - return - } - if err := os.Remove(t.LocalPath); err != nil && !os.IsNotExist(err) { - t.Error = "WebDAV 上传成功，但删除本地文件失败: " + err.Error() - a.db.Save(t) - } -} -func (a *App) removeLowerQualityVersions(client *gowebdav.Client, remote string, t *Task) { - levels := map[string]int{"standard": 1, "exhigh": 2, "sq": 3, "lossless": 4, "hires": 5, "jyeffect": 6, "sky": 7, "jymaster": 8, "atmos": 9, "atmos_plus": 10, "master": 11} - current := levels[t.ActualQuality] - if current == 0 { - current = levels[t.ResolvedQuality] - } - if current == 0 { - current = levels[t.Quality] - } - if current == 0 { - return - } - var old []Task - a.db.Where("artist = ? AND title = ? AND album = ? AND status = ? AND id <> ?", t.Artist, t.Title, t.Album, "done", t.ID).Find(&old) - for _, candidate := range old { - q := candidate.ActualQuality - if q == "" { - q = candidate.ResolvedQuality - } - if q == "" { - q = candidate.Quality - } - if levels[q] <= 0 || levels[q] >= current || candidate.RemotePath == "" || candidate.RemotePath == remote { - continue - } - if _, err := client.Stat(candidate.RemotePath); err == nil { - _ = client.Remove(candidate.RemotePath) - } - } -} - -func (a *App) fail(t *Task, err error) { - if t.LocalPath != "" { - _ = os.Remove(t.LocalPath + ".part") - } - t.Status = "error" - t.Error = err.Error() - a.db.Save(t) -} - -func (a *App) getSettings(c *gin.Context) { - c.JSON(200, gin.H{"amApi": a.cfg.AMAPI, "downloadDir": a.cfg.DownloadDir, "webdavUrl": a.cfg.WebDAVURL, "webdavUser": a.cfg.WebDAVUser, "webdavPass": a.cfg.WebDAVPass, "webdavPath": a.cfg.WebDAVPath, "quality": a.cfg.DefaultQuality, "webdavVersionPolicy": a.cfg.WebDAVVersionPolicy, "localFilePolicy": a.cfg.LocalFilePolicy}) -} -func (a *App) putSettings(c *gin.Context) { - var v map[string]string - if c.BindJSON(&v) != nil { - c.Status(400) - return - } - if x := v["amApi"]; x != "" { - a.cfg.AMAPI = x - a.save("am_api", x) - } - if p := v["webdavVersionPolicy"]; p != "keep" && p != "replace_lower" { - v["webdavVersionPolicy"] = "keep" - } - if p := v["localFilePolicy"]; p != "keep" && p != "delete_after_upload" { - v["localFilePolicy"] = "keep" - } - if x := v["downloadDir"]; x != "" { - a.cfg.DownloadDir = x - a.save("download_dir", x) - os.MkdirAll(x, 0755) - } - a.cfg.WebDAVURL = v["webdavUrl"] - a.cfg.WebDAVUser = v["webdavUser"] - a.cfg.WebDAVPass = v["webdavPass"] - a.cfg.WebDAVPath = strings.Trim(v["webdavPath"], "/") - a.cfg.DefaultQuality = v["quality"] - if p := v["webdavVersionPolicy"]; p == "keep" || p == "replace_lower" { - a.cfg.WebDAVVersionPolicy = p - } - if p := v["localFilePolicy"]; p == "keep" || p == "delete_after_upload" { - a.cfg.LocalFilePolicy = p - } - for k, x := range map[string]string{"webdav_url": a.cfg.WebDAVURL, "webdav_user": a.cfg.WebDAVUser, "webdav_pass": a.cfg.WebDAVPass, "webdav_path": a.cfg.WebDAVPath, "quality": a.cfg.DefaultQuality, "webdav_version_policy": a.cfg.WebDAVVersionPolicy, "local_file_policy": a.cfg.LocalFilePolicy} { - a.save(k, x) - } - c.JSON(200, gin.H{"ok": true}) -} -func (a *App) testWebDAV(c *gin.Context) { - var v struct{ URL, User, Pass, Path string } - if c.BindJSON(&v) != nil { - c.Status(400) - return - } - client := gowebdav.NewClient(v.URL, v.User, v.Pass) - client.SetTimeout(20 * time.Second) - if err := client.Connect(); err != nil { - c.JSON(502, gin.H{"ok": false, "error": err.Error()}) - return - } - c.JSON(200, gin.H{"ok": true, "path": v.Path}) -} - -var bad = regexp.MustCompile(`[/\\:*?"<>|]+`) - -func safe(s string) string { - s = strings.TrimSpace(s) - s = bad.ReplaceAllString(s, "_") - s = strings.Trim(s, ". ") - if s == "" { - return "Unknown" - } - return s -} -func getenv(k, d string) string { - if v := os.Getenv(k); v != "" { - return v - } - return d -} -func atoi(s string, d int) int { - v, e := strconv.Atoi(s) - if e != nil || v < 1 { - return d - } - return v -} -func must(e error) { - if e != nil { - panic(e) + panic(err) } } diff --git a/models.go b/models.go new file mode 100644 index 0000000..853a5f9 --- /dev/null +++ b/models.go @@ -0,0 +1,54 @@ +package main + +import "time" + +type Task struct { + ID uint `gorm:"primaryKey" json:"id"` + Source string `gorm:"uniqueIndex:idx_tasks_source_rid_quality" json:"source"` + RID string `gorm:"uniqueIndex:idx_tasks_source_rid_quality" json:"rid"` + Title string `json:"title"` + Artist string `json:"artist"` + Album string `json:"album"` + Pic string `json:"pic"` + Quality string `gorm:"uniqueIndex:idx_tasks_source_rid_quality" json:"quality"` + ResolvedQuality string `json:"resolvedQuality,omitempty"` + ActualQuality string `json:"actualQuality,omitempty"` + Format string `json:"format"` + Status string `json:"status"` + Error string `json:"error,omitempty"` + LocalPath string `json:"localPath,omitempty"` + RemotePath string `json:"remotePath,omitempty"` + TotalBytes int64 `json:"totalBytes"` + DownloadedBytes int64 `json:"downloadedBytes"` + Progress float64 `json:"progress"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +type Setting struct { + ID uint `gorm:"primaryKey"` + Key string `gorm:"uniqueIndex"` + Value string +} + +type SongRequest struct { + Source string `json:"source"` + RID string `json:"rid"` + Title string `json:"title"` + Artist string `json:"artist"` + Album string `json:"album"` + Pic string `json:"pic"` + Quality string `json:"quality"` +} + +type Config struct { + AMAPI, DownloadDir, WebDAVURL, WebDAVUser, WebDAVPass, WebDAVPath, DefaultQuality, WebDAVVersionPolicy, LocalFilePolicy string + WorkerCount int +} + +type playInfo struct { + URL string `json:"url"` + Level string `json:"level"` + Format string `json:"format"` + Headers map[string]string `json:"headers"` +} diff --git a/remote_storage.go b/remote_storage.go new file mode 100644 index 0000000..782501e --- /dev/null +++ b/remote_storage.go @@ -0,0 +1,41 @@ +package main + +import ( + "os" + "time" + + "github.com/studio-b12/gowebdav" +) + +type WebDAVStorage struct{ client *gowebdav.Client } + +func NewWebDAVStorage(cfg Config) *WebDAVStorage { + c := gowebdav.NewClient(cfg.WebDAVURL, cfg.WebDAVUser, cfg.WebDAVPass) + c.SetTimeout(10 * time.Minute) + return &WebDAVStorage{client: c} +} + +func (s *WebDAVStorage) MkdirAll(remoteDir string) error { return s.client.MkdirAll(remoteDir, 0755) } + +func (s *WebDAVStorage) ExistsSameSize(remotePath string, size int64) (bool, error) { + fi, err := s.client.Stat(remotePath) + if err != nil { + return false, err + } + return fi.Size() == size, nil +} + +func (s *WebDAVStorage) Upload(localPath, remotePath string) error { + f, err := os.Open(localPath) + if err != nil { + return err + } + defer f.Close() + return s.client.WriteStream(remotePath, f, 0644) +} + +func (s *WebDAVStorage) Remove(remotePath string) error { return s.client.Remove(remotePath) } +func (s *WebDAVStorage) Stat(remotePath string) error { + _, err := s.client.Stat(remotePath) + return err +} diff --git a/settings.go b/settings.go new file mode 100644 index 0000000..4894eff --- /dev/null +++ b/settings.go @@ -0,0 +1,85 @@ +package main + +import ( + "strings" + "time" + + "github.com/gin-gonic/gin" + "github.com/studio-b12/gowebdav" +) + +func (a *App) getSettings(c *gin.Context) { + c.JSON(200, gin.H{ + "amApi": a.cfg.AMAPI, "downloadDir": a.cfg.DownloadDir, + "webdavUrl": a.cfg.WebDAVURL, "webdavUser": a.cfg.WebDAVUser, + "webdavPass": a.cfg.WebDAVPass, "webdavPath": a.cfg.WebDAVPath, + "quality": a.cfg.DefaultQuality, "webdavVersionPolicy": a.cfg.WebDAVVersionPolicy, + "localFilePolicy": a.cfg.LocalFilePolicy, + }) +} + +func (a *App) putSettings(c *gin.Context) { + var v map[string]string + if c.BindJSON(&v) != nil { + c.Status(400) + return + } + if x := v["amApi"]; x != "" { + a.cfg.AMAPI = x + a.saveSetting("am_api", x) + a.am = &AMAPIClient{BaseURL: x, HTTP: a.httpClient} + } + if x := v["downloadDir"]; x != "" { + a.cfg.DownloadDir = x + a.saveSetting("download_dir", x) + } + if p := v["webdavVersionPolicy"]; p == "keep" || p == "replace_lower" { + a.cfg.WebDAVVersionPolicy = p + } + if p := v["localFilePolicy"]; p == "keep" || p == "delete_after_upload" { + a.cfg.LocalFilePolicy = p + } + if x, ok := v["webdavUrl"]; ok { + a.cfg.WebDAVURL = x + } + if x, ok := v["webdavUser"]; ok { + a.cfg.WebDAVUser = x + } + if x, ok := v["webdavPass"]; ok { + a.cfg.WebDAVPass = x + } + if x, ok := v["webdavPath"]; ok { + a.cfg.WebDAVPath = strings.Trim(x, "/") + } + if x, ok := v["quality"]; ok && x != "" { + a.cfg.DefaultQuality = x + } + for k, x := range map[string]string{ + "webdav_url": a.cfg.WebDAVURL, "webdav_user": a.cfg.WebDAVUser, "webdav_pass": a.cfg.WebDAVPass, + "webdav_path": a.cfg.WebDAVPath, "quality": a.cfg.DefaultQuality, + "webdav_version_policy": a.cfg.WebDAVVersionPolicy, "local_file_policy": a.cfg.LocalFilePolicy, + } { + a.saveSetting(k, x) + } + if a.cfg.WebDAVURL != "" { + a.storage = NewWebDAVStorage(a.cfg) + } else { + a.storage = nil + } + c.JSON(200, gin.H{"ok": true}) +} + +func (a *App) testWebDAV(c *gin.Context) { + var v struct{ URL, User, Pass, Path string } + if c.BindJSON(&v) != nil { + c.Status(400) + return + } + client := gowebdav.NewClient(v.URL, v.User, v.Pass) + client.SetTimeout(20 * time.Second) + if err := client.Connect(); err != nil { + c.JSON(502, gin.H{"ok": false, "error": err.Error()}) + return + } + c.JSON(200, gin.H{"ok": true, "path": v.Path}) +} diff --git a/util.go b/util.go new file mode 100644 index 0000000..a54d702 --- /dev/null +++ b/util.go @@ -0,0 +1,37 @@ +package main + +import ( + "errors" + "os" + "regexp" + "strings" + + "path/filepath" +) + +func (a *App) fail(t *Task, err error) { + if t.LocalPath != "" { + _ = os.Remove(t.LocalPath + ".part") + } + if err == nil { + err = errors.New("unknown download error") + } + t.Status, t.Error = "error", err.Error() + a.db.Save(t) +} + +var bad = regexp.MustCompile(`[/\\:*?"<>|]+`) + +func safe(s string) string { + s = strings.TrimSpace(s) + s = bad.ReplaceAllString(s, "_") + s = strings.Trim(s, ". ") + if s == "" { + return "Unknown" + } + return s +} + +func audioPath(dir, artist, title, format string) string { + return filepath.Join(dir, safe(artist)+" - "+safe(title)+"."+normalizedFormat(format)) +} diff --git a/webdav.go b/webdav.go new file mode 100644 index 0000000..259e99d --- /dev/null +++ b/webdav.go @@ -0,0 +1,76 @@ +package main + +import ( + "os" + "path/filepath" + "strings" +) + +func (a *App) upload(t *Task) { + if a.storage == nil || a.cfg.WebDAVURL == "" { + t.Status, t.Progress = "done", 100 + a.db.Save(t) + return + } + a.db.Model(t).Updates(map[string]any{"status": "uploading"}) + remote := strings.TrimRight(a.cfg.WebDAVPath, "/") + "/" + safe(t.Artist) + "/" + safe(t.Album) + "/" + safe(t.Title) + "." + safe(t.Format) + t.RemotePath = remote + if err := a.storage.MkdirAll(filepath.ToSlash(filepath.Dir(remote))); err != nil { + a.fail(t, err) + return + } + if fi, err := os.Stat(t.LocalPath); err == nil { + if same, _ := a.storage.ExistsSameSize(remote, fi.Size()); same { + t.Status, t.Progress = "done", 100 + a.db.Save(t) + a.cleanupLocalFile(t) + return + } + } + if err := a.storage.Upload(t.LocalPath, remote); err != nil { + a.fail(t, err) + return + } + if a.cfg.WebDAVVersionPolicy == "replace_lower" { + a.cleanupLowerQualityVersions(t) + } + t.Status, t.Progress = "done", 100 + a.db.Save(t) + a.cleanupLocalFile(t) +} + +func (a *App) cleanupLocalFile(t *Task) { + if a.cfg.LocalFilePolicy != "delete_after_upload" || t.LocalPath == "" { + return + } + if err := os.Remove(t.LocalPath); err != nil && !os.IsNotExist(err) { + t.Error = "WebDAV 上传成功，但删除本地文件失败: " + err.Error() + a.db.Save(t) + } +} + +func (a *App) cleanupLowerQualityVersions(t *Task) { + levels := map[string]int{"standard": 1, "exhigh": 2, "sq": 3, "lossless": 4, "hires": 5, "jyeffect": 6, "sky": 7, "jymaster": 8, "atmos": 9, "atmos_plus": 10, "master": 11} + current := qualityRank(levels, t.ActualQuality, t.ResolvedQuality, t.Quality) + if current == 0 { + return + } + var old []Task + a.db.Where("artist = ? AND title = ? AND album = ? AND status = ? AND id <> ?", t.Artist, t.Title, t.Album, "done", t.ID).Find(&old) + for _, candidate := range old { + q := qualityRank(levels, candidate.ActualQuality, candidate.ResolvedQuality, candidate.Quality) + if q == 0 || q >= current || candidate.RemotePath == "" || candidate.RemotePath == t.RemotePath { + continue + } + _ = a.storage.Remove(candidate.RemotePath) + } +} + +func qualityRank(levels map[string]int, values ...string) int { + for _, q := range values { + if n := levels[q]; n > 0 { + return n + } + } + return 0 +}