Compare commits
6 Commits
6ee81c3581
..
dev
| Author | SHA1 | Date | |
|---|---|---|---|
| 1ec1557a9a | |||
| 974ba8d638 | |||
| fa16d4080a | |||
| 6ba964275e | |||
| e4f7c4a04c | |||
| 51a1a6aa82 |
@@ -23,7 +23,7 @@ import (
|
||||
"tm/pkg/schedule"
|
||||
)
|
||||
|
||||
// aiPipelineDailyRunMu ensures only one AI pipeline daily-run executes at a time (startup catch-up and cron).
|
||||
// aiPipelineDailyRunMu ensures only one AI pipeline auto run executes at a time (startup catch-up and cron).
|
||||
var aiPipelineDailyRunMu sync.Mutex
|
||||
|
||||
// Init Application Configuration
|
||||
@@ -98,8 +98,8 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
|
||||
"translation_enabled": config.Worker.TranslationEnabled,
|
||||
"translation_interval": config.Worker.TranslationInterval,
|
||||
"notification_interval": config.Worker.NotificationInterval,
|
||||
"ai_pipeline_daily_enabled": config.Worker.AIPipelineAutoEnabled,
|
||||
"ai_pipeline_daily_interval": config.Worker.AIPipelineAutoInterval,
|
||||
"ai_pipeline_auto_enabled": config.Worker.AIPipelineAutoEnabled,
|
||||
"ai_pipeline_auto_interval": config.Worker.AIPipelineAutoInterval,
|
||||
})
|
||||
// Initialize repositories
|
||||
noticeRepo := notice.NewRepository(mongoManager, appLogger)
|
||||
@@ -231,8 +231,8 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
|
||||
})
|
||||
|
||||
if !config.Worker.AIPipelineAutoEnabled {
|
||||
appLogger.Info("AI pipeline daily-run worker disabled by configuration", map[string]interface{}{
|
||||
"ai_pipeline_daily_enabled": false,
|
||||
appLogger.Info("AI pipeline auto worker disabled by configuration", map[string]interface{}{
|
||||
"ai_pipeline_auto_enabled": false,
|
||||
})
|
||||
} else if aiClient != nil {
|
||||
dailyJobTracker := schedule.NewDailyJobTracker(mongoManager, appLogger)
|
||||
@@ -246,13 +246,13 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
|
||||
|
||||
completed, checkErr := dailyJobTracker.IsCompleted(ctx, schedule.AIPipelineDailyJobName, today)
|
||||
if checkErr != nil {
|
||||
appLogger.Error("Failed to check AI pipeline daily-run completion status", map[string]interface{}{
|
||||
appLogger.Error("Failed to check AI pipeline auto completion status", map[string]interface{}{
|
||||
"trigger": trigger,
|
||||
"error": checkErr.Error(),
|
||||
"date": today.Format("02/01/2006"),
|
||||
})
|
||||
} else if completed {
|
||||
appLogger.Info("AI pipeline daily-run skipped: already completed today", map[string]interface{}{
|
||||
appLogger.Info("AI pipeline auto skipped: already completed today", map[string]interface{}{
|
||||
"trigger": trigger,
|
||||
"date": today.Format("02/01/2006"),
|
||||
})
|
||||
@@ -260,11 +260,11 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
|
||||
}
|
||||
|
||||
if trigger == "startup" {
|
||||
appLogger.Info("Running startup catch-up for today's AI pipeline daily-run", map[string]interface{}{
|
||||
appLogger.Info("Running startup catch-up for today's AI pipeline auto run", map[string]interface{}{
|
||||
"date": today.Format("02/01/2006"),
|
||||
})
|
||||
} else {
|
||||
appLogger.Info("Running scheduled AI pipeline daily-run", map[string]interface{}{
|
||||
appLogger.Info("Running scheduled AI pipeline auto run", map[string]interface{}{
|
||||
"date": today.Format("02/01/2006"),
|
||||
})
|
||||
}
|
||||
@@ -272,12 +272,12 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
|
||||
worker := workers.NewAIPipelineDailyWorker(appLogger, aiClient)
|
||||
if err := worker.Run(); err != nil {
|
||||
if errors.Is(err, ai_summarizer.ErrPipelineJobRunning) {
|
||||
appLogger.Info("AI pipeline daily-run deferred: job already running", map[string]interface{}{
|
||||
appLogger.Info("AI pipeline auto deferred: job already running", map[string]interface{}{
|
||||
"trigger": trigger,
|
||||
"date": today.Format("02/01/2006"),
|
||||
})
|
||||
} else {
|
||||
appLogger.Error("AI pipeline daily-run failed", map[string]interface{}{
|
||||
appLogger.Error("AI pipeline auto failed", map[string]interface{}{
|
||||
"trigger": trigger,
|
||||
"date": today.Format("02/01/2006"),
|
||||
"error": err.Error(),
|
||||
@@ -287,14 +287,14 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
|
||||
}
|
||||
|
||||
if markErr := dailyJobTracker.MarkCompleted(ctx, schedule.AIPipelineDailyJobName, today); markErr != nil {
|
||||
appLogger.Error("Failed to mark AI pipeline daily-run as completed", map[string]interface{}{
|
||||
appLogger.Error("Failed to mark AI pipeline auto as completed", map[string]interface{}{
|
||||
"trigger": trigger,
|
||||
"date": today.Format("02/01/2006"),
|
||||
"error": markErr.Error(),
|
||||
})
|
||||
}
|
||||
|
||||
appLogger.Info("AI pipeline daily-run completed successfully", map[string]interface{}{
|
||||
appLogger.Info("AI pipeline auto completed successfully", map[string]interface{}{
|
||||
"trigger": trigger,
|
||||
"date": today.Format("02/01/2006"),
|
||||
})
|
||||
@@ -305,11 +305,11 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
|
||||
}
|
||||
|
||||
scheduler.AddJob(schedule.Job{
|
||||
Name: "AI Pipeline Daily Run Job",
|
||||
Name: "AI Pipeline Auto Job",
|
||||
Func: func() { runAIPipelineDaily("scheduled") },
|
||||
Expr: config.Worker.AIPipelineAutoInterval,
|
||||
})
|
||||
appLogger.Info("Scheduled AI pipeline daily-run worker", map[string]interface{}{
|
||||
appLogger.Info("Scheduled AI pipeline auto worker", map[string]interface{}{
|
||||
"interval": config.Worker.AIPipelineAutoInterval,
|
||||
})
|
||||
|
||||
@@ -317,7 +317,7 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
|
||||
runAIPipelineDaily("startup")
|
||||
}()
|
||||
} else {
|
||||
appLogger.Warn("AI summarizer client not available, AI pipeline daily-run worker is disabled", map[string]interface{}{})
|
||||
appLogger.Warn("AI summarizer client not available, AI pipeline auto worker is disabled", map[string]interface{}{})
|
||||
}
|
||||
|
||||
// After a server restart, process any unprocessed notices (including today's) without blocking startup.
|
||||
|
||||
@@ -36,12 +36,12 @@ type WorkerConfig struct {
|
||||
TranslationEnabled bool `env:"WORKER_TRANSLATION_ENABLED" envDefault:"true"`
|
||||
// NotificationInterval schedules promotion of due scheduled notifications from pending to sent.
|
||||
NotificationInterval string `env:"WORKER_NOTIFICATION_INTERVAL" envDefault:"0 * * * * *"`
|
||||
// AIPipelineAutoEnabled schedules the daily AI pipeline daily-run on the worker.
|
||||
// AIPipelineAutoEnabled schedules the daily AI pipeline auto run on the worker.
|
||||
// On-demand pipeline routes on web are unaffected.
|
||||
AIPipelineAutoEnabled bool `env:"WORKER_AI_PIPELINE_AUTO_ENABLED" envDefault:"true"`
|
||||
// AIPipelineAutoInterval runs daily-run once per day by default at 11:00 UTC (after TED scrape at 10:00).
|
||||
// AIPipelineAutoInterval runs pipeline auto once per day by default at 11:00 UTC (after TED scrape at 10:00).
|
||||
AIPipelineAutoInterval string `env:"WORKER_AI_PIPELINE_AUTO_INTERVAL" envDefault:"0 0 11 * * *"`
|
||||
// RecommendationRefreshAfterPipelineEnabled refreshes cached company recommendations after the daily AI pipeline run.
|
||||
// RecommendationRefreshAfterPipelineEnabled refreshes cached company recommendations after the daily AI pipeline auto run.
|
||||
RecommendationRefreshAfterPipelineEnabled bool `env:"WORKER_RECOMMENDATION_REFRESH" envDefault:"true"`
|
||||
}
|
||||
|
||||
|
||||
@@ -8,13 +8,13 @@ import (
|
||||
"tm/pkg/logger"
|
||||
)
|
||||
|
||||
// AIPipelineDailyWorker triggers the Opplens AI pipeline daily run (sync + full pipeline on newly synced tenders).
|
||||
// AIPipelineDailyWorker triggers the scheduled Opplens AI pipeline auto run.
|
||||
type AIPipelineDailyWorker struct {
|
||||
Logger logger.Logger
|
||||
AIClient *ai_summarizer.Client
|
||||
}
|
||||
|
||||
// NewAIPipelineDailyWorker creates a worker that calls POST /pipeline/daily-run on the AI service.
|
||||
// NewAIPipelineDailyWorker creates a worker that calls POST /pipeline/auto on the AI service.
|
||||
func NewAIPipelineDailyWorker(log logger.Logger, aiClient *ai_summarizer.Client) *AIPipelineDailyWorker {
|
||||
return &AIPipelineDailyWorker{
|
||||
Logger: log,
|
||||
@@ -22,28 +22,28 @@ func NewAIPipelineDailyWorker(log logger.Logger, aiClient *ai_summarizer.Client)
|
||||
}
|
||||
}
|
||||
|
||||
// Run starts the pipeline daily-run job. A 409 response means a run is already in progress and is returned as an error
|
||||
// Run starts the pipeline auto job. A 409 response means a run is already in progress and is returned as an error
|
||||
// so the daily job tracker does not mark today as completed before the run actually starts.
|
||||
func (w *AIPipelineDailyWorker) Run() error {
|
||||
if w.AIClient == nil {
|
||||
w.Logger.Warn("AI pipeline daily worker skipped: AI client is not configured", map[string]interface{}{})
|
||||
w.Logger.Warn("AI pipeline auto worker skipped: AI client is not configured", map[string]interface{}{})
|
||||
return nil
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
w.Logger.Info("AI pipeline daily worker started", map[string]interface{}{})
|
||||
w.Logger.Info("AI pipeline auto worker started", map[string]interface{}{})
|
||||
|
||||
resp, err := w.AIClient.PipelineDailyRun(ctx)
|
||||
resp, err := w.AIClient.PipelineAuto(ctx)
|
||||
if err != nil {
|
||||
w.Logger.Error("AI pipeline daily worker failed", map[string]interface{}{
|
||||
w.Logger.Error("AI pipeline auto worker failed", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
w.Logger.Info("AI pipeline daily worker triggered successfully", map[string]interface{}{
|
||||
w.Logger.Info("AI pipeline auto worker triggered successfully", map[string]interface{}{
|
||||
"status": resp.Status,
|
||||
"report": resp.Report,
|
||||
})
|
||||
|
||||
@@ -44,11 +44,11 @@ type (
|
||||
|
||||
// AddressForm represents the form for company address
|
||||
AddressForm struct {
|
||||
Street string `json:"street" valid:"required,length(5|200)" example:"123 Main St"`
|
||||
City string `json:"city" valid:"required,length(2|100)" example:"New York"`
|
||||
State string `json:"state" valid:"required,length(2|100)" example:"NY"`
|
||||
PostalCode string `json:"postal_code" valid:"required,length(3|20)" example:"10001"`
|
||||
Country string `json:"country" valid:"required,length(2|100)" example:"US"`
|
||||
Street string `json:"street" valid:"optional,length(5|200)" example:"123 Main St"`
|
||||
City string `json:"city" valid:"optional,length(2|100)" example:"New York"`
|
||||
State string `json:"state" valid:"optional,length(2|100)" example:"NY"`
|
||||
PostalCode string `json:"postal_code" valid:"optional,length(3|20)" example:"10001"`
|
||||
Country string `json:"country" valid:"optional,length(2|100)" example:"US"`
|
||||
}
|
||||
|
||||
// CompanyTagsForm represents the form for company tags
|
||||
|
||||
@@ -13,7 +13,7 @@ import (
|
||||
|
||||
// AIPipelineStatusClient reports pipeline job status from the AI service.
|
||||
type AIPipelineStatusClient interface {
|
||||
GetPipelineLastRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error)
|
||||
GetPipelineLastAutoRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error)
|
||||
}
|
||||
|
||||
const (
|
||||
@@ -22,7 +22,7 @@ const (
|
||||
recommendationRefreshConcurrency = 3
|
||||
)
|
||||
|
||||
// ScheduleRefreshCachedAIRecommendationsAfterPipeline waits for the AI daily pipeline to finish,
|
||||
// ScheduleRefreshCachedAIRecommendationsAfterPipeline waits for the AI pipeline auto run to finish,
|
||||
// then re-fetches ranked tenders for every company that already has a recommendation cache.
|
||||
func (s *companyService) ScheduleRefreshCachedAIRecommendationsAfterPipeline() {
|
||||
if s.aiRecommendationClient == nil {
|
||||
@@ -45,7 +45,7 @@ func (s *companyService) ScheduleRefreshCachedAIRecommendationsAfterPipeline() {
|
||||
}
|
||||
|
||||
func (s *companyService) refreshCachedAIRecommendationsAfterPipeline(ctx context.Context) error {
|
||||
if err := s.waitForPipelineDailyRunCompletion(ctx); err != nil {
|
||||
if err := s.waitForPipelineAutoCompletion(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -95,7 +95,7 @@ func (s *companyService) refreshCachedAIRecommendationsAfterPipeline(ctx context
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *companyService) waitForPipelineDailyRunCompletion(ctx context.Context) error {
|
||||
func (s *companyService) waitForPipelineAutoCompletion(ctx context.Context) error {
|
||||
if s.aiPipelineStatusClient == nil {
|
||||
s.logger.Info("AI pipeline status client not configured; waiting before recommendation refresh", map[string]interface{}{
|
||||
"delay_sec": int(recommendationPipelineWaitInitialDelay.Seconds()),
|
||||
@@ -134,9 +134,9 @@ func (s *companyService) waitForPipelineDailyRunCompletion(ctx context.Context)
|
||||
}
|
||||
}
|
||||
|
||||
report, err := s.aiPipelineStatusClient.GetPipelineLastRun(ctx)
|
||||
report, err := s.aiPipelineStatusClient.GetPipelineLastAutoRun(ctx)
|
||||
if err != nil {
|
||||
s.logger.Warn("Failed to read AI pipeline last-run status", map[string]interface{}{
|
||||
s.logger.Warn("Failed to read AI pipeline last-auto-run status", map[string]interface{}{
|
||||
"attempt": attempt,
|
||||
"error": err.Error(),
|
||||
})
|
||||
@@ -147,8 +147,8 @@ func (s *companyService) waitForPipelineDailyRunCompletion(ctx context.Context)
|
||||
continue
|
||||
}
|
||||
|
||||
if !isPipelineDailyRunInProgress(report) {
|
||||
s.logger.Info("AI pipeline daily-run finished; refreshing recommendation caches", map[string]interface{}{
|
||||
if !isPipelineRunInProgress(report) {
|
||||
s.logger.Info("AI pipeline auto run finished; refreshing recommendation caches", map[string]interface{}{
|
||||
"attempt": attempt,
|
||||
"status": strings.TrimSpace(report.Status),
|
||||
})
|
||||
@@ -167,7 +167,7 @@ func (s *companyService) waitForPipelineDailyRunCompletion(ctx context.Context)
|
||||
return nil
|
||||
}
|
||||
|
||||
func isPipelineDailyRunInProgress(report *ai_summarizer.PipelineReportResponse) bool {
|
||||
func isPipelineRunInProgress(report *ai_summarizer.PipelineReportResponse) bool {
|
||||
if report == nil {
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -6,7 +6,7 @@ import (
|
||||
"tm/pkg/ai_summarizer"
|
||||
)
|
||||
|
||||
func TestIsPipelineDailyRunInProgress(t *testing.T) {
|
||||
func TestIsPipelineRunInProgress(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
report *ai_summarizer.PipelineReportResponse
|
||||
@@ -22,8 +22,8 @@ func TestIsPipelineDailyRunInProgress(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := isPipelineDailyRunInProgress(tt.report); got != tt.want {
|
||||
t.Fatalf("isPipelineDailyRunInProgress() = %v, want %v", got, tt.want)
|
||||
if got := isPipelineRunInProgress(tt.report); got != tt.want {
|
||||
t.Fatalf("isPipelineRunInProgress() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -226,6 +226,7 @@ func (s *TEDScraper) mapPriorInformationNoticeToTender(pin *PriorInformationNoti
|
||||
t.BuyerOrganization = s.mapOrganization(buyerOrg, "buyer")
|
||||
}
|
||||
}
|
||||
backfillNoticeLocationFromBuyer(t)
|
||||
|
||||
reviewID := pin.GetReviewOrganizationID()
|
||||
if reviewID != "" {
|
||||
@@ -407,6 +408,7 @@ func (s *TEDScraper) mapToTender(cn *ContractNotice, sourceFileURL, sourceFileNa
|
||||
t.BuyerOrganization = s.mapOrganization(buyerOrg, "buyer")
|
||||
}
|
||||
}
|
||||
backfillNoticeLocationFromBuyer(t)
|
||||
|
||||
// Set review organization
|
||||
reviewID := cn.GetReviewOrganizationID()
|
||||
@@ -570,6 +572,7 @@ func (s *TEDScraper) mapContractAwardNoticeToTender(can *ContractAwardNotice, so
|
||||
t.BuyerOrganization = s.mapOrganization(buyerOrg, "buyer")
|
||||
}
|
||||
}
|
||||
backfillNoticeLocationFromBuyer(t)
|
||||
|
||||
// Set all organizations
|
||||
if orgs := can.GetOrganizations(); orgs != nil {
|
||||
@@ -667,6 +670,25 @@ func (s *TEDScraper) mapOrganization(tedOrg *Organization, role string) *notice.
|
||||
return org
|
||||
}
|
||||
|
||||
func backfillNoticeLocationFromBuyer(t *notice.Notice) {
|
||||
if t == nil || t.BuyerOrganization == nil {
|
||||
return
|
||||
}
|
||||
buyer := t.BuyerOrganization.Address
|
||||
if strings.TrimSpace(t.CountryCode) == "" && strings.TrimSpace(buyer.CountryCode) != "" {
|
||||
t.CountryCode = strings.TrimSpace(buyer.CountryCode)
|
||||
}
|
||||
if strings.TrimSpace(t.CityName) == "" && strings.TrimSpace(buyer.CityName) != "" {
|
||||
t.CityName = strings.TrimSpace(buyer.CityName)
|
||||
}
|
||||
if strings.TrimSpace(t.PostalCode) == "" && strings.TrimSpace(buyer.PostalZone) != "" {
|
||||
t.PostalCode = strings.TrimSpace(buyer.PostalZone)
|
||||
}
|
||||
if strings.TrimSpace(t.RegionCode) == "" && strings.TrimSpace(buyer.CountrySubentityCode) != "" {
|
||||
t.RegionCode = strings.TrimSpace(buyer.CountrySubentityCode)
|
||||
}
|
||||
}
|
||||
|
||||
func backfillNoticeEstimatedValueFromLots(t *notice.Notice) {
|
||||
if t == nil || t.EstimatedValue > 0 {
|
||||
return
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
package ted
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"tm/internal/notice"
|
||||
)
|
||||
|
||||
func TestBackfillNoticeLocationFromBuyer(t *testing.T) {
|
||||
n := ¬ice.Notice{
|
||||
BuyerOrganization: ¬ice.Organization{
|
||||
Address: notice.Address{
|
||||
CountryCode: "POL",
|
||||
CityName: "Warsaw",
|
||||
PostalZone: "00-001",
|
||||
CountrySubentityCode: "PL14",
|
||||
},
|
||||
},
|
||||
}
|
||||
backfillNoticeLocationFromBuyer(n)
|
||||
|
||||
if n.CountryCode != "POL" {
|
||||
t.Fatalf("expected country POL, got %q", n.CountryCode)
|
||||
}
|
||||
if n.CityName != "Warsaw" {
|
||||
t.Fatalf("expected city Warsaw, got %q", n.CityName)
|
||||
}
|
||||
if n.PostalCode != "00-001" {
|
||||
t.Fatalf("expected postal 00-001, got %q", n.PostalCode)
|
||||
}
|
||||
if n.RegionCode != "PL14" {
|
||||
t.Fatalf("expected region PL14, got %q", n.RegionCode)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackfillNoticeLocationFromBuyerDoesNotOverwrite(t *testing.T) {
|
||||
n := ¬ice.Notice{
|
||||
CountryCode: "DEU",
|
||||
CityName: "Berlin",
|
||||
PostalCode: "10115",
|
||||
RegionCode: "DE300",
|
||||
BuyerOrganization: ¬ice.Organization{
|
||||
Address: notice.Address{
|
||||
CountryCode: "POL",
|
||||
CityName: "Warsaw",
|
||||
PostalZone: "00-001",
|
||||
CountrySubentityCode: "PL14",
|
||||
},
|
||||
},
|
||||
}
|
||||
backfillNoticeLocationFromBuyer(n)
|
||||
|
||||
if n.CountryCode != "DEU" {
|
||||
t.Fatalf("expected country DEU, got %q", n.CountryCode)
|
||||
}
|
||||
if n.CityName != "Berlin" {
|
||||
t.Fatalf("expected city Berlin, got %q", n.CityName)
|
||||
}
|
||||
if n.PostalCode != "10115" {
|
||||
t.Fatalf("expected postal 10115, got %q", n.PostalCode)
|
||||
}
|
||||
if n.RegionCode != "DE300" {
|
||||
t.Fatalf("expected region DE300, got %q", n.RegionCode)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user