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