Files
eh-downloader/app.go
T
toor 0447875b6d feat: 双层并发 + 全局限流 + 整轮扫尾 + 画廊目录清单
原实现只能「多画廊并行,每个画廊串行下」,单张大画廊下得很慢。
按用户多年使用习惯(5 画廊 × 每画廊 8 张)改造为双层并发。

- 双层并发:画廊并发(默认5) × 页并发(默认8) = 最多 40 张同时下载,
  两个值都在设置面板可调(1-20)。页级 worker 池 + mutex 保护计数。
- 全局限流:对齐参考实现的 nextRequestTime 算法,限所有请求
  (含 /s/ 解析与图片本体),默认 0 秒关闭。
- 整轮扫尾:跑完后自动重扫失败页 1 轮(同样走完整 4 次退避重试),
  仍失败则停下并标记 partial,换线路后手动点开始即可续传。
- 画廊目录清单 .ehdl.json:记录 gid/token/总页数/已完成页,
  即使 state.json 丢失也能算出「应该多少张、还差多少张」。

顺带修复:
- Start() 双重加锁导致死锁(引擎完全无法启动)
- clampInt 越界时返回默认值而非边界值(设 30 会静默变 8)
- 直链正则写死 hath.network,CDN 换域名会全军覆没 -> 加图片 URL 兜底
- ReconcileTask 会误报「有改动」

测试从 37 增至 62,其中并发/限流/扫尾用 httptest 模拟 EH 实测:
- 页并发峰值 = 8、画廊并发峰值 = 5(完全生效)
- 12 次请求限流 80ms 实测 966ms(期望 880ms)
- 断点续传 4 张已存在时只发 5 次请求
2026-10-09 00:11:38 +08:00

