package workers import ( "context" "errors" "strings" "tm/internal/tender" "tm/pkg/ai_summarizer" "tm/pkg/logger" "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. type TranslationWorker struct { 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 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, AIStorage: aiStorage, TargetLanguages: langs, BatchSize: batchSize, } } // Run starts backfilling translations for tenders missing target-language content. func (w *TranslationWorker) Run() { if w.AIClient == nil { w.Logger.Warn("Translation worker skipped: AI client is not configured", map[string]interface{}{ "target_languages": w.TargetLanguages, }) return } w.Logger.Info("Translation worker started", map[string]interface{}{ "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) { 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{}{ "target_language": language, "error": err.Error(), }) return } w.Logger.Info("Fetched tenders for translation backfill", map[string]interface{}{ "target_language": language, "batch_count": len(tenders), "total_count": totalCount, "skip": skip, }) if len(tenders) == 0 { return } for _, t := range tenders { w.translateTender(&t, language) } skip += len(tenders) if int64(skip) >= totalCount { return } } } 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 { return false } stored, err := w.AIStorage.GetTranslationFromStorage( context.Background(), t.ContractFolderID, t.NoticePublicationID, language, ) 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, }) if apiErr != nil { return "", "", "api", apiErr } return resp.TranslatedTitle, resp.TranslatedDescription, "api", nil }