package workers import ( "context" "fmt" "strings" "time" "tm/internal/notice" "tm/internal/tender" "tm/pkg/glm" "tm/pkg/logger" "tm/pkg/mongo" "tm/pkg/notification" "tm/pkg/scraper" "go.mongodb.org/mongo-driver/v2/bson" mongodriver "go.mongodb.org/mongo-driver/v2/mongo" ) type NoticeWorker struct { Mongo *mongo.ConnectionManager Logger logger.Logger Notify *notification.SDK NoticeRepo notice.Repository TenderRepo tender.TenderRepository GLM *glm.SDK Scraper scraper.SDK ProcessingLimit int } func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, glmService *glm.SDK, scraper scraper.SDK, processingLimit int) *NoticeWorker { if processingLimit < 0 { processingLimit = 5 // Default limit } return &NoticeWorker{ Mongo: mongo, Logger: logger, Notify: notify, NoticeRepo: noticeRepo, TenderRepo: tenderRepo, GLM: glmService, Scraper: scraper, ProcessingLimit: processingLimit, } } // cleanupProcessedNotices removes notices that have been successfully processed func (w *NoticeWorker) cleanupProcessedNotices() { w.Logger.Info("Starting cleanup of processed notices", map[string]interface{}{}) // Get all processed notices (limit to avoid memory issues) limit := 100 skip := 0 for { notices, _, err := w.NoticeRepo.GetProcessedNotices(context.Background(), limit, skip) if err != nil { w.Logger.Error("Failed to get processed notices for cleanup", map[string]interface{}{ "error": err.Error(), }) break } if len(notices) == 0 { break } deletedCount := 0 for _, notice := range notices { if err := w.NoticeRepo.Delete(context.Background(), notice.ID.Hex()); err != nil { w.Logger.Warn("Failed to delete processed notice during cleanup", map[string]interface{}{ "notice_id": notice.ID.Hex(), "error": err.Error(), }) } else { deletedCount++ } } w.Logger.Info("Cleanup batch completed", map[string]interface{}{ "processed": len(notices), "deleted": deletedCount, "skip": skip, }) skip += limit // Safety limit to avoid infinite loops if skip > 10000 { w.Logger.Warn("Cleanup reached safety limit, stopping", map[string]interface{}{ "skip": skip, }) break } } w.Logger.Info("Notice cleanup completed", map[string]interface{}{}) } func (w *NoticeWorker) Run() { w.Logger.Info("Notice worker started", map[string]interface{}{}) w.cleanupProcessedNotices() skip := 0 processedCount := 0 maxToProcess := w.ProcessingLimit for maxToProcess == 0 || processedCount < maxToProcess { notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), skip, maxToProcess) if err != nil { w.Logger.Error("Failed to get notices", map[string]interface{}{ "error": err.Error(), }) continue } if len(notices) == 0 { w.Logger.Info("No more unprocessed notices found", map[string]interface{}{ "processed_count": processedCount, }) break } for _, n := range notices { if maxToProcess > 0 && processedCount >= maxToProcess { w.Logger.Info("Reached maximum processing limit, stopping", map[string]interface{}{ "processed_count": processedCount, "max_to_process": maxToProcess, }) break } w.Logger.Info("Notice", map[string]interface{}{ "notice": n.ID.Hex(), }) t, err := w.ToTender(&n) if err != nil { w.Logger.Error("Failed to create tender", map[string]interface{}{ "error": err.Error(), }) continue } if t.ID.IsZero() { err = w.TenderRepo.Create(context.Background(), t) if err != nil { w.Logger.Error("Failed to create tender", map[string]interface{}{ "error": err.Error(), }) } } else { err = w.TenderRepo.Update(context.Background(), t) if err != nil { w.Logger.Error("Failed to update tender", map[string]interface{}{ "error": err.Error(), }) } } 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(), }) } else { processedCount++ w.Logger.Info("Notice processed successfully", map[string]interface{}{ "notice": n.ID.Hex(), "processed_count": processedCount, }) // Clean up: Delete processed notice from MongoDB if err := w.NoticeRepo.Delete(context.Background(), n.ID.Hex()); err != nil { w.Logger.Warn("Failed to delete processed notice", map[string]interface{}{ "notice_id": n.ID.Hex(), "error": err.Error(), }) } else { w.Logger.Debug("Successfully deleted processed notice", map[string]interface{}{ "notice_id": n.ID.Hex(), }) } } } skip += maxToProcess // Check if we need to exit outer loop (break from inner loop means we hit the limit) if maxToProcess > 0 && processedCount >= maxToProcess { break } } } // scrapeDocumentsForNotice calls the scraper SDK to scrape documents for a processed notice func (w *NoticeWorker) scrapeDocumentsForNotice(notice *notice.Notice) { // Skip if no NoticePublicationID if notice.NoticePublicationID == "" { w.Logger.Warn("Skipping document scraping - no NoticePublicationID", map[string]interface{}{ "notice_id": notice.ID.Hex(), }) return } // Get the tender URL noticeURL := notice.GetTenderURL() if noticeURL == "" { w.Logger.Warn("Skipping document scraping - no tender URL available", map[string]interface{}{ "notice_id": notice.ID.Hex(), "notice_publication_id": notice.NoticePublicationID, }) return } // Call scraper SDK scrapeReq := &scraper.ScrapeRequest{ NoticeID: notice.NoticePublicationID, // Use NoticePublicationID as notice_id NoticeURL: noticeURL, } w.Logger.Info("Starting document scraping for processed notice", map[string]interface{}{ "notice_id": notice.ID.Hex(), "notice_publication_id": notice.NoticePublicationID, "notice_url": noticeURL, }) response, err := w.Scraper.ScrapeDocuments(context.Background(), scrapeReq) if err != nil { w.Logger.Error("Failed to scrape documents", map[string]interface{}{ "notice_id": notice.ID.Hex(), "notice_publication_id": notice.NoticePublicationID, "error": err.Error(), }) return } if !response.Success { w.Logger.Warn("Document scraping failed", map[string]interface{}{ "notice_id": notice.ID.Hex(), "notice_publication_id": notice.NoticePublicationID, "errors": response.Errors, }) return } w.Logger.Info("Document scraping completed successfully", map[string]interface{}{ "notice_id": notice.ID.Hex(), "notice_publication_id": notice.NoticePublicationID, "uploaded_count": response.UploadedCount, "bucket": response.Buckets, }) } 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 { buyerOrg = &tender.Organization{ Name: n.BuyerOrganization.Name, CompanyID: n.BuyerOrganization.CompanyID, WebsiteURI: n.BuyerOrganization.WebsiteURI, ContactName: n.BuyerOrganization.ContactName, ContactTelephone: n.BuyerOrganization.ContactTelephone, ContactEmail: n.BuyerOrganization.ContactEmail, ContactFax: n.BuyerOrganization.ContactFax, Role: n.BuyerOrganization.Role, Address: tender.Address{ StreetName: n.BuyerOrganization.Address.StreetName, CityName: n.BuyerOrganization.Address.CityName, PostalZone: n.BuyerOrganization.Address.PostalZone, CountrySubentityCode: n.BuyerOrganization.Address.CountrySubentityCode, Department: n.BuyerOrganization.Address.Department, Region: n.BuyerOrganization.Address.Region, CountryCode: n.BuyerOrganization.Address.CountryCode, }, } } // Review Organization reviewOrg := &tender.Organization{} if n.ReviewOrganization != nil { reviewOrg = &tender.Organization{ Name: n.ReviewOrganization.Name, CompanyID: n.ReviewOrganization.CompanyID, WebsiteURI: n.ReviewOrganization.WebsiteURI, ContactName: n.ReviewOrganization.ContactName, ContactTelephone: n.ReviewOrganization.ContactTelephone, ContactEmail: n.ReviewOrganization.ContactEmail, ContactFax: n.ReviewOrganization.ContactFax, Role: n.ReviewOrganization.Role, Address: tender.Address{ StreetName: n.ReviewOrganization.Address.StreetName, CityName: n.ReviewOrganization.Address.CityName, PostalZone: n.ReviewOrganization.Address.PostalZone, CountrySubentityCode: n.ReviewOrganization.Address.CountrySubentityCode, Department: n.ReviewOrganization.Address.Department, Region: n.ReviewOrganization.Address.Region, CountryCode: n.ReviewOrganization.Address.CountryCode, }, } } // Organizations organizations := make([]tender.Organization, len(n.Organizations)) for i, org := range n.Organizations { organizations[i] = tender.Organization{ Name: org.Name, CompanyID: org.CompanyID, WebsiteURI: org.WebsiteURI, ContactName: org.ContactName, ContactTelephone: org.ContactTelephone, ContactEmail: org.ContactEmail, ContactFax: org.ContactFax, Role: org.Role, Address: tender.Address{ StreetName: org.Address.StreetName, CityName: org.Address.CityName, PostalZone: org.Address.PostalZone, CountrySubentityCode: org.Address.CountrySubentityCode, Department: org.Address.Department, Region: org.Address.Region, CountryCode: org.Address.CountryCode, }, } } // Selection Criteria selectionCriteria := make([]tender.SelectionCriterion, len(n.SelectionCriteria)) for i, criterion := range n.SelectionCriteria { selectionCriteria[i] = tender.SelectionCriterion{ TypeCode: criterion.TypeCode, Description: criterion.Description, LanguageID: criterion.LanguageID, } } // Winning Tenderer winningTenderer := &tender.Organization{} if n.WinningTenderer != nil { winningTenderer = &tender.Organization{ Name: n.WinningTenderer.Name, CompanyID: n.WinningTenderer.CompanyID, WebsiteURI: n.WinningTenderer.WebsiteURI, ContactName: n.WinningTenderer.ContactName, ContactTelephone: n.WinningTenderer.ContactTelephone, ContactEmail: n.WinningTenderer.ContactEmail, ContactFax: n.WinningTenderer.ContactFax, Role: n.WinningTenderer.Role, Address: tender.Address{ StreetName: n.WinningTenderer.Address.StreetName, CityName: n.WinningTenderer.Address.CityName, PostalZone: n.WinningTenderer.Address.PostalZone, CountrySubentityCode: n.WinningTenderer.Address.CountrySubentityCode, Department: n.WinningTenderer.Address.Department, Region: n.WinningTenderer.Address.Region, CountryCode: n.WinningTenderer.Address.CountryCode, }, } } // Modifications modifications := make([]tender.TenderModification, len(n.Modifications)) for i, modification := range n.Modifications { modifications[i] = tender.TenderModification{ ModificationDate: modification.ModificationDate, ModificationReason: modification.ModificationReason, Description: modification.Description, LanguageID: modification.LanguageID, } } // Awarded Entities awardedEntities := make([]tender.Awarded, len(n.AwardedEntities)) for i, awarded := range n.AwardedEntities { awardedEntities[i] = tender.Awarded{ Name: awarded.Name, Address: awarded.Address, Country: awarded.Country, Amount: awarded.Amount, Currency: awarded.Currency, Share: awarded.Share, AwardDate: awarded.AwardDate, ContractID: awarded.ContractID, TenderID: awarded.TenderID, LotID: awarded.LotID, CompanyID: awarded.CompanyID, OrganizationID: awarded.OrganizationID, } } var title string if w.GLM != nil { 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 } } else { w.Logger.Warn("GLM service not available, using original title", map[string]interface{}{ "notice_publication_id": n.NoticePublicationID, "country_code": n.CountryCode, }) title = n.Title } var description string if w.GLM != nil { 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 } } else { w.Logger.Warn("GLM service not available, using original description", map[string]interface{}{ "notice_publication_id": n.NoticePublicationID, "country_code": n.CountryCode, }) description = n.Description } t.Title = title t.Description = description t.NoticePublicationID = n.NoticePublicationID t.ProcurementTypeCode = n.ProcurementTypeCode t.ProcedureCode = n.ProcedureCode t.EstimatedValue = n.EstimatedValue t.Currency = n.Currency t.TenderDeadline = n.TenderDeadline t.SubmissionURL = n.SubmissionURL t.BuyerOrganization = buyerOrg t.ReviewOrganization = reviewOrg t.Organizations = organizations t.SelectionCriteria = selectionCriteria t.WinningTenderer = winningTenderer t.Modifications = modifications t.AwardedEntities = awardedEntities t.NoticeTypeCode = n.NoticeTypeCode t.NoticeSubTypeCode = n.NoticeSubTypeCode t.NoticeLanguageCode = n.NoticeLanguageCode t.IssueDate = n.IssueDate t.IssueTime = n.IssueTime t.PublicationDate = n.PublicationDate t.SubmissionDeadline = n.SubmissionDeadline t.ApplicationDeadline = n.ApplicationDeadline t.GazetteID = n.GazetteID t.PlaceOfPerformance = n.PlaceOfPerformance t.CountryCode = n.CountryCode t.RegionCode = n.RegionCode t.CityName = n.CityName t.PostalCode = n.PostalCode t.DocumentURI = n.DocumentURI t.TenderURL = n.TenderURL t.MainClassification = n.MainClassification t.AdditionalClassifications = n.AdditionalClassifications t.OfficialLanguages = n.OfficialLanguages t.Status = tender.TenderStatus(n.Status) t.Source = tender.TenderSource(n.Source) t.ContractNoticeID = n.ContractNoticeID t.ContractFolderID = n.ContractFolderID t.SourceFileURL = n.SourceFileURL t.SourceFileName = n.SourceFileName // Preserve document-scraping/summarization flags on the tender (managed by // downstream workers), but refresh the notice-side processing metadata. t.ProcessingMetadata.ScrapedAt = n.ProcessingMetadata.ScrapedAt t.ProcessingMetadata.ProcessedAt = n.ProcessingMetadata.ProcessedAt t.ProcessingMetadata.ProcessingVersion = n.ProcessingMetadata.ProcessingVersion t.ProcessingMetadata.ParsingErrors = n.ProcessingMetadata.ParsingErrors t.ProcessingMetadata.ValidationErrors = n.ProcessingMetadata.ValidationErrors t.ProcessingMetadata.EnrichmentData = n.ProcessingMetadata.EnrichmentData t.ProcessingMetadata.TranslatedData = n.ProcessingMetadata.TranslatedData t.ProcessingMetadata.TranslatedAt = n.ProcessingMetadata.TranslatedAt t.ProcessingMetadata.Processed = n.ProcessingMetadata.Processed t.CancellationReason = n.CancellationReason t.CancellationDate = n.CancellationDate t.AwardDate = n.AwardDate t.AwardedValue = n.AwardedValue t.ContractNumber = n.ContractNumber t.SuspensionReason = n.SuspensionReason t.SuspensionDate = n.SuspensionDate t.WinningTenderer = winningTenderer t.Modifications = modifications t.AwardedEntities = awardedEntities t.TenderID = w.GenerateTenderID(context.Background(), n) t.ProjectName = w.GenerateProjectName(t) return t, nil } // GenerateTenderID generates a unique tender ID using the PBL naming convention: SCDYYNNN // SCD = Source Code (3 letters), YY = Year (2 digits), NNN = Sequential number (3 digits) func (w *NoticeWorker) GenerateTenderID(ctx context.Context, t *notice.Notice) string { sourceCode := getSourceCode(t) year := getCurrentYear() sequentialNumber := w.getNextSequentialNumber(ctx, sourceCode, year) return fmt.Sprintf("%s%s%03d", sourceCode, year, sequentialNumber) } // getSourceCode determines the 3-letter source code based on tender characteristics func getSourceCode(t *notice.Notice) string { // Since all notices come from TED (Tenders Electronic Daily), use PTD for Public Tender return "PTD" } // getCurrentYear returns the last two digits of the current year func getCurrentYear() string { now := time.Now() return fmt.Sprintf("%02d", now.Year()%100) } // GenerateProjectName generates the project name using PBL naming convention: // __ func (w *NoticeWorker) GenerateProjectName(t *tender.Tender) string { clientName := w.getClientName(t) opportunityName := w.getOpportunityName(t) // Clean and format the names (replace spaces with hyphens, remove special chars) clientName = cleanForProjectName(clientName) opportunityName = cleanForProjectName(opportunityName) // If client name and opportunity name are the same, just use one of them if clientName == opportunityName { return fmt.Sprintf("%s_%s_PBL", t.TenderID, clientName) } return fmt.Sprintf("%s_%s-%s_PBL", t.TenderID, clientName, opportunityName) } // getClientName extracts the client name from the tender func (w *NoticeWorker) getClientName(t *tender.Tender) string { if t.BuyerOrganization != nil && t.BuyerOrganization.Name != "" { return t.BuyerOrganization.Name } return "UnknownClient" } // getOpportunityName extracts the opportunity name from the tender title func (w *NoticeWorker) getOpportunityName(t *tender.Tender) string { if t.Title != "" { // Take first few words of the title as opportunity name words := strings.Fields(t.Title) if len(words) > 3 { return strings.Join(words[:3], " ") } return t.Title } return "UnknownOpportunity" } // cleanForProjectName cleans a string for use in project names (removes special chars, replaces spaces with hyphens) func cleanForProjectName(s string) string { // Replace spaces with hyphens s = strings.ReplaceAll(s, " ", "-") // Remove special characters except hyphens and alphanumeric clean := "" for _, r := range s { if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '-' { clean += string(r) } } // Remove multiple consecutive hyphens for strings.Contains(clean, "--") { clean = strings.ReplaceAll(clean, "--", "-") } // Trim hyphens from start and end clean = strings.Trim(clean, "-") return clean } // getNextSequentialNumber returns the next sequential number for the given source code and year func (w *NoticeWorker) getNextSequentialNumber(ctx context.Context, sourceCode, year string) int { // Query tenders with tender IDs that match the pattern: {sourceCode}{year}* // We need to find the highest NNN value for this source code and year // Pattern to match: e.g., "PTD25" followed by 3 digits pattern := fmt.Sprintf("^%s%s\\d{3}$", sourceCode, year) // Create aggregation pipeline to find the highest sequential number pipeline := mongodriver.Pipeline{ bson.D{{Key: "$match", Value: bson.M{ "tender_id": bson.M{"$regex": pattern}, }}}, bson.D{{Key: "$project", Value: bson.M{ "sequential_num": bson.M{"$substr": bson.A{"$tender_id", 5, 3}}, // Extract last 3 chars (NNN) }}}, bson.D{{Key: "$group", Value: bson.M{ "_id": nil, "max_num": bson.M{"$max": bson.M{"$toInt": "$sequential_num"}}, }}}, } // Execute aggregation cursor, err := w.Mongo.GetCollection("tenders").Aggregate(ctx, pipeline) if err != nil { w.Logger.Error("Failed to aggregate tenders for sequential numbering", map[string]interface{}{ "error": err.Error(), "source_code": sourceCode, "year": year, }) return 1 // Return 1 as fallback } defer cursor.Close(ctx) // Parse result var result struct { MaxNum int `bson:"max_num"` } if cursor.Next(ctx) { if err := cursor.Decode(&result); err == nil { return result.MaxNum + 1 } } return 1 // No existing tenders found, start with 1 } // 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") }