Spark/spark-server
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(squarebot): 接入媒体引导与 Bot 附件/锁

Browse Source 
完善 bootstrap 媒体装配、配置开关，并补充 squarebot 附件与锁逻辑。
This commit is contained in:
DESKTOP-L8OUL9F\gfl0034 2026-09-24 23:37:44 +08:00
parent 6cfd6309d8
commit 5fda6be8ec
13 changed files with 348 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
configs/config.yaml
View File

@ -106,7 +106,7 @@ sms:
username: "noreply@sparkmail.wyrs365.com"
password: "gD9UNvSt5ilagfa1"
from: "Spark IM <noreply@sparkmail.wyrs365.com>"
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"

45
internal/bootstrap/app.go
View File

@ -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())

16
internal/bootstrap/config.go
View File

@ -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) {

21
internal/bootstrap/media.go Normal file
View File

@ -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(),
}
}

2
internal/infrastructure/media/image.go
View File

@ -8,7 +8,7 @@ import (
"image"
_ "image/gif"
"image/jpeg"
_ "image/png"
"image/png"
"strings"
"golang.org/x/image/draw"

11
internal/module/square/logic/bot_publish.go
View File

@ -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(),

2
internal/module/square/logic/service.go
View File

@ -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)

0
internal/module/squarebot/logic/.keep
View File

90
internal/module/squarebot/logic/attach.go Normal file
View File

@ -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"
}
}

30
internal/module/squarebot/logic/lock.go Normal file
View File

@ -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()
}

103
internal/module/squarebot/logic/rss.go
View File

@ -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 <img>
}
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)<img[^>]+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 {

23
internal/module/squarebot/logic/service.go
View File

@ -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
}

6
internal/module/squarebot/logic/service_test.go
View File

@ -24,7 +24,8 @@ func TestParseRSSAndFormat(t *testing.T) {
<item>
<title>Hello &amp; World</title>
<link>https://example.com/a</link>
<description>&lt;p&gt;Short summary here.&lt;/p&gt;</description>
<description>&lt;p&gt;Short summary here.&lt;/p&gt;&lt;img src="https://example.com/pic.jpg"/&gt;</description>
<enclosure url="https://example.com/cover.jpg" type="image/jpeg"/>
</item>
</channel></rss>`)
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)