From ca2a1b44253dce3fa022d33224bd11784b8fe13f Mon Sep 17 00:00:00 2001 From: Mazyar Date: Sun, 14 Jun 2026 14:59:37 +0330 Subject: [PATCH] Enhance AI pipeline service with document metadata synchronization - Updated the AI pipeline service to include a new `ScrapedDocumentMetadataSyncer` interface for persisting scraped document metadata onto tender records. - Modified the `NewService` function to accept the new metadata syncer dependency. - Implemented synchronization of scraped document metadata in the `ScrapeDocuments` and `Run` methods. - Enhanced the tender service to enrich search filters based on scraped documents and added a new method for syncing scraped documents from storage. - Updated the `SearchForm` to include `ContractFolderIDsWithDocuments` for better handling of scraped documents in queries. This update improves the integration of scraped document handling within the AI pipeline, enhancing data consistency and operational efficiency in the tender management system. --- cmd/web/main.go | 2 +- internal/ai_pipeline/service.go | 36 ++++++-- internal/tender/form.go | 2 + internal/tender/handler.go | 6 +- internal/tender/repository.go | 9 +- internal/tender/service.go | 146 +++++++++++++++++++++++++++++--- 6 files changed, 174 insertions(+), 27 deletions(-) diff --git a/cmd/web/main.go b/cmd/web/main.go index ce99dc4..a69678b 100644 --- a/cmd/web/main.go +++ b/cmd/web/main.go @@ -232,7 +232,7 @@ func main() { documentScraperService := document_scraper.NewService(tenderRepository, logger) dashboardRepository := dashboard.NewRepository(mongoManager, logger) dashboardService := dashboard.NewService(dashboardRepository, logger) - aiPipelineService := ai_pipeline.NewService(aiPipelineClient, logger) + aiPipelineService := ai_pipeline.NewService(aiPipelineClient, logger, tenderService) logger.Info("Services initialized successfully", map[string]interface{}{ "services": []string{"customer", "user", "company", "category", "tender", "feedback", "tender_approval", "inquiry", "flag", "notification", "contact", "cms", "scraper", "kanban", "document_scraper", "dashboard", "ai_pipeline"}, }) diff --git a/internal/ai_pipeline/service.go b/internal/ai_pipeline/service.go index 35fe1bc..4fafaad 100644 --- a/internal/ai_pipeline/service.go +++ b/internal/ai_pipeline/service.go @@ -46,16 +46,23 @@ type Service interface { GetMinioStats(ctx context.Context) (ai_summarizer.PipelineMinioStatsResponse, error) } +// ScrapedDocumentMetadataSyncer persists MinIO scraped-document metadata onto tender records. +type ScrapedDocumentMetadataSyncer interface { + SyncScrapedDocumentsFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) error +} + type service struct { - client Client - logger logger.Logger + client Client + logger logger.Logger + metadataSyncer ScrapedDocumentMetadataSyncer } // NewService creates an AI pipeline admin service. -func NewService(client Client, log logger.Logger) Service { +func NewService(client Client, log logger.Logger, metadataSyncer ScrapedDocumentMetadataSyncer) Service { return &service{ - client: client, - logger: log, + client: client, + logger: log, + metadataSyncer: metadataSyncer, } } @@ -89,6 +96,10 @@ func (s *service) ScrapeDocuments(ctx context.Context, form ScrapeDocumentsForm) return nil, fmt.Errorf("scrape documents failed: %w", err) } + if resp != nil && resp.Success && resp.FilesDownloaded > 0 { + s.syncScrapedDocumentMetadata(ctx, form.ContractFolderID, form.NoticePublicationID) + } + return resp, nil } @@ -222,6 +233,8 @@ func (s *service) Run(ctx context.Context, form PipelineRunForm) (map[string]int return nil, fmt.Errorf("pipeline run failed: %w", err) } + s.syncScrapedDocumentMetadata(ctx, req.ContractFolderID, req.NoticePublicationID) + return resp, nil } @@ -398,3 +411,16 @@ func validateTenderBatch(forms []TenderRefForm) ([]ai_summarizer.TenderRef, erro } return toTenderRefs(forms), nil } + +func (s *service) syncScrapedDocumentMetadata(ctx context.Context, contractFolderID, noticePublicationID string) { + if s.metadataSyncer == nil { + return + } + if err := s.metadataSyncer.SyncScrapedDocumentsFromStorage(ctx, contractFolderID, noticePublicationID); err != nil { + s.logger.Warn("Failed to sync scraped document metadata to Mongo", map[string]interface{}{ + "contract_folder_id": contractFolderID, + "notice_publication_id": noticePublicationID, + "error": err.Error(), + }) + } +} diff --git a/internal/tender/form.go b/internal/tender/form.go index 8cd8f23..a80c453 100644 --- a/internal/tender/form.go +++ b/internal/tender/form.go @@ -62,6 +62,8 @@ type SearchForm struct { OnlyActiveDeadlines bool `query:"-" valid:"optional"` Language *string `query:"lang" valid:"optional"` DocumentsScraped bool `query:"documents_scraped" valid:"optional"` + // ContractFolderIDsWithDocuments is populated by the service from MinIO when DocumentsScraped is true. + ContractFolderIDsWithDocuments []string `query:"-" valid:"optional"` } // SearchResponse represents the response for listing tenders diff --git a/internal/tender/handler.go b/internal/tender/handler.go index dc092c3..db45636 100644 --- a/internal/tender/handler.go +++ b/internal/tender/handler.go @@ -158,7 +158,7 @@ func (h *TenderHandler) Delete(c echo.Context) error { // @Param submission_date_to query number false "Submission deadline to (Unix timestamp, seconds)" // @Param languages query []string false "Filter by languages (comma-separated)" // @Param buyer_organization_id query string false "Filter by buyer organization ID" -// @Param documents_scraped query bool false "When true, only tenders with scraped documents" +// @Param documents_scraped query bool false "When true, only tenders whose contract_folder_id has documents in MinIO" // @Param limit query int false "Number of items per page (default: 20, max: 100)" // @Param offset query int false "Number of items to skip (default: 0). Prefer cursor for deep pages." // @Param cursor query string false "Opaque cursor from previous page meta.next_cursor" @@ -229,7 +229,7 @@ func (h *TenderHandler) Search(c echo.Context) error { // @Param submission_date_to query number false "Submission deadline to (Unix seconds)" // @Param languages query []string false "Filter by languages (comma-separated)" // @Param buyer_organization_id query string false "Filter by buyer organization ID" -// @Param documents_scraped query bool false "When true, only tenders with scraped documents" +// @Param documents_scraped query bool false "When true, only tenders whose contract_folder_id has documents in MinIO" // @Param limit query int false "Number of items per page (default: 20, max: 100)" // @Param offset query int false "Number of items to skip (default: 0). Prefer cursor for deep pages." // @Param cursor query string false "Opaque cursor from previous page meta.next_cursor" @@ -312,7 +312,7 @@ func (h *TenderHandler) GetPublicTenders(c echo.Context) error { // @Param submission_date_to query number false "Submission deadline to (Unix seconds)" // @Param languages query []string false "Filter by languages (comma-separated)" // @Param buyer_organization_id query string false "Filter by buyer organization ID" -// @Param documents_scraped query bool false "When true, only tenders with scraped documents" +// @Param documents_scraped query bool false "When true, only tenders whose contract_folder_id has documents in MinIO" // @Param limit query int false "Number of items per page (default: 20, max: 100)" // @Param offset query int false "Number of items to skip (default: 0). Prefer cursor for deep pages." // @Param cursor query string false "Opaque cursor from previous page meta.next_cursor" diff --git a/internal/tender/repository.go b/internal/tender/repository.go index 6e9f57a..fcd22d3 100644 --- a/internal/tender/repository.go +++ b/internal/tender/repository.go @@ -1060,11 +1060,10 @@ func (r *tenderRepository) buildSearchFilter(form *SearchForm) bson.M { } if form.DocumentsScraped { - scrapedClause := bson.M{ - "$or": []bson.M{ - {"processing_metadata.documents_scraped": true}, - {"scraped_documents.0": bson.M{"$exists": true}}, - }, + folderIDs := dedupeNonEmptyStrings(form.ContractFolderIDsWithDocuments) + scrapedClause := bson.M{"contract_folder_id": bson.M{"$in": folderIDs}} + if len(folderIDs) == 0 { + scrapedClause = bson.M{"_id": bson.M{"$exists": false}} } if len(filter) == 0 { return scrapedClause diff --git a/internal/tender/service.go b/internal/tender/service.go index 8132f87..5c02d4b 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -70,6 +70,8 @@ type Service interface { // ListTendersWithScrapedDocuments lists tenders that have documents in MinIO. When syncMongoMetadata is true, // each tender is re-listed from MinIO and scraped document metadata is persisted to Mongo (slow). ListTendersWithScrapedDocuments(ctx context.Context, pagination *response.Pagination, syncMongoMetadata bool) (*TendersWithScrapedDocumentsResponse, error) + // SyncScrapedDocumentsFromStorage lists documents in MinIO for a tender and persists scraped metadata to Mongo. + SyncScrapedDocumentsFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) error // Maintenance operations UpdateExpiredTenders(ctx context.Context) error @@ -886,6 +888,15 @@ func (s *tenderService) Search(ctx context.Context, form *SearchForm, pagination } } + if form.DocumentsScraped { + if err := s.enrichDocumentsScrapedFilter(ctx, form); err != nil { + return nil, err + } + if len(form.ContractFolderIDsWithDocuments) == 0 { + return s.emptySearchResponse(pagination), nil + } + } + page, err := s.repository.Search(ctx, form, pagination) if err != nil { s.logger.Error("Failed to list tenders", map[string]interface{}{ @@ -997,6 +1008,15 @@ func (s *tenderService) Recommend(ctx context.Context, form *SearchForm, paginat } } + if form.DocumentsScraped { + if err := s.enrichDocumentsScrapedFilter(ctx, form); err != nil { + return nil, err + } + if len(form.ContractFolderIDsWithDocuments) == 0 { + return s.emptySearchResponse(pagination), nil + } + } + page, err := s.repository.Search(ctx, form, pagination) if err != nil { s.logger.Error("Failed to list tenders", map[string]interface{}{ @@ -1410,25 +1430,13 @@ func (s *tenderService) GetAllAISummaries(ctx context.Context) ([]ai_summarizer. // documents in MinIO. By default uses counts from the initial MinIO procedure scan only (fast). // Set syncMongoMetadata=true to list every notice folder again and persist ScrapedDocuments to Mongo (slow). func (s *tenderService) ListTendersWithScrapedDocuments(ctx context.Context, pagination *response.Pagination, syncMongoMetadata bool) (*TendersWithScrapedDocumentsResponse, error) { - if s.aiSummarizerStorage == nil { - return nil, fmt.Errorf("document storage is not configured") - } - - type procedureLister interface { - ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) - } - storage, ok := s.aiSummarizerStorage.(procedureLister) - if !ok { - return nil, fmt.Errorf("document storage does not support procedure listing") - } - s.logger.Info("Listing tenders with scraped documents from MinIO", map[string]interface{}{ "limit": pagination.Limit, "offset": pagination.Offset, "sync_mongo_metadata": syncMongoMetadata, }) - procedures, err := storage.ListProceduresWithDocuments(ctx) + procedures, err := s.listProceduresWithDocuments(ctx) if err != nil { fields := map[string]interface{}{"error": err.Error()} if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) { @@ -1564,6 +1572,118 @@ func (s *tenderService) persistScrapedDocuments(ctx context.Context, t *Tender, } } +type procedureDocumentLister interface { + ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) +} + +func (s *tenderService) procedureDocumentStorage() (procedureDocumentLister, error) { + if s.aiSummarizerStorage == nil { + return nil, fmt.Errorf("document storage is not configured") + } + storage, ok := s.aiSummarizerStorage.(procedureDocumentLister) + if !ok { + return nil, fmt.Errorf("document storage does not support procedure listing") + } + return storage, nil +} + +func (s *tenderService) listProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) { + storage, err := s.procedureDocumentStorage() + if err != nil { + return nil, err + } + procedures, err := storage.ListProceduresWithDocuments(ctx) + if err != nil { + return nil, err + } + return procedures, nil +} + +func (s *tenderService) enrichDocumentsScrapedFilter(ctx context.Context, form *SearchForm) error { + procedures, err := s.listProceduresWithDocuments(ctx) + if err != nil { + s.logger.Error("Failed to resolve tenders with scraped documents from MinIO", map[string]interface{}{ + "error": err.Error(), + }) + if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) { + return fmt.Errorf("failed to resolve tenders with scraped documents: MinIO connection unavailable: %w", err) + } + return fmt.Errorf("failed to resolve tenders with scraped documents: %w", err) + } + + folderIDs := make([]string, 0, len(procedures)) + for _, proc := range procedures { + if proc.DocumentCount <= 0 { + continue + } + if id := strings.TrimSpace(proc.ContractFolderID); id != "" { + folderIDs = append(folderIDs, id) + } + } + form.ContractFolderIDsWithDocuments = folderIDs + return nil +} + +func (s *tenderService) emptySearchResponse(pagination *response.Pagination) *SearchResponse { + return &SearchResponse{ + Tenders: []TenderResponse{}, + Metadata: pagination.ListMeta(0, "", false, pagination.Offset), + } +} + +// SyncScrapedDocumentsFromStorage lists documents in MinIO for a tender and persists scraped metadata to Mongo. +func (s *tenderService) SyncScrapedDocumentsFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) error { + contractFolderID = strings.TrimSpace(contractFolderID) + noticePublicationID = strings.TrimSpace(noticePublicationID) + if contractFolderID == "" { + return fmt.Errorf("contract folder ID is required") + } + + var t *Tender + var err error + if noticePublicationID != "" { + t, err = s.repository.GetByNoticePublicationID(ctx, noticePublicationID) + } else { + byFolder, lookupErr := s.repository.GetLatestByContractFolderIDs(ctx, []string{contractFolderID}) + err = lookupErr + if lookupErr == nil { + t = byFolder[contractFolderID] + } + } + if err != nil { + s.logger.Error("Failed to load tender for scraped document sync", map[string]interface{}{ + "contract_folder_id": contractFolderID, + "notice_publication_id": noticePublicationID, + "error": err.Error(), + }) + return fmt.Errorf("failed to load tender for scraped document sync: %w", err) + } + if t == nil { + s.logger.Warn("Skipping scraped document metadata sync: tender not found", map[string]interface{}{ + "contract_folder_id": contractFolderID, + "notice_publication_id": noticePublicationID, + }) + return nil + } + + docs, err := s.listDocumentsForTender(ctx, t) + if err != nil { + s.logger.Error("Failed to list scraped documents from storage", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "contract_folder_id": contractFolderID, + "notice_publication_id": noticePublicationID, + "error": err.Error(), + }) + return fmt.Errorf("failed to list scraped documents from storage: %w", err) + } + if len(docs) == 0 { + return nil + } + + s.persistScrapedDocuments(ctx, t, docs) + return nil +} + // TriggerAITranslate triggers on-demand AI translation for a tender. Results are // read from MinIO when available, otherwise fetched from the AI service (which persists to MinIO). func (s *tenderService) TriggerAITranslate(ctx context.Context, id, language string) (*AITranslateResponse, error) {