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) {