From 5da343cc71e65cad5b64e7db94f37a3150b07955 Mon Sep 17 00:00:00 2001 From: yanweidong Date: Thu, 23 Jul 2026 20:47:54 +0800 Subject: [PATCH] feat: collect JSON and XML feeds --- apps/web_v1/scripts/visual-check.mjs | 12 +- apps/web_v1/src/pages/workspace-explore.tsx | 14 +- backend/internal/logic/dataset/feed.go | 347 ++++++++++++++++++ backend/internal/logic/dataset/feed_test.go | 126 +++++++ backend/internal/logic/dataset/handlers.go | 4 +- .../internal/logic/dataset/handlers_test.go | 30 +- backend/internal/logic/dataset/service.go | 135 +++++-- backend/internal/models/dataset_item.go | 1 + 8 files changed, 627 insertions(+), 42 deletions(-) create mode 100644 backend/internal/logic/dataset/feed.go create mode 100644 backend/internal/logic/dataset/feed_test.go diff --git a/apps/web_v1/scripts/visual-check.mjs b/apps/web_v1/scripts/visual-check.mjs index 8f604f0..8098751 100644 --- a/apps/web_v1/scripts/visual-check.mjs +++ b/apps/web_v1/scripts/visual-check.mjs @@ -265,16 +265,16 @@ await page.route('http://localhost:9150/api/v1/**', async (route) => { } if (url.pathname === '/api/v1/dataset-crons/sync') { await route.fulfill({ - status: 202, + status: 200, json: visualDatasetSources.map((source, index) => ({ id: `019b0000-0000-7000-8000-${String(70 + index).padStart(12, '0')}`, sourceId: source.id, schedule: '@once', - status: 'pending', - enabled: true, - nextRunAt: new Date().toISOString(), - lastRunAt: null, - lastResult: '', + status: 'completed', + enabled: false, + nextRunAt: null, + lastRunAt: new Date().toISOString(), + lastResult: 'format=rss fetched=2 inserted=0', createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), })), diff --git a/apps/web_v1/src/pages/workspace-explore.tsx b/apps/web_v1/src/pages/workspace-explore.tsx index 4e92f84..468a1ca 100644 --- a/apps/web_v1/src/pages/workspace-explore.tsx +++ b/apps/web_v1/src/pages/workspace-explore.tsx @@ -231,12 +231,20 @@ export function WorkspaceExplorePage({ try { const crons = await onQueueSync() if (crons.length === 0) { - Message.info('暂无已启用的数据源') + Message.info('暂无已启用的 RSS 数据源') } else { - Message.success(`已提交 ${crons.length} 个采集任务`) + const failed = crons.filter((cron) => cron.status === 'failed').length + const [nextSources, nextItems] = await Promise.all([onListSources(), onListItems()]) + setSourceRecords(nextSources) + setItems(nextItems) + if (failed > 0) { + Message.warning(`${crons.length - failed} 个数据源采集完成，${failed} 个失败`) + } else { + Message.success(`已完成 ${crons.length} 个数据源采集任务`) + } } } catch (error) { - Message.error(error instanceof Error ? error.message : '同步任务提交失败') + Message.error(error instanceof Error ? error.message : '数据源采集失败') } finally { setSyncing(false) } diff --git a/backend/internal/logic/dataset/feed.go b/backend/internal/logic/dataset/feed.go new file mode 100644 index 0000000..9a0dd72 --- /dev/null +++ b/backend/internal/logic/dataset/feed.go @@ -0,0 +1,347 @@ +package dataset + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "encoding/xml" + "errors" + "fmt" + "html" + "io" + "net" + "net/http" + "net/url" + "regexp" + "strings" + "time" +) + +const maxFeedBytes = 5 << 20 + +var ( + ErrFeedFormat = errors.New("unsupported feed format") + ErrFeedTooLarge = errors.New("feed response is too large") + ErrUnsafeFeedURL = errors.New("feed url resolves to a private or local address") + htmlTagPattern = regexp.MustCompile(`<[^>]*>`) +) + +type FeedItem struct { + ExternalID string + Title string + Summary string + Content string + URL string + PublishedAt *time.Time +} + +type ParsedFeed struct { + Format string + Items []FeedItem +} + +type FeedFetcher interface { + Fetch(context.Context, string) (ParsedFeed, error) +} + +type HTTPFeedFetcher struct { + client *http.Client +} + +func NewHTTPFeedFetcher() *HTTPFeedFetcher { + transport := &http.Transport{ + DialContext: func(ctx context.Context, network, address string) (net.Conn, error) { + host, port, err := net.SplitHostPort(address) + if err != nil { + return nil, err + } + ips, err := net.DefaultResolver.LookupIPAddr(ctx, host) + if err != nil { + return nil, err + } + dialer := &net.Dialer{Timeout: 10 * time.Second} + var lastErr error + for _, candidate := range ips { + if !isPublicIP(candidate.IP) { + continue + } + connection, err := dialer.DialContext(ctx, network, net.JoinHostPort(candidate.IP.String(), port)) + if err == nil { + return connection, nil + } + lastErr = err + } + if lastErr != nil { + return nil, lastErr + } + return nil, ErrUnsafeFeedURL + }, + TLSHandshakeTimeout: 10 * time.Second, + ResponseHeaderTimeout: 15 * time.Second, + } + fetcher := &HTTPFeedFetcher{} + fetcher.client = &http.Client{ + Transport: transport, + Timeout: 30 * time.Second, + CheckRedirect: func(request *http.Request, via []*http.Request) error { + if len(via) >= 5 { + return errors.New("too many feed redirects") + } + return validateFeedURL(request.URL) + }, + } + return fetcher +} + +func (f *HTTPFeedFetcher) Fetch(ctx context.Context, value string) (ParsedFeed, error) { + parsedURL, err := url.Parse(value) + if err != nil { + return ParsedFeed{}, err + } + if err := validateFeedURL(parsedURL); err != nil { + return ParsedFeed{}, err + } + request, err := http.NewRequestWithContext(ctx, http.MethodGet, parsedURL.String(), nil) + if err != nil { + return ParsedFeed{}, err + } + request.Header.Set("Accept", "application/feed+json, application/json, application/rss+xml, application/atom+xml, application/xml, text/xml;q=0.9") + request.Header.Set("User-Agent", "SenlinAI-Agent/1.0") + response, err := f.client.Do(request) + if err != nil { + return ParsedFeed{}, err + } + defer response.Body.Close() + if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { + return ParsedFeed{}, fmt.Errorf("feed request returned %s", response.Status) + } + body, err := io.ReadAll(io.LimitReader(response.Body, maxFeedBytes+1)) + if err != nil { + return ParsedFeed{}, err + } + if len(body) > maxFeedBytes { + return ParsedFeed{}, ErrFeedTooLarge + } + return ParseFeed(body, response.Header.Get("Content-Type")) +} + +func ParseFeed(body []byte, contentType string) (ParsedFeed, error) { + body = bytes.TrimSpace(bytes.TrimPrefix(body, []byte{0xef, 0xbb, 0xbf})) + if len(body) == 0 { + return ParsedFeed{}, ErrFeedFormat + } + lowerContentType := strings.ToLower(contentType) + if body[0] == '{' || strings.Contains(lowerContentType, "json") { + if feed, err := parseJSONFeed(body); err == nil { + return feed, nil + } + } + if body[0] == '<' || strings.Contains(lowerContentType, "xml") || strings.Contains(lowerContentType, "rss") || strings.Contains(lowerContentType, "atom") { + return parseXMLFeed(body) + } + return ParsedFeed{}, ErrFeedFormat +} + +type jsonFeedDocument struct { + Version string `json:"version"` + Items []jsonFeedItem `json:"items"` +} + +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"` + DatePublished string `json:"date_published"` +} + +func parseJSONFeed(body []byte) (ParsedFeed, error) { + var document jsonFeedDocument + if err := json.Unmarshal(body, &document); err != nil { + return ParsedFeed{}, err + } + if document.Version == "" { + return ParsedFeed{}, ErrFeedFormat + } + items := make([]FeedItem, 0, len(document.Items)) + for _, input := range document.Items { + content := strings.TrimSpace(input.ContentText) + if content == "" { + content = plainText(input.ContentHTML) + } + item := normalizeFeedItem(FeedItem{ + ExternalID: input.ID, Title: plainText(input.Title), Summary: plainText(input.Summary), + Content: content, URL: input.URL, PublishedAt: parseFeedTime(input.DatePublished), + }) + if item.Title != "" { + items = append(items, item) + } + } + return ParsedFeed{Format: "json_feed", Items: items}, nil +} + +type xmlFeedDocument struct { + XMLName xml.Name + Channel struct { + Items []xmlFeedItem `xml:"item"` + } `xml:"channel"` + Items []xmlFeedItem `xml:"item"` + Entries []xmlAtomEntry `xml:"entry"` +} + +type xmlFeedItem struct { + Title string `xml:"title"` + Link string `xml:"link"` + GUID string `xml:"guid"` + Description string `xml:"description"` + Content string `xml:"encoded"` + PubDate string `xml:"pubDate"` + Date string `xml:"date"` +} + +type xmlAtomEntry struct { + Title string `xml:"title"` + ID string `xml:"id"` + Links []xmlAtomLink `xml:"link"` + Summary string `xml:"summary"` + Content string `xml:"content"` + Published string `xml:"published"` + Updated string `xml:"updated"` +} + +type xmlAtomLink struct { + Href string `xml:"href,attr"` + Rel string `xml:"rel,attr"` +} + +func parseXMLFeed(body []byte) (ParsedFeed, error) { + var document xmlFeedDocument + if err := xml.Unmarshal(body, &document); err != nil { + return ParsedFeed{}, err + } + switch strings.ToLower(document.XMLName.Local) { + case "rss", "rdf": + inputs := document.Channel.Items + if len(inputs) == 0 { + inputs = document.Items + } + items := make([]FeedItem, 0, len(inputs)) + for _, input := range inputs { + content := plainText(input.Content) + if content == "" { + content = plainText(input.Description) + } + item := normalizeFeedItem(FeedItem{ + ExternalID: input.GUID, Title: plainText(input.Title), + Summary: plainText(input.Description), Content: content, + URL: strings.TrimSpace(input.Link), PublishedAt: parseFeedTime(firstNonEmpty(input.PubDate, input.Date)), + }) + if item.Title != "" { + items = append(items, item) + } + } + return ParsedFeed{Format: "rss", Items: items}, nil + case "feed": + items := make([]FeedItem, 0, len(document.Entries)) + for _, input := range document.Entries { + item := normalizeFeedItem(FeedItem{ + ExternalID: input.ID, Title: plainText(input.Title), + Summary: plainText(input.Summary), Content: plainText(input.Content), + URL: atomLink(input.Links), PublishedAt: parseFeedTime(firstNonEmpty(input.Published, input.Updated)), + }) + if item.Title != "" { + items = append(items, item) + } + } + return ParsedFeed{Format: "atom", Items: items}, nil + default: + return ParsedFeed{}, ErrFeedFormat + } +} + +func normalizeFeedItem(item FeedItem) FeedItem { + item.Title = truncateRunes(strings.TrimSpace(item.Title), 500) + item.Summary = strings.TrimSpace(item.Summary) + item.Content = strings.TrimSpace(item.Content) + item.URL = truncateRunes(strings.TrimSpace(item.URL), 2048) + key := firstNonEmpty(strings.TrimSpace(item.ExternalID), item.URL) + if key == "" { + published := "" + if item.PublishedAt != nil { + published = item.PublishedAt.UTC().Format(time.RFC3339Nano) + } + key = item.Title + "\x00" + published + } + sum := sha256.Sum256([]byte(key)) + item.ExternalID = hex.EncodeToString(sum[:]) + return item +} + +func parseFeedTime(value string) *time.Time { + value = strings.TrimSpace(value) + if value == "" { + return nil + } + layouts := []string{ + time.RFC3339Nano, time.RFC3339, time.RFC1123Z, time.RFC1123, + time.RFC822Z, time.RFC822, time.RFC850, time.ANSIC, + "Mon, 2 Jan 2006 15:04:05 -0700", + } + for _, layout := range layouts { + if parsed, err := time.Parse(layout, value); err == nil { + utc := parsed.UTC() + return &utc + } + } + return nil +} + +func plainText(value string) string { + value = htmlTagPattern.ReplaceAllString(value, " ") + return strings.Join(strings.Fields(html.UnescapeString(value)), " ") +} + +func atomLink(links []xmlAtomLink) string { + for _, link := range links { + if link.Rel == "" || link.Rel == "alternate" { + return strings.TrimSpace(link.Href) + } + } + if len(links) > 0 { + return strings.TrimSpace(links[0].Href) + } + return "" +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if strings.TrimSpace(value) != "" { + return value + } + } + return "" +} + +func validateFeedURL(value *url.URL) error { + if value == nil || (value.Scheme != "http" && value.Scheme != "https") || value.Hostname() == "" || value.User != nil { + return ErrURLInvalid + } + if ip := net.ParseIP(value.Hostname()); ip != nil && !isPublicIP(ip) { + return ErrUnsafeFeedURL + } + return nil +} + +func isPublicIP(ip net.IP) bool { + return ip != nil && + !ip.IsLoopback() && + !ip.IsPrivate() && + !ip.IsLinkLocalUnicast() && + !ip.IsLinkLocalMulticast() && + !ip.IsUnspecified() && + !ip.IsMulticast() +} diff --git a/backend/internal/logic/dataset/feed_test.go b/backend/internal/logic/dataset/feed_test.go new file mode 100644 index 0000000..7e72ec2 --- /dev/null +++ b/backend/internal/logic/dataset/feed_test.go @@ -0,0 +1,126 @@ +package dataset + +import ( + "context" + "errors" + "net/url" + "testing" + + "github.com/stretchr/testify/require" + "senlinai-agent/backend/internal/models" +) + +func TestParseFeedDetectsJSONFeedFromContent(t *testing.T) { + feed, err := ParseFeed([]byte(`{ + "version": "https://jsonfeed.org/version/1.1", + "items": [{ + "id": "json-1", + "url": "https://example.com/json-1", + "title": "JSON item", + "summary": "JSON summary", + "content_html": "

JSON content

", + "date_published": "2026-07-23T08:00:00Z" + }] + }`), "text/plain") + + require.NoError(t, err) + require.Equal(t, "json_feed", feed.Format) + require.Len(t, feed.Items, 1) + require.Equal(t, "JSON item", feed.Items[0].Title) + require.Equal(t, "JSON content", feed.Items[0].Content) + require.Equal(t, "https://example.com/json-1", feed.Items[0].URL) + require.NotEmpty(t, feed.Items[0].ExternalID) + require.NotNil(t, feed.Items[0].PublishedAt) +} + +func TestParseFeedDetectsRSSXML(t *testing.T) { + feed, err := ParseFeed([]byte(` + + + + RSS item + https://example.com/rss-1 + rss-1 + RSS summary

]]> + RSS content

]]> + Thu, 23 Jul 2026 08:00:00 +0000 + + + `), "application/xml") + + require.NoError(t, err) + require.Equal(t, "rss", feed.Format) + require.Len(t, feed.Items, 1) + require.Equal(t, "RSS summary", feed.Items[0].Summary) + require.Equal(t, "RSS content", feed.Items[0].Content) + require.NotNil(t, feed.Items[0].PublishedAt) +} + +func TestParseFeedDetectsAtomXML(t *testing.T) { + feed, err := ParseFeed([]byte(` + + + Atom item + atom-1 + + Atom summary + Atom content + 2026-07-23T08:00:00Z + + `), "application/atom+xml") + + require.NoError(t, err) + require.Equal(t, "atom", feed.Format) + require.Len(t, feed.Items, 1) + require.Equal(t, "https://example.com/atom-1", feed.Items[0].URL) + require.Equal(t, "Atom content", feed.Items[0].Content) +} + +func TestParseFeedRejectsUnknownContent(t *testing.T) { + _, err := ParseFeed([]byte("not a feed"), "text/plain") + require.ErrorIs(t, err, ErrFeedFormat) +} + +func TestValidateFeedURLRejectsLocalAddressesAndCredentials(t *testing.T) { + for _, value := range []string{ + "http://127.0.0.1/feed", + "http://[::1]/feed", + "http://169.254.169.254/latest/meta-data", + "https://user:password@example.com/feed", + } { + parsed, err := url.Parse(value) + require.NoError(t, err) + require.Error(t, validateFeedURL(parsed), value) + } + + parsed, err := url.Parse("https://rsshub.app/cls/hot") + require.NoError(t, err) + require.NoError(t, validateFeedURL(parsed)) +} + +func TestSyncSourcesRecordsFeedFailure(t *testing.T) { + database := newDatasetTestDatabase(t) + user := createDatasetTestUser(t, database, "feed-owner@example.com") + source := models.SaDatasetSource{ + CreatedBy: user.ID, + Name: "Broken feed", + Kind: "rss", + URL: "https://example.com/feed", + Enabled: true, + } + require.NoError(t, database.Create(&source).Error) + + crons, err := newServiceWithFetcher(database, failingFeedFetcher{}).SyncSources(context.Background(), user.ID) + + require.NoError(t, err) + require.Len(t, crons, 1) + require.Equal(t, "failed", crons[0].Status) + require.False(t, crons[0].Enabled) + require.Contains(t, crons[0].LastResult, "feed unavailable") +} + +type failingFeedFetcher struct{} + +func (failingFeedFetcher) Fetch(context.Context, string) (ParsedFeed, error) { + return ParsedFeed{}, errors.New("feed unavailable") +} diff --git a/backend/internal/logic/dataset/handlers.go b/backend/internal/logic/dataset/handlers.go index 72130f5..7bc8f66 100644 --- a/backend/internal/logic/dataset/handlers.go +++ b/backend/internal/logic/dataset/handlers.go @@ -272,7 +272,7 @@ func (h *Handler) queueSync(c *gin.Context) { if !ok { return } - crons, err := h.service.QueueSync(userID) + crons, err := h.service.SyncSources(c.Request.Context(), userID) if err != nil { writeError(c, err) return @@ -281,7 +281,7 @@ func (h *Handler) queueSync(c *gin.Context) { for _, cron := range crons { result = append(result, cronDTO(cron)) } - c.JSON(http.StatusAccepted, result) + c.JSON(http.StatusOK, result) } func currentUser(c *gin.Context) (uint, bool) { diff --git a/backend/internal/logic/dataset/handlers_test.go b/backend/internal/logic/dataset/handlers_test.go index f384e29..dec7674 100644 --- a/backend/internal/logic/dataset/handlers_test.go +++ b/backend/internal/logic/dataset/handlers_test.go @@ -2,6 +2,7 @@ package dataset import ( "bytes" + "context" "encoding/json" "fmt" "net/http" @@ -63,17 +64,21 @@ func TestDatasetAPIConnectsSourcesItemsAndSyncTasks(t *testing.T) { require.True(t, item.Starred) firstSync := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-crons/sync", map[string]any{}) - require.Equal(t, http.StatusAccepted, firstSync.Code) + require.Equal(t, http.StatusOK, firstSync.Code) var firstCrons []CronDTO require.NoError(t, json.Unmarshal(firstSync.Body.Bytes(), &firstCrons)) require.Len(t, firstCrons, 1) - require.Equal(t, "pending", firstCrons[0].Status) + require.Equal(t, "completed", firstCrons[0].Status) + require.Contains(t, firstCrons[0].LastResult, "format=rss") secondSync := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-crons/sync", map[string]any{}) - require.Equal(t, http.StatusAccepted, secondSync.Code) + require.Equal(t, http.StatusOK, secondSync.Code) var secondCrons []CronDTO require.NoError(t, json.Unmarshal(secondSync.Body.Bytes(), &secondCrons)) - require.Equal(t, firstCrons[0].ID, secondCrons[0].ID) + require.NotEqual(t, firstCrons[0].ID, secondCrons[0].ID) + var collectedItems []models.SaDatasetItem + require.NoError(t, database.Where("source_id = ?", 1).Find(&collectedItems).Error) + require.Len(t, collectedItems, 2) otherRouter := datasetTestRouter(database, other.ID) otherSources := performDatasetRequest(t, otherRouter, http.MethodGet, "/api/v1/dataset-sources", nil) @@ -117,10 +122,25 @@ func datasetTestRouter(database *gorm.DB, userID uint) http.Handler { return httpx.NewProtectedRouter( config.Config{Env: "test"}, func(string) (uint, error) { return userID, nil }, - NewHandler(NewService(database)), + NewHandler(newServiceWithFetcher(database, staticFeedFetcher{})), ) } +type staticFeedFetcher struct{} + +func (staticFeedFetcher) Fetch(context.Context, string) (ParsedFeed, error) { + return ParsedFeed{ + Format: "rss", + Items: []FeedItem{{ + ExternalID: "stable-feed-item", + Title: "Collected item", + Summary: "Collected summary", + Content: "Collected content", + URL: "https://example.com/collected", + }}, + }, nil +} + func performDatasetRequest(t *testing.T, router http.Handler, method, path string, body any) *httptest.ResponseRecorder { t.Helper() var encoded []byte diff --git a/backend/internal/logic/dataset/service.go b/backend/internal/logic/dataset/service.go index fef6b54..2a84415 100644 --- a/backend/internal/logic/dataset/service.go +++ b/backend/internal/logic/dataset/service.go @@ -1,7 +1,9 @@ package dataset import ( + "context" "errors" + "fmt" "net/url" "strings" "time" @@ -19,7 +21,8 @@ var ( ) type Service struct { - db *gorm.DB + db *gorm.DB + fetcher FeedFetcher } type SourceInput struct { @@ -51,7 +54,11 @@ type SourceRecord struct { } func NewService(database *gorm.DB) *Service { - return &Service{db: database} + return &Service{db: database, fetcher: NewHTTPFeedFetcher()} +} + +func newServiceWithFetcher(database *gorm.DB, fetcher FeedFetcher) *Service { + return &Service{db: database, fetcher: fetcher} } func (s *Service) ListSources(userID uint) ([]SourceRecord, error) { @@ -208,36 +215,111 @@ func (s *Service) ListCrons(userID uint) ([]models.SaDatasetCron, error) { return crons, nil } -func (s *Service) QueueSync(userID uint) ([]models.SaDatasetCron, error) { +func (s *Service) SyncSources(ctx context.Context, userID uint) ([]models.SaDatasetCron, error) { result := make([]models.SaDatasetCron, 0) - err := s.db.Transaction(func(tx *gorm.DB) error { - var sources []models.SaDatasetSource - if err := tx.Where("created_by = ? AND enabled = ?", userID, true).Order("id asc").Find(&sources).Error; err != nil { - return err - } + var sources []models.SaDatasetSource + if err := s.db.Where("created_by = ? AND enabled = ? AND kind = ?", userID, true, "rss"). + Order("id asc").Find(&sources).Error; err != nil { + return nil, err + } + for _, source := range sources { now := time.Now().UTC() - for _, source := range sources { - var cron models.SaDatasetCron - err := tx.Where("source_id = ? AND created_by = ? AND status = ?", source.ID, userID, "pending").First(&cron).Error - if err == nil { - result = append(result, cron) - continue - } - if !errors.Is(err, gorm.ErrRecordNotFound) { - return err - } - cron = models.SaDatasetCron{ - SourceID: source.ID, CreatedBy: userID, Schedule: "@once", - Status: "pending", Enabled: true, NextRunAt: &now, - } - if err := tx.Create(&cron).Error; err != nil { - return err + cron := models.SaDatasetCron{ + SourceID: source.ID, CreatedBy: userID, Schedule: "@once", + Status: "running", Enabled: true, NextRunAt: &now, LastRunAt: &now, + } + if err := s.db.Create(&cron).Error; err != nil { + return nil, err + } + + feed, fetchErr := s.fetcher.Fetch(ctx, source.URL) + if fetchErr != nil { + cron.Status = "failed" + cron.Enabled = false + cron.NextRunAt = nil + cron.LastResult = truncateResult(fetchErr.Error()) + if err := s.db.Model(&cron).Updates(map[string]any{ + "status": cron.Status, "last_result": cron.LastResult, "enabled": false, "next_run_at": nil, + }).Error; err != nil { + return nil, err } result = append(result, cron) + continue + } + + inserted, err := s.storeFeedItems(userID, source, feed.Items) + if err != nil { + cron.Status = "failed" + cron.Enabled = false + cron.NextRunAt = nil + cron.LastResult = truncateResult("store feed items: " + err.Error()) + if updateErr := s.db.Model(&cron).Updates(map[string]any{ + "status": cron.Status, "last_result": cron.LastResult, "enabled": false, "next_run_at": nil, + }).Error; updateErr != nil { + return nil, updateErr + } + result = append(result, cron) + continue + } + completedAt := time.Now().UTC() + cron.Status = "completed" + cron.Enabled = false + cron.NextRunAt = nil + cron.LastResult = fmt.Sprintf("format=%s fetched=%d inserted=%d", feed.Format, len(feed.Items), inserted) + if err := s.db.Transaction(func(tx *gorm.DB) error { + if err := tx.Model(&cron).Updates(map[string]any{ + "status": cron.Status, "last_result": cron.LastResult, "enabled": false, "next_run_at": nil, + }).Error; err != nil { + return err + } + return tx.Model(&source).Update("last_synced_at", completedAt).Error + }); err != nil { + return nil, err + } + source.LastSyncedAt = &completedAt + result = append(result, cron) + } + return result, nil +} + +func (s *Service) storeFeedItems(userID uint, source models.SaDatasetSource, items []FeedItem) (int, error) { + inserted := 0 + err := s.db.Transaction(func(tx *gorm.DB) error { + for _, input := range items { + var count int64 + if err := tx.Model(&models.SaDatasetItem{}). + Where("source_id = ? AND external_id = ?", source.ID, input.ExternalID). + Count(&count).Error; err != nil { + return err + } + if count > 0 { + continue + } + item := models.SaDatasetItem{ + SourceID: source.ID, CreatedBy: userID, ExternalID: input.ExternalID, + Title: input.Title, Summary: input.Summary, Content: input.Content, + URL: input.URL, Status: "unread", PublishedAt: input.PublishedAt, + } + if err := tx.Create(&item).Error; err != nil { + return err + } + inserted++ } return nil }) - return result, err + return inserted, err +} + +func truncateResult(value string) string { + return truncateRunes(value, 2000) +} + +func truncateRunes(value string, limit int) string { + runes := []rune(value) + if len(runes) <= limit { + return value + } + return string(runes[:limit]) } func (s *Service) findOwnedSource(userID uint, identity string) (*models.SaDatasetSource, error) { @@ -272,7 +354,8 @@ func normalizeSourceInput(input SourceInput) (SourceInput, error) { func validHTTPURL(value string) bool { parsed, err := url.ParseRequestURI(value) - return err == nil && (parsed.Scheme == "http" || parsed.Scheme == "https") && parsed.Host != "" + return err == nil && (parsed.Scheme == "http" || parsed.Scheme == "https") && + parsed.Host != "" && parsed.User == nil } func utcOptionalTime(value *time.Time) *time.Time { diff --git a/backend/internal/models/dataset_item.go b/backend/internal/models/dataset_item.go index a1dab9c..8cf28b0 100644 --- a/backend/internal/models/dataset_item.go +++ b/backend/internal/models/dataset_item.go @@ -9,6 +9,7 @@ type SaDatasetItem struct { SourceIdentity string `gorm:"type:char(36);index"` CreatedBy uint `gorm:"index;not null"` CreatedByIdentity string `gorm:"type:char(36);index"` + ExternalID string `gorm:"size:64;index"` Title string `gorm:"size:500;not null"` Summary string `gorm:"type:text"` Content string `gorm:"type:text"`