diff --git a/configs/config.yaml b/configs/config.yaml index 31fb1e0..8470e0c 100644 --- a/configs/config.yaml +++ b/configs/config.yaml @@ -106,7 +106,7 @@ sms: username: "noreply@sparkmail.wyrs365.com" password: "gD9UNvSt5ilagfa1" from: "Spark IM " - aliyun_sms: + aliyun_sms: access_key_id: "" access_key_secret: "" sign_name: "" @@ -119,3 +119,33 @@ sms: mobile: "" unicom: "" telecom: "" + +# 用户上传：检测/校验（默认仅日志；user_reject=true 则拒凭证/发帖）。主压缩仍在 App。 +# 机器人媒体：bot_compress=true 时入库前压到图≤1080 / 视频≤720（视频需 ffmpeg）。 +media: + enabled: true + user_reject: false + bot_compress: true + image_max_edge: 1080 + video_max_edge: 720 + avatar_max_edge: 200 + image_max_bytes: 10485760 + video_max_bytes: 104857600 + avatar_max_bytes: 2097152 + jpeg_quality: 85 + video_crf: 23 + ffmpeg_path: "ffmpeg" + +# 广场机器人：每 interval 随机 bot 发一条海外主流媒体公开 RSS（标题+短摘要+原文链接）。 +# 需先跑 cmd/seed 确保 bot001@spark.bot … 存在。机器人帖 ttl 后自动清理。 +square_bot: + enabled: true + interval: 5m + ttl: 360h + bot_indexes: [] + feeds: + - "https://feeds.bbci.co.uk/news/world/rss.xml" + - "https://www.theguardian.com/world/rss" + - "https://feeds.npr.org/1004/rss.xml" + - "https://www.aljazeera.com/xml/rss/all.xml" + - "https://rss.nytimes.com/services/xml/rss/nyt/World.xml" diff --git a/internal/bootstrap/app.go b/internal/bootstrap/app.go index cd36106..7ce0b78 100644 --- a/internal/bootstrap/app.go +++ b/internal/bootstrap/app.go @@ -48,6 +48,7 @@ import ( smslogic "spark-server/internal/module/sms/logic" squarelogic "spark-server/internal/module/square/logic" squaremodel "spark-server/internal/module/square/model" + squarebotlogic "spark-server/internal/module/squarebot/logic" synclogic "spark-server/internal/module/sync/logic" syncmodel "spark-server/internal/module/sync/model" uploadlogic "spark-server/internal/module/upload/logic" @@ -71,6 +72,7 @@ type App struct { idgen *id.Generator sessions session.Manager deliver *deliverlogic.Worker + squareBot *squarebotlogic.Worker accepting atomic.Bool shutdownOnce sync.Once } @@ -189,10 +191,18 @@ func finishApp(cfg *Config, log *zap.Logger, logClose func() error, idgen *id.Ge friends := wireFriends(idgen, db, users, deliverQueue, chat) groups := wireGroups(idgen, db, users, friends, deliverQueue, objects, chat) messages := wireMessages(idgen, friends, groups, deliverQueue, chat) - files := wireFiles(cfg, objects, groups) + files := wireFiles(cfg, log, objects, groups) favorites := wireFavorites(idgen, db, objects, chat) moments := wireMoments(idgen, db, objects, friends, users) square := wireSquare(idgen, db, objects, friends, users) + mediaPol := MediaPolicy(cfg.Media) + if moments != nil { + moments.ConfigureMedia(mediaPol, log) + } + if square != nil { + square.ConfigureMedia(mediaPol, log) + } + app.squareBot = wireSquareBot(cfg, log, rdb, objects, square, users) music := wireMusic(cfg, db, objects) users.BindAccountCleanup(jwtSvc, sessions, accountCleaners(friends, moments, square, groups, favorites, inbox, mailbox, chat)...) adminSvc, adminTokens := wireAdmin(cfg, db, groups, idgen) @@ -325,7 +335,7 @@ func wireGroups(idgen *id.Generator, db *mysql.DB, users *userlogic.Service, fri return grouplogic.NewService(repo, users, friends, idgen, chat.msgs, q, objects) } -func wireFiles(cfg *Config, store *objectstore.Store, groups *grouplogic.Service) *uploadlogic.Service { +func wireFiles(cfg *Config, log *zap.Logger, store *objectstore.Store, groups *grouplogic.Service) *uploadlogic.Service { if store == nil || !store.Ready() { return nil } @@ -333,9 +343,33 @@ func wireFiles(cfg *Config, store *objectstore.Store, groups *grouplogic.Service TTL: cfg.ObjectStorage.PresignTTL, Limit: ratelimit.New(cfg.RateLimit.UploadPerSecond), Groups: groups, + Media: MediaPolicy(cfg.Media), + Log: log, }) } +func wireSquareBot( + cfg *Config, log *zap.Logger, rdb *redisx.Client, store *objectstore.Store, + square *squarelogic.Service, users *userlogic.Service, +) *squarebotlogic.Worker { + if cfg == nil || !cfg.SquareBot.Enabled || square == nil || users == nil { + return nil + } + var locker squarebotlogic.Locker + if rdb != nil && rdb.Ready() { + locker = squarebotlogic.NewRedisLocker(rdb.RDB()) + } + svc := squarebotlogic.NewService(log, square, users, locker, squarebotlogic.Options{ + Enabled: true, Interval: cfg.SquareBot.Interval, TTL: cfg.SquareBot.TTL, + Feeds: cfg.SquareBot.Feeds, BotIndexes: cfg.SquareBot.BotIndexes, + }) + if cfg.Media.BotCompress && store != nil && store.Ready() { + lim, jq, crf, ff := MediaCompressOptions(cfg.Media) + svc.BindMedia(squarebotlogic.NewMediaCompressor(lim, jq, crf, ff), store, true) + } + return squarebotlogic.NewWorker(log, svc) +} + func wireFavorites(idgen *id.Generator, db *mysql.DB, store *objectstore.Store, chat chatStores) *favlogic.Service { if idgen == nil { return nil @@ -522,6 +556,10 @@ func (a *App) Run() error { if a.deliver != nil { a.deliver.Start(context.Background()) } + if a.squareBot != nil { + a.squareBot.Start() + a.log.Info("squarebot started") + } errCh := make(chan error, 2) go func() { errCh <- a.http.Start() }() go func() { errCh <- a.tcp.Start() }() @@ -588,6 +626,9 @@ func (a *App) doShutdown() error { if a.deliver != nil { a.deliver.Stop() } + if a.squareBot != nil { + a.squareBot.Stop() + } err = errors.Join(err, a.redis.Close(), a.mysql.Close()) if a.logClose != nil { err = errors.Join(err, a.logClose()) diff --git a/internal/bootstrap/config.go b/internal/bootstrap/config.go index e3500e9..2ec079e 100644 --- a/internal/bootstrap/config.go +++ b/internal/bootstrap/config.go @@ -26,6 +26,7 @@ type Config struct { Message MessageConfig `yaml:"message"` ObjectStorage ObjectStorageConfig `yaml:"object_storage"` Media MediaConfig `yaml:"media"` + SquareBot SquareBotConfig `yaml:"square_bot"` SMS SMSConfig `yaml:"sms"` } @@ -153,6 +154,15 @@ type MediaConfig struct { FFmpegPath string `yaml:"ffmpeg_path"` } +// SquareBotConfig drives the overseas media RSS publisher on the square feed. +type SquareBotConfig struct { + Enabled bool `yaml:"enabled"` + Interval time.Duration `yaml:"interval"` + TTL time.Duration `yaml:"ttl"` + Feeds []string `yaml:"feeds"` + BotIndexes []int `yaml:"bot_indexes"` +} + type MessageConfig struct { MaxContentSize int `yaml:"max_content_size"` SyncLimit int `yaml:"sync_limit"` @@ -333,6 +343,12 @@ func applyDefaults(cfg *Config) { cfg.RateLimit.UploadPerSecond = 10 } applyMediaDefaults(&cfg.Media) + if cfg.SquareBot.Interval <= 0 { + cfg.SquareBot.Interval = 5 * time.Minute + } + if cfg.SquareBot.TTL <= 0 { + cfg.SquareBot.TTL = 15 * 24 * time.Hour + } } func applyMediaDefaults(m *MediaConfig) { diff --git a/internal/bootstrap/media.go b/internal/bootstrap/media.go new file mode 100644 index 0000000..7be7d41 --- /dev/null +++ b/internal/bootstrap/media.go @@ -0,0 +1,21 @@ +package bootstrap + +import ( + "spark-server/internal/pkg/media" +) + +// MediaPolicy builds the shared user-check / bot-compress limits from config. +func MediaPolicy(cfg MediaConfig) media.Policy { + return media.Policy{ + Enabled: cfg.Enabled, + UserReject: cfg.UserReject, + Limits: media.Limits{ + ImageMaxEdge: cfg.ImageMaxEdge, + VideoMaxEdge: cfg.VideoMaxEdge, + AvatarMaxEdge: cfg.AvatarMaxEdge, + ImageMaxBytes: cfg.ImageMaxBytes, + VideoMaxBytes: cfg.VideoMaxBytes, + AvatarMaxBytes: cfg.AvatarMaxBytes, + }.Normalize(), + } +} diff --git a/internal/infrastructure/media/image.go b/internal/infrastructure/media/image.go index 211a215..a5e4075 100644 --- a/internal/infrastructure/media/image.go +++ b/internal/infrastructure/media/image.go @@ -8,7 +8,7 @@ import ( "image" _ "image/gif" "image/jpeg" - _ "image/png" + "image/png" "strings" "golang.org/x/image/draw" diff --git a/internal/module/square/logic/bot_publish.go b/internal/module/square/logic/bot_publish.go index babd161..b19ac31 100644 --- a/internal/module/square/logic/bot_publish.go +++ b/internal/module/square/logic/bot_publish.go @@ -13,13 +13,15 @@ import ( ) // BotPublishInput is an internal publish from the media bot worker (no rate limit). +// Images must already be compressed and stored under perm/square/image/. type BotPublishInput struct { Text string SourceURL string TTL time.Duration + Images []*entity.Image } -// PublishBotMedia publishes a public text post as a bot user with TTL and URL dedupe. +// PublishBotMedia publishes a public post as a bot user with TTL and URL dedupe. func (s *Service) PublishBotMedia(ctx context.Context, userID int64, in *BotPublishInput) (*View, error) { if !s.ready() || userID <= 0 { return nil, errs.ErrUnauthorized @@ -39,7 +41,10 @@ func (s *Service) PublishBotMedia(ctx context.Context, userID int64, in *BotPubl if ttl <= 0 { ttl = 15 * 24 * time.Hour } - content := &entity.Content{} + content := &entity.Content{Images: in.Images} + if err := content.Validate(); err != nil { + return nil, errs.ErrInvalidArgument.WithMsg(i18n.T(reqctx.Lang(ctx), i18n.KeySquareContentInvalid, nil)) + } raw, err := content.Marshal() if err != nil { return nil, errs.ErrInvalidArgument.WithMsg(i18n.T(reqctx.Lang(ctx), i18n.KeySquareContentInvalid, nil)) @@ -50,7 +55,7 @@ func (s *Service) PublishBotMedia(ctx context.Context, userID int64, in *BotPubl } now := time.Now() post := &entity.Post{ - ID: id, UserID: userID, Kind: entity.KindText, Text: text, Content: raw, + ID: id, UserID: userID, Kind: content.Kind(), Text: text, Content: raw, Visibility: entity.VisibilityPublic, AuditStatus: entity.AuditApproved, Status: entity.StatusNormal, Source: entity.SourceBotRSS, SourceURL: url, ExpiresAtMS: now.Add(ttl).UnixMilli(), CreatedAtMS: now.UnixMilli(), diff --git a/internal/module/square/logic/service.go b/internal/module/square/logic/service.go index f52d083..421df91 100644 --- a/internal/module/square/logic/service.go +++ b/internal/module/square/logic/service.go @@ -14,10 +14,12 @@ import ( // Repository persists square data. Model implements it. type Repository interface { CreatePost(ctx context.Context, post *entity.Post, audience []int64) error + CreateBotPost(ctx context.Context, post *entity.Post) error GetPost(ctx context.Context, id int64) (*entity.Post, error) ListPublicFeed(ctx context.Context, cursor int64, limit int) ([]*entity.Post, error) ListByAuthor(ctx context.Context, authorID, cursor int64, limit int) ([]*entity.Post, error) SoftDeletePost(ctx context.Context, userID, postID int64) (*entity.Post, error) + SoftDeleteExpiredBotPosts(ctx context.Context, nowMS int64, limit int) ([]*entity.Post, error) ListAudience(ctx context.Context, postIDs []int64) (map[int64][]int64, error) ListLikes(ctx context.Context, postIDs []int64) (map[int64][]*entity.Like, error) AddLike(ctx context.Context, like *entity.Like) (created bool, err error) diff --git a/internal/module/squarebot/logic/.keep b/internal/module/squarebot/logic/.keep deleted file mode 100644 index e69de29..0000000 diff --git a/internal/module/squarebot/logic/attach.go b/internal/module/squarebot/logic/attach.go new file mode 100644 index 0000000..1cdbd28 --- /dev/null +++ b/internal/module/squarebot/logic/attach.go @@ -0,0 +1,90 @@ +package logic + +import ( + "context" + "strings" + "time" + + "go.uber.org/zap" + + infmedia "spark-server/internal/infrastructure/media" + squareentity "spark-server/internal/module/square/entity" + uploadentity "spark-server/internal/module/upload/entity" + "spark-server/internal/pkg/reqctx" +) + +// ObjectStore writes compressed bot assets before square publish. +type ObjectStore interface { + Put(ctx context.Context, key string, body []byte, contentType string) error + PutLarge(ctx context.Context, key string, body []byte, contentType string, maxBytes int) error +} + +// BindMedia attaches optional compressor + object store for RSS enclosure images/videos. +func (s *Service) BindMedia(comp *MediaCompressor, store ObjectStore, botCompress bool) { + if s == nil { + return + } + s.comp = comp + s.store = store + s.botCompress = botCompress +} + +// attachMedia downloads/compresses optional enclosure and returns square Image refs (best-effort). +func (s *Service) attachMedia(ctx context.Context, item FeedItem) []*squareentity.Image { + if s == nil || !s.botCompress || s.comp == nil || s.store == nil { + return nil + } + imgURL := strings.TrimSpace(item.ImageURL) + if imgURL == "" { + return nil + } + asset, err := s.comp.FetchImage(ctx, imgURL) + if err != nil { + s.log.Warn("squarebot compress image", zap.String("url", imgURL), zap.Error(err)) + return nil + } + ext := extForMIME(asset.MIME) + key := uploadentity.BuildObjectKey( + uploadentity.PurposeSquareImage, 0, reqctx.NewObjectName(), + time.Now().UTC().Format("20060102"), ext, + ) + if err := s.store.Put(ctx, key, asset.Data, asset.MIME); err != nil { + s.log.Warn("squarebot put image", zap.String("key", key), zap.Error(err)) + return nil + } + return []*squareentity.Image{{ + Key: key, Mime: asset.MIME, Width: asset.Width, Height: asset.Height, Size: asset.Size, + }} +} + +// PrepareVideoAsset compresses a remote video and uploads it (bot video paths; requires ffmpeg). +func (s *Service) PrepareVideoAsset(ctx context.Context, videoURL string) (*infmedia.BotAsset, string, error) { + if s == nil || !s.botCompress || s.comp == nil || s.store == nil { + return nil, "", nil + } + asset, err := s.comp.FetchVideo(ctx, videoURL) + if err != nil { + return nil, "", err + } + key := uploadentity.BuildObjectKey( + uploadentity.PurposeSquareVideo, 0, reqctx.NewObjectName(), + time.Now().UTC().Format("20060102"), "mp4", + ) + if err := s.store.PutLarge(ctx, key, asset.Data, asset.MIME, int(asset.Size)+1); err != nil { + return nil, "", err + } + return asset, key, nil +} + +func extForMIME(mime string) string { + switch strings.ToLower(strings.TrimSpace(mime)) { + case "image/png": + return "png" + case "image/gif": + return "gif" + case "image/webp": + return "webp" + default: + return "jpg" + } +} diff --git a/internal/module/squarebot/logic/lock.go b/internal/module/squarebot/logic/lock.go new file mode 100644 index 0000000..c557826 --- /dev/null +++ b/internal/module/squarebot/logic/lock.go @@ -0,0 +1,30 @@ +package logic + +import ( + "context" + "time" + + "github.com/redis/go-redis/v9" +) + +// RedisLocker uses SET NX EX so multi-instance servers only run one tick. +type RedisLocker struct { + rdb *redis.Client +} + +func NewRedisLocker(rdb *redis.Client) *RedisLocker { + if rdb == nil { + return nil + } + return &RedisLocker{rdb: rdb} +} + +func (l *RedisLocker) TryLock(ctx context.Context, key string, ttl time.Duration) (bool, error) { + if l == nil || l.rdb == nil { + return true, nil + } + if ttl <= 0 { + ttl = time.Minute + } + return l.rdb.SetNX(ctx, key, "1", ttl).Result() +} diff --git a/internal/module/squarebot/logic/rss.go b/internal/module/squarebot/logic/rss.go index a5bb7e8..9e90f8b 100644 --- a/internal/module/squarebot/logic/rss.go +++ b/internal/module/squarebot/logic/rss.go @@ -22,10 +22,11 @@ const ( // FeedItem is one public RSS/Atom entry (title + summary + link). type FeedItem struct { - Title string - Summary string - Link string - Source string + Title string + Summary string + Link string + Source string + ImageURL string // optional enclosure / media:content / first } type rssFeed struct { @@ -36,10 +37,24 @@ type rssFeed struct { } type rssItem struct { - Title string `xml:"title"` - Link string `xml:"link"` - Description string `xml:"description"` - GUID string `xml:"guid"` + Title string `xml:"title"` + Link string `xml:"link"` + Description string `xml:"description"` + GUID string `xml:"guid"` + Enclosure *rssEnclosure `xml:"enclosure"` + MediaContent []rssMediaContent `xml:"http://search.yahoo.com/mrss/ content"` + MediaThumb []rssMediaContent `xml:"http://search.yahoo.com/mrss/ thumbnail"` +} + +type rssEnclosure struct { + URL string `xml:"url,attr"` + Type string `xml:"type,attr"` +} + +type rssMediaContent struct { + URL string `xml:"url,attr"` + Type string `xml:"type,attr"` + Medium string `xml:"medium,attr"` } type atomFeed struct { @@ -58,9 +73,11 @@ type atomEntry struct { type atomLink struct { Href string `xml:"href,attr"` Rel string `xml:"rel,attr"` + Type string `xml:"type,attr"` } var tagRe = regexp.MustCompile(`(?s)<[^>]*>`) +var imgSrcRe = regexp.MustCompile(`(?i)]+src=["']([^"']+)["']`) // FetchFeed loads and parses a public RSS 2.0 or Atom URL. func FetchFeed(ctx context.Context, client *http.Client, feedURL string) ([]FeedItem, error) { @@ -100,10 +117,11 @@ func parseFeed(body []byte, feedURL string) ([]FeedItem, error) { for _, it := range rss.Channel.Items { link := firstNonEmpty(strings.TrimSpace(it.Link), strings.TrimSpace(it.GUID)) item := FeedItem{ - Title: cleanText(it.Title), - Summary: cleanText(it.Description), - Link: normalizeURL(link), - Source: src, + Title: cleanText(it.Title), + Summary: cleanText(it.Description), + Link: normalizeURL(link), + Source: src, + ImageURL: rssImageURL(it), } if item.Title == "" || item.Link == "" { continue @@ -127,10 +145,11 @@ func parseFeed(body []byte, feedURL string) ([]FeedItem, error) { } sum := firstNonEmpty(e.Summary, e.Content) item := FeedItem{ - Title: cleanText(e.Title), - Summary: cleanText(sum), - Link: normalizeURL(link), - Source: src, + Title: cleanText(e.Title), + Summary: cleanText(sum), + Link: normalizeURL(link), + Source: src, + ImageURL: atomImageURL(e), } if item.Title == "" || item.Link == "" { continue @@ -143,6 +162,58 @@ func parseFeed(body []byte, feedURL string) ([]FeedItem, error) { return out, nil } +func rssImageURL(it rssItem) string { + if it.Enclosure != nil && isImageMIME(it.Enclosure.Type) { + if u := normalizeURL(it.Enclosure.URL); u != "" { + return u + } + } + for _, m := range it.MediaContent { + if isImageMedia(m) { + if u := normalizeURL(m.URL); u != "" { + return u + } + } + } + for _, m := range it.MediaThumb { + if u := normalizeURL(m.URL); u != "" { + return u + } + } + return firstImgSrc(it.Description) +} + +func atomImageURL(e atomEntry) string { + for _, l := range e.Links { + if strings.EqualFold(l.Rel, "enclosure") || strings.HasPrefix(strings.ToLower(l.Type), "image/") { + if u := normalizeURL(l.Href); u != "" { + return u + } + } + } + return firstImgSrc(firstNonEmpty(e.Summary, e.Content)) +} + +func isImageMIME(t string) bool { + t = strings.ToLower(strings.TrimSpace(t)) + return strings.HasPrefix(t, "image/") +} + +func isImageMedia(m rssMediaContent) bool { + if isImageMIME(m.Type) { + return true + } + return strings.EqualFold(strings.TrimSpace(m.Medium), "image") +} + +func firstImgSrc(htmlBody string) string { + m := imgSrcRe.FindStringSubmatch(htmlBody) + if len(m) < 2 { + return "" + } + return normalizeURL(m[1]) +} + func atomBestLink(links []atomLink) string { var alt, self, any string for _, l := range links { diff --git a/internal/module/squarebot/logic/service.go b/internal/module/squarebot/logic/service.go index a6d1404..2045b50 100644 --- a/internal/module/squarebot/logic/service.go +++ b/internal/module/squarebot/logic/service.go @@ -43,14 +43,17 @@ type Locker interface { // Service runs one RSS publish + TTL cleanup tick. type Service struct { - log *zap.Logger - square Square - users Users - locker Locker - opts Options - client *http.Client - rand *rand.Rand - lockKey string + log *zap.Logger + square Square + users Users + locker Locker + opts Options + client *http.Client + rand *rand.Rand + lockKey string + comp *MediaCompressor + store ObjectStore + botCompress bool } func NewService(log *zap.Logger, square Square, users Users, locker Locker, opts Options) *Service { @@ -143,8 +146,9 @@ func (s *Service) Tick(ctx context.Context) error { return err } text := FormatPostText(item) + images := s.attachMedia(ctx, item) view, err := s.square.PublishBotMedia(ctx, bot.ID, &squarelogic.BotPublishInput{ - Text: text, SourceURL: item.Link, TTL: s.opts.TTL, + Text: text, SourceURL: item.Link, TTL: s.opts.TTL, Images: images, }) if err != nil { if errors.Is(err, errs.ErrDuplicate) { @@ -158,6 +162,7 @@ func (s *Service) Tick(ctx context.Context) error { zap.String("bot", bot.Nickname), zap.Int64("post_id", view.ID), zap.String("url", item.Link), + zap.Int("images", len(images)), ) return nil } diff --git a/internal/module/squarebot/logic/service_test.go b/internal/module/squarebot/logic/service_test.go index efb63bf..cf00801 100644 --- a/internal/module/squarebot/logic/service_test.go +++ b/internal/module/squarebot/logic/service_test.go @@ -24,7 +24,8 @@ func TestParseRSSAndFormat(t *testing.T) { Hello & World https://example.com/a -<p>Short summary here.</p> +<p>Short summary here.</p><img src="https://example.com/pic.jpg"/> + `) items, err := parseFeed(body, "https://example.com/rss") @@ -34,6 +35,9 @@ func TestParseRSSAndFormat(t *testing.T) { if items[0].Title != "Hello & World" || items[0].Link != "https://example.com/a" { t.Fatalf("%+v", items[0]) } + if items[0].ImageURL != "https://example.com/cover.jpg" { + t.Fatalf("image=%q", items[0].ImageURL) + } text := FormatPostText(items[0]) if !strings.Contains(text, "Demo News") || !strings.Contains(text, "https://example.com/a") { t.Fatalf("text=%q", text)