diff --git a/cmd/web/router/routes.go b/cmd/web/router/routes.go index 5845b91..296955b 100644 --- a/cmd/web/router/routes.go +++ b/cmd/web/router/routes.go @@ -233,6 +233,7 @@ func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler aiPipelineGP.Use(userHandler.AuthMiddleware()) aiPipelineGP.POST("/scrape/documents", aiPipelineHandler.ScrapeDocuments) aiPipelineGP.POST("/scrape/documents/batch", aiPipelineHandler.ScrapeDocumentsBatch) + aiPipelineGP.GET("/scrape/portals", aiPipelineHandler.GetScrapePortals) aiPipelineGP.POST("/summarize/batch", aiPipelineHandler.SummarizeBatch) aiPipelineGP.POST("/translate/batch", aiPipelineHandler.TranslateBatch) aiPipelineGP.POST("/sync", aiPipelineHandler.Sync) diff --git a/internal/ai_pipeline/handler.go b/internal/ai_pipeline/handler.go index 5057e0a..7b55e08 100644 --- a/internal/ai_pipeline/handler.go +++ b/internal/ai_pipeline/handler.go @@ -78,6 +78,26 @@ func (h *Handler) ScrapeDocumentsBatch(c echo.Context) error { return response.Accepted(c, result, "Scrape batch accepted") } +// GetScrapePortals lists document scraping portals supported by the Opplens AI service. +// @Summary List document scraping portals +// @Description Retrieve the list of document scraping portals supported by the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Produce json +// @Success 200 {object} response.APIResponse{data=ai_summarizer.ScrapePortalsResponse} "Scrape portals retrieved successfully" +// @Failure 401 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/scrape/portals [get] +func (h *Handler) GetScrapePortals(c echo.Context) error { + result, err := h.service.GetScrapePortals(c.Request().Context()) + if err != nil { + return mapPipelineHTTPError(c, err, "Get scrape portals") + } + + return response.Success(c, result, "Scrape portals retrieved successfully") +} + // SummarizeBatch enqueues background summarization for many tenders. // @Summary Batch summarize tenders // @Description Enqueue background AI summarization for multiple tenders via the Opplens AI service diff --git a/internal/ai_pipeline/service.go b/internal/ai_pipeline/service.go index 4fafaad..0cfd5ae 100644 --- a/internal/ai_pipeline/service.go +++ b/internal/ai_pipeline/service.go @@ -12,6 +12,7 @@ import ( type Client interface { ScrapeDocuments(ctx context.Context, reqBody ai_summarizer.ScrapeDocumentsRequest) (*ai_summarizer.ScrapeDocumentsResponse, error) ScrapeDocumentsBatch(ctx context.Context, tenders []ai_summarizer.TenderRef) (*ai_summarizer.BatchAcceptedResponse, error) + GetScrapePortals(ctx context.Context) (*ai_summarizer.ScrapePortalsResponse, error) TriggerSummarizeBatch(ctx context.Context, tenders []ai_summarizer.TenderRef) (*ai_summarizer.BatchAcceptedResponse, error) TriggerTranslateBatch(ctx context.Context, tenders []ai_summarizer.TenderRef, languages []string) (*ai_summarizer.BatchAcceptedResponse, error) PipelineSync(ctx context.Context) (*ai_summarizer.PipelineSyncResponse, error) @@ -31,6 +32,7 @@ type Client interface { type Service interface { ScrapeDocuments(ctx context.Context, form ScrapeDocumentsForm) (*ai_summarizer.ScrapeDocumentsResponse, error) ScrapeDocumentsBatch(ctx context.Context, form TenderBatchForm) (*ai_summarizer.BatchAcceptedResponse, error) + GetScrapePortals(ctx context.Context) (*ai_summarizer.ScrapePortalsResponse, error) SummarizeBatch(ctx context.Context, form TenderBatchForm) (*ai_summarizer.BatchAcceptedResponse, error) TranslateBatch(ctx context.Context, form TranslateBatchForm) (*ai_summarizer.BatchAcceptedResponse, error) Sync(ctx context.Context) (*ai_summarizer.PipelineSyncResponse, error) @@ -128,6 +130,24 @@ func (s *service) ScrapeDocumentsBatch(ctx context.Context, form TenderBatchForm return resp, nil } +func (s *service) GetScrapePortals(ctx context.Context) (*ai_summarizer.ScrapePortalsResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + + s.logger.Info("Admin get scrape portals requested", map[string]interface{}{}) + + resp, err := s.client.GetScrapePortals(ctx) + if err != nil { + s.logger.Error("Admin get scrape portals failed", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("get scrape portals failed: %w", err) + } + + return resp, nil +} + func (s *service) SummarizeBatch(ctx context.Context, form TenderBatchForm) (*ai_summarizer.BatchAcceptedResponse, error) { if err := s.requireClient(); err != nil { return nil, err diff --git a/internal/document_scraper/errors.go b/internal/document_scraper/errors.go new file mode 100644 index 0000000..58e3a7a --- /dev/null +++ b/internal/document_scraper/errors.go @@ -0,0 +1,8 @@ +package document_scraper + +import "errors" + +var ( + // ErrInvalidDateRange is returned when created_at_from is after created_at_to. + ErrInvalidDateRange = errors.New("invalid date range: created_at_from must be before or equal to created_at_to") +) diff --git a/internal/document_scraper/form.go b/internal/document_scraper/form.go index 8c0aad6..a18d9cc 100644 --- a/internal/document_scraper/form.go +++ b/internal/document_scraper/form.go @@ -2,8 +2,24 @@ package document_scraper // DocumentScraperListRequest represents the request to list tenders pending document scraping type DocumentScraperListRequest struct { - Limit int `query:"limit" valid:"optional,range(1|100)" default:"20"` - Offset int `query:"offset" valid:"optional,range(0|1000000)" default:"0"` + Limit int `query:"limit" valid:"optional,range(1|100)"` + Offset int `query:"offset" valid:"optional,range(0|1000000)"` + CreatedAtFrom *int64 `query:"created_at_from" valid:"optional"` + CreatedAtTo *int64 `query:"created_at_to" valid:"optional"` + IncludeExpired bool `query:"include_expired" valid:"optional"` +} + +// Normalize applies default pagination bounds. +func (r *DocumentScraperListRequest) Normalize() { + if r.Limit <= 0 { + r.Limit = 20 + } + if r.Limit > 100 { + r.Limit = 100 + } + if r.Offset < 0 { + r.Offset = 0 + } } // DocumentScraperGetRequest represents the request to get a specific tender by notice publication ID diff --git a/internal/document_scraper/handler.go b/internal/document_scraper/handler.go index c5fa551..1dfa30f 100644 --- a/internal/document_scraper/handler.go +++ b/internal/document_scraper/handler.go @@ -1,7 +1,7 @@ package document_scraper import ( - "strconv" + "errors" "tm/pkg/logger" "tm/pkg/response" @@ -24,36 +24,31 @@ func NewHandler(service Service, logger logger.Logger) *Handler { // ListPendingTenders retrieves tenders that need document scraping // @Summary List pending tenders for document scraping -// @Description Retrieve tenders that have not yet been scraped for documents. Each item includes contract_folder_id, notice_publication_id, document_url, title, and description. +// @Description Retrieve tenders that have not yet been scraped for documents. Each item includes contract_folder_id, notice_publication_id, document_url, title, and description. Optionally filter by created_at range (Unix timestamps in seconds). By default only tenders with active deadlines are returned; set include_expired=true to include expired tenders. // @Tags Document-Scraper // @Produce json // @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)" +// @Param created_at_from query int64 false "Filter by created_at lower bound (Unix timestamp, seconds)" +// @Param created_at_to query int64 false "Filter by created_at upper bound (Unix timestamp, seconds)" +// @Param include_expired query bool false "When true, include tenders whose deadline has passed (default: false)" // @Success 200 {object} response.APIResponse{data=DocumentScraperListResponse} "List of pending tenders" +// @Failure 400 {object} response.APIResponse "Invalid query parameters" // @Failure 401 {object} response.APIResponse "Invalid or missing API key" // @Failure 500 {object} response.APIResponse "Internal server error" // @Security ApiKeyAuth // @Router /api/v1/scraper/tenders [get] func (h *Handler) ListPendingTenders(c echo.Context) error { - // Parse pagination parameters - limit, err := strconv.Atoi(c.QueryParam("limit")) - if err != nil || limit <= 0 { - limit = 20 - } - if limit > 100 { - limit = 100 - } - - offset, err := strconv.Atoi(c.QueryParam("offset")) - if err != nil || offset < 0 { - offset = 0 - } - - tenders, meta, err := h.service.ListPendingTenders(c.Request().Context(), limit, offset) + form, err := response.Parse[DocumentScraperListRequest](c) if err != nil { - h.logger.Error("Failed to list pending tenders", map[string]interface{}{ - "error": err.Error(), - }) + return response.ValidationError(c, "Invalid request data", err.Error()) + } + + tenders, meta, err := h.service.ListPendingTenders(c.Request().Context(), *form) + if err != nil { + if errors.Is(err, ErrInvalidDateRange) { + return response.BadRequest(c, "Invalid date range", err.Error()) + } return response.InternalServerError(c, "Failed to list pending tenders") } @@ -84,10 +79,6 @@ func (h *Handler) GetTenderByNoticeID(c echo.Context) error { if err.Error() == "tender not found" { return response.NotFound(c, "Tender not found") } - h.logger.Error("Failed to get tender by notice ID", map[string]interface{}{ - "notice_id": noticeID, - "error": err.Error(), - }) return response.InternalServerError(c, "Failed to get tender") } diff --git a/internal/document_scraper/service.go b/internal/document_scraper/service.go index 3a819d6..94f5fc3 100644 --- a/internal/document_scraper/service.go +++ b/internal/document_scraper/service.go @@ -23,7 +23,7 @@ func isDocumentScraperCountry(countryCode string) bool { // Service defines business logic for the document scraper API type Service interface { // ListPendingTenders retrieves tenders that need document scraping - ListPendingTenders(ctx context.Context, limit, offset int) (*DocumentScraperListResponse, *response.Meta, error) + ListPendingTenders(ctx context.Context, req DocumentScraperListRequest) (*DocumentScraperListResponse, *response.Meta, error) // GetTenderByNoticeID retrieves a specific tender by its notice publication ID GetTenderByNoticeID(ctx context.Context, noticeID string) (*DocumentScraperTenderResponse, error) } @@ -42,16 +42,36 @@ func NewService(tenderRepo tender.TenderRepository, logger logger.Logger) Servic } // ListPendingTenders retrieves tenders that have not yet been scraped for documents. -// Only returns tenders from Sweden, Finland, and Norway with active deadlines (deadline not yet reached). -func (s *service) ListPendingTenders(ctx context.Context, limit, offset int) (*DocumentScraperListResponse, *response.Meta, error) { +// Only returns tenders from Sweden, Finland, and Norway. By default only tenders with active +// deadlines are included; set IncludeExpired to include tenders whose deadline has passed. +func (s *service) ListPendingTenders(ctx context.Context, req DocumentScraperListRequest) (*DocumentScraperListResponse, *response.Meta, error) { + req.Normalize() + + if req.CreatedAtFrom != nil && req.CreatedAtTo != nil && *req.CreatedAtFrom > *req.CreatedAtTo { + return nil, nil, ErrInvalidDateRange + } + s.logger.Info("Listing pending tenders for document scraper", map[string]interface{}{ - "limit": limit, - "offset": offset, + "limit": req.Limit, + "offset": req.Offset, + "created_at_from": req.CreatedAtFrom, + "created_at_to": req.CreatedAtTo, + "include_expired": req.IncludeExpired, }) - now := time.Now().Unix() + filter := tender.PendingDocumentScrapeFilter{ + CountryCodes: documentScraperCountryCodes, + CreatedAtFrom: req.CreatedAtFrom, + CreatedAtTo: req.CreatedAtTo, + Limit: req.Limit, + Skip: req.Offset, + } + if !req.IncludeExpired { + now := time.Now().Unix() + filter.MinDeadline = &now + } - tenders, hasMore, err := s.tenderRepo.GetPendingDocumentScrapeTenders(ctx, documentScraperCountryCodes, now, limit, offset) + tenders, hasMore, err := s.tenderRepo.GetPendingDocumentScrapeTenders(ctx, filter) if err != nil { s.logger.Error("Failed to list pending tenders", map[string]interface{}{ "error": err.Error(), @@ -69,12 +89,12 @@ func (s *service) ListPendingTenders(ctx context.Context, limit, offset int) (*D meta := &response.Meta{ Total: len(tenders), - Limit: limit, - Offset: offset, + Limit: req.Limit, + Offset: req.Offset, HasMore: hasMore, } if meta.Limit > 0 { - meta.Page = (offset / meta.Limit) + 1 + meta.Page = (req.Offset / meta.Limit) + 1 } s.logger.Info("Pending tenders retrieved successfully", map[string]interface{}{ diff --git a/internal/tender/repository.go b/internal/tender/repository.go index 9ff17eb..3339166 100644 --- a/internal/tender/repository.go +++ b/internal/tender/repository.go @@ -31,7 +31,7 @@ type TenderRepository interface { GetTenderCountByClassification(ctx context.Context) (map[string]int64, error) Search(ctx context.Context, form *SearchForm, pagination *response.Pagination) (*SearchPageResult, error) GetExpiredTenders(ctx context.Context, beforeDate int64) ([]Tender, error) - GetPendingDocumentScrapeTenders(ctx context.Context, countryCodes []string, minDeadline int64, limit, skip int) ([]Tender, bool, error) + GetPendingDocumentScrapeTenders(ctx context.Context, filter PendingDocumentScrapeFilter) ([]Tender, bool, error) GetUnSummarizedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error) GetUnTranslatedTenders(ctx context.Context, language string, limit, skip int) ([]Tender, int64, error) Update(ctx context.Context, tender *Tender) error @@ -555,37 +555,64 @@ func (r *tenderRepository) GetExpiredTenders(ctx context.Context, beforeDate int return result.Items, nil } -// GetPendingDocumentScrapeTenders returns unscraped tenders for the given countries with an active deadline. -func (r *tenderRepository) GetPendingDocumentScrapeTenders(ctx context.Context, countryCodes []string, minDeadline int64, limit, skip int) ([]Tender, bool, error) { - if len(countryCodes) == 0 { +// PendingDocumentScrapeFilter defines query criteria for pending document scrape tenders. +type PendingDocumentScrapeFilter struct { + CountryCodes []string + MinDeadline *int64 // when set, tender_deadline must be greater than this value + CreatedAtFrom *int64 + CreatedAtTo *int64 + Limit int + Skip int +} + +// GetPendingDocumentScrapeTenders returns unscraped tenders for the given countries. +// When MinDeadline is set, only tenders with tender_deadline greater than that value are returned. +func (r *tenderRepository) GetPendingDocumentScrapeTenders(ctx context.Context, filter PendingDocumentScrapeFilter) ([]Tender, bool, error) { + if len(filter.CountryCodes) == 0 { return nil, false, nil } - filter := bson.M{ - "country_code": bson.M{"$in": countryCodes}, - "tender_deadline": bson.M{ - "$gt": minDeadline, - }, + query := bson.M{ + "country_code": bson.M{"$in": filter.CountryCodes}, "processing_metadata.documents_scraped": bson.M{"$ne": true}, } + if filter.MinDeadline != nil { + query["tender_deadline"] = bson.M{ + "$gt": *filter.MinDeadline, + } + } + + if filter.CreatedAtFrom != nil || filter.CreatedAtTo != nil { + createdFilter := bson.M{} + if filter.CreatedAtFrom != nil { + createdFilter["$gte"] = *filter.CreatedAtFrom + } + if filter.CreatedAtTo != nil { + createdFilter["$lte"] = *filter.CreatedAtTo + } + query["created_at"] = createdFilter + } + pagination := orm.Pagination{ - Limit: limit, - Skip: skip, + Limit: filter.Limit, + Skip: filter.Skip, SortField: "created_at", SortOrder: 1, SkipCount: true, Projection: documentScraperListProjection(), } - result, err := r.ormRepo.FindAll(ctx, filter, pagination) + result, err := r.ormRepo.FindAll(ctx, query, pagination) if err != nil { r.logger.Error("Failed to get pending document scrape tenders", map[string]interface{}{ - "country_codes": countryCodes, - "min_deadline": minDeadline, - "limit": limit, - "skip": skip, - "error": err.Error(), + "country_codes": filter.CountryCodes, + "min_deadline": filter.MinDeadline, + "created_at_from": filter.CreatedAtFrom, + "created_at_to": filter.CreatedAtTo, + "limit": filter.Limit, + "skip": filter.Skip, + "error": err.Error(), }) return nil, false, err } diff --git a/pkg/ai_summarizer/client.go b/pkg/ai_summarizer/client.go index 895d0a8..f3f4fb6 100644 --- a/pkg/ai_summarizer/client.go +++ b/pkg/ai_summarizer/client.go @@ -195,6 +195,29 @@ func (c *Client) ScrapeDocumentsBatch(ctx context.Context, tenders []TenderRef) }) } +// GetScrapePortals calls GET /scrape/portals. +func (c *Client) GetScrapePortals(ctx context.Context) (*ScrapePortalsResponse, error) { + url := c.config.APIBaseURL + "/scrape/portals" + httpResp, bodyBytes, err := c.doRawGet(ctx, url) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + if httpResp.StatusCode >= 400 { + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) + } + + var result ScrapePortalsResponse + if len(bodyBytes) > 0 { + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode scrape portals response: %w", err) + } + } + + return &result, nil +} + // PipelineSync calls POST /pipeline/sync to pull the tender list from the backend. func (c *Client) PipelineSync(ctx context.Context) (*PipelineSyncResponse, error) { url := c.config.APIBaseURL + "/pipeline/sync" diff --git a/pkg/ai_summarizer/entities.go b/pkg/ai_summarizer/entities.go index 393e2a2..3379bac 100644 --- a/pkg/ai_summarizer/entities.go +++ b/pkg/ai_summarizer/entities.go @@ -173,6 +173,19 @@ type ScrapeDocumentsBatchRequest struct { Tenders []TenderRef `json:"tenders"` } +// ScrapePortal describes one document scraping portal supported by the AI service. +type ScrapePortal struct { + ID string `json:"id"` + Name string `json:"name"` + CountryCode string `json:"country_code,omitempty"` + Countries []string `json:"countries,omitempty"` +} + +// ScrapePortalsResponse is returned by GET /scrape/portals. +type ScrapePortalsResponse struct { + Portals []ScrapePortal `json:"portals"` +} + // SummarizeBatchRequest is the payload for POST /ai/summarize/batch. type SummarizeBatchRequest struct { Tenders []TenderRef `json:"tenders"`