Refactor AI summarization functionality by removing document summarization worker

- Removed the DocumentSummarizationWorker and its related scheduling logic from the worker bootstrap. - Updated the AI summarizer client initialization comment for clarity. - Added a new error type for cases when tender documents have not been scraped yet, enhancing error handling in the tender service. - Modified API documentation to reflect changes in AI summary retrieval logic, ensuring accurate descriptions of on-demand summarization behavior. This update streamlines the AI summarization process by eliminating the document summarization worker, improving overall system efficiency and clarity in error handling.
2026-06-21 11:20:28 +03:30
parent e31bccced6
commit e04cdd1d10
6 changed files with 163 additions and 302 deletions
@@ -95,20 +95,6 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
 	// Create a single shared cron scheduler for all recurring jobs
 	scheduler := schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo)

-	// Document summarization via external AI service (requires AI_SUMMARIZER_API_BASE_URL)
-	if aiClient != nil {
-		scheduler.AddJob(schedule.Job{
-			Name: "Document Summarization Worker Job",
-			Func: func() {
-				worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, aiClient, aiStorage)
-				worker.Run()
-			},
-			Expr: "0 11 * * * *", // Run at 11 AM every day (after AI pipeline scrape)
-		})
-	} else {
-		appLogger.Warn("Document summarization worker not scheduled: AI summarizer API is not configured", map[string]interface{}{})
-	}
-
 	// Schedule Notice Worker job (no blocking run at startup; first tick runs on schedule)
 	workerInterval := config.Worker.Interval
 	if workerInterval == "" {
@@ -37,7 +37,7 @@ func main() {
 	// Initialize notification service
 	notificationService := bootstrap.InitNotificationService(config.Notification, appLogger)

-	// Initialize AI summarizer client (translation + document summarization workers)
+	// Initialize AI summarizer client (translation worker)
 	aiSummarizerClient := bootstrap.InitAISummarizerClient(config.AISummarizer, mongoManager, appLogger)
 	aiSummarizerStorage := bootstrap.InitAISummarizerStorage(config.AISummarizer, appLogger)

@@ -1,208 +0,0 @@
-package workers
-
-import (
-	"context"
-	"strings"
-	"time"
-	"tm/internal/tender"
-	ai_summarizer "tm/pkg/ai_summarizer"
-	"tm/pkg/logger"
-	"tm/pkg/mongo"
-)
-
-// DocumentSummarizationWorker triggers AI summarization for scraped tenders.
-// The AI service owns MinIO artefacts; this worker marks Mongo processing flags
-// after the pipeline or on-demand summarize API completes.
-type DocumentSummarizationWorker struct {
-	Mongo      *mongo.ConnectionManager
-	Logger     logger.Logger
-	TenderRepo tender.TenderRepository
-	AIClient   *ai_summarizer.Client
-	AIStorage  *ai_summarizer.StorageClient
-}
-
-// NewDocumentSummarizationWorker creates a document summarization worker.
-func NewDocumentSummarizationWorker(
-	mongo *mongo.ConnectionManager,
-	logger logger.Logger,
-	tenderRepo tender.TenderRepository,
-	aiClient *ai_summarizer.Client,
-	aiStorage *ai_summarizer.StorageClient,
-) *DocumentSummarizationWorker {
-	return &DocumentSummarizationWorker{
-		Mongo:      mongo,
-		Logger:     logger,
-		TenderRepo: tenderRepo,
-		AIClient:   aiClient,
-		AIStorage:  aiStorage,
-	}
-}
-
-// Run starts the document summarization process.
-func (w *DocumentSummarizationWorker) Run() {
-	w.Logger.Info("Document summarization worker started", map[string]interface{}{})
-
-	if w.AIClient == nil {
-		w.Logger.Warn("AI summarizer client not configured, skipping document summarization run", map[string]interface{}{})
-		return
-	}
-
-	if refs := w.collectUnSummarizedTenderRefs(context.Background()); len(refs) > 0 {
-		if _, err := w.AIClient.TriggerSummarizeBatch(context.Background(), refs); err != nil {
-			w.Logger.Warn("Failed to trigger AI summarize batch (continuing with per-tender sync)", map[string]interface{}{
-				"tender_count": len(refs),
-				"error":        err.Error(),
-			})
-		}
-	}
-
-	limit := 5
-	skip := 0
-	for {
-		tenders, totalCount, err := w.TenderRepo.GetUnSummarizedTenders(context.Background(), limit, skip)
-		if err != nil {
-			w.Logger.Error("Failed to get un-summarized tenders", map[string]interface{}{
-				"error": err.Error(),
-			})
-			return
-		}
-
-		w.Logger.Info("Found tenders for document summarization", map[string]interface{}{
-			"count":       len(tenders),
-			"total_count": totalCount,
-			"skip":        skip,
-		})
-
-		if len(tenders) == 0 {
-			break
-		}
-
-		for i := range tenders {
-			t := &tenders[i]
-			w.Logger.Info("Processing tender for document summarization", map[string]interface{}{
-				"tender_id":             t.ID.Hex(),
-				"notice_publication_id": t.NoticePublicationID,
-			})
-			w.summarizeTender(t)
-		}
-
-		skip += len(tenders)
-		if int64(skip) >= totalCount {
-			break
-		}
-	}
-
-	w.Logger.Info("Document summarization worker completed", map[string]interface{}{})
-}
-
-func (w *DocumentSummarizationWorker) collectUnSummarizedTenderRefs(ctx context.Context) []ai_summarizer.TenderRef {
-	limit := 100
-	skip := 0
-	refs := make([]ai_summarizer.TenderRef, 0)
-
-	for {
-		tenders, totalCount, err := w.TenderRepo.GetUnSummarizedTenders(ctx, limit, skip)
-		if err != nil {
-			w.Logger.Warn("Failed to collect tenders for summarize batch", map[string]interface{}{
-				"error": err.Error(),
-			})
-			return refs
-		}
-		if len(tenders) == 0 {
-			return refs
-		}
-
-		for i := range tenders {
-			t := &tenders[i]
-			if strings.TrimSpace(t.ContractFolderID) == "" || strings.TrimSpace(t.NoticePublicationID) == "" {
-				continue
-			}
-			refs = append(refs, ai_summarizer.NewTenderRef(t.ContractFolderID, t.NoticePublicationID))
-		}
-
-		skip += len(tenders)
-		if int64(skip) >= totalCount {
-			return refs
-		}
-	}
-}
-
-func (w *DocumentSummarizationWorker) summarizeTender(t *tender.Tender) {
-	if strings.TrimSpace(t.NoticePublicationID) == "" {
-		w.Logger.Warn("Skipping tender summarization: missing notice_publication_id", map[string]interface{}{
-			"tender_id": t.ID.Hex(),
-		})
-		w.markTenderSummarized(t)
-		return
-	}
-	if strings.TrimSpace(t.ContractFolderID) == "" {
-		w.Logger.Warn("Skipping tender summarization: missing contract_folder_id", map[string]interface{}{
-			"tender_id": t.ID.Hex(),
-		})
-		w.markTenderSummarized(t)
-		return
-	}
-
-	ctx := context.Background()
-	if w.isSummaryReadyInStorage(ctx, t) {
-		w.markTenderSummarized(t)
-		return
-	}
-
-	req := ai_summarizer.NewSummarizeRequest(t.ContractFolderID, t.NoticePublicationID)
-	if _, err := w.AIClient.FetchSummaryOnDemand(ctx, req); err != nil {
-		w.Logger.Error("Failed to summarize tender via AI service", map[string]interface{}{
-			"tender_id":             t.ID.Hex(),
-			"notice_publication_id": t.NoticePublicationID,
-			"contract_folder_id":    t.ContractFolderID,
-			"error":                 err.Error(),
-		})
-		return
-	}
-
-	if !w.isSummaryReadyInStorage(ctx, t) {
-		w.Logger.Warn("AI summarize completed but storage is not ready yet", map[string]interface{}{
-			"tender_id":             t.ID.Hex(),
-			"notice_publication_id": t.NoticePublicationID,
-		})
-	}
-
-	w.markTenderSummarized(t)
-}
-
-func (w *DocumentSummarizationWorker) isSummaryReadyInStorage(ctx context.Context, t *tender.Tender) bool {
-	if w.AIStorage == nil {
-		return false
-	}
-	ready, err := w.AIStorage.IsSummaryReady(ctx, t.ContractFolderID, t.NoticePublicationID)
-	if err != nil {
-		w.Logger.Debug("Could not check summarization status in storage", map[string]interface{}{
-			"tender_id":             t.ID.Hex(),
-			"notice_publication_id": t.NoticePublicationID,
-			"error":                 err.Error(),
-		})
-		return false
-	}
-	return ready
-}
-
-func (w *DocumentSummarizationWorker) markTenderSummarized(t *tender.Tender) {
-	now := time.Now().Unix()
-	t.ProcessingMetadata.DocumentsSummarized = true
-	t.ProcessingMetadata.DocumentsSummarizedAt = now
-	t.UpdatedAt = now
-
-	if err := w.TenderRepo.Update(context.Background(), t); err != nil {
-		w.Logger.Error("Failed to update tender summarization status", map[string]interface{}{
-			"tender_id":             t.ID.Hex(),
-			"notice_publication_id": t.NoticePublicationID,
-			"error":                 err.Error(),
-		})
-		return
-	}
-
-	w.Logger.Info("Tender marked as summarized", map[string]interface{}{
-		"tender_id":             t.ID.Hex(),
-		"notice_publication_id": t.NoticePublicationID,
-	})
-}