AI scraper prefix

cmd/worker/bootstrap/config.go

@@ -38,7 +38,7 @@ type WorkerConfig struct {
 	Interval                string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"`
 	NoticeProcessingLimit   int    `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"0"`
 	DeleteProcessedNotices  bool   `env:"WORKER_DELETE_PROCESSED_NOTICES" envDefault:"true"`
-	NoticeWorkerConcurrency int    `env:"WORKER_NOTICE_CONCURRENCY" envDefault:"8"`
+	NoticeWorkerConcurrency int    `env:"WORKER_NOTICE_CONCURRENCY" envDefault:"1"`
 	NoticeFetchBatchSize    int    `env:"WORKER_NOTICE_FETCH_BATCH" envDefault:"50"`
 	// NoticeBatchPause is slept after each notice batch finishes, before loading the next batch (0 = no pause).
 	NoticeBatchPause time.Duration `env:"WORKER_NOTICE_BATCH_PAUSE" envDefault:"0s"`

cmd/worker/workers/notice.go

@@ -21,6 +21,10 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+// noticeRunSerialMu ensures only one NoticeWorker.Run executes at a time (startup
+// goroutine + cron ticks can otherwise overlap and double load on MongoDB).
+var noticeRunSerialMu sync.Mutex
+
 const (
 	// tenderIDCountersCollection holds per-(sourceCode+year) atomic counters used to allocate
 	// unique tender_id values (e.g. PTD26827). Using a Mongo counter document is the standard way
@@ -39,7 +43,7 @@ type NoticeWorker struct {
 	TenderRepo             tender.TenderRepository
 	ProcessingLimit        int
 	DeleteProcessedNotices bool
-	// Concurrency is the maximum number of notices processed in parallel per batch (default 8).
+	// Concurrency is the maximum number of notices processed in parallel per batch (default 1).
 	Concurrency int
 	// FetchBatchSize is how many unprocessed notices to load from MongoDB per batch (default 50).
 	FetchBatchSize int
@@ -60,7 +64,7 @@ func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notif
 		processingLimit = 5 // Default limit
 	}
 	if concurrency < 1 {
-		concurrency = 8
+		concurrency = 1
 	}
 	if fetchBatchSize < 1 {
 		fetchBatchSize = 50
@@ -141,6 +145,9 @@ func (w *NoticeWorker) cleanupProcessedNotices() {
 }
 
 func (w *NoticeWorker) Run() {
+	noticeRunSerialMu.Lock()
+	defer noticeRunSerialMu.Unlock()
+
 	w.Logger.Info("Notice worker started", map[string]interface{}{
 		"concurrency":         w.Concurrency,
 		"fetch_batch":         w.FetchBatchSize,

pkg/ai_summarizer/storage.go

@@ -126,10 +126,14 @@ func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolde
 		"path":   objectPath,
 	})
 
-	// Attempt to get the object
+	return s.getSummaryFromObjectKey(ctx, objectPath, contractFolderID, noticePublicationID)
-	object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectPath, minio.GetObjectOptions{})
+}
+
+// getSummaryFromObjectKey reads tender.json at objectKey and returns overall_summary when ready.
+// contractFolderID and noticePublicationID are used for logging only (may be empty when unknown).
+func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey, contractFolderID, noticePublicationID string) (string, error) {
+	object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectKey, minio.GetObjectOptions{})
 	if err != nil {
-		// minio-go returns an error response that we can inspect
 		var minioErr minio.ErrorResponse
 		if errors.As(err, &minioErr) {
 			switch minioErr.StatusCode {
@@ -146,24 +150,20 @@ func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolde
 	}
 	defer object.Close()
 
-	// Read the object content
 	data, err := io.ReadAll(object)
 	if err != nil {
 		return "", fmt.Errorf("ai_summarizer: failed to read tender.json: %w", err)
 	}
 
-	// Parse the JSON
 	var tenderJSON TenderJSON
 	if err := json.Unmarshal(data, &tenderJSON); err != nil {
 		return "", fmt.Errorf("ai_summarizer: failed to parse tender.json: %w", err)
 	}
 
-	// Check if summarization is complete
 	if !tenderJSON.SummarizeStatus.Done {
 		return "", ErrSummaryNotReady
 	}
 
-	// Check if overall_summary is available
 	if tenderJSON.OverallSummary == nil || *tenderJSON.OverallSummary == "" {
 		return "", ErrNoOverallSummary
 	}
@@ -171,6 +171,7 @@ func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolde
 	s.logger.Info("Retrieved overall_summary from storage", map[string]interface{}{
 		"contract_folder_id": contractFolderID,
 		"notice_id":          noticePublicationID,
+		"object_key":         objectKey,
 		"summary_length":     len(*tenderJSON.OverallSummary),
 	})
 
@@ -302,65 +303,86 @@ func (s *StorageClient) DownloadDocument(ctx context.Context, objectName string)
 	return object, nil
 }
 
-// GetAllSummaries walks the MinIO bucket for the AI team's procedure-scoped
+// GetAllSummaries walks the MinIO bucket for tender.json objects. It first lists
-// layout (PROC_<contractFolderID>/<noticePublicationID>/tender.json) and
+// under the PROC_ prefix; if none match, it falls back to a full-bucket recursive
-// returns one TenderSummaryItem per notice. Tenders whose summarize_status.done
+// scan (slower on large buckets) so alternate key layouts still work.
-// is true are returned with their overall_summary; tenders still in progress
-// are returned with Done=false and an empty summary.
 func (s *StorageClient) GetAllSummaries(ctx context.Context) ([]TenderSummaryItem, error) {
-	s.logger.Debug("Listing all tender.json files from MinIO", map[string]interface{}{
+	results := make([]TenderSummaryItem, 0)
-		"bucket": s.config.MinioBucket,
+	seen := make(map[string]struct{})
-	})
 
-	objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{
+	appendFromPrefix := func(prefix string) {
-		Prefix:    procedurePrefix,
+		s.logger.Debug("Listing tender.json files from MinIO", map[string]interface{}{
-		Recursive: true,
+			"bucket": s.config.MinioBucket,
-	})
+			"prefix": prefix,
-
-	var results []TenderSummaryItem
-
-	for object := range objectCh {
-		if object.Err != nil {
-			s.logger.Error("Error listing MinIO objects", map[string]interface{}{
-				"error": object.Err.Error(),
-			})
-			continue
-		}
-
-		// Expected key shape: PROC_<contractFolderID>/<noticePublicationID>/tender.json
-		if !strings.HasSuffix(object.Key, "/tender.json") {
-			continue
-		}
-
-		contractFolderID, noticePublicationID, ok := parseTenderJSONKey(object.Key)
-		if !ok {
-			continue
-		}
-
-		summary, err := s.GetSummaryFromStorage(ctx, contractFolderID, noticePublicationID)
-		if err != nil {
-			s.logger.Debug("Skipping tender summary", map[string]interface{}{
-				"contract_folder_id": contractFolderID,
-				"notice_id":          noticePublicationID,
-				"error":              err.Error(),
-			})
-
-			if errors.Is(err, ErrSummaryNotReady) {
-				results = append(results, TenderSummaryItem{
-					ContractFolderID: contractFolderID,
-					NoticeID:         noticePublicationID,
-					Done:             false,
-				})
-			}
-			continue
-		}
-
-		results = append(results, TenderSummaryItem{
-			ContractFolderID: contractFolderID,
-			NoticeID:         noticePublicationID,
-			OverallSummary:   summary,
-			Done:             true,
 		})
+
+		objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{
+			Prefix:    prefix,
+			Recursive: true,
+		})
+
+		for object := range objectCh {
+			if object.Err != nil {
+				s.logger.Error("Error listing MinIO objects", map[string]interface{}{
+					"prefix": prefix,
+					"error":  object.Err.Error(),
+				})
+				continue
+			}
+			key := object.Key
+			if !strings.HasSuffix(key, "/tender.json") {
+				continue
+			}
+			if _, dup := seen[key]; dup {
+				continue
+			}
+
+			contractFolderID, noticePublicationID, ok := parseTenderJSONKey(key)
+			if !ok {
+				s.logger.Debug("Skipping unparseable tender.json key", map[string]interface{}{
+					"key": key,
+				})
+				continue
+			}
+			seen[key] = struct{}{}
+
+			summary, err := s.GetSummaryFromStorage(ctx, contractFolderID, noticePublicationID)
+			if errors.Is(err, ErrObjectNotFound) {
+				summary, err = s.getSummaryFromObjectKey(ctx, key, contractFolderID, noticePublicationID)
+			}
+			if err != nil {
+				s.logger.Debug("Skipping tender summary", map[string]interface{}{
+					"object_key":         key,
+					"contract_folder_id": contractFolderID,
+					"notice_id":          noticePublicationID,
+					"error":              err.Error(),
+				})
+
+				if errors.Is(err, ErrSummaryNotReady) {
+					results = append(results, TenderSummaryItem{
+						ContractFolderID: contractFolderID,
+						NoticeID:         noticePublicationID,
+						Done:             false,
+					})
+				}
+				continue
+			}
+
+			results = append(results, TenderSummaryItem{
+				ContractFolderID: contractFolderID,
+				NoticeID:         noticePublicationID,
+				OverallSummary:   summary,
+				Done:             true,
+			})
+		}
+	}
+
+	appendFromPrefix(procedurePrefix)
+	if len(seen) == 0 {
+		s.logger.Info("No tender.json keys under PROC_; listing entire bucket for */tender.json (may be slower on large buckets)", map[string]interface{}{
+			"bucket": s.config.MinioBucket,
+		})
+		appendFromPrefix("")
 	}
 
 	s.logger.Info("Retrieved all summaries from storage", map[string]interface{}{
@@ -371,26 +393,38 @@ func (s *StorageClient) GetAllSummaries(ctx context.Context) ([]TenderSummaryIte
 	return results, nil
 }
 
-// parseTenderJSONKey splits a key of the form
+// parseTenderJSONKey splits an object key ending in /tender.json into contract folder
-// PROC_<contractFolderID>/<noticePublicationID>/tender.json into its components.
+// id and notice publication id. Accepts PROC_/proc_ prefix or plain <folder>/<notice>/tender.json.
-// The boolean is false if the key does not match the expected layout.
 func parseTenderJSONKey(key string) (contractFolderID, noticePublicationID string, ok bool) {
-	if !strings.HasPrefix(key, procedurePrefix) {
+	key = strings.TrimPrefix(strings.TrimSpace(key), "/")
+	if !strings.HasSuffix(key, "/tender.json") {
 		return "", "", false
 	}
 	trimmed := strings.TrimSuffix(key, "/tender.json")
-	if trimmed == key {
+	if trimmed == "" {
 		return "", "", false
 	}
-	parts := strings.SplitN(trimmed, "/", 3)
+	last := strings.LastIndex(trimmed, "/")
-	if len(parts) != 2 {
+	if last < 0 {
 		return "", "", false
 	}
-	cfid := strings.TrimPrefix(parts[0], procedurePrefix)
+	noticePublicationID = strings.TrimSpace(trimmed[last+1:])
-	if cfid == "" || parts[1] == "" {
+	parent := strings.TrimSpace(trimmed[:last])
+	if noticePublicationID == "" || parent == "" {
 		return "", "", false
 	}
-	return cfid, parts[1], true
+	contractFolderID = stripLeadingProcPrefix(parent)
+	if contractFolderID == "" {
+		return "", "", false
+	}
+	return contractFolderID, noticePublicationID, true
+}
+
+func stripLeadingProcPrefix(parent string) string {
+	if len(parent) >= 5 && strings.EqualFold(parent[:5], "proc_") {
+		return strings.TrimSpace(parent[5:])
+	}
+	return parent
 }
 
 // countDone counts how many items have Done=true