GithubBackup/SaveAny-Bot
1
0
Fork 0
mirror of https://github.com/krau/SaveAny-Bot.git synced 2026-05-12 02:39:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: configurable parser and refactor config

Browse Source
This commit is contained in:
krau
2025-08-23 14:29:32 +08:00
parent 03eb4f8a18
commit e5d1e143e0
28 changed files with 181 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
core/core.go
View File

@@ -20,7 +20,7 @@ type Exectable interface {
func worker(ctx context.Context, qe *queue.TaskQueue[Exectable], semaphore chan struct{}) {
logger := log.FromContext(ctx)
execHooks := config.Cfg.Hook.Exec
execHooks := config.C().Hook.Exec
for {
semaphore <- struct{}{}
qtask, err := qe.Get()
@@ -58,11 +58,11 @@ func worker(ctx context.Context, qe *queue.TaskQueue[Exectable], semaphore chan
func Run(ctx context.Context) {
log.FromContext(ctx).Info("Start processing tasks...")
semaphore := make(chan struct{}, config.Cfg.Workers)
semaphore := make(chan struct{}, config.C().Workers)
if queueInstance == nil {
queueInstance = queue.NewTaskQueue[Exectable]()
}
for range config.Cfg.Workers {
for range config.C().Workers {
go worker(ctx, queueInstance, semaphore)
}

4
core/tasks/batchtfile/execute.go
View File

@@ -21,7 +21,7 @@ func (t *Task) Execute(ctx context.Context) error {
logger := log.FromContext(ctx).WithPrefix(fmt.Sprintf("batch_file[%s]", t.ID))
logger.Info("Starting batch file task")
t.Progress.OnStart(ctx, t)
workers := config.Cfg.Workers
workers := config.C().Workers
eg, gctx := errgroup.WithContext(ctx)
eg.SetLimit(workers)
for _, elem := range t.Elems {
@@ -124,6 +124,6 @@ func (t *Task) processElement(ctx context.Context, elem TaskElement) error {
return err
}
return nil
}, retry.Context(vctx), retry.RetryTimes(uint(config.Cfg.Retry)))
}, retry.Context(vctx), retry.RetryTimes(uint(config.C().Retry)))
return err
}

4
core/tasks/batchtfile/task.go
View File

@@ -47,8 +47,8 @@ func NewTaskElement(
) (*TaskElement, error) {
id := xid.New().String()
_, ok := stor.(storage.StorageCannotStream)
if !config.Cfg.Stream || ok {
cachePath, err := filepath.Abs(filepath.Join(config.Cfg.Temp.BasePath, fmt.Sprintf("%s_%s", id, file.Name())))
if !config.C().Stream || ok {
cachePath, err := filepath.Abs(filepath.Join(config.C().Temp.BasePath, fmt.Sprintf("%s_%s", id, file.Name())))
if err != nil {
return nil, fmt.Errorf("failed to get absolute path for cache: %w", err)
}

6
core/tasks/parsed/execute.go
View File

@@ -26,7 +26,7 @@ func (t *Task) Execute(ctx context.Context) error {
t.progress.OnStart(ctx, t)
}
eg, gctx := errgroup.WithContext(ctx)
eg.SetLimit(config.Cfg.Workers)
eg.SetLimit(config.C().Workers)
for _, resource := range t.item.Resources {
eg.Go(func() error {
t.processingMu.RLock()
@@ -96,7 +96,7 @@ func (t *Task) processResource(ctx context.Context, resource parser.Resource) er
if t.stream {
return t.Stor.Save(ctx, resp.Body, path.Join(t.StorPath, resource.Filename))
}
cacheFile, err := fsutil.CreateFile(filepath.Join(config.Cfg.Temp.BasePath,
cacheFile, err := fsutil.CreateFile(filepath.Join(config.C().Temp.BasePath,
fmt.Sprintf("resource_%s_%s", t.ID, resource.Filename)))
if err != nil {
return fmt.Errorf("failed to create cache file for resource %s: %w", resource.URL, err)
@@ -131,7 +131,7 @@ func (t *Task) processResource(ctx context.Context, resource parser.Resource) er
return fmt.Errorf("failed to seek cache file for resource %s: %w", resource.URL, err)
}
return t.Stor.Save(ctx, cacheFile, path.Join(t.StorPath, resource.Filename))
}, retry.Context(ctx), retry.RetryTimes(uint(config.Cfg.Retry)))
}, retry.Context(ctx), retry.RetryTimes(uint(config.C().Retry)))
if ctx.Err() != nil {
return ctx.Err()
}

2
core/tasks/parsed/task.go
View File

@@ -54,7 +54,7 @@ func NewTask(
},
}
_, ok := stor.(storage.StorageCannotStream)
stream := config.Cfg.Stream && !ok
stream := config.C().Stream && !ok
return &Task{
ID: id,
Ctx: ctx,

6
core/tasks/telegraph/execute.go
View File

@@ -20,7 +20,7 @@ func (t *Task) Execute(ctx context.Context) error {
logger.Infof("Starting Telegraph task %s", t.PhPath)
t.progress.OnStart(ctx, t)
eg, gctx := errgroup.WithContext(ctx)
eg.SetLimit(config.Cfg.Workers)
eg.SetLimit(config.C().Workers)
for i, pic := range t.Pics {
eg.Go(func() error {
err := t.processPic(gctx, pic, i)
@@ -46,7 +46,7 @@ func (t *Task) Execute(ctx context.Context) error {
func (t *Task) processPic(ctx context.Context, picUrl string, index int) error {
retryOpts := []retry.Option{
retry.Context(ctx),
retry.RetryTimes(uint(config.Cfg.Retry)),
retry.RetryTimes(uint(config.C().Retry)),
}
var lastErr error
err := retry.Retry(func() error {
@@ -59,7 +59,7 @@ func (t *Task) processPic(ctx context.Context, picUrl string, index int) error {
defer body.Close()
filename := fmt.Sprintf("%d%s", index+1, path.Ext(picUrl))
if t.cannotStream {
cacheFile, err := fsutil.CreateFile(filepath.Join(config.Cfg.Temp.BasePath,
cacheFile, err := fsutil.CreateFile(filepath.Join(config.C().Temp.BasePath,
fmt.Sprintf("tph_%s_%s", t.TaskID(), filename),
))
if err != nil {

4
core/tasks/tfile/execute.go
View File

@@ -57,7 +57,7 @@ func (t *Task) Execute(ctx context.Context) error {
return fmt.Errorf("failed to get file stat: %w", err)
}
vctx := context.WithValue(ctx, ctxkey.ContentLength, fileStat.Size())
for i := range config.Cfg.Retry + 1 {
for i := range config.C().Retry + 1 {
if err = vctx.Err(); err != nil {
return fmt.Errorf("context canceled while saving file: %w", err)
}
@@ -68,7 +68,7 @@ func (t *Task) Execute(ctx context.Context) error {
}
defer file.Close()
if err = t.Storage.Save(vctx, file, t.Path); err != nil {
if i == config.Cfg.Retry {
if i == config.C().Retry {
return fmt.Errorf("failed to save file: %w", err)
}
logger.Errorf("Failed to save file: %s, retrying...", err)

4
core/tasks/tfile/tftask.go
View File

@@ -35,8 +35,8 @@ func NewTGFileTask(
progress ProgressTracker,
) (*Task, error) {
_, ok := stor.(storage.StorageCannotStream)
if !config.Cfg.Stream || ok {
cachePath, err := filepath.Abs(filepath.Join(config.Cfg.Temp.BasePath, fmt.Sprintf("%s_%s", id, file.Name())))
if !config.C().Stream || ok {
cachePath, err := filepath.Abs(filepath.Join(config.C().Temp.BasePath, fmt.Sprintf("%s_%s", id, file.Name())))
if err != nil {
return nil, fmt.Errorf("failed to get absolute path for cache: %w", err)
}