get scraped documents list performance update
This commit is contained in:
@@ -5,6 +5,7 @@ import (
|
|||||||
"mime"
|
"mime"
|
||||||
"net/http"
|
"net/http"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
"tm/pkg/logger"
|
"tm/pkg/logger"
|
||||||
orm "tm/pkg/mongo"
|
orm "tm/pkg/mongo"
|
||||||
@@ -661,11 +662,12 @@ func (h *TenderHandler) TriggerPublicAITranslate(c echo.Context) error {
|
|||||||
|
|
||||||
// ListTendersWithScrapedDocuments lists tenders that have scraped documents in the database
|
// ListTendersWithScrapedDocuments lists tenders that have scraped documents in the database
|
||||||
// @Summary List tenders with scraped documents
|
// @Summary List tenders with scraped documents
|
||||||
// @Description Lists database tenders whose contract_folder_id has documents in MinIO (PROC_<contract_folder_id>/…) and syncs scraped document metadata
|
// @Description Lists database tenders whose contract_folder_id has documents in MinIO. By default returns quickly using the procedure scan only. Use sync=1 to re-list each procedure in MinIO and update Mongo scraped document metadata (slow).
|
||||||
// @Tags Admin-Tenders
|
// @Tags Admin-Tenders
|
||||||
// @Produce json
|
// @Produce json
|
||||||
// @Param limit query int false "Number of items per page (default: 10, max: 100)"
|
// @Param limit query int false "Number of items per page (default: 10, max: 100)"
|
||||||
// @Param offset query int false "Number of items to skip (default: 0)"
|
// @Param offset query int false "Number of items to skip (default: 0)"
|
||||||
|
// @Param sync query bool false "When true or 1, list each tender's documents in MinIO and persist scraped metadata to Mongo (slow)"
|
||||||
// @Success 200 {object} response.APIResponse{data=TendersWithScrapedDocumentsResponse}
|
// @Success 200 {object} response.APIResponse{data=TendersWithScrapedDocumentsResponse}
|
||||||
// @Failure 400 {object} response.APIResponse
|
// @Failure 400 {object} response.APIResponse
|
||||||
// @Failure 500 {object} response.APIResponse
|
// @Failure 500 {object} response.APIResponse
|
||||||
@@ -677,7 +679,8 @@ func (h *TenderHandler) ListTendersWithScrapedDocuments(c echo.Context) error {
|
|||||||
return response.PaginationBadRequest(c, err)
|
return response.PaginationBadRequest(c, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
result, err := h.service.ListTendersWithScrapedDocuments(c.Request().Context(), pagination)
|
syncMongo := c.QueryParam("sync") == "1" || strings.EqualFold(c.QueryParam("sync"), "true")
|
||||||
|
result, err := h.service.ListTendersWithScrapedDocuments(c.Request().Context(), pagination, syncMongo)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return response.InternalServerError(c, "Failed to list tenders with scraped documents")
|
return response.InternalServerError(c, "Failed to list tenders with scraped documents")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -21,6 +21,8 @@ type TenderRepository interface {
|
|||||||
GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Tender, error)
|
GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Tender, error)
|
||||||
GetByNoticePublicationID(ctx context.Context, noticePublicationID string) (*Tender, error)
|
GetByNoticePublicationID(ctx context.Context, noticePublicationID string) (*Tender, error)
|
||||||
GetByContractFolderID(ctx context.Context, contractFolderID string) (*Tender, error)
|
GetByContractFolderID(ctx context.Context, contractFolderID string) (*Tender, error)
|
||||||
|
// GetLatestByContractFolderIDs returns the most recently updated tender per distinct contract_folder_id.
|
||||||
|
GetLatestByContractFolderIDs(ctx context.Context, contractFolderIDs []string) (map[string]*Tender, error)
|
||||||
GetByProcurementProjectID(ctx context.Context, procurementProjectID string) (*Tender, error)
|
GetByProcurementProjectID(ctx context.Context, procurementProjectID string) (*Tender, error)
|
||||||
FindTendersWithContentXML(ctx context.Context, limit, skip int) ([]Tender, error)
|
FindTendersWithContentXML(ctx context.Context, limit, skip int) ([]Tender, error)
|
||||||
GetTenderCountByCountry(ctx context.Context) (map[string]int64, error)
|
GetTenderCountByCountry(ctx context.Context) (map[string]int64, error)
|
||||||
@@ -289,6 +291,76 @@ func (r *tenderRepository) GetByContractFolderID(ctx context.Context, contractFo
|
|||||||
return &result.Items[0], nil
|
return &result.Items[0], nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// GetLatestByContractFolderIDs returns one tender per distinct contract_folder_id (latest updated_at).
|
||||||
|
func (r *tenderRepository) GetLatestByContractFolderIDs(ctx context.Context, contractFolderIDs []string) (map[string]*Tender, error) {
|
||||||
|
out := make(map[string]*Tender)
|
||||||
|
if len(contractFolderIDs) == 0 {
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
uniq := uniqueNonEmptyTrimmedStrings(contractFolderIDs)
|
||||||
|
if len(uniq) == 0 {
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Allow a few rows per folder in case duplicate folder rows exist despite partial uniques.
|
||||||
|
findLimit := len(uniq) * 3
|
||||||
|
if findLimit < len(uniq)+1 {
|
||||||
|
findLimit = len(uniq) + 1
|
||||||
|
}
|
||||||
|
if findLimit > 100000 {
|
||||||
|
findLimit = 100000
|
||||||
|
}
|
||||||
|
|
||||||
|
result, err := r.ormRepo.FindAll(ctx,
|
||||||
|
bson.M{"contract_folder_id": bson.M{"$in": uniq}},
|
||||||
|
orm.Pagination{
|
||||||
|
Limit: findLimit,
|
||||||
|
Skip: 0,
|
||||||
|
SortField: "updated_at",
|
||||||
|
SortOrder: -1,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
r.logger.Error("Failed to load tenders by contract folder ids", map[string]interface{}{
|
||||||
|
"folder_count": len(uniq),
|
||||||
|
"error": err.Error(),
|
||||||
|
})
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
for i := range result.Items {
|
||||||
|
t := &result.Items[i]
|
||||||
|
cid := strings.TrimSpace(t.ContractFolderID)
|
||||||
|
if cid == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
prev, ok := out[cid]
|
||||||
|
if !ok || t.UpdatedAt > prev.UpdatedAt {
|
||||||
|
out[cid] = t
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func uniqueNonEmptyTrimmedStrings(ids []string) []string {
|
||||||
|
seen := make(map[string]struct{}, len(ids))
|
||||||
|
out := make([]string, 0, len(ids))
|
||||||
|
for _, id := range ids {
|
||||||
|
id = strings.TrimSpace(id)
|
||||||
|
if id == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, dup := seen[id]; dup {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
seen[id] = struct{}{}
|
||||||
|
out = append(out, id)
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
// GetByProcurementProjectID finds a tender by UBL ProcurementProject ID (shared across per-lot notices).
|
// GetByProcurementProjectID finds a tender by UBL ProcurementProject ID (shared across per-lot notices).
|
||||||
func (r *tenderRepository) GetByProcurementProjectID(ctx context.Context, procurementProjectID string) (*Tender, error) {
|
func (r *tenderRepository) GetByProcurementProjectID(ctx context.Context, procurementProjectID string) (*Tender, error) {
|
||||||
filter := bson.M{"procurement_project_id": procurementProjectID}
|
filter := bson.M{"procurement_project_id": procurementProjectID}
|
||||||
|
|||||||
@@ -59,7 +59,9 @@ type Service interface {
|
|||||||
TriggerAISummarize(ctx context.Context, id string) (*AISummarizeResponse, error)
|
TriggerAISummarize(ctx context.Context, id string) (*AISummarizeResponse, error)
|
||||||
TriggerAITranslate(ctx context.Context, id, language string) (*AITranslateResponse, error)
|
TriggerAITranslate(ctx context.Context, id, language string) (*AITranslateResponse, error)
|
||||||
GetAllAISummaries(ctx context.Context) ([]ai_summarizer.TenderSummaryItem, error)
|
GetAllAISummaries(ctx context.Context) ([]ai_summarizer.TenderSummaryItem, error)
|
||||||
ListTendersWithScrapedDocuments(ctx context.Context, pagination *response.Pagination) (*TendersWithScrapedDocumentsResponse, error)
|
// 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)
|
||||||
|
|
||||||
// Maintenance operations
|
// Maintenance operations
|
||||||
UpdateExpiredTenders(ctx context.Context) error
|
UpdateExpiredTenders(ctx context.Context) error
|
||||||
@@ -1161,8 +1163,9 @@ func (s *tenderService) GetAllAISummaries(ctx context.Context) ([]ai_summarizer.
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ListTendersWithScrapedDocuments lists database tenders whose contract_folder_id has
|
// ListTendersWithScrapedDocuments lists database tenders whose contract_folder_id has
|
||||||
// documents in MinIO, and syncs scraped document metadata onto each tender.
|
// documents in MinIO. By default uses counts from the initial MinIO procedure scan only (fast).
|
||||||
func (s *tenderService) ListTendersWithScrapedDocuments(ctx context.Context, pagination *response.Pagination) (*TendersWithScrapedDocumentsResponse, error) {
|
// 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 {
|
if s.aiSummarizerStorage == nil {
|
||||||
return nil, fmt.Errorf("document storage is not configured")
|
return nil, fmt.Errorf("document storage is not configured")
|
||||||
}
|
}
|
||||||
@@ -1178,6 +1181,7 @@ func (s *tenderService) ListTendersWithScrapedDocuments(ctx context.Context, pag
|
|||||||
s.logger.Info("Listing tenders with scraped documents from MinIO", map[string]interface{}{
|
s.logger.Info("Listing tenders with scraped documents from MinIO", map[string]interface{}{
|
||||||
"limit": pagination.Limit,
|
"limit": pagination.Limit,
|
||||||
"offset": pagination.Offset,
|
"offset": pagination.Offset,
|
||||||
|
"sync_mongo_metadata": syncMongoMetadata,
|
||||||
})
|
})
|
||||||
|
|
||||||
procedures, err := storage.ListProceduresWithDocuments(ctx)
|
procedures, err := storage.ListProceduresWithDocuments(ctx)
|
||||||
@@ -1191,13 +1195,31 @@ func (s *tenderService) ListTendersWithScrapedDocuments(ctx context.Context, pag
|
|||||||
return nil, fmt.Errorf("failed to list tenders with scraped documents: %w", err)
|
return nil, fmt.Errorf("failed to list tenders with scraped documents: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
folderIDs := make([]string, 0, len(procedures))
|
||||||
|
for _, proc := range procedures {
|
||||||
|
folderIDs = append(folderIDs, proc.ContractFolderID)
|
||||||
|
}
|
||||||
|
|
||||||
|
tenderByFolder, err := s.repository.GetLatestByContractFolderIDs(ctx, folderIDs)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to load tenders by contract_folder_id: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
matched := make([]TenderScrapedDocumentsItem, 0, len(procedures))
|
matched := make([]TenderScrapedDocumentsItem, 0, len(procedures))
|
||||||
for _, proc := range procedures {
|
for _, proc := range procedures {
|
||||||
tender, err := s.repository.GetByContractFolderID(ctx, proc.ContractFolderID)
|
if proc.DocumentCount <= 0 {
|
||||||
if err != nil || tender == nil {
|
continue
|
||||||
|
}
|
||||||
|
tender, ok := tenderByFolder[strings.TrimSpace(proc.ContractFolderID)]
|
||||||
|
if tender == nil || !ok {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
item := tender.ToTenderScrapedDocumentsItem()
|
||||||
|
item.DocumentCount = proc.DocumentCount
|
||||||
|
item.DocumentsScrapedAt = proc.LatestModified
|
||||||
|
|
||||||
|
if syncMongoMetadata {
|
||||||
docs, listErr := s.listDocumentsForTender(ctx, tender)
|
docs, listErr := s.listDocumentsForTender(ctx, tender)
|
||||||
if listErr != nil {
|
if listErr != nil {
|
||||||
if errors.Is(listErr, ai_summarizer.ErrMinIOUnavailable) {
|
if errors.Is(listErr, ai_summarizer.ErrMinIOUnavailable) {
|
||||||
@@ -1213,11 +1235,13 @@ func (s *tenderService) ListTendersWithScrapedDocuments(ctx context.Context, pag
|
|||||||
if len(docs) == 0 {
|
if len(docs) == 0 {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
s.persistScrapedDocuments(ctx, tender, docs)
|
s.persistScrapedDocuments(ctx, tender, docs)
|
||||||
|
|
||||||
item := tender.ToTenderScrapedDocumentsItem()
|
|
||||||
item.DocumentCount = len(docs)
|
item.DocumentCount = len(docs)
|
||||||
|
if tender.ProcessingMetadata.DocumentsScrapedAt != 0 {
|
||||||
|
item.DocumentsScrapedAt = tender.ProcessingMetadata.DocumentsScrapedAt
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
matched = append(matched, item)
|
matched = append(matched, item)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user