rogee/tg_exporter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix group msg

Browse Source
This commit is contained in:
Rogee
2024-09-03 09:43:21 +08:00
parent ba16785b43
commit df2ca18f19
5 changed files with 138 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
internal/channel_message.go
View File

@@ -13,15 +13,18 @@ type ChannelMessage struct {
ID int
GroupID int64
Message string
Medias []ChannelMessageMedia
Medias ChannelMessageMedia
PublishAt time.Time
}
type ChannelMessageMedia struct {
Photo *string
Video *string
Document *ChannelMessageDocument
WebPage *ChannelMessageMediaWebPage
MsgID int `json:"msg_id,omitempty"`
AssetID int64 `json:"asset_id,omitempty"`
Photo *string `json:"photo,omitempty"`
Video *string `json:"video,omitempty"`
Document *ChannelMessageDocument `json:"document,omitempty"`
WebPage *ChannelMessageMediaWebPage `json:"web_page,omitempty"`
}
type ChannelMessageDocument struct {
@@ -57,31 +60,37 @@ func (c *ChannelMessage) WithMessage(message string) *ChannelMessage {
}
func (c *ChannelMessage) WithPhoto(assetID int64, ext string) *ChannelMessage {
c.Medias = append(c.Medias, ChannelMessageMedia{
Photo: lo.ToPtr(fmt.Sprintf("%d.%s", assetID, strings.Trim(ext, "."))),
})
c.Medias = ChannelMessageMedia{
MsgID: c.ID,
AssetID: assetID,
Photo: lo.ToPtr(fmt.Sprintf("%d.%s", assetID, strings.Trim(ext, "."))),
}
return c
}
func (c *ChannelMessage) WithVideo(video string) *ChannelMessage {
c.Medias = append(c.Medias, ChannelMessageMedia{Video: lo.ToPtr(video)})
return c
}
func (c *ChannelMessage) WithDocument(d ChannelMessageDocument) *ChannelMessage {
c.Medias = append(c.Medias, ChannelMessageMedia{
func (c *ChannelMessage) WithDocument(assetID int64, d ChannelMessageDocument) *ChannelMessage {
c.Medias = ChannelMessageMedia{
MsgID: c.ID,
AssetID: assetID,
Document: lo.ToPtr(d),
})
}
return c
}
func (c *ChannelMessage) WithWebPage(title, url string) *ChannelMessage {
c.Medias = append(c.Medias, ChannelMessageMedia{
func (c *ChannelMessage) WithWebPage(assetID int64, title, url string) *ChannelMessage {
c.Medias = ChannelMessageMedia{
MsgID: c.ID,
AssetID: assetID,
WebPage: lo.ToPtr(ChannelMessageMediaWebPage{Title: title, URL: url}),
})
}
return c
}
func (c *ChannelMessage) GetMedias() string {
b, _ := json.Marshal([]ChannelMessageMedia{c.Medias})
return string(b)
}
func (c *ChannelMessage) GetMedia() string {
b, _ := json.Marshal(c.Medias)
return string(b)

4
internal/client_channel.go
View File

@@ -87,11 +87,11 @@ func (t *TClient) Channel(ctx context.Context, channel *tg.Channel, cfg *DBChann
logger.Error("save document failed", zap.Error(err))
return err
}
channelMessage.WithDocument(data)
channelMessage.WithDocument(doc.GetID(), data)
}
case *tg.MessageMediaWebPage:
if page, ok := mediaClass.(*tg.MessageMediaWebPage).GetWebpage().(*tg.WebPage); ok {
channelMessage.WithWebPage(page.Title, page.URL)
channelMessage.WithWebPage(page.GetID(), page.Title, page.URL)
} else {
logger.Warn("web_page", zap.String("url", mediaClass.(*tg.MessageMediaWebPage).GetWebpage().String()))
}

48
internal/db_channel.go
View File

@@ -119,21 +119,55 @@ func (c *DBChannel) SaveMessage(ctx context.Context, msg *ChannelMessage) error
GroupID: msg.GroupID,
UUID: int64(msg.ID),
Content: lo.ToPtr(msg.Message),
Media: msg.GetMedia(),
Media: msg.GetMedias(),
PublishedAt: msg.PublishAt,
CreatedAt: time.Now(),
}
tbl := table.ChannelMessages
_, err := tbl.INSERT(tbl.AllColumns.Except(tbl.ID)).MODEL(message).ExecContext(ctx, db)
var m model.ChannelMessages
err := tbl.
SELECT(tbl.ID).
WHERE(
tbl.GroupID.EQ(Int(message.GroupID)).AND(
tbl.UUID.EQ(Int64(message.UUID)),
),
).
LIMIT(1).
QueryContext(ctx, db, &m)
if err != nil {
if e, ok := err.(*pq.Error); ok {
if e.Code == "23505" {
return nil
// 如果没有找到记录，那么插入新记录
if errors.Is(err, qrm.ErrNoRows) {
_, err = tbl.INSERT(tbl.AllColumns.Except(tbl.ID)).MODEL(message).ExecContext(ctx, db)
if err != nil {
if e, ok := err.(*pq.Error); ok {
if e.Code == "23505" {
return nil
}
}
return errors.Wrap(err, "insert message")
}
return nil
}
return errors.Wrap(err, "insert message")
return errors.Wrap(err, "select message")
}
return nil
// 如果找到记录，那么更新记录
stmt := tbl.UPDATE().SET(
tbl.Content.SET(RawString(`CONCAT(content, #var)`, RawArgs{
"#var": *message.Content,
})),
tbl.Media.SET(RawString(`media || #var::jsonb`, RawArgs{
"#var": msg.GetMedia(),
})),
).WHERE(
tbl.GroupID.EQ(Int(message.GroupID)).AND(
tbl.UUID.EQ(Int(message.UUID)),
),
)
_, err = db.ExecContext(ctx, stmt.DebugSql())
return err
}

50
internal/db_channel_test.go Normal file
View File

@@ -0,0 +1,50 @@
package internal
import (
"context"
"testing"
"time"
"github.com/samber/lo"
)
func TestDBChannel_SaveMessage(t *testing.T) {
dsn := "postgresql://postgres:xixi0202@10.1.1.3:5432/telegram_resource?sslmode=disable"
if err := InitDB(dsn); err != nil {
t.Error(err)
}
db.Exec(`truncate channel_messages`)
msg := &ChannelMessage{
ID: 1,
GroupID: 1,
Message: "Hello",
Medias: ChannelMessageMedia{
MsgID: 1,
AssetID: 1,
Photo: lo.ToPtr("photo"),
},
PublishAt: time.Now(),
}
msg1 := &ChannelMessage{
ID: 1,
GroupID: 1,
Message: "Hello",
Medias: ChannelMessageMedia{
MsgID: 2,
AssetID: 3,
Photo: lo.ToPtr("Man"),
},
PublishAt: time.Now(),
}
c := NewDBChannel(123, "test", "hello")
if err := c.SaveMessage(context.Background(), msg); err != nil {
t.Error(err)
}
if err := c.SaveMessage(context.Background(), msg1); err != nil {
t.Logf("%+v", err)
}
}

17
main_test.go
View File

@@ -4,7 +4,10 @@ import (
"mime"
"testing"
"exporter/database/telegram_resource/public/table"
"github.com/dustin/go-humanize"
. "github.com/go-jet/jet/v2/postgres"
"github.com/spf13/viper"
)
@@ -27,3 +30,17 @@ func Test_Size(t *testing.T) {
t.Logf("Size: %d", s)
t.Logf("Vize: %d", b)
}
func Test_Sql(t *testing.T) {
tbl := table.ChannelMessages
stmt := tbl.UPDATE().SET(
tbl.Content.SET(RawString(`CONCAT(content, $var)`, RawArgs{"$var": "hello"})),
tbl.Media.SET(RawString(`media || $var::jsonb`, RawArgs{"$var": `{"Rogee": "Hello"}`})),
).WHERE(
tbl.GroupID.EQ(Int(1)).AND(
tbl.UUID.EQ(Int64(1)),
),
)
t.Log(stmt.DebugSql())
}