From 6701428b09b7f39e28d3fd3410395d4ebb5c5093 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Sat, 16 May 2026 12:56:10 +0330 Subject: [PATCH] AI translate refactor --- cmd/worker/bootstrap/bootstrap.go | 77 ++++++++++-- cmd/worker/bootstrap/config.go | 17 ++- cmd/worker/main.go | 3 +- cmd/worker/workers/translation.go | 200 ++++++++++++++++++++---------- internal/tender/form.go | 3 +- internal/tender/service.go | 93 +++++++++----- pkg/ai_summarizer/client.go | 91 ++++++++++++-- pkg/ai_summarizer/entities.go | 99 +++++++++++++-- pkg/ai_summarizer/errors.go | 8 ++ pkg/ai_summarizer/storage.go | 85 ++++++++----- 10 files changed, 510 insertions(+), 166 deletions(-) diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 9e6befb..1bb88df 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -2,6 +2,7 @@ package bootstrap import ( "fmt" + "strings" "time" "tm/cmd/worker/workers" "tm/internal/notice" @@ -77,7 +78,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) *schedule.CronScheduler { +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 { // Debug: Log worker config appLogger.Info("Worker configuration", map[string]interface{}{ "worker_interval": config.Worker.Interval, @@ -192,9 +193,9 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger Expr: workerInterval, }) - // Initialize translation backfill worker for default language + // Initialize translation backfill worker for configured languages if aiClient != nil { - targetLanguage := config.AISummarizer.DefaultLanguage + targetLanguages := parseTranslationLanguages(config.AISummarizer) scheduler.AddJob(schedule.Job{ Name: "Tender Translation Worker Job", Func: func() { @@ -203,7 +204,8 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger appLogger, tenderRepo, aiClient, - targetLanguage, + aiStorage, + targetLanguages, config.Worker.TranslationBatchSize, ) worker.Run() @@ -211,9 +213,9 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger Expr: config.Worker.TranslationInterval, }) appLogger.Info("Scheduled tender translation worker", map[string]interface{}{ - "target_language": targetLanguage, - "interval": config.Worker.TranslationInterval, - "batch_size": config.Worker.TranslationBatchSize, + "target_languages": targetLanguages, + "interval": config.Worker.TranslationInterval, + "batch_size": config.Worker.TranslationBatchSize, }) } else { appLogger.Warn("AI summarizer client not available, tender translation worker is disabled", map[string]interface{}{}) @@ -274,6 +276,67 @@ func InitAISummarizerClient(conf AISummarizerConfig, log logger.Logger) *ai_summ return client } +// InitAISummarizerStorage initializes the AI summarizer MinIO storage client for workers. +func InitAISummarizerStorage(conf AISummarizerConfig, log logger.Logger) *ai_summarizer.StorageClient { + if conf.MinioEndpoint == "" { + log.Warn("AI Summarizer MinIO endpoint not configured, translation sync from storage will be unavailable", map[string]interface{}{}) + return nil + } + + cfg := &ai_summarizer.Config{ + APIBaseURL: conf.APIBaseURL, + APITimeout: conf.APITimeout, + APIRetryCount: conf.APIRetryCount, + APIRetryDelay: conf.APIRetryDelay, + MinioEndpoint: conf.MinioEndpoint, + MinioAccessKey: conf.MinioAccessKey, + MinioSecretKey: conf.MinioSecretKey, + MinioUseSSL: conf.MinioUseSSL, + MinioRegion: conf.MinioRegion, + MinioBucket: conf.MinioBucket, + } + + storage, err := ai_summarizer.NewStorageClient(cfg, log) + if err != nil { + log.Error("Failed to initialize AI Summarizer storage client for worker", map[string]interface{}{ + "error": err.Error(), + }) + return nil + } + + log.Info("AI Summarizer storage client initialized for worker", map[string]interface{}{ + "endpoint": conf.MinioEndpoint, + "bucket": conf.MinioBucket, + }) + + return storage +} + +func parseTranslationLanguages(conf AISummarizerConfig) []string { + raw := strings.TrimSpace(conf.TranslationLanguages) + if raw == "" { + raw = conf.DefaultLanguage + } + parts := strings.Split(raw, ",") + langs := make([]string, 0, len(parts)) + seen := make(map[string]struct{}, len(parts)) + for _, part := range parts { + lang := strings.ToLower(strings.TrimSpace(part)) + if lang == "" { + continue + } + if _, dup := seen[lang]; dup { + continue + } + seen[lang] = struct{}{} + langs = append(langs, lang) + } + if len(langs) == 0 { + return []string{"en"} + } + return langs +} + // Init Notification Service func InitNotificationService(conf config.NotificationConfig, log logger.Logger) notification.SDK { notificationConfig := ¬ification.Config{ diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index a63d69e..874631e 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -50,9 +50,16 @@ type WorkerConfig struct { // AISummarizerConfig holds configuration for the external AI summarizer service. type AISummarizerConfig struct { - APIBaseURL string `env:"AI_SUMMARIZER_API_BASE_URL" envDefault:""` - APITimeout time.Duration `env:"AI_SUMMARIZER_API_TIMEOUT" envDefault:"120s"` - APIRetryCount int `env:"AI_SUMMARIZER_API_RETRY_COUNT" envDefault:"2"` - APIRetryDelay time.Duration `env:"AI_SUMMARIZER_API_RETRY_DELAY" envDefault:"3s"` - DefaultLanguage string `env:"AI_SUMMARIZER_DEFAULT_LANGUAGE" envDefault:"en"` + APIBaseURL string `env:"AI_SUMMARIZER_API_BASE_URL" envDefault:""` + APITimeout time.Duration `env:"AI_SUMMARIZER_API_TIMEOUT" envDefault:"120s"` + APIRetryCount int `env:"AI_SUMMARIZER_API_RETRY_COUNT" envDefault:"2"` + APIRetryDelay time.Duration `env:"AI_SUMMARIZER_API_RETRY_DELAY" envDefault:"3s"` + DefaultLanguage string `env:"AI_SUMMARIZER_DEFAULT_LANGUAGE" envDefault:"en"` + TranslationLanguages string `env:"AI_SUMMARIZER_TRANSLATION_LANGUAGES" envDefault:"en"` + MinioEndpoint string `env:"AI_SUMMARIZER_MINIO_ENDPOINT" envDefault:""` + MinioAccessKey string `env:"AI_SUMMARIZER_MINIO_ACCESS_KEY" envDefault:""` + MinioSecretKey string `env:"AI_SUMMARIZER_MINIO_SECRET_KEY" envDefault:""` + MinioUseSSL bool `env:"AI_SUMMARIZER_MINIO_USE_SSL" envDefault:"false"` + MinioRegion string `env:"AI_SUMMARIZER_MINIO_REGION" envDefault:"us-east-1"` + MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens-documents"` } diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 8bf8521..e312f1b 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -45,9 +45,10 @@ func main() { // Initialize AI summarizer client (translation + document summarization workers) aiSummarizerClient := bootstrap.InitAISummarizerClient(config.AISummarizer, appLogger) + aiSummarizerStorage := bootstrap.InitAISummarizerStorage(config.AISummarizer, appLogger) // Initialize Worker - scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, minioService, aiSummarizerClient) + scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, minioService, aiSummarizerClient, aiSummarizerStorage) // Set up signal handling for graceful shutdown signalChan := make(chan os.Signal, 1) diff --git a/cmd/worker/workers/translation.go b/cmd/worker/workers/translation.go index 4923378..815b268 100644 --- a/cmd/worker/workers/translation.go +++ b/cmd/worker/workers/translation.go @@ -2,6 +2,7 @@ package workers import ( "context" + "errors" "strings" "time" "tm/internal/tender" @@ -10,32 +11,56 @@ import ( "tm/pkg/mongo" ) -// TranslationWorker backfills missing tender translations using AI summarizer API. +// TranslationWorker backfills missing tender translations using the AI pipeline +// and on-demand translate API, syncing results from MinIO when available. type TranslationWorker struct { - Mongo *mongo.ConnectionManager - Logger logger.Logger - TenderRepo tender.TenderRepository - AIClient *ai_summarizer.Client - TargetLanguage string - BatchSize int + Mongo *mongo.ConnectionManager + Logger logger.Logger + TenderRepo tender.TenderRepository + AIClient *ai_summarizer.Client + AIStorage *ai_summarizer.StorageClient + TargetLanguages []string + BatchSize int } -// NewTranslationWorker creates a translation worker for a target language. -func NewTranslationWorker(mongo *mongo.ConnectionManager, logger logger.Logger, tenderRepo tender.TenderRepository, aiClient *ai_summarizer.Client, targetLanguage string, batchSize int) *TranslationWorker { - if strings.TrimSpace(targetLanguage) == "" { - targetLanguage = "en" +// NewTranslationWorker creates a translation worker for the given target languages. +func NewTranslationWorker( + mongo *mongo.ConnectionManager, + logger logger.Logger, + tenderRepo tender.TenderRepository, + aiClient *ai_summarizer.Client, + aiStorage *ai_summarizer.StorageClient, + targetLanguages []string, + batchSize int, +) *TranslationWorker { + langs := make([]string, 0, len(targetLanguages)) + seen := make(map[string]struct{}, len(targetLanguages)) + for _, lang := range targetLanguages { + clean := strings.ToLower(strings.TrimSpace(lang)) + if clean == "" { + continue + } + if _, dup := seen[clean]; dup { + continue + } + seen[clean] = struct{}{} + langs = append(langs, clean) + } + if len(langs) == 0 { + langs = []string{"en"} } if batchSize <= 0 { batchSize = 20 } return &TranslationWorker{ - Mongo: mongo, - Logger: logger, - TenderRepo: tenderRepo, - AIClient: aiClient, - TargetLanguage: strings.ToLower(strings.TrimSpace(targetLanguage)), - BatchSize: batchSize, + Mongo: mongo, + Logger: logger, + TenderRepo: tenderRepo, + AIClient: aiClient, + AIStorage: aiStorage, + TargetLanguages: langs, + BatchSize: batchSize, } } @@ -43,48 +68,60 @@ func NewTranslationWorker(mongo *mongo.ConnectionManager, logger logger.Logger, func (w *TranslationWorker) Run() { if w.AIClient == nil { w.Logger.Warn("Translation worker skipped: AI client is not configured", map[string]interface{}{ - "target_language": w.TargetLanguage, + "target_languages": w.TargetLanguages, }) return } w.Logger.Info("Translation worker started", map[string]interface{}{ - "target_language": w.TargetLanguage, - "batch_size": w.BatchSize, + "target_languages": w.TargetLanguages, + "batch_size": w.BatchSize, }) + 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{}{ + "target_languages": w.TargetLanguages, + "error": err.Error(), + }) + } + + for _, language := range w.TargetLanguages { + w.runForLanguage(language) + } + + w.Logger.Info("Translation worker completed", map[string]interface{}{ + "target_languages": w.TargetLanguages, + }) +} + +func (w *TranslationWorker) runForLanguage(language string) { for { - // Always use skip=0 because successfully translated tenders are excluded by filter. - tenders, totalCount, err := w.TenderRepo.GetUnTranslatedTenders(context.Background(), w.TargetLanguage, w.BatchSize, 0) + tenders, totalCount, err := w.TenderRepo.GetUnTranslatedTenders(context.Background(), language, w.BatchSize, 0) if err != nil { w.Logger.Error("Failed to fetch untranslated tenders", map[string]interface{}{ - "target_language": w.TargetLanguage, + "target_language": language, "error": err.Error(), }) - break + return } w.Logger.Info("Fetched untranslated tenders batch", map[string]interface{}{ - "target_language": w.TargetLanguage, + "target_language": language, "batch_count": len(tenders), "total_count": totalCount, }) if len(tenders) == 0 { - break + return } for _, t := range tenders { - w.translateTender(&t) + w.translateTender(&t, language) } } - - w.Logger.Info("Translation worker completed", map[string]interface{}{ - "target_language": w.TargetLanguage, - }) } -func (w *TranslationWorker) translateTender(t *tender.Tender) { +func (w *TranslationWorker) translateTender(t *tender.Tender, language string) { if t.NoticePublicationID == "" { return } @@ -94,64 +131,97 @@ func (w *TranslationWorker) translateTender(t *tender.Tender) { }) return } - if existing, ok := t.Translations[w.TargetLanguage]; ok && strings.TrimSpace(existing.Title) != "" { + if existing, ok := t.Translations[language]; ok && strings.TrimSpace(existing.Title) != "" { w.Logger.Debug("Skipping tender translation because target language already exists", map[string]interface{}{ "tender_id": t.ID.Hex(), "notice_id": t.NoticePublicationID, - "target_language": w.TargetLanguage, + "target_language": language, }) return } - resp, err := w.AIClient.FetchTranslationOnDemand(context.Background(), ai_summarizer.TranslateRequest{ - NoticeID: t.NoticePublicationID, - ContractFolderID: t.ContractFolderID, - NoticePublicationID: t.NoticePublicationID, - Title: t.Title, - Description: t.Description, - Language: w.TargetLanguage, - }) + 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": w.TargetLanguage, + "target_language": language, + "translation_source": source, "translation_error": err.Error(), }) return } - if t.Translations == nil { - t.Translations = make(map[string]tender.TranslationEntry) - } - now := time.Now().Unix() - // Only update the target-language slot. Existing translations in other - // languages (e.g., "de", "fr") are preserved and never modified. - t.Translations[w.TargetLanguage] = tender.TranslationEntry{ - Title: resp.TranslatedTitle, - Description: resp.TranslatedDescription, - Language: w.TargetLanguage, - TranslatedAt: now, - } - if t.ProcessingMetadata.TranslatedData == nil { - t.ProcessingMetadata.TranslatedData = make(map[string]string) - } - t.ProcessingMetadata.TranslatedData[w.TargetLanguage] = "done" - t.ProcessingMetadata.TranslatedAt = now - + persistTranslation(t, language, title, description) if err := w.TenderRepo.Update(context.Background(), t); err != nil { w.Logger.Error("Failed to persist translated tender", map[string]interface{}{ "tender_id": t.ID.Hex(), "notice_id": t.NoticePublicationID, - "target_language": w.TargetLanguage, + "target_language": language, + "translation_source": source, "persistence_error": err.Error(), }) return } w.Logger.Info("Tender translated successfully", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_id": t.NoticePublicationID, - "target_language": w.TargetLanguage, + "tender_id": t.ID.Hex(), + "notice_id": t.NoticePublicationID, + "target_language": language, + "translation_source": source, }) } + +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, + }) + if apiErr != nil { + return "", "", "api", apiErr + } + return resp.TranslatedTitle, resp.TranslatedDescription, "api", nil +} + +func persistTranslation(t *tender.Tender, language, title, description string) { + if t.Translations == nil { + t.Translations = make(map[string]tender.TranslationEntry) + } + now := time.Now().Unix() + t.Translations[language] = tender.TranslationEntry{ + Title: title, + Description: description, + Language: language, + TranslatedAt: now, + } + if t.ProcessingMetadata.TranslatedData == nil { + t.ProcessingMetadata.TranslatedData = make(map[string]string) + } + t.ProcessingMetadata.TranslatedData[language] = "done" + t.ProcessingMetadata.TranslatedAt = now +} diff --git a/internal/tender/form.go b/internal/tender/form.go index faf2a5b..e097dd9 100644 --- a/internal/tender/form.go +++ b/internal/tender/form.go @@ -517,7 +517,8 @@ type AITranslateRequest struct { // AITranslateResponse represents the response for AI translation trigger. type AITranslateResponse struct { - NoticeID string `json:"notice_id"` + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` Language string `json:"language"` TranslatedTitle string `json:"translated_title"` TranslatedDescription string `json:"translated_description"` diff --git a/internal/tender/service.go b/internal/tender/service.go index 521159d..f81471e 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -21,6 +21,7 @@ import ( 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) + TriggerPipelineTranslate(ctx context.Context, languages []string) (*ai_summarizer.PipelineTranslateResponse, error) } // AISummarizerStorage defines the interface for fetching summaries from MinIO storage. @@ -28,6 +29,7 @@ type AISummarizerClient interface { // PROC_// in the shared bucket. type AISummarizerStorage interface { GetSummaryFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) (string, 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) } @@ -1051,25 +1053,13 @@ func (s *tenderService) TriggerAITranslate(ctx context.Context, id, language str return nil, fmt.Errorf("tender has no contract folder ID") } - if tr, ok := t.Translations[targetLanguage]; ok { - return &AITranslateResponse{ - NoticeID: t.NoticePublicationID, - Language: targetLanguage, - TranslatedTitle: tr.Title, - TranslatedDescription: tr.Description, - }, nil + if tr, ok := t.Translations[targetLanguage]; ok && strings.TrimSpace(tr.Title) != "" { + return s.buildAITranslateResponse(t, targetLanguage, tr.Title, tr.Description), nil } - resp, err := s.aiSummarizerClient.FetchTranslationOnDemand(ctx, ai_summarizer.TranslateRequest{ - NoticeID: t.NoticePublicationID, - ContractFolderID: t.ContractFolderID, - NoticePublicationID: t.NoticePublicationID, - Title: t.Title, - Description: t.Description, - Language: targetLanguage, - }) + title, description, err := s.resolveTranslation(ctx, t, targetLanguage) if err != nil { - s.logger.Error("On-demand AI translation failed", map[string]interface{}{ + s.logger.Error("AI translation failed", map[string]interface{}{ "tender_id": id, "notice_id": t.NoticePublicationID, "language": targetLanguage, @@ -1078,33 +1068,70 @@ func (s *tenderService) TriggerAITranslate(ctx context.Context, id, language str return nil, fmt.Errorf("AI translation request failed: %w", err) } + s.storeTranslation(t, targetLanguage, title, description) + if err := s.repository.Update(ctx, t); err != nil { + return nil, fmt.Errorf("failed to persist translation: %w", err) + } + + return s.buildAITranslateResponse(t, targetLanguage, title, description), nil +} + +func (s *tenderService) buildAITranslateResponse(t *Tender, language, title, description string) *AITranslateResponse { + return &AITranslateResponse{ + ContractFolderID: t.ContractFolderID, + NoticePublicationID: t.NoticePublicationID, + Language: language, + TranslatedTitle: title, + TranslatedDescription: description, + } +} + +func (s *tenderService) resolveTranslation(ctx context.Context, t *Tender, language string) (string, string, error) { + if s.aiSummarizerStorage != nil { + stored, err := s.aiSummarizerStorage.GetTranslationFromStorage(ctx, t.ContractFolderID, t.NoticePublicationID, language) + if err == nil { + return stored.Title, stored.Description, nil + } + if !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{}{ + "tender_id": t.ID.Hex(), + "language": language, + "error": err.Error(), + }) + } + } + + resp, err := s.aiSummarizerClient.FetchTranslationOnDemand(ctx, ai_summarizer.TranslateRequest{ + ContractFolderID: t.ContractFolderID, + NoticePublicationID: t.NoticePublicationID, + Title: t.Title, + Description: t.Description, + Language: language, + }) + if err != nil { + return "", "", err + } + return resp.TranslatedTitle, resp.TranslatedDescription, nil +} + +func (s *tenderService) storeTranslation(t *Tender, language, title, description string) { if t.Translations == nil { t.Translations = make(map[string]TranslationEntry) } - now := time.Now().Unix() - t.Translations[targetLanguage] = TranslationEntry{ - Title: resp.TranslatedTitle, - Description: resp.TranslatedDescription, - Language: targetLanguage, + t.Translations[language] = TranslationEntry{ + Title: title, + Description: description, + Language: language, TranslatedAt: now, } if t.ProcessingMetadata.TranslatedData == nil { t.ProcessingMetadata.TranslatedData = make(map[string]string) } - t.ProcessingMetadata.TranslatedData[targetLanguage] = "done" + t.ProcessingMetadata.TranslatedData[language] = "done" t.ProcessingMetadata.TranslatedAt = now - - if err := s.repository.Update(ctx, t); err != nil { - return nil, fmt.Errorf("failed to persist translation: %w", err) - } - - return &AITranslateResponse{ - NoticeID: resp.NoticeID, - Language: targetLanguage, - TranslatedTitle: resp.TranslatedTitle, - TranslatedDescription: resp.TranslatedDescription, - }, nil } func (s *tenderService) pickResponseLanguage(language *string) string { diff --git a/pkg/ai_summarizer/client.go b/pkg/ai_summarizer/client.go index 9beaed1..61329ee 100644 --- a/pkg/ai_summarizer/client.go +++ b/pkg/ai_summarizer/client.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "net/http" + "strings" "time" "tm/pkg/logger" @@ -93,8 +94,9 @@ func (c *Client) FetchTranslationOnDemand(ctx context.Context, reqBody Translate for attempt := 0; attempt <= c.config.APIRetryCount; attempt++ { if attempt > 0 { c.logger.Warn("Retrying AI translate request", map[string]interface{}{ - "notice_id": reqBody.NoticeID, - "attempt": attempt, + "notice_publication_id": reqBody.NoticePublicationID, + "language": reqBody.Language, + "attempt": attempt, }) select { case <-ctx.Done(): @@ -107,9 +109,10 @@ func (c *Client) FetchTranslationOnDemand(ctx context.Context, reqBody Translate if err != nil { lastErr = err c.logger.Error("AI translate request failed", map[string]interface{}{ - "notice_id": reqBody.NoticeID, - "attempt": attempt, - "error": err.Error(), + "notice_publication_id": reqBody.NoticePublicationID, + "language": reqBody.Language, + "attempt": attempt, + "error": err.Error(), }) continue } @@ -120,6 +123,50 @@ 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} + jsonBody, err := json.Marshal(reqBody) + if err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to marshal pipeline translate request: %w", err) + } + + url := c.config.APIBaseURL + "/pipeline/translate" + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(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) + } + 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 >= 400 { + return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + } + + var result PipelineTranslateResponse + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode pipeline translate response: %w", err) + } + + c.logger.Info("Pipeline translate triggered", map[string]interface{}{ + "status": result.Status, + "languages": result.Languages, + }) + + return &result, 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)) @@ -193,15 +240,37 @@ func (c *Client) doTranslatePost(ctx context.Context, url string, jsonBody []byt return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) } - var result TranslateResponse - if err := json.Unmarshal(bodyBytes, &result); err != nil { - return nil, fmt.Errorf("ai_summarizer: failed to decode translation response JSON: %w", err) + result, err := decodeTranslateResponse(bodyBytes) + if err != nil { + return nil, err } c.logger.Info("AI translate request succeeded", map[string]interface{}{ - "notice_id": result.NoticeID, - "language": result.Language, + "notice_publication_id": result.NoticePublicationID, + "language": result.Language, }) - return &result, nil + return result, nil +} + +type translateAPIEnvelope struct { + Success bool `json:"success"` + Message string `json:"message"` + Data TranslateResponse `json:"data"` +} + +func decodeTranslateResponse(bodyBytes []byte) (*TranslateResponse, error) { + var envelope translateAPIEnvelope + if err := json.Unmarshal(bodyBytes, &envelope); err == nil && envelope.Success && envelope.Data.Language != "" { + return &envelope.Data, nil + } + + var direct TranslateResponse + if err := json.Unmarshal(bodyBytes, &direct); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode translation response JSON: %w", err) + } + if strings.TrimSpace(direct.Language) == "" { + return nil, fmt.Errorf("ai_summarizer: translation response missing language") + } + return &direct, nil } diff --git a/pkg/ai_summarizer/entities.go b/pkg/ai_summarizer/entities.go index 4431885..fcb0414 100644 --- a/pkg/ai_summarizer/entities.go +++ b/pkg/ai_summarizer/entities.go @@ -44,29 +44,72 @@ type StoredDocument struct { // TranslateRequest represents the request payload sent to the AI service // POST /ai/translate endpoint. type TranslateRequest struct { - NoticeID string `json:"notice_id"` // required (TED-style notice key) - ContractFolderID string `json:"contract_folder_id"` // required by AI service (procedure / MinIO prefix) - NoticePublicationID string `json:"notice_publication_id"` // required by AI service - Title string `json:"title"` // required - Description string `json:"description"` // required - Language string `json:"language,omitempty"` // optional target language override + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` + Title string `json:"title"` + Description string `json:"description"` + Language string `json:"language"` } -// TranslateResponse represents the response from the AI translation API. +// TranslateResponse represents the translation payload from POST /ai/translate. type TranslateResponse struct { - NoticeID string `json:"notice_id"` + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` + Language string `json:"language"` TranslatedTitle string `json:"translated_title"` TranslatedDescription string `json:"translated_description"` - Language string `json:"language"` +} + +// PipelineTranslateRequest represents the request payload for POST /pipeline/translate. +type PipelineTranslateRequest struct { + Languages []string `json:"languages"` +} + +// PipelineTranslateResponse represents the response from POST /pipeline/translate. +type PipelineTranslateResponse struct { + Status string `json:"status"` + Languages []string `json:"languages"` +} + +// StoredTranslation is one language entry inside tender.json translations map. +type StoredTranslation struct { + Title string `json:"title"` + Description string `json:"description"` +} + +// TranslationStatus tracks which languages the async pipeline has finished. +type TranslationStatus struct { + Done []string `json:"done"` + DoneUpper []string `json:"Done"` +} + +func (s TranslationStatus) languagesDone() []string { + if len(s.Done) > 0 { + return s.Done + } + return s.DoneUpper +} + +func languageInDone(done []string, language string) bool { + lang := strings.ToLower(strings.TrimSpace(language)) + for _, item := range done { + if strings.ToLower(strings.TrimSpace(item)) == lang { + return true + } + } + return false } // TenderJSON represents the structure of the /tender.json file // stored in the AI service's MinIO bucket. type TenderJSON struct { - OverallSummary *string `json:"overall_summary"` - OverallSummaryCamel *string `json:"overallSummary"` - SummarizeStatus SummarizeStatus `json:"summarize_status"` - SummarizeStatusCamel SummarizeStatus `json:"summarizeStatus"` + OverallSummary *string `json:"overall_summary"` + OverallSummaryCamel *string `json:"overallSummary"` + SummarizeStatus SummarizeStatus `json:"summarize_status"` + SummarizeStatusCamel SummarizeStatus `json:"summarizeStatus"` + Translations map[string]StoredTranslation `json:"translations"` + TranslationStatus TranslationStatus `json:"translation_status"` + TranslationStatusCamel TranslationStatus `json:"translationStatus"` } // resolved returns whether summarization is marked done and the best-effort @@ -90,3 +133,33 @@ type SummarizeStatus struct { func (s SummarizeStatus) isDone() bool { return s.Done || s.DoneUpper } + +// translationsDone returns languages marked complete by the translation pipeline. +func (t *TenderJSON) translationsDone() []string { + done := t.TranslationStatus.languagesDone() + if len(done) > 0 { + return done + } + return t.TranslationStatusCamel.languagesDone() +} + +// translationForLanguage returns stored translation text 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 +} diff --git a/pkg/ai_summarizer/errors.go b/pkg/ai_summarizer/errors.go index 0570212..0e48086 100644 --- a/pkg/ai_summarizer/errors.go +++ b/pkg/ai_summarizer/errors.go @@ -12,6 +12,14 @@ var ( // overall_summary is empty or null. ErrNoOverallSummary = errors.New("ai_summarizer: overall_summary is empty") + // ErrTranslationNotReady is returned when tender.json exists but the + // requested language is not listed in translation_status.done yet. + ErrTranslationNotReady = errors.New("ai_summarizer: translation not ready yet") + + // ErrNoTranslation is returned when translation_status marks a language done + // but translations[language] is missing or empty. + ErrNoTranslation = errors.New("ai_summarizer: translation is empty") + // ErrObjectNotFound is returned when the tender.json object does not // exist in the MinIO bucket (HTTP 404 / NoSuchKey). ErrObjectNotFound = errors.New("ai_summarizer: tender.json not found in storage") diff --git a/pkg/ai_summarizer/storage.go b/pkg/ai_summarizer/storage.go index 448dadf..5588bdb 100644 --- a/pkg/ai_summarizer/storage.go +++ b/pkg/ai_summarizer/storage.go @@ -295,9 +295,46 @@ 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. +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") + } + if strings.TrimSpace(noticePublicationID) == "" { + return StoredTranslation{}, fmt.Errorf("ai_summarizer: noticePublicationID is required") + } + if strings.TrimSpace(language) == "" { + return StoredTranslation{}, fmt.Errorf("ai_summarizer: language is required") + } + + prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID) + if err != nil { + return StoredTranslation{}, err + } + return s.getTranslationFromObjectKey(ctx, prefix+"tender.json", language) +} + +// getTranslationFromObjectKey reads tender.json at objectKey and returns translation for language. +func (s *StorageClient) getTranslationFromObjectKey(ctx context.Context, objectKey, language string) (StoredTranslation, error) { + tenderJSON, err := s.readTenderJSONAtKey(ctx, objectKey) + if err != nil { + return StoredTranslation{}, err + } + + if !languageInDone(tenderJSON.translationsDone(), language) { + return StoredTranslation{}, ErrTranslationNotReady + } + entry, ok := tenderJSON.translationForLanguage(language) + if !ok { + return StoredTranslation{}, ErrNoTranslation + } + return entry, nil +} + // getSummaryFromObjectKey reads tender.json at objectKey and returns overall_summary when ready. // contractFolderID and noticePublicationID are used for logging only (may be empty when unknown). -func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey, contractFolderID, noticePublicationID string) (string, error) { +func (s *StorageClient) readTenderJSONAtKey(ctx context.Context, objectKey string) (*TenderJSON, error) { object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectKey, minio.GetObjectOptions{}) if err != nil { if isMinioNotFound(err) { @@ -305,13 +342,13 @@ func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey, "bucket": s.config.MinioBucket, "object_key": objectKey, }) - return "", ErrObjectNotFound + return nil, ErrObjectNotFound } var minioErr minio.ErrorResponse if errors.As(err, &minioErr) && minioErr.StatusCode == http.StatusForbidden { - return "", fmt.Errorf("ai_summarizer: access denied to bucket %s: %w", s.config.MinioBucket, err) + return nil, fmt.Errorf("ai_summarizer: access denied to bucket %s: %w", s.config.MinioBucket, err) } - return "", fmt.Errorf("ai_summarizer: failed to get object: %w", err) + return nil, fmt.Errorf("ai_summarizer: failed to get object: %w", err) } defer object.Close() @@ -322,14 +359,22 @@ func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey, "bucket": s.config.MinioBucket, "object_key": objectKey, }) - return "", ErrObjectNotFound + return nil, ErrObjectNotFound } - return "", fmt.Errorf("ai_summarizer: failed to read tender.json: %w", err) + return nil, fmt.Errorf("ai_summarizer: failed to read tender.json: %w", err) } var tenderJSON TenderJSON if err := json.Unmarshal(data, &tenderJSON); err != nil { - return "", fmt.Errorf("ai_summarizer: failed to parse tender.json: %w", err) + return nil, fmt.Errorf("ai_summarizer: failed to parse tender.json: %w", err) + } + return &tenderJSON, nil +} + +func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey, contractFolderID, noticePublicationID string) (string, error) { + tenderJSON, err := s.readTenderJSONAtKey(ctx, objectKey) + if err != nil { + return "", err } done, overall := tenderJSON.resolved() @@ -368,32 +413,12 @@ func (s *StorageClient) IsSummaryReady(ctx context.Context, contractFolderID, no } objectPath := prefix + "tender.json" - object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectPath, minio.GetObjectOptions{}) + tenderJSON, err := s.readTenderJSONAtKey(ctx, objectPath) if err != nil { - if isMinioNotFound(err) { + if errors.Is(err, ErrObjectNotFound) { return false, nil } - return false, fmt.Errorf("ai_summarizer: failed to check summary readiness: %w", err) - } - - data, readErr := io.ReadAll(object) - closeErr := object.Close() - if readErr != nil { - if isMinioNotFound(readErr) { - return false, nil - } - return false, fmt.Errorf("ai_summarizer: failed to read tender.json: %w", readErr) - } - if closeErr != nil { - s.logger.Warn("Failed to close tender.json object after readiness check", map[string]interface{}{ - "object_key": objectPath, - "error": closeErr.Error(), - }) - } - - var tenderJSON TenderJSON - if err := json.Unmarshal(data, &tenderJSON); err != nil { - return false, fmt.Errorf("ai_summarizer: failed to parse tender.json: %w", err) + return false, err } done, _ := tenderJSON.resolved()