813 lines
26 KiB
Go
813 lines
26 KiB
Go
// Package app owns application orchestration and top-level use cases.
|
|
package app
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"path/filepath"
|
|
"time"
|
|
|
|
distributoradapter "gitea.maximumdirect.net/eric/weatherreporter/internal/adapters/distributor"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/briefing"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/collect"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/config"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/facts"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/forecast"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/module"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/promptdebug"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/promptexec"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/promptinput"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/report"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/timeutil"
|
|
"gitea.maximumdirect.net/eric/weatherreporter/internal/weatherdata"
|
|
)
|
|
|
|
type ReportKind string
|
|
|
|
const (
|
|
ReportDaily ReportKind = ReportKind(report.CommandNameDaily)
|
|
ReportToday ReportKind = ReportKind(report.CommandNameToday)
|
|
ReportTomorrow ReportKind = ReportKind(report.CommandNameTomorrow)
|
|
ReportHourly ReportKind = ReportKind(report.CommandNameHourly)
|
|
)
|
|
|
|
type BatchKind string
|
|
|
|
const (
|
|
BatchMorning BatchKind = BatchKind(report.BatchNameMorning)
|
|
BatchEvening BatchKind = BatchKind(report.BatchNameEvening)
|
|
)
|
|
|
|
type GenerateRequest struct {
|
|
Config config.Config
|
|
Report ReportKind
|
|
WorkingDir string
|
|
OutputPath string
|
|
LLMDebugDir string
|
|
Now time.Time
|
|
Date time.Time
|
|
Collector Collector
|
|
Notifier Notifier
|
|
Executor promptexec.Executor
|
|
}
|
|
|
|
type BatchRequest struct {
|
|
Config config.Config
|
|
Batch BatchKind
|
|
Now time.Time
|
|
WorkingDir string
|
|
OutputDir string
|
|
LLMDebugDir string
|
|
Collector Collector
|
|
Executor promptexec.Executor
|
|
Notifier Notifier
|
|
}
|
|
|
|
type ModuleSnapshotRequest struct {
|
|
Config config.Config
|
|
Resolved report.Resolved
|
|
}
|
|
|
|
type ReportFacts struct {
|
|
Collected facts.CollectedFacts
|
|
Derived facts.DerivedFacts
|
|
}
|
|
|
|
type ReportResult struct {
|
|
ReportID report.ID
|
|
ReportName string
|
|
PromptID string
|
|
PromptVersion string
|
|
RunID string
|
|
GeneratedAt time.Time
|
|
Timezone string
|
|
ValidPeriod timeutil.Period
|
|
ProfileID string
|
|
BackendID string
|
|
ModelName string
|
|
SourceWarnings []weatherdata.SourceWarning
|
|
ValidationStatus promptexec.ValidationStatus
|
|
LLMDebugPath string
|
|
OutputPath string
|
|
Notification *NotificationResult
|
|
}
|
|
|
|
type BatchResult struct {
|
|
Batch BatchKind `json:"batch"`
|
|
StartedAt time.Time `json:"startedAt"`
|
|
FinishedAt time.Time `json:"finishedAt"`
|
|
Total int `json:"total"`
|
|
Succeeded int `json:"succeeded"`
|
|
Failed int `json:"failed"`
|
|
Notification *BatchNotificationResult `json:"notification,omitempty"`
|
|
Reports []BatchReportResult `json:"reports"`
|
|
}
|
|
|
|
type BatchNotificationResult struct {
|
|
Status string `json:"status"`
|
|
Reason string `json:"reason,omitempty"`
|
|
RunID string `json:"runId,omitempty"`
|
|
PipelineID string `json:"pipelineId,omitempty"`
|
|
BundleID string `json:"bundleId,omitempty"`
|
|
IdempotencyKey string `json:"idempotencyKey,omitempty"`
|
|
IncludedReports []BatchNotificationReport `json:"includedReports,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
type BatchNotificationReport struct {
|
|
ReportID report.ID `json:"reportId"`
|
|
RunID string `json:"runId"`
|
|
SourcePath string `json:"sourcePath"`
|
|
BundlePaths []string `json:"bundlePaths"`
|
|
}
|
|
|
|
type BatchReportResult struct {
|
|
ReportID report.ID `json:"reportId"`
|
|
ReportName string `json:"reportName"`
|
|
PromptID string `json:"promptId"`
|
|
RunID string `json:"runId"`
|
|
Status string `json:"status"`
|
|
Error string `json:"error,omitempty"`
|
|
GeneratedAt time.Time `json:"generatedAt"`
|
|
ValidPeriod timeutil.Period `json:"validPeriod"`
|
|
Timezone string `json:"timezone"`
|
|
ProfileID string `json:"profileId,omitempty"`
|
|
BackendID string `json:"backendId,omitempty"`
|
|
ModelName string `json:"modelName,omitempty"`
|
|
SourceWarnings []weatherdata.SourceWarning `json:"sourceWarnings,omitempty"`
|
|
ValidationStatus promptexec.ValidationStatus `json:"validationStatus,omitempty"`
|
|
LLMDebugPath string `json:"llmDebugPath,omitempty"`
|
|
OutputPath string `json:"outputPath,omitempty"`
|
|
}
|
|
|
|
type BatchError struct {
|
|
Result *BatchResult
|
|
}
|
|
|
|
func (e BatchError) Error() string {
|
|
if e.Result == nil {
|
|
return "batch failed"
|
|
}
|
|
failedReports := batchReportFailures(e.Result)
|
|
if batchNotificationFailed(e.Result) && failedReports == 0 {
|
|
if e.Result.Notification.Error != "" {
|
|
return fmt.Sprintf("batch %s notification failed: %s", e.Result.Batch, e.Result.Notification.Error)
|
|
}
|
|
return fmt.Sprintf("batch %s notification failed", e.Result.Batch)
|
|
}
|
|
return fmt.Sprintf("batch %s failed: %d of %d reports failed", e.Result.Batch, failedReports, len(e.Result.Reports))
|
|
}
|
|
|
|
func batchNotificationFailed(result *BatchResult) bool {
|
|
return result != nil && result.Notification != nil && result.Notification.Status == "failed"
|
|
}
|
|
|
|
func batchReportFailures(result *BatchResult) int {
|
|
if result == nil {
|
|
return 0
|
|
}
|
|
failures := 0
|
|
for _, item := range result.Reports {
|
|
if item.Status == "failed" {
|
|
failures++
|
|
}
|
|
}
|
|
return failures
|
|
}
|
|
|
|
type Collector interface {
|
|
Run(context.Context, collect.Request) (*collect.Result, error)
|
|
}
|
|
|
|
type defaultCollector struct{}
|
|
|
|
func (defaultCollector) Run(ctx context.Context, req collect.Request) (*collect.Result, error) {
|
|
return collect.Run(ctx, req)
|
|
}
|
|
|
|
type Notifier interface {
|
|
Notify(context.Context, NotificationRequest) (*NotificationResult, error)
|
|
}
|
|
|
|
type NotificationRequest struct {
|
|
ReportID report.ID
|
|
RunID string
|
|
PipelineID string
|
|
BundleID string
|
|
IdempotencyKey string
|
|
ReportPath string
|
|
BundlePaths []string
|
|
CreatedAt time.Time
|
|
}
|
|
|
|
type NotificationResult struct {
|
|
BundleID string
|
|
IdempotencyKey string
|
|
RunID string
|
|
Status string
|
|
UploadStatus string
|
|
StatusError string
|
|
PipelineID string
|
|
AcceptedAt time.Time
|
|
StartedAt *time.Time
|
|
FinishedAt *time.Time
|
|
Report []byte
|
|
Error string
|
|
}
|
|
|
|
type NotificationError struct {
|
|
Request NotificationRequest
|
|
Err error
|
|
}
|
|
|
|
func (e *NotificationError) Error() string {
|
|
if e == nil || e.Err == nil {
|
|
return "notification failed"
|
|
}
|
|
return e.Err.Error()
|
|
}
|
|
|
|
func (e *NotificationError) Unwrap() error {
|
|
if e == nil {
|
|
return nil
|
|
}
|
|
return e.Err
|
|
}
|
|
|
|
func Generate(ctx context.Context, req GenerateRequest) error {
|
|
_, err := GenerateDetailed(ctx, req)
|
|
return err
|
|
}
|
|
|
|
func GenerateDetailed(ctx context.Context, req GenerateRequest) (*ReportResult, error) {
|
|
now := req.Now
|
|
if now.IsZero() {
|
|
now = time.Now()
|
|
}
|
|
resolved, err := ResolveGenerate(req, now)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result := initialReportResult(req, resolved, PromptInspectionResult{})
|
|
outputPath, err := resolveReportOutputPath(req.WorkingDir, req.OutputPath, req.Config.Output.Directory, resolved)
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
req.OutputPath = outputPath
|
|
debugWriter, err := promptdebug.NewPromptDebugWriter(req.LLMDebugDir)
|
|
if err != nil {
|
|
return result, promptexec.NewError(promptexec.InvalidConfiguration, "initialize prompt debug", err)
|
|
}
|
|
defer func() { _ = debugWriter.Close() }()
|
|
inspection, err := InspectPromptExecution(ctx, PromptInspectionRequest{
|
|
Resolved: resolved,
|
|
Executor: req.Executor,
|
|
Promptkit: req.Config.Promptkit,
|
|
})
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
result.ProfileID, result.BackendID, result.ModelName = inspection.ProfileID, inspection.BackendID, inspection.ModelName
|
|
collection, err := collectWeather(ctx, req.Config, req.Collector)
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
return generatePromptReport(ctx, promptReportRequest{
|
|
GenerateRequest: req,
|
|
Resolved: resolved,
|
|
Collection: *collection,
|
|
Inspection: inspection,
|
|
DebugWriter: debugWriter,
|
|
Result: result,
|
|
})
|
|
}
|
|
|
|
func RunBatch(ctx context.Context, req BatchRequest) error {
|
|
result, err := RunBatchDetailed(ctx, req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if result.Failed > 0 || batchNotificationFailed(result) {
|
|
return BatchError{Result: result}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func RunBatchDetailed(ctx context.Context, req BatchRequest) (*BatchResult, error) {
|
|
now := req.Now
|
|
if now.IsZero() {
|
|
now = time.Now()
|
|
}
|
|
if _, err := report.BatchForCommandName(string(req.Batch)); err != nil {
|
|
return nil, err
|
|
}
|
|
outputDir, err := resolveOutputDirWithConfigured(req.WorkingDir, req.OutputDir, req.Config.Output.Directory)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
req.OutputDir = outputDir
|
|
debugWriter, err := promptdebug.NewPromptDebugWriter(req.LLMDebugDir)
|
|
if err != nil {
|
|
return nil, promptexec.NewError(promptexec.InvalidConfiguration, "initialize prompt debug", err)
|
|
}
|
|
defer func() { _ = debugWriter.Close() }()
|
|
candidates, err := batchInspectionCandidates(req, now)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
inspections, err := InspectPromptExecutions(ctx, PromptExecutionsInspectionRequest{
|
|
Resolved: candidates,
|
|
Executor: req.Executor,
|
|
Promptkit: req.Config.Promptkit,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
collection, err := collectWeather(ctx, req.Config, req.Collector)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
plannedReports, err := planBatchRun(req, now, *collection)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := prepareBatchOutputs(req.OutputDir, plannedReports); err != nil {
|
|
return nil, err
|
|
}
|
|
if req.Batch == BatchEvening || req.Batch == BatchMorning {
|
|
startedAt := now
|
|
result := &BatchResult{Batch: req.Batch, StartedAt: startedAt}
|
|
for _, planned := range plannedReports {
|
|
resolved := planned.Resolved
|
|
item := batchReportResult(planned)
|
|
reportResult, err := generatePromptReport(ctx, promptReportRequest{
|
|
GenerateRequest: GenerateRequest{
|
|
Config: req.Config,
|
|
OutputPath: planned.OutputPath,
|
|
Notifier: req.Notifier,
|
|
Executor: req.Executor,
|
|
},
|
|
Resolved: resolved,
|
|
Collection: *collection,
|
|
Inspection: inspections[resolved.Definition.ID],
|
|
DebugWriter: debugWriter,
|
|
noNotify: true,
|
|
})
|
|
if reportResult != nil {
|
|
copyBatchReportDetails(&item, reportResult)
|
|
}
|
|
if err != nil {
|
|
item.Status = "failed"
|
|
item.Error = err.Error()
|
|
result.Failed++
|
|
} else {
|
|
item.Status = "succeeded"
|
|
result.Succeeded++
|
|
}
|
|
result.Reports = append(result.Reports, item)
|
|
}
|
|
result.Total = len(result.Reports)
|
|
batchNotification := notifyBatch(ctx, req.Config, req.Batch, batchRunID(startedAt, req.Batch), startedAt, result, plannedReports, req.Notifier)
|
|
if batchNotification != nil {
|
|
result.Notification = batchNotification
|
|
}
|
|
result.FinishedAt = time.Now()
|
|
return result, nil
|
|
}
|
|
return nil, fmt.Errorf("run is not implemented")
|
|
}
|
|
|
|
func copyBatchReportDetails(item *BatchReportResult, result *ReportResult) {
|
|
item.LLMDebugPath = result.LLMDebugPath
|
|
item.OutputPath = result.OutputPath
|
|
item.ProfileID = result.ProfileID
|
|
item.BackendID = result.BackendID
|
|
item.ModelName = result.ModelName
|
|
item.Timezone = result.Timezone
|
|
item.SourceWarnings = append([]weatherdata.SourceWarning(nil), result.SourceWarnings...)
|
|
item.ValidationStatus = result.ValidationStatus
|
|
}
|
|
|
|
func batchInspectionCandidates(req BatchRequest, now time.Time) ([]report.Resolved, error) {
|
|
location, err := timeutil.LoadLocation(req.Config.WeatherAPI.Timezone)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
registry, err := reportRegistry(req.Config)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ids := []report.ID{report.Tomorrow, report.Daily}
|
|
if req.Batch == BatchMorning {
|
|
ids = []report.ID{report.Today, report.Tomorrow, report.Daily}
|
|
}
|
|
date := timeutil.LocalDate(now, location).AddDate(0, 0, 2)
|
|
candidates := make([]report.Resolved, 0, len(ids))
|
|
for _, id := range ids {
|
|
resolveReq := report.ResolveRequest{Now: now, Location: location}
|
|
if id == report.Daily {
|
|
resolveReq.Date = date
|
|
}
|
|
resolved, err := registry.Resolve(id, resolveReq)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
candidates = append(candidates, resolved)
|
|
}
|
|
return candidates, nil
|
|
}
|
|
|
|
func batchReportResult(planned plannedBatchReport) BatchReportResult {
|
|
resolved := planned.Resolved
|
|
metadata := resolved.Metadata()
|
|
return BatchReportResult{
|
|
ReportID: resolved.Definition.ID,
|
|
ReportName: resolved.Definition.Name,
|
|
PromptID: resolved.Definition.PromptID,
|
|
RunID: metadata.RunID,
|
|
GeneratedAt: metadata.GeneratedAt,
|
|
ValidPeriod: metadata.ValidPeriod,
|
|
Timezone: "",
|
|
}
|
|
}
|
|
|
|
func ResolveGenerate(req GenerateRequest, now time.Time) (report.Resolved, error) {
|
|
location, err := timeutil.LoadLocation(req.Config.WeatherAPI.Timezone)
|
|
if err != nil {
|
|
return report.Resolved{}, err
|
|
}
|
|
id, err := report.IDForCommandName(string(req.Report))
|
|
if err != nil {
|
|
return report.Resolved{}, err
|
|
}
|
|
registry, err := reportRegistry(req.Config)
|
|
if err != nil {
|
|
return report.Resolved{}, err
|
|
}
|
|
return registry.Resolve(id, report.ResolveRequest{
|
|
Now: now,
|
|
Location: location,
|
|
Date: req.Date,
|
|
})
|
|
}
|
|
|
|
func reportRegistry(cfg config.Config) (report.Registry, error) {
|
|
overrides, err := cfg.ReportModuleOverrides()
|
|
if err != nil {
|
|
return report.Registry{}, err
|
|
}
|
|
registry, err := report.DefaultRegistry().WithModuleOverrides(overrides)
|
|
if err != nil {
|
|
return report.Registry{}, err
|
|
}
|
|
return registry, nil
|
|
}
|
|
|
|
func collectWeather(ctx context.Context, cfg config.Config, collector Collector) (*collect.Result, error) {
|
|
if collector == nil {
|
|
collector = defaultCollector{}
|
|
}
|
|
result, err := collector.Run(ctx, collect.Request{Config: cfg})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if result == nil {
|
|
return nil, fmt.Errorf("collect weather bundle: collector returned nil result")
|
|
}
|
|
if result.Bundle == nil {
|
|
return nil, fmt.Errorf("collect weather bundle: collector returned nil bundle")
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func notifyReport(ctx context.Context, cfg config.Config, resolved report.Resolved, outputPath, runID string, generatedAt time.Time, notifier Notifier) (*NotificationResult, error) {
|
|
notifier, enabled := reportNotifier(cfg, notifier)
|
|
if !enabled {
|
|
return nil, nil
|
|
}
|
|
notificationRequest, err := buildNotificationRequest(cfg, resolved, outputPath, runID, generatedAt)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result, err := notifier.Notify(ctx, notificationRequest)
|
|
if err != nil {
|
|
return result, &NotificationError{
|
|
Request: notificationRequest,
|
|
Err: fmt.Errorf("notify report %q run %q from output %q: %w", resolved.Definition.ID, runID, outputPath, err),
|
|
}
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func reportNotifier(cfg config.Config, notifier Notifier) (Notifier, bool) {
|
|
if !cfg.Notify.Distributor.Enabled {
|
|
return noopNotifier{}, false
|
|
}
|
|
if notifier != nil {
|
|
return notifier, true
|
|
}
|
|
return distributorNotifier{
|
|
client: distributoradapter.New(cfg.Notify.Distributor),
|
|
}, true
|
|
}
|
|
|
|
func buildNotificationRequest(cfg config.Config, resolved report.Resolved, outputPath, runID string, generatedAt time.Time) (NotificationRequest, error) {
|
|
values, err := distributorTemplateValuesForReport(cfg, resolved, runID, filepath.Base(outputPath))
|
|
if err != nil {
|
|
return NotificationRequest{}, err
|
|
}
|
|
bundleID, err := config.RenderDistributorBundleID(cfg.Notify.Distributor.BundleIDTemplate, values)
|
|
if err != nil {
|
|
return NotificationRequest{}, err
|
|
}
|
|
values.BundleID = bundleID
|
|
pipelineID, err := config.RenderDistributorPipelineID(cfg.Notify.Distributor.PipelineIDTemplate, values)
|
|
if err != nil {
|
|
return NotificationRequest{}, err
|
|
}
|
|
idempotencyKey, err := config.RenderDistributorIdempotencyKey(cfg.Notify.Distributor.IdempotencyKeyTemplate, values)
|
|
if err != nil {
|
|
return NotificationRequest{}, err
|
|
}
|
|
bundlePaths, err := renderDistributorReportBundlePaths(cfg, resolved, runID, outputPath, values)
|
|
if err != nil {
|
|
return NotificationRequest{}, err
|
|
}
|
|
return NotificationRequest{
|
|
ReportID: resolved.Definition.ID,
|
|
RunID: runID,
|
|
PipelineID: pipelineID,
|
|
BundleID: bundleID,
|
|
IdempotencyKey: idempotencyKey,
|
|
ReportPath: outputPath,
|
|
BundlePaths: bundlePaths,
|
|
CreatedAt: generatedAt,
|
|
}, nil
|
|
}
|
|
|
|
func distributorTemplateValuesForReport(cfg config.Config, resolved report.Resolved, runID string, outputName string) (config.DistributorTemplateValues, error) {
|
|
values := config.DistributorTemplateValues{
|
|
LocationID: cfg.Location.ID,
|
|
ReportID: string(resolved.Definition.ID),
|
|
RunID: runID,
|
|
ArtifactGroup: resolved.Definition.ArtifactGroup,
|
|
BatchOutputName: outputName,
|
|
}
|
|
if values.BatchOutputName == "" {
|
|
var err error
|
|
values.BatchOutputName, err = resolved.OutputName()
|
|
if err != nil {
|
|
return config.DistributorTemplateValues{}, err
|
|
}
|
|
}
|
|
if err := addDistributorValidPeriodValues(&values, resolved.ValidPeriod, cfg.WeatherAPI.Timezone); err != nil {
|
|
return config.DistributorTemplateValues{}, err
|
|
}
|
|
return values, nil
|
|
}
|
|
|
|
func renderDistributorReportBundlePaths(cfg config.Config, resolved report.Resolved, runID string, sourcePath string, values config.DistributorTemplateValues) ([]string, error) {
|
|
templates, name, err := distributorPathTemplatesForReport(cfg, resolved.Definition)
|
|
if err != nil {
|
|
return nil, distributorReportPathError(resolved.Definition.ID, runID, sourcePath, err)
|
|
}
|
|
paths, err := config.RenderDistributorReportPaths(name, templates, values)
|
|
if err != nil {
|
|
return nil, distributorReportPathError(resolved.Definition.ID, runID, sourcePath, err)
|
|
}
|
|
return paths, nil
|
|
}
|
|
|
|
func distributorPathTemplatesForReport(cfg config.Config, definition report.Definition) ([]string, string, error) {
|
|
overrides, err := cfg.ReportDistributorPathOverrides()
|
|
if err != nil {
|
|
return nil, "", err
|
|
}
|
|
if templates, ok := overrides[definition.ID]; ok {
|
|
return append([]string(nil), templates...), fmt.Sprintf("reports.%s.distributor.path_templates", definition.ID), nil
|
|
}
|
|
if len(definition.DistributorPathTemplates) > 0 {
|
|
return append([]string(nil), definition.DistributorPathTemplates...), fmt.Sprintf("report.%s.distributor_path_templates", definition.ID), nil
|
|
}
|
|
return nil, "", fmt.Errorf("no distributor path templates configured")
|
|
}
|
|
|
|
func distributorReportPathError(id report.ID, runID string, sourcePath string, err error) error {
|
|
if sourcePath != "" {
|
|
return fmt.Errorf("report %q run %q source path %q: %w", id, runID, sourcePath, err)
|
|
}
|
|
return fmt.Errorf("report %q run %q: %w", id, runID, err)
|
|
}
|
|
|
|
func addDistributorValidPeriodValues(values *config.DistributorTemplateValues, period timeutil.Period, timezone string) error {
|
|
location, err := timeutil.LoadLocation(timezone)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
start := period.Start.In(location)
|
|
end := period.End.In(location)
|
|
values.ValidStartDate = start.Format(timeutil.DateLayout)
|
|
values.ValidEndDate = end.Format(timeutil.DateLayout)
|
|
values.ValidStartTime = start.Format("1504")
|
|
values.ValidEndTime = end.Format("1504")
|
|
values.ValidStartStamp = start.Format("2006-01-02T1504")
|
|
values.ValidEndStamp = end.Format("2006-01-02T1504")
|
|
return nil
|
|
}
|
|
|
|
type noopNotifier struct{}
|
|
|
|
func (noopNotifier) Notify(context.Context, NotificationRequest) (*NotificationResult, error) {
|
|
return nil, nil
|
|
}
|
|
|
|
type distributorNotifier struct {
|
|
client *distributoradapter.Client
|
|
}
|
|
|
|
func (n distributorNotifier) Notify(ctx context.Context, req NotificationRequest) (*NotificationResult, error) {
|
|
result, err := n.client.Upload(ctx, distributoradapter.UploadRequest{
|
|
PipelineID: req.PipelineID,
|
|
BundleID: req.BundleID,
|
|
IdempotencyKey: req.IdempotencyKey,
|
|
Files: distributorUploadFiles(req.ReportPath, req.BundlePaths),
|
|
CreatedAt: req.CreatedAt,
|
|
})
|
|
notification := &NotificationResult{
|
|
PipelineID: req.PipelineID,
|
|
BundleID: req.BundleID,
|
|
IdempotencyKey: req.IdempotencyKey,
|
|
RunID: result.RunID,
|
|
Status: result.Status,
|
|
UploadStatus: result.UploadStatus,
|
|
StatusError: result.StatusError,
|
|
}
|
|
if result.RunStatus != nil {
|
|
if result.RunStatus.PipelineID != "" {
|
|
notification.PipelineID = result.RunStatus.PipelineID
|
|
}
|
|
notification.AcceptedAt = result.RunStatus.AcceptedAt
|
|
notification.StartedAt = result.RunStatus.StartedAt
|
|
notification.FinishedAt = result.RunStatus.FinishedAt
|
|
notification.Report = append([]byte(nil), result.RunStatus.Report...)
|
|
notification.Error = result.RunStatus.Error
|
|
}
|
|
if err != nil {
|
|
return notification, err
|
|
}
|
|
return notification, nil
|
|
}
|
|
|
|
func (n distributorNotifier) NotifyBatch(ctx context.Context, req batchNotificationRequest) (*NotificationResult, error) {
|
|
result, err := n.client.Upload(ctx, batchDistributorUploadRequest(req))
|
|
notification := notificationResultFromUpload(req.PipelineID, req.BundleID, req.IdempotencyKey, result)
|
|
if err != nil {
|
|
return notification, err
|
|
}
|
|
return notification, nil
|
|
}
|
|
|
|
func notificationResultFromUpload(pipelineID string, bundleID string, idempotencyKey string, result distributoradapter.UploadResult) *NotificationResult {
|
|
notification := &NotificationResult{
|
|
PipelineID: pipelineID,
|
|
BundleID: bundleID,
|
|
IdempotencyKey: idempotencyKey,
|
|
RunID: result.RunID,
|
|
Status: result.Status,
|
|
UploadStatus: result.UploadStatus,
|
|
StatusError: result.StatusError,
|
|
}
|
|
if result.RunStatus != nil {
|
|
if result.RunStatus.PipelineID != "" {
|
|
notification.PipelineID = result.RunStatus.PipelineID
|
|
}
|
|
notification.AcceptedAt = result.RunStatus.AcceptedAt
|
|
notification.StartedAt = result.RunStatus.StartedAt
|
|
notification.FinishedAt = result.RunStatus.FinishedAt
|
|
notification.Report = append([]byte(nil), result.RunStatus.Report...)
|
|
notification.Error = result.RunStatus.Error
|
|
}
|
|
return notification
|
|
}
|
|
|
|
func distributorUploadFiles(sourcePath string, bundlePaths []string) []distributoradapter.UploadFile {
|
|
files := make([]distributoradapter.UploadFile, 0, len(bundlePaths))
|
|
for _, bundlePath := range bundlePaths {
|
|
files = append(files, distributoradapter.UploadFile{
|
|
SourcePath: sourcePath,
|
|
BundlePath: bundlePath,
|
|
})
|
|
}
|
|
return files
|
|
}
|
|
|
|
func BuildModuleSnapshot(req ModuleSnapshotRequest, bundle *weatherdata.Bundle) (module.Snapshot, error) {
|
|
reportFacts, err := BuildReportFacts(req, bundle)
|
|
if err != nil {
|
|
return module.Snapshot{}, err
|
|
}
|
|
return BuildModuleSnapshotFromFacts(req, reportFacts)
|
|
}
|
|
|
|
func BuildReportFacts(req ModuleSnapshotRequest, bundle *weatherdata.Bundle) (ReportFacts, error) {
|
|
collected := facts.BuildCollected(bundle)
|
|
derived, err := buildDerivedFacts(req.Config, req.Resolved, collected)
|
|
if err != nil {
|
|
return ReportFacts{}, err
|
|
}
|
|
return ReportFacts{
|
|
Collected: collected,
|
|
Derived: derived,
|
|
}, nil
|
|
}
|
|
|
|
func BuildModuleSnapshotFromFacts(req ModuleSnapshotRequest, reportFacts ReportFacts) (module.Snapshot, error) {
|
|
if !req.Resolved.ValidPeriod.IsValid() {
|
|
return module.Snapshot{}, fmt.Errorf("resolved valid period is required")
|
|
}
|
|
registry, err := briefing.DefaultModuleRegistry()
|
|
if err != nil {
|
|
return module.Snapshot{}, err
|
|
}
|
|
moduleContext := briefing.ModuleContext{
|
|
Resolved: req.Resolved,
|
|
Collected: reportFacts.Collected,
|
|
Derived: reportFacts.Derived,
|
|
Units: req.Config.WeatherAPI.Units,
|
|
Timezone: req.Config.WeatherAPI.Timezone,
|
|
Location: briefingLocation(req.Config),
|
|
}
|
|
var outputs []module.Output
|
|
for _, item := range req.Resolved.Definition.Modules {
|
|
output, err := registry.BuildModule(moduleContext, item)
|
|
if err != nil {
|
|
return module.Snapshot{}, err
|
|
}
|
|
if output == nil {
|
|
continue
|
|
}
|
|
outputs = append(outputs, *output)
|
|
}
|
|
return module.NewSnapshot(outputs)
|
|
}
|
|
|
|
func briefingBuildContext(cfg config.Config, resolved report.Resolved, collected facts.CollectedFacts) briefing.BuildContext {
|
|
return briefing.BuildContext{
|
|
Resolved: resolved,
|
|
Bundle: collected.Bundle(),
|
|
Units: cfg.WeatherAPI.Units,
|
|
Timezone: cfg.WeatherAPI.Timezone,
|
|
Location: briefingLocation(cfg),
|
|
}
|
|
}
|
|
|
|
func promptMetadata(metadata briefing.Metadata) promptinput.Metadata {
|
|
return promptinput.Metadata{
|
|
RunID: metadata.RunID,
|
|
ReportID: metadata.ReportID,
|
|
Variant: metadata.Variant,
|
|
PromptID: metadata.PromptID,
|
|
GeneratedAt: metadata.GeneratedAt,
|
|
Timezone: metadata.Timezone,
|
|
ValidPeriod: metadata.ValidPeriod,
|
|
SourceWarnings: metadata.SourceWarnings,
|
|
}
|
|
}
|
|
|
|
func buildDerivedFacts(cfg config.Config, resolved report.Resolved, collected facts.CollectedFacts) (facts.DerivedFacts, error) {
|
|
dayparts := make([]forecast.DaypartDefinition, 0, len(cfg.Dayparts))
|
|
for _, daypart := range cfg.Dayparts {
|
|
dayparts = append(dayparts, forecast.DaypartDefinition{
|
|
Name: daypart.Name,
|
|
Start: daypart.Start,
|
|
End: daypart.End,
|
|
})
|
|
}
|
|
return facts.BuildDerived(facts.BuildDerivedRequest{
|
|
Resolved: resolved,
|
|
Timezone: cfg.WeatherAPI.Timezone,
|
|
Dayparts: dayparts,
|
|
Collected: collected,
|
|
})
|
|
}
|
|
|
|
func briefingLocation(cfg config.Config) *briefing.LocationContext {
|
|
location := briefing.LocationContext{
|
|
ID: cfg.Location.ID,
|
|
Name: cfg.Location.Name,
|
|
Region: cfg.Location.Region,
|
|
Timezone: cfg.WeatherAPI.Timezone,
|
|
}
|
|
if location.ID == "" && location.Name == "" && location.Region == "" && location.Timezone == "" {
|
|
return nil
|
|
}
|
|
return &location
|
|
}
|
|
|
|
func generatedReportError(resolved report.Resolved, runID string, operation string, err error) error {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("generate report %q run %q: %s: %w", resolved.Definition.ID, runID, operation, err)
|
|
}
|