senlinai/agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden exploration data flow

Browse Source
This commit is contained in:
yanweidong
2026-07-23 23:29:14 +08:00
parent 400d768ecf
commit bfcfd8d32e
7 changed files with 163 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
backend/internal/crontab/dataset.go
View File

@@ -79,6 +79,11 @@ func (s *DatasetScheduler) collect(ctx context.Context) {
completed++
case "failed":
failed++
s.logger.Printf(
"dataset source collection failed: source=%s error=%s",
result.SourceIdentity,
result.Result,
)
}
}
if err != nil {

41
backend/internal/logic/dataset/feed.go
View File

@@ -152,15 +152,21 @@ type jsonFeedDocument struct {
}
type jsonFeedItem struct {
ID string `json:"id"`
URL string `json:"url"`
Title string `json:"title"`
Summary string `json:"summary"`
ContentText string `json:"content_text"`
ContentHTML string `json:"content_html"`
Image string `json:"image"`
BannerImage string `json:"banner_image"`
DatePublished string `json:"date_published"`
ID string `json:"id"`
URL string `json:"url"`
Title string `json:"title"`
Summary string `json:"summary"`
ContentText string `json:"content_text"`
ContentHTML string `json:"content_html"`
Image string `json:"image"`
BannerImage string `json:"banner_image"`
Attachments []jsonFeedAttachment `json:"attachments"`
DatePublished string `json:"date_published"`
}
type jsonFeedAttachment struct {
URL string `json:"url"`
MIMEType string `json:"mime_type"`
}
func parseJSONFeed(body []byte) (ParsedFeed, error) {
@@ -179,7 +185,13 @@ func parseJSONFeed(body []byte) (ParsedFeed, error) {
}
item := normalizeFeedItem(FeedItem{
ExternalID: input.ID, Title: plainText(input.Title), Summary: plainText(input.Summary),
Content: content, URL: input.URL, ImageURL: firstNonEmpty(input.Image, input.BannerImage),
Content: content, URL: input.URL,
ImageURL: firstNonEmpty(
input.Image,
input.BannerImage,
firstJSONImageAttachment(input.Attachments),
htmlImageURL(input.ContentHTML),
),
PublishedAt: parseFeedTime(input.DatePublished),
})
if item.Title != "" {
@@ -338,6 +350,15 @@ func htmlImageURL(value string) string {
return strings.TrimSpace(html.UnescapeString(match[1]))
}
func firstJSONImageAttachment(attachments []jsonFeedAttachment) string {
for _, attachment := range attachments {
if strings.HasPrefix(strings.ToLower(strings.TrimSpace(attachment.MIMEType)), "image/") {
return strings.TrimSpace(attachment.URL)
}
}
return ""
}
func imageMediaURL(media xmlMedia) string {
mediaType := strings.ToLower(strings.TrimSpace(media.Type))
medium := strings.ToLower(strings.TrimSpace(media.Medium))

44
backend/internal/logic/dataset/feed_test.go
View File

@@ -189,6 +189,50 @@ func TestSyncSourcesCollectsMultipleSourcesForOneOwner(t *testing.T) {
}
}
func TestStoreFeedItemsDeduplicatesAndRefreshesMetadata(t *testing.T) {
database := newDatasetTestDatabase(t)
user := createDatasetTestUser(t, database, "refresh-feed-owner@example.com")
source := models.SaDatasetSource{
OwnerID: user.ID, Name: "Refresh feed", Kind: "rss",
URL: "https://example.com/feeds/main.xml", Enabled: true,
}
require.NoError(t, database.Create(&source).Error)
inserted, err := storeFeedItems(database, source, []FeedItem{{
ExternalID: "stable-item", Title: "Old title", Summary: "Old summary",
URL: "/articles/1",
}})
require.NoError(t, err)
require.Equal(t, 1, inserted)
var stored models.SaDatasetItem
require.NoError(t, database.Where("source_id = ?", source.ID).First(&stored).Error)
require.NoError(t, database.Model(&stored).Updates(map[string]any{
"status": "read", "starred": true,
}).Error)
inserted, err = storeFeedItems(database, source, []FeedItem{{
ExternalID: "stable-item", Title: "Updated title", Content: "Updated content",
URL: "/articles/1", ImageURL: "../images/cover.jpg",
}})
require.NoError(t, err)
require.Zero(t, inserted)
require.NoError(t, database.First(&stored, stored.ID).Error)
require.Equal(t, "Updated title", stored.Title)
require.Equal(t, "Old summary", stored.Summary)
require.Equal(t, "Updated content", stored.Content)
require.Equal(t, "https://example.com/articles/1", stored.URL)
require.Equal(t, "https://example.com/images/cover.jpg", stored.ImageURL)
require.Equal(t, "read", stored.Status)
require.True(t, stored.Starred)
var count int64
require.NoError(t, database.Model(&models.SaDatasetItem{}).
Where("source_id = ?", source.ID).
Count(&count).Error)
require.Equal(t, int64(1), count)
}
func TestSyncSourcesRejectsOverlap(t *testing.T) {
database := newDatasetTestDatabase(t)
user := createDatasetTestUser(t, database, "queued-feed-owner@example.com")

5
backend/internal/logic/dataset/handlers_test.go
View File

@@ -93,6 +93,7 @@ func TestDatasetAPIConnectsSourcesItemsAndSyncTasks(t *testing.T) {
var depositedNote models.SaNote
require.NoError(t, database.Where("project_id = ?", project.ID).First(&depositedNote).Error)
require.Equal(t, item.Title, depositedNote.Title)
require.Contains(t, depositedNote.Markdown, "![文章图片](https://example.com/article.jpg)")
require.Contains(t, depositedNote.Markdown, "原文链接")
firstSync := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-sources/sync", map[string]any{})
@@ -237,10 +238,6 @@ func newDatasetTestDatabase(t *testing.T) *gorm.DB {
&models.SaProject{},
&models.SaNote{},
))
require.NoError(t, database.Exec(`
CREATE UNIQUE INDEX IF NOT EXISTS idx_sa_dataset_item_source_external
ON sa_dataset_items (source_id, external_id)
`).Error)
return database
}

59
backend/internal/logic/dataset/service.go
View File

@@ -350,6 +350,13 @@ func (s *Service) DepositItem(userID uint, itemIdentity, projectIdentity string)
if body == "" {
body = strings.TrimSpace(item.Summary)
}
if item.ImageURL != "" {
if body != "" {
body = "![文章图片](" + item.ImageURL + ")\n\n" + body
} else {
body = "![文章图片](" + item.ImageURL + ")"
}
}
if item.URL != "" {
if body != "" {
body += "\n\n"
@@ -464,10 +471,12 @@ func storeFeedItems(tx *gorm.DB, source models.SaDatasetSource, items []FeedItem
inserted := 0
for _, input := range items {
externalID := input.ExternalID
itemURL := resolveFeedReference(source.URL, input.URL)
imageURL := resolveFeedReference(source.URL, input.ImageURL)
item := models.SaDatasetItem{
SourceID: source.ID, ExternalID: &externalID,
Title: input.Title, Summary: input.Summary, Content: input.Content,
URL: input.URL, ImageURL: input.ImageURL,
URL: itemURL, ImageURL: imageURL,
Status: "unread", PublishedAt: input.PublishedAt,
}
result := tx.Clauses(clause.OnConflict{
@@ -478,10 +487,58 @@ func storeFeedItems(tx *gorm.DB, source models.SaDatasetSource, items []FeedItem
return inserted, result.Error
}
inserted += int(result.RowsAffected)
if result.RowsAffected == 0 {
updates := map[string]any{
"title": item.Title,
}
if item.Summary != "" {
updates["summary"] = item.Summary
}
if item.Content != "" {
updates["content"] = item.Content
}
if item.URL != "" {
updates["url"] = item.URL
}
if item.ImageURL != "" {
updates["image_url"] = item.ImageURL
}
if item.PublishedAt != nil {
updates["published_at"] = item.PublishedAt
}
if err := tx.Model(&models.SaDatasetItem{}).
Where("source_id = ? AND external_id = ?", source.ID, externalID).
Updates(updates).Error; err != nil {
return inserted, err
}
}
}
return inserted, nil
}
func resolveFeedReference(baseURL, reference string) string {
reference = strings.TrimSpace(reference)
if reference == "" {
return ""
}
parsedReference, err := url.Parse(reference)
if err != nil {
return ""
}
if !parsedReference.IsAbs() {
parsedBase, err := url.Parse(baseURL)
if err != nil {
return ""
}
parsedReference = parsedBase.ResolveReference(parsedReference)
}
resolved := parsedReference.String()
if !validHTTPURL(resolved) {
return ""
}
return truncateRunes(resolved, 2048)
}
func truncateResult(value string) string {
return truncateRunes(value, 2000)
}

4
backend/internal/models/dataset_item.go
View File

@@ -5,9 +5,9 @@ import "time"
type SaDatasetItem struct {
ID uint `gorm:"primaryKey"`
Identity string `gorm:"type:char(36);uniqueIndex"`
SourceID uint `gorm:"index;not null"`
SourceID uint `gorm:"index;not null;uniqueIndex:idx_sa_dataset_item_source_external,priority:1"`
SourceIdentity string `gorm:"type:char(36);index"`
ExternalID *string `gorm:"size:64;index"`
ExternalID *string `gorm:"size:64;uniqueIndex:idx_sa_dataset_item_source_external,priority:2"`
Title string `gorm:"size:500;not null"`
Summary string `gorm:"type:text"`
Content string `gorm:"type:text"`