From fafccd0d74eecbeb07ce90463124560cf2159fde Mon Sep 17 00:00:00 2001 From: "n.nakhostin" Date: Tue, 4 Nov 2025 16:51:45 +0330 Subject: [PATCH] Refactor Worker Initialization to Integrate GLM SDK - Updated the worker initialization process to include the new GLM SDK, enhancing the worker's capabilities for translation tasks. - Modified the InitWorker function and NewNoticeWorker constructor to accept the GLM service, ensuring a cohesive integration. - Implemented the GLM service initialization and logging for successful setup, improving maintainability and usability. - Updated the NoticeWorker to utilize the GLM SDK for translating notice titles and descriptions, enhancing functionality and user experience. --- cmd/worker/bootstrap/bootstrap.go | 52 +++++++++++- cmd/worker/bootstrap/config.go | 31 +++++++ cmd/worker/main.go | 13 +-- cmd/worker/workers/notice.go | 129 ++++++++++++++++++++++++++---- internal/notice/entity.go | 1 - internal/notice/repository.go | 43 +++++----- pkg/glm/sdk.go | 72 ++++++++++++++++- ted/mapper.go | 9 --- ted/scraper.go | 104 +++++++++++++++++++++--- ted/service.go | 59 -------------- 10 files changed, 379 insertions(+), 134 deletions(-) diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 9452de6..f1e1121 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -7,6 +7,7 @@ import ( "tm/internal/notice" "tm/internal/tender" "tm/pkg/config" + "tm/pkg/glm" "tm/pkg/logger" "tm/pkg/mongo" "tm/pkg/notification" @@ -74,19 +75,19 @@ func InitMongoDB(conf config.MongoConfig, log logger.Logger) *mongo.ConnectionMa } // Init Worker -func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK) { +func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, glmService *glm.SDK) { // Initialize repositories noticeRepo := notice.NewRepository(mongoManager, appLogger) tenderRepo := tender.NewRepository(mongoManager, appLogger) - worker := workers.NewNoticeWorker(mongoManager, appLogger, notify, noticeRepo, tenderRepo) + worker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService) worker.Run() // start Worker job schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{ Name: "Worker Job", Func: func() { - worker := workers.NewNoticeWorker(mongoManager, appLogger, notify, noticeRepo, tenderRepo) + worker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService) worker.Run() }, Expr: config.Worker.Interval, @@ -148,3 +149,48 @@ func InitOllamaService(conf config.OllamaConfig, log logger.Logger) *ollama.SDK return ollamaSDK } + +// Init GLM Service +func InitGLMService(conf GLMConfig, log logger.Logger) *glm.SDK { + glmSDK, err := glm.New( + &glm.Config{ + BaseURL: conf.BaseURL, + APIKey: conf.APIKey, + Timeout: conf.Timeout, + RetryAttempts: conf.RetryAttempts, + RetryDelay: conf.RetryDelay, + EnableLogging: conf.EnableLogging, + UserAgent: conf.UserAgent, + DefaultModel: conf.DefaultModel, + DefaultMaxTokens: conf.DefaultMaxTokens, + DefaultTopP: conf.DefaultTopP, + DefaultTemperature: conf.DefaultTemperature, + DefaultFrequencyPenalty: conf.DefaultFrequencyPenalty, + DefaultPresencePenalty: conf.DefaultPresencePenalty, + EnableStreaming: conf.EnableStreaming, + EnableThinking: conf.EnableThinking, + TranslationOptions: glm.TranslationOption{ + Model: conf.TranslateOptions.Model, + MaxTokens: conf.TranslateOptions.MaxTokens, + Temperature: conf.TranslateOptions.Temperature, + FrequencyPenalty: conf.TranslateOptions.FrequencyPenalty, + PresencePenalty: conf.TranslateOptions.PresencePenalty, + TopP: conf.TranslateOptions.TopP, + EnableThinking: conf.TranslateOptions.EnableThinking, + }, + }, log) + + if err != nil { + log.Error("Failed to initialize GLM SDK", map[string]interface{}{ + "error": err.Error(), + }) + panic(fmt.Sprintf("Failed to initialize GLM SDK: %v", err)) + } + + log.Info("GLM SDK initialized successfully", map[string]interface{}{ + "base_url": conf.BaseURL, + "timeout": conf.Timeout, + }) + + return glmSDK +} diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 8c3485c..4fce431 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -1,6 +1,7 @@ package bootstrap import ( + "time" "tm/pkg/config" ) @@ -11,9 +12,39 @@ type Config struct { Notification config.NotificationConfig Ollama config.OllamaConfig Worker WorkerConfig + GLM GLMConfig AlertMail string `env:"ALERT_MAIL" envDefault:"alerts@opplens.com"` } type WorkerConfig struct { Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"` } + +type GLMConfig struct { + BaseURL string `env:"GLM_BASE_URL" envDefault:"https://api.z.ai"` + APIKey string `env:"GLM_API_KEY" envDefault:""` + Timeout time.Duration `env:"GLM_TIMEOUT" envDefault:"300s"` + RetryAttempts int `env:"GLM_RETRY_ATTEMPTS" envDefault:"3"` + RetryDelay time.Duration `env:"GLM_RETRY_DELAY" envDefault:"2s"` + EnableLogging bool `env:"GLM_ENABLE_LOGGING" envDefault:"true"` + UserAgent string `env:"GLM_USER_AGENT" envDefault:"Opplens-GLMSDK/1.0"` + DefaultModel string `env:"GLM_DEFAULT_MODEL" envDefault:"glm-4.5"` + DefaultMaxTokens int `env:"GLM_DEFAULT_MAX_TOKENS" envDefault:"4096"` + DefaultTopP float64 `env:"GLM_DEFAULT_TOP_P" envDefault:"1.0"` + DefaultTemperature float64 `env:"GLM_DEFAULT_TEMPERATURE" envDefault:"0.7"` + DefaultFrequencyPenalty float64 `env:"GLM_DEFAULT_FREQUENCY_PENALTY" envDefault:"0"` + DefaultPresencePenalty float64 `env:"GLM_DEFAULT_PRESENCE_PENALTY" envDefault:"0"` + EnableStreaming bool `env:"GLM_ENABLE_STREAMING" envDefault:"false"` + EnableThinking bool `env:"GLM_ENABLE_THINKING" envDefault:"false"` + TranslateOptions TranslateOptions +} + +type TranslateOptions struct { + Model string `env:"GLM_TRANSLATION_MODEL" envDefault:"glm-4.5"` + MaxTokens int `env:"GLM_TRANSLATION_MAX_TOKENS" envDefault:"2048"` + Temperature float64 `env:"GLM_TRANSLATION_TEMPERATURE" envDefault:"0.3"` + FrequencyPenalty float64 `env:"GLM_TRANSLATION_FREQUENCY_PENALTY" envDefault:"0.1"` + PresencePenalty float64 `env:"GLM_TRANSLATION_PRESENCE_PENALTY" envDefault:"0.1"` + TopP float64 `env:"GLM_TRANSLATION_TOP_P" envDefault:"1.0"` + EnableThinking bool `env:"GLM_TRANSLATION_ENABLE_THINKING" envDefault:"false"` +} diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 18cb77b..d4995ae 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -37,18 +37,11 @@ func main() { // Initialize notification service notificationService := bootstrap.InitNotificationService(config.Notification, appLogger) - // Initialize ollama service - // ollamaSDK := bootstrap.InitOllamaService(config.Ollama, appLogger) - // aiModels, err := ollamaSDK.ListModels(context.Background()) - // if err != nil { - // appLogger.Error("Failed to list ollama models", map[string]interface{}{ - // "error": err.Error(), - // }) - // } - // appLogger.Info("Ollama models", map[string]interface{}{"models": aiModels}) + // Initialize GLM service + glmService := bootstrap.InitGLMService(config.GLM, appLogger) // Initialize Worker - bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService) + bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, glmService) // Set up signal handling for graceful shutdown signalChan := make(chan os.Signal, 1) diff --git a/cmd/worker/workers/notice.go b/cmd/worker/workers/notice.go index 4274386..479434c 100644 --- a/cmd/worker/workers/notice.go +++ b/cmd/worker/workers/notice.go @@ -2,30 +2,33 @@ package workers import ( "context" + "strings" + "time" "tm/internal/notice" "tm/internal/tender" + "tm/pkg/glm" "tm/pkg/logger" "tm/pkg/mongo" "tm/pkg/notification" - "tm/pkg/ollama" ) type NoticeWorker struct { Mongo *mongo.ConnectionManager Logger logger.Logger - Notify notification.SDK - Ollama *ollama.SDK + Notify *notification.SDK NoticeRepo notice.Repository TenderRepo tender.TenderRepository + GLM *glm.SDK } -func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository) *NoticeWorker { +func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, glmService *glm.SDK) *NoticeWorker { return &NoticeWorker{ Mongo: mongo, Logger: logger, Notify: notify, NoticeRepo: noticeRepo, TenderRepo: tenderRepo, + GLM: glmService, } } @@ -52,12 +55,7 @@ func (w *NoticeWorker) Run() { "notice": n.ID.Hex(), }) - t, err := w.TenderRepo.GetByNoticePublicationID(context.Background(), n.NoticePublicationID) - if err == nil { - continue - } - - err = w.ToTender(&n, t) + t, err := w.ToTender(&n) if err != nil { w.Logger.Error("Failed to create tender", map[string]interface{}{ "error": err.Error(), @@ -80,13 +78,33 @@ func (w *NoticeWorker) Run() { }) } } + n.ProcessingMetadata.Processed = true + n.ProcessingMetadata.ProcessedAt = time.Now().Unix() + err = w.NoticeRepo.Update(context.Background(), &n) + if err != nil { + w.Logger.Error("Failed to update notice", map[string]interface{}{ + "error": err.Error(), + }) + } + w.Logger.Info("Notice processed", map[string]interface{}{ + "notice": n.ID.Hex(), + }) } skip += limit } } -func (w *NoticeWorker) ToTender(n *notice.Notice, t *tender.Tender) error { +func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) { + t, err := w.TenderRepo.GetByNoticePublicationID(context.Background(), n.NoticePublicationID) + if err != nil { + w.Logger.Info("New tender raised", map[string]interface{}{ + "notice_publication_id": n.NoticePublicationID, + "country_code": n.CountryCode, + }) + t = new(tender.Tender) + } + // Buyer Organization buyerOrg := &tender.Organization{} if n.BuyerOrganization != nil { @@ -223,9 +241,32 @@ func (w *NoticeWorker) ToTender(n *notice.Notice, t *tender.Tender) error { } } + title, err := w.GLM.Translate(context.Background(), n.Title, n.NoticeLanguageCode, "en") + if err != nil { + w.Logger.Error("Failed to translate title", map[string]interface{}{ + "notice_publication_id": n.NoticePublicationID, + "country_code": n.CountryCode, + "content": n.Title, + "error": err.Error(), + }) + title = n.Title + // return err + } + + description, err := w.GLM.Translate(context.Background(), n.Description, n.NoticeLanguageCode, "en") + if err != nil { + w.Logger.Error("Failed to translate description", map[string]interface{}{ + "notice_publication_id": n.NoticePublicationID, + "country_code": n.CountryCode, + "content": n.Description, + "error": err.Error(), + }) + description = n.Description + // return err + } + t.Title = title + t.Description = description t.NoticePublicationID = n.NoticePublicationID - t.Title = n.Title - t.Description = n.Description t.ProcurementTypeCode = n.ProcurementTypeCode t.ProcedureCode = n.ProcedureCode t.EstimatedValue = n.EstimatedValue @@ -268,5 +309,65 @@ func (w *NoticeWorker) ToTender(n *notice.Notice, t *tender.Tender) error { t.WinningTenderer = winningTenderer t.Modifications = modifications t.AwardedEntities = awardedEntities - return nil + t.TenderID = GenerateTenderID(n) + + return t, nil +} + +// generateTenderID generates a unique tender ID using the format: +// {Source}{BUYER_CODE}{TED_CODE}{DATE}{LOCATION_CODE} +func GenerateTenderID(t *notice.Notice) string { + var parts []string + + // Source (TED) + parts = append(parts, "1") + + // TED code (Notice Publication ID, first 8 chars) + tedCode := "00000000" + if t.NoticePublicationID != "" { + tedCode = PadOrTruncate(t.NoticePublicationID, 8) + } + parts = append(parts, tedCode) + + // Buyer code (first 8 chars of buyer organization ID, or 8 zeros if not available) + // buyerCode := "00000000" + // if t.BuyerOrganization != nil && t.BuyerOrganization.ID != "" { + // buyerCode = padOrTruncate(t.BuyerOrganization.ID, 8) + // } + // parts = append(parts, buyerCode) + + // Date (YYYYMMDD format from issue date) + parts = append(parts, PadOrTruncate(UnixToDateString(t.IssueDate), 4)) + + // Location code (country code + region code, padded to 6 chars) + locationCode := "EU" + if t.CountryCode != "" { + locationCode = t.CountryCode + } + parts = append(parts, locationCode) + + return strings.Join(parts, "") +} + +// padOrTruncate pads a string with zeros or truncates it to the specified length +func PadOrTruncate(str string, length int) string { + // Remove any non-alphanumeric characters and convert to uppercase + clean := strings.ToUpper(strings.ReplaceAll(str, "-", "")) + clean = strings.ReplaceAll(clean, " ", "") + + if len(clean) >= length { + return clean[:length] + } + + // Pad with zeros + return clean + strings.Repeat("0", length-len(clean)) +} + +// unixToDateString converts Unix to date string for TenderID generation +func UnixToDateString(unix int64) string { + if unix == 0 { + return "00000000" + } + + return time.Unix(unix, 0).Format("20060102") } diff --git a/internal/notice/entity.go b/internal/notice/entity.go index 5f8b0bf..104aa5d 100644 --- a/internal/notice/entity.go +++ b/internal/notice/entity.go @@ -52,7 +52,6 @@ type Notice struct { SourceFileURL string `bson:"source_file_url" json:"source_file_url"` SourceFileName string `bson:"source_file_name" json:"source_file_name"` ProcessingMetadata ProcessingMetadata `bson:"processing_metadata" json:"processing_metadata"` - TenderID string `bson:"tender_id" json:"tender_id"` // Implement logic to generate a unique tender_id using the following format: {Source}{BUYER_CODE}{TED_CODE}{DATE}{LOCATION_CODE} // Status-specific fields CancellationReason string `bson:"cancellation_reason,omitempty" json:"cancellation_reason,omitempty"` diff --git a/internal/notice/repository.go b/internal/notice/repository.go index bcf9551..3b66ca0 100644 --- a/internal/notice/repository.go +++ b/internal/notice/repository.go @@ -12,6 +12,7 @@ import ( // NoticeRepository interface defines notice data access methods type Repository interface { Import(ctx context.Context, notice *Notice) error + Update(ctx context.Context, notice *Notice) error BulkImport(ctx context.Context, notices []Notice) error GetByID(ctx context.Context, id string) (*Notice, error) GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Notice, error) @@ -87,6 +88,22 @@ func (r *noticeRepository) BulkImport(ctx context.Context, notices []Notice) err return nil } +// Update updates an existing notice +func (r *noticeRepository) Update(ctx context.Context, notice *Notice) error { + notice.UpdatedAt = time.Now().Unix() + + err := r.ormRepo.Update(ctx, notice) + if err != nil { + r.logger.Error("Failed to update notice", map[string]interface{}{ + "notice_id": notice.ID, + "error": err.Error(), + }) + return err + } + + return nil +} + // GetByID retrieves a notice by ID func (r *noticeRepository) GetByID(ctx context.Context, id string) (*Notice, error) { notice, err := r.ormRepo.FindByID(ctx, id) @@ -128,8 +145,10 @@ func (r *noticeRepository) GetUnProcessedNotices(ctx context.Context, limit int, filter := bson.M{"processing_metadata.processed": false} pagination := orm.Pagination{ - Limit: limit, - Skip: skip, + Limit: limit, + Skip: skip, + SortField: "created_at", + SortOrder: 1, } result, err := r.ormRepo.FindAll(ctx, filter, pagination) @@ -143,26 +162,6 @@ func (r *noticeRepository) GetUnProcessedNotices(ctx context.Context, limit int, return result.Items, result.TotalCount, nil } -// Update updates an existing notice -func (r *noticeRepository) Update(ctx context.Context, notice *Notice) error { - notice.UpdatedAt = time.Now().Unix() - - err := r.ormRepo.Update(ctx, notice) - if err != nil { - r.logger.Error("Failed to update notice", map[string]interface{}{ - "notice_id": notice.ID, - "error": err.Error(), - }) - return err - } - - r.logger.Info("Notice updated successfully", map[string]interface{}{ - "notice_id": notice.ID, - }) - - return nil -} - // Delete deletes a notice by ID func (r *noticeRepository) Delete(ctx context.Context, id string) error { err := r.ormRepo.Delete(ctx, id) diff --git a/pkg/glm/sdk.go b/pkg/glm/sdk.go index ec47a77..6f98aa4 100644 --- a/pkg/glm/sdk.go +++ b/pkg/glm/sdk.go @@ -15,11 +15,75 @@ type SDK struct { // New creates a new GLM SDK with the provided configuration and logger func New(config *Config, logger logger.Logger) (*SDK, error) { - if config == nil { - config = DefaultConfig() + cfg := DefaultConfig() + if config.BaseURL != "" { + cfg.BaseURL = config.BaseURL + } + if config.APIKey != "" { + cfg.APIKey = config.APIKey + } + if config.Timeout != 0 { + cfg.Timeout = config.Timeout + } + if config.RetryAttempts != 0 { + cfg.RetryAttempts = config.RetryAttempts + } + if config.RetryDelay != 0 { + cfg.RetryDelay = config.RetryDelay + } + if config.EnableLogging { + cfg.EnableLogging = config.EnableLogging + } + if config.UserAgent != "" { + cfg.UserAgent = config.UserAgent + } + if config.DefaultModel != "" { + cfg.DefaultModel = config.DefaultModel + } + if config.DefaultMaxTokens != 0 { + cfg.DefaultMaxTokens = config.DefaultMaxTokens + } + if config.DefaultTopP != 0 { + cfg.DefaultTopP = config.DefaultTopP + } + if config.DefaultTemperature != 0 { + cfg.DefaultTemperature = config.DefaultTemperature + } + if config.DefaultFrequencyPenalty != 0 { + cfg.DefaultFrequencyPenalty = config.DefaultFrequencyPenalty + } + if config.DefaultPresencePenalty != 0 { + cfg.DefaultPresencePenalty = config.DefaultPresencePenalty + } + if config.EnableStreaming { + cfg.EnableStreaming = config.EnableStreaming + } + if config.EnableThinking { + cfg.EnableThinking = config.EnableThinking + } + if config.TranslationOptions.Model != "" { + cfg.TranslationOptions.Model = config.TranslationOptions.Model + } + if config.TranslationOptions.MaxTokens != 0 { + cfg.TranslationOptions.MaxTokens = config.TranslationOptions.MaxTokens + } + if config.TranslationOptions.Temperature != 0 { + cfg.TranslationOptions.Temperature = config.TranslationOptions.Temperature + } + if config.TranslationOptions.FrequencyPenalty != 0 { + cfg.TranslationOptions.FrequencyPenalty = config.TranslationOptions.FrequencyPenalty + } + if config.TranslationOptions.PresencePenalty != 0 { + cfg.TranslationOptions.PresencePenalty = config.TranslationOptions.PresencePenalty + } + if config.TranslationOptions.TopP != 0 { + cfg.TranslationOptions.TopP = config.TranslationOptions.TopP + } + if config.TranslationOptions.EnableThinking { + cfg.TranslationOptions.EnableThinking = config.TranslationOptions.EnableThinking } - client, err := NewClient(config, logger) + client, err := NewClient(cfg, logger) if err != nil { return nil, err } @@ -29,7 +93,7 @@ func New(config *Config, logger logger.Logger) (*SDK, error) { return &SDK{ service: service, client: client, - config: config, + config: cfg, }, nil } diff --git a/ted/mapper.go b/ted/mapper.go index 2d254c4..3aaa4d2 100644 --- a/ted/mapper.go +++ b/ted/mapper.go @@ -79,9 +79,6 @@ func (s *TEDScraper) mapGenericNoticeToTender(doc TEDDocument, sourceFileURL, so }, } - // Generate the tender ID after creating the tender - t.TenderID = GenerateTenderID(t) - return t } @@ -228,9 +225,6 @@ func (s *TEDScraper) mapToTender(cn *ContractNotice, sourceFileURL, sourceFileNa t.OfficialLanguages = languages } - // Generate unique tender ID - t.TenderID = GenerateTenderID(t) - return t } @@ -308,9 +302,6 @@ func (s *TEDScraper) mapContractAwardNoticeToTender(can *ContractAwardNotice, so } } - // Generate unique tender ID - t.TenderID = GenerateTenderID(t) - return t } diff --git a/ted/scraper.go b/ted/scraper.go index 3f28d15..425d365 100644 --- a/ted/scraper.go +++ b/ted/scraper.go @@ -345,7 +345,6 @@ func (s *TEDScraper) processXMLContent(ctx context.Context, content []byte, file // // Create new tender s.logger.Info("Creating new tender", map[string]interface{}{ "contract_notice_id": t.ContractNoticeID, - "tender_id": t.TenderID, }) t.CreatedAt = time.Now().Unix() @@ -355,14 +354,12 @@ func (s *TEDScraper) processXMLContent(ctx context.Context, content []byte, file if err != nil { s.logger.Error("Failed to create new tender", map[string]interface{}{ "contract_notice_id": t.ContractNoticeID, - "tender_id": t.TenderID, "error": err.Error(), }) return fmt.Errorf("failed to create tender: %w", err) } s.logger.Info("Successfully created new tender", map[string]interface{}{ - "tender_id": t.TenderID, "contract_notice_id": t.ContractNoticeID, }) // } @@ -372,7 +369,6 @@ func (s *TEDScraper) processXMLContent(ctx context.Context, content []byte, file "notice_id": id, "notice_type": noticeType, "file_name": fileName, - "tender_id": t.TenderID, "tender_url": t.TenderURL, }) @@ -380,34 +376,118 @@ func (s *TEDScraper) processXMLContent(ctx context.Context, content []byte, file return nil } -func (s *TEDScraper) Run(ctx context.Context) error { - now := time.Now().Local() - s.logger.Info("Running periodic scraping", map[string]interface{}{ - "time": now.Format(time.RFC3339), +func (s *TEDScraper) Run(ctx context.Context, from, to *time.Time) error { + // If no date range is provided, use current date + if from == nil && to == nil { + now := time.Now().Local() + s.logger.Info("Running periodic scraping for current date", map[string]interface{}{ + "date": now.Format("02/01/2006"), + }) + return s.processSingleDate(ctx, now) + } + + // Validate date range + if from == nil || to == nil { + return fmt.Errorf("both from and to dates must be provided for date range scraping") + } + + if from.After(*to) { + return fmt.Errorf("from date cannot be after to date") + } + + s.logger.Info("Running date range scraping", map[string]interface{}{ + "from_date": from.Format("02/01/2006"), + "to_date": to.Format("02/01/2006"), + }) + + // Process each date in the range + current := *from + totalDays := 0 + successfulDays := 0 + var errors []string + + for !current.After(*to) { + totalDays++ + + s.logger.Info("Processing date", map[string]interface{}{ + "date": current.Format("02/01/2006"), + "progress": fmt.Sprintf("%d/%d", totalDays, int(to.Sub(*from).Hours()/24)+1), + }) + + err := s.processSingleDate(ctx, current) + if err != nil { + errorMsg := fmt.Sprintf("failed to process date %s: %v", current.Format("02/01/2006"), err) + errors = append(errors, errorMsg) + s.logger.Error("Failed to process date", map[string]interface{}{ + "date": current.Format("02/01/2006"), + "error": err.Error(), + }) + } else { + successfulDays++ + } + + // Check for context cancellation + select { + case <-ctx.Done(): + s.logger.Warn("Date range scraping cancelled", map[string]interface{}{ + "processed_days": totalDays, + "successful_days": successfulDays, + "cancelled_at": current.Format("02/01/2006"), + }) + return ctx.Err() + default: + } + + // Move to next day + current = current.AddDate(0, 0, 1) + } + + s.logger.Info("Date range scraping completed", map[string]interface{}{ + "total_days": totalDays, + "successful_days": successfulDays, + "failed_days": len(errors), + "errors": errors, + }) + + // Return error if no days were processed successfully + if successfulDays == 0 && len(errors) > 0 { + return fmt.Errorf("failed to process any dates in range: %v", errors) + } + + return nil +} + +// processSingleDate processes data for a single date +func (s *TEDScraper) processSingleDate(ctx context.Context, date time.Time) error { + s.logger.Info("Processing single date", map[string]interface{}{ + "date": date.Format("02/01/2006"), }) // get ojs - ojs, err := GetOJS(s.config.BaseURL, now.Year(), now.Format("02/01/2006")) + ojs, err := GetOJS(s.config.BaseURL, date.Year(), date.Format("02/01/2006")) if err != nil { s.logger.Error("Failed to get OJS", map[string]interface{}{ + "date": date.Format("02/01/2006"), "error": err.Error(), }) - return nil + return fmt.Errorf("failed to get OJS for date %s: %w", date.Format("02/01/2006"), err) } // download daily package result, err := s.DownloadDailyPackage(ctx, ojs) if err != nil { s.logger.Error("Failed to download daily package", map[string]interface{}{ + "date": date.Format("02/01/2006"), + "ojs": ojs, "error": err.Error(), }) - return err + return fmt.Errorf("failed to download daily package for date %s: %w", date.Format("02/01/2006"), err) } // process daily package s.logger.Info("Processed daily package successfully", map[string]interface{}{ "result": result, - "time": now.Format(time.RFC3339), + "date": date.Format("02/01/2006"), "ojs": ojs, }) diff --git a/ted/service.go b/ted/service.go index ef691d0..8f75eed 100644 --- a/ted/service.go +++ b/ted/service.go @@ -5,7 +5,6 @@ import ( "strconv" "strings" "time" - "tm/internal/notice" ) // parseDate parses various date formats used in TED XML @@ -69,15 +68,6 @@ func ParseTimeToUnix(timeStr string) int64 { return 0 } -// unixToDateString converts Unix to date string for TenderID generation -func UnixToDateString(unix int64) string { - if unix == 0 { - return "00000000" - } - - return time.Unix(unix, 0).Format("20060102") -} - func CalculateSubmissionDeadline(publicationDate time.Time) time.Time { deadline := publicationDate workingHoursAdded := 0 @@ -108,20 +98,6 @@ func CalculateApplicationDeadline(tenderDeadline time.Time) time.Time { return applicationDeadline } -// padOrTruncate pads a string with zeros or truncates it to the specified length -func PadOrTruncate(str string, length int) string { - // Remove any non-alphanumeric characters and convert to uppercase - clean := strings.ToUpper(strings.ReplaceAll(str, "-", "")) - clean = strings.ReplaceAll(clean, " ", "") - - if len(clean) >= length { - return clean[:length] - } - - // Pad with zeros - return clean + strings.Repeat("0", length-len(clean)) -} - // generateTenderURL generates the TED tender URL from NoticePublicationID func GenerateTenderURL(noticePublicationID string) string { if noticePublicationID == "" { @@ -143,38 +119,3 @@ func GenerateTenderURL(noticePublicationID string) string { return "" } - -// generateTenderID generates a unique tender ID using the format: -// {Source}{BUYER_CODE}{TED_CODE}{DATE}{LOCATION_CODE} -func GenerateTenderID(t *notice.Notice) string { - var parts []string - - // Source (TED) - parts = append(parts, "1") - - // TED code (Notice Publication ID, first 8 chars) - tedCode := "00000000" - if t.NoticePublicationID != "" { - tedCode = PadOrTruncate(t.NoticePublicationID, 8) - } - parts = append(parts, tedCode) - - // Buyer code (first 8 chars of buyer organization ID, or 8 zeros if not available) - // buyerCode := "00000000" - // if t.BuyerOrganization != nil && t.BuyerOrganization.ID != "" { - // buyerCode = padOrTruncate(t.BuyerOrganization.ID, 8) - // } - // parts = append(parts, buyerCode) - - // Date (YYYYMMDD format from issue date) - parts = append(parts, PadOrTruncate(UnixToDateString(t.IssueDate), 4)) - - // Location code (country code + region code, padded to 6 chars) - locationCode := "EU" - if t.CountryCode != "" { - locationCode = t.CountryCode - } - parts = append(parts, locationCode) - - return strings.Join(parts, "") -}