internal/aggregator/aggregator.go (view raw)
1package aggregator
2
3import (
4 "context"
5 "fmt"
6 "io"
7 "net/http"
8 "os"
9 "strings"
10 "sync"
11 "time"
12
13 "codeberg.org/maxwelljensen/llm_aggregator/internal/progress"
14
15 "github.com/PuerkitoBio/goquery"
16 "github.com/mmcdole/gofeed"
17 "golang.org/x/sync/errgroup"
18)
19
20// FeedAggregator aggregates articles from multiple RSS feeds.
21type FeedAggregator struct {
22 maxArticlesPerFeed int
23 maxDaysOld int
24 maxContentLength int
25 client *http.Client
26 userAgent string
27 progressCtx *progress.Context
28}
29
30// NewFeedAggregatorWithProgress creates a FeedAggregator with progress reporting.
31// progressCtx may be nil (e.g. in dry-run mode with verbose disabled).
32func NewFeedAggregatorWithProgress(maxArticlesPerFeed, maxDaysOld, maxContentLength int, progressCtx *progress.Context) *FeedAggregator {
33 return &FeedAggregator{
34 maxArticlesPerFeed: maxArticlesPerFeed,
35 maxDaysOld: maxDaysOld,
36 maxContentLength: maxContentLength,
37 client: &http.Client{
38 Timeout: 30 * time.Second,
39 },
40 userAgent: "LLM-Aggregator/0.1.0 (+https://codeberg.org/maxwelljensen/llm-aggregator)",
41 progressCtx: progressCtx,
42 }
43}
44
45// ParseFeedsFromFile parses RSS feeds from a file containing one URL per line.
46// Feeds are fetched concurrently for improved performance.
47func (fa *FeedAggregator) ParseFeedsFromFile(filePath string) ([]*Article, error) {
48 file, err := os.Open(filePath)
49 if err != nil {
50 return nil, fmt.Errorf("failed to open feeds file: %w", err)
51 }
52 defer file.Close() //nolint:errcheck
53
54 return fa.parseFeedsFromReader(file, filePath)
55}
56
57// ParseFeedFromStdin parses a single RSS/Atom feed from stdin.
58func (fa *FeedAggregator) ParseFeedFromStdin() ([]*Article, error) {
59 return fa.ParseFeedFromReader(os.Stdin, "stdin")
60}
61
62// ParseFeedFromReader parses a single RSS/Atom feed from an io.Reader.
63// The sourceName is used for progress/logging and to identify the feed source.
64func (fa *FeedAggregator) ParseFeedFromReader(reader io.Reader, sourceName string) ([]*Article, error) {
65 fp := gofeed.NewParser()
66 feed, err := fp.Parse(reader)
67 if err != nil {
68 return nil, fmt.Errorf("failed to parse feed from %s: %w", sourceName, err)
69 }
70
71 feedTitle := feed.Title
72 if feedTitle == "" {
73 feedTitle = sourceName
74 }
75
76 if fa.progressCtx != nil {
77 fa.progressCtx.Logf("Parsing feed: %s (%d entries)", feedTitle, len(feed.Items))
78 }
79
80 var cutoffTime time.Time
81 if fa.maxDaysOld > 0 {
82 cutoffTime = time.Now().Add(-time.Duration(fa.maxDaysOld) * 24 * time.Hour)
83 }
84
85 maxItems := min(fa.maxArticlesPerFeed, len(feed.Items))
86 articles := make([]*Article, 0, maxItems)
87
88 for i := range feed.Items[:maxItems] {
89 article := fa.extractArticle(feed.Items[i], feedTitle, cutoffTime)
90 if article != nil {
91 articles = append(articles, article)
92 }
93 }
94
95 return articles, nil
96}
97
98// parseFeedsFromReader reads feed URLs line-by-line from any io.Reader,
99// then fetches each feed concurrently. Used by both file and stdin paths.
100func (fa *FeedAggregator) parseFeedsFromReader(reader io.Reader, sourceName string) ([]*Article, error) {
101 content, err := io.ReadAll(reader)
102 if err != nil {
103 return nil, fmt.Errorf("failed to read feeds from %s: %w", sourceName, err)
104 }
105
106 lines := strings.Split(string(content), "\n")
107 feedURLs := []string{}
108 for _, line := range lines {
109 line = strings.TrimSpace(line)
110 if line != "" && !strings.HasPrefix(line, "#") {
111 feedURLs = append(feedURLs, line)
112 }
113 }
114
115 if fa.progressCtx != nil {
116 fa.progressCtx.Logf("Found %d feed URLs in %s", len(feedURLs), sourceName)
117 }
118
119 // Use mutex to protect shared state
120 var mu sync.Mutex
121 var allArticles []*Article
122 var feedErrors []string
123
124 // Limit concurrency to avoid overwhelming servers
125 // NOTE: maxConcurrency is hardcoded to 10. Changing this requires a code
126 // modification.
127 const maxConcurrency = 10
128 sem := make(chan struct{}, maxConcurrency)
129
130 g, _ := errgroup.WithContext(context.Background())
131 for i, feedURL := range feedURLs {
132 currentIndex := i
133 sem <- struct{}{} // Acquire semaphore
134 g.Go(func() error {
135 defer func() { <-sem }() // Release semaphore
136 if fa.progressCtx != nil {
137 fa.progressCtx.Logf("Processing feed %d/%d: %s", currentIndex+1, len(feedURLs), feedURL)
138 }
139 feedArticles, err := fa.ParseFeed(feedURL)
140 if err != nil {
141 if fa.progressCtx != nil {
142 fa.progressCtx.Warningf("Failed to parse feed %s: %v", feedURL, err)
143 }
144 mu.Lock()
145 feedErrors = append(feedErrors, fmt.Sprintf("%s: %v", feedURL, err))
146 mu.Unlock()
147 return nil // Don't return error to continue processing other feeds
148 }
149 mu.Lock()
150 allArticles = append(allArticles, feedArticles...)
151 mu.Unlock()
152 return nil
153 })
154 }
155
156 if err := g.Wait(); err != nil {
157 return nil, fmt.Errorf("failed to process feeds: %w", err)
158 }
159
160 if len(feedErrors) > 0 {
161 if fa.progressCtx != nil {
162 fa.progressCtx.Warningf("Encountered %d feed errors: %v", len(feedErrors), feedErrors)
163 }
164 }
165
166 return allArticles, nil
167}
168
169// ParseFeed parses a single RSS feed and extracts articles.
170func (fa *FeedAggregator) ParseFeed(feedURL string) ([]*Article, error) {
171 articles := []*Article{}
172
173 fp := gofeed.NewParser()
174 feed, err := fp.ParseURL(feedURL)
175 if err != nil {
176 return nil, fmt.Errorf("failed to parse feed URL %s: %w", feedURL, err)
177 }
178
179 feedTitle := feed.Title
180 if feedTitle == "" {
181 feedTitle = feedURL
182 }
183
184 if fa.progressCtx != nil {
185 fa.progressCtx.Logf("Parsing feed: %s (%d entries)", feedTitle, len(feed.Items))
186 }
187
188 var cutoffTime time.Time
189 if fa.maxDaysOld > 0 {
190 cutoffTime = time.Now().Add(-time.Duration(fa.maxDaysOld) * 24 * time.Hour)
191 }
192
193 maxItems := min(fa.maxArticlesPerFeed, len(feed.Items))
194
195 for _, item := range feed.Items[:maxItems] {
196 article := fa.extractArticle(item, feedTitle, cutoffTime)
197 if article != nil {
198 articles = append(articles, article)
199 }
200 }
201
202 return articles, nil
203}
204
205func (fa *FeedAggregator) extractArticle(item *gofeed.Item, feedTitle string, cutoffTime time.Time) *Article {
206 // Extract metadata
207 title := item.Title
208 if title == "" {
209 title = "Untitled"
210 }
211
212 link := item.Link
213 if link == "" {
214 return nil
215 }
216
217 // Parse publication date
218 published := fa.parsePublishedDate(item)
219
220 // Check if article is too old
221 if !cutoffTime.IsZero() && !published.IsZero() && published.Before(cutoffTime) {
222 if fa.progressCtx != nil {
223 fa.progressCtx.Debugf("Skipping old article: %s (%s)", title, published.Format("2006-01-02"))
224 }
225 return nil
226 }
227
228 // Extract author
229 author := ""
230 if item.Author != nil {
231 author = item.Author.Name
232 if author == "" {
233 author = item.Author.Email
234 }
235 }
236 if author == "" && item.DublinCoreExt != nil && len(item.DublinCoreExt.Creator) > 0 {
237 author = item.DublinCoreExt.Creator[0]
238 }
239
240 // Get content
241 content := fa.extractContent(item, link)
242
243 // Truncate content if too long
244 if len(content) > fa.maxContentLength {
245 content = content[:fa.maxContentLength] + "... [truncated]"
246 }
247
248 return &Article{
249 Title: title,
250 Link: link,
251 Content: content,
252 Published: published,
253 Author: author,
254 SourceFeed: feedTitle,
255 }
256}
257
258func (fa *FeedAggregator) parsePublishedDate(item *gofeed.Item) time.Time {
259 if item.PublishedParsed != nil {
260 return *item.PublishedParsed
261 }
262 if item.UpdatedParsed != nil {
263 return *item.UpdatedParsed
264 }
265 // Try to parse from string
266 if item.Published != "" {
267 if t, err := time.Parse(time.RFC1123, item.Published); err == nil {
268 return t
269 }
270 if t, err := time.Parse(time.RFC1123Z, item.Published); err == nil {
271 return t
272 }
273 if t, err := time.Parse(time.RFC3339, item.Published); err == nil {
274 return t
275 }
276 }
277 return time.Time{}
278}
279
280func (fa *FeedAggregator) extractContent(item *gofeed.Item, link string) string {
281 content := ""
282
283 // Feed item content fields are tried in order; gofeed unifies atom Content/Description
284 if item.Content != "" {
285 content = item.Content
286 } else if item.Description != "" {
287 content = item.Description
288 }
289
290 // If still no content or very short, fetch webpage
291 if content == "" || len(content) < 100 {
292 fetchedContent, err := fa.fetchWebpageContent(link)
293 if err == nil && fetchedContent != "" {
294 content = fetchedContent
295 }
296 }
297
298 return content
299}
300
301func (fa *FeedAggregator) fetchWebpageContent(url string) (string, error) {
302 req, err := http.NewRequest(http.MethodGet, url, nil)
303 if err != nil {
304 return "", err
305 }
306 req.Header.Set("User-Agent", fa.userAgent)
307
308 resp, err := fa.client.Do(req)
309 if err != nil {
310 return "", err
311 }
312 defer resp.Body.Close() //nolint:errcheck
313
314 if resp.StatusCode != http.StatusOK {
315 return "", fmt.Errorf("HTTP status %d", resp.StatusCode)
316 }
317
318 doc, err := goquery.NewDocumentFromReader(resp.Body)
319 if err != nil {
320 return "", err
321 }
322
323 // Remove noise elements before extracting text
324 doc.Find("script, style, nav, footer, header").Remove()
325
326 // Try article-shaped selectors first, then fall back to paragraphs, then body
327 articleSelectors := []string{
328 "article",
329 "main",
330 ".article-content",
331 ".post-content",
332 ".entry-content",
333 "#content",
334 ".content",
335 }
336
337 for _, selector := range articleSelectors {
338 articleElement := doc.Find(selector).First()
339 if articleElement.Length() > 0 {
340 text := articleElement.Text()
341 text = strings.Join(strings.Fields(text), " ") // Normalise whitespace
342 if len(text) > 200 {
343 return text, nil
344 }
345 }
346 }
347
348 // Fallback: collect all paragraphs
349 paragraphs := doc.Find("p")
350 if paragraphs.Length() > 0 {
351 var texts []string
352 paragraphs.Each(func(i int, s *goquery.Selection) {
353 texts = append(texts, strings.TrimSpace(s.Text()))
354 })
355 text := strings.Join(texts, " ")
356 if len(text) > 200 {
357 return text, nil
358 }
359 }
360
361 // Last resort: raw body text
362 text := doc.Text()
363 text = strings.Join(strings.Fields(text), " ")
364 if len(text) > 200 {
365 return text, nil
366 }
367
368 return "", nil
369}