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 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}