List and download tender documents - worker auto translation to english job
This commit is contained in:
Mazyar
@@ -6,6 +6,7 @@ import (
|
||||
"tm/cmd/worker/workers"
|
||||
"tm/internal/notice"
|
||||
"tm/internal/tender"
|
||||
ai_summarizer "tm/pkg/ai_summarizer"
|
||||
"tm/pkg/config"
|
||||
"tm/pkg/glm"
|
||||
"tm/pkg/logger"
|
||||
@@ -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, glmService *glm.SDK, minioService *minio.Service) *schedule.CronScheduler {
|
||||
func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, glmService *glm.SDK, minioService *minio.Service, aiClient *ai_summarizer.Client) *schedule.CronScheduler {
|
||||
// Debug: Log worker config
|
||||
appLogger.Info("Worker configuration", map[string]interface{}{
|
||||
"worker_interval": config.Worker.Interval,
|
||||
@@ -186,6 +187,33 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
|
||||
Expr: workerInterval,
|
||||
})
|
||||
|
||||
// Initialize translation backfill worker for default language
|
||||
if aiClient != nil {
|
||||
targetLanguage := config.AISummarizer.DefaultLanguage
|
||||
scheduler.AddJob(schedule.Job{
|
||||
Name: "Tender Translation Worker Job",
|
||||
Func: func() {
|
||||
worker := workers.NewTranslationWorker(
|
||||
mongoManager,
|
||||
appLogger,
|
||||
tenderRepo,
|
||||
aiClient,
|
||||
targetLanguage,
|
||||
config.Worker.TranslationBatchSize,
|
||||
)
|
||||
worker.Run()
|
||||
},
|
||||
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,
|
||||
})
|
||||
} else {
|
||||
appLogger.Warn("AI summarizer client not available, tender translation worker is disabled", map[string]interface{}{})
|
||||
}
|
||||
|
||||
// Start the cron scheduler so all recurring jobs actually run
|
||||
scheduler.Start()
|
||||
appLogger.Info("Cron scheduler started with all worker jobs", map[string]interface{}{})
|
||||
@@ -193,6 +221,36 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
|
||||
return scheduler
|
||||
}
|
||||
|
||||
// InitAISummarizerClient initializes the AI summarizer HTTP client used by worker jobs.
|
||||
func InitAISummarizerClient(conf AISummarizerConfig, log logger.Logger) *ai_summarizer.Client {
|
||||
if conf.APIBaseURL == "" {
|
||||
log.Warn("AI Summarizer API base URL not configured, translation worker will be unavailable", map[string]interface{}{})
|
||||
return nil
|
||||
}
|
||||
|
||||
cfg := &ai_summarizer.Config{
|
||||
APIBaseURL: conf.APIBaseURL,
|
||||
APITimeout: conf.APITimeout,
|
||||
APIRetryCount: conf.APIRetryCount,
|
||||
APIRetryDelay: conf.APIRetryDelay,
|
||||
}
|
||||
|
||||
client, err := ai_summarizer.NewClient(cfg, log)
|
||||
if err != nil {
|
||||
log.Error("Failed to initialize AI Summarizer client for worker", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Info("AI Summarizer client initialized for worker", map[string]interface{}{
|
||||
"base_url": conf.APIBaseURL,
|
||||
"timeout": conf.APITimeout,
|
||||
})
|
||||
|
||||
return client
|
||||
}
|
||||
|
||||
// Init Notification Service
|
||||
func InitNotificationService(conf config.NotificationConfig, log logger.Logger) notification.SDK {
|
||||
notificationConfig := ¬ification.Config{
|
||||
|
||||
@@ -15,6 +15,7 @@ type Config struct {
|
||||
GLM GLMConfig
|
||||
Scraper config.ScraperConfig
|
||||
MinIO config.MinIOConfig
|
||||
AISummarizer AISummarizerConfig
|
||||
Queue QueueConfig `env:"QUEUE_" envPrefix:"QUEUE_"`
|
||||
AlertMail string `env:"ALERT_MAIL" envDefault:"alerts@opplens.com"`
|
||||
}
|
||||
@@ -37,6 +38,17 @@ type QueueConfig struct {
|
||||
type WorkerConfig struct {
|
||||
Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"`
|
||||
NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"0"`
|
||||
TranslationInterval string `env:"WORKER_TRANSLATION_INTERVAL" envDefault:"0 */30 * * * *"`
|
||||
TranslationBatchSize int `env:"WORKER_TRANSLATION_BATCH_SIZE" envDefault:"20"`
|
||||
}
|
||||
|
||||
// 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"`
|
||||
}
|
||||
|
||||
type GLMConfig struct {
|
||||
|
||||
+4
-1
@@ -54,8 +54,11 @@ func main() {
|
||||
appLogger.Info("MinIO service initialized successfully", map[string]interface{}{})
|
||||
}
|
||||
|
||||
// Initialize AI summarizer client (used by translation worker)
|
||||
aiSummarizerClient := bootstrap.InitAISummarizerClient(config.AISummarizer, appLogger)
|
||||
|
||||
// Initialize Worker
|
||||
scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, glmService, minioService)
|
||||
scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, glmService, minioService, aiSummarizerClient)
|
||||
|
||||
// Set up signal handling for graceful shutdown
|
||||
signalChan := make(chan os.Signal, 1)
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
package workers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"time"
|
||||
"tm/internal/tender"
|
||||
"tm/pkg/ai_summarizer"
|
||||
"tm/pkg/logger"
|
||||
"tm/pkg/mongo"
|
||||
)
|
||||
|
||||
// TranslationWorker backfills missing tender translations using AI summarizer API.
|
||||
type TranslationWorker struct {
|
||||
Mongo *mongo.ConnectionManager
|
||||
Logger logger.Logger
|
||||
TenderRepo tender.TenderRepository
|
||||
AIClient *ai_summarizer.Client
|
||||
TargetLanguage 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"
|
||||
}
|
||||
if batchSize <= 0 {
|
||||
batchSize = 20
|
||||
}
|
||||
|
||||
return &TranslationWorker{
|
||||
Mongo: mongo,
|
||||
Logger: logger,
|
||||
TenderRepo: tenderRepo,
|
||||
AIClient: aiClient,
|
||||
TargetLanguage: strings.ToLower(strings.TrimSpace(targetLanguage)),
|
||||
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_language": w.TargetLanguage,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
w.Logger.Info("Translation worker started", map[string]interface{}{
|
||||
"target_language": w.TargetLanguage,
|
||||
"batch_size": w.BatchSize,
|
||||
})
|
||||
|
||||
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)
|
||||
if err != nil {
|
||||
w.Logger.Error("Failed to fetch untranslated tenders", map[string]interface{}{
|
||||
"target_language": w.TargetLanguage,
|
||||
"error": err.Error(),
|
||||
})
|
||||
break
|
||||
}
|
||||
|
||||
w.Logger.Info("Fetched untranslated tenders batch", map[string]interface{}{
|
||||
"target_language": w.TargetLanguage,
|
||||
"batch_count": len(tenders),
|
||||
"total_count": totalCount,
|
||||
})
|
||||
|
||||
if len(tenders) == 0 {
|
||||
break
|
||||
}
|
||||
|
||||
for _, t := range tenders {
|
||||
w.translateTender(&t)
|
||||
}
|
||||
}
|
||||
|
||||
w.Logger.Info("Translation worker completed", map[string]interface{}{
|
||||
"target_language": w.TargetLanguage,
|
||||
})
|
||||
}
|
||||
|
||||
func (w *TranslationWorker) translateTender(t *tender.Tender) {
|
||||
if t.NoticePublicationID == "" {
|
||||
return
|
||||
}
|
||||
if existing, ok := t.Translations[w.TargetLanguage]; 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,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
resp, err := w.AIClient.FetchTranslationOnDemand(context.Background(), ai_summarizer.TranslateRequest{
|
||||
NoticeID: t.NoticePublicationID,
|
||||
Title: t.Title,
|
||||
Description: t.Description,
|
||||
Language: w.TargetLanguage,
|
||||
})
|
||||
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,
|
||||
"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
|
||||
|
||||
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,
|
||||
"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,
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user