GithubBackup/SaveAny-Bot
1
0
Fork 0
mirror of https://github.com/krau/SaveAny-Bot.git synced 2026-05-10 17:52:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: migrate to gotd (wip)

Browse Source
This commit is contained in:
krau
2024-11-08 23:00:57 +08:00
parent 32bd391129
commit fbdfc04ad8
14 changed files with 465 additions and 387 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
core/core.go
View File

@@ -3,55 +3,56 @@ package core
import (
"context"
"errors"
"os"
"time"
"github.com/amarnathcjd/gogram/telegram"
"github.com/krau/SaveAny-Bot/bot"
"github.com/krau/SaveAny-Bot/common"
"github.com/celestix/gotgproto/ext"
"github.com/gotd/td/tg"
"github.com/krau/SaveAny-Bot/config"
"github.com/krau/SaveAny-Bot/logger"
"github.com/krau/SaveAny-Bot/queue"
"github.com/krau/SaveAny-Bot/storage"
"github.com/krau/SaveAny-Bot/types"
)
func processPendingTask(task types.Task) error {
os.MkdirAll(config.Cfg.Temp.BasePath, os.ModePerm)
logger.L.Debugf("Start processing task: %s", task.FileName)
time.Sleep(10 * time.Second)
logger.L.Debugf("Task done: %s", task.FileName)
message, err := bot.Client.GetMessageByID(task.ChatID, task.MessageID)
if err != nil {
return err
}
logger.L.Debugf("Start downloading file: %s", task.FileName)
bot.Client.EditMessage(task.ChatID, task.ReplyMessageID, "正在下载文件...")
dest, err := message.Download(&telegram.DownloadOptions{
FileName: common.GetCacheFilePath(task.FileName),
Threads: config.Cfg.Threads,
ChunkSize: config.Cfg.ChunkSize,
// ProgressCallback: func(totalBytes, downloadedBytes int64) {},
})
if err != nil {
return err
}
// os.MkdirAll(config.Cfg.Temp.BasePath, os.ModePerm)
defer func() {
if config.Cfg.Temp.CacheTTL > 0 {
common.RmFileAfter(dest, time.Duration(config.Cfg.Temp.CacheTTL)*time.Second)
} else {
if err := os.Remove(dest); err != nil {
logger.L.Errorf("Failed to purge file: %s", err)
}
}
}()
if task.StoragePath == "" {
task.StoragePath = task.FileName
}
// message, err := bot.Client.GetMessageByID(task.ChatID, task.MessageID)
// if err != nil {
// return err
// }
// logger.L.Debugf("Start downloading file: %s", task.FileName)
// bot.Client.EditMessage(task.ChatID, task.ReplyMessageID, "正在下载文件...")
// dest, err := message.Download(&telegram.DownloadOptions{
// FileName: common.GetCacheFilePath(task.FileName),
// Threads: config.Cfg.Threads,
// ChunkSize: config.Cfg.ChunkSize,
// // ProgressCallback: func(totalBytes, downloadedBytes int64) {},
// })
// if err != nil {
// return err
// }
bot.Client.EditMessage(task.ChatID, task.ReplyMessageID, "下载完成, 正在转存文件...")
if err := storage.Save(task.Storage, task.Ctx, dest, task.StoragePath); err != nil {
return err
}
// defer func() {
// if config.Cfg.Temp.CacheTTL > 0 {
// common.RmFileAfter(dest, time.Duration(config.Cfg.Temp.CacheTTL)*time.Second)
// } else {
// if err := os.Remove(dest); err != nil {
// logger.L.Errorf("Failed to purge file: %s", err)
// }
// }
// }()
// if task.StoragePath == "" {
// task.StoragePath = task.FileName
// }
// bot.Client.EditMessage(task.ChatID, task.ReplyMessageID, "下载完成, 正在转存文件...")
// if err := storage.Save(task.Storage, task.Ctx, dest, task.StoragePath); err != nil {
// return err
// }
return nil
}
@@ -69,6 +70,7 @@ func worker(queue *queue.TaskQueue, semaphore chan struct{}) {
logger.L.Errorf("Failed to do task: %s", err)
task.Error = err
if errors.Is(err, context.Canceled) {
logger.L.Debugf("Task canceled: %s", task.String())
task.Status = types.Canceled
} else {
task.Status = types.Failed
@@ -79,10 +81,16 @@ func worker(queue *queue.TaskQueue, semaphore chan struct{}) {
queue.AddTask(task)
case types.Succeeded:
logger.L.Infof("Task succeeded: %s", task.String())
bot.Client.EditMessage(task.ChatID, task.ReplyMessageID, "文件保存成功")
task.Ctx.(*ext.Context).EditMessage(task.ChatID, &tg.MessagesEditMessageRequest{
Message: "文件保存成功",
ID: task.ReplyMessageID,
})
case types.Failed:
logger.L.Errorf("Task failed: %s", task.String())
bot.Client.EditMessage(task.ChatID, task.ReplyMessageID, "文件保存失败")
task.Ctx.(*ext.Context).EditMessage(task.ChatID, &tg.MessagesEditMessageRequest{
Message: "文件保存失败",
ID: task.ReplyMessageID,
})
case types.Canceled:
logger.L.Infof("Task canceled: %s", task.String())
default:
@@ -99,4 +107,5 @@ func Run() {
for i := 0; i < config.Cfg.Workers; i++ {
go worker(queue.Queue, semaphore)
}
}