diff --git a/cmd/web/main.go b/cmd/web/main.go index 8b6d241..ce99dc4 100644 --- a/cmd/web/main.go +++ b/cmd/web/main.go @@ -59,6 +59,9 @@ package main // @tag.name Admin-CMS // @tag.description Administrative CMS management operations for web panel including CRUD operations, search, statistics, and comprehensive filtering with pagination +// @tag.name Admin-AI-Pipeline +// @tag.description Administrative Opplens AI pipeline operations including scrape, summarize, translate, sync, and background pipeline runs + // @tag.name Authorization // @tag.description Customer authentication and authorization operations for mobile application including login, logout, token refresh, and profile access @@ -98,6 +101,7 @@ import ( "net/http" "time" "tm/internal/assets" + "tm/internal/ai_pipeline" "tm/internal/cms" "tm/internal/company" "tm/internal/company_category" @@ -168,9 +172,11 @@ func main() { var aiSummarizerClient tender.AISummarizerClient var aiSummarizerStorage tender.AISummarizerStorage var aiRecommendationClient company.AIRecommendationClient + var aiPipelineClient ai_pipeline.Client if c := bootstrap.InitAISummarizerClient(conf.AISummarizer, mongoManager, logger); c != nil { aiSummarizerClient = c + aiPipelineClient = c aiRecommendationClient = c } if s := bootstrap.InitAISummarizerStorage(conf.AISummarizer, logger); s != nil { @@ -226,8 +232,9 @@ func main() { documentScraperService := document_scraper.NewService(tenderRepository, logger) dashboardRepository := dashboard.NewRepository(mongoManager, logger) dashboardService := dashboard.NewService(dashboardRepository, logger) + aiPipelineService := ai_pipeline.NewService(aiPipelineClient, logger) 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"}, + "services": []string{"customer", "user", "company", "category", "tender", "feedback", "tender_approval", "inquiry", "flag", "notification", "contact", "cms", "scraper", "kanban", "document_scraper", "dashboard", "ai_pipeline"}, }) // Initialize handlers with services @@ -247,8 +254,9 @@ func main() { fileStoreHandler := filestore.NewHandler(fileStoreService, logger) documentScraperHandler := document_scraper.NewHandler(documentScraperService, logger) dashboardHandler := dashboard.NewHandler(dashboardService) + aiPipelineHandler := ai_pipeline.NewHandler(aiPipelineService) logger.Info("Handlers initialized successfully", map[string]interface{}{ - "handlers": []string{"customer", "user", "company", "category", "tender", "feedback", "tender_approval", "inquiry", "flag", "notification", "contact", "cms", "scraper", "kanban", "filestore", "document_scraper", "dashboard"}, + "handlers": []string{"customer", "user", "company", "category", "tender", "feedback", "tender_approval", "inquiry", "flag", "notification", "contact", "cms", "scraper", "kanban", "filestore", "document_scraper", "dashboard", "ai_pipeline"}, }) // Initialize HTTP server @@ -257,7 +265,7 @@ func main() { router.SetupCSPSecurity(e) // Register routes - router.RegisterAdminRoutes(e, userHandler, companyHandler, customerHandler, tenderHandler, feedbackHandler, tenderApprovalHandler, inquiryHandler, categoryHandler, flagHandler, notificationHandler, contactHandler, cmsHandler, kanbanHandler, fileStoreHandler, dashboardHandler, xssPolicy) + router.RegisterAdminRoutes(e, userHandler, companyHandler, customerHandler, tenderHandler, feedbackHandler, tenderApprovalHandler, inquiryHandler, categoryHandler, flagHandler, notificationHandler, contactHandler, cmsHandler, kanbanHandler, fileStoreHandler, dashboardHandler, aiPipelineHandler, xssPolicy) router.RegisterPublicRoutes(e, customerHandler, tenderHandler, feedbackHandler, tenderApprovalHandler, companyHandler, inquiryHandler, categoryHandler, flagHandler, notificationHandler, contactHandler, cmsHandler, kanbanHandler, fileStoreHandler, xssPolicy) router.RegisterDocumentScraperRoutes(e, documentScraperHandler, conf.DocumentScraper.APIKey) diff --git a/cmd/web/router/routes.go b/cmd/web/router/routes.go index 9d4aaf7..5845b91 100644 --- a/cmd/web/router/routes.go +++ b/cmd/web/router/routes.go @@ -2,6 +2,7 @@ package router import ( "tm/internal/assets" + "tm/internal/ai_pipeline" "tm/internal/cms" "tm/internal/company" "tm/internal/company_category" @@ -34,7 +35,7 @@ func SetupCSPSecurity(e *echo.Echo) { e.POST("/api/v1/csp-report", security.CSPReportHandler) } -func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler *company.Handler, customerHandler *customer.Handler, tenderHandler *tender.TenderHandler, feedbackHandler *feedback.Handler, tenderApprovalHandler *tender_approval.Handler, inquiryHandler *inquiry.Handler, categoryHandler *company_category.Handler, flagHandler *assets.Handler, notificationHandler *notification.NotificationHandler, contactHandler *contact.Handler, cmsHandler *cms.Handler, kanbanHandler *kanban.Handler, fileStoreHandler *filestore.Handler, dashboardHandler *dashboard.Handler, xssPolicy *security.XSSPolicy) { +func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler *company.Handler, customerHandler *customer.Handler, tenderHandler *tender.TenderHandler, feedbackHandler *feedback.Handler, tenderApprovalHandler *tender_approval.Handler, inquiryHandler *inquiry.Handler, categoryHandler *company_category.Handler, flagHandler *assets.Handler, notificationHandler *notification.NotificationHandler, contactHandler *contact.Handler, cmsHandler *cms.Handler, kanbanHandler *kanban.Handler, fileStoreHandler *filestore.Handler, dashboardHandler *dashboard.Handler, aiPipelineHandler *ai_pipeline.Handler, xssPolicy *security.XSSPolicy) { adminV1 := e.Group("/admin/v1") // Admin Users Routes @@ -225,6 +226,27 @@ func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler dashboardGP.GET("/recent", dashboardHandler.Recent) dashboardGP.GET("/statistics", dashboardHandler.Statistics) } + + // AI Pipeline Routes (proxy to Opplens AI service) + aiPipelineGP := adminV1.Group("/ai-pipeline") + { + aiPipelineGP.Use(userHandler.AuthMiddleware()) + aiPipelineGP.POST("/scrape/documents", aiPipelineHandler.ScrapeDocuments) + aiPipelineGP.POST("/scrape/documents/batch", aiPipelineHandler.ScrapeDocumentsBatch) + aiPipelineGP.POST("/summarize/batch", aiPipelineHandler.SummarizeBatch) + aiPipelineGP.POST("/translate/batch", aiPipelineHandler.TranslateBatch) + aiPipelineGP.POST("/sync", aiPipelineHandler.Sync) + aiPipelineGP.POST("/run", aiPipelineHandler.Run) + aiPipelineGP.POST("/run/batch", aiPipelineHandler.RunBatch) + aiPipelineGP.POST("/analyze", aiPipelineHandler.Analyze) + aiPipelineGP.POST("/daily-run", aiPipelineHandler.DailyRun) + aiPipelineGP.POST("/auto", aiPipelineHandler.Auto) + aiPipelineGP.GET("/last-run", aiPipelineHandler.GetLastRun) + aiPipelineGP.GET("/last-auto-run", aiPipelineHandler.GetLastAutoRun) + aiPipelineGP.GET("/report", aiPipelineHandler.GetReport) + aiPipelineGP.GET("/stats", aiPipelineHandler.GetStats) + aiPipelineGP.GET("/minio-stats", aiPipelineHandler.GetMinioStats) + } } func RegisterPublicRoutes(e *echo.Echo, customerHandler *customer.Handler, tenderHandler *tender.TenderHandler, feedbackHandler *feedback.Handler, tenderApprovalHandler *tender_approval.Handler, companyHandler *company.Handler, inquiryHandler *inquiry.Handler, categoryHandler *company_category.Handler, flagHandler *assets.Handler, notificationHandler *notification.NotificationHandler, contactHandler *contact.Handler, cmsHandler *cms.Handler, kanbanHandler *kanban.Handler, fileStoreHandler *filestore.Handler, xssPolicy *security.XSSPolicy) { diff --git a/cmd/worker/workers/summarizer.go b/cmd/worker/workers/summarizer.go index dcc352f..4a4de77 100644 --- a/cmd/worker/workers/summarizer.go +++ b/cmd/worker/workers/summarizer.go @@ -47,10 +47,13 @@ func (w *DocumentSummarizationWorker) Run() { return } - if _, err := w.AIClient.TriggerPipelineSummarize(context.Background()); err != nil { - w.Logger.Warn("Failed to trigger AI pipeline summarize (continuing with per-tender sync)", map[string]interface{}{ - "error": err.Error(), - }) + 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 @@ -92,6 +95,38 @@ func (w *DocumentSummarizationWorker) Run() { 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{}{ diff --git a/cmd/worker/workers/translation.go b/cmd/worker/workers/translation.go index 9f94f63..158baf6 100644 --- a/cmd/worker/workers/translation.go +++ b/cmd/worker/workers/translation.go @@ -10,7 +10,7 @@ import ( ) // TranslationWorker syncs translation completion from the AI pipeline MinIO storage. -// Batch translation is triggered via POST /pipeline/translate; this worker only +// Batch translation is triggered via POST /ai/translate/batch; this worker only // verifies which tenders now have translations available in storage. type TranslationWorker struct { Mongo *mongo.ConnectionManager @@ -82,11 +82,14 @@ func (w *TranslationWorker) Run() { startCount := w.getSuccessfulTranslationCount(mongo.AITranslationSuccessCounterKeyDailyJob) - if _, err := w.AIClient.TriggerPipelineTranslate(context.Background(), w.TargetLanguages); err != nil { - w.Logger.Warn("Failed to trigger AI pipeline translate", map[string]interface{}{ - "target_languages": w.TargetLanguages, - "error": err.Error(), - }) + if refs := w.collectUnTranslatedTenderRefs(context.Background()); len(refs) > 0 { + if _, err := w.AIClient.TriggerTranslateBatch(context.Background(), refs, w.TargetLanguages); err != nil { + w.Logger.Warn("Failed to trigger AI translate batch", map[string]interface{}{ + "tender_count": len(refs), + "target_languages": w.TargetLanguages, + "error": err.Error(), + }) + } } readyCount := 0 @@ -186,3 +189,44 @@ func (w *TranslationWorker) hasTranslationInStorage(t *tender.Tender, language s ) return err == nil && strings.TrimSpace(stored.Title) != "" } + +func (w *TranslationWorker) collectUnTranslatedTenderRefs(ctx context.Context) []ai_summarizer.TenderRef { + seen := make(map[string]struct{}) + refs := make([]ai_summarizer.TenderRef, 0) + + for _, language := range w.TargetLanguages { + skip := 0 + for { + tenders, totalCount, err := w.TenderRepo.GetUnTranslatedTenders(ctx, language, w.BatchSize, skip) + if err != nil { + w.Logger.Warn("Failed to collect tenders for translate batch", map[string]interface{}{ + "target_language": language, + "error": err.Error(), + }) + break + } + if len(tenders) == 0 { + break + } + + for _, t := range tenders { + if strings.TrimSpace(t.ContractFolderID) == "" || strings.TrimSpace(t.NoticePublicationID) == "" { + continue + } + key := t.ContractFolderID + "|" + t.NoticePublicationID + if _, dup := seen[key]; dup { + continue + } + seen[key] = struct{}{} + refs = append(refs, ai_summarizer.NewTenderRef(t.ContractFolderID, t.NoticePublicationID)) + } + + skip += len(tenders) + if int64(skip) >= totalCount { + break + } + } + } + + return refs +} diff --git a/internal/ai_pipeline/errors.go b/internal/ai_pipeline/errors.go new file mode 100644 index 0000000..4cabd18 --- /dev/null +++ b/internal/ai_pipeline/errors.go @@ -0,0 +1,14 @@ +package ai_pipeline + +import "errors" + +var ( + // ErrAINotConfigured is returned when the Opplens AI client is not configured. + ErrAINotConfigured = errors.New("ai_pipeline: AI service not configured") + + // ErrEmptyTenderBatch is returned when a batch request has no tenders. + ErrEmptyTenderBatch = errors.New("ai_pipeline: at least one tender is required") + + // ErrEmptyLanguages is returned when languages are required but missing. + ErrEmptyLanguages = errors.New("ai_pipeline: at least one language is required") +) diff --git a/internal/ai_pipeline/form.go b/internal/ai_pipeline/form.go new file mode 100644 index 0000000..21d9b9a --- /dev/null +++ b/internal/ai_pipeline/form.go @@ -0,0 +1,73 @@ +package ai_pipeline + +import ( + "strings" + + "tm/pkg/ai_summarizer" +) + +// TenderRefForm identifies one tender in batch and pipeline requests. +type TenderRefForm struct { + ContractFolderID string `json:"contract_folder_id" valid:"required"` + NoticePublicationID string `json:"notice_publication_id" valid:"required"` +} + +// ScrapeDocumentsForm is the request for POST /admin/v1/ai-pipeline/scrape/documents. +type ScrapeDocumentsForm struct { + ContractFolderID string `json:"contract_folder_id" valid:"required"` + NoticePublicationID string `json:"notice_publication_id" valid:"required"` +} + +// TenderBatchForm is the request body for batch scrape and summarize endpoints. +type TenderBatchForm struct { + Tenders []TenderRefForm `json:"tenders" valid:"required"` +} + +// TranslateBatchForm is the request body for POST /admin/v1/ai-pipeline/translate/batch. +type TranslateBatchForm struct { + Tenders []TenderRefForm `json:"tenders" valid:"required"` + Languages []string `json:"languages" valid:"required"` +} + +// PipelineRunForm is the request body for POST /admin/v1/ai-pipeline/run. +type PipelineRunForm struct { + ContractFolderID string `json:"contract_folder_id" valid:"required"` + NoticePublicationID string `json:"notice_publication_id" valid:"required"` + Languages []string `json:"languages" valid:"required"` +} + +// PipelineRunBatchForm is the request body for POST /admin/v1/ai-pipeline/run/batch. +type PipelineRunBatchForm struct { + Tenders []TenderRefForm `json:"tenders" valid:"required"` + Languages []string `json:"languages" valid:"required"` +} + +// PipelineWindowQueryForm is the query form for report and stats endpoints. +type PipelineWindowQueryForm struct { + Window string `query:"window" valid:"optional,in(last_hour|last_24h|today|yesterday|last_7_days|last_30_days|all|daily)"` +} + +func toTenderRefs(forms []TenderRefForm) []ai_summarizer.TenderRef { + refs := make([]ai_summarizer.TenderRef, 0, len(forms)) + for _, form := range forms { + refs = append(refs, ai_summarizer.NewTenderRef(form.ContractFolderID, form.NoticePublicationID)) + } + return refs +} + +func normalizeLanguages(languages []string) []string { + out := make([]string, 0, len(languages)) + seen := make(map[string]struct{}, len(languages)) + for _, language := range languages { + lang := strings.ToLower(strings.TrimSpace(language)) + if lang == "" { + continue + } + if _, dup := seen[lang]; dup { + continue + } + seen[lang] = struct{}{} + out = append(out, lang) + } + return out +} diff --git a/internal/ai_pipeline/handler.go b/internal/ai_pipeline/handler.go new file mode 100644 index 0000000..c497bd9 --- /dev/null +++ b/internal/ai_pipeline/handler.go @@ -0,0 +1,401 @@ +package ai_pipeline + +import ( + "errors" + "net/http" + + "tm/pkg/ai_summarizer" + "tm/pkg/response" + + "github.com/labstack/echo/v4" +) + +// Handler handles admin HTTP requests for Opplens AI pipeline operations. +type Handler struct { + service Service +} + +// NewHandler creates a new AI pipeline handler. +func NewHandler(service Service) *Handler { + return &Handler{service: service} +} + +// ScrapeDocuments scrapes documents for one tender via the Opplens AI service. +// @Summary Scrape tender documents +// @Description Download all documents for one tender through the Opplens AI service (synchronous, cached) +// @Tags Admin-AI-Pipeline +// @Accept json +// @Produce json +// @Param request body ScrapeDocumentsForm true "Tender identifiers" +// @Success 200 {object} response.APIResponse{data=ai_summarizer.ScrapeDocumentsResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 404 {object} response.APIResponse +// @Failure 422 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/scrape/documents [post] +func (h *Handler) ScrapeDocuments(c echo.Context) error { + form, err := response.Parse[ScrapeDocumentsForm](c) + if err != nil { + return response.ValidationError(c, "Invalid request data", err.Error()) + } + + result, err := h.service.ScrapeDocuments(c.Request().Context(), *form) + if err != nil { + return mapPipelineHTTPError(c, err, "Scrape documents") + } + + return response.Success(c, result, "Documents scraped successfully") +} + +// ScrapeDocumentsBatch enqueues background document scraping for many tenders. +// @Summary Batch scrape tender documents +// @Description Enqueue background document scraping for multiple tenders via the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Accept json +// @Produce json +// @Param request body TenderBatchForm true "Tenders to scrape" +// @Success 202 {object} response.APIResponse{data=ai_summarizer.BatchAcceptedResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 422 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/scrape/documents/batch [post] +func (h *Handler) ScrapeDocumentsBatch(c echo.Context) error { + form, err := response.Parse[TenderBatchForm](c) + if err != nil { + return response.ValidationError(c, "Invalid request data", err.Error()) + } + + result, err := h.service.ScrapeDocumentsBatch(c.Request().Context(), *form) + if err != nil { + return mapPipelineHTTPError(c, err, "Scrape documents batch") + } + + return response.Accepted(c, result, "Scrape batch accepted") +} + +// SummarizeBatch enqueues background summarization for many tenders. +// @Summary Batch summarize tenders +// @Description Enqueue background AI summarization for multiple tenders via the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Accept json +// @Produce json +// @Param request body TenderBatchForm true "Tenders to summarize" +// @Success 202 {object} response.APIResponse{data=ai_summarizer.BatchAcceptedResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 422 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/summarize/batch [post] +func (h *Handler) SummarizeBatch(c echo.Context) error { + form, err := response.Parse[TenderBatchForm](c) + if err != nil { + return response.ValidationError(c, "Invalid request data", err.Error()) + } + + result, err := h.service.SummarizeBatch(c.Request().Context(), *form) + if err != nil { + return mapPipelineHTTPError(c, err, "Summarize batch") + } + + return response.Accepted(c, result, "Summarize batch accepted") +} + +// TranslateBatch enqueues background translation for many tenders. +// @Summary Batch translate tenders +// @Description Enqueue background AI translation for multiple tenders via the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Accept json +// @Produce json +// @Param request body TranslateBatchForm true "Tenders and target languages" +// @Success 202 {object} response.APIResponse{data=ai_summarizer.BatchAcceptedResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 422 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/translate/batch [post] +func (h *Handler) TranslateBatch(c echo.Context) error { + form, err := response.Parse[TranslateBatchForm](c) + if err != nil { + return response.ValidationError(c, "Invalid request data", err.Error()) + } + + result, err := h.service.TranslateBatch(c.Request().Context(), *form) + if err != nil { + return mapPipelineHTTPError(c, err, "Translate batch") + } + + return response.Accepted(c, result, "Translate batch accepted") +} + +// Sync pulls the tender list from this backend into the Opplens AI service. +// @Summary Sync tenders to AI pipeline +// @Description Pull the tender list from the backend and store it in the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Produce json +// @Success 200 {object} response.APIResponse{data=ai_summarizer.PipelineSyncResponse} +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/sync [post] +func (h *Handler) Sync(c echo.Context) error { + result, err := h.service.Sync(c.Request().Context()) + if err != nil { + return mapPipelineHTTPError(c, err, "Pipeline sync") + } + + return response.Success(c, result, "Pipeline sync completed successfully") +} + +// Run runs the full AI pipeline for one tender (scrape → translate → summarize). +// @Summary Run full AI pipeline for one tender +// @Description Run scrape, translate, and summarize for one tender via the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Accept json +// @Produce json +// @Param request body PipelineRunForm true "Tender and languages" +// @Success 200 {object} response.APIResponse{data=object} +// @Failure 400 {object} response.APIResponse +// @Failure 404 {object} response.APIResponse +// @Failure 422 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/run [post] +func (h *Handler) Run(c echo.Context) error { + form, err := response.Parse[PipelineRunForm](c) + if err != nil { + return response.ValidationError(c, "Invalid request data", err.Error()) + } + + result, err := h.service.Run(c.Request().Context(), *form) + if err != nil { + return mapPipelineHTTPError(c, err, "Pipeline run") + } + + return response.Success(c, result, "Pipeline run completed successfully") +} + +// RunBatch enqueues the full AI pipeline for many tenders. +// @Summary Batch run full AI pipeline +// @Description Enqueue scrape, translate, and summarize for multiple tenders via the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Accept json +// @Produce json +// @Param request body PipelineRunBatchForm true "Tenders and languages" +// @Success 202 {object} response.APIResponse{data=ai_summarizer.BatchAcceptedResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 422 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/run/batch [post] +func (h *Handler) RunBatch(c echo.Context) error { + form, err := response.Parse[PipelineRunBatchForm](c) + if err != nil { + return response.ValidationError(c, "Invalid request data", err.Error()) + } + + result, err := h.service.RunBatch(c.Request().Context(), *form) + if err != nil { + return mapPipelineHTTPError(c, err, "Pipeline run batch") + } + + return response.Accepted(c, result, "Pipeline run batch accepted") +} + +// Analyze triggers analysis across stored tenders in the Opplens AI service. +// @Summary Trigger AI analysis pipeline +// @Description Trigger agentic analysis across stored tenders via the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Produce json +// @Success 200 {object} response.APIResponse{data=ai_summarizer.PipelineActionResponse} +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/analyze [post] +func (h *Handler) Analyze(c echo.Context) error { + result, err := h.service.Analyze(c.Request().Context()) + if err != nil { + return mapPipelineHTTPError(c, err, "Pipeline analyze") + } + + return response.Success(c, result, "Pipeline analyze triggered successfully") +} + +// DailyRun syncs tenders and runs the full pipeline on newly synced tenders. +// @Summary Start daily AI pipeline run +// @Description Sync tenders, then run the full pipeline on newly synced tenders (background, single-flight) +// @Tags Admin-AI-Pipeline +// @Produce json +// @Success 202 {object} response.APIResponse{data=ai_summarizer.PipelineStartedResponse} +// @Failure 409 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/daily-run [post] +func (h *Handler) DailyRun(c echo.Context) error { + result, err := h.service.DailyRun(c.Request().Context()) + if err != nil { + return mapPipelineHTTPError(c, err, "Pipeline daily-run") + } + + return response.Accepted(c, result, "Pipeline daily-run started") +} + +// Auto syncs tenders and completes all missing AI work across stored tenders. +// @Summary Start auto AI pipeline run +// @Description Sync tenders, then complete all missing scrape/translate/summarize work (background, single-flight, idempotent) +// @Tags Admin-AI-Pipeline +// @Produce json +// @Success 202 {object} response.APIResponse{data=ai_summarizer.PipelineStartedResponse} +// @Failure 409 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/auto [post] +func (h *Handler) Auto(c echo.Context) error { + result, err := h.service.Auto(c.Request().Context()) + if err != nil { + return mapPipelineHTTPError(c, err, "Pipeline auto") + } + + return response.Accepted(c, result, "Pipeline auto started") +} + +// GetLastRun returns the result of the last daily-run. +// @Summary Get last daily-run report +// @Description Retrieve the report for the last pipeline daily-run from the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Produce json +// @Success 200 {object} response.APIResponse{data=ai_summarizer.PipelineReportResponse} +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/last-run [get] +func (h *Handler) GetLastRun(c echo.Context) error { + result, err := h.service.GetLastRun(c.Request().Context()) + if err != nil { + return mapPipelineHTTPError(c, err, "Get last pipeline run") + } + + return response.Success(c, result, "Last pipeline run retrieved successfully") +} + +// GetLastAutoRun returns the result of the last auto run. +// @Summary Get last auto-run report +// @Description Retrieve the report for the last pipeline auto run from the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Produce json +// @Success 200 {object} response.APIResponse{data=ai_summarizer.PipelineReportResponse} +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/last-auto-run [get] +func (h *Handler) GetLastAutoRun(c echo.Context) error { + result, err := h.service.GetLastAutoRun(c.Request().Context()) + if err != nil { + return mapPipelineHTTPError(c, err, "Get last pipeline auto run") + } + + return response.Success(c, result, "Last pipeline auto run retrieved successfully") +} + +// GetReport returns an event-log summary for a time window. +// @Summary Get AI pipeline event report +// @Description Retrieve an event-log summary for a time window from the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Produce json +// @Param window query string false "Time window" Enums(last_hour,last_24h,today,yesterday,last_7_days,last_30_days,all,daily) default(last_24h) +// @Success 200 {object} response.APIResponse{data=ai_summarizer.PipelineReportResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 422 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/report [get] +func (h *Handler) GetReport(c echo.Context) error { + form, err := response.Parse[PipelineWindowQueryForm](c) + if err != nil { + return response.ValidationError(c, "Invalid query parameters", err.Error()) + } + + result, err := h.service.GetReport(c.Request().Context(), form.Window) + if err != nil { + return mapPipelineHTTPError(c, err, "Get pipeline report") + } + + return response.Success(c, result, "Pipeline report retrieved successfully") +} + +// GetStats returns event-log and storage inventory for a time window. +// @Summary Get AI pipeline stats +// @Description Retrieve event-log summary and MinIO inventory for a time window from the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Produce json +// @Param window query string false "Time window" Enums(last_hour,last_24h,today,yesterday,last_7_days,last_30_days,all,daily) default(last_24h) +// @Success 200 {object} response.APIResponse{data=ai_summarizer.PipelineStatsResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 422 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/stats [get] +func (h *Handler) GetStats(c echo.Context) error { + form, err := response.Parse[PipelineWindowQueryForm](c) + if err != nil { + return response.ValidationError(c, "Invalid query parameters", err.Error()) + } + + result, err := h.service.GetStats(c.Request().Context(), form.Window) + if err != nil { + return mapPipelineHTTPError(c, err, "Get pipeline stats") + } + + return response.Success(c, result, "Pipeline stats retrieved successfully") +} + +// GetMinioStats returns storage inventory from the Opplens AI service. +// @Summary Get AI pipeline MinIO stats +// @Description Retrieve storage inventory only from the Opplens AI service +// @Tags Admin-AI-Pipeline +// @Produce json +// @Success 200 {object} response.APIResponse{data=ai_summarizer.PipelineMinioStatsResponse} +// @Failure 500 {object} response.APIResponse +// @Failure 503 {object} response.APIResponse +// @Security BearerAuth +// @Router /admin/v1/ai-pipeline/minio-stats [get] +func (h *Handler) GetMinioStats(c echo.Context) error { + result, err := h.service.GetMinioStats(c.Request().Context()) + if err != nil { + return mapPipelineHTTPError(c, err, "Get pipeline minio stats") + } + + return response.Success(c, result, "Pipeline MinIO stats retrieved successfully") +} + +func mapPipelineHTTPError(c echo.Context, err error, action string) error { + switch { + case errors.Is(err, ErrAINotConfigured): + return response.ServiceUnavailable(c, "AI service is not configured") + case errors.Is(err, ErrEmptyTenderBatch), errors.Is(err, ErrEmptyLanguages): + return response.BadRequest(c, "Invalid request", err.Error()) + case errors.Is(err, ai_summarizer.ErrPipelineJobRunning): + return response.Conflict(c, "A pipeline job is already running") + case errors.Is(err, ai_summarizer.ErrObjectNotFound): + return response.NotFound(c, "Tender not found in AI service") + default: + if errors.Is(err, ai_summarizer.ErrAPINonSuccess) { + return c.JSON(http.StatusBadGateway, response.APIResponse{ + Success: false, + Message: action + " failed: upstream AI service error", + }) + } + return response.InternalServerError(c, action+" failed") + } +} diff --git a/internal/ai_pipeline/service.go b/internal/ai_pipeline/service.go new file mode 100644 index 0000000..35fe1bc --- /dev/null +++ b/internal/ai_pipeline/service.go @@ -0,0 +1,400 @@ +package ai_pipeline + +import ( + "context" + "fmt" + + "tm/pkg/ai_summarizer" + "tm/pkg/logger" +) + +// Client defines Opplens AI operations exposed to the admin API. +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) + 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) + PipelineRun(ctx context.Context, reqBody ai_summarizer.PipelineRunRequest) (map[string]interface{}, error) + PipelineRunBatch(ctx context.Context, tenders []ai_summarizer.TenderRef, languages []string) (*ai_summarizer.BatchAcceptedResponse, error) + PipelineAnalyze(ctx context.Context) (*ai_summarizer.PipelineActionResponse, error) + PipelineDailyRun(ctx context.Context) (*ai_summarizer.PipelineStartedResponse, error) + PipelineAuto(ctx context.Context) (*ai_summarizer.PipelineStartedResponse, error) + GetPipelineLastRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error) + GetPipelineLastAutoRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error) + GetPipelineReport(ctx context.Context, window string) (*ai_summarizer.PipelineReportResponse, error) + GetPipelineStats(ctx context.Context, window string) (*ai_summarizer.PipelineStatsResponse, error) + GetPipelineMinioStats(ctx context.Context) (ai_summarizer.PipelineMinioStatsResponse, error) +} + +// Service proxies admin pipeline operations to the Opplens AI service. +type Service interface { + ScrapeDocuments(ctx context.Context, form ScrapeDocumentsForm) (*ai_summarizer.ScrapeDocumentsResponse, error) + ScrapeDocumentsBatch(ctx context.Context, form TenderBatchForm) (*ai_summarizer.BatchAcceptedResponse, 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) + Run(ctx context.Context, form PipelineRunForm) (map[string]interface{}, error) + RunBatch(ctx context.Context, form PipelineRunBatchForm) (*ai_summarizer.BatchAcceptedResponse, error) + Analyze(ctx context.Context) (*ai_summarizer.PipelineActionResponse, error) + DailyRun(ctx context.Context) (*ai_summarizer.PipelineStartedResponse, error) + Auto(ctx context.Context) (*ai_summarizer.PipelineStartedResponse, error) + GetLastRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error) + GetLastAutoRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error) + GetReport(ctx context.Context, window string) (*ai_summarizer.PipelineReportResponse, error) + GetStats(ctx context.Context, window string) (*ai_summarizer.PipelineStatsResponse, error) + GetMinioStats(ctx context.Context) (ai_summarizer.PipelineMinioStatsResponse, error) +} + +type service struct { + client Client + logger logger.Logger +} + +// NewService creates an AI pipeline admin service. +func NewService(client Client, log logger.Logger) Service { + return &service{ + client: client, + logger: log, + } +} + +func (s *service) requireClient() error { + if s.client == nil { + return ErrAINotConfigured + } + return nil +} + +func (s *service) ScrapeDocuments(ctx context.Context, form ScrapeDocumentsForm) (*ai_summarizer.ScrapeDocumentsResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + + s.logger.Info("Admin scrape documents requested", map[string]interface{}{ + "contract_folder_id": form.ContractFolderID, + "notice_publication_id": form.NoticePublicationID, + }) + + resp, err := s.client.ScrapeDocuments(ctx, ai_summarizer.ScrapeDocumentsRequest{ + ContractFolderID: form.ContractFolderID, + NoticePublicationID: form.NoticePublicationID, + }) + if err != nil { + s.logger.Error("Admin scrape documents failed", map[string]interface{}{ + "contract_folder_id": form.ContractFolderID, + "notice_publication_id": form.NoticePublicationID, + "error": err.Error(), + }) + return nil, fmt.Errorf("scrape documents failed: %w", err) + } + + return resp, nil +} + +func (s *service) ScrapeDocumentsBatch(ctx context.Context, form TenderBatchForm) (*ai_summarizer.BatchAcceptedResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + refs, err := validateTenderBatch(form.Tenders) + if err != nil { + return nil, err + } + + s.logger.Info("Admin scrape documents batch requested", map[string]interface{}{ + "tender_count": len(refs), + }) + + resp, err := s.client.ScrapeDocumentsBatch(ctx, refs) + if err != nil { + s.logger.Error("Admin scrape documents batch failed", map[string]interface{}{ + "tender_count": len(refs), + "error": err.Error(), + }) + return nil, fmt.Errorf("scrape documents batch 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 + } + refs, err := validateTenderBatch(form.Tenders) + if err != nil { + return nil, err + } + + s.logger.Info("Admin summarize batch requested", map[string]interface{}{ + "tender_count": len(refs), + }) + + resp, err := s.client.TriggerSummarizeBatch(ctx, refs) + if err != nil { + s.logger.Error("Admin summarize batch failed", map[string]interface{}{ + "tender_count": len(refs), + "error": err.Error(), + }) + return nil, fmt.Errorf("summarize batch failed: %w", err) + } + + return resp, nil +} + +func (s *service) TranslateBatch(ctx context.Context, form TranslateBatchForm) (*ai_summarizer.BatchAcceptedResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + refs, err := validateTenderBatch(form.Tenders) + if err != nil { + return nil, err + } + languages := normalizeLanguages(form.Languages) + if len(languages) == 0 { + return nil, ErrEmptyLanguages + } + + s.logger.Info("Admin translate batch requested", map[string]interface{}{ + "tender_count": len(refs), + "languages": languages, + }) + + resp, err := s.client.TriggerTranslateBatch(ctx, refs, languages) + if err != nil { + s.logger.Error("Admin translate batch failed", map[string]interface{}{ + "tender_count": len(refs), + "languages": languages, + "error": err.Error(), + }) + return nil, fmt.Errorf("translate batch failed: %w", err) + } + + return resp, nil +} + +func (s *service) Sync(ctx context.Context) (*ai_summarizer.PipelineSyncResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + + s.logger.Info("Admin pipeline sync requested", map[string]interface{}{}) + + resp, err := s.client.PipelineSync(ctx) + if err != nil { + s.logger.Error("Admin pipeline sync failed", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("pipeline sync failed: %w", err) + } + + return resp, nil +} + +func (s *service) Run(ctx context.Context, form PipelineRunForm) (map[string]interface{}, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + languages := normalizeLanguages(form.Languages) + if len(languages) == 0 { + return nil, ErrEmptyLanguages + } + + req := ai_summarizer.PipelineRunRequest{ + ContractFolderID: form.ContractFolderID, + NoticePublicationID: form.NoticePublicationID, + Languages: languages, + } + + s.logger.Info("Admin pipeline run requested", map[string]interface{}{ + "contract_folder_id": req.ContractFolderID, + "notice_publication_id": req.NoticePublicationID, + "languages": languages, + }) + + resp, err := s.client.PipelineRun(ctx, req) + if err != nil { + s.logger.Error("Admin pipeline run failed", map[string]interface{}{ + "contract_folder_id": req.ContractFolderID, + "notice_publication_id": req.NoticePublicationID, + "error": err.Error(), + }) + return nil, fmt.Errorf("pipeline run failed: %w", err) + } + + return resp, nil +} + +func (s *service) RunBatch(ctx context.Context, form PipelineRunBatchForm) (*ai_summarizer.BatchAcceptedResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + refs, err := validateTenderBatch(form.Tenders) + if err != nil { + return nil, err + } + languages := normalizeLanguages(form.Languages) + if len(languages) == 0 { + return nil, ErrEmptyLanguages + } + + s.logger.Info("Admin pipeline run batch requested", map[string]interface{}{ + "tender_count": len(refs), + "languages": languages, + }) + + resp, err := s.client.PipelineRunBatch(ctx, refs, languages) + if err != nil { + s.logger.Error("Admin pipeline run batch failed", map[string]interface{}{ + "tender_count": len(refs), + "languages": languages, + "error": err.Error(), + }) + return nil, fmt.Errorf("pipeline run batch failed: %w", err) + } + + return resp, nil +} + +func (s *service) Analyze(ctx context.Context) (*ai_summarizer.PipelineActionResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + + s.logger.Info("Admin pipeline analyze requested", map[string]interface{}{}) + + resp, err := s.client.PipelineAnalyze(ctx) + if err != nil { + s.logger.Error("Admin pipeline analyze failed", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("pipeline analyze failed: %w", err) + } + + return resp, nil +} + +func (s *service) DailyRun(ctx context.Context) (*ai_summarizer.PipelineStartedResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + + s.logger.Info("Admin pipeline daily-run requested", map[string]interface{}{}) + + resp, err := s.client.PipelineDailyRun(ctx) + if err != nil { + s.logger.Error("Admin pipeline daily-run failed", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("pipeline daily-run failed: %w", err) + } + + return resp, nil +} + +func (s *service) Auto(ctx context.Context) (*ai_summarizer.PipelineStartedResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + + s.logger.Info("Admin pipeline auto requested", map[string]interface{}{}) + + resp, err := s.client.PipelineAuto(ctx) + if err != nil { + s.logger.Error("Admin pipeline auto failed", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("pipeline auto failed: %w", err) + } + + return resp, nil +} + +func (s *service) GetLastRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + + resp, err := s.client.GetPipelineLastRun(ctx) + if err != nil { + s.logger.Error("Admin get pipeline last-run failed", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("get pipeline last-run failed: %w", err) + } + + return resp, nil +} + +func (s *service) GetLastAutoRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + + resp, err := s.client.GetPipelineLastAutoRun(ctx) + if err != nil { + s.logger.Error("Admin get pipeline last-auto-run failed", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("get pipeline last-auto-run failed: %w", err) + } + + return resp, nil +} + +func (s *service) GetReport(ctx context.Context, window string) (*ai_summarizer.PipelineReportResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + + resp, err := s.client.GetPipelineReport(ctx, window) + if err != nil { + s.logger.Error("Admin get pipeline report failed", map[string]interface{}{ + "window": window, + "error": err.Error(), + }) + return nil, fmt.Errorf("get pipeline report failed: %w", err) + } + + return resp, nil +} + +func (s *service) GetStats(ctx context.Context, window string) (*ai_summarizer.PipelineStatsResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + + resp, err := s.client.GetPipelineStats(ctx, window) + if err != nil { + s.logger.Error("Admin get pipeline stats failed", map[string]interface{}{ + "window": window, + "error": err.Error(), + }) + return nil, fmt.Errorf("get pipeline stats failed: %w", err) + } + + return resp, nil +} + +func (s *service) GetMinioStats(ctx context.Context) (ai_summarizer.PipelineMinioStatsResponse, error) { + if err := s.requireClient(); err != nil { + return nil, err + } + + resp, err := s.client.GetPipelineMinioStats(ctx) + if err != nil { + s.logger.Error("Admin get pipeline minio stats failed", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("get pipeline minio stats failed: %w", err) + } + + return resp, nil +} + +func validateTenderBatch(forms []TenderRefForm) ([]ai_summarizer.TenderRef, error) { + if len(forms) == 0 { + return nil, ErrEmptyTenderBatch + } + return toTenderRefs(forms), nil +} diff --git a/internal/tender/service.go b/internal/tender/service.go index 840c9e2..8132f87 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -24,7 +24,8 @@ type AISummarizerClient interface { FetchSummaryOnDemand(ctx context.Context, reqBody ai_summarizer.SummarizeRequest) (*ai_summarizer.SummarizeResponse, error) FetchTranslationOnDemand(ctx context.Context, reqBody ai_summarizer.TranslateRequest) (*ai_summarizer.TranslateResponse, error) FetchAnalyzeOnDemand(ctx context.Context, reqBody ai_summarizer.AnalyzeRequest) (*ai_summarizer.AnalyzeResponse, error) - TriggerPipelineTranslate(ctx context.Context, languages []string) (*ai_summarizer.PipelineTranslateResponse, error) + TriggerTranslateBatch(ctx context.Context, tenders []ai_summarizer.TenderRef, languages []string) (*ai_summarizer.BatchAcceptedResponse, error) + TriggerSummarizeBatch(ctx context.Context, tenders []ai_summarizer.TenderRef) (*ai_summarizer.BatchAcceptedResponse, error) } // AISummarizerStorage defines the interface for fetching summaries from MinIO storage. @@ -1646,12 +1647,15 @@ func (s *tenderService) resolveTranslation(ctx context.Context, t *Tender, langu } } - resp, err := s.aiSummarizerClient.FetchTranslationOnDemand(ctx, ai_summarizer.TranslateRequest{ - ContractFolderID: t.ContractFolderID, - NoticePublicationID: t.NoticePublicationID, - Language: language, - RequestSource: requestSource, - }) + req := ai_summarizer.NewTranslateRequest( + t.ContractFolderID, + t.NoticePublicationID, + language, + t.Title, + t.Description, + ) + req.RequestSource = requestSource + resp, err := s.aiSummarizerClient.FetchTranslationOnDemand(ctx, req) if err != nil { return "", "", false, err } diff --git a/pkg/ai_summarizer/client.go b/pkg/ai_summarizer/client.go index 6d4f620..5eae5e8 100644 --- a/pkg/ai_summarizer/client.go +++ b/pkg/ai_summarizer/client.go @@ -125,50 +125,349 @@ func (c *Client) FetchTranslationOnDemand(ctx context.Context, reqBody Translate return nil, fmt.Errorf("ai_summarizer: all %d translation attempts failed, last error: %w", c.config.APIRetryCount+1, lastErr) } -// TriggerPipelineTranslate calls POST /pipeline/translate to enqueue batch translation -// for the given target languages. -func (c *Client) TriggerPipelineTranslate(ctx context.Context, languages []string) (*PipelineTranslateResponse, error) { - reqBody := PipelineTranslateRequest{Languages: languages} +// TriggerTranslateBatch calls POST /ai/translate/batch to enqueue background translation. +func (c *Client) TriggerTranslateBatch(ctx context.Context, tenders []TenderRef, languages []string) (*BatchAcceptedResponse, error) { + reqBody := TranslateBatchRequest{ + Tenders: tenders, + Languages: languages, + } + return c.postBatchAccepted(ctx, "/ai/translate/batch", reqBody, "translate batch", map[string]interface{}{ + "tender_count": len(tenders), + "languages": languages, + }) +} + +// TriggerSummarizeBatch calls POST /ai/summarize/batch to enqueue background summarization. +func (c *Client) TriggerSummarizeBatch(ctx context.Context, tenders []TenderRef) (*BatchAcceptedResponse, error) { + reqBody := SummarizeBatchRequest{Tenders: tenders} + return c.postBatchAccepted(ctx, "/ai/summarize/batch", reqBody, "summarize batch", map[string]interface{}{ + "tender_count": len(tenders), + }) +} + +// ScrapeDocuments calls POST /scrape/documents for one tender (synchronous, cached). +func (c *Client) ScrapeDocuments(ctx context.Context, reqBody ScrapeDocumentsRequest) (*ScrapeDocumentsResponse, error) { jsonBody, err := json.Marshal(reqBody) if err != nil { - return nil, fmt.Errorf("ai_summarizer: failed to marshal pipeline translate request: %w", err) + return nil, fmt.Errorf("ai_summarizer: failed to marshal scrape request: %w", err) } - url := c.config.APIBaseURL + "/pipeline/translate" - req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(jsonBody)) + url := c.config.APIBaseURL + "/scrape/documents" + httpResp, bodyBytes, err := c.doRawPost(ctx, url, jsonBody) if err != nil { - return nil, fmt.Errorf("ai_summarizer: failed to create pipeline translate request: %w", err) - } - req.Header.Set("Content-Type", "application/json") - req.Header.Set("Accept", "application/json") - - httpResp, err := c.httpClient.Do(req) - if err != nil { - return nil, fmt.Errorf("ai_summarizer: pipeline translate request error: %w", err) + return nil, err } defer httpResp.Body.Close() - bodyBytes, err := io.ReadAll(httpResp.Body) - if err != nil { - return nil, fmt.Errorf("ai_summarizer: failed to read pipeline translate response: %w", err) + if httpResp.StatusCode == http.StatusNotFound { + return nil, fmt.Errorf("%w: status 404, body: %s", ErrObjectNotFound, string(bodyBytes)) } if httpResp.StatusCode >= 400 { return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) } - var result PipelineTranslateResponse + var result ScrapeDocumentsResponse if err := json.Unmarshal(bodyBytes, &result); err != nil { - return nil, fmt.Errorf("ai_summarizer: failed to decode pipeline translate response: %w", err) + return nil, fmt.Errorf("ai_summarizer: failed to decode scrape response: %w", err) } - c.logger.Info("Pipeline translate triggered", map[string]interface{}{ - "status": result.Status, - "languages": result.Languages, + c.logger.Info("AI scrape documents request succeeded", map[string]interface{}{ + "contract_folder_id": reqBody.ContractFolderID, + "notice_publication_id": reqBody.NoticePublicationID, + "files_downloaded": result.FilesDownloaded, }) return &result, nil } +// ScrapeDocumentsBatch calls POST /scrape/documents/batch. +func (c *Client) ScrapeDocumentsBatch(ctx context.Context, tenders []TenderRef) (*BatchAcceptedResponse, error) { + reqBody := ScrapeDocumentsBatchRequest{Tenders: tenders} + return c.postBatchAccepted(ctx, "/scrape/documents/batch", reqBody, "scrape batch", map[string]interface{}{ + "tender_count": len(tenders), + }) +} + +// 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" + httpResp, bodyBytes, err := c.doRawPost(ctx, url, nil) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + if httpResp.StatusCode >= 400 { + return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + } + + var result PipelineSyncResponse + if len(bodyBytes) > 0 { + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode pipeline sync response: %w", err) + } + } + + c.logger.Info("AI pipeline sync completed", map[string]interface{}{ + "stored": result.Stored, + }) + + return &result, nil +} + +// PipelineRun calls POST /pipeline/run for one tender (scrape → translate → summarize). +func (c *Client) PipelineRun(ctx context.Context, reqBody PipelineRunRequest) (map[string]interface{}, error) { + jsonBody, err := json.Marshal(reqBody) + if err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to marshal pipeline run request: %w", err) + } + + url := c.config.APIBaseURL + "/pipeline/run" + httpResp, bodyBytes, err := c.doRawPost(ctx, url, jsonBody) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + if httpResp.StatusCode == http.StatusNotFound { + return nil, fmt.Errorf("%w: status 404, body: %s", ErrObjectNotFound, string(bodyBytes)) + } + if httpResp.StatusCode >= 400 { + return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + } + + var result map[string]interface{} + if len(bodyBytes) > 0 { + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode pipeline run response: %w", err) + } + } + + c.logger.Info("AI pipeline run completed", map[string]interface{}{ + "contract_folder_id": reqBody.ContractFolderID, + "notice_publication_id": reqBody.NoticePublicationID, + "languages": reqBody.Languages, + }) + + return result, nil +} + +// PipelineRunBatch calls POST /pipeline/run/batch. +func (c *Client) PipelineRunBatch(ctx context.Context, tenders []TenderRef, languages []string) (*BatchAcceptedResponse, error) { + reqBody := PipelineRunBatchRequest{ + Tenders: tenders, + Languages: languages, + } + return c.postBatchAccepted(ctx, "/pipeline/run/batch", reqBody, "pipeline run batch", map[string]interface{}{ + "tender_count": len(tenders), + "languages": languages, + }) +} + +// PipelineAnalyze calls POST /pipeline/analyze to trigger analysis across stored tenders. +func (c *Client) PipelineAnalyze(ctx context.Context) (*PipelineActionResponse, error) { + return c.triggerPipelineAction(ctx, "/pipeline/analyze", "pipeline analyze") +} + +// PipelineDailyRun calls POST /pipeline/daily-run (sync + full pipeline on newly synced tenders). +func (c *Client) PipelineDailyRun(ctx context.Context) (*PipelineStartedResponse, error) { + return c.triggerPipelineStarted(ctx, "/pipeline/daily-run", "pipeline daily-run") +} + +// PipelineAuto calls POST /pipeline/auto (sync + complete all missing work, idempotent). +func (c *Client) PipelineAuto(ctx context.Context) (*PipelineStartedResponse, error) { + return c.triggerPipelineStarted(ctx, "/pipeline/auto", "pipeline auto") +} + +// GetPipelineLastRun returns the result of the last daily-run (GET /pipeline/last-run). +func (c *Client) GetPipelineLastRun(ctx context.Context) (*PipelineReportResponse, error) { + return c.getPipelineReport(ctx, "/pipeline/last-run") +} + +// GetPipelineLastAutoRun returns the result of the last auto run (GET /pipeline/last-auto-run). +func (c *Client) GetPipelineLastAutoRun(ctx context.Context) (*PipelineReportResponse, error) { + return c.getPipelineReport(ctx, "/pipeline/last-auto-run") +} + +// GetPipelineReport calls GET /pipeline/report. +func (c *Client) GetPipelineReport(ctx context.Context, window string) (*PipelineReportResponse, error) { + return c.getPipelineReportWithQuery(ctx, "/pipeline/report", window) +} + +// GetPipelineStats calls GET /pipeline/stats. +func (c *Client) GetPipelineStats(ctx context.Context, window string) (*PipelineStatsResponse, error) { + url := c.config.APIBaseURL + "/pipeline/stats" + if window = strings.TrimSpace(window); window != "" { + url += "?window=" + window + } + + httpResp, bodyBytes, err := c.doRawGet(ctx, url) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + if httpResp.StatusCode == http.StatusBadRequest { + return nil, fmt.Errorf("%w: status 400, body: %s", ErrAPINonSuccess, string(bodyBytes)) + } + if httpResp.StatusCode >= 400 { + return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + } + + var result PipelineStatsResponse + if len(bodyBytes) > 0 { + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode pipeline stats response: %w", err) + } + } + + return &result, nil +} + +// GetPipelineMinioStats calls GET /pipeline/minio-stats. +func (c *Client) GetPipelineMinioStats(ctx context.Context) (PipelineMinioStatsResponse, error) { + url := c.config.APIBaseURL + "/pipeline/minio-stats" + httpResp, bodyBytes, err := c.doRawGet(ctx, url) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + if httpResp.StatusCode >= 400 { + return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + } + + var result PipelineMinioStatsResponse + if len(bodyBytes) > 0 { + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode pipeline minio stats response: %w", err) + } + } + + return result, nil +} + +func (c *Client) postBatchAccepted(ctx context.Context, path string, reqBody any, label string, fields map[string]interface{}) (*BatchAcceptedResponse, error) { + jsonBody, err := json.Marshal(reqBody) + if err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to marshal %s request: %w", label, err) + } + + url := c.config.APIBaseURL + path + httpResp, bodyBytes, err := c.doRawPost(ctx, url, jsonBody) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + if httpResp.StatusCode >= 400 { + return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + } + + var result BatchAcceptedResponse + if len(bodyBytes) > 0 { + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode %s response: %w", label, err) + } + } + + logFields := map[string]interface{}{ + "path": path, + "status": result.Status, + "count": result.Count, + } + for k, v := range fields { + logFields[k] = v + } + c.logger.Info("AI batch request accepted", logFields) + + return &result, nil +} + +func (c *Client) triggerPipelineStarted(ctx context.Context, path, label string) (*PipelineStartedResponse, error) { + url := c.config.APIBaseURL + path + httpResp, bodyBytes, err := c.doRawPost(ctx, url, nil) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + if httpResp.StatusCode == http.StatusConflict { + return nil, fmt.Errorf("%w: status 409, body: %s", ErrPipelineJobRunning, string(bodyBytes)) + } + if httpResp.StatusCode >= 400 { + return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + } + + var result PipelineStartedResponse + if len(bodyBytes) > 0 { + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode %s response: %w", label, err) + } + } + + c.logger.Info("AI pipeline job started", map[string]interface{}{ + "path": path, + "status": result.Status, + "report": result.Report, + }) + + return &result, nil +} + +func (c *Client) getPipelineReport(ctx context.Context, path string) (*PipelineReportResponse, error) { + return c.getPipelineReportWithQuery(ctx, path, "") +} + +func (c *Client) getPipelineReportWithQuery(ctx context.Context, path, window string) (*PipelineReportResponse, error) { + url := c.config.APIBaseURL + path + if window = strings.TrimSpace(window); window != "" { + url += "?window=" + window + } + + httpResp, bodyBytes, err := c.doRawGet(ctx, url) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + if httpResp.StatusCode == http.StatusBadRequest { + return nil, fmt.Errorf("%w: status 400, body: %s", ErrAPINonSuccess, string(bodyBytes)) + } + if httpResp.StatusCode >= 400 { + return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + } + + var result PipelineReportResponse + if len(bodyBytes) > 0 { + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode pipeline report response: %w", err) + } + } + + return &result, nil +} + +func (c *Client) doRawGet(ctx context.Context, url string) (*http.Response, []byte, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, nil, fmt.Errorf("ai_summarizer: failed to create request: %w", err) + } + req.Header.Set("Accept", "application/json") + + httpResp, err := c.httpClient.Do(req) + if err != nil { + return nil, nil, fmt.Errorf("ai_summarizer: request error: %w", err) + } + + bodyBytes, err := io.ReadAll(httpResp.Body) + if err != nil { + httpResp.Body.Close() + return nil, nil, fmt.Errorf("ai_summarizer: failed to read response body: %w", err) + } + return httpResp, bodyBytes, nil +} + // FetchAnalyzeOnDemand calls POST /ai/analyze for one tender. func (c *Client) FetchAnalyzeOnDemand(ctx context.Context, reqBody AnalyzeRequest) (*AnalyzeResponse, error) { jsonBody, err := json.Marshal(reqBody) @@ -204,11 +503,6 @@ func (c *Client) FetchAnalyzeOnDemand(ctx context.Context, reqBody AnalyzeReques return &result, nil } -// TriggerPipelineSummarize calls POST /pipeline/summarize. -func (c *Client) TriggerPipelineSummarize(ctx context.Context) (*PipelineActionResponse, error) { - return c.triggerPipelineAction(ctx, "/pipeline/summarize", "pipeline summarize") -} - // StartOnboarding calls POST /onboarding to start company profile indexing for recommendations. func (c *Client) StartOnboarding(ctx context.Context, reqBody OnboardingRequest) (*OnboardingResponse, error) { jsonBody, err := json.Marshal(reqBody) diff --git a/pkg/ai_summarizer/entities.go b/pkg/ai_summarizer/entities.go index d751f49..393e2a2 100644 --- a/pkg/ai_summarizer/entities.go +++ b/pkg/ai_summarizer/entities.go @@ -4,6 +4,20 @@ import ( "strings" ) +// TenderRef identifies a tender for batch and pipeline requests. +type TenderRef struct { + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` +} + +// NewTenderRef builds a TenderRef from contract folder and notice publication ids. +func NewTenderRef(contractFolderID, noticePublicationID string) TenderRef { + return TenderRef{ + ContractFolderID: strings.TrimSpace(contractFolderID), + NoticePublicationID: strings.TrimSpace(noticePublicationID), + } +} + // SummarizeRequest is the payload for POST /ai/summarize. type SummarizeRequest struct { ContractFolderID string `json:"contract_folder_id"` @@ -70,9 +84,22 @@ type TranslateRequest struct { ContractFolderID string `json:"contract_folder_id"` NoticePublicationID string `json:"notice_publication_id"` Language string `json:"language"` + Title string `json:"title,omitempty"` + Description string `json:"description,omitempty"` RequestSource string `json:"-"` // internal: daily_job | manual_trigger } +// NewTranslateRequest builds a TranslateRequest for POST /ai/translate. +func NewTranslateRequest(contractFolderID, noticePublicationID, language, title, description string) TranslateRequest { + return TranslateRequest{ + ContractFolderID: strings.TrimSpace(contractFolderID), + NoticePublicationID: strings.TrimSpace(noticePublicationID), + Language: strings.ToLower(strings.TrimSpace(language)), + Title: strings.TrimSpace(title), + Description: strings.TrimSpace(description), + } +} + const ( TranslationRequestSourceDailyJob = "daily_job" TranslationRequestSourceManualTrigger = "manual_trigger" @@ -122,15 +149,69 @@ type AnalyzeResponse struct { Model string `json:"model"` } -// PipelineTranslateRequest represents the request payload for POST /pipeline/translate. -type PipelineTranslateRequest struct { - Languages []string `json:"languages"` +// ScrapeDocumentsRequest is the payload for POST /scrape/documents. +type ScrapeDocumentsRequest struct { + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` } -// PipelineTranslateResponse represents the response from POST /pipeline/translate. -type PipelineTranslateResponse struct { - Status string `json:"status"` - Languages []string `json:"languages"` +// ScrapeDocumentsFile describes one downloaded document. +type ScrapeDocumentsFile struct { + Filename string `json:"filename"` + Size int64 `json:"size"` +} + +// ScrapeDocumentsResponse is returned by POST /scrape/documents. +type ScrapeDocumentsResponse struct { + Success bool `json:"success"` + FilesDownloaded int `json:"files_downloaded"` + Files []ScrapeDocumentsFile `json:"files"` +} + +// ScrapeDocumentsBatchRequest is the payload for POST /scrape/documents/batch. +type ScrapeDocumentsBatchRequest struct { + Tenders []TenderRef `json:"tenders"` +} + +// SummarizeBatchRequest is the payload for POST /ai/summarize/batch. +type SummarizeBatchRequest struct { + Tenders []TenderRef `json:"tenders"` +} + +// TranslateBatchRequest is the payload for POST /ai/translate/batch. +type TranslateBatchRequest struct { + Tenders []TenderRef `json:"tenders"` + Languages []string `json:"languages"` +} + +// BatchAcceptedResponse is returned by background batch endpoints (HTTP 202). +type BatchAcceptedResponse struct { + Status string `json:"status"` + Count int `json:"count"` +} + +// PipelineSyncResponse is returned by POST /pipeline/sync. +type PipelineSyncResponse struct { + Stored int `json:"stored"` +} + +// PipelineRunRequest is the payload for POST /pipeline/run. +type PipelineRunRequest struct { + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` + Languages []string `json:"languages"` +} + +// PipelineRunBatchRequest is the payload for POST /pipeline/run/batch. +type PipelineRunBatchRequest struct { + Tenders []TenderRef `json:"tenders"` + Languages []string `json:"languages"` +} + +// PipelineStartedResponse is returned by POST /pipeline/daily-run and POST /pipeline/auto. +type PipelineStartedResponse struct { + Status string `json:"status"` + Report string `json:"report"` } // PipelineActionResponse is returned by fire-and-forget pipeline endpoints. @@ -138,6 +219,31 @@ type PipelineActionResponse struct { Status string `json:"status"` } +// PipelineReportQuery holds optional query params for GET /pipeline/report and GET /pipeline/stats. +type PipelineReportQuery struct { + Window string `json:"window,omitempty"` +} + +// PipelineReportResponse is returned by GET /pipeline/report. +type PipelineReportResponse struct { + Window string `json:"window"` + From string `json:"from"` + To string `json:"to"` + Summary map[string]interface{} `json:"summary,omitempty"` + Days map[string]interface{} `json:"days,omitempty"` + Status string `json:"status,omitempty"` +} + +// PipelineStatsResponse is returned by GET /pipeline/stats. +type PipelineStatsResponse struct { + GeneratedAt string `json:"generated_at"` + EventLog map[string]interface{} `json:"event_log"` + Minio map[string]interface{} `json:"minio"` +} + +// PipelineMinioStatsResponse is returned by GET /pipeline/minio-stats. +type PipelineMinioStatsResponse map[string]interface{} + // OnboardingRequest is the payload for POST /onboarding. type OnboardingRequest struct { CompanyID string `json:"company_id"` diff --git a/pkg/ai_summarizer/errors.go b/pkg/ai_summarizer/errors.go index 4b42cb7..8c62888 100644 --- a/pkg/ai_summarizer/errors.go +++ b/pkg/ai_summarizer/errors.go @@ -33,4 +33,7 @@ var ( // ErrAPINonSuccess is returned when the AI API returns a non-2xx status code. ErrAPINonSuccess = errors.New("ai_summarizer: API returned non-success status") + + // ErrPipelineJobRunning is returned when a single-flight pipeline job is already running (HTTP 409). + ErrPipelineJobRunning = errors.New("ai_summarizer: pipeline job already running") )