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) } // ScrapedDocumentMetadataSyncer persists MinIO scraped-document metadata onto tender records. type ScrapedDocumentMetadataSyncer interface { SyncScrapedDocumentsFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) error } type service struct { client Client logger logger.Logger metadataSyncer ScrapedDocumentMetadataSyncer } // NewService creates an AI pipeline admin service. func NewService(client Client, log logger.Logger, metadataSyncer ScrapedDocumentMetadataSyncer) Service { return &service{ client: client, logger: log, metadataSyncer: metadataSyncer, } } 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) } if resp != nil && resp.Success && resp.FilesDownloaded > 0 { s.syncScrapedDocumentMetadata(ctx, form.ContractFolderID, form.NoticePublicationID) } 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) } s.syncScrapedDocumentMetadata(ctx, req.ContractFolderID, req.NoticePublicationID) 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 } func (s *service) syncScrapedDocumentMetadata(ctx context.Context, contractFolderID, noticePublicationID string) { if s.metadataSyncer == nil { return } if err := s.metadataSyncer.SyncScrapedDocumentsFromStorage(ctx, contractFolderID, noticePublicationID); err != nil { s.logger.Warn("Failed to sync scraped document metadata to Mongo", map[string]interface{}{ "contract_folder_id": contractFolderID, "notice_publication_id": noticePublicationID, "error": err.Error(), }) } }