From 692b7fb9df1c12faebfec360588a40c6afc7bd5c Mon Sep 17 00:00:00 2001 From: yebuwudong <1573876229@qq.com> Date: Thu, 18 Jun 2026 00:05:38 +0800 Subject: [PATCH] Improve Telegram notifications --- internal/service/downloads.go | 101 +++++++++-- internal/service/downloads_test.go | 38 ++++ internal/service/notify_channels.go | 36 +++- internal/service/notify_channels_test.go | 54 ++++++ internal/service/notify_telegram.go | 221 +++++++++++++++++++++-- internal/service/subscription.go | 20 +- 6 files changed, 435 insertions(+), 35 deletions(-) diff --git a/internal/service/downloads.go b/internal/service/downloads.go index a7eeb59..ade9197 100644 --- a/internal/service/downloads.go +++ b/internal/service/downloads.go @@ -931,8 +931,10 @@ func (d *DownloadService) processDownloadSnapshot(ctx context.Context, live []QB stateKey := completedTorrentQueueKey(torrent) complete := torrent.Progress >= 1.0 matchedTask, hasTask := findMatchingTaskByTorrentIdentity(torrent.Name, taskByKey) + autoOrganize := d.downloadAutoOrganizeEnabled(ctx) catchupRecorded := hasTask && d.completedTorrentCatchupRecorded(ctx, torrent) taskNeedsOrganize := hasTask && !catchupRecorded && + autoOrganize && (downloadTaskNeedsCompletion(matchedTask) || recentlyCompletedTorrent(torrent, time.Now())) d.syncDownloadTaskProgress(ctx, torrent, taskByKey) if stateKey == "" { @@ -1060,6 +1062,20 @@ func (d *DownloadService) markCompletedTorrentOrganizeDone(torrent QBitTorrent) const completedTorrentCatchupWindow = 24 * time.Hour const completedTorrentCatchupSettingPrefix = "download.auto_organized." +const completedTorrentNotifySettingPrefix = "download.completed_notified." + +func (d *DownloadService) downloadAutoOrganizeEnabled(ctx context.Context) bool { + if d == nil || d.repo == nil || d.repo.Setting == nil { + return false + } + if v, err := d.repo.Setting.Get(ctx, "organizer.auto_after_download"); err == nil && parseBoolSetting(v, false) { + return true + } + if v, err := d.repo.Setting.Get(ctx, "organize.auto"); err == nil && parseBoolSetting(v, false) { + return true + } + return false +} // recentlyCompletedTorrent 报告该种子是否在补整理时间窗内完成。 // qBittorrent 未提供 completion_on 时保守地返回 false。 @@ -1111,6 +1127,46 @@ func completedTorrentCatchupSettingKey(torrent QBitTorrent) string { return completedTorrentCatchupSettingPrefix + fmt.Sprintf("%x", sum[:]) } +func (d *DownloadService) completedTorrentNotified(ctx context.Context, torrent QBitTorrent) bool { + if d == nil || d.repo == nil || d.repo.Setting == nil { + return false + } + key := completedTorrentNotifySettingKey(torrent) + if key == "" { + return false + } + value, err := d.repo.Setting.Get(ctx, key) + if err != nil { + return false + } + return parseBoolSetting(value, false) +} + +func (d *DownloadService) markCompletedTorrentNotified(ctx context.Context, torrent QBitTorrent) { + if d == nil || d.repo == nil || d.repo.Setting == nil { + return + } + key := completedTorrentNotifySettingKey(torrent) + if key == "" { + return + } + if err := d.repo.Setting.Set(ctx, key, "true"); err != nil && d.log != nil { + d.log.Debug("mark completed torrent notification failed", + zap.String("hash", torrent.Hash), + zap.String("name", torrent.Name), + zap.Error(err)) + } +} + +func completedTorrentNotifySettingKey(torrent QBitTorrent) string { + key := completedTorrentQueueKey(torrent) + if key == "" { + return "" + } + sum := sha1.Sum([]byte(key)) + return completedTorrentNotifySettingPrefix + fmt.Sprintf("%x", sum[:]) +} + func completedTorrentQueueKey(torrent QBitTorrent) string { hash := strings.ToLower(strings.TrimSpace(torrent.Hash)) if hash != "" { @@ -1221,22 +1277,15 @@ func downloadTaskNeedsCompletion(task model.DownloadTask) bool { // Media rows is too late for freshly-downloaded files: they usually have not // been scanned into the library yet. func (d *DownloadService) onTorrentComplete(ctx context.Context, torrent QBitTorrent) { - d.notifyDownloadComplete(torrent) + taskRow, hasTask := d.completedTorrentTask(ctx, torrent) + d.notifyDownloadComplete(ctx, torrent, taskRow) if d.organizer == nil { return } // 仅当显式开启 organizer.auto_after_download / organize.auto 时才在下载完成后整理。 // 之前的代码错误地把 organizer.smart_classify 也当成"自动整理"开关， // 让操作员只想启用"分类子目录"就被动触发了文件 move。 - autoOrganize := false - if v, err := d.repo.Setting.Get(ctx, "organizer.auto_after_download"); err == nil { - autoOrganize = v == "true" || v == "1" || v == "on" - } - if !autoOrganize { - if v, err := d.repo.Setting.Get(ctx, "organize.auto"); err == nil { - autoOrganize = v == "true" || v == "1" || v == "on" - } - } + autoOrganize := d.downloadAutoOrganizeEnabled(ctx) if !autoOrganize { d.log.Info("download completed, auto-organize disabled", zap.String("hash", torrent.Hash)) return @@ -1250,7 +1299,6 @@ func (d *DownloadService) onTorrentComplete(ctx context.Context, torrent QBitTor zap.String("content_path", torrent.ContentPath)) return } - taskRow, hasTask := d.completedTorrentTask(ctx, torrent) allowReplace := hasTask && taskRow.AllowExistingLibrary d.log.Info("download completed, triggering directory organize", zap.String("hash", torrent.Hash), @@ -1289,22 +1337,49 @@ func (d *DownloadService) onTorrentComplete(ctx context.Context, torrent QBitTor zap.Int("errors", len(res.Errors))) } -func (d *DownloadService) notifyDownloadComplete(torrent QBitTorrent) { +func (d *DownloadService) notifyDownloadComplete(ctx context.Context, torrent QBitTorrent, task *model.DownloadTask) { if d == nil || d.notify == nil { return } + if d.completedTorrentNotified(ctx, torrent) { + return + } + d.markCompletedTorrentNotified(ctx, torrent) name := strings.TrimSpace(torrent.Name) if name == "" { name = strings.TrimSpace(filepath.Base(torrent.ContentPath)) } + if task != nil && strings.TrimSpace(task.Title) != "" { + name = strings.TrimSpace(task.Title) + } if name == "" { name = "下载任务" } body := fmt.Sprintf("任务：%s\n保存路径：%s\nHash：%s", name, firstNonEmpty(torrent.ContentPath, torrent.SavePath), torrent.Hash) + data := map[string]interface{}{} + if task != nil { + if strings.TrimSpace(task.PosterURL) != "" { + data["poster_url"] = task.PosterURL + } + if strings.TrimSpace(task.BackdropURL) != "" { + data["backdrop_url"] = task.BackdropURL + } + if strings.TrimSpace(task.MediaType) != "" { + data["media_type"] = task.MediaType + } + if strings.TrimSpace(task.MediaCategory) != "" { + data["media_category"] = task.MediaCategory + } + } go func() { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() - d.notify.Broadcast(ctx, "MediaStationGo 下载完成", body, EventDownloadComplete) + d.notify.BroadcastEvent(ctx, NotifyEvent{ + Type: EventDownloadComplete, + Title: "MediaStationGo 下载完成", + Message: body, + Data: data, + }) }() } diff --git a/internal/service/downloads_test.go b/internal/service/downloads_test.go index e8d9f11..9529121 100644 --- a/internal/service/downloads_test.go +++ b/internal/service/downloads_test.go @@ -145,6 +145,9 @@ func TestProcessDownloadSnapshotQueuesCompletedPendingTaskOnFirstSnapshot(t *tes if err := repos.Download.Create(t.Context(), task); err != nil { t.Fatal(err) } + if err := repos.Setting.Set(t.Context(), "organizer.auto_after_download", "true"); err != nil { + t.Fatal(err) + } svc := NewDownloadService(zap.NewNop(), repos, NewHub(zap.NewNop()), nil) svc.processDownloadSnapshot(t.Context(), []QBitTorrent{{ @@ -444,6 +447,9 @@ func TestDownloadPollCatchesUpRecentlyCompletedTorrents(t *testing.T) { if err := repos.Download.Create(t.Context(), task); err != nil { t.Fatal(err) } + if err := repos.Setting.Set(t.Context(), "organizer.auto_after_download", "true"); err != nil { + t.Fatal(err) + } svc := NewDownloadService(zap.NewNop(), repos, NewHub(zap.NewNop()), nil) svc.processDownloadSnapshot(t.Context(), []QBitTorrent{ @@ -458,6 +464,38 @@ func TestDownloadPollCatchesUpRecentlyCompletedTorrents(t *testing.T) { } } +func TestDownloadPollDoesNotCatchUpWhenAutoOrganizeDisabled(t *testing.T) { + repos := newOrganizerTestRepo(t) + if err := repos.DB.AutoMigrate(&model.DownloadTask{}); err != nil { + t.Fatal(err) + } + task := &model.DownloadTask{ + Source: "qbittorrent", + URL: "magnet:?xt=urn:btih:fresh", + Title: "Fresh Complete S01E01", + SavePath: "/downloads", + Status: "queued", + Progress: 0, + } + if err := repos.Download.Create(t.Context(), task); err != nil { + t.Fatal(err) + } + svc := NewDownloadService(zap.NewNop(), repos, NewHub(zap.NewNop()), nil) + torrent := QBitTorrent{ + Hash: "fresh-complete", + Name: "Fresh Complete S01E01", + Progress: 1, + CompletionOn: time.Now().Add(-time.Hour).Unix(), + } + + svc.processDownloadSnapshot(t.Context(), []QBitTorrent{torrent}, tasksByTorrentIdentity([]model.DownloadTask{*task})) + svc.processDownloadSnapshot(t.Context(), []QBitTorrent{torrent}, tasksByTorrentIdentity([]model.DownloadTask{*task})) + + if got := len(svc.organizeQueue); got != 0 { + t.Fatalf("auto-organize disabled queued %d completed jobs, want 0", got) + } +} + func TestDownloadPollSkipsRecordedCompletedTorrentCatchup(t *testing.T) { repos := newOrganizerTestRepo(t) torrent := QBitTorrent{ diff --git a/internal/service/notify_channels.go b/internal/service/notify_channels.go index 8ba3e78..46e93e0 100644 --- a/internal/service/notify_channels.go +++ b/internal/service/notify_channels.go @@ -192,16 +192,27 @@ const ( // NotifyEventNone sentinel means the channel stays enabled but receives no // event push. func (s *NotifyChannelService) Broadcast(ctx context.Context, title, body, event string) { + s.BroadcastEvent(ctx, NotifyEvent{ + Type: event, + Title: title, + Message: body, + }) +} + +// BroadcastEvent sends one structured event to every subscribed enabled +// channel. Rich channels such as Telegram can use Data fields for artwork and +// cleaner formatting while simpler channels keep receiving title/body text. +func (s *NotifyChannelService) BroadcastEvent(ctx context.Context, event NotifyEvent) { rows, err := s.repo.NotifyChannel.ListEnabled(ctx) if err != nil { s.log.Warn("notify list failed", zap.Error(err)) return } for _, r := range rows { - if !channelSubscribes(r, event) { + if !channelSubscribes(r, event.Type) { continue } - if err := s.dispatchOne(ctx, r, title, body); err != nil { + if err := s.dispatchOneEvent(ctx, r, event); err != nil { s.log.Warn("notify dispatch failed", zap.String("channel", r.Name), zap.Error(err)) } } @@ -237,8 +248,14 @@ func channelSubscribes(n model.NotifyChannel, event string) bool { // dispatchOne is the inner dispatcher; the channel type drives which // HTTP request gets built. func (s *NotifyChannelService) dispatchOne(ctx context.Context, n model.NotifyChannel, title, body string) error { + return s.dispatchOneEvent(ctx, n, NotifyEvent{Title: title, Message: body}) +} + +func (s *NotifyChannelService) dispatchOneEvent(ctx context.Context, n model.NotifyChannel, event NotifyEvent) error { cfg := map[string]any{} _ = json.Unmarshal([]byte(n.Config), &cfg) + title := event.Title + body := event.Message switch n.Type { case "telegram": @@ -248,9 +265,22 @@ func (s *NotifyChannelService) dispatchOne(ctx context.Context, n model.NotifyCh if token == "" || len(chats) == 0 { return errors.New("telegram missing bot_token / group_chat_id / channel_chat_id") } - text := fmt.Sprintf("%s\n\n%s", escapeHTML(title), escapeHTML(body)) + text := formatTelegramNotification(event) + photoURL := telegramEventPhotoURL(event) var firstErr error for _, chat := range chats { + if photoURL != "" && len(text) <= 1024 { + form := url.Values{} + form.Set("chat_id", chat) + form.Set("photo", photoURL) + form.Set("caption", text) + form.Set("parse_mode", "HTML") + if err := telegramPostForm(ctx, telegramCfg, "sendPhoto", form, 15*time.Second); err == nil { + continue + } else if firstErr == nil { + firstErr = err + } + } form := url.Values{} form.Set("chat_id", chat) form.Set("text", text) diff --git a/internal/service/notify_channels_test.go b/internal/service/notify_channels_test.go index 2580f9f..aa267cc 100644 --- a/internal/service/notify_channels_test.go +++ b/internal/service/notify_channels_test.go @@ -1,9 +1,13 @@ package service import ( + "net/http" + "net/http/httptest" + "strings" "testing" "github.com/ShukeBta/MediaStationGo/internal/model" + "go.uber.org/zap" ) func TestChannelSubscribesCanDisableAllEvents(t *testing.T) { @@ -37,3 +41,53 @@ func TestChannelSubscribesSupportsExplicitAllAndSpecificEvents(t *testing.T) { t.Fatal("unlisted event should not be subscribed") } } + +func TestTelegramDispatchUsesPhotoAndFormattedCaption(t *testing.T) { + var gotPath string + var gotForm map[string]string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + if err := r.ParseForm(); err != nil { + t.Fatalf("parse form: %v", err) + } + gotForm = map[string]string{} + for key := range r.Form { + gotForm[key] = r.Form.Get(key) + } + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer server.Close() + + svc := NewNotifyChannelService(zap.NewNop(), nil) + channel := model.NotifyChannel{ + Type: "telegram", + Config: `{ + "bot_token":"123456:ABC", + "group_chat_id":"-10001", + "api_base_url":"` + server.URL + `" + }`, + } + err := svc.dispatchOneEvent(t.Context(), channel, NotifyEvent{ + Type: EventDownloadComplete, + Title: "MediaStationGo 下载完成", + Message: "任务：示例影片\n保存路径：/downloads/movie\nHash：abcdef", + Data: map[string]interface{}{ + "poster_url": server.URL + "/poster.jpg", + }, + }) + if err != nil { + t.Fatal(err) + } + if gotPath != "/bot123456:ABC/sendPhoto" { + t.Fatalf("path = %q, want sendPhoto", gotPath) + } + if gotForm["chat_id"] != "-10001" || gotForm["photo"] == "" { + t.Fatalf("telegram form = %#v", gotForm) + } + caption := gotForm["caption"] + for _, want := range []string{"MediaStationGo", "下载完成", "任务: 示例影片", "/downloads/movie", "abcdef"} { + if !strings.Contains(caption, want) { + t.Fatalf("caption missing %q: %s", want, caption) + } + } +} diff --git a/internal/service/notify_telegram.go b/internal/service/notify_telegram.go index 5847c8e..2c4c55e 100644 --- a/internal/service/notify_telegram.go +++ b/internal/service/notify_telegram.go @@ -4,6 +4,8 @@ package service import ( "context" "fmt" + "net/url" + "sort" "strings" "time" ) @@ -25,9 +27,23 @@ func (p *TelegramProvider) Send(ctx context.Context, cfg map[string]string, even } text := formatTelegramMessage(event, parseMode) + photoURL := telegramEventPhotoURL(event) var firstErr error for _, chatID := range chatIDs { + if photoURL != "" && len(text) <= 1024 { + payload := map[string]string{ + "chat_id": chatID, + "photo": photoURL, + "caption": text, + "parse_mode": parseMode, + } + if err := telegramPostJSON(ctx, cfg, "sendPhoto", payload, 15*time.Second); err == nil { + continue + } else if firstErr == nil { + firstErr = err + } + } payload := map[string]string{ "chat_id": chatID, "text": text, @@ -53,27 +69,196 @@ func (p *TelegramProvider) ValidateConfig(cfg map[string]string) error { // formatTelegramMessage 格式化消息内容。 func formatTelegramMessage(event NotifyEvent, parseMode string) string { - var sb strings.Builder - sb.WriteString(fmt.Sprintf("%s\n\n", escapeHTML(event.Title))) - sb.WriteString(escapeHTML(event.Message)) + text := formatTelegramNotification(event) + if parseMode == "HTML" || parseMode == "" { + return text + } + result := text + result = strings.ReplaceAll(result, "", "**") + result = strings.ReplaceAll(result, "", "**") + result = strings.ReplaceAll(result, "", "`") + result = strings.ReplaceAll(result, "", "`") + result = strings.ReplaceAll(result, "<", "<") + result = strings.ReplaceAll(result, ">", ">") + result = strings.ReplaceAll(result, "&", "&") + return result +} - if len(event.Data) > 0 { +func formatTelegramNotification(event NotifyEvent) string { + var sb strings.Builder + title := strings.TrimSpace(event.Title) + if strings.HasPrefix(title, "MediaStationGo ") { + sb.WriteString("MediaStationGo\n") + sb.WriteString(fmt.Sprintf("%s", escapeHTML(strings.TrimSpace(strings.TrimPrefix(title, "MediaStationGo "))))) + } else if title != "" { + sb.WriteString(fmt.Sprintf("%s", escapeHTML(title))) + } else { + sb.WriteString("MediaStationGo") + } + + message := strings.TrimSpace(event.Message) + if message != "" { sb.WriteString("\n\n") - for k, v := range event.Data { - sb.WriteString(fmt.Sprintf("• %s: %v\n", escapeHTML(k), v)) + sb.WriteString(formatTelegramBody(message)) + } + + fields := telegramDisplayData(event.Data) + if len(fields) > 0 { + if message != "" { + sb.WriteString("\n") + } + sb.WriteString("\n") + for _, field := range fields { + sb.WriteString(formatTelegramField(field.key, field.value)) + sb.WriteString("\n") } } - if parseMode != "HTML" { - // Markdown 模式 - result := sb.String() - result = strings.ReplaceAll(result, "", "**") - result = strings.ReplaceAll(result, "", "**") - result = strings.ReplaceAll(result, "<", "<") - result = strings.ReplaceAll(result, ">", ">") - result = strings.ReplaceAll(result, "&", "&") - return result - } - - return sb.String() + return strings.TrimSpace(sb.String()) +} + +func formatTelegramBody(message string) string { + lines := strings.Split(message, "\n") + out := make([]string, 0, len(lines)) + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" { + out = append(out, "") + continue + } + if strings.HasPrefix(line, "- ") { + out = append(out, "- "+escapeHTML(strings.TrimSpace(strings.TrimPrefix(line, "- ")))) + continue + } + if key, value, ok := splitTelegramField(line); ok { + out = append(out, formatTelegramField(key, value)) + continue + } + out = append(out, escapeHTML(line)) + } + return strings.TrimSpace(strings.Join(out, "\n")) +} + +func splitTelegramField(line string) (string, string, bool) { + idx := strings.Index(line, "：") + sepLen := len("：") + if idx < 0 { + idx = strings.Index(line, ":") + sepLen = len(":") + } + if idx <= 0 { + return "", "", false + } + key := strings.TrimSpace(line[:idx]) + value := strings.TrimSpace(line[idx+sepLen:]) + if key == "" || value == "" || len([]rune(key)) > 16 { + return "", "", false + } + return key, value, true +} + +func formatTelegramField(key, value string) string { + key = telegramFieldLabel(key) + escapedValue := escapeHTML(strings.TrimSpace(value)) + if telegramCodeField(key) { + escapedValue = "" + escapedValue + "" + } + return fmt.Sprintf("%s: %s", escapeHTML(key), escapedValue) +} + +func telegramCodeField(key string) bool { + key = strings.ToLower(strings.TrimSpace(key)) + return strings.Contains(key, "hash") || + strings.Contains(key, "路径") || + strings.Contains(key, "path") || + strings.Contains(key, "id") +} + +type telegramDataField struct { + key string + value string +} + +func telegramDisplayData(data map[string]interface{}) []telegramDataField { + if len(data) == 0 { + return nil + } + keys := make([]string, 0, len(data)) + for key := range data { + if telegramHiddenDataKey(key) { + continue + } + keys = append(keys, key) + } + sort.Strings(keys) + fields := make([]telegramDataField, 0, len(keys)) + for _, key := range keys { + value := strings.TrimSpace(fmt.Sprint(data[key])) + if value == "" || value == "" { + continue + } + fields = append(fields, telegramDataField{key: key, value: value}) + } + return fields +} + +func telegramHiddenDataKey(key string) bool { + switch strings.ToLower(strings.TrimSpace(key)) { + case "photo_url", "poster_url", "poster", "image_url", "backdrop_url": + return true + default: + return false + } +} + +func telegramFieldLabel(key string) string { + switch strings.ToLower(strings.TrimSpace(key)) { + case "title", "name": + return "标题" + case "save_path": + return "保存路径" + case "hash": + return "Hash" + case "media_type": + return "媒体类型" + case "media_category": + return "分类" + case "subscription": + return "订阅" + case "queued": + return "新增资源" + default: + return strings.TrimSpace(key) + } +} + +func telegramEventPhotoURL(event NotifyEvent) string { + for _, key := range []string{"photo_url", "poster_url", "poster", "image_url", "backdrop_url"} { + value := telegramDataString(event.Data, key) + if isTelegramRemotePhotoURL(value) { + return value + } + } + return "" +} + +func telegramDataString(data map[string]interface{}, key string) string { + if len(data) == 0 { + return "" + } + for k, value := range data { + if strings.EqualFold(strings.TrimSpace(k), key) { + return strings.TrimSpace(fmt.Sprint(value)) + } + } + return "" +} + +func isTelegramRemotePhotoURL(raw string) bool { + raw = strings.TrimSpace(raw) + if raw == "" { + return false + } + u, err := url.Parse(raw) + return err == nil && (u.Scheme == "http" || u.Scheme == "https") && u.Host != "" } diff --git a/internal/service/subscription.go b/internal/service/subscription.go index 09aeaa9..349a951 100644 --- a/internal/service/subscription.go +++ b/internal/service/subscription.go @@ -421,7 +421,25 @@ func (s *SubscriptionService) notifySubscriptionHit(sub *model.Subscription, que go func() { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() - s.notify.Broadcast(ctx, "MediaStationGo 订阅命中新资源", body, EventSubscriptionHit) + data := map[string]interface{}{} + if strings.TrimSpace(sub.PosterURL) != "" { + data["poster_url"] = sub.PosterURL + } + if strings.TrimSpace(sub.BackdropURL) != "" { + data["backdrop_url"] = sub.BackdropURL + } + if strings.TrimSpace(sub.MediaType) != "" { + data["media_type"] = sub.MediaType + } + if strings.TrimSpace(sub.MediaCategory) != "" { + data["media_category"] = sub.MediaCategory + } + s.notify.BroadcastEvent(ctx, NotifyEvent{ + Type: EventSubscriptionHit, + Title: "MediaStationGo 订阅命中新资源", + Message: body, + Data: data, + }) }() }