Implemented the integration with scraper python server, tenders summarise, and some fixes.

2025-12-27 12:47:08 +03:30
parent e4da3b5bb2
commit f6bdcc6d7c
18 changed files with 1054 additions and 45 deletions
@@ -11,6 +11,7 @@ import (
 	"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"
@@ -23,9 +24,10 @@ type NoticeWorker struct {
 	NoticeRepo notice.Repository
 	TenderRepo tender.TenderRepository
 	GLM        *glm.SDK
+	Scraper    scraper.SDK
 }

-func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, glmService *glm.SDK) *NoticeWorker {
+func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, glmService *glm.SDK, scraper scraper.SDK) *NoticeWorker {
 	return &NoticeWorker{
 		Mongo:      mongo,
 		Logger:     logger,
@@ -33,15 +35,74 @@ func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notif
 		NoticeRepo: noticeRepo,
 		TenderRepo: tenderRepo,
 		GLM:        glmService,
+		Scraper:    scraper,
 	}
 }

+// 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()
+
 	limit := 10
 	skip := 0
-	for {
+	processedCount := 0
+	maxToProcess := 5 // Limit processing to allow scraper worker to run
+
+	for processedCount < maxToProcess {
 		notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), limit, skip)
 		if err != nil {
 			w.Logger.Error("Failed to get notices", map[string]interface{}{
@@ -51,6 +112,9 @@ func (w *NoticeWorker) Run() {
 		}

 		if len(notices) == 0 {
+			w.Logger.Info("No more unprocessed notices found", map[string]interface{}{
+				"processed_count": processedCount,
+			})
 			break
 		}

@@ -89,16 +153,99 @@ func (w *NoticeWorker) Run() {
 				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(),
+					})
+				}
 			}
-			w.Logger.Info("Notice processed", map[string]interface{}{
-				"notice": n.ID.Hex(),
-			})
 		}

 		skip += limit
+
+		// Check if we've reached the processing limit
+		if processedCount >= maxToProcess {
+			w.Logger.Info("Reached maximum processing limit, stopping", map[string]interface{}{
+				"processed_count": processedCount,
+				"max_to_process":  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 {