Compare commits

...

6 Commits

Author SHA1 Message Date
m.nazemi 1ec1557a9a Merge pull request 'Update AddressForm fields to optional in company form validation' (#54) from TM-687 into develop
continuous-integration/drone/push Build is passing
Reviewed-on: https://repo.ravanertebat.com/TM/tm_back/pulls/54
Reviewed-by: Hadi Barzegar <barzagarhadi@gmail.com>
2026-07-14 00:58:06 +03:30
m.nazemi 974ba8d638 Merge branch 'develop' into TM-687 2026-07-14 00:57:12 +03:30
m.nazemi fa16d4080a Merge pull request 'Add unit tests for backfilling notice location from buyer' (#53) from TM-485 into develop
continuous-integration/drone/push Build is passing
Reviewed-on: https://repo.ravanertebat.com/TM/tm_back/pulls/53
Reviewed-by: Hadi Barzegar <barzagarhadi@gmail.com>
2026-07-13 14:20:08 +03:30
Mazyar 6ba964275e Update AddressForm fields to optional in company form validation
- Changed validation rules for AddressForm fields from required to optional, allowing for more flexible input during company address submissions.
- This adjustment enhances the usability of the form by accommodating scenarios where address details may not be provided.
2026-07-13 14:17:46 +03:30
Mazyar e4f7c4a04c Add unit tests for backfilling notice location from buyer
- Introduced `mapper_test.go` to validate the functionality of backfilling notice location details from the buyer organization.
- Implemented two test cases: one to ensure the location fields are correctly populated from the buyer when they are empty, and another to verify that existing values are not overwritten.
- Enhanced the `mapper.go` file by adding the `backfillNoticeLocationFromBuyer` function to facilitate this logic.

This update improves the reliability of notice data handling within the tender management system by ensuring accurate location information is maintained.
2026-07-13 13:44:31 +03:30
Mazyar 51a1a6aa82 Refactor AI pipeline terminology for consistency
continuous-integration/drone/push Build is passing
- Updated comments and logging messages in the worker and related files to replace "daily-run" with "auto run" for clarity and consistency.
- Adjusted the `WorkerConfig` struct to reflect the new terminology in configuration settings.
- Renamed functions and test cases to align with the updated terminology, enhancing code readability and maintainability.

This change improves the clarity of the AI pipeline's functionality within the tender management system.
2026-07-13 11:54:58 +03:30
8 changed files with 130 additions and 44 deletions
+16 -16
View File
@@ -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.
+3 -3
View File
@@ -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 -8
View File
@@ -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,
})
+5 -5
View File
@@ -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)
}
})
}
+22
View File
@@ -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
+64
View File
@@ -0,0 +1,64 @@
package ted
import (
"testing"
"tm/internal/notice"
)
func TestBackfillNoticeLocationFromBuyer(t *testing.T) {
n := &notice.Notice{
BuyerOrganization: &notice.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 := &notice.Notice{
CountryCode: "DEU",
CityName: "Berlin",
PostalCode: "10115",
RegionCode: "DE300",
BuyerOrganization: &notice.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)
}
}