524 lines
14 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package main
// Wails 绑定层:把 Go 侧能力暴露给前端。
// 前端通过 window.go.main.App.XXX() 调用(见 frontend/js/api.js)。
import (
"context"
"fmt"
"os"
"path/filepath"
"time"
wruntime "github.com/wailsapp/wails/v2/pkg/runtime"
"eh-downloader-wails/internal/core"
"eh-downloader-wails/internal/eh"
)
// App 是 Wails 应用主体
type App struct {
ctx context.Context
store *core.Store
engine *core.Engine
bridge *core.Bridge
// 事件序号(前端用来丢弃旧日志)
logSeq int64
logMu chan struct{}
engineCtx context.Context
engineStop context.CancelFunc
}
// LogEntry 是一条日志
type LogEntry struct {
Seq int64 `json:"seq"`
Level string `json:"level"`
Msg string `json:"msg"`
At int64 `json:"at"`
}
// AddTasksResult 导入结果
type AddTasksResult struct {
Added int `json:"added"`
Skipped int `json:"skipped"`
Warning string `json:"warning"`
}
// StartResult 开始下载的结果
type StartResult struct {
Started bool `json:"started"`
Reason string `json:"reason"`
Count int `json:"count"`
}
// StatusResult 引擎状态
type StatusResult struct {
Running bool `json:"running"`
Queued int `json:"queued"`
}
// DirPreviewResult 目录预览
type DirPreviewResult struct {
OK bool `json:"ok"`
Path string `json:"path"`
Error string `json:"error"`
}
// statePath 状态文件路径(放用户配置目录)
func statePath() string {
dir, err := os.UserConfigDir()
if err != nil {
dir = "."
}
return filepath.Join(dir, core.AppName, "state.json")
}
// NewApp 建应用
func NewApp() *App {
return &App{logMu: make(chan struct{}, 1)}
}
// ---------------------------------------------------------------- 生命周期
// startup Wails 启动钩子
func (a *App) startup(ctx context.Context) {
a.ctx = ctx
a.store = core.NewStore(statePath())
s := a.store.Settings()
if s.DownloadRoot == "" {
s.DownloadRoot = core.DefaultDownloadRoot()
a.store.UpdateSettings(map[string]interface{}{"downloadRoot": s.DownloadRoot})
}
// 首次启动:EH 直连基本不通,自动探测本机常见代理端口
if s.ProxyMode == "off" && s.Proxy == "" && !s.Probed {
found := core.DetectLocalProxy([]int{7890, 7897, 1087, 1080, 8080, 10808, 10809, 8888})
a.store.UpdateSettings(map[string]interface{}{"_probed": true})
if found != "" {
a.store.UpdateSettings(map[string]interface{}{"proxyMode": "http", "proxy": found})
a.emitLog("info", "自动检测到本机代理 "+found+",已填入设置")
} else {
a.emitLog("warn", "没探测到本机代理,EH 在国内通常需要代理,请在设置里填")
}
a.store.Flush()
}
if n := a.store.RecoverOnStart(); n > 0 {
a.emitLog("info", fmt.Sprintf("启动时归位了 %d 个中断的任务", n))
}
a.engine = core.NewEngine(a.store, &sink{a}, s)
a.engineCtx, a.engineStop = context.WithCancel(context.Background())
a.bridge = core.NewBridge(
a.store,
a.emitLog,
func(t core.Task, added bool, batch bool) {
a.emitTaskUpdate(t)
how := "已在列表中"
if added {
how = "已加入待下载"
}
title := ""
if t.Title != "" {
title = " · " + t.Title
}
a.emitLog("ok", fmt.Sprintf("油猴推送:%s %s%s", t.GID, how, title))
// 窗口最小化/隐藏时唤起一下,让用户看得到
wruntime.WindowUnminimise(a.ctx)
wruntime.WindowShow(a.ctx)
},
)
a.bridge.Start(s.BridgePort)
}
// shutdown Wails 退出钩子
func (a *App) shutdown(ctx context.Context) {
if a.engineStop != nil {
a.engineStop()
}
if a.bridge != nil {
a.bridge.Stop()
}
if a.store != nil {
a.store.Stop()
a.store.Flush()
}
}
// sink 把引擎事件转成 Wails 事件
type sink struct{ a *App }
func (s *sink) TaskUpdate(t core.Task) { s.a.emitTaskUpdate(t) }
func (s *sink) Log(level, msg string) { s.a.emitLog(level, msg) }
func (s *sink) Done() { s.a.emitEvent("engine:done", map[string]interface{}{}) }
// ---------------------------------------------------------------- 事件
func (a *App) emitEvent(name string, data interface{}) {
if a.ctx == nil {
return
}
wruntime.EventsEmit(a.ctx, name, data)
}
func (a *App) emitLog(level, msg string) {
// 串行化自增,避免并发下序号乱序
a.logMu <- struct{}{}
a.logSeq++
entry := LogEntry{Seq: a.logSeq, Level: level, Msg: msg, At: time.Now().UnixMilli()}
<-a.logMu
a.emitEvent("log", entry)
}
func (a *App) emitTaskUpdate(t core.Task) {
a.emitEvent("task:update", t)
}
// ---------------------------------------------------------------- 设置
// GetSettings 取设置
func (a *App) GetSettings() core.Settings {
return a.store.Settings()
}
// SetSettings 更新设置
func (a *App) SetSettings(patch map[string]interface{}) core.Settings {
next := a.store.UpdateSettings(patch)
// 同步到引擎
if _, ok := patch["galleryConcurrency"]; ok {
a.engine.SetGalleryConcurrency(next.GalleryConcurrency)
}
if _, ok := patch["pageConcurrency"]; ok {
a.engine.SetPageConcurrency(next.PageConcurrency)
}
if _, ok := patch["requestIntervalMs"]; ok {
a.engine.SetRequestInterval(next.RequestIntervalMs)
}
if _, ok := patch["timeout"]; ok {
a.engine.Client().SetTimeout(time.Duration(next.Timeout) * time.Millisecond)
}
if _, ok := patch["xres"]; ok {
a.engine.Client().SetXres(next.Xres)
}
if _, ok := patch["proxy"]; ok {
a.engine.Client().SetProxy(next.Proxy, next.ProxyMode)
}
if _, ok := patch["proxyMode"]; ok {
a.engine.Client().SetProxy(next.Proxy, next.ProxyMode)
}
// 端口变了就重启桥口
if _, ok := patch["bridgePort"]; ok && a.bridge.Port() != next.BridgePort {
a.bridge.Stop()
a.bridge.Start(next.BridgePort)
}
return next
}
// PreviewDir 目录模板实时预览
func (a *App) PreviewDir(tpl string) DirPreviewResult {
g := previewGallery()
return DirPreviewResult{OK: true, Path: core.BuildDir(tpl, g)}
}
func previewGallery() *eh.Gallery {
return &eh.Gallery{
GID: "4239274",
Title: "[Noxe] 0820 Qingxiao [AI Generated]",
Meta: eh.GalleryMeta{
Artist: "Noxe",
Group: "CriticalHAI",
Series: "wuthering waves",
Category: "misc",
Uploader: "velius11",
Tags: []string{"qingxiao", "rover", "big breasts", "x-ray", "ai generated"},
},
}
}
// Placeholders 可用占位符
func (a *App) Placeholders() []string {
return core.ListPlaceholders()
}
// EngineInfo 引擎当前的并发配置(前端用来显示"总上限 = 画廊 × 页")
type EngineInfo struct {
Running bool `json:"running"`
Queued int `json:"queued"`
GalleryConcurrency int `json:"galleryConcurrency"`
PageConcurrency int `json:"pageConcurrency"`
MaxConcurrent int `json:"maxConcurrent"`
RequestIntervalMs int `json:"requestIntervalMs"`
}
// EngineInfo 取引擎状态和并发配置
func (a *App) EngineInfo() EngineInfo {
g := a.engine.GalleryConcurrency()
p := a.engine.PageConcurrency()
return EngineInfo{
Running: a.engine.IsRunning(),
Queued: a.engine.QueueLength(),
GalleryConcurrency: g,
PageConcurrency: p,
MaxConcurrent: g * p,
RequestIntervalMs: int(a.engine.Client().RequestInterval().Milliseconds()),
}
}
// ScanManifestsResult 扫描下载目录的结果
type ScanManifestsResult struct {
Found int `json:"found"`
Adopted int `json:"adopted"`
Root string `json:"root"`
Note string `json:"note"`
}
// ScanDownloadDir 扫描下载目录里的画廊清单,把 state.json 里缺的信息补上。
// 场景:用户删了 state.json,靠清单重建"这个画廊应该有多少张"的认知。
func (a *App) ScanDownloadDir() ScanManifestsResult {
root := a.store.Settings().DownloadRoot
if root == "" {
root = core.DefaultDownloadRoot()
}
manifests := core.ScanManifests(root)
res := ScanManifestsResult{Root: root, Found: len(manifests)}
if len(manifests) == 0 {
res.Note = "下载目录里没找到画廊清单(.ehdl.json)"
return res
}
for _, m := range manifests {
// 找到对应任务(按 gid 匹配,token 可能有差异所以只按 gid)
t := a.store.FindTaskByGID(m.GID)
if t == nil {
continue
}
if core.ReconcileTask(t, m) {
a.store.UpsertTask(*t)
res.Adopted++
}
}
res.Note = fmt.Sprintf("扫描到 %d 个画廊清单,补全了 %d 个任务", res.Found, res.Adopted)
return res
}
// ---------------------------------------------------------------- 任务
// ListTasks 全部任务
func (a *App) ListTasks() []core.Task {
return a.store.Tasks()
}
// AddTasks 导入链接文本
func (a *App) AddTasks(text string) AddTasksResult {
parsed := eh.ParseGalleryInput(text)
if len(parsed) == 0 {
for _, g := range eh.ExtractGIDs(text) {
parsed = append(parsed, eh.GalleryRef{GID: g, Raw: g})
}
}
if len(parsed) == 0 {
return AddTasksResult{
Warning: "没解析出任何画廊链接。支持 https://e-hentai.org/g/{gid}/{token}/ 格式或裸 gid。",
}
}
noToken := 0
for _, p := range parsed {
if p.Token == "" {
noToken++
}
}
added := a.store.AddTasks(parsed)
warning := ""
if noToken > 0 {
warning = fmt.Sprintf("%d 个只有 gid 没有 token,可能取不到画廊(EH 需要完整链接)", noToken)
}
a.emitLog("info", fmt.Sprintf("导入 %d 个画廊,新增 %d 个", len(parsed), added))
return AddTasksResult{Added: added, Skipped: len(parsed) - added, Warning: warning}
}
// RemoveTask 删单个
func (a *App) RemoveTask(gid, token string) []core.Task {
a.store.RemoveTask(gid, token)
return a.store.Tasks()
}
// ClearFinished 清除已完成(不动本地文件)
func (a *App) ClearFinished() map[string]interface{} {
n := a.store.ClearFinished()
a.emitLog("info", fmt.Sprintf("清除了 %d 个已完成任务(本地已下载文件未动)", n))
return map[string]interface{}{"removed": n, "tasks": a.store.Tasks()}
}
// CountDone 已完成数
func (a *App) CountDone() int { return a.store.CountDone() }
// ClearAll 清空任务列表(不动本地文件)
func (a *App) ClearAll(onlySelected bool) map[string]interface{} {
scope := "全部"
if onlySelected {
scope = "选中的"
}
n := a.store.ClearAll(onlySelected)
if n > 0 {
a.emitLog("warn", fmt.Sprintf("删除了%s %d 个任务(本地已下载文件未动)", scope, n))
} else {
a.emitLog("info", "没有可删除的任务")
}
return map[string]interface{}{"removed": n, "tasks": a.store.Tasks()}
}
// RequeueFailed 失败重排
func (a *App) RequeueFailed() []core.Task {
n := a.store.RequeueFailed()
a.emitLog("info", fmt.Sprintf("重新排队 %d 个任务", n))
return a.store.Tasks()
}
// SelectAll 标记勾选
func (a *App) SelectAll(gids []string) []core.Task {
return a.store.SelectAll(gids)
}
// ---------------------------------------------------------------- 引擎
// Status 引擎状态
func (a *App) Status() StatusResult {
return StatusResult{Running: a.engine.IsRunning(), Queued: a.engine.QueueLength()}
}
// Start 开始下载
func (a *App) Start(onlyGids []string) StartResult {
if a.engine.IsRunning() {
return StartResult{Started: false, Reason: "已经在跑了"}
}
all := a.store.Tasks()
targets := make([]core.Task, 0, len(all))
withToken := 0
for _, t := range all {
switch t.Status {
case "queued", "failed", "partial":
default:
continue
}
if len(onlyGids) > 0 && !containsStr(onlyGids, t.GID) {
continue
}
if t.Token != "" {
withToken++
}
targets = append(targets, t)
}
if len(targets) == 0 {
return StartResult{Started: false, Reason: "没有待处理的任务"}
}
if withToken == 0 {
return StartResult{Started: false, Reason: "任务里没有带 token 的画廊,无法开始"}
}
a.engine.Enqueue(targets)
ctx := a.engineCtx
if ctx == nil {
a.engineCtx, a.engineStop = context.WithCancel(context.Background())
ctx = a.engineCtx
}
go a.engine.Start(ctx)
g := a.engine.GalleryConcurrency()
p := a.engine.PageConcurrency()
iv := a.engine.Client().RequestInterval().Milliseconds()
msg := fmt.Sprintf("开始下载 %d 个画廊(画廊并发 %d × 页并发 %d,最多 %d 张同时下载)",
len(targets), g, p, g*p)
if iv > 0 {
msg += fmt.Sprintf(",请求间隔 %dms", iv)
}
a.emitLog("info", msg)
return StartResult{Started: true, Count: len(targets)}
}
// Stop 停止下载
func (a *App) Stop() map[string]interface{} {
a.engine.Stop()
a.emitLog("warn", "已请求停止,等待当前请求结束")
return map[string]interface{}{"stopped": true}
}
// ---------------------------------------------------------------- 系统
// PickDir 选目录
func (a *App) PickDir() string {
p, _ := wruntime.OpenDirectoryDialog(a.ctx, wruntime.OpenDialogOptions{
Title: "选择下载目录",
})
return p
}
// OpenPath 打开路径
func (a *App) OpenPath(p string) {
wruntime.BrowserOpenURL(a.ctx, "file:///"+filepath.ToSlash(p))
}
// ShowItem 在资源管理器里定位
func (a *App) ShowItem(p string) {
wruntime.BrowserOpenURL(a.ctx, "file:///"+filepath.ToSlash(filepath.Dir(p)))
}
// ExportFailed 导出失败清单
func (a *App) ExportFailed(gid string) string {
var target *core.Task
for _, x := range a.store.Tasks() {
if x.GID == gid {
c := x
target = &c
break
}
}
if target == nil || len(target.FailedList) == 0 {
return ""
}
path, _ := wruntime.SaveFileDialog(a.ctx, wruntime.SaveDialogOptions{
Title: "导出失败清单",
DefaultFilename: gid + "-failed.txt",
Filters: []wruntime.FileFilter{
{DisplayName: "文本文件 (*.txt)", Pattern: "*.txt"},
},
})
if path == "" {
return ""
}
var sb string
for _, f := range target.FailedList {
sb += fmt.Sprintf("# page %d %s\t%s\n", f.Page, f.File, f.Error)
}
if err := os.WriteFile(path, []byte(sb), 0o644); err != nil {
a.emitLog("error", "导出失败:"+err.Error())
return ""
}
return path
}
func containsStr(list []string, s string) bool {
for _, x := range list {
if x == s {
return true
}
}
return false
}