1package runtime
  2
  3import (
  4	"context"
  5	"fmt"
  6	"io"
  7	"os"
  8	"time"
  9
 10	"llm_aggregator/internal/aggregator"
 11
 12	"llm_aggregator/internal/llm"
 13	"llm_aggregator/internal/output"
 14	"llm_aggregator/internal/processor"
 15	"llm_aggregator/internal/progress"
 16)
 17
 18// Runtime holds the execution context for the aggregator
 19type Runtime struct {
 20	// Configuration
 21	FeedsFile          string
 22	Stdin              bool
 23	MaxArticlesPerFeed int
 24	MaxDaysOld         int
 25	MaxTotalArticles   int
 26	IncludeKeywords    []string
 27	ExcludeKeywords    []string
 28	APIKey             string
 29	BaseURL            string
 30	Model              string
 31	MaxTokens          int
 32	Temperature        float64
 33	LLMTimeout         int // seconds; 0 means no timeout
 34	Prompt             string
 35	SystemPrompt       string
 36	Output             string
 37	OutputFile         string
 38	IncludeArticles    bool
 39	Plain              bool
 40	Verbose            bool
 41
 42	// State
 43	Articles    []map[string]any
 44	Summary     string
 45	Error       error
 46	Interrupted bool // Set to true when a termination signal is received
 47
 48	// Logger for verbose output
 49	Progress progress.Progress
 50}
 51
 52// Execute runs the full aggregation pipeline
 53func (r *Runtime) Execute(ctx context.Context) error {
 54	// The logger/progress handler is injected, so we don't create it here.
 55	// We wrap it in a context to pass to sub-modules that expect a *progress.Context.
 56	progCtx := progress.NewContext(r.Progress)
 57
 58	feedAgg := aggregator.NewFeedAggregatorWithProgress(
 59		r.MaxArticlesPerFeed,
 60		r.MaxDaysOld,
 61		5000,    // max content length
 62		progCtx, // Pass the new progress context
 63	)
 64
 65	// Step 1: Aggregate feeds
 66	r.Progress.SetStage("Aggregating feeds")
 67
 68	var articles []*aggregator.Article
 69	var err error
 70
 71	if r.Stdin && r.FeedsFile != "" {
 72		// Both stdin and feeds file: collate results
 73		r.Progress.SetSubStage("Parsing stdin feed and feeds file")
 74
 75		var stdinArticles []*aggregator.Article
 76		var fileArticles []*aggregator.Article
 77
 78		if stdinArticles, err = feedAgg.ParseFeedFromStdin(); err != nil {
 79			return fmt.Errorf("error parsing stdin feed: %w", err)
 80		}
 81
 82		if fileArticles, err = feedAgg.ParseFeedsFromFile(r.FeedsFile); err != nil {
 83			return fmt.Errorf("error aggregating feeds: %w", err)
 84		}
 85
 86		articles = append(stdinArticles, fileArticles...)
 87	} else if r.Stdin {
 88		// Stdin only
 89		r.Progress.SetSubStage("Parsing stdin feed")
 90		if articles, err = feedAgg.ParseFeedFromStdin(); err != nil {
 91			return fmt.Errorf("error parsing stdin feed: %w", err)
 92		}
 93	} else if r.FeedsFile != "" {
 94		// Feeds file only
 95		r.Progress.SetSubStage(fmt.Sprintf("Parsing feeds from %s", r.FeedsFile))
 96		if articles, err = feedAgg.ParseFeedsFromFile(r.FeedsFile); err != nil {
 97			return fmt.Errorf("error aggregating feeds: %w", err)
 98		}
 99	} else {
100		return fmt.Errorf("no feed source specified: use --feeds-file or --stdin")
101	}
102
103	if len(articles) == 0 {
104		return fmt.Errorf("no articles found after parsing feeds")
105	}
106	r.Progress.SetArticleCount(len(articles), 0)
107
108	// Step 2: Process content
109	r.Progress.SetStage("Processing articles")
110	r.Progress.SetSubStage(fmt.Sprintf("Filtering and sorting %d articles", len(articles)))
111
112	contentProc := processor.NewContentProcessor(
113		r.MaxTotalArticles,
114		3000, // max content per article
115		r.IncludeKeywords,
116		r.ExcludeKeywords,
117	)
118	contentProc.SetLogger(progCtx) // Pass the new progress context
119
120	processedArticles := contentProc.ProcessArticles(articles, "date", true)
121
122	if len(processedArticles) == 0 {
123		return fmt.Errorf("no articles passed filtering")
124	}
125	r.Progress.SetArticleCount(len(articles), len(articles)-len(processedArticles))
126
127	r.Articles = processedArticles
128
129	// Estimate tokens for progress tracking
130	tokenEst := contentProc.EstimateTokenCount(processedArticles, r.Model)
131	r.Progress.SetTokenEstimate(tokenEst, tokenEst)
132
133	// Step 3: Initialise LLM client
134	r.Progress.SetStage("Connecting to LLM")
135	r.Progress.SetSubStage(fmt.Sprintf("Using model: %s", r.Model))
136
137	llm, err := llm.NewLLMClient(
138		r.APIKey,
139		r.BaseURL,
140		r.Model,
141		r.MaxTokens,
142		r.Temperature,
143		r.LLMTimeout,
144	)
145	if err != nil {
146		return fmt.Errorf("error initialising LLM client: %w", err)
147	}
148	llm.SetLogger(progCtx) // Pass the new progress context
149
150	// Step 4: Get summary from LLM
151	r.Progress.SetStage("Getting summary")
152	r.Progress.SetSubStage(fmt.Sprintf("Sending %d articles to LLM", len(processedArticles)))
153	r.Progress.StartWaiting()
154
155	// Derive a timeout context for the LLM call from the parent context.
156	// The parent context carries signal cancellation, so both signal interrupts
157	// and timeouts will abort the call.
158	callCtx := ctx
159	if r.LLMTimeout > 0 {
160		var cancel context.CancelFunc
161		callCtx, cancel = context.WithTimeout(ctx, time.Duration(r.LLMTimeout)*time.Second)
162		defer cancel()
163	}
164
165	summary, usage, err := llm.SummariseArticles(
166		processedArticles,
167		r.Prompt,
168		r.SystemPrompt,
169		callCtx,
170	)
171	if err != nil {
172		return fmt.Errorf("error getting summary from LLM: %w", err)
173	}
174
175	r.Summary = summary
176	if usage != nil {
177		r.Progress.SetTokenEstimate(tokenEst, tokenEst+usage.CompletionTokens)
178	}
179	return nil
180}
181
182// WasInterrupted returns true if a termination signal was received during execution.
183func (r *Runtime) WasInterrupted() bool {
184	return r.Interrupted
185}
186func (r *Runtime) WriteOutput(writer io.Writer) error {
187	if r.Plain {
188		if _, err := fmt.Fprint(writer, r.Summary); err != nil {
189			return fmt.Errorf("error writing plain output: %w", err)
190		}
191		return nil
192	}
193
194	formatter, err := output.NewFormatter(r.Output)
195	if err != nil {
196		return fmt.Errorf("error creating formatter: %w", err)
197	}
198
199	outputData := map[string]any{
200		"title":          fmt.Sprintf("LLM Aggregator Summary - %s", time.Now().Format("2006-01-02 15:04")),
201		"prompt":         r.Prompt,
202		"model":          r.Model,
203		"articles_count": len(r.Articles),
204		"summary":        r.Summary,
205		"timestamp":      time.Now().Format(time.RFC3339),
206	}
207
208	if r.IncludeArticles {
209		outputData["articles"] = r.Articles
210	}
211
212	formattedOutput, err := formatter.FormatData(outputData)
213	if err != nil {
214		return fmt.Errorf("error formatting output: %w", err)
215	}
216
217	if _, err := fmt.Fprint(writer, formattedOutput); err != nil {
218		return fmt.Errorf("error writing output: %w", err)
219	}
220
221	return nil
222}
223
224// WriteOutputToFile writes the output to the specified file
225func (r *Runtime) WriteOutputToFile() error {
226	if r.OutputFile == "" {
227		return r.WriteOutput(os.Stdout)
228	}
229
230	file, err := os.Create(r.OutputFile)
231	if err != nil {
232		return fmt.Errorf("error creating output file: %w", err)
233	}
234	defer file.Close()
235
236	return r.WriteOutput(file)
237}