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