Improve Telegram notifications

This commit is contained in:
yebuwudong
2026-06-18 00:05:38 +08:00
committed by ShukeBta
parent e4b1488e11
commit 692b7fb9df
6 changed files with 435 additions and 35 deletions
+88 -13
View File
@@ -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,
})
}()
}
+38
View File
@@ -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{
+33 -3
View File
@@ -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("<b>%s</b>\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)
+54
View File
@@ -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{"<b>MediaStationGo</b>", "<b>下载完成</b>", "<b>任务</b>: 示例影片", "<code>/downloads/movie</code>", "<code>abcdef</code>"} {
if !strings.Contains(caption, want) {
t.Fatalf("caption missing %q: %s", want, caption)
}
}
}
+203 -18
View File
@@ -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("<b>%s</b>\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, "<b>", "**")
result = strings.ReplaceAll(result, "</b>", "**")
result = strings.ReplaceAll(result, "<code>", "`")
result = strings.ReplaceAll(result, "</code>", "`")
result = strings.ReplaceAll(result, "&lt;", "<")
result = strings.ReplaceAll(result, "&gt;", ">")
result = strings.ReplaceAll(result, "&amp;", "&")
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("<b>MediaStationGo</b>\n")
sb.WriteString(fmt.Sprintf("<b>%s</b>", escapeHTML(strings.TrimSpace(strings.TrimPrefix(title, "MediaStationGo ")))))
} else if title != "" {
sb.WriteString(fmt.Sprintf("<b>%s</b>", escapeHTML(title)))
} else {
sb.WriteString("<b>MediaStationGo</b>")
}
message := strings.TrimSpace(event.Message)
if message != "" {
sb.WriteString("\n\n")
for k, v := range event.Data {
sb.WriteString(fmt.Sprintf("• <b>%s</b>: %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, "<b>", "**")
result = strings.ReplaceAll(result, "</b>", "**")
result = strings.ReplaceAll(result, "&lt;", "<")
result = strings.ReplaceAll(result, "&gt;", ">")
result = strings.ReplaceAll(result, "&amp;", "&")
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 = "<code>" + escapedValue + "</code>"
}
return fmt.Sprintf("<b>%s</b>: %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 == "<nil>" {
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 != ""
}
+19 -1
View File
@@ -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,
})
}()
}