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