refactor: streamline storage configuration loading and remove redundant code

feat: add Minio storage support

.github/workflows/build-release.yml

@@ -1,3 +1,5 @@
+name: Build Release
+
 on:
   push:
     tags:
@@ -36,6 +38,9 @@ jobs:
       matrix:
         goos: [linux, darwin, windows]
         goarch: [amd64, arm64]
+        exclude:
+          - goos: windows
+            goarch: arm64
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -53,6 +58,7 @@ jobs:
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           github_token: ${{ secrets.GITHUB_TOKEN }}
+          executable_compression: upx
           extra_files: |
             LICENSE
             README.md

.github/workflows/docs.yml

@@ -0,0 +1,22 @@
+name: Deploy Docs
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - "docs/**"
+  workflow_dispatch:
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+      - uses: actions/cache@v4
+        with:
+          key: ${{ github.ref }}
+          path: .cache
+      - run: pip install mkdocs-material
+      - run: cd docs && mkdocs gh-deploy --force

README.md

@@ -1,6 +1,5 @@
 <div align="center">
 
-
 # <img src="docs/logo.jpg" width="45" align="center"> Save Any Bot
 
 **简体中文** | [English](README_EN.md) 
@@ -9,15 +8,6 @@
 
 > _就像 PikPak Bot 一样_
 
-</div
-
-Demo Video:
-
-<div align="center">
-
-
-[SaveAny-Bot 演示视频 ｜ The Demo of SaveAny-Bot.webm](https://github.com/user-attachments/assets/a0de2453-a4d1-4a12-81fb-9d84856dce09)
-
 </div>
 
 ## 部署

README_EN.md

@@ -10,21 +10,13 @@ Save Telegram files to various storage endpoints.
 
 </div>
 
-Demo Video:
-
-<div align="center">
-
-[SaveAny-Bot Demo Video.webm](https://github.com/user-attachments/assets/a0de2453-a4d1-4a12-81fb-9d84856dce09)
-
-</div>
-
 ## Deployment
 
 ### Deploy from Binary
 
 Download the binary file for your platform from the [Release](https://github.com/krau/SaveAny-Bot/releases) page.
 
-Create a `config.toml` file in the extracted directory, refer to [config.toml.example](https://github.com/krau/SaveAny-Bot/blob/main/config.example.toml) for configuration.
+Create a `config.toml` file in the extracted directory, refer to [config.example.toml](https://github.com/krau/SaveAny-Bot/blob/main/config.example.toml) for configuration.
 
 Run:
 
@@ -62,7 +54,7 @@ systemctl enable --now saveany-bot
 
 #### Docker Compose
 
-Download [docker-compose.yml](https://github.com/krau/SaveAny-Bot/blob/main/docker-compose.yml) file and create a `config.toml` file in the same directory, refer to [config.toml.example](https://github.com/krau/SaveAny-Bot/blob/main/config.example.toml) for configuration.
+Download [docker-compose.yml](https://github.com/krau/SaveAny-Bot/blob/main/docker-compose.yml) file and create a `config.toml` file in the same directory, refer to [config.example.toml](https://github.com/krau/SaveAny-Bot/blob/main/config.example.toml) for configuration.
 
 Run:
 

bot/handle_add_task.go

@@ -153,7 +153,6 @@ func AddToQueue(ctx *ext.Context, update *ext.Update) error {
 		Status:         types.Pending,
 		File:           file,
 		StorageName:    storageName,
-		StoragePath:    path.Join(),
 		FileChatID:     record.ChatID,
 		ReplyMessageID: record.ReplyMessageID,
 		FileMessageID:  record.MessageID,
@@ -164,7 +163,7 @@ func AddToQueue(ctx *ext.Context, update *ext.Update) error {
 		task.StoragePath = path.Join(dir.Path, file.FileName)
 	}
 
-	queue.AddTask(task)
+	queue.AddTask(&task)
 
 	entityBuilder := entity.Builder{}
 	var entities []tg.MessageEntityClass

bot/handle_cancel_task.go

@@ -0,0 +1,27 @@
+package bot
+
+import (
+	"strings"
+
+	"github.com/celestix/gotgproto/dispatcher"
+	"github.com/celestix/gotgproto/ext"
+	"github.com/gotd/td/tg"
+	"github.com/krau/SaveAny-Bot/queue"
+)
+
+func cancelTask(ctx *ext.Context, update *ext.Update) error {
+	key := strings.Split(string(update.CallbackQuery.Data), " ")[1]
+	ok := queue.CancelTask(key)
+	if ok {
+		ctx.AnswerCallback(&tg.MessagesSetBotCallbackAnswerRequest{
+			QueryID: update.CallbackQuery.QueryID,
+			Message: "任务已取消",
+		})
+		return dispatcher.EndGroups
+	}
+	ctx.AnswerCallback(&tg.MessagesSetBotCallbackAnswerRequest{
+		QueryID: update.CallbackQuery.QueryID,
+		Message: "任务取消失败",
+	})
+	return dispatcher.EndGroups
+}

bot/handlers.go

@@ -22,5 +22,6 @@ func RegisterHandlers(dispatcher dispatcher.Dispatcher) {
 	dispatcher.AddHandler(handlers.NewMessage(linkRegexFilter, handleLinkMessage))
 	dispatcher.AddHandler(handlers.NewCallbackQuery(filters.CallbackQuery.Prefix("add"), AddToQueue))
 	dispatcher.AddHandler(handlers.NewCallbackQuery(filters.CallbackQuery.Prefix("set_default"), setDefaultStorage))
+	dispatcher.AddHandler(handlers.NewCallbackQuery(filters.CallbackQuery.Prefix("cancel"), cancelTask))
 	dispatcher.AddHandler(handlers.NewMessage(filters.Message.Media, handleFileMessage))
 }

bot/utils.go

@@ -264,7 +264,7 @@ func HandleSilentAddTask(ctx *ext.Context, update *ext.Update, user *dao.User, t
 		})
 		return dispatcher.EndGroups
 	}
-	queue.AddTask(*task)
+	queue.AddTask(task)
 	ctx.EditMessage(update.EffectiveChat().GetID(), &tg.MessagesEditMessageRequest{
 		Message: fmt.Sprintf("已添加到队列: %s\n当前排队任务数: %d", task.FileName(), queue.Len()),
 		ID:      task.ReplyMessageID,

config.example.toml

@@ -2,6 +2,7 @@
 workers = 4 # 同时下载文件数
 retry = 3   # 下载失败重试次数
 threads = 4 # 单个任务下载最大线程数
+stream = false # 使用stream模式, 详情请查看文档
 
 [telegram]
 # Bot Token
@@ -22,7 +23,7 @@ url = "socks5://127.0.0.1:7890"
 [[storages]]
 # 标识名, 需要唯一
 name = "本机1"
-# 存储类型, 目前可用: local , alist , webdav
+# 存储类型, 目前可用: local, alist, webdav, minio
 type = "local"
 # 启用存储
 enable = true
@@ -58,6 +59,16 @@ url = 'https://example.com/dav'
 username = 'username'
 password = 'password'
 
+[[storages]]
+name = "MyMinio"
+type = "minio"
+enable = true
+endpoint = 'play.min.io'
+use_ssl = true
+access_key_id = 'Q3AM3UQ867SPQQA43P2F'
+secret_access_key = 'zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG'
+bucket_name = 'saveanybot'
+base_path = '/path/telegram'
 
 # 用户列表
 [[users]]
@@ -90,4 +101,4 @@ storages = ["本机1"]
 # cache_ttl = 30
 
 # [db]
-# path = "data/data.db" # 数据库文件路径
+# path = "data/data.db" # 数据库文件路径

config/storage_factory.go

@@ -1,5 +1,3 @@
-// storage_config.go
-
 package config
 
 import (
@@ -18,24 +16,49 @@ type StorageConfig interface {
 
 // Base storage config
 type NewStorageConfig struct {
-	Name      string                 `toml:"name" mapstructure:"name" json:"name"`
-	Type      string                 `toml:"type" mapstructure:"type" json:"type"`
-	Enable    bool                   `toml:"enable" mapstructure:"enable" json:"enable"`
-	RawConfig map[string]interface{} `toml:"-" mapstructure:",remain"`
+	Name      string         `toml:"name" mapstructure:"name" json:"name"`
+	Type      string         `toml:"type" mapstructure:"type" json:"type"`
+	Enable    bool           `toml:"enable" mapstructure:"enable" json:"enable"`
+	RawConfig map[string]any `toml:"-" mapstructure:",remain"`
 }
 
-type StorageConfigFactory func(cfg *NewStorageConfig) (StorageConfig, error)
-
-var storageFactories = make(map[string]StorageConfigFactory)
-
-func RegisterStorageFactory(storageType string, factory StorageConfigFactory) {
-	storageFactories[storageType] = factory
+var storageFactories = map[types.StorageType]func(cfg *NewStorageConfig) (StorageConfig, error){
+	types.StorageTypeLocal:  newLocalStorageConfig,
+	types.StorageTypeAlist:  newAlistStorageConfig,
+	types.StorageTypeWebdav: newWebdavStorageConfig,
+	types.StorageTypeMinio:  newMinioStorageConfig,
 }
 
-func init() {
-	RegisterStorageFactory(string(types.StorageTypeLocal), newLocalStorageConfig)
-	RegisterStorageFactory(string(types.StorageTypeAlist), newAlistStorageConfig)
-	RegisterStorageFactory(string(types.StorageTypeWebdav), newWebdavStorageConfig)
+func LoadStorageConfigs(v *viper.Viper) ([]StorageConfig, error) {
+	var baseConfigs []NewStorageConfig
+	if err := v.UnmarshalKey("storages", &baseConfigs); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal storage configs: %w", err)
+	}
+
+	var configs []StorageConfig
+	for _, baseCfg := range baseConfigs {
+		if !baseCfg.Enable {
+			continue
+		}
+
+		factory, ok := storageFactories[types.StorageType(baseCfg.Type)]
+		if !ok {
+			return nil, fmt.Errorf("unsupported storage type: %s", baseCfg.Type)
+		}
+
+		cfg, err := factory(&baseCfg)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create storage config for %s: %w", baseCfg.Name, err)
+		}
+
+		if err := cfg.Validate(); err != nil {
+			return nil, fmt.Errorf("invalid storage config for %s: %w", baseCfg.Name, err)
+		}
+
+		configs = append(configs, cfg)
+	}
+
+	return configs, nil
 }
 
 func newLocalStorageConfig(cfg *NewStorageConfig) (StorageConfig, error) {
@@ -71,34 +94,11 @@ func newWebdavStorageConfig(cfg *NewStorageConfig) (StorageConfig, error) {
 	return &webdavCfg, nil
 }
 
-func LoadStorageConfigs(v *viper.Viper) ([]StorageConfig, error) {
-	var baseConfigs []NewStorageConfig
-	if err := v.UnmarshalKey("storages", &baseConfigs); err != nil {
-		return nil, fmt.Errorf("failed to unmarshal storage configs: %w", err)
+func newMinioStorageConfig(cfg *NewStorageConfig) (StorageConfig, error) {
+	var minioCfg MinioStorageConfig
+	minioCfg.NewStorageConfig = *cfg
+	if err := mapstructure.Decode(cfg.RawConfig, &minioCfg); err != nil {
+		return nil, fmt.Errorf("failed to decode minio storage config: %w", err)
 	}
-
-	var configs []StorageConfig
-	for _, baseCfg := range baseConfigs {
-		if !baseCfg.Enable {
-			continue
-		}
-
-		factory, ok := storageFactories[baseCfg.Type]
-		if !ok {
-			return nil, fmt.Errorf("unsupported storage type: %s", baseCfg.Type)
-		}
-
-		cfg, err := factory(&baseCfg)
-		if err != nil {
-			return nil, fmt.Errorf("failed to create storage config for %s: %w", baseCfg.Name, err)
-		}
-
-		if err := cfg.Validate(); err != nil {
-			return nil, fmt.Errorf("invalid storage config for %s: %w", baseCfg.Name, err)
-		}
-
-		configs = append(configs, cfg)
-	}
-
-	return configs, nil
+	return &minioCfg, nil
 }

config/storages.go

@@ -104,3 +104,37 @@ func (w *WebdavStorageConfig) GetType() types.StorageType {
 func (w *WebdavStorageConfig) GetName() string {
 	return w.Name
 }
+
+type MinioStorageConfig struct {
+	NewStorageConfig
+	Endpoint        string `toml:"endpoint" mapstructure:"endpoint" json:"endpoint"`
+	AccessKeyID     string `toml:"access_key_id" mapstructure:"access_key_id" json:"access_key_id"`
+	SecretAccessKey string `toml:"secret_access_key" mapstructure:"secret_access_key" json:"secret_access_key"`
+	BucketName      string `toml:"bucket_name" mapstructure:"bucket_name" json:"bucket_name"`
+	UseSSL          bool   `toml:"use_ssl" mapstructure:"use_ssl" json:"use_ssl"`
+	BasePath        string `toml:"base_path" mapstructure:"base_path" json:"base_path"`
+}
+
+func (m *MinioStorageConfig) Validate() error {
+	if m.Endpoint == "" {
+		return fmt.Errorf("endpoint is required for minio storage")
+	}
+	if m.AccessKeyID == "" || m.SecretAccessKey == "" {
+		return fmt.Errorf("access_key_id and secret_access_key are required for minio storage")
+	}
+	if m.BucketName == "" {
+		return fmt.Errorf("bucket_name is required for minio storage")
+	}
+	if m.BasePath == "" {
+		return fmt.Errorf("base_path is required for minio storage")
+	}
+	return nil
+}
+
+func (m *MinioStorageConfig) GetType() types.StorageType {
+	return types.StorageTypeMinio
+}
+
+func (m *MinioStorageConfig) GetName() string {
+	return m.Name
+}

config/viper.go

@@ -13,6 +13,7 @@ type Config struct {
 	Retry        int  `toml:"retry" mapstructure:"retry"`
 	NoCleanCache bool `toml:"no_clean_cache" mapstructure:"no_clean_cache" json:"no_clean_cache"`
 	Threads      int  `toml:"threads" mapstructure:"threads" json:"threads"`
+	Stream       bool `toml:"stream" mapstructure:"stream" json:"stream"`
 
 	Users []userConfig `toml:"users" mapstructure:"users" json:"users"`
 

core/core.go

@@ -22,13 +22,12 @@ func worker(queue *queue.TaskQueue, semaphore chan struct{}) {
 		switch task.Status {
 		case types.Pending:
 			logger.L.Infof("Processing task: %s", task.String())
-			if err := processPendingTask(&task); err != nil {
-				logger.L.Errorf("Failed to do task: %s", err)
+			if err := processPendingTask(task); err != nil {
 				task.Error = err
 				if errors.Is(err, context.Canceled) {
-					logger.L.Debugf("Task canceled: %s", task.String())
 					task.Status = types.Canceled
 				} else {
+					logger.L.Errorf("Failed to do task: %s", err)
 					task.Status = types.Failed
 				}
 			} else {
@@ -37,23 +36,43 @@ func worker(queue *queue.TaskQueue, semaphore chan struct{}) {
 			queue.AddTask(task)
 		case types.Succeeded:
 			logger.L.Infof("Task succeeded: %s", task.String())
-			task.Ctx.(*ext.Context).EditMessage(task.ReplyChatID, &tg.MessagesEditMessageRequest{
-				Message: fmt.Sprintf("文件保存成功\n [%s]: %s", task.StorageName, task.StoragePath),
-				ID:      task.ReplyMessageID,
-			})
+			extCtx, ok := task.Ctx.(*ext.Context)
+			if !ok {
+				logger.L.Errorf("Context is not *ext.Context: %T", task.Ctx)
+			} else {
+				extCtx.EditMessage(task.ReplyChatID, &tg.MessagesEditMessageRequest{
+					Message: fmt.Sprintf("文件保存成功\n [%s]: %s", task.StorageName, task.StoragePath),
+					ID:      task.ReplyMessageID,
+				})
+			}
 		case types.Failed:
 			logger.L.Errorf("Task failed: %s", task.String())
-			task.Ctx.(*ext.Context).EditMessage(task.ReplyChatID, &tg.MessagesEditMessageRequest{
-				Message: "文件保存失败\n" + task.Error.Error(),
-				ID:      task.ReplyMessageID,
-			})
+			extCtx, ok := task.Ctx.(*ext.Context)
+			if !ok {
+				logger.L.Errorf("Context is not *ext.Context: %T", task.Ctx)
+			} else {
+				extCtx.EditMessage(task.ReplyChatID, &tg.MessagesEditMessageRequest{
+					Message: "文件保存失败\n" + task.Error.Error(),
+					ID:      task.ReplyMessageID,
+				})
+			}
 		case types.Canceled:
 			logger.L.Infof("Task canceled: %s", task.String())
+			extCtx, ok := task.Ctx.(*ext.Context)
+			if !ok {
+				logger.L.Errorf("Context is not *ext.Context: %T", task.Ctx)
+			} else {
+				extCtx.EditMessage(task.ReplyChatID, &tg.MessagesEditMessageRequest{
+					Message: "任务已取消",
+					ID:      task.ReplyMessageID,
+				})
+			}
 		default:
 			logger.L.Errorf("Unknown task status: %s", task.Status)
 		}
 		<-semaphore
-		logger.L.Debugf("Task done: %s", task.String())
+		logger.L.Debugf("Task done: %s; status: %s", task.String(), task.Status)
+		queue.DoneTask(task)
 	}
 }
 

core/download.go

@@ -1,6 +1,7 @@
 package core
 
 import (
+	"context"
 	"fmt"
 	"path/filepath"
 	"time"
@@ -48,26 +49,62 @@ func processPendingTask(task *types.Task) error {
 		return fmt.Errorf("context is not *ext.Context: %T", task.Ctx)
 	}
 
+	cancelCtx, cancel := context.WithCancel(ctx)
+	task.Cancel = cancel
+
+	downloadBuider := Downloader.Download(bot.Client.API(), task.File.Location).WithThreads(getTaskThreads(task.File.FileSize))
+
+	taskStreamStorage, isStreamStorage := taskStorage.(storage.StreamStorage)
+	if config.Cfg.Stream {
+		if !isStreamStorage {
+			logger.L.Warnf("存储 %s 不支持流式上传", taskStorage.Name())
+		} else {
+			text, entities := buildProgressMessageEntity(task, 0, task.StartTime, 0)
+			ctx.EditMessage(task.ReplyChatID, &tg.MessagesEditMessageRequest{
+				Message:     text,
+				Entities:    entities,
+				ID:          task.ReplyMessageID,
+				ReplyMarkup: getCancelTaskMarkup(task),
+			})
+			uploadStream, err := taskStreamStorage.NewUploadStream(cancelCtx, task.StoragePath)
+			if err != nil {
+				return fmt.Errorf("创建上传流失败: %w", err)
+			}
+			defer uploadStream.Close()
+
+			task.StartTime = time.Now()
+			progressCallback := buildProgressCallback(ctx, task, getProgressUpdateCount(task.File.FileSize))
+
+			progressStream := NewProgressStream(uploadStream, task.File.FileSize, progressCallback)
+
+			_, err = downloadBuider.Stream(cancelCtx, progressStream)
+			if err != nil {
+				return fmt.Errorf("下载文件失败: %w", err)
+			}
+			logger.L.Infof("Uploaded file: %s", task.StoragePath)
+			return nil
+		}
+	}
+
 	text, entities := buildProgressMessageEntity(task, 0, task.StartTime, 0)
 	ctx.EditMessage(task.ReplyChatID, &tg.MessagesEditMessageRequest{
-		Message:  text,
-		Entities: entities,
-		ID:       task.ReplyMessageID,
+		Message:     text,
+		Entities:    entities,
+		ID:          task.ReplyMessageID,
+		ReplyMarkup: getCancelTaskMarkup(task),
 	})
-	progressCallback := buildProgressCallback(ctx, task, getProgressUpdateCount(task.File.FileSize))
 
+	progressCallback := buildProgressCallback(ctx, task, getProgressUpdateCount(task.File.FileSize))
 	dest, err := NewTaskLocalFile(cacheDestPath, task.File.FileSize, progressCallback)
 	if err != nil {
 		return fmt.Errorf("创建文件失败: %w", err)
 	}
 	defer dest.Close()
 	task.StartTime = time.Now()
-	downloadBuider := Downloader.Download(bot.Client.API(), task.File.Location).WithThreads(getTaskThreads(task.File.FileSize))
-	_, err = downloadBuider.Parallel(ctx, dest)
+	_, err = downloadBuider.Parallel(cancelCtx, dest)
 	if err != nil {
 		return fmt.Errorf("下载文件失败: %w", err)
 	}
-
 	defer cleanCacheFile(cacheDestPath)
 
 	fixTaskFileExt(task, cacheDestPath)
@@ -78,5 +115,5 @@ func processPendingTask(task *types.Task) error {
 		ID:      task.ReplyMessageID,
 	})
 
-	return saveFileWithRetry(task, taskStorage, cacheDestPath)
+	return saveFileWithRetry(cancelCtx, task, taskStorage, cacheDestPath)
 }

core/utils.go

@@ -1,7 +1,9 @@
 package core
 
 import (
+	"context"
 	"fmt"
+	"io"
 	"os"
 	"path"
 	"time"
@@ -19,13 +21,21 @@ import (
 	"github.com/krau/SaveAny-Bot/types"
 )
 
-func saveFileWithRetry(task *types.Task, taskStorage storage.Storage, localFilePath string) error {
+func saveFileWithRetry(ctx context.Context, task *types.Task, taskStorage storage.Storage, localFilePath string) error {
 	for i := 0; i <= config.Cfg.Retry; i++ {
-		if err := taskStorage.Save(task.Ctx, localFilePath, task.StoragePath); err != nil {
+		if err := ctx.Err(); err != nil {
+			return fmt.Errorf("context canceled while saving file: %w", err)
+		}
+		if err := taskStorage.Save(ctx, localFilePath, task.StoragePath); err != nil {
 			if i == config.Cfg.Retry {
 				return fmt.Errorf("failed to save file: %w", err)
 			}
 			logger.L.Errorf("Failed to save file: %s, retrying...", err)
+			select {
+			case <-ctx.Done():
+				return fmt.Errorf("context canceled during retry delay: %w", ctx.Err())
+			case <-time.After(time.Duration(i*500) * time.Millisecond):
+			}
 			continue
 		}
 		return nil
@@ -56,22 +66,9 @@ func processPhoto(task *types.Task, taskStorage storage.Storage, cachePath strin
 
 	logger.L.Infof("Downloaded file: %s", cachePath)
 
-	return saveFileWithRetry(task, taskStorage, cachePath)
+	return saveFileWithRetry(task.Ctx, task, taskStorage, cachePath)
 }
 
-// func getProgressBar(progress float64, updateCount int) string {
-// 	bar := ""
-// 	barSize := 100 / updateCount
-// 	for i := 0; i < updateCount; i++ {
-// 		if progress >= float64(barSize*(i+1)) {
-// 			bar += "█"
-// 		} else {
-// 			bar += "░"
-// 		}
-// 	}
-// 	return bar
-// }
-
 func cleanCacheFile(destPath string) {
 	if config.Cfg.Temp.CacheTTL > 0 {
 		common.RmFileAfter(destPath, time.Duration(config.Cfg.Temp.CacheTTL)*time.Second)
@@ -139,13 +136,20 @@ func buildProgressCallback(ctx *ext.Context, task *types.Task, updateCount int)
 		}
 		text, entities := buildProgressMessageEntity(task, bytesRead, task.StartTime, progress)
 		ctx.EditMessage(task.ReplyChatID, &tg.MessagesEditMessageRequest{
-			Message:  text,
-			Entities: entities,
-			ID:       task.ReplyMessageID,
+			Message:     text,
+			Entities:    entities,
+			ID:          task.ReplyMessageID,
+			ReplyMarkup: getCancelTaskMarkup(task),
 		})
 	}
 }
 
+func getCancelTaskMarkup(task *types.Task) *tg.ReplyInlineMarkup {
+	return &tg.ReplyInlineMarkup{
+		Rows: []tg.KeyboardButtonRow{{Buttons: []tg.KeyboardButtonClass{&tg.KeyboardButtonCallback{Text: "取消任务", Data: fmt.Appendf(nil, "cancel %s", task.Key())}}}},
+	}
+}
+
 func fixTaskFileExt(task *types.Task, localFilePath string) {
 	if path.Ext(task.FileName()) == "" {
 		mimeType, err := mimetype.DetectFile(localFilePath)
@@ -217,3 +221,40 @@ func NewTaskLocalFile(filePath string, fileSize int64, progressCallback func(byt
 		callbackInterval: callbackInterval,
 	}, nil
 }
+
+type ProgressStream struct {
+	writer   io.Writer
+	size     int64
+	done     int64
+	callback func(bytesRead, contentLength int64)
+	nextAt   int64
+	interval int64
+}
+
+func (ps *ProgressStream) Write(p []byte) (n int, err error) {
+	n, err = ps.writer.Write(p)
+	if err != nil {
+		return n, err
+	}
+	ps.done += int64(n)
+	if ps.callback != nil && ps.done >= ps.nextAt {
+		ps.callback(ps.done, ps.size)
+		ps.nextAt += ps.interval
+	}
+	return n, nil
+}
+
+func NewProgressStream(writer io.Writer, size int64, callback func(bytesRead, contentLength int64)) *ProgressStream {
+	var interval int64
+	interval = size / 100
+	if interval == 0 {
+		interval = 1
+	}
+	return &ProgressStream{
+		writer:   writer,
+		size:     size,
+		callback: callback,
+		nextAt:   interval,
+		interval: interval,
+	}
+}

docker-compose.yml

@@ -7,4 +7,7 @@ services:
       - ./data:/app/data
       - ./config.toml:/app/config.toml
       - ./downloads:/app/downloads
-      - ./cache:/app/cache
+      - ./cache:/app/cache
+    # 使用 host 模式以便访问宿主机服务 (如代理)
+    # 如果你对 Docker 网络模式熟悉, 可以自行修改
+    network_mode: host

docs/docs/deploy.md

@@ -0,0 +1,94 @@
+# 部署指南
+
+## 从二进制文件部署
+
+在 [Release](https://github.com/krau/SaveAny-Bot/releases) 页面下载对应平台的二进制文件.
+
+在解压后目录新建 `config.toml` 文件, 参考 [config.example.toml](https://github.com/krau/SaveAny-Bot/blob/main/config.example.toml) 编辑配置文件.
+
+运行:
+
+```bash
+chmod +x saveany-bot
+./saveany-bot
+```
+
+### 添加为 systemd 服务
+
+创建文件 `/etc/systemd/system/saveany-bot.service` 并写入以下内容:
+
+```
+[Unit]
+Description=SaveAnyBot
+After=systemd-user-sessions.service
+
+[Service]
+Type=simple
+WorkingDirectory=/yourpath/
+ExecStart=/yourpath/saveany-bot
+Restart=on-failure
+
+[Install]
+WantedBy=multi-user.target
+```
+
+设为开机启动并启动服务:
+
+```bash
+systemctl enable --now saveany-bot
+```
+
+### 为OpenWrt及衍生系统添加开机自启动服务
+
+创建文件 ` /etc/init.d/saveanybot` ，参考[saveanybot](https://github.com/krau/SaveAny-Bot/blob/main/docs/saveanybot)自行修改.
+
+`chmod +x /etc/init.d/saveanybot`
+
+完成后，将文件复制到 `/etc/rc.d`并重命名为`S99saveanybot`.
+
+`chmod +x /etc/rc.d/S99saveanybot`
+
+### 为OpenWrt及衍生系统添加快捷指令
+
+创建文件` /usr/bin/sabot` ，参考[sabot](https://github.com/krau/SaveAny-Bot/blob/main/docs/sabot)自行配置修改，注意此处文件编码仅支持 ANSI 936 .
+
+`chmod +x /usr/bin/sabot`
+
+之后，终端输入`sabot start|stop|restart|status|enable|disable`即可.
+
+
+## 使用 Docker 部署
+
+### Docker Compose
+
+下载 [docker-compose.yml](https://github.com/krau/SaveAny-Bot/blob/main/docker-compose.yml) 文件, 在同目录下新建 `config.toml` 文件, 参考 [config.example.toml](https://github.com/krau/SaveAny-Bot/blob/main/config.example.toml) 编辑配置文件.
+
+启动:
+
+```bash
+docker compose up -d
+```
+
+### Docker
+
+```shell
+docker run -d --name saveany-bot \
+    -v /path/to/config.toml:/app/config.toml \
+    -v /path/to/downloads:/app/downloads \
+    ghcr.io/krau/saveany-bot:latest
+```
+
+## 更新
+
+使用 `upgrade` 或 `up` 升级到最新版
+
+```bash
+./saveany-bot upgrade
+```
+
+如果是 Docker 部署, 使用以下命令更新:
+
+```bash
+docker pull ghcr.io/krau/saveany-bot:latest
+docker restart saveany-bot
+```

docs/docs/faq.md

@@ -0,0 +1,16 @@
+# 常见问题
+
+## 上传 alist 失败也会显示成功
+
+在 alist 管理页面适当调整上传分片大小, 为 alist 使用更稳定的网络环境部署, 都可以减少这种情况的发生.
+
+## Bot 提示下载成功但是 alist 未显示
+
+alist 缓存了目录结构, 参考文档可以调整缓存时间
+
+https://alist.nn.ci/zh/guide/drivers/common.html#缓存过期
+
+## docker部署配置了代理后仍无法连接 telegram (初始化客户端超时)
+
+docker 不能直接访问宿主机网络, 如果你不熟悉其用法, 请将容器设为 host 模式:
+

docs/docs/help.md

@@ -0,0 +1,35 @@
+# 使用帮助
+
+## 保存文件
+
+Bot 接受两种消息: 文件和链接.
+
+目前, 链接仅支持公开频道 (具有用户名) 的链接, 例如: `https://t.me/acherkrau/1097`.
+
+**即使频道禁止了转发和保存, Bot 依然可以下载其文件.**
+
+## 静默模式 (silent)
+
+使用 `/silent` 命令可以开关静默模式.
+
+默认情况下不开启静默模式, Bot 会询问你每个文件的保存位置.
+
+开启静默模式后, Bot 会直接保存文件到默认位置, 无需确认.
+
+在开启静默模式之前, 需要使用 `/storage` 命令设置默认保存位置.
+
+## Stream 模式
+
+在配置文件中将 `stream` 设置为 `true` 可以开启 Stream 模式.
+
+未开启时, Bot 处理任务分为两步: 下载和上传. Bot 会将文件暂存到本地, 然后上传到对应存储位置, 最后删除本地文件.
+
+开启后, Bot 将直接将文件流式传输到存储端, 不需要下载到本地.
+
+该功能对于硬盘空间有限的部署环境十分有用, 然而相较于普通模式也具有一些弊端:
+
+- 无法使用多线程从 telegram 下载文件, 速度较慢.
+- 网络不稳定时, 任务失败率高.
+- 无法在中间层对文件进行处理, 例如自动文件类型识别.
+
+虽然目前 Bot 适配的所有存储端 (Alist, 本地磁盘, Webdav) 都支持 Stream 模式, 但今后可能会有不支持的存储端, 此时即使开启 Stream 模式, Bot 也会自动切换到普通模式.

docs/docs/index.md

@@ -0,0 +1,7 @@
+# SaveAnyBot 文档
+
+SaveAnyBot 是一个可以保存 Telegram 上的文件到云存储的机器人, 就像 PikPak Bot 一样.
+
+不同的是, SaveAnyBot 提供更灵活的存储端选择, 并实现一些更强大的功能.
+
+本项目以 AGPL-3.0 协议开源, 请遵守协议使用.

docs/mkdocs.yml

@@ -0,0 +1,33 @@
+site_name: SaveAnyBot 官方文档
+site_author: Krau
+site_description: SaveAnyBot 是一个可以保存 Telegram 上的文件到多种云存储的机器人, 本文档将帮助你了解如何部署和使用它.
+repo_name: krau/saveany-bot
+repo_url: https://github.com/krau/saveany-bot
+copyright: CC BY-NC-SA 4.0
+theme:
+  name: material
+  language: zh
+  highlightjs: true
+  palette:
+    - media: "(prefers-color-scheme)"
+      toggle:
+        icon: material/brightness-auto
+        name: 切换主题
+    - media: "(prefers-color-scheme: light)"
+      scheme: default
+      primary: indigo
+      toggle:
+        icon: material/brightness-7
+        name: 暗色模式
+    - media: "(prefers-color-scheme: dark)"
+      scheme: slate
+      primary: blue grey
+      toggle:
+        icon: material/brightness-4
+        name: 亮色模式
+
+nav:
+  - index.md
+  - deploy.md
+  - help.md
+  - faq.md

go.mod

@@ -9,10 +9,10 @@ require (
 	github.com/gookit/slog v0.5.7
 	github.com/gotd/contrib v0.21.0
 	github.com/gotd/td v0.120.0
+	github.com/minio/minio-go/v7 v7.0.81
 	github.com/rhysd/go-github-selfupdate v1.2.3
 	github.com/spf13/cobra v1.8.1
 	github.com/spf13/viper v1.19.0
-	github.com/studio-b12/gowebdav v0.10.0
 	golang.org/x/net v0.35.0
 	golang.org/x/time v0.10.0
 )
@@ -32,7 +32,9 @@ require (
 	github.com/go-faster/jx v1.1.0 // indirect
 	github.com/go-faster/xor v1.0.0 // indirect
 	github.com/go-faster/yaml v0.4.6 // indirect
+	github.com/go-ini/ini v1.67.0 // indirect
 	github.com/go-sql-driver/mysql v1.8.1 // indirect
+	github.com/goccy/go-json v0.10.3 // indirect
 	github.com/google/go-github/v30 v30.1.0 // indirect
 	github.com/google/go-querystring v1.1.0 // indirect
 	github.com/google/pprof v0.0.0-20250128161936-077ca0a936bf // indirect
@@ -42,13 +44,16 @@ require (
 	github.com/inconshreveable/go-update v0.0.0-20160112193335-8152e7eb6ccf // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.5 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.8 // indirect
 	github.com/mattn/go-colorable v0.1.14 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/minio/md5-simd v1.1.2 // indirect
 	github.com/ncruces/go-strftime v0.1.9 // indirect
 	github.com/ogen-go/ogen v1.10.0 // indirect
 	github.com/onsi/gomega v1.36.2 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
+	github.com/rs/xid v1.6.0 // indirect
 	github.com/segmentio/asm v1.2.0 // indirect
 	github.com/tcnksm/go-gitconfig v0.1.2 // indirect
 	github.com/ulikunitz/xz v0.5.12 // indirect

go.sum

@@ -49,6 +49,8 @@ github.com/go-faster/xor v1.0.0 h1:2o8vTOgErSGHP3/7XwA5ib1FTtUsNtwCoLLBjl31X38=
 github.com/go-faster/xor v1.0.0/go.mod h1:x5CaDY9UKErKzqfRfFZdfu+OSTfoZny3w5Ak7UxcipQ=
 github.com/go-faster/yaml v0.4.6 h1:lOK/EhI04gCpPgPhgt0bChS6bvw7G3WwI8xxVe0sw9I=
 github.com/go-faster/yaml v0.4.6/go.mod h1:390dRIvV4zbnO7qC9FGo6YYutc+wyyUSHBgbXL52eXk=
+github.com/go-ini/ini v1.67.0 h1:z6ZrTEZqSWOTyH2FlglNbNgARyHG8oLW9gMELqKr06A=
+github.com/go-ini/ini v1.67.0/go.mod h1:ByCAeIL28uOIIG0E3PJtZPDL8WnHpFKFOtgjp+3Ies8=
 github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
 github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
@@ -56,6 +58,8 @@ github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre
 github.com/go-sql-driver/mysql v1.7.0/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI=
 github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y=
 github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg=
+github.com/goccy/go-json v0.10.3 h1:KZ5WoDbxAIgm2HNbYckL0se1fHD6rz5j4ywS6ebzDqA=
+github.com/goccy/go-json v0.10.3/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
 github.com/golang-sql/civil v0.0.0-20220223132316-b832511892a9 h1:au07oEsX2xN0ktxqI+Sida1w446QrXBRJ0nee3SNZlA=
 github.com/golang-sql/civil v0.0.0-20220223132316-b832511892a9/go.mod h1:8vg3r2VgvsThLBIFL93Qb5yWzgyZWhEmBwUJWevAkK0=
 github.com/golang-sql/sqlexp v0.1.0 h1:ZCD6MBpcuOVfGVqsEmY5/4FtYiKz6tSyUv9LPEDei6A=
@@ -111,6 +115,9 @@ github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
 github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
 github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
 github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
+github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
+github.com/klauspost/cpuid/v2 v2.2.8 h1:+StwCXwm9PdpiEkPyzBXIy+M9KUb4ODm0Zarf1kS5BM=
+github.com/klauspost/cpuid/v2 v2.2.8/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
@@ -128,6 +135,10 @@ github.com/mattn/go-sqlite3 v1.14.17 h1:mCRHCLDUBXgpKAqIKsaAaAsrAlbkeomtRFKXh2L6
 github.com/mattn/go-sqlite3 v1.14.17/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
 github.com/microsoft/go-mssqldb v1.7.2 h1:CHkFJiObW7ItKTJfHo1QX7QBBD1iV+mn1eOyRP3b/PA=
 github.com/microsoft/go-mssqldb v1.7.2/go.mod h1:kOvZKUdrhhFQmxLZqbwUV0rHkNkZpthMITIb2Ko1IoA=
+github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
+github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM=
+github.com/minio/minio-go/v7 v7.0.81 h1:SzhMN0TQ6T/xSBu6Nvw3M5M8voM+Ht8RH3hE8S7zxaA=
+github.com/minio/minio-go/v7 v7.0.81/go.mod h1:84gmIilaX4zcvAWWzJ5Z1WI5axN+hAbM5w25xf8xvC0=
 github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
 github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4=
@@ -150,6 +161,8 @@ github.com/rhysd/go-github-selfupdate v1.2.3 h1:iaa+J202f+Nc+A8zi75uccC8Wg3omaM7
 github.com/rhysd/go-github-selfupdate v1.2.3/go.mod h1:mp/N8zj6jFfBQy/XMYoWsmfzxazpPAODuqarmPDe2Rg=
 github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
 github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
+github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU=
+github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/sagikazarmark/locafero v0.7.0 h1:5MqpDsTGNDhY8sGp0Aowyf0qKsPrhewaLSsFaodPcyo=
 github.com/sagikazarmark/locafero v0.7.0/go.mod h1:2za3Cg5rMaTMoG/2Ulr9AwtFaIppKXTRYnozin4aB5k=
@@ -172,8 +185,6 @@ github.com/spf13/viper v1.19.0 h1:RWq5SEjt8o25SROyN3z2OrDB9l7RPd3lwTWU8EcEdcI=
 github.com/spf13/viper v1.19.0/go.mod h1:GQUN9bilAbhU/jgc1bKs99f/suXKeUMct8Adx5+Ntkg=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-github.com/studio-b12/gowebdav v0.10.0 h1:Yewz8FFiadcGEu4hxS/AAJQlHelndqln1bns3hcJIYc=
-github.com/studio-b12/gowebdav v0.10.0/go.mod h1:bHA7t77X/QFExdeAnDzK6vKM34kEZAcE1OX4MfiwjkE=
 github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8=
 github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU=
 github.com/tcnksm/go-gitconfig v0.1.2 h1:iiDhRitByXAEyjgBqsKi9QU4o2TNtv9kPP3RgPgXBPw=
@@ -226,6 +237,7 @@ golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
 golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=

queue/queue.go

@@ -8,30 +8,65 @@ import (
 )
 
 type TaskQueue struct {
-	list  *list.List
-	cond  *sync.Cond
-	mutex *sync.Mutex
+	list      *list.List
+	cond      *sync.Cond
+	mutex     *sync.Mutex
+	activeMap map[string]*types.Task
 }
 
-func (q *TaskQueue) AddTask(task types.Task) {
+func (q *TaskQueue) AddTask(task *types.Task) {
 	q.mutex.Lock()
 	defer q.mutex.Unlock()
 	q.list.PushBack(task)
 	q.cond.Signal()
+	if task.Status != types.Pending {
+		delete(q.activeMap, task.Key())
+	}
 }
 
-func (q *TaskQueue) GetTask() types.Task {
+func (q *TaskQueue) GetTask() *types.Task {
 	q.mutex.Lock()
 	defer q.mutex.Unlock()
 	for q.list.Len() == 0 {
 		q.cond.Wait()
 	}
 	e := q.list.Front()
-	task := e.Value.(types.Task)
+	task := e.Value.(*types.Task)
 	q.list.Remove(e)
+	if task.Status == types.Pending {
+		q.activeMap[task.Key()] = task
+	}
 	return task
 }
 
+func (q *TaskQueue) DoneTask(task *types.Task) {
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+	delete(q.activeMap, task.Key())
+}
+
+func (q *TaskQueue) CancelTask(key string) bool {
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+	if task, ok := q.activeMap[key]; ok {
+		if task.Cancel != nil {
+			task.Cancel()
+			return true
+		}
+	}
+	for e := q.list.Front(); e != nil; e = e.Next() {
+		task := e.Value.(*types.Task)
+		if task.Key() == key {
+			if task.Cancel != nil {
+				task.Cancel()
+			}
+			q.list.Remove(e)
+			return true
+		}
+	}
+	return false
+}
+
 func (q *TaskQueue) Len() int {
 	q.mutex.Lock()
 	defer q.mutex.Unlock()
@@ -47,20 +82,29 @@ func init() {
 func NewQueue() *TaskQueue {
 	m := &sync.Mutex{}
 	return &TaskQueue{
-		list:  list.New(),
-		cond:  sync.NewCond(m),
-		mutex: m,
+		list:      list.New(),
+		cond:      sync.NewCond(m),
+		mutex:     m,
+		activeMap: make(map[string]*types.Task),
 	}
 }
 
-func AddTask(task types.Task) {
+func AddTask(task *types.Task) {
 	Queue.AddTask(task)
 }
 
-func GetTask() types.Task {
+func GetTask() *types.Task {
 	return Queue.GetTask()
 }
 
 func Len() int {
 	return Queue.Len()
 }
+
+func CancelTask(key string) bool {
+	return Queue.CancelTask(key)
+}
+
+func DoneTask(task *types.Task) {
+	Queue.DoneTask(task)
+}

storage/alist/alist.go

@@ -9,6 +9,7 @@ import (
 	"net/url"
 	"os"
 	"path"
+	"sync"
 	"time"
 
 	"github.com/krau/SaveAny-Bot/config"
@@ -116,7 +117,7 @@ func (a *Alist) Save(ctx context.Context, filePath, storagePath string) error {
 	}
 	req.Header.Set("Authorization", a.token)
 	req.Header.Set("File-Path", url.PathEscape(storagePath))
-	req.Header.Set("As-Task", "true")
+	// req.Header.Set("As-Task", "true")
 	req.Header.Set("Content-Type", "application/octet-stream")
 	req.ContentLength = filestat.Size()
 
@@ -150,3 +151,87 @@ func (a *Alist) Save(ctx context.Context, filePath, storagePath string) error {
 func (a *Alist) JoinStoragePath(task types.Task) string {
 	return path.Join(a.config.BasePath, task.StoragePath)
 }
+
+type uploadStream struct {
+	ctx         context.Context
+	client      *http.Client
+	token       string
+	storagePath string
+	baseURL     string
+	pr          *io.PipeReader
+	pw          *io.PipeWriter
+	errChan     chan error
+	once        sync.Once
+}
+
+func (us *uploadStream) Write(p []byte) (int, error) {
+	return us.pw.Write(p)
+}
+
+func (us *uploadStream) Close() error {
+	var uploadErr error
+	us.once.Do(func() {
+		if err := us.pw.Close(); err != nil {
+			uploadErr = fmt.Errorf("failed to close pipe writer: %w", err)
+			return
+		}
+
+		if err := <-us.errChan; err != nil {
+			uploadErr = err
+		}
+	})
+	return uploadErr
+}
+
+func (a *Alist) NewUploadStream(ctx context.Context, storagePath string) (io.WriteCloser, error) {
+	if a.token == "" {
+		if err := a.getToken(); err != nil {
+			return nil, fmt.Errorf("not logged in to Alist: %w", err)
+		}
+	}
+
+	pr, pw := io.Pipe()
+
+	us := &uploadStream{
+		ctx:         ctx,
+		client:      a.client,
+		token:       a.token,
+		storagePath: storagePath,
+		baseURL:     a.baseURL,
+		pr:          pr,
+		pw:          pw,
+		errChan:     make(chan error, 1),
+	}
+
+	go func() {
+		defer close(us.errChan)
+
+		req, err := http.NewRequestWithContext(ctx, http.MethodPut, a.baseURL+"/api/fs/put", pr)
+		if err != nil {
+			us.errChan <- fmt.Errorf("failed to create request: %w", err)
+			return
+		}
+
+		req.Header.Set("Authorization", a.token)
+		req.Header.Set("File-Path", url.PathEscape(storagePath))
+		// req.Header.Set("As-Task", "true")
+		req.Header.Set("Content-Type", "application/octet-stream")
+
+		resp, err := a.client.Do(req)
+		if err != nil {
+			us.errChan <- fmt.Errorf("failed to send request: %w", err)
+			return
+		}
+		defer resp.Body.Close()
+
+		if resp.StatusCode != http.StatusOK {
+			body, _ := io.ReadAll(resp.Body)
+			us.errChan <- fmt.Errorf("failed to upload file, status code: %d, response: %s", resp.StatusCode, string(body))
+			return
+		}
+
+		us.errChan <- nil
+	}()
+
+	return us, nil
+}

storage/local/local.go

@@ -3,6 +3,7 @@ package local
 import (
 	"context"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 
@@ -55,3 +56,18 @@ func (l *Local) Save(ctx context.Context, filePath, storagePath string) error {
 func (l *Local) JoinStoragePath(task types.Task) string {
 	return filepath.Join(l.config.BasePath, task.StoragePath)
 }
+
+func (l *Local) NewUploadStream(ctx context.Context, path string) (io.WriteCloser, error) {
+	absPath, err := filepath.Abs(path)
+	if err != nil {
+		return nil, err
+	}
+	if err := fileutil.CreateDir(filepath.Dir(absPath)); err != nil {
+		return nil, err
+	}
+	file, err := os.Create(absPath)
+	if err != nil {
+		return nil, err
+	}
+	return file, nil
+}

storage/minio/client.go

@@ -0,0 +1,71 @@
+package minio
+
+import (
+	"context"
+	"fmt"
+	"path"
+
+	"github.com/krau/SaveAny-Bot/config"
+	"github.com/krau/SaveAny-Bot/logger"
+	"github.com/krau/SaveAny-Bot/types"
+	"github.com/minio/minio-go/v7"
+	"github.com/minio/minio-go/v7/pkg/credentials"
+)
+
+type Minio struct {
+	config config.MinioStorageConfig
+	client *minio.Client
+}
+
+func (m *Minio) Init(cfg config.StorageConfig) error {
+	minioConfig, ok := cfg.(*config.MinioStorageConfig)
+	if !ok {
+		return fmt.Errorf("failed to cast minio config")
+	}
+	if err := minioConfig.Validate(); err != nil {
+		return err
+	}
+	m.config = *minioConfig
+
+	client, err := minio.New(m.config.Endpoint, &minio.Options{
+		Creds:  credentials.NewStaticV4(m.config.AccessKeyID, m.config.SecretAccessKey, ""),
+		Secure: m.config.UseSSL,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to create minio client: %w", err)
+	}
+
+	exists, err := client.BucketExists(context.Background(), m.config.BucketName)
+	if err != nil {
+		return fmt.Errorf("failed to check bucket existence: %w", err)
+	}
+	if !exists {
+		return fmt.Errorf("bucket %s does not exist", m.config.BucketName)
+	}
+
+	m.client = client
+	return nil
+}
+
+func (m *Minio) Type() types.StorageType {
+	return types.StorageTypeMinio
+}
+
+func (m *Minio) Name() string {
+	return m.config.Name
+}
+
+func (m *Minio) JoinStoragePath(task types.Task) string {
+	return path.Join(m.config.BasePath, task.StoragePath)
+}
+
+func (m *Minio) Save(ctx context.Context, localFilePath, storagePath string) error {
+	logger.L.Infof("Saving file %s to %s", localFilePath, storagePath)
+
+	_, err := m.client.FPutObject(ctx, m.config.BucketName, storagePath, localFilePath, minio.PutObjectOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to upload file to minio: %w", err)
+	}
+
+	return nil
+}

storage/minio/stream.go

@@ -0,0 +1,92 @@
+package minio
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/krau/SaveAny-Bot/logger"
+	"github.com/minio/minio-go/v7"
+)
+
+type MinioWriter struct {
+	pipeWriter *io.PipeWriter
+	done       chan error
+	path       string
+	ctx        context.Context
+	closed     bool
+}
+
+func (w *MinioWriter) Write(p []byte) (n int, err error) {
+	select {
+	case <-w.ctx.Done():
+		return 0, w.ctx.Err()
+	default:
+		return w.pipeWriter.Write(p)
+	}
+}
+
+func (w *MinioWriter) Close() error {
+	if w.closed {
+		return nil
+	}
+	w.closed = true
+
+	if err := w.pipeWriter.Close(); err != nil {
+		return fmt.Errorf("failed to close pipe writer: %w", err)
+	}
+
+	select {
+	case err := <-w.done:
+		if err != nil {
+			return fmt.Errorf("upload failed: %w", err)
+		}
+		return nil
+	case <-w.ctx.Done():
+		return fmt.Errorf("upload cancelled: %w", w.ctx.Err())
+	}
+}
+
+func (m *Minio) NewUploadStream(ctx context.Context, storagePath string) (io.WriteCloser, error) {
+	logger.L.Infof("Creating upload stream for %s", storagePath)
+
+	uploadCtx, cancel := context.WithCancel(ctx)
+	pipeReader, pipeWriter := io.Pipe()
+	done := make(chan error, 1)
+
+	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				done <- fmt.Errorf("panic during upload: %v", r)
+			}
+			pipeReader.Close()
+			cancel()
+		}()
+
+		info, err := m.client.PutObject(
+			uploadCtx,
+			m.config.BucketName,
+			storagePath,
+			pipeReader,
+			-1,
+			minio.PutObjectOptions{},
+		)
+
+		if err != nil {
+			logger.L.Errorf("Failed to upload to %s: %v", storagePath, err)
+			done <- err
+			return
+		}
+
+		logger.L.Infof("uploaded %d bytes to %s", info.Size, storagePath)
+		done <- nil
+	}()
+
+	return &MinioWriter{
+		pipeWriter: pipeWriter,
+		done:       done,
+		path:       storagePath,
+		ctx:        uploadCtx,
+		closed:     false,
+	}, nil
+}

storage/storage.go

@@ -3,12 +3,14 @@ package storage
 import (
 	"context"
 	"fmt"
+	"io"
 
 	"github.com/krau/SaveAny-Bot/config"
 	"github.com/krau/SaveAny-Bot/logger"
 	"github.com/krau/SaveAny-Bot/storage/alist"
 	"github.com/krau/SaveAny-Bot/storage/local"
 	"github.com/krau/SaveAny-Bot/storage/webdav"
+	"github.com/krau/SaveAny-Bot/storage/minio"
 	"github.com/krau/SaveAny-Bot/types"
 )
 
@@ -20,6 +22,11 @@ type Storage interface {
 	Save(cttx context.Context, localFilePath, storagePath string) error
 }
 
+type StreamStorage interface {
+	Storage
+	NewUploadStream(ctx context.Context, path string) (io.WriteCloser, error)
+}
+
 var Storages = make(map[string]Storage)
 
 var UserStorages = make(map[int64][]Storage)
@@ -84,6 +91,7 @@ var storageConstructors = map[string]StorageConstructor{
 	string(types.StorageTypeAlist):  func() Storage { return new(alist.Alist) },
 	string(types.StorageTypeLocal):  func() Storage { return new(local.Local) },
 	string(types.StorageTypeWebdav): func() Storage { return new(webdav.Webdav) },
+	string(types.StorageTypeMinio):  func() Storage { return new(minio.Minio) },
 }
 
 func NewStorage(cfg config.StorageConfig) (Storage, error) {

storage/webdav/client.go

@@ -0,0 +1,70 @@
+package webdav
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+)
+
+type Client struct {
+	BaseURL    string
+	Username   string
+	Password   string
+	httpClient *http.Client
+}
+
+func NewClient(baseURL, username, password string, httpClient *http.Client) *Client {
+	if !strings.HasSuffix(baseURL, "/") {
+		baseURL += "/"
+	}
+	if httpClient == nil {
+		httpClient = http.DefaultClient
+	}
+	return &Client{
+		BaseURL:    baseURL,
+		Username:   username,
+		Password:   password,
+		httpClient: httpClient,
+	}
+}
+
+func (c *Client) doRequest(ctx context.Context, method, url string, body io.Reader) (*http.Response, error) {
+	req, err := http.NewRequestWithContext(ctx, method, url, body)
+	if err != nil {
+		return nil, err
+	}
+	if c.Username != "" && c.Password != "" {
+		req.SetBasicAuth(c.Username, c.Password)
+	}
+	return c.httpClient.Do(req)
+}
+
+func (c *Client) MkDir(ctx context.Context, dirPath string) error {
+	url := c.BaseURL + dirPath
+	resp, err := c.doRequest(ctx, "MKCOL", url, nil)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 200 && resp.StatusCode < 300 {
+		return nil
+	}
+	return fmt.Errorf("MKCOL: %s", resp.Status)
+}
+
+func (c *Client) WriteFile(ctx context.Context, remotePath string, content io.Reader) error {
+	url := c.BaseURL + remotePath
+	resp, err := c.doRequest(ctx, "PUT", url, content)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 200 && resp.StatusCode < 300 {
+		return nil
+	}
+	return fmt.Errorf("PUT: %s", resp.Status)
+}

storage/webdav/stream.go

@@ -0,0 +1,58 @@
+package webdav
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"path"
+
+	"github.com/krau/SaveAny-Bot/logger"
+)
+
+type WebdavWriter struct {
+	pipeWriter *io.PipeWriter
+	done       chan error
+	path       string
+}
+
+func (w *WebdavWriter) Write(p []byte) (n int, err error) {
+	return w.pipeWriter.Write(p)
+}
+
+func (w *WebdavWriter) Close() error {
+	if err := w.pipeWriter.Close(); err != nil {
+		return err
+	}
+	if err := <-w.done; err != nil {
+		return fmt.Errorf("upload failed: %w", err)
+	}
+
+	return nil
+}
+
+func (w *Webdav) NewUploadStream(ctx context.Context, storagePath string) (io.WriteCloser, error) {
+	if err := w.client.MkDir(ctx, path.Dir(storagePath)); err != nil {
+		logger.L.Errorf("Failed to create directory %s: %v", path.Dir(storagePath), err)
+		return nil, ErrFailedToCreateDirectory
+	}
+	pipeReader, pipeWriter := io.Pipe()
+	done := make(chan error, 1)
+	go func() {
+		defer func() {
+			if err := recover(); err != nil {
+				done <- fmt.Errorf("panic during upload: %v", err)
+			}
+		}()
+
+		err := w.client.WriteFile(ctx, storagePath, pipeReader)
+
+		pipeReader.Close()
+		done <- err
+	}()
+
+	return &WebdavWriter{
+		pipeWriter: pipeWriter,
+		done:       done,
+		path:       storagePath,
+	}, nil
+}

storage/webdav/webdav.go

@@ -3,6 +3,7 @@ package webdav
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"os"
 	"path"
 	"time"
@@ -10,12 +11,11 @@ import (
 	"github.com/krau/SaveAny-Bot/config"
 	"github.com/krau/SaveAny-Bot/logger"
 	"github.com/krau/SaveAny-Bot/types"
-	"github.com/studio-b12/gowebdav"
 )
 
 type Webdav struct {
 	config config.WebdavStorageConfig
-	client *gowebdav.Client
+	client *Client
 }
 
 func (w *Webdav) Init(cfg config.StorageConfig) error {
@@ -27,12 +27,9 @@ func (w *Webdav) Init(cfg config.StorageConfig) error {
 		return err
 	}
 	w.config = *webdavConfig
-	client := gowebdav.NewClient(webdavConfig.URL, webdavConfig.Username, webdavConfig.Password)
-	if err := client.Connect(); err != nil {
-		return fmt.Errorf("failed to connect to webdav server: %w", err)
-	}
-	client.SetTimeout(12 * time.Hour)
-	w.client = client
+	w.client = NewClient(w.config.URL, w.config.Username, w.config.Password, &http.Client{
+		Timeout: time.Hour * 12,
+	})
 	return nil
 }
 
@@ -46,7 +43,7 @@ func (w *Webdav) Name() string {
 
 func (w *Webdav) Save(ctx context.Context, filePath, storagePath string) error {
 	logger.L.Infof("Saving file %s to %s", filePath, storagePath)
-	if err := w.client.MkdirAll(path.Dir(storagePath), os.ModePerm); err != nil {
+	if err := w.client.MkDir(ctx, path.Dir(storagePath)); err != nil {
 		logger.L.Errorf("Failed to create directory %s: %v", path.Dir(storagePath), err)
 		return ErrFailedToCreateDirectory
 	}
@@ -57,7 +54,7 @@ func (w *Webdav) Save(ctx context.Context, filePath, storagePath string) error {
 	}
 	defer file.Close()
 
-	if err := w.client.WriteStream(storagePath, file, os.ModePerm); err != nil {
+	if err := w.client.WriteFile(ctx, storagePath, file); err != nil {
 		logger.L.Errorf("Failed to write file %s: %v", storagePath, err)
 		return ErrFailedToWriteFile
 	}

types/types.go

@@ -25,17 +25,20 @@ var (
 	StorageTypeLocal  StorageType = "local"
 	StorageTypeWebdav StorageType = "webdav"
 	StorageTypeAlist  StorageType = "alist"
+	StorageTypeMinio  StorageType = "minio"
 )
 
-var StorageTypes = []StorageType{StorageTypeLocal, StorageTypeAlist, StorageTypeWebdav}
+var StorageTypes = []StorageType{StorageTypeLocal, StorageTypeAlist, StorageTypeWebdav, StorageTypeMinio}
 var StorageTypeDisplay = map[StorageType]string{
 	StorageTypeLocal:  "本地磁盘",
 	StorageTypeWebdav: "WebDAV",
 	StorageTypeAlist:  "Alist",
+	StorageTypeMinio:  "Minio",
 }
 
 type Task struct {
 	Ctx         context.Context
+	Cancel      context.CancelFunc
 	Error       error
 	Status      TaskStatus
 	File        *File
@@ -52,6 +55,10 @@ type Task struct {
 	UserID int64
 }
 
+func (t Task) Key() string {
+	return fmt.Sprintf("%d:%d", t.FileChatID, t.FileMessageID)
+}
+
 func (t Task) String() string {
 	return fmt.Sprintf("[%d:%d]:%s", t.FileChatID, t.FileMessageID, t.File.FileName)
 }