toor
0f927147cd
原 Electron 版在这台 Windows 机器上无法启动（0x80000003 崩在 V8 初始化）， 整个业务逻辑用 Go 重写，前端界面与协议结论完全保留。 - internal/eh: 协议核心，画廊页分页 -> /s/ -> hath 直链 -> 下载 - internal/core: 并发引擎 / 状态存储 / 目录模板 / 油猴桥口 - app.go: Wails 绑定层，替代 Electron IPC - frontend/js/api.js: 适配层，renderer.js 一行未改 - 37 个测试，含 httptest 模拟 EH 的端到端链路验证 相比 Electron 版：90MB -> 12.4MB，180MB 内存 -> 27MB，构建 90s -> 2s。 顺带修掉原版代理探测的 bug（永远返回第一个端口，从不等连接结果）。
421 lines
8.8 KiB
Go
421 lines
8.8 KiB
Go
package core
|
|
|
|
// 下载引擎：任务编排 / 并发控制 / 指数退避重试 / 断点续传
|
|
//
|
|
// 断点续传策略：以「文件系统为准」——
|
|
//
|
|
// 目标文件已存在且 size > 0 → 视为已完成，直接跳过。
|
|
//
|
|
// 因此即使任务状态 JSON 丢失或损坏，重跑也能正确续传。
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"math"
|
|
"math/rand"
|
|
"os"
|
|
"path/filepath"
|
|
"regexp"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"eh-downloader-wails/internal/eh"
|
|
)
|
|
|
|
// ErrCanceled 取消标记（用户主动停止不算失败）
|
|
var ErrCanceled = errors.New("已取消")
|
|
|
|
// retryableErrs 可重试的错误特征（对应 JS 版 RETRYABLE 集合）
|
|
var retryablePatterns = []string{
|
|
"connection reset", "connection refused", "timeout", "i/o timeout",
|
|
"broken pipe", "eof", "no such host", "tls handshake",
|
|
"unexpected status", "server closed", "too many redirects",
|
|
"429", "502", "503", "504", "socket hang up",
|
|
}
|
|
|
|
// EventSink 是引擎向 UI 发事件的方式（由 app 层注入）
|
|
type EventSink interface {
|
|
TaskUpdate(t Task)
|
|
Log(level, msg string)
|
|
Done()
|
|
}
|
|
|
|
// Engine 下载引擎
|
|
type Engine struct {
|
|
store *Store
|
|
client *eh.Client
|
|
sink EventSink
|
|
|
|
mu sync.Mutex
|
|
concurrency int
|
|
running bool
|
|
cancel context.CancelFunc
|
|
queue []Task
|
|
}
|
|
|
|
// NewEngine 建引擎
|
|
func NewEngine(store *Store, sink EventSink, settings Settings) *Engine {
|
|
e := &Engine{
|
|
store: store,
|
|
sink: sink,
|
|
client: eh.NewClient(time.Duration(settings.Timeout)*time.Millisecond,
|
|
settings.Proxy, settings.ProxyMode, settings.Xres),
|
|
concurrency: settings.Concurrency,
|
|
}
|
|
if e.concurrency < 1 {
|
|
e.concurrency = 3
|
|
}
|
|
if e.concurrency > 8 {
|
|
e.concurrency = 8
|
|
}
|
|
return e
|
|
}
|
|
|
|
// Client 暴露给 app 层改配置用
|
|
func (e *Engine) Client() *eh.Client { return e.client }
|
|
|
|
// SetConcurrency 运行时改并发
|
|
func (e *Engine) SetConcurrency(n int) {
|
|
if n < 1 {
|
|
n = 1
|
|
}
|
|
if n > 8 {
|
|
n = 8
|
|
}
|
|
e.mu.Lock()
|
|
e.concurrency = n
|
|
e.mu.Unlock()
|
|
}
|
|
|
|
func (e *Engine) IsRunning() bool {
|
|
e.mu.Lock()
|
|
defer e.mu.Unlock()
|
|
return e.running
|
|
}
|
|
|
|
func (e *Engine) QueueLength() int {
|
|
e.mu.Lock()
|
|
defer e.mu.Unlock()
|
|
return len(e.queue)
|
|
}
|
|
|
|
// Stop 停止：取消 context，立刻打断在途请求
|
|
func (e *Engine) Stop() {
|
|
e.mu.Lock()
|
|
if e.cancel != nil {
|
|
e.cancel()
|
|
e.cancel = nil
|
|
}
|
|
e.mu.Unlock()
|
|
}
|
|
|
|
// Enqueue 把任务加入队列（已排队的跳过）
|
|
func (e *Engine) Enqueue(tasks []Task) {
|
|
e.mu.Lock()
|
|
defer e.mu.Unlock()
|
|
for _, t := range tasks {
|
|
dup := false
|
|
for _, q := range e.queue {
|
|
if q.GID == t.GID && q.Token == t.Token {
|
|
dup = true
|
|
break
|
|
}
|
|
}
|
|
if !dup {
|
|
e.queue = append(e.queue, t)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Start 开始跑队列（阻塞直到队列空或被取消）
|
|
func (e *Engine) Start(ctx context.Context) {
|
|
e.mu.Lock()
|
|
if e.running {
|
|
e.mu.Unlock()
|
|
return
|
|
}
|
|
e.running = true
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
e.cancel = cancel
|
|
n := e.concurrency
|
|
e.mu.Unlock()
|
|
|
|
defer func() {
|
|
cancel()
|
|
e.mu.Lock()
|
|
e.running = false
|
|
e.cancel = nil
|
|
e.queue = nil
|
|
e.mu.Unlock()
|
|
if e.sink != nil {
|
|
e.sink.Done()
|
|
}
|
|
}()
|
|
|
|
// 并发 worker 共享一个 channel 队列
|
|
work := make(chan Task)
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < n; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case t, ok := <-work:
|
|
if !ok {
|
|
return
|
|
}
|
|
e.runTask(ctx, t)
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
// 派发
|
|
go func() {
|
|
defer close(work)
|
|
for {
|
|
e.mu.Lock()
|
|
if len(e.queue) == 0 {
|
|
e.mu.Unlock()
|
|
return
|
|
}
|
|
t := e.queue[0]
|
|
e.queue = e.queue[1:]
|
|
e.mu.Unlock()
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case work <- t:
|
|
}
|
|
}
|
|
}()
|
|
|
|
wg.Wait()
|
|
}
|
|
|
|
func (e *Engine) runTask(ctx context.Context, task Task) {
|
|
task.Status = "running"
|
|
task.Error = ""
|
|
task.Done = 0
|
|
task.Failed = 0
|
|
task.Pages = map[string]string{}
|
|
task.FailedList = []FailedItem{}
|
|
e.save(task)
|
|
|
|
// ① 画廊页（内部自动翻 ?p=N 分页收集全部页）
|
|
gallery, err := e.client.FetchGalleryPage(ctx, task.GID, task.Token)
|
|
if err != nil {
|
|
if ctx.Err() != nil {
|
|
e.markCanceled(task)
|
|
return
|
|
}
|
|
e.fail(task, fmt.Errorf("解析画廊页失败: %w", err))
|
|
return
|
|
}
|
|
|
|
task.Title = gallery.Title
|
|
task.TotalPages = gallery.TotalPages
|
|
task.Dir = BuildDir(e.store.Settings().DirTemplate, gallery)
|
|
|
|
task.Status = "downloading"
|
|
e.save(task)
|
|
|
|
// 目录按模板生成
|
|
root := e.store.Settings().DownloadRoot
|
|
if root == "" {
|
|
root = DefaultDownloadRoot()
|
|
}
|
|
dir := filepath.Join(root, task.Dir)
|
|
|
|
// 老版本一律用「标题」当目录名。用户中途换了模板时，
|
|
// 先看老的「标题目录」在不在 —— 在就继续往那儿下，
|
|
// 否则换个模板就会把几千张图全部重下一遍。
|
|
// ⚠️ 必须在 mkdir 之前判断，否则会白建出一个新目录。
|
|
legacyDir := filepath.Join(root, SanitizeSegment(gallery.Title, task.GID))
|
|
if dir != legacyDir {
|
|
if _, err := os.Stat(legacyDir); err == nil {
|
|
e.log("info", "沿用旧目录（模板改过，已有图在老位置）："+gallery.Title)
|
|
dir = legacyDir
|
|
rel, err := filepath.Rel(root, dir)
|
|
if err == nil {
|
|
task.Dir = rel
|
|
}
|
|
}
|
|
}
|
|
|
|
if err := os.MkdirAll(dir, 0o755); err != nil {
|
|
e.fail(task, fmt.Errorf("创建目录失败: %w", err))
|
|
return
|
|
}
|
|
e.save(task)
|
|
|
|
// ② ③ 逐页：解析 URL → 下载
|
|
total := len(gallery.Pages)
|
|
for _, p := range gallery.Pages {
|
|
if ctx.Err() != nil {
|
|
e.markCanceled(task)
|
|
return
|
|
}
|
|
|
|
fname := targetName(p, task.GID, total)
|
|
dest := filepath.Join(dir, fname)
|
|
|
|
// 断点续传：文件在就算完成（文件系统即真相，状态文件丢了也能续）
|
|
if st, err := os.Stat(dest); err == nil && st.Size() > 0 {
|
|
task.Done++
|
|
task.Pages[strconv.Itoa(p.Page)] = "done"
|
|
e.progress(task)
|
|
continue
|
|
}
|
|
|
|
page := p.Page
|
|
err := e.withRetry(ctx, func() error {
|
|
imgURL, err := e.client.ResolveImageURL(ctx, p.SPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = e.client.DownloadImage(ctx, imgURL, dest, nil)
|
|
return err
|
|
})
|
|
if err != nil {
|
|
if ctx.Err() != nil || errors.Is(err, ErrCanceled) {
|
|
e.markCanceled(task)
|
|
return
|
|
}
|
|
task.Failed++
|
|
task.Pages[strconv.Itoa(page)] = "failed"
|
|
task.FailedList = append(task.FailedList, FailedItem{
|
|
Page: page, File: fname, Error: err.Error(),
|
|
})
|
|
e.progress(task)
|
|
continue
|
|
}
|
|
|
|
task.Done++
|
|
task.Pages[strconv.Itoa(page)] = "done"
|
|
e.progress(task)
|
|
}
|
|
|
|
if task.Failed > 0 {
|
|
task.Status = "partial"
|
|
} else {
|
|
task.Status = "done"
|
|
}
|
|
e.save(task)
|
|
}
|
|
|
|
// targetName 页码补零，保证排序
|
|
func targetName(p eh.GalleryPageInfo, gid string, total int) string {
|
|
width := len(strconv.Itoa(total))
|
|
n := strconv.Itoa(p.Page)
|
|
for len(n) < width {
|
|
n = "0" + n
|
|
}
|
|
return SanitizeFilename(p.Filename, gid+"-"+n+".webp")
|
|
}
|
|
|
|
// withRetry 指数退避重试
|
|
func (e *Engine) withRetry(ctx context.Context, fn func() error) error {
|
|
const (
|
|
tries = 4
|
|
base = 800 * time.Millisecond
|
|
max = 15 * time.Second
|
|
)
|
|
|
|
var lastErr error
|
|
for i := 0; i < tries; i++ {
|
|
if ctx.Err() != nil {
|
|
return ErrCanceled
|
|
}
|
|
err := fn()
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
if ctx.Err() != nil {
|
|
return ErrCanceled
|
|
}
|
|
lastErr = err
|
|
|
|
if i == tries-1 || !isRetryable(err) {
|
|
return err
|
|
}
|
|
wait := time.Duration(float64(base) * math.Pow(2, float64(i)))
|
|
if wait > max {
|
|
wait = max
|
|
}
|
|
wait += time.Duration(rand.Intn(400)) * time.Millisecond
|
|
e.log("warn", fmt.Sprintf("第 %d 次失败，%dms 后重试：%v", i+1, wait.Milliseconds(), err))
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
return ErrCanceled
|
|
case <-time.After(wait):
|
|
}
|
|
}
|
|
return lastErr
|
|
}
|
|
|
|
// isRetryable 判断是否值得重试
|
|
func isRetryable(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
if errors.Is(err, ErrCanceled) || errors.Is(err, context.Canceled) {
|
|
return false
|
|
}
|
|
s := strings.ToLower(err.Error())
|
|
for _, p := range retryablePatterns {
|
|
if strings.Contains(s, p) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// markCanceled 用户主动取消：退回排队，等续传
|
|
func (e *Engine) markCanceled(task Task) {
|
|
task.Status = "queued"
|
|
task.Error = ""
|
|
e.save(task)
|
|
e.log("warn", fmt.Sprintf("画廊 %s 已取消，再点开始可继续", task.GID))
|
|
}
|
|
|
|
func (e *Engine) fail(task Task, err error) {
|
|
msg := err.Error()
|
|
task.Status = "failed"
|
|
task.Error = msg
|
|
e.save(task)
|
|
e.log("error", fmt.Sprintf("画廊 %s 失败：%s", task.GID, msg))
|
|
}
|
|
|
|
func (e *Engine) progress(task Task) {
|
|
if e.sink != nil {
|
|
e.sink.TaskUpdate(task)
|
|
}
|
|
}
|
|
|
|
func (e *Engine) save(task Task) {
|
|
task = e.store.UpsertTask(task)
|
|
if e.sink != nil {
|
|
e.sink.TaskUpdate(task)
|
|
}
|
|
}
|
|
|
|
func (e *Engine) log(level, msg string) {
|
|
if e.sink != nil {
|
|
e.sink.Log(level, msg)
|
|
}
|
|
}
|
|
|
|
// reSpaced 去掉路径里的多余空白（清洗用）
|
|
var reSpaced = regexp.MustCompile(`\s+`)
|
|
|
|
func cleanSpaces(s string) string { return reSpaced.ReplaceAllString(s, " ") }
|