From 7d383c36c359fae2eb6b1061bffd0ba6c7a064dd Mon Sep 17 00:00:00 2001 From: Mazyar Date: Mon, 8 Jun 2026 18:01:15 +0330 Subject: [PATCH] Implement AI analysis trigger endpoint and refactor related components - Added a new endpoint to trigger on-demand agentic analysis for tenders via the AI service. - Introduced `TriggerAIAnalyze` method in the `TenderHandler` to handle requests and responses for AI analysis. - Updated the `tenderService` to include `TriggerAIAnalyze` method, which validates input and interacts with the AI summarizer client. - Enhanced the `AIAnalyzeResponse` and `AIAnalyzeDocument` structures to support the new analysis feature. - Refactored the `DocumentSummarizationWorker` and `TranslationWorker` to remove deprecated MinIO dependencies, streamlining the AI service interactions. This update improves the functionality of the tender management system by allowing users to trigger AI analysis on-demand, enhancing the overall user experience and system capabilities. --- cmd/web/router/routes.go | 1 + cmd/worker/bootstrap/bootstrap.go | 115 +------------- cmd/worker/main.go | 8 +- cmd/worker/workers/summarizer.go | 251 +++++++++--------------------- cmd/worker/workers/translation.go | 145 +++++------------ internal/tender/form.go | 20 +++ internal/tender/handler.go | 32 +++- internal/tender/service.go | 187 ++++++++++++++++++---- pkg/ai_summarizer/client.go | 112 +++++++++++++ pkg/ai_summarizer/entities.go | 140 +++++++++-------- pkg/ai_summarizer/storage.go | 50 +++++- 11 files changed, 559 insertions(+), 502 deletions(-) diff --git a/cmd/web/router/routes.go b/cmd/web/router/routes.go index f76826a..78625b8 100644 --- a/cmd/web/router/routes.go +++ b/cmd/web/router/routes.go @@ -121,6 +121,7 @@ func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler tendersGP.GET("/:id/documents/download", tenderHandler.DownloadDocuments) tendersGP.GET("/:id/ai-summary", tenderHandler.GetAISummary) tendersGP.POST("/:id/ai-summarize", tenderHandler.TriggerAISummarize) + tendersGP.POST("/:id/ai-analyze", tenderHandler.TriggerAIAnalyze) tendersGP.POST("/:id/ai-translate", tenderHandler.TriggerAITranslate) } diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index eda46b9..21012c5 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -10,11 +10,9 @@ import ( ai_summarizer "tm/pkg/ai_summarizer" "tm/pkg/config" "tm/pkg/logger" - "tm/pkg/minio" "tm/pkg/mongo" "tm/pkg/notification" "tm/pkg/ollama" - "tm/pkg/queue" "tm/pkg/schedule" ) @@ -78,7 +76,7 @@ func InitMongoDB(conf config.MongoConfig, log logger.Logger) *mongo.ConnectionMa } // Init Worker -func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, minioService *minio.Service, aiClient *ai_summarizer.Client, aiStorage *ai_summarizer.StorageClient) *schedule.CronScheduler { +func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, aiClient *ai_summarizer.Client, aiStorage *ai_summarizer.StorageClient) *schedule.CronScheduler { // Debug: Log worker config appLogger.Info("Worker configuration", map[string]interface{}{ "worker_interval": config.Worker.Interval, @@ -97,70 +95,24 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger // Create a single shared cron scheduler for all recurring jobs scheduler := schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo) - var queueService queue.Queue if config.Queue.Enabled { - var err error - queueConfig := queue.QueueConfig{ - RedisURL: config.Queue.RedisURL, - MaxRetries: config.Queue.MaxRetries, - RetryBackoff: config.Queue.RetryBackoff, - JobTimeout: config.Queue.JobTimeout, - PoolSize: config.Queue.PoolSize, - IsDLQEnabled: config.Queue.IsDLQEnabled, - MaxQueueLength: 0, // Unlimited - } - - queueService, err = queue.NewRedisQueue(queueConfig, appLogger) - if err != nil { - appLogger.Error("Failed to initialize queue service", map[string]interface{}{ - "error": err.Error(), - }) - appLogger.Warn("Queue-based scraping is disabled, falling back to scheduled scraping", map[string]interface{}{}) - config.Queue.Enabled = false - } else { - appLogger.Info("Queue service initialized successfully", map[string]interface{}{ - "mode": config.Queue.Mode, - }) - } - } - - // Start Queue Producer Worker (enqueues unscraped tenders periodically) - if config.Queue.Enabled && (config.Queue.Mode == "producer" || config.Queue.Mode == "both") { - appLogger.Info("Scheduling queue producer worker", map[string]interface{}{ - "batch_size": config.Queue.ProducerBatch, - "interval": config.Queue.ProducerInterval, - }) - - scheduler.AddJob(schedule.Job{ - Name: "Queue Producer Worker Job", - Func: func() { - producer := workers.NewQueueProducerWorker( - mongoManager, - appLogger, - tenderRepo, - queueService, - config.Queue.ProducerBatch, - ) - producer.Run() - }, - Expr: config.Queue.ProducerInterval, + appLogger.Warn("Redis document-scraper queue is deprecated; document scraping is owned by the AI service pipeline", map[string]interface{}{ + "queue_mode": config.Queue.Mode, }) } - // Document summarization via external AI service (requires MinIO + AI_SUMMARIZER_API_BASE_URL) - if minioService != nil && aiClient != nil { + // Document summarization via external AI service (requires AI_SUMMARIZER_API_BASE_URL) + if aiClient != nil { scheduler.AddJob(schedule.Job{ Name: "Document Summarization Worker Job", Func: func() { - worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, minioService, aiClient) + worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, aiClient, aiStorage) worker.Run() }, - Expr: "0 11 * * * *", // Run at 11 AM every day (1 hour after document scraping) + Expr: "0 11 * * * *", // Run at 11 AM every day (after AI pipeline scrape) }) - } else if minioService != nil { - appLogger.Warn("Document summarization worker not scheduled: AI summarizer API is not configured", map[string]interface{}{}) } else { - appLogger.Warn("MinIO service not available, document summarization will be skipped", map[string]interface{}{}) + appLogger.Warn("Document summarization worker not scheduled: AI summarizer API is not configured", map[string]interface{}{}) } // Schedule Notice Worker job (no blocking run at startup; first tick runs on schedule) @@ -395,54 +347,3 @@ func InitOllamaService(conf config.OllamaConfig, log logger.Logger) *ollama.SDK return ollamaSDK } - -// Init MinIO Service -func InitMinIOService(conf config.MinIOConfig, log logger.Logger) *minio.Service { - // Convert bootstrap config to minio config - minioConfig := minio.Config{ - Endpoint: conf.Endpoint, - AccessKeyID: conf.AccessKeyID, - SecretAccessKey: conf.SecretAccessKey, - UseSSL: conf.UseSSL, - Region: conf.Region, - DefaultBucket: conf.DefaultBucket, - HierarchicalBucket: conf.HierarchicalBucket, - } - - // Validate config - if err := minioConfig.Validate(); err != nil { - log.Warn("Invalid MinIO configuration, MinIO features will be disabled", map[string]interface{}{ - "error": err.Error(), - }) - return nil // Return nil instead of panicking - } - - // Create connection manager - connManager, err := minio.NewConnectionManager(minioConfig, log) - if err != nil { - log.Error("Failed to create MinIO connection manager", map[string]interface{}{ - "error": err.Error(), - }) - panic(fmt.Sprintf("Failed to create MinIO connection manager: %v", err)) - } - - // Connect to MinIO - if err := connManager.Connect(); err != nil { - log.Warn("Failed to connect to MinIO, MinIO features will be disabled", map[string]interface{}{ - "endpoint": minioConfig.Endpoint, - "error": err.Error(), - }) - return nil // Return nil instead of panicking - } - - // Create service - service := minio.NewService(connManager, minioConfig, log) - - log.Info("MinIO service initialized successfully", map[string]interface{}{ - "endpoint": conf.Endpoint, - "default_bucket": conf.DefaultBucket, - "hierarchical_bucket": conf.HierarchicalBucket, - }) - - return service -} diff --git a/cmd/worker/main.go b/cmd/worker/main.go index a401a5e..2b60f27 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -37,18 +37,12 @@ func main() { // Initialize notification service notificationService := bootstrap.InitNotificationService(config.Notification, appLogger) - // Initialize MinIO service - minioService := bootstrap.InitMinIOService(config.MinIO, appLogger) - if minioService != nil { - appLogger.Info("MinIO service initialized successfully", map[string]interface{}{}) - } - // Initialize AI summarizer client (translation + document summarization workers) aiSummarizerClient := bootstrap.InitAISummarizerClient(config.AISummarizer, mongoManager, appLogger) aiSummarizerStorage := bootstrap.InitAISummarizerStorage(config.AISummarizer, appLogger) // Initialize Worker - scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, minioService, aiSummarizerClient, aiSummarizerStorage) + scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, aiSummarizerClient, aiSummarizerStorage) // Set up signal handling for graceful shutdown signalChan := make(chan os.Signal, 1) diff --git a/cmd/worker/workers/summarizer.go b/cmd/worker/workers/summarizer.go index 991d917..51f80f7 100644 --- a/cmd/worker/workers/summarizer.go +++ b/cmd/worker/workers/summarizer.go @@ -2,35 +2,39 @@ package workers import ( "context" - "path/filepath" "strings" "time" "tm/internal/tender" ai_summarizer "tm/pkg/ai_summarizer" "tm/pkg/logger" - "tm/pkg/minio" "tm/pkg/mongo" ) -const presignedDocumentURLSeconds int64 = 7200 - -// DocumentSummarizationWorker asks the external AI service to summarize scraped tender documents. +// DocumentSummarizationWorker triggers AI summarization for scraped tenders. +// The AI service owns MinIO artefacts; this worker marks Mongo processing flags +// after the pipeline or on-demand summarize API completes. type DocumentSummarizationWorker struct { Mongo *mongo.ConnectionManager Logger logger.Logger TenderRepo tender.TenderRepository - MinioSDK *minio.Service AIClient *ai_summarizer.Client + AIStorage *ai_summarizer.StorageClient } // NewDocumentSummarizationWorker creates a document summarization worker. -func NewDocumentSummarizationWorker(mongo *mongo.ConnectionManager, logger logger.Logger, tenderRepo tender.TenderRepository, minioSDK *minio.Service, aiClient *ai_summarizer.Client) *DocumentSummarizationWorker { +func NewDocumentSummarizationWorker( + mongo *mongo.ConnectionManager, + logger logger.Logger, + tenderRepo tender.TenderRepository, + aiClient *ai_summarizer.Client, + aiStorage *ai_summarizer.StorageClient, +) *DocumentSummarizationWorker { return &DocumentSummarizationWorker{ Mongo: mongo, Logger: logger, TenderRepo: tenderRepo, - MinioSDK: minioSDK, AIClient: aiClient, + AIStorage: aiStorage, } } @@ -42,9 +46,11 @@ func (w *DocumentSummarizationWorker) Run() { w.Logger.Warn("AI summarizer client not configured, skipping document summarization run", map[string]interface{}{}) return } - if w.MinioSDK == nil { - w.Logger.Warn("MinIO not configured, skipping document summarization run", map[string]interface{}{}) - 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(), + }) } limit := 5 @@ -54,7 +60,7 @@ func (w *DocumentSummarizationWorker) Run() { w.Logger.Error("Failed to get un-summarized tenders", map[string]interface{}{ "error": err.Error(), }) - continue + return } w.Logger.Info("Found tenders for document summarization", map[string]interface{}{ @@ -63,7 +69,6 @@ func (w *DocumentSummarizationWorker) Run() { }) if len(tenders) == 0 { - w.Logger.Info("No more tenders to process for document summarization", map[string]interface{}{}) break } @@ -72,78 +77,78 @@ func (w *DocumentSummarizationWorker) Run() { w.Logger.Info("Processing tender for document summarization", map[string]interface{}{ "tender_id": t.ID.Hex(), "notice_publication_id": t.NoticePublicationID, - "document_count": len(t.ScrapedDocuments), }) - w.summarizeDocumentsForTender(t) + w.summarizeTender(t) } } w.Logger.Info("Document summarization worker completed", map[string]interface{}{}) } -func (w *DocumentSummarizationWorker) summarizeDocumentsForTender(t *tender.Tender) { - if len(t.ScrapedDocuments) == 0 { - w.Logger.Warn("No scraped documents found for tender", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_publication_id": t.NoticePublicationID, +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{}{ + "tender_id": t.ID.Hex(), + }) + return + } + if strings.TrimSpace(t.ContractFolderID) == "" { + w.Logger.Warn("Skipping tender summarization: missing contract_folder_id", map[string]interface{}{ + "tender_id": t.ID.Hex(), }) - t.ProcessingMetadata.DocumentsSummarized = true - t.ProcessingMetadata.DocumentsSummarizedAt = time.Now().Unix() - _ = w.updateTenderSummarizationStatus(t) return } - summaries := make([]tender.DocumentSummary, 0, len(t.ScrapedDocuments)) - summarizedAt := time.Now().Unix() + ctx := context.Background() + if w.isSummaryReadyInStorage(ctx, t) { + w.markTenderSummarized(t) + return + } - for _, doc := range t.ScrapedDocuments { - summaryText, model, errStr := w.summarizeDocumentViaAI(context.Background(), t, doc) - if errStr != "" { - w.Logger.Error("Failed to summarize document", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_publication_id": t.NoticePublicationID, - "document_name": doc.Filename, - "object_name": doc.ObjectName, - "error": errStr, - }) - summaries = append(summaries, tender.DocumentSummary{ - DocumentName: doc.Filename, - ObjectName: doc.ObjectName, - BucketName: doc.BucketName, - Summary: "", - SummaryLanguage: "en", - DocumentType: w.getDocumentType(doc.Filename), - SummarizedAt: summarizedAt, - SummaryModel: model, - Error: errStr, - }) - continue - } - - summaries = append(summaries, tender.DocumentSummary{ - DocumentName: doc.Filename, - ObjectName: doc.ObjectName, - BucketName: doc.BucketName, - Summary: summaryText, - SummaryLanguage: "en", - DocumentType: w.getDocumentType(doc.Filename), - SummarizedAt: summarizedAt, - SummaryModel: model, - }) - - w.Logger.Info("Document summarized successfully", map[string]interface{}{ + req := ai_summarizer.NewSummarizeRequest(t.ContractFolderID, t.NoticePublicationID) + if _, err := w.AIClient.FetchSummaryOnDemand(ctx, req); err != nil { + w.Logger.Error("Failed to summarize tender via AI service", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "notice_publication_id": t.NoticePublicationID, + "contract_folder_id": t.ContractFolderID, + "error": err.Error(), + }) + return + } + + if !w.isSummaryReadyInStorage(ctx, t) { + w.Logger.Warn("AI summarize completed but storage is not ready yet", map[string]interface{}{ "tender_id": t.ID.Hex(), "notice_publication_id": t.NoticePublicationID, - "document_name": doc.Filename, - "summary_length": len(summaryText), }) } - t.DocumentSummaries = summaries - t.ProcessingMetadata.DocumentsSummarized = true - t.ProcessingMetadata.DocumentsSummarizedAt = summarizedAt + w.markTenderSummarized(t) +} - if err := w.updateTenderSummarizationStatus(t); err != nil { +func (w *DocumentSummarizationWorker) isSummaryReadyInStorage(ctx context.Context, t *tender.Tender) bool { + if w.AIStorage == nil { + return false + } + ready, err := w.AIStorage.IsSummaryReady(ctx, t.ContractFolderID, t.NoticePublicationID) + if err != nil { + w.Logger.Debug("Could not check summarization status in storage", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "notice_publication_id": t.NoticePublicationID, + "error": err.Error(), + }) + return false + } + return ready +} + +func (w *DocumentSummarizationWorker) markTenderSummarized(t *tender.Tender) { + now := time.Now().Unix() + t.ProcessingMetadata.DocumentsSummarized = true + t.ProcessingMetadata.DocumentsSummarizedAt = now + t.UpdatedAt = now + + if err := w.TenderRepo.Update(context.Background(), t); err != nil { w.Logger.Error("Failed to update tender summarization status", map[string]interface{}{ "tender_id": t.ID.Hex(), "notice_publication_id": t.NoticePublicationID, @@ -152,118 +157,8 @@ func (w *DocumentSummarizationWorker) summarizeDocumentsForTender(t *tender.Tend return } - w.Logger.Info("Document summarization completed for tender", map[string]interface{}{ + w.Logger.Info("Tender marked as summarized", map[string]interface{}{ "tender_id": t.ID.Hex(), "notice_publication_id": t.NoticePublicationID, - "summarized_count": len(summaries), - "total_documents": len(t.ScrapedDocuments), }) } - -func (w *DocumentSummarizationWorker) summarizeDocumentViaAI(ctx context.Context, t *tender.Tender, doc tender.ScrapedDocument) (summaryText, model, errStr string) { - docURL, err := w.MinioSDK.GetPresignedURL(ctx, doc.BucketName, doc.ObjectName, presignedDocumentURLSeconds) - if err != nil { - return "", "", err.Error() - } - - if strings.TrimSpace(t.NoticePublicationID) == "" { - return "", "", "tender has no notice_publication_id for AI summarize request" - } - if strings.TrimSpace(t.ContractFolderID) == "" { - return "", "", "tender has no contract_folder_id for AI summarize request" - } - - req := ai_summarizer.NewSummarizeRequest(ai_summarizer.SummarizeRequestInput{ - ContractFolderID: t.ContractFolderID, - NoticePublicationID: t.NoticePublicationID, - DocumentURL: docURL, - Title: t.Title, - Description: t.Description, - Country: t.CountryCode, - Currency: t.Currency, - SubmissionDeadline: t.SubmissionDeadline, - EstimatedValue: t.EstimatedValue, - AuthorityName: authorityNameFromTender(t), - }) - - resp, err := w.AIClient.FetchSummaryOnDemand(ctx, req) - if err != nil { - return "", "", err.Error() - } - if resp == nil { - return "", "", "AI summarize returned nil response" - } - - summaryText, model = pickDocumentSummary(resp.Summaries, doc.Filename) - if summaryText == "" && len(resp.Summaries) > 0 { - // Fall back to first summary if filename did not match - summaryText = resp.Summaries[0].Summary - model = resp.Summaries[0].SummaryModel - } - if summaryText == "" { - return "", model, "AI summarize returned no summary text for document" - } - return summaryText, model, "" -} - -func authorityNameFromTender(t *tender.Tender) string { - if t == nil || t.BuyerOrganization == nil { - return "" - } - return strings.TrimSpace(t.BuyerOrganization.Name) -} - -func pickDocumentSummary(items []ai_summarizer.DocumentSummary, filename string) (text, model string) { - filename = strings.TrimSpace(filename) - for _, s := range items { - if strings.EqualFold(strings.TrimSpace(s.DocumentName), filename) { - return strings.TrimSpace(s.Summary), strings.TrimSpace(s.SummaryModel) - } - } - for _, s := range items { - if strings.TrimSpace(s.Summary) != "" { - return strings.TrimSpace(s.Summary), strings.TrimSpace(s.SummaryModel) - } - } - return "", "" -} - -func (w *DocumentSummarizationWorker) getDocumentType(filename string) string { - ext := strings.ToLower(filepath.Ext(filename)) - switch ext { - case ".xlsx": - return "xlsx" - case ".pdf": - return "pdf" - case ".docx": - return "docx" - case ".doc": - return "doc" - case ".txt": - return "txt" - case ".html", ".htm": - return "html" - default: - return "unknown" - } -} - -func (w *DocumentSummarizationWorker) updateTenderSummarizationStatus(tender *tender.Tender) error { - err := w.TenderRepo.Update(context.Background(), tender) - if err != nil { - w.Logger.Error("Failed to update tender summarization status", map[string]interface{}{ - "tender_id": tender.ID.Hex(), - "error": err.Error(), - }) - return err - } - - w.Logger.Debug("Updated tender summarization status", map[string]interface{}{ - "tender_id": tender.ID.Hex(), - "documents_summarized": tender.ProcessingMetadata.DocumentsSummarized, - "documents_summarized_at": tender.ProcessingMetadata.DocumentsSummarizedAt, - "summary_count": len(tender.DocumentSummaries), - }) - - return nil -} diff --git a/cmd/worker/workers/translation.go b/cmd/worker/workers/translation.go index bb8e6a5..9f94f63 100644 --- a/cmd/worker/workers/translation.go +++ b/cmd/worker/workers/translation.go @@ -2,7 +2,6 @@ package workers import ( "context" - "errors" "strings" "tm/internal/tender" "tm/pkg/ai_summarizer" @@ -10,17 +9,18 @@ import ( "tm/pkg/mongo" ) -// TranslationWorker backfills missing tender translations using the AI pipeline -// and on-demand translate API. Results are persisted by the AI service in MinIO only. +// TranslationWorker syncs translation completion from the AI pipeline MinIO storage. +// Batch translation is triggered via POST /pipeline/translate; this worker only +// verifies which tenders now have translations available in storage. type TranslationWorker struct { - Mongo *mongo.ConnectionManager - Logger logger.Logger - TenderRepo tender.TenderRepository - AIClient *ai_summarizer.Client - AIStorage *ai_summarizer.StorageClient + Mongo *mongo.ConnectionManager + Logger logger.Logger + TenderRepo tender.TenderRepository + AIClient *ai_summarizer.Client + AIStorage *ai_summarizer.StorageClient TranslationSuccessCounter *mongo.Counter - TargetLanguages []string - BatchSize int + TargetLanguages []string + BatchSize int } // NewTranslationWorker creates a translation worker for the given target languages. @@ -66,7 +66,7 @@ func NewTranslationWorker( } } -// Run starts backfilling translations for tenders missing target-language content. +// Run triggers the AI translation pipeline and logs tenders with ready translations in MinIO. func (w *TranslationWorker) Run() { if w.AIClient == nil { w.Logger.Warn("Translation worker skipped: AI client is not configured", map[string]interface{}{ @@ -83,14 +83,15 @@ 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 (continuing with per-tender sync)", map[string]interface{}{ + w.Logger.Warn("Failed to trigger AI pipeline translate", map[string]interface{}{ "target_languages": w.TargetLanguages, "error": err.Error(), }) } + readyCount := 0 for _, language := range w.TargetLanguages { - w.runForLanguage(language) + readyCount += w.syncLanguageFromStorage(language) } endCount := w.getSuccessfulTranslationCount(mongo.AITranslationSuccessCounterKeyDailyJob) @@ -101,10 +102,11 @@ func (w *TranslationWorker) Run() { } w.Logger.Info("Translation worker completed", map[string]interface{}{ - "target_languages": w.TargetLanguages, - "successful_ai_translation_requests_run": runCount, - "successful_ai_translation_requests_daily_job": endCount, - "successful_ai_translation_requests_total": totalCount, + "target_languages": w.TargetLanguages, + "translations_ready_in_storage": readyCount, + "successful_ai_translation_requests_run": runCount, + "successful_ai_translation_requests_daily_job": endCount, + "successful_ai_translation_requests_total": totalCount, }) } @@ -123,19 +125,27 @@ func (w *TranslationWorker) getSuccessfulTranslationCount(key string) int64 { return count } -func (w *TranslationWorker) runForLanguage(language string) { +func (w *TranslationWorker) syncLanguageFromStorage(language string) int { + if w.AIStorage == nil { + w.Logger.Warn("Translation storage client not configured; skipping MinIO sync", map[string]interface{}{ + "target_language": language, + }) + return 0 + } + + readyCount := 0 skip := 0 for { tenders, totalCount, err := w.TenderRepo.GetUnTranslatedTenders(context.Background(), language, w.BatchSize, skip) if err != nil { - w.Logger.Error("Failed to fetch tenders for translation backfill", map[string]interface{}{ + w.Logger.Error("Failed to fetch tenders for translation sync", map[string]interface{}{ "target_language": language, "error": err.Error(), }) - return + return readyCount } - w.Logger.Info("Fetched tenders for translation backfill", map[string]interface{}{ + w.Logger.Info("Fetched tenders for translation sync", map[string]interface{}{ "target_language": language, "batch_count": len(tenders), "total_count": totalCount, @@ -143,63 +153,29 @@ func (w *TranslationWorker) runForLanguage(language string) { }) if len(tenders) == 0 { - return + return readyCount } for _, t := range tenders { - w.translateTender(&t, language) + if w.hasTranslationInStorage(&t, language) { + readyCount++ + w.Logger.Debug("Translation available in MinIO", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "notice_id": t.NoticePublicationID, + "target_language": language, + }) + } } skip += len(tenders) if int64(skip) >= totalCount { - return + return readyCount } } } -func (w *TranslationWorker) translateTender(t *tender.Tender, language string) { - if t.NoticePublicationID == "" { - return - } - if strings.TrimSpace(t.ContractFolderID) == "" { - w.Logger.Debug("Skipping tender translation: missing contract_folder_id", map[string]interface{}{ - "tender_id": t.ID.Hex(), - }) - return - } - if w.hasTranslationInStorage(t, language) { - w.Logger.Debug("Skipping tender translation: already available in MinIO", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_id": t.NoticePublicationID, - "target_language": language, - }) - return - } - - title, description, source, err := w.resolveTranslation(t, language) - if err != nil { - w.Logger.Error("Failed to translate tender", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_id": t.NoticePublicationID, - "target_language": language, - "translation_source": source, - "translation_error": err.Error(), - }) - return - } - - w.Logger.Info("Tender translated successfully", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_id": t.NoticePublicationID, - "target_language": language, - "translation_source": source, - "title_len": len(title), - "description_len": len(description), - }) -} - func (w *TranslationWorker) hasTranslationInStorage(t *tender.Tender, language string) bool { - if w.AIStorage == nil { + if strings.TrimSpace(t.ContractFolderID) == "" || t.NoticePublicationID == "" { return false } stored, err := w.AIStorage.GetTranslationFromStorage( @@ -210,40 +186,3 @@ func (w *TranslationWorker) hasTranslationInStorage(t *tender.Tender, language s ) return err == nil && strings.TrimSpace(stored.Title) != "" } - -func (w *TranslationWorker) resolveTranslation(t *tender.Tender, language string) (title, description, source string, err error) { - if w.AIStorage != nil { - stored, storageErr := w.AIStorage.GetTranslationFromStorage( - context.Background(), - t.ContractFolderID, - t.NoticePublicationID, - language, - ) - if storageErr == nil { - return stored.Title, stored.Description, "storage", nil - } - if !errors.Is(storageErr, ai_summarizer.ErrTranslationNotReady) && - !errors.Is(storageErr, ai_summarizer.ErrObjectNotFound) && - !errors.Is(storageErr, ai_summarizer.ErrNoTranslation) { - w.Logger.Debug("Translation not available from storage yet", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_id": t.NoticePublicationID, - "target_language": language, - "error": storageErr.Error(), - }) - } - } - - resp, apiErr := w.AIClient.FetchTranslationOnDemand(context.Background(), ai_summarizer.TranslateRequest{ - ContractFolderID: t.ContractFolderID, - NoticePublicationID: t.NoticePublicationID, - Title: t.Title, - Description: t.Description, - Language: language, - RequestSource: ai_summarizer.TranslationRequestSourceDailyJob, - }) - if apiErr != nil { - return "", "", "api", apiErr - } - return resp.TranslatedTitle, resp.TranslatedDescription, "api", nil -} diff --git a/internal/tender/form.go b/internal/tender/form.go index 4a2bfaa..8cd8f23 100644 --- a/internal/tender/form.go +++ b/internal/tender/form.go @@ -573,3 +573,23 @@ type AIDocumentSummary struct { SummaryModel string `json:"summary_model"` Error string `json:"error,omitempty"` } + +// AIAnalyzeResponse represents the response for on-demand agentic analysis. +type AIAnalyzeResponse struct { + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` + Summary string `json:"summary"` + Documents []AIAnalyzeDocument `json:"documents"` + ToolCallsLog []string `json:"tool_calls_log"` + Iterations int `json:"iterations"` + Model string `json:"model"` +} + +// AIAnalyzeDocument describes one document referenced during analysis. +type AIAnalyzeDocument struct { + Filename string `json:"filename"` + SizeBytes int64 `json:"size_bytes"` + TimesRead int `json:"times_read"` + TimesQueried int `json:"times_queried"` + Description string `json:"description"` +} diff --git a/internal/tender/handler.go b/internal/tender/handler.go index 83863de..c239f1f 100644 --- a/internal/tender/handler.go +++ b/internal/tender/handler.go @@ -398,7 +398,7 @@ func (h *TenderHandler) GetPublicTenderDetails(c echo.Context) error { // GetDocumentSummaries retrieves document summaries for a tender // @Summary Get document summaries for a tender -// @Description Retrieve AI-generated summaries of documents for a specific tender +// @Description Retrieve per-document AI summaries from the AI service MinIO storage for a specific tender // @Tags Tenders // @Produce json // @Param id path string true "Tender ID" @@ -595,6 +595,36 @@ func (h *TenderHandler) TriggerAISummarize(c echo.Context) error { return response.Success(c, result, "AI summarization completed successfully") } +// TriggerAIAnalyze triggers on-demand agentic analysis for a tender +// @Summary Trigger AI analysis +// @Description Trigger on-demand agentic document analysis for a tender via the external AI service +// @Tags Admin-Tenders +// @Produce json +// @Param id path string true "Tender ID" +// @Success 200 {object} response.APIResponse{data=AIAnalyzeResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 404 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Router /admin/v1/tenders/{id}/ai-analyze [post] +// @Security BearerAuth +func (h *TenderHandler) TriggerAIAnalyze(c echo.Context) error { + id := c.Param("id") + if id == "" { + return response.BadRequest(c, "Tender ID is required", "ID parameter cannot be empty") + } + + result, err := h.service.TriggerAIAnalyze(c.Request().Context(), id) + if err != nil { + h.logger.Error("Failed to trigger AI analysis", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return response.InternalServerError(c, err.Error()) + } + + return response.Success(c, result, "AI analysis completed successfully") +} + // TriggerAITranslate triggers on-demand AI translation for a tender // @Summary Trigger AI translation // @Description Trigger on-demand AI translation for tender title and description (cached in MinIO by the AI service) diff --git a/internal/tender/service.go b/internal/tender/service.go index a3e7c02..44cd986 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -18,10 +18,11 @@ import ( "tm/pkg/response" ) -// AISummarizerClient defines the interface for on-demand AI summarization +// AISummarizerClient defines the interface for on-demand AI operations. 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) } @@ -30,6 +31,7 @@ type AISummarizerClient interface { // PROC_// in the shared bucket. type AISummarizerStorage interface { GetSummaryFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) (string, error) + GetDocumentSummariesFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) ([]ai_summarizer.DocumentSummary, error) GetTranslationFromStorage(ctx context.Context, contractFolderID, noticePublicationID, language string) (ai_summarizer.StoredTranslation, error) ListDocuments(ctx context.Context, contractFolderID, noticePublicationID string) ([]ai_summarizer.StoredDocument, error) DownloadDocument(ctx context.Context, objectName string) (io.ReadCloser, error) @@ -60,6 +62,7 @@ type Service interface { // AI summarization operations GetAISummary(ctx context.Context, id string) (*AISummaryResponse, error) TriggerAISummarize(ctx context.Context, id string) (*AISummarizeResponse, error) + TriggerAIAnalyze(ctx context.Context, id string) (*AIAnalyzeResponse, error) TriggerAITranslate(ctx context.Context, id, language string) (*AITranslateResponse, error) GetAllAISummaries(ctx context.Context) ([]ai_summarizer.TenderSummaryItem, error) // ListTendersWithScrapedDocuments lists tenders that have documents in MinIO. When syncMongoMetadata is true, @@ -451,14 +454,13 @@ func (s *tenderService) Delete(ctx context.Context, id string) error { return nil } -// GetDocumentSummaries retrieves document summaries for a specific tender +// GetDocumentSummaries retrieves per-document summaries from the AI service MinIO storage. func (s *tenderService) GetDocumentSummaries(ctx context.Context, id string) ([]DocumentSummary, error) { if id == "" { return nil, fmt.Errorf("tender ID cannot be empty") } - // Get tender from repository - tender, err := s.repository.GetByID(ctx, id) + t, err := s.repository.GetByID(ctx, id) if err != nil { s.logger.Error("Failed to get tender for document summaries", map[string]interface{}{ "tender_id": id, @@ -466,22 +468,60 @@ func (s *tenderService) GetDocumentSummaries(ctx context.Context, id string) ([] }) return nil, fmt.Errorf("failed to get tender: %w", err) } - - if tender == nil { + if t == nil { return nil, fmt.Errorf("tender not found") } - - // Return document summaries (empty slice if none exist) - if tender.DocumentSummaries == nil { + if strings.TrimSpace(t.NoticePublicationID) == "" || strings.TrimSpace(t.ContractFolderID) == "" { + return []DocumentSummary{}, nil + } + if s.aiSummarizerStorage == nil { return []DocumentSummary{}, nil } + stored, err := s.aiSummarizerStorage.GetDocumentSummariesFromStorage(ctx, t.ContractFolderID, t.NoticePublicationID) + if errors.Is(err, ai_summarizer.ErrSummaryNotReady) { + return []DocumentSummary{}, nil + } + if errors.Is(err, ai_summarizer.ErrObjectNotFound) { + return []DocumentSummary{}, nil + } + if err != nil { + if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) { + return nil, fmt.Errorf("failed to get document summaries: MinIO connection unavailable: %w", err) + } + s.logger.Error("Failed to get document summaries from storage", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to get document summaries: %w", err) + } + + summaries := mapAIDocumentSummaries(stored) s.logger.Info("Retrieved document summaries for tender", map[string]interface{}{ "tender_id": id, - "summaries_count": len(tender.DocumentSummaries), + "summaries_count": len(summaries), + "source": "storage", }) - return tender.DocumentSummaries, nil + return summaries, nil +} + +func mapAIDocumentSummaries(items []ai_summarizer.DocumentSummary) []DocumentSummary { + if len(items) == 0 { + return []DocumentSummary{} + } + out := make([]DocumentSummary, 0, len(items)) + for _, item := range items { + out = append(out, DocumentSummary{ + DocumentName: item.DocumentName, + Summary: item.Summary, + SummaryLanguage: "en", + DocumentType: item.DocumentType, + SummaryModel: item.SummaryModel, + Error: item.Error, + }) + } + return out } // GetDocuments retrieves scraped document metadata for a tender. @@ -1175,23 +1215,7 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS return nil, fmt.Errorf("tender has no contract folder ID") } - authorityName := "" - if t.BuyerOrganization != nil { - authorityName = t.BuyerOrganization.Name - } - - reqBody := ai_summarizer.NewSummarizeRequest(ai_summarizer.SummarizeRequestInput{ - ContractFolderID: t.ContractFolderID, - NoticePublicationID: t.NoticePublicationID, - DocumentURL: t.DocumentURI, - Title: t.Title, - Description: t.Description, - Country: t.CountryCode, - Currency: t.Currency, - SubmissionDeadline: t.SubmissionDeadline, - EstimatedValue: t.EstimatedValue, - AuthorityName: authorityName, - }) + reqBody := ai_summarizer.NewSummarizeRequest(t.ContractFolderID, t.NoticePublicationID) s.logger.Info("Triggering on-demand AI summarization", map[string]interface{}{ "tender_id": id, @@ -1247,6 +1271,89 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS }, nil } +// TriggerAIAnalyze triggers on-demand agentic analysis for a tender. +func (s *tenderService) TriggerAIAnalyze(ctx context.Context, id string) (*AIAnalyzeResponse, error) { + if id == "" { + return nil, fmt.Errorf("tender ID cannot be empty") + } + if s.aiSummarizerClient == nil { + return nil, fmt.Errorf("AI summarizer service is not configured") + } + + t, err := s.repository.GetByID(ctx, id) + if err != nil { + s.logger.Error("Failed to get tender for on-demand analysis", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to get tender: %w", err) + } + if t == nil { + return nil, fmt.Errorf("tender not found") + } + if t.NoticePublicationID == "" { + return nil, fmt.Errorf("tender has no notice publication ID") + } + if strings.TrimSpace(t.ContractFolderID) == "" { + return nil, fmt.Errorf("tender has no contract folder ID") + } + + reqBody := ai_summarizer.NewAnalyzeRequest(t.ContractFolderID, t.NoticePublicationID) + s.logger.Info("Triggering on-demand AI analysis", map[string]interface{}{ + "tender_id": id, + "contract_folder_id": t.ContractFolderID, + "notice_publication_id": t.NoticePublicationID, + }) + + resp, err := s.aiSummarizerClient.FetchAnalyzeOnDemand(ctx, reqBody) + if err != nil { + s.logger.Error("On-demand AI analysis failed", map[string]interface{}{ + "tender_id": id, + "contract_folder_id": t.ContractFolderID, + "notice_publication_id": t.NoticePublicationID, + "error": err.Error(), + }) + return nil, fmt.Errorf("AI analysis request failed: %w", err) + } + + documents := make([]AIAnalyzeDocument, 0, len(resp.Documents)) + for _, doc := range resp.Documents { + documents = append(documents, AIAnalyzeDocument{ + Filename: doc.Filename, + SizeBytes: doc.SizeBytes, + TimesRead: doc.TimesRead, + TimesQueried: doc.TimesQueried, + Description: doc.Description, + }) + } + + noticePublicationID := strings.TrimSpace(resp.NoticePublicationID) + if noticePublicationID == "" { + noticePublicationID = t.NoticePublicationID + } + contractFolderID := strings.TrimSpace(resp.ContractFolderID) + if contractFolderID == "" { + contractFolderID = t.ContractFolderID + } + + s.logger.Info("On-demand AI analysis completed", map[string]interface{}{ + "tender_id": id, + "contract_folder_id": contractFolderID, + "notice_publication_id": noticePublicationID, + "iterations": resp.Iterations, + }) + + return &AIAnalyzeResponse{ + ContractFolderID: contractFolderID, + NoticePublicationID: noticePublicationID, + Summary: resp.Summary, + Documents: documents, + ToolCallsLog: resp.ToolCallsLog, + Iterations: resp.Iterations, + Model: resp.Model, + }, nil +} + // GetAllAISummaries retrieves all AI-generated summaries from the MinIO storage. // It lists all tender.json files in the bucket and returns summaries for tenders // where the AI pipeline has completed processing. @@ -1510,11 +1617,12 @@ func (s *tenderService) buildAITranslateResponse(t *Tender, language, title, des func (s *tenderService) resolveTranslation(ctx context.Context, t *Tender, language, requestSource string) (string, string, bool, error) { if s.aiSummarizerStorage != nil { - stored, err := s.aiSummarizerStorage.GetTranslationFromStorage(ctx, t.ContractFolderID, t.NoticePublicationID, language) - if err == nil { + stored, _, err := s.lookupTranslationForTender(ctx, t, language) + if err == nil && strings.TrimSpace(stored.Title) != "" { return stored.Title, stored.Description, false, nil } - if !errors.Is(err, ai_summarizer.ErrTranslationNotReady) && + if err != nil && + !errors.Is(err, ai_summarizer.ErrTranslationNotReady) && !errors.Is(err, ai_summarizer.ErrObjectNotFound) && !errors.Is(err, ai_summarizer.ErrNoTranslation) { s.logger.Debug("Translation not available from storage", map[string]interface{}{ @@ -1528,15 +1636,26 @@ 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, - Title: t.Title, - Description: t.Description, Language: language, RequestSource: requestSource, }) if err != nil { return "", "", false, err } - return resp.TranslatedTitle, resp.TranslatedDescription, true, nil + + title := strings.TrimSpace(resp.TranslatedTitle) + description := resp.TranslatedDescription + if s.aiSummarizerStorage != nil { + if stored, _, storageErr := s.lookupTranslationForTender(ctx, t, language); storageErr == nil && strings.TrimSpace(stored.Title) != "" { + title = stored.Title + description = stored.Description + } + } + if title == "" { + title = strings.TrimSpace(resp.TranslatedTitle) + } + + return title, description, true, nil } func (s *tenderService) pickResponseLanguage(language *string) string { diff --git a/pkg/ai_summarizer/client.go b/pkg/ai_summarizer/client.go index 27ed873..fc1a90d 100644 --- a/pkg/ai_summarizer/client.go +++ b/pkg/ai_summarizer/client.go @@ -169,6 +169,118 @@ func (c *Client) TriggerPipelineTranslate(ctx context.Context, languages []strin return &result, 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) + if err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to marshal analyze request body: %w", err) + } + + url := c.config.APIBaseURL + "/ai/analyze" + 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 AnalyzeResponse + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode analyze response JSON: %w", err) + } + + c.logger.Info("AI analyze request succeeded", map[string]interface{}{ + "contract_folder_id": result.ContractFolderID, + "notice_publication_id": result.NoticePublicationID, + "iterations": result.Iterations, + }) + + return &result, nil +} + +// TriggerPipelineSync calls POST /pipeline/sync. +func (c *Client) TriggerPipelineSync(ctx context.Context) (*PipelineActionResponse, error) { + return c.triggerPipelineAction(ctx, "/pipeline/sync", "pipeline sync") +} + +// TriggerPipelineScrape calls POST /pipeline/scrape. +func (c *Client) TriggerPipelineScrape(ctx context.Context) (*PipelineActionResponse, error) { + return c.triggerPipelineAction(ctx, "/pipeline/scrape", "pipeline scrape") +} + +// TriggerPipelineSummarize calls POST /pipeline/summarize. +func (c *Client) TriggerPipelineSummarize(ctx context.Context) (*PipelineActionResponse, error) { + return c.triggerPipelineAction(ctx, "/pipeline/summarize", "pipeline summarize") +} + +// TriggerPipelineAnalyze calls POST /pipeline/analyze. +func (c *Client) TriggerPipelineAnalyze(ctx context.Context) (*PipelineActionResponse, error) { + return c.triggerPipelineAction(ctx, "/pipeline/analyze", "pipeline analyze") +} + +// TriggerPipelineDailyRun calls POST /pipeline/daily-run. +func (c *Client) TriggerPipelineDailyRun(ctx context.Context) (*PipelineActionResponse, error) { + return c.triggerPipelineAction(ctx, "/pipeline/daily-run", "pipeline daily-run") +} + +func (c *Client) triggerPipelineAction(ctx context.Context, path, label string) (*PipelineActionResponse, 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 >= 400 { + return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + } + + var result PipelineActionResponse + 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 action triggered", map[string]interface{}{ + "path": path, + "status": result.Status, + }) + + return &result, nil +} + +func (c *Client) doRawPost(ctx context.Context, url string, jsonBody []byte) (*http.Response, []byte, error) { + var body io.Reader + if len(jsonBody) > 0 { + body = bytes.NewReader(jsonBody) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, body) + if err != nil { + return nil, nil, fmt.Errorf("ai_summarizer: failed to create 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, 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 +} + // doPost performs a single POST request and parses the response. func (c *Client) doPost(ctx context.Context, url string, jsonBody []byte) (*SummarizeResponse, error) { req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(jsonBody)) diff --git a/pkg/ai_summarizer/entities.go b/pkg/ai_summarizer/entities.go index 9b04ee0..430d04b 100644 --- a/pkg/ai_summarizer/entities.go +++ b/pkg/ai_summarizer/entities.go @@ -2,58 +2,20 @@ package ai_summarizer import ( "strings" - "time" ) -// SummarizeRequest represents the request payload sent to the AI service -// POST /ai/summarize endpoint for on-demand summarization. +// SummarizeRequest is the payload for POST /ai/summarize. type SummarizeRequest struct { - ContractFolderID string `json:"contract_folder_id"` - NoticePublicationID string `json:"notice_publication_id"` - DocumentURL string `json:"document_url,omitempty"` - Title string `json:"title,omitempty"` - Description string `json:"description,omitempty"` - Country string `json:"country,omitempty"` - EstimatedValue *float64 `json:"estimated_value,omitempty"` - Currency string `json:"currency,omitempty"` - SubmissionDeadline string `json:"submission_deadline,omitempty"` - AuthorityName string `json:"authority_name,omitempty"` -} - -// SummarizeRequestInput collects fields used to build SummarizeRequest. -type SummarizeRequestInput struct { - ContractFolderID string - NoticePublicationID string - DocumentURL string - Title string - Description string - Country string - Currency string - SubmissionDeadline int64 // Unix seconds; omitted from request when zero - EstimatedValue float64 - AuthorityName string + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` } // NewSummarizeRequest builds a SummarizeRequest for POST /ai/summarize. -func NewSummarizeRequest(in SummarizeRequestInput) SummarizeRequest { - req := SummarizeRequest{ - ContractFolderID: strings.TrimSpace(in.ContractFolderID), - NoticePublicationID: strings.TrimSpace(in.NoticePublicationID), - DocumentURL: strings.TrimSpace(in.DocumentURL), - Title: in.Title, - Description: in.Description, - Country: in.Country, - Currency: in.Currency, - AuthorityName: strings.TrimSpace(in.AuthorityName), +func NewSummarizeRequest(contractFolderID, noticePublicationID string) SummarizeRequest { + return SummarizeRequest{ + ContractFolderID: strings.TrimSpace(contractFolderID), + NoticePublicationID: strings.TrimSpace(noticePublicationID), } - if in.EstimatedValue > 0 { - v := in.EstimatedValue - req.EstimatedValue = &v - } - if in.SubmissionDeadline > 0 { - req.SubmissionDeadline = time.Unix(in.SubmissionDeadline, 0).UTC().Format(time.RFC3339) - } - return req } // SummarizeResponse represents the response from the AI summarization API. @@ -63,6 +25,8 @@ type SummarizeResponse struct { NoticeID string `json:"notice_id"` // legacy field name Summaries []DocumentSummary `json:"summaries"` OverallSummary *string `json:"overall_summary"` + RollupS float64 `json:"rollup_s"` + TotalS float64 `json:"total_s"` } // ResolvedNoticePublicationID returns the notice publication ID from the response. @@ -82,7 +46,7 @@ type DocumentSummary struct { DocumentType string `json:"document_type"` Summary string `json:"summary"` SummaryModel string `json:"summary_model"` - Error string `json:"error"` // empty string when no error + Error string `json:"error"` // empty when no error } // ProcedureDocumentsSummary reports scraped documents stored under one procedure folder in MinIO. @@ -101,13 +65,10 @@ type StoredDocument struct { DocumentType string `json:"document_type,omitempty"` } -// TranslateRequest represents the request payload sent to the AI service -// POST /ai/translate endpoint. +// TranslateRequest is the payload for POST /ai/translate. type TranslateRequest struct { ContractFolderID string `json:"contract_folder_id"` NoticePublicationID string `json:"notice_publication_id"` - Title string `json:"title"` - Description string `json:"description"` Language string `json:"language"` RequestSource string `json:"-"` // internal: daily_job | manual_trigger } @@ -126,6 +87,41 @@ type TranslateResponse struct { TranslatedDescription string `json:"translated_description"` } +// AnalyzeRequest is the payload for POST /ai/analyze. +type AnalyzeRequest struct { + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` +} + +// NewAnalyzeRequest builds an AnalyzeRequest for POST /ai/analyze. +func NewAnalyzeRequest(contractFolderID, noticePublicationID string) AnalyzeRequest { + return AnalyzeRequest{ + ContractFolderID: strings.TrimSpace(contractFolderID), + NoticePublicationID: strings.TrimSpace(noticePublicationID), + } +} + +// AnalyzeDocumentStats describes one document touched during agentic analysis. +type AnalyzeDocumentStats struct { + Filename string `json:"filename"` + SizeBytes int64 `json:"size_bytes"` + TimesRead int `json:"times_read"` + TimesQueried int `json:"times_queried"` + Description string `json:"description"` +} + +// AnalyzeResponse is the payload from POST /ai/analyze. +type AnalyzeResponse struct { + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` + Summary string `json:"summary"` + Documents []AnalyzeDocumentStats `json:"documents"` + Scratchpad map[string]interface{} `json:"scratchpad"` + ToolCallsLog []string `json:"tool_calls_log"` + Iterations int `json:"iterations"` + Model string `json:"model"` +} + // PipelineTranslateRequest represents the request payload for POST /pipeline/translate. type PipelineTranslateRequest struct { Languages []string `json:"languages"` @@ -137,6 +133,11 @@ type PipelineTranslateResponse struct { Languages []string `json:"languages"` } +// PipelineActionResponse is returned by fire-and-forget pipeline endpoints. +type PipelineActionResponse struct { + Status string `json:"status"` +} + // StoredTranslation is one language entry inside tender.json translations map. type StoredTranslation struct { Title string `json:"title"` @@ -173,6 +174,7 @@ type TenderJSON struct { Summary string `json:"summary"` OverallSummary *string `json:"overall_summary"` OverallSummaryCamel *string `json:"overallSummary"` + Summaries []DocumentSummary `json:"summaries"` SummarizeStatus SummarizeStatus `json:"summarize_status"` SummarizeStatusCamel SummarizeStatus `json:"summarizeStatus"` Translations map[string]StoredTranslation `json:"translations"` @@ -225,23 +227,31 @@ func (t *TenderJSON) translationsDone() []string { return t.TranslationStatusCamel.languagesDone() } -// translationForLanguage returns stored translation text when the pipeline marked -// the language done and non-empty title exists. +// storedTranslationText returns translation text from the translations map when +// title is non-empty. It does not require translation_status.done — on-demand +// writes may persist translations before updating status. +func (t *TenderJSON) storedTranslationText(language string) (StoredTranslation, bool) { + if t == nil || t.Translations == nil { + return StoredTranslation{}, false + } + lang := strings.ToLower(strings.TrimSpace(language)) + for key, entry := range t.Translations { + if strings.ToLower(strings.TrimSpace(key)) != lang { + continue + } + if strings.TrimSpace(entry.Title) == "" { + continue + } + return entry, true + } + return StoredTranslation{}, false +} + +// translationForLanguage returns stored translation when the pipeline marked the +// language done and non-empty title exists. func (t *TenderJSON) translationForLanguage(language string) (StoredTranslation, bool) { if t == nil || !languageInDone(t.translationsDone(), language) { return StoredTranslation{}, false } - lang := strings.ToLower(strings.TrimSpace(language)) - if t.Translations != nil { - for key, entry := range t.Translations { - if strings.ToLower(strings.TrimSpace(key)) != lang { - continue - } - if strings.TrimSpace(entry.Title) == "" { - continue - } - return entry, true - } - } - return StoredTranslation{}, false + return t.storedTranslationText(language) } diff --git a/pkg/ai_summarizer/storage.go b/pkg/ai_summarizer/storage.go index c0bfa5f..2dd75e4 100644 --- a/pkg/ai_summarizer/storage.go +++ b/pkg/ai_summarizer/storage.go @@ -326,8 +326,9 @@ func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolde return s.getSummaryFromObjectKey(ctx, prefix+"tender.json", contractFolderID, noticePublicationID) } -// GetTranslationFromStorage reads tender.json and returns the translation for a language -// when translation_status.done includes that language. +// GetTranslationFromStorage reads tender.json and returns the translation for a language. +// It prefers entries marked done in translation_status; if missing, it falls back to +// translations[] when title is non-empty (on-demand API may lag status updates). func (s *StorageClient) GetTranslationFromStorage(ctx context.Context, contractFolderID, noticePublicationID, language string) (StoredTranslation, error) { if strings.TrimSpace(contractFolderID) == "" { return StoredTranslation{}, fmt.Errorf("ai_summarizer: contractFolderID is required") @@ -353,14 +354,16 @@ func (s *StorageClient) getTranslationFromObjectKey(ctx context.Context, objectK return StoredTranslation{}, err } + if entry, ok := tenderJSON.translationForLanguage(language); ok { + return entry, nil + } + if entry, ok := tenderJSON.storedTranslationText(language); ok { + return entry, nil + } if !languageInDone(tenderJSON.translationsDone(), language) { return StoredTranslation{}, ErrTranslationNotReady } - entry, ok := tenderJSON.translationForLanguage(language) - if !ok { - return StoredTranslation{}, ErrNoTranslation - } - return entry, nil + return StoredTranslation{}, ErrNoTranslation } // getSummaryFromObjectKey reads tender.json at objectKey and returns summary text when ready. @@ -432,6 +435,39 @@ func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey, return summaryText, nil } +// GetDocumentSummariesFromStorage reads per-document summaries from tender.json +// when summarize_status.done is true. +func (s *StorageClient) GetDocumentSummariesFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) ([]DocumentSummary, error) { + if strings.TrimSpace(contractFolderID) == "" { + return nil, fmt.Errorf("ai_summarizer: contractFolderID is required") + } + if strings.TrimSpace(noticePublicationID) == "" { + return nil, fmt.Errorf("ai_summarizer: noticePublicationID is required") + } + + prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID) + if err != nil { + return nil, err + } + + tenderJSON, err := s.readTenderJSONAtKey(ctx, prefix+"tender.json") + if err != nil { + return nil, err + } + + done, _ := tenderJSON.resolved() + if !done { + return nil, ErrSummaryNotReady + } + if len(tenderJSON.Summaries) == 0 { + return []DocumentSummary{}, nil + } + + out := make([]DocumentSummary, len(tenderJSON.Summaries)) + copy(out, tenderJSON.Summaries) + return out, nil +} + // IsSummaryReady is a convenience method that checks whether the tender.json // exists and the summarize_status.done flag is true, without returning the // actual summary text.