first commit
This commit is contained in:
@@ -0,0 +1,954 @@
|
||||
package core
|
||||
|
||||
// 下载引擎：任务编排 / 并发控制 / 指数退避重试 / 断点续传
|
||||
//
|
||||
// 断点续传策略：以「文件系统为准」——
|
||||
//
|
||||
// 目标文件已存在且 size > 0 → 视为已完成，直接跳过。
|
||||
//
|
||||
// 因此即使任务状态 JSON 丢失或损坏，重跑也能正确续传。
|
||||
//
|
||||
// ── 与参考项目（EH）的关键差异 ─────────────────────────────────
|
||||
//
|
||||
// ① 图片 URL 是解析期一次性拿全的（4KHD 的直链稳定，不像 EH 那样每页
|
||||
// 都要现去 /s/ 换 hath 直链）。所以下载阶段只有「下载」一件事，
|
||||
// 没有 EH 那两步（ResolveImageURL → DownloadImage）。
|
||||
// ② 因此「清单失效自愈」不再是必需的：清单里的 URL 和重新解析拿到的一样。
|
||||
// 但保留重解析路径，用于「站点改图后作者补了新图」的情况。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"math"
|
||||
"math/rand"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"hd4k-downloader/internal/hd4k"
|
||||
)
|
||||
|
||||
// ErrCanceled 取消标记（用户主动停止不算失败）
|
||||
var ErrCanceled = errors.New("已取消")
|
||||
|
||||
// retryablePatterns 可重试的错误特征
|
||||
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", "500", "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 *hd4k.Client
|
||||
sink EventSink
|
||||
|
||||
mu sync.Mutex
|
||||
galleryWorkers int
|
||||
pageConcurrency 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: hd4k.NewClient(time.Duration(settings.Timeout)*time.Millisecond, settings.Proxy, settings.ProxyMode, settings.Sites),
|
||||
galleryWorkers: clampInt(settings.GalleryConcurrency, 1, 20, 5),
|
||||
pageConcurrency: clampInt(settings.PageConcurrency, 1, 20, 8),
|
||||
}
|
||||
e.client.SetRequestInterval(time.Duration(settings.RequestIntervalMs) * time.Millisecond)
|
||||
return e
|
||||
}
|
||||
|
||||
// clampInt 把值夹到 [lo, hi]。0 或负数视为「没设置」，用默认值。
|
||||
func clampInt(v, lo, hi, def int) int {
|
||||
if v <= 0 {
|
||||
return def
|
||||
}
|
||||
if v < lo {
|
||||
return lo
|
||||
}
|
||||
if v > hi {
|
||||
return hi
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
// Client 暴露给 app 层改配置用
|
||||
func (e *Engine) Client() *hd4k.Client { return e.client }
|
||||
|
||||
// SetConcurrency 运行时改画廊并发
|
||||
func (e *Engine) SetConcurrency(n int) {
|
||||
e.mu.Lock()
|
||||
e.galleryWorkers = clampInt(n, 1, 20, 5)
|
||||
e.mu.Unlock()
|
||||
}
|
||||
|
||||
// SetGalleryConcurrency 改画廊并发
|
||||
func (e *Engine) SetGalleryConcurrency(n int) { e.SetConcurrency(n) }
|
||||
|
||||
// SetPageConcurrency 改单画廊图片并发
|
||||
func (e *Engine) SetPageConcurrency(n int) {
|
||||
e.mu.Lock()
|
||||
e.pageConcurrency = clampInt(n, 1, 20, 8)
|
||||
e.mu.Unlock()
|
||||
}
|
||||
|
||||
// SetRequestInterval 改全局请求间隔（毫秒）
|
||||
func (e *Engine) SetRequestInterval(ms int) {
|
||||
if ms < 0 {
|
||||
ms = 0
|
||||
}
|
||||
e.client.SetRequestInterval(time.Duration(ms) * time.Millisecond)
|
||||
}
|
||||
|
||||
// GalleryConcurrency 当前画廊并发
|
||||
func (e *Engine) GalleryConcurrency() int {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
return e.galleryWorkers
|
||||
}
|
||||
|
||||
// PageConcurrency 当前图片并发
|
||||
func (e *Engine) PageConcurrency() int {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
return e.pageConcurrency
|
||||
}
|
||||
|
||||
func (e *Engine) emitTask(t Task) {
|
||||
if e.sink != nil {
|
||||
e.sink.TaskUpdate(t)
|
||||
}
|
||||
}
|
||||
|
||||
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.ID == t.ID {
|
||||
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
|
||||
e.mu.Unlock()
|
||||
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
e.mu.Lock()
|
||||
e.cancel = cancel
|
||||
n := e.galleryWorkers
|
||||
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()
|
||||
}
|
||||
}()
|
||||
|
||||
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.safeRunTask(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()
|
||||
}
|
||||
|
||||
// safeRunTask worker 崩溃保护：
|
||||
// 把这个画廊标记为失败并记一条日志，而不是让整个应用无声消失。
|
||||
func (e *Engine) safeRunTask(ctx context.Context, t Task) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
// 恢复代码本身也可能 panic，再包一层保证永不二次崩溃
|
||||
func() {
|
||||
defer func() { _ = recover() }()
|
||||
e.log("error", fmt.Sprintf("画廊 %s 处理时发生内部错误：%v", t.Slug, r))
|
||||
e.fail(t, fmt.Errorf("内部错误: %v", r))
|
||||
}()
|
||||
}
|
||||
}()
|
||||
e.runTask(ctx, t)
|
||||
}
|
||||
|
||||
type existingGallery struct {
|
||||
dir string
|
||||
manifest *Manifest
|
||||
}
|
||||
|
||||
// reportParseProgress 解析阶段进度反馈
|
||||
func (e *Engine) reportParseProgress(task Task, done, total, page int) {
|
||||
task.Status = "parsing"
|
||||
task.Total = total
|
||||
task.Done = 0
|
||||
task.Pages = map[string]string{}
|
||||
if done > 0 {
|
||||
task.Pages["_parsing"] = strconv.Itoa(done)
|
||||
}
|
||||
e.save(task)
|
||||
}
|
||||
|
||||
// findExistingGallery 在下载根目录里找这个画廊已存在的目录
|
||||
func (e *Engine) findExistingGallery(root string, task Task) *existingGallery {
|
||||
// 1) 按清单里的 ID 精确匹配（最可靠，拿到目录绝对路径）
|
||||
if md := FindManifestByID(root, task.ID); md != nil {
|
||||
return &existingGallery{dir: md.Dir, manifest: md.Manifest}
|
||||
}
|
||||
// 2) 没有清单：按可能的目录名猜（换过模板/旧版标题目录的情况）
|
||||
for _, name := range e.guessDirNames(task, nil) {
|
||||
p := filepath.Join(root, name)
|
||||
if st, err := os.Stat(p); err == nil && st.IsDir() {
|
||||
m, _ := ReadManifest(p)
|
||||
return &existingGallery{dir: p, manifest: m}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// guessDirNames 猜这个画廊可能的目录名
|
||||
func (e *Engine) guessDirNames(task Task, meta *hd4k.GalleryMeta) []string {
|
||||
tpl := e.store.Settings().DirTemplate
|
||||
if tpl == "" {
|
||||
tpl = "{title}"
|
||||
}
|
||||
|
||||
probe := hd4k.Gallery{
|
||||
ID: task.ID,
|
||||
URL: task.URL,
|
||||
Host: task.Host,
|
||||
Path: task.Path,
|
||||
Slug: task.Slug,
|
||||
Title: task.Title,
|
||||
}
|
||||
if meta != nil {
|
||||
probe.Meta = *meta
|
||||
}
|
||||
|
||||
var names []string
|
||||
if d := BuildDir(tpl, &probe); d != "" {
|
||||
names = append(names, d)
|
||||
}
|
||||
if task.Title != "" {
|
||||
if s := SanitizeSegment(task.Title, task.Slug); s != "" {
|
||||
names = append(names, s)
|
||||
}
|
||||
}
|
||||
return names
|
||||
}
|
||||
|
||||
// refFromTask 从任务还原一个 GalleryRef
|
||||
func refFromTask(t Task) hd4k.GalleryRef {
|
||||
return hd4k.GalleryRef{
|
||||
ID: t.ID,
|
||||
URL: t.URL,
|
||||
Host: t.Host,
|
||||
Path: t.Path,
|
||||
Slug: t.Slug,
|
||||
Raw: t.Raw,
|
||||
Title: t.Title,
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
|
||||
root := e.store.Settings().DownloadRoot
|
||||
if root == "" {
|
||||
root = DefaultDownloadRoot()
|
||||
}
|
||||
|
||||
// 先看磁盘上有没有这个画廊的目录和清单。
|
||||
// 4KHD 的图片直链是稳定的，存清单后**0 次请求**就能开始下载。
|
||||
existing := e.findExistingGallery(root, task)
|
||||
|
||||
var gallery *hd4k.Gallery
|
||||
usedManifest := false
|
||||
|
||||
if existing != nil {
|
||||
if g, ok := existing.manifest.GalleryFromManifest(); ok {
|
||||
gallery = g
|
||||
usedManifest = true
|
||||
e.log("info", fmt.Sprintf(
|
||||
"画廊 %s 用本地清单（%d 张），跳过重新解析", task.Slug, len(g.Images)))
|
||||
} else if existing.manifest != nil {
|
||||
e.log("info", fmt.Sprintf(
|
||||
"画廊 %s 有目录但清单不含图片直链，仍需重新解析", task.Slug))
|
||||
}
|
||||
}
|
||||
|
||||
if gallery == nil {
|
||||
g, err := e.client.FetchGallery(ctx, refFromTask(task),
|
||||
func(done, total, page int) {
|
||||
e.reportParseProgress(task, done, total, page)
|
||||
})
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
e.markCanceled(task)
|
||||
return
|
||||
}
|
||||
e.fail(task, fmt.Errorf("解析画廊页失败: %w", err))
|
||||
return
|
||||
}
|
||||
gallery = g
|
||||
}
|
||||
|
||||
if gallery.Title != "" {
|
||||
task.Title = gallery.Title
|
||||
}
|
||||
task.Total = gallery.Total
|
||||
task.Dir = BuildDir(e.store.Settings().DirTemplate, gallery)
|
||||
|
||||
task.Status = "downloading"
|
||||
e.save(task)
|
||||
|
||||
dir := filepath.Join(root, task.Dir)
|
||||
|
||||
// 用户中途换了模板时，先看老的「标题目录」在不在 —— 在就继续往那儿下，
|
||||
// 否则换个模板就会把几千张图全部重下一遍。
|
||||
legacyDir := filepath.Join(root, SanitizeSegment(gallery.Title, task.Slug))
|
||||
if dir != legacyDir {
|
||||
if _, err := os.Stat(legacyDir); err == nil {
|
||||
e.log("info", "沿用旧目录（模板改过，已有图在老位置）："+gallery.Title)
|
||||
dir = legacyDir
|
||||
if rel, err := filepath.Rel(root, dir); err == nil {
|
||||
task.Dir = rel
|
||||
}
|
||||
}
|
||||
}
|
||||
// 清单里的目录名对不上时，以磁盘上实际存在的为准
|
||||
if existing != nil && existing.dir != "" && existing.dir != dir {
|
||||
dir = existing.dir
|
||||
if rel, err := filepath.Rel(root, dir); err == nil {
|
||||
task.Dir = rel
|
||||
}
|
||||
}
|
||||
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
e.fail(task, fmt.Errorf("创建目录失败: %w", err))
|
||||
return
|
||||
}
|
||||
e.save(task)
|
||||
|
||||
st := &pageState{
|
||||
task: &task,
|
||||
dir: dir,
|
||||
id: task.ID,
|
||||
host: task.Host,
|
||||
title: gallery.Title,
|
||||
total: len(gallery.Images),
|
||||
images: gallery.Images,
|
||||
meta: gallery.Meta,
|
||||
engine: e,
|
||||
}
|
||||
|
||||
// 先快速扫描：已存在的直接算完成，剩下的才进队列
|
||||
st.scanExisting()
|
||||
st.saveManifest()
|
||||
|
||||
runImageWorkers(ctx, e, st, st.pending)
|
||||
|
||||
// ---- 整轮扫尾：把刚失败的图再扫一遍 ----
|
||||
if st.failedCount() > 0 && ctx.Err() == nil {
|
||||
e.log("info", fmt.Sprintf("画廊 %s：%d 张首轮失败，开始扫尾重试", task.Slug, st.failedCount()))
|
||||
|
||||
retry := st.collectFailed()
|
||||
st.clearFailures()
|
||||
runImageWorkers(ctx, e, st, retry)
|
||||
|
||||
if st.failedCount() > 0 {
|
||||
e.log("warn", fmt.Sprintf(
|
||||
"画廊 %s：扫尾后仍有 %d 张失败（换个线路再点「开始下载」可续传）",
|
||||
task.Slug, st.failedCount()))
|
||||
} else {
|
||||
e.log("ok", fmt.Sprintf("画廊 %s：扫尾重试全部成功", task.Slug))
|
||||
}
|
||||
}
|
||||
|
||||
// ---- 清单自愈 ----
|
||||
// 作者可能给画廊补了新图。全部待下图都失败时，重新解析一次拿最新列表。
|
||||
if usedManifest && ctx.Err() == nil && st.allFailed() {
|
||||
e.log("warn", fmt.Sprintf(
|
||||
"画廊 %s：本地清单的直链似乎已失效，重新解析画廊页", task.Slug))
|
||||
|
||||
if fresh, err := e.client.FetchGallery(ctx, refFromTask(task),
|
||||
func(done, total, page int) { e.reportParseProgress(task, done, total, page) }); err == nil {
|
||||
st.reset(fresh)
|
||||
st.scanExisting()
|
||||
st.saveManifest()
|
||||
|
||||
runImageWorkers(ctx, e, st, st.collectAllPending())
|
||||
|
||||
if st.failedCount() == 0 {
|
||||
e.log("ok", fmt.Sprintf("画廊 %s：重新解析后全部下载完成", task.Slug))
|
||||
}
|
||||
} else if ctx.Err() == nil {
|
||||
e.log("error", fmt.Sprintf("画廊 %s：重新解析失败：%v", task.Slug, err))
|
||||
}
|
||||
}
|
||||
|
||||
st.saveManifest()
|
||||
|
||||
if ctx.Err() != nil {
|
||||
e.markCanceled(task)
|
||||
return
|
||||
}
|
||||
|
||||
t := st.snapshot()
|
||||
if t.Failed > 0 {
|
||||
t.Status = "partial"
|
||||
} else {
|
||||
t.Status = "done"
|
||||
}
|
||||
e.save(t)
|
||||
}
|
||||
|
||||
// runImageWorkers 用并发下载给定图片（阻塞直到完成）
|
||||
func runImageWorkers(ctx context.Context, e *Engine, st *pageState, imgs []hd4k.ImageInfo) {
|
||||
if len(imgs) == 0 {
|
||||
return
|
||||
}
|
||||
n := e.PageConcurrency()
|
||||
if n > len(imgs) {
|
||||
n = len(imgs)
|
||||
}
|
||||
|
||||
work := make(chan hd4k.ImageInfo, len(imgs))
|
||||
for _, im := range imgs {
|
||||
work <- im
|
||||
}
|
||||
close(work)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < n; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for im := range work {
|
||||
st.safeDownloadImage(ctx, im)
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// pageState 是「一个画廊的图片级下载状态」，所有字段都由 mutex 保护，
|
||||
// 因为图片级 worker 是并发的。
|
||||
type pageState struct {
|
||||
mu sync.Mutex
|
||||
engine *Engine
|
||||
|
||||
task *Task
|
||||
dir string
|
||||
id string
|
||||
host string
|
||||
title string
|
||||
total int
|
||||
|
||||
images []hd4k.ImageInfo
|
||||
pending []hd4k.ImageInfo
|
||||
meta hd4k.GalleryMeta
|
||||
|
||||
done map[int]bool
|
||||
failed map[int]string
|
||||
failedL []FailedItem
|
||||
}
|
||||
|
||||
func (s *pageState) reset(g *hd4k.Gallery) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.images = g.Images
|
||||
s.pending = nil
|
||||
s.total = len(g.Images)
|
||||
s.title = g.Title
|
||||
s.meta = g.Meta
|
||||
}
|
||||
|
||||
// allFailed 判断是否「有待下图且全部失败」
|
||||
func (s *pageState) allFailed() bool {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if len(s.pending) == 0 {
|
||||
return false
|
||||
}
|
||||
if len(s.pending) == 1 && len(s.done) > 0 {
|
||||
return false
|
||||
}
|
||||
return len(s.done) == 0 && len(s.failed) > 0
|
||||
}
|
||||
|
||||
func (s *pageState) collectAllPending() []hd4k.ImageInfo {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
var out []hd4k.ImageInfo
|
||||
for _, im := range s.images {
|
||||
if !s.done[im.Index] {
|
||||
out = append(out, im)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// scanExisting 扫描已有文件（断点续传第一依据：文件在且非空 = 完成）
|
||||
func (s *pageState) scanExisting() {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
s.done = map[int]bool{}
|
||||
s.failed = map[int]string{}
|
||||
s.failedL = nil
|
||||
s.pending = nil
|
||||
|
||||
if s.task.Pages == nil {
|
||||
s.task.Pages = map[string]string{}
|
||||
}
|
||||
|
||||
for _, im := range s.images {
|
||||
fname := targetName(im, s.id, s.total)
|
||||
dest := filepath.Join(s.dir, fname)
|
||||
if st, err := os.Stat(dest); err == nil && st.Size() > 0 {
|
||||
s.done[im.Index] = true
|
||||
s.task.Pages[strconv.Itoa(im.Index)] = "done"
|
||||
} else {
|
||||
s.pending = append(s.pending, im)
|
||||
}
|
||||
}
|
||||
s.task.Done = len(s.done)
|
||||
}
|
||||
|
||||
// safeDownloadImage 包一层 recover：单张图的意外 panic 不该带走整个 worker。
|
||||
func (s *pageState) safeDownloadImage(ctx context.Context, im hd4k.ImageInfo) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
fname := targetName(im, s.id, s.total)
|
||||
s.mu.Lock()
|
||||
s.failed[im.Index] = fmt.Sprintf("内部错误: %v", r)
|
||||
s.failedL = append(s.failedL, FailedItem{
|
||||
Page: im.Index, File: fname, Error: fmt.Sprintf("内部错误: %v", r),
|
||||
})
|
||||
s.task.Pages[strconv.Itoa(im.Index)] = "failed"
|
||||
s.syncCountsLocked()
|
||||
snap := s.snapshotLocked()
|
||||
s.mu.Unlock()
|
||||
s.engine.emitTask(snap)
|
||||
}
|
||||
}()
|
||||
s.downloadImage(ctx, im)
|
||||
}
|
||||
|
||||
// downloadImage 下载单张图，内部含 4 次退避重试
|
||||
func (s *pageState) downloadImage(ctx context.Context, im hd4k.ImageInfo) {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
fname := targetName(im, s.id, s.total)
|
||||
dest := filepath.Join(s.dir, fname)
|
||||
// Referer 取画廊自身的 origin（用户可能配了 http 或带端口的站点，
|
||||
// 硬拼 "https://"+host 会对不上）
|
||||
referer := hd4k.OriginOf(s.task.URL) + "/"
|
||||
|
||||
err := s.engine.withRetry(ctx, func() error {
|
||||
_, e := s.engine.client.DownloadImage(ctx, im.URL, dest, referer, nil)
|
||||
return e
|
||||
})
|
||||
|
||||
s.mu.Lock()
|
||||
k := strconv.Itoa(im.Index)
|
||||
if err != nil {
|
||||
if ctx.Err() != nil || errors.Is(err, ErrCanceled) {
|
||||
s.mu.Unlock()
|
||||
s.markCanceled()
|
||||
return
|
||||
}
|
||||
s.failed[im.Index] = err.Error()
|
||||
s.failedL = append(s.failedL, FailedItem{Page: im.Index, File: fname, Error: err.Error()})
|
||||
s.task.Pages[k] = "failed"
|
||||
} else {
|
||||
s.done[im.Index] = true
|
||||
s.task.Pages[k] = "done"
|
||||
}
|
||||
s.syncCountsLocked()
|
||||
snap := s.snapshotLocked()
|
||||
s.mu.Unlock()
|
||||
|
||||
s.engine.emitTask(snap)
|
||||
}
|
||||
|
||||
func (s *pageState) syncCountsLocked() {
|
||||
s.task.Done = len(s.done)
|
||||
s.task.Failed = len(s.failed)
|
||||
s.task.FailedList = append([]FailedItem(nil), s.failedL...)
|
||||
}
|
||||
|
||||
// snapshotLocked 生成任务快照。
|
||||
//
|
||||
// ⚠️ 必须深拷贝 Pages / FailedList —— map 和 slice 浅拷贝会把底层内存
|
||||
// 一起交出去，而快照是要送到 UI 做 JSON 序列化的，其他 worker 此刻还在写
|
||||
// 同一份数据 → fatal error: concurrent map read and map write（进程闪退）。
|
||||
func (s *pageState) snapshotLocked() Task {
|
||||
s.syncCountsLocked()
|
||||
t := *s.task
|
||||
|
||||
pages := make(map[string]string, len(s.task.Pages))
|
||||
for k, v := range s.task.Pages {
|
||||
pages[k] = v
|
||||
}
|
||||
t.Pages = pages
|
||||
t.FailedList = append([]FailedItem(nil), s.task.FailedList...)
|
||||
|
||||
return t
|
||||
}
|
||||
|
||||
func (s *pageState) snapshot() Task {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.snapshotLocked()
|
||||
}
|
||||
|
||||
func (s *pageState) failedCount() int {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return len(s.failed)
|
||||
}
|
||||
|
||||
func (s *pageState) clearFailures() {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.failed = map[int]string{}
|
||||
s.failedL = nil
|
||||
s.task.Failed = 0
|
||||
s.task.FailedList = []FailedItem{}
|
||||
}
|
||||
|
||||
func (s *pageState) collectFailed() []hd4k.ImageInfo {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
failed := intSet(keysOf(s.failed))
|
||||
var out []hd4k.ImageInfo
|
||||
for _, im := range s.images {
|
||||
if failed[im.Index] {
|
||||
out = append(out, im)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (s *pageState) markCanceled() {
|
||||
s.engine.markCanceled(s.snapshot())
|
||||
}
|
||||
|
||||
// saveManifest 写画廊目录清单。
|
||||
//
|
||||
// 4KHD 的图片直链是 Jetpack Photon 的哈希路径，完全稳定，存下来随时能用 ——
|
||||
// 下次重开程序，读这份清单就能 0 请求直接下载。
|
||||
func (s *pageState) saveManifest() {
|
||||
s.mu.Lock()
|
||||
done := keysOf(s.done)
|
||||
failed := keysOf(s.failed)
|
||||
t := *s.task
|
||||
|
||||
meta := s.meta
|
||||
pageMap := make(map[string]ManifestPage, len(s.images))
|
||||
for _, im := range s.images {
|
||||
k := strconv.Itoa(im.Index)
|
||||
errMsg := ""
|
||||
if msg, bad := s.failed[im.Index]; bad {
|
||||
errMsg = msg
|
||||
}
|
||||
pageMap[k] = ManifestPage{
|
||||
Page: im.Index,
|
||||
File: targetName(im, s.id, s.total),
|
||||
URL: im.URL,
|
||||
Finished: s.done[im.Index],
|
||||
Error: errMsg,
|
||||
}
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
WriteManifest(s.dir, Manifest{
|
||||
ID: s.id,
|
||||
URL: t.URL,
|
||||
Host: s.host,
|
||||
Path: t.Path,
|
||||
Slug: t.Slug,
|
||||
Title: s.title,
|
||||
TotalPages: 0,
|
||||
Total: s.total,
|
||||
DonePages: done,
|
||||
FailedPage: failed,
|
||||
PageMap: pageMap,
|
||||
Meta: &meta,
|
||||
CreatedAt: t.CreatedAt,
|
||||
})
|
||||
}
|
||||
|
||||
func keysOf[V any](m map[int]V) []int {
|
||||
out := make([]int, 0, len(m))
|
||||
for k := range m {
|
||||
out = append(out, k)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// targetName 图片文件名。
|
||||
//
|
||||
// 4KHD 的图片本身带原名和序号（tbox-vol52-4khd.com-001.webp），
|
||||
// 优先用它 —— 可读且天然有序。只有原名缺失时才用页码补零兜底。
|
||||
func targetName(im hd4k.ImageInfo, id string, total int) string {
|
||||
if im.Filename != "" {
|
||||
return SanitizeFilename(im.Filename, fallbackName(im, id, total))
|
||||
}
|
||||
return fallbackName(im, id, total)
|
||||
}
|
||||
|
||||
func fallbackName(im hd4k.ImageInfo, id string, total int) string {
|
||||
width := len(strconv.Itoa(total))
|
||||
n := strconv.Itoa(im.Index)
|
||||
for len(n) < width {
|
||||
n = "0" + n
|
||||
}
|
||||
short := id
|
||||
if i := strings.LastIndex(short, "/"); i >= 0 {
|
||||
short = short[i+1:]
|
||||
}
|
||||
if j := strings.Index(short, "."); j > 0 {
|
||||
short = short[:j]
|
||||
}
|
||||
if short == "" {
|
||||
short = "image"
|
||||
}
|
||||
return SanitizeFilename(short+"-"+n+".webp", n+".webp")
|
||||
}
|
||||
|
||||
// withRetry 指数退避重试
|
||||
//
|
||||
// ⚠️ 限流（429/503/403）走另一条路：触发**全局冷却**，让所有 worker
|
||||
// 一起停，而不是各自退避。原因见 Limiter.Cooldown 的注释 ——
|
||||
// 40 个 worker 错峰重试会持续往已限流的服务器打请求，永远缓不过来。
|
||||
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
|
||||
}
|
||||
|
||||
// 限流：全局冷却 + 更长退避，不再各退各的
|
||||
if hd4k.IsRateLimited(err) {
|
||||
wait := e.rateLimitBackoff(err, i)
|
||||
e.log("warn", fmt.Sprintf("触发限流，全部 worker 冷却 %s（第 %d 次）：%v",
|
||||
wait.Round(time.Second), i+1, err))
|
||||
e.client.Cooldown(wait)
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ErrCanceled
|
||||
case <-time.After(wait):
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
// rateLimitBackoff 算限流后的等待时长。
|
||||
//
|
||||
// 优先用服务器给的 Retry-After（它最清楚该等多久），否则按尝试次数指数退避。
|
||||
// 下限 10s 是因为实测 CDN 的限流窗口通常以十几秒计，退避太短只会一直撞墙。
|
||||
func (e *Engine) rateLimitBackoff(err error, attempt int) time.Duration {
|
||||
var se *hd4k.StatusError
|
||||
if errors.As(err, &se) && se.RetryAfter > 0 {
|
||||
// 服务器明说了，就按它的来（再加一点余量，避免卡着点又撞上）
|
||||
d := se.RetryAfter + time.Duration(rand.Intn(2000))*time.Millisecond
|
||||
if d > 10*time.Minute {
|
||||
d = 10 * time.Minute
|
||||
}
|
||||
return d
|
||||
}
|
||||
wait := time.Duration(10<<uint(attempt)) * time.Second // 10s → 20s → 40s → 80s
|
||||
if wait > 5*time.Minute {
|
||||
wait = 5 * time.Minute
|
||||
}
|
||||
return wait + time.Duration(rand.Intn(2000))*time.Millisecond
|
||||
}
|
||||
|
||||
// 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.Slug))
|
||||
}
|
||||
|
||||
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.Slug, msg))
|
||||
}
|
||||
|
||||
func (e *Engine) save(task Task) {
|
||||
task = e.store.UpsertTask(task)
|
||||
e.emitTask(task)
|
||||
}
|
||||
|
||||
func (e *Engine) log(level, msg string) {
|
||||
if e.sink != nil {
|
||||
e.sink.Log(level, msg)
|
||||
}
|
||||
}
|
||||
|
||||
var reSpaced = regexp.MustCompile(`\s+`)
|
||||
|
||||
func cleanSpaces(s string) string { return reSpaced.ReplaceAllString(s, " ") }
|
||||
Reference in New Issue
Block a user