diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 1b3a4cc..43156e1 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -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 == "" { diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 2b60f27..5c0f69a 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -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) diff --git a/cmd/worker/workers/summarizer.go b/cmd/worker/workers/summarizer.go deleted file mode 100644 index 4a4de77..0000000 --- a/cmd/worker/workers/summarizer.go +++ /dev/null @@ -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, - }) -} diff --git a/internal/tender/errors.go b/internal/tender/errors.go index a3f3486..ad0d5c5 100644 --- a/internal/tender/errors.go +++ b/internal/tender/errors.go @@ -9,5 +9,6 @@ var ( ErrTenderNotFound = errors.New("tender: not found") ErrTenderMissingNoticeID = errors.New("tender: missing notice publication id") ErrTenderMissingContractFolderID = errors.New("tender: missing contract folder id") + ErrTenderDocumentsNotScraped = errors.New("tender: documents not scraped") ErrAINotConfigured = errors.New("tender: ai service not configured") ) diff --git a/internal/tender/handler.go b/internal/tender/handler.go index 92e6196..a6b3b89 100644 --- a/internal/tender/handler.go +++ b/internal/tender/handler.go @@ -537,7 +537,7 @@ func (h *TenderHandler) streamDocumentDownload(c echo.Context, tenderID, filenam // GetAISummary retrieves the AI-generated overall summary for a tender // @Summary Get AI-generated tender summary -// @Description Retrieve the AI-generated overall summary for a tender from the AI pipeline storage +// @Description Retrieve the AI-generated overall summary for a tender. When documents are scraped but no summary exists yet, on-demand summarization is triggered automatically. // @Tags Admin-Tenders // @Produce json // @Param id path string true "Tender ID" @@ -637,6 +637,8 @@ func mapAITriggerHTTPError(c echo.Context, err error, action string) error { return response.NotFound(c, "Tender not found") case errors.Is(err, ErrTenderMissingNoticeID), errors.Is(err, ErrTenderMissingContractFolderID): return response.BadRequest(c, "Tender not eligible", "Tender is missing required identifiers for AI processing") + case errors.Is(err, ErrTenderDocumentsNotScraped): + return response.BadRequest(c, "Tender not eligible", "Tender documents have not been scraped yet") case errors.Is(err, ErrAINotConfigured): return response.InternalServerError(c, "AI service is not available") default: @@ -683,7 +685,7 @@ func (h *TenderHandler) TriggerAITranslate(c echo.Context) error { // GetPublicAISummary retrieves the AI-generated overall summary for a tender (public endpoint) // @Summary Get AI-generated tender summary (public) -// @Description Retrieve the AI-generated overall summary for a tender from the AI pipeline storage +// @Description Retrieve the AI-generated overall summary for a tender. When documents are scraped but no summary exists yet, on-demand summarization is triggered automatically. // @Tags Tenders // @Produce json // @Param id path string true "Tender ID" diff --git a/internal/tender/service.go b/internal/tender/service.go index 5c02d4b..5b305c9 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -168,44 +168,27 @@ func (s *tenderService) buildDetailResponse(ctx context.Context, tender *Tender, return resp } -// enrichWithAISummary loads overall_summary from MinIO (PROC_//tender.json) -// and attaches it to the response. The primary notice publication id is tried first, then related ids. -// A previously stored ai_overall_summary on the tender (already mapped in ToResponse) is kept when MinIO -// has no ready summary. On success the tender document is updated so list/search can return it without -// another MinIO round-trip. +// enrichWithAISummary attaches the overall AI summary to a tender detail response. +// It reads from MinIO when available; for scraped tenders without a summary it triggers +// on-demand summarization via the AI service (same path as GetAISummary). func (s *tenderService) enrichWithAISummary(ctx context.Context, t *Tender, resp *TenderResponse) { if t == nil || resp == nil { return } - if s.aiSummarizerStorage == nil || strings.TrimSpace(t.ContractFolderID) == "" { - return - } - enrichCtx, cancel := context.WithTimeout(ctx, aiSummaryEnrichmentTimeout) - defer cancel() - - summary, usedNotice, err := s.lookupSummaryForTender(enrichCtx, t) + result, err := s.resolveAISummary(ctx, t) if err != nil { - s.logger.Debug("AI summary not available for tender", map[string]interface{}{ - "tender_id": resp.ID, - "contract_folder_id": t.ContractFolderID, - "notice_id": t.NoticePublicationID, - "error": err.Error(), + s.logger.Debug("AI summary resolution failed for tender detail", map[string]interface{}{ + "tender_id": resp.ID, + "error": err.Error(), }) return } - - resp.OverallSummary = summary - if usedNotice != t.NoticePublicationID { - s.logger.Debug("AI summary resolved from related notice", map[string]interface{}{ - "tender_id": resp.ID, - "contract_folder_id": t.ContractFolderID, - "primary_notice_id": t.NoticePublicationID, - "used_notice_id": usedNotice, - }) + if result == nil || strings.TrimSpace(result.OverallSummary) == "" { + return } - s.persistAIOverallSummary(enrichCtx, t, summary) + resp.OverallSummary = result.OverallSummary } // enrichWithTranslation overlays title/description from MinIO when the pipeline has finished @@ -1131,8 +1114,8 @@ func (s *tenderService) CleanupOldData(ctx context.Context, olderThan time.Durat } // GetAISummary retrieves the AI-generated summary for a tender. -// It loads from MinIO storage (async pipeline result). When no summary exists yet, -// the pipeline is still running, or storage is unavailable, it returns 200-compatible +// It reads from MinIO when available; for scraped tenders without a summary it triggers +// on-demand summarization. When no summary can be produced it returns 200-compatible // data with an empty overall_summary and a Source hint instead of an error. func (s *tenderService) GetAISummary(ctx context.Context, id string) (*AISummaryResponse, error) { if id == "" { @@ -1151,6 +1134,12 @@ func (s *tenderService) GetAISummary(ctx context.Context, id string) (*AISummary return nil, fmt.Errorf("tender not found") } + return s.resolveAISummary(ctx, t) +} + +// resolveAISummary returns the overall AI summary for a tender, triggering on-demand +// summarization when documents are scraped but no summary exists in storage yet. +func (s *tenderService) resolveAISummary(ctx context.Context, t *Tender) (*AISummaryResponse, error) { noticeID := strings.TrimSpace(t.NoticePublicationID) emptyResp := func(source string) *AISummaryResponse { return &AISummaryResponse{ @@ -1164,45 +1153,146 @@ func (s *tenderService) GetAISummary(ctx context.Context, id string) (*AISummary return emptyResp("unavailable"), nil } - if s.aiSummarizerStorage == nil { - s.logger.Debug("AI summarizer storage not configured; returning empty AI summary", map[string]interface{}{ - "tender_id": id, + if s.aiSummarizerStorage == nil && s.aiSummarizerClient == nil { + s.logger.Debug("AI summarizer not configured; returning empty AI summary", map[string]interface{}{ + "tender_id": t.ID.Hex(), }) return emptyResp("unavailable"), nil } - summary, usedNotice, err := s.lookupSummaryForTender(ctx, t) - if err == nil { - return &AISummaryResponse{ - NoticeID: usedNotice, - OverallSummary: summary, - Source: "storage", - }, nil + lookupCtx := ctx + if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) > aiSummaryEnrichmentTimeout { + var cancel context.CancelFunc + lookupCtx, cancel = context.WithTimeout(ctx, aiSummaryEnrichmentTimeout) + defer cancel() } - s.logger.Info("AI summary not available from storage", map[string]interface{}{ - "tender_id": id, - "contract_folder_id": t.ContractFolderID, - "notice_id": noticeID, - "error": err.Error(), - }) + if s.aiSummarizerStorage != nil { + summary, usedNotice, err := s.lookupSummaryForTender(lookupCtx, t) + if err == nil { + s.persistAIOverallSummary(ctx, t, summary) + return &AISummaryResponse{ + NoticeID: usedNotice, + OverallSummary: summary, + Source: "storage", + }, nil + } - if errors.Is(err, ai_summarizer.ErrSummaryNotReady) { - return emptyResp("pending"), nil + if !errors.Is(err, ai_summarizer.ErrSummaryNotReady) && + !errors.Is(err, ai_summarizer.ErrObjectNotFound) && + !errors.Is(err, ai_summarizer.ErrNoOverallSummary) { + if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) { + s.logger.Error("Cannot retrieve AI summary: MinIO connection unavailable", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "contract_folder_id": t.ContractFolderID, + "notice_id": noticeID, + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to get AI summary: MinIO connection unavailable: %w", err) + } + return nil, fmt.Errorf("failed to get AI summary: %w", err) + } } - if errors.Is(err, ai_summarizer.ErrObjectNotFound) || errors.Is(err, ai_summarizer.ErrNoOverallSummary) { + + if !s.tenderHasScrapedDocuments(t) { return emptyResp("unavailable"), nil } - if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) { - s.logger.Error("Cannot retrieve AI summary: MinIO connection unavailable", map[string]interface{}{ - "tender_id": id, + + if s.aiSummarizerClient == nil { + return emptyResp("unavailable"), nil + } + + resp, err := s.fetchAISummaryOnDemand(ctx, t) + if err != nil { + s.logger.Error("On-demand AI summarization on tender read failed", map[string]interface{}{ + "tender_id": t.ID.Hex(), "contract_folder_id": t.ContractFolderID, "notice_id": noticeID, "error": err.Error(), }) - return nil, fmt.Errorf("failed to get AI summary: MinIO connection unavailable: %w", err) + return emptyResp("pending"), nil } - return nil, fmt.Errorf("failed to get AI summary: %w", err) + + overall := derefString(resp.OverallSummary) + usedNotice := resp.ResolvedNoticePublicationID() + if usedNotice == "" { + usedNotice = noticeID + } + + if overall == "" && s.aiSummarizerStorage != nil { + if summary, notice, lookupErr := s.lookupSummaryForTender(ctx, t); lookupErr == nil { + overall = summary + usedNotice = notice + } + } + + if overall != "" { + s.persistAIOverallSummary(ctx, t, overall) + return &AISummaryResponse{ + NoticeID: usedNotice, + OverallSummary: overall, + Source: "generated", + }, nil + } + + return emptyResp("pending"), nil +} + +func (s *tenderService) tenderHasScrapedDocuments(t *Tender) bool { + if t == nil { + return false + } + if t.ProcessingMetadata.DocumentsScraped { + return true + } + return len(t.ScrapedDocuments) > 0 +} + +func (s *tenderService) fetchAISummaryOnDemand(ctx context.Context, t *Tender) (*ai_summarizer.SummarizeResponse, error) { + if s.aiSummarizerClient == nil { + return nil, ErrAINotConfigured + } + if strings.TrimSpace(t.NoticePublicationID) == "" { + return nil, ErrTenderMissingNoticeID + } + if strings.TrimSpace(t.ContractFolderID) == "" { + return nil, ErrTenderMissingContractFolderID + } + + reqBody := ai_summarizer.NewSummarizeRequest(t.ContractFolderID, t.NoticePublicationID) + s.logger.Info("Triggering on-demand AI summarization", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "contract_folder_id": t.ContractFolderID, + "notice_publication_id": t.NoticePublicationID, + }) + + resp, err := s.aiSummarizerClient.FetchSummaryOnDemand(ctx, reqBody) + if err != nil { + return nil, fmt.Errorf("AI summarization request failed: %w", err) + } + + s.markTenderSummarized(ctx, t) + return resp, nil +} + +func (s *tenderService) markTenderSummarized(ctx context.Context, t *Tender) { + now := time.Now().Unix() + t.ProcessingMetadata.DocumentsSummarized = true + t.ProcessingMetadata.DocumentsSummarizedAt = now + t.UpdatedAt = now + if err := s.repository.Update(ctx, t); err != nil { + s.logger.Warn("Failed to update tender summarization status", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "error": err.Error(), + }) + } +} + +func derefString(value *string) string { + if value == nil { + return "" + } + return strings.TrimSpace(*value) } // TriggerAISummarize triggers on-demand AI summarization for a tender. @@ -1216,7 +1306,6 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS return nil, ErrAINotConfigured } - // Get tender to build the request t, err := s.repository.GetByID(ctx, id) if err != nil { s.logger.Error("Failed to get tender for on-demand summarization", map[string]interface{}{ @@ -1236,23 +1325,11 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS return nil, ErrTenderNotFound } - if t.NoticePublicationID == "" { - return nil, ErrTenderMissingNoticeID - } - if strings.TrimSpace(t.ContractFolderID) == "" { - return nil, ErrTenderMissingContractFolderID + if !s.tenderHasScrapedDocuments(t) { + return nil, ErrTenderDocumentsNotScraped } - reqBody := ai_summarizer.NewSummarizeRequest(t.ContractFolderID, t.NoticePublicationID) - - s.logger.Info("Triggering on-demand AI summarization", map[string]interface{}{ - "tender_id": id, - "contract_folder_id": t.ContractFolderID, - "notice_publication_id": t.NoticePublicationID, - }) - - // Call the AI service - resp, err := s.aiSummarizerClient.FetchSummaryOnDemand(ctx, reqBody) + resp, err := s.fetchAISummaryOnDemand(ctx, t) if err != nil { s.logger.Error("On-demand AI summarization failed", map[string]interface{}{ "tender_id": id, @@ -1260,18 +1337,17 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS "notice_publication_id": t.NoticePublicationID, "error": err.Error(), }) - return nil, fmt.Errorf("AI summarization request failed: %w", err) + return nil, err } - // Map the response to our response type summaries := make([]AIDocumentSummary, 0, len(resp.Summaries)) - for _, s := range resp.Summaries { + for _, item := range resp.Summaries { summaries = append(summaries, AIDocumentSummary{ - DocumentName: s.DocumentName, - DocumentType: s.DocumentType, - Summary: s.Summary, - SummaryModel: s.SummaryModel, - Error: s.Error, + DocumentName: item.DocumentName, + DocumentType: item.DocumentType, + Summary: item.Summary, + SummaryModel: item.SummaryModel, + Error: item.Error, }) } @@ -1291,6 +1367,10 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS contractFolderID = t.ContractFolderID } + if overall := derefString(resp.OverallSummary); overall != "" { + s.persistAIOverallSummary(ctx, t, overall) + } + return &AISummarizeResponse{ ContractFolderID: contractFolderID, NoticePublicationID: noticePublicationID,