diff --git a/cmd/web/bootstrap/bootstrap.go b/cmd/web/bootstrap/bootstrap.go index 17d1dc0..a67ee5f 100644 --- a/cmd/web/bootstrap/bootstrap.go +++ b/cmd/web/bootstrap/bootstrap.go @@ -7,7 +7,6 @@ import ( "tm/pkg/authorization" "tm/pkg/config" "tm/pkg/filestore" - "tm/pkg/glm" "tm/pkg/hcaptcha" "tm/pkg/logger" "tm/pkg/mongo" @@ -272,52 +271,6 @@ func InitHCaptchaService(conf config.HCaptchaConfig, log logger.Logger) hcaptcha return hcaptchaVerifier } -// InitGLMService initializes the GLM service -func InitGLMService(conf GLMConfig, log logger.Logger) *glm.SDK { - glmSDK, err := glm.New( - &glm.Config{ - BaseURL: conf.BaseURL, - APIKey: conf.APIKey, - Timeout: conf.Timeout, - RetryAttempts: conf.RetryAttempts, - RetryDelay: conf.RetryDelay, - EnableLogging: conf.EnableLogging, - UserAgent: conf.UserAgent, - DefaultModel: conf.DefaultModel, - DefaultMaxTokens: conf.DefaultMaxTokens, - DefaultTopP: conf.DefaultTopP, - DefaultTemperature: conf.DefaultTemperature, - DefaultFrequencyPenalty: conf.DefaultFrequencyPenalty, - DefaultPresencePenalty: conf.DefaultPresencePenalty, - EnableStreaming: conf.EnableStreaming, - EnableThinking: conf.EnableThinking, - TranslationOptions: glm.TranslationOption{ - Model: conf.TranslateOptions.Model, - MaxTokens: conf.TranslateOptions.MaxTokens, - Temperature: conf.TranslateOptions.Temperature, - FrequencyPenalty: conf.TranslateOptions.FrequencyPenalty, - PresencePenalty: conf.TranslateOptions.PresencePenalty, - TopP: conf.TranslateOptions.TopP, - EnableThinking: conf.TranslateOptions.EnableThinking, - }, - }, log) - - if err != nil { - log.Error("Failed to initialize GLM SDK", map[string]interface{}{ - "error": err.Error(), - }) - log.Warn("GLM service will be disabled, keyword generation features will not be available", map[string]interface{}{}) - return nil - } - - log.Info("GLM SDK initialized successfully", map[string]interface{}{ - "base_url": conf.BaseURL, - "timeout": conf.Timeout, - }) - - return glmSDK -} - // InitFileStore initializes the GridFS file store service func InitFileStore(mongoManager *mongo.ConnectionManager, log logger.Logger) (*filestore.GridFSService, error) { fileStoreService, err := filestore.NewGridFSService(mongoManager, log) diff --git a/cmd/web/bootstrap/config.go b/cmd/web/bootstrap/config.go index fca6513..06bd2c7 100644 --- a/cmd/web/bootstrap/config.go +++ b/cmd/web/bootstrap/config.go @@ -16,7 +16,6 @@ type Config struct { Notification config.NotificationConfig Ollama config.OllamaConfig HCaptcha config.HCaptchaConfig - GLM GLMConfig Scraper ScraperConfig TED TEDConfig UserAuth AuthConfig @@ -25,35 +24,6 @@ type Config struct { AISummarizer AISummarizerConfig } -type GLMConfig struct { - BaseURL string `env:"GLM_BASE_URL" envDefault:"https://api.z.ai"` - APIKey string `env:"GLM_API_KEY" envDefault:""` - Timeout time.Duration `env:"GLM_TIMEOUT" envDefault:"300s"` - RetryAttempts int `env:"GLM_RETRY_ATTEMPTS" envDefault:"3"` - RetryDelay time.Duration `env:"GLM_RETRY_DELAY" envDefault:"2s"` - EnableLogging bool `env:"GLM_ENABLE_LOGGING" envDefault:"true"` - UserAgent string `env:"GLM_USER_AGENT" envDefault:"Opplens-GLMSDK/1.0"` - DefaultModel string `env:"GLM_DEFAULT_MODEL" envDefault:"glm-4.5"` - DefaultMaxTokens int `env:"GLM_DEFAULT_MAX_TOKENS" envDefault:"4096"` - DefaultTopP float64 `env:"GLM_DEFAULT_TOP_P" envDefault:"1.0"` - DefaultTemperature float64 `env:"GLM_DEFAULT_TEMPERATURE" envDefault:"0.7"` - DefaultFrequencyPenalty float64 `env:"GLM_DEFAULT_FREQUENCY_PENALTY" envDefault:"0"` - DefaultPresencePenalty float64 `env:"GLM_DEFAULT_PRESENCE_PENALTY" envDefault:"0"` - EnableStreaming bool `env:"GLM_ENABLE_STREAMING" envDefault:"false"` - EnableThinking bool `env:"GLM_ENABLE_THINKING" envDefault:"false"` - TranslateOptions TranslateOptions -} - -type TranslateOptions struct { - Model string `env:"GLM_TRANSLATION_MODEL" envDefault:"glm-4.5"` - MaxTokens int `env:"GLM_TRANSLATION_MAX_TOKENS" envDefault:"2048"` - Temperature float64 `env:"GLM_TRANSLATION_TEMPERATURE" envDefault:"0.3"` - FrequencyPenalty float64 `env:"GLM_TRANSLATION_FREQUENCY_PENALTY" envDefault:"0.1"` - PresencePenalty float64 `env:"GLM_TRANSLATION_PRESENCE_PENALTY" envDefault:"0.1"` - TopP float64 `env:"GLM_TRANSLATION_TOP_P" envDefault:"1.0"` - EnableThinking bool `env:"GLM_TRANSLATION_ENABLE_THINKING" envDefault:"false"` -} - type ScraperConfig struct { BaseURL string `env:"SCRAPER_BASE_URL"` Timeout time.Duration `env:"SCRAPER_TIMEOUT"` diff --git a/cmd/web/main.go b/cmd/web/main.go index 5f773dd..8c5fa9b 100644 --- a/cmd/web/main.go +++ b/cmd/web/main.go @@ -159,9 +159,6 @@ func main() { // Initialize hCaptcha verifier hcaptchaVerifier := bootstrap.InitHCaptchaService(conf.HCaptcha, logger) - // Initialize GLM service - glmSDK := bootstrap.InitGLMService(conf.GLM, logger) - // Initialize AI Summarizer service var aiSummarizerClient tender.AISummarizerClient var aiSummarizerStorage tender.AISummarizerStorage @@ -206,7 +203,7 @@ func main() { userService := user.NewService(userRepository, logger, userAuthService, userValidator, notificationSDK) categoryService := company_category.NewService(categoryRepository, logger) companyService := company.NewService(companyRepository, categoryService, logger) - customerService := customer.NewService(customerRepository, logger, customerAuthService, companyService, redisClient, notificationSDK, customerValidator, glmSDK) + customerService := customer.NewService(customerRepository, logger, customerAuthService, companyService, redisClient, notificationSDK, customerValidator) tenderService := tender.NewService(tenderRepository, companyService, customerService, logger, aiSummarizerClient, aiSummarizerStorage, conf.AISummarizer.DefaultLanguage) feedbackService := feedback.NewService(feedbackRepo, tenderService, companyService, logger) tenderApprovalService := tender_approval.NewService(tenderApprovalRepository, tenderService, logger) diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 1ba6bfa..266f905 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -8,7 +8,6 @@ import ( "tm/internal/tender" ai_summarizer "tm/pkg/ai_summarizer" "tm/pkg/config" - "tm/pkg/glm" "tm/pkg/logger" "tm/pkg/minio" "tm/pkg/mongo" @@ -78,12 +77,15 @@ 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, 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) *schedule.CronScheduler { // Debug: Log worker config appLogger.Info("Worker configuration", map[string]interface{}{ - "worker_interval": config.Worker.Interval, - "queue_enabled": config.Queue.Enabled, - "queue_mode": config.Queue.Mode, + "worker_interval": config.Worker.Interval, + "notice_concurrency": config.Worker.NoticeWorkerConcurrency, + "notice_fetch_batch": config.Worker.NoticeFetchBatchSize, + "notice_processing_limit": config.Worker.NoticeProcessingLimit, + "queue_enabled": config.Queue.Enabled, + "queue_mode": config.Queue.Mode, }) // Initialize repositories noticeRepo := notice.NewRepository(mongoManager, appLogger) @@ -92,14 +94,6 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger // Create a single shared cron scheduler for all recurring jobs scheduler := schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo) - // Initialize Notice Worker - appLogger.Info("Starting notice worker", map[string]interface{}{ - "processing_limit": config.Worker.NoticeProcessingLimit, - }) - noticeWorker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, config.Worker.NoticeProcessingLimit, config.Worker.DeleteProcessedNotices) - noticeWorker.Run() - - // Initialize Queue-based scraping if enabled var queueService queue.Queue if config.Queue.Enabled { var err error @@ -150,26 +144,23 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger }) } - // Initialize Document Summarization Worker (only if MinIO is available) - if minioService != nil { - appLogger.Info("Starting document summarization worker", map[string]interface{}{}) - summarizerWorker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, minioService, glmService) - summarizerWorker.Run() - - // Schedule Document Summarization Worker job + // Document summarization via external AI service (requires MinIO + AI_SUMMARIZER_API_BASE_URL) + if minioService != nil && aiClient != nil { scheduler.AddJob(schedule.Job{ Name: "Document Summarization Worker Job", Func: func() { - worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, minioService, glmService) + worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, minioService, aiClient) worker.Run() }, Expr: "0 11 * * * *", // Run at 11 AM every day (1 hour after document scraping) }) + } 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{}{}) } - // Schedule Notice Worker job + // Schedule Notice Worker job (no blocking run at startup; first tick runs on schedule) workerInterval := config.Worker.Interval if workerInterval == "" { workerInterval = "* 10 * * * *" // Default: every hour at minute 10 @@ -181,7 +172,17 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger scheduler.AddJob(schedule.Job{ Name: "Notice Worker Job", Func: func() { - worker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, config.Worker.NoticeProcessingLimit, config.Worker.DeleteProcessedNotices) + worker := workers.NewNoticeWorker( + mongoManager, + appLogger, + ¬ify, + noticeRepo, + tenderRepo, + config.Worker.NoticeProcessingLimit, + config.Worker.NoticeWorkerConcurrency, + config.Worker.NoticeFetchBatchSize, + config.Worker.DeleteProcessedNotices, + ) worker.Run() }, Expr: workerInterval, @@ -214,6 +215,22 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger appLogger.Warn("AI summarizer client not available, tender translation worker is disabled", map[string]interface{}{}) } + // Kick off one notice-processing pass without blocking startup (cron continues on schedule) + go func() { + w := workers.NewNoticeWorker( + mongoManager, + appLogger, + ¬ify, + noticeRepo, + tenderRepo, + config.Worker.NoticeProcessingLimit, + config.Worker.NoticeWorkerConcurrency, + config.Worker.NoticeFetchBatchSize, + config.Worker.DeleteProcessedNotices, + ) + w.Run() + }() + // Start the cron scheduler so all recurring jobs actually run scheduler.Start() appLogger.Info("Cron scheduler started with all worker jobs", map[string]interface{}{}) @@ -306,52 +323,6 @@ func InitOllamaService(conf config.OllamaConfig, log logger.Logger) *ollama.SDK return ollamaSDK } -// Init GLM Service -func InitGLMService(conf GLMConfig, log logger.Logger) *glm.SDK { - glmSDK, err := glm.New( - &glm.Config{ - BaseURL: conf.BaseURL, - APIKey: conf.APIKey, - Timeout: conf.Timeout, - RetryAttempts: conf.RetryAttempts, - RetryDelay: conf.RetryDelay, - EnableLogging: conf.EnableLogging, - UserAgent: conf.UserAgent, - DefaultModel: conf.DefaultModel, - DefaultMaxTokens: conf.DefaultMaxTokens, - DefaultTopP: conf.DefaultTopP, - DefaultTemperature: conf.DefaultTemperature, - DefaultFrequencyPenalty: conf.DefaultFrequencyPenalty, - DefaultPresencePenalty: conf.DefaultPresencePenalty, - EnableStreaming: conf.EnableStreaming, - EnableThinking: conf.EnableThinking, - TranslationOptions: glm.TranslationOption{ - Model: conf.TranslateOptions.Model, - MaxTokens: conf.TranslateOptions.MaxTokens, - Temperature: conf.TranslateOptions.Temperature, - FrequencyPenalty: conf.TranslateOptions.FrequencyPenalty, - PresencePenalty: conf.TranslateOptions.PresencePenalty, - TopP: conf.TranslateOptions.TopP, - EnableThinking: conf.TranslateOptions.EnableThinking, - }, - }, log) - - if err != nil { - log.Error("Failed to initialize GLM SDK", map[string]interface{}{ - "error": err.Error(), - }) - log.Warn("GLM service will be disabled, translation features will not be available", map[string]interface{}{}) - return nil - } - - log.Info("GLM SDK initialized successfully", map[string]interface{}{ - "base_url": conf.BaseURL, - "timeout": conf.Timeout, - }) - - return glmSDK -} - // Init MinIO Service func InitMinIOService(conf config.MinIOConfig, log logger.Logger) *minio.Service { // Convert bootstrap config to minio config diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 349ece3..65fbcbd 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -12,7 +12,6 @@ type Config struct { Notification config.NotificationConfig Ollama config.OllamaConfig Worker WorkerConfig - GLM GLMConfig Scraper config.ScraperConfig MinIO config.MinIOConfig AISummarizer AISummarizerConfig @@ -39,8 +38,10 @@ type WorkerConfig struct { Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"` NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"0"` DeleteProcessedNotices bool `env:"WORKER_DELETE_PROCESSED_NOTICES" envDefault:"true"` + NoticeWorkerConcurrency int `env:"WORKER_NOTICE_CONCURRENCY" envDefault:"8"` + NoticeFetchBatchSize int `env:"WORKER_NOTICE_FETCH_BATCH" envDefault:"50"` TranslationInterval string `env:"WORKER_TRANSLATION_INTERVAL" envDefault:"0 */30 * * * *"` - TranslationBatchSize int `env:"WORKER_TRANSLATION_BATCH_SIZE" envDefault:"20"` + TranslationBatchSize int `env:"WORKER_TRANSLATION_BATCH_SIZE" envDefault:"20"` } // AISummarizerConfig holds configuration for the external AI summarizer service. @@ -51,32 +52,3 @@ type AISummarizerConfig struct { APIRetryDelay time.Duration `env:"AI_SUMMARIZER_API_RETRY_DELAY" envDefault:"3s"` DefaultLanguage string `env:"AI_SUMMARIZER_DEFAULT_LANGUAGE" envDefault:"en"` } - -type GLMConfig struct { - BaseURL string `env:"GLM_BASE_URL" envDefault:"https://api.z.ai"` - APIKey string `env:"GLM_API_KEY" envDefault:""` - Timeout time.Duration `env:"GLM_TIMEOUT" envDefault:"300s"` - RetryAttempts int `env:"GLM_RETRY_ATTEMPTS" envDefault:"3"` - RetryDelay time.Duration `env:"GLM_RETRY_DELAY" envDefault:"2s"` - EnableLogging bool `env:"GLM_ENABLE_LOGGING" envDefault:"true"` - UserAgent string `env:"GLM_USER_AGENT" envDefault:"Opplens-GLMSDK/1.0"` - DefaultModel string `env:"GLM_DEFAULT_MODEL" envDefault:"glm-4.5"` - DefaultMaxTokens int `env:"GLM_DEFAULT_MAX_TOKENS" envDefault:"4096"` - DefaultTopP float64 `env:"GLM_DEFAULT_TOP_P" envDefault:"1.0"` - DefaultTemperature float64 `env:"GLM_DEFAULT_TEMPERATURE" envDefault:"0.7"` - DefaultFrequencyPenalty float64 `env:"GLM_DEFAULT_FREQUENCY_PENALTY" envDefault:"0"` - DefaultPresencePenalty float64 `env:"GLM_DEFAULT_PRESENCE_PENALTY" envDefault:"0"` - EnableStreaming bool `env:"GLM_ENABLE_STREAMING" envDefault:"false"` - EnableThinking bool `env:"GLM_ENABLE_THINKING" envDefault:"false"` - TranslateOptions TranslateOptions -} - -type TranslateOptions struct { - Model string `env:"GLM_TRANSLATION_MODEL" envDefault:"glm-4.5"` - MaxTokens int `env:"GLM_TRANSLATION_MAX_TOKENS" envDefault:"2048"` - Temperature float64 `env:"GLM_TRANSLATION_TEMPERATURE" envDefault:"0.3"` - FrequencyPenalty float64 `env:"GLM_TRANSLATION_FREQUENCY_PENALTY" envDefault:"0.1"` - PresencePenalty float64 `env:"GLM_TRANSLATION_PRESENCE_PENALTY" envDefault:"0.1"` - TopP float64 `env:"GLM_TRANSLATION_TOP_P" envDefault:"1.0"` - EnableThinking bool `env:"GLM_TRANSLATION_ENABLE_THINKING" envDefault:"false"` -} diff --git a/cmd/worker/main.go b/cmd/worker/main.go index e0a1682..8bf8521 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -37,28 +37,17 @@ func main() { // Initialize notification service notificationService := bootstrap.InitNotificationService(config.Notification, appLogger) - // Initialize GLM service - glmService := bootstrap.InitGLMService(config.GLM, appLogger) - if glmService != nil { - appLogger.Info("GLM service initialized successfully", map[string]interface{}{}) - } else { - appLogger.Warn("GLM service not available, translation features will be disabled", map[string]interface{}{}) - } - - // Initialize scraper service - // scraperService := bootstrap.InitScraperService(*config, 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 (used by translation worker) + // Initialize AI summarizer client (translation + document summarization workers) aiSummarizerClient := bootstrap.InitAISummarizerClient(config.AISummarizer, appLogger) // Initialize Worker - scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, glmService, minioService, aiSummarizerClient) + scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, minioService, aiSummarizerClient) // Set up signal handling for graceful shutdown signalChan := make(chan os.Signal, 1) diff --git a/cmd/worker/workers/notice.go b/cmd/worker/workers/notice.go index 764692c..4686f3b 100644 --- a/cmd/worker/workers/notice.go +++ b/cmd/worker/workers/notice.go @@ -2,44 +2,76 @@ package workers import ( "context" + "errors" "fmt" + "strconv" "strings" + "sync" + "sync/atomic" "time" "tm/internal/notice" "tm/internal/tender" - "tm/pkg/glm" "tm/pkg/logger" "tm/pkg/mongo" "tm/pkg/notification" "go.mongodb.org/mongo-driver/v2/bson" mongodriver "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" + "golang.org/x/sync/errgroup" +) + +const ( + // tenderIDCountersCollection holds per-(sourceCode+year) atomic counters used to allocate + // unique tender_id values (e.g. PTD26827). Using a Mongo counter document is the standard way + // to avoid race conditions when several goroutines (or processes) generate ids in parallel. + tenderIDCountersCollection = "tender_id_counters" + // tenderIDMaxRetries caps how many times we retry allocating a new tender id when an insert + // fails with a duplicate-key error (e.g. counter lagging behind existing rows). + tenderIDMaxRetries = 5 ) type NoticeWorker struct { - Mongo *mongo.ConnectionManager - Logger logger.Logger - Notify *notification.SDK - NoticeRepo notice.Repository - TenderRepo tender.TenderRepository - GLM *glm.SDK - ProcessingLimit int - DeleteProcessedNotices bool + Mongo *mongo.ConnectionManager + Logger logger.Logger + Notify *notification.SDK + NoticeRepo notice.Repository + TenderRepo tender.TenderRepository + ProcessingLimit int + DeleteProcessedNotices bool + // Concurrency is the maximum number of notices processed in parallel per batch (default 8). + Concurrency int + // FetchBatchSize is how many unprocessed notices to load from MongoDB per batch (default 50). + FetchBatchSize int + + // tenderIDSeedMu / tenderIDSeeded track which (sourceCode+year) counters have been seeded + // from the current tenders collection during this process, so we only run the seeding + // $max-upsert once per key. + tenderIDSeedMu sync.Mutex + tenderIDSeeded map[string]bool } -func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, glmService *glm.SDK, processingLimit int, deleteProcessedNotices bool) *NoticeWorker { +func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, processingLimit, concurrency, fetchBatchSize int, deleteProcessedNotices bool) *NoticeWorker { if processingLimit < 0 { processingLimit = 5 // Default limit } + if concurrency < 1 { + concurrency = 8 + } + if fetchBatchSize < 1 { + fetchBatchSize = 50 + } return &NoticeWorker{ Mongo: mongo, Logger: logger, Notify: notify, NoticeRepo: noticeRepo, TenderRepo: tenderRepo, - GLM: glmService, ProcessingLimit: processingLimit, DeleteProcessedNotices: deleteProcessedNotices, + Concurrency: concurrency, + FetchBatchSize: fetchBatchSize, + tenderIDSeeded: make(map[string]bool), } } @@ -97,23 +129,27 @@ func (w *NoticeWorker) cleanupProcessedNotices() { } func (w *NoticeWorker) Run() { - w.Logger.Info("Notice worker started", map[string]interface{}{}) + w.Logger.Info("Notice worker started", map[string]interface{}{ + "concurrency": w.Concurrency, + "fetch_batch": w.FetchBatchSize, + "processing_limit": w.ProcessingLimit, + }) - processedCount := 0 + var processedTotal atomic.Int64 maxToProcess := w.ProcessingLimit - batchSize := 10 - for maxToProcess == 0 || processedCount < maxToProcess { - // Calculate how many to fetch in this batch - fetchLimit := batchSize + for maxToProcess == 0 || int(processedTotal.Load()) < maxToProcess { + fetchLimit := w.FetchBatchSize if maxToProcess > 0 { - remaining := maxToProcess - processedCount + remaining := maxToProcess - int(processedTotal.Load()) + if remaining < 1 { + break + } if remaining < fetchLimit { fetchLimit = remaining } } - // Always skip=0 since processed notices are excluded by the filter notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), fetchLimit, 0) if err != nil { w.Logger.Error("Failed to get notices", map[string]interface{}{ @@ -124,77 +160,75 @@ func (w *NoticeWorker) Run() { if len(notices) == 0 { w.Logger.Info("No more unprocessed notices found", map[string]interface{}{ - "processed_count": processedCount, + "processed_count": processedTotal.Load(), }) break } - for _, n := range notices { - if maxToProcess > 0 && processedCount >= maxToProcess { - w.Logger.Info("Reached maximum processing limit, stopping", map[string]interface{}{ - "processed_count": processedCount, - "max_to_process": maxToProcess, - }) - break - } - - w.Logger.Info("Notice", map[string]interface{}{ - "notice": n.ID.Hex(), - }) - - t, err := w.ToTender(&n) - if err != nil { - w.Logger.Error("Failed to convert notice to tender, skipping notice", map[string]interface{}{ - "error": err.Error(), + var g errgroup.Group + g.SetLimit(w.Concurrency) + for i := range notices { + n := notices[i] + g.Go(func() error { + w.Logger.Debug("Processing notice", map[string]interface{}{ "notice_id": n.ID.Hex(), }) - // Mark notice as processed so it doesn't block the queue + + t, convErr := w.ToTender(&n) + if convErr != nil { + w.Logger.Error("Failed to convert notice to tender, skipping notice", map[string]interface{}{ + "error": convErr.Error(), + "notice_id": n.ID.Hex(), + }) + n.ProcessingMetadata.Processed = true + n.ProcessingMetadata.ProcessedAt = time.Now().Unix() + if updateErr := w.NoticeRepo.Update(context.Background(), &n); updateErr != nil { + w.Logger.Error("Failed to update notice after conversion error", map[string]interface{}{ + "error": updateErr.Error(), + "notice_id": n.ID.Hex(), + }) + } + return nil + } + + persistErr := w.persistTender(context.Background(), t, &n) + if persistErr != nil { + // Leave the notice unprocessed so it is retried on the next worker tick. + w.Logger.Error("Failed to persist tender, notice left unprocessed for retry", map[string]interface{}{ + "error": persistErr.Error(), + "notice_id": n.ID.Hex(), + }) + return nil + } + n.ProcessingMetadata.Processed = true n.ProcessingMetadata.ProcessedAt = time.Now().Unix() if updateErr := w.NoticeRepo.Update(context.Background(), &n); updateErr != nil { - w.Logger.Error("Failed to update notice after conversion error", map[string]interface{}{ + w.Logger.Error("Failed to update notice", map[string]interface{}{ "error": updateErr.Error(), "notice_id": n.ID.Hex(), }) - } - continue - } - - if t.ID.IsZero() { - err = w.TenderRepo.Create(context.Background(), t) - if err != nil { - w.Logger.Error("Failed to create tender", map[string]interface{}{ - "error": err.Error(), + } else { + processedTotal.Add(1) + w.Logger.Debug("Notice processed successfully", map[string]interface{}{ + "notice": n.ID.Hex(), + "processed_count": processedTotal.Load(), }) } - } else { - err = w.TenderRepo.Update(context.Background(), t) - if err != nil { - w.Logger.Error("Failed to update tender", map[string]interface{}{ - "error": err.Error(), - }) - } - } - - // Mark notice as processed so it's excluded from future queries - n.ProcessingMetadata.Processed = true - n.ProcessingMetadata.ProcessedAt = time.Now().Unix() - err = w.NoticeRepo.Update(context.Background(), &n) - if err != nil { - w.Logger.Error("Failed to update notice", map[string]interface{}{ - "error": err.Error(), - }) - } else { - processedCount++ - w.Logger.Info("Notice processed successfully", map[string]interface{}{ - "notice": n.ID.Hex(), - "processed_count": processedCount, - }) - } + return nil + }) + } + if err := g.Wait(); err != nil { + w.Logger.Error("Notice worker batch failed", map[string]interface{}{ + "error": err.Error(), + }) } - // Check if we need to exit outer loop (break from inner loop means we hit the limit) - if maxToProcess > 0 && processedCount >= maxToProcess { + if maxToProcess > 0 && int(processedTotal.Load()) >= maxToProcess { + w.Logger.Info("Reached maximum processing limit, stopping", map[string]interface{}{ + "processed_count": processedTotal.Load(), + "max_to_process": maxToProcess, + }) break } } @@ -208,19 +242,47 @@ func (w *NoticeWorker) Run() { } func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) { - t, err := w.TenderRepo.GetByNoticePublicationID(context.Background(), n.NoticePublicationID) - if err != nil { + t := w.resolveExistingTender(context.Background(), n) + if t == nil { w.Logger.Info("New tender raised", map[string]interface{}{ "notice_publication_id": n.NoticePublicationID, + "contract_folder_id": n.ContractFolderID, "country_code": n.CountryCode, }) t = new(tender.Tender) } - var preserveTenderID, preserveProjectName string - if t != nil && !t.ID.IsZero() { + // Guard against out-of-order or duplicate notice ingestion: if this notice is older than the data + // already on the tender (same notice id with an older BT-757 version, or a strictly earlier issue + // date), don't overwrite the richer record. The notice is still marked processed by the caller. + if !t.ID.IsZero() && isStaleNotice(t, n) { + w.Logger.Info("Stale notice ignored, preserving existing tender", map[string]interface{}{ + "tender_id": t.GetID(), + "contract_folder_id": n.ContractFolderID, + "incoming_notice_id": n.ContractNoticeID, + "incoming_version": n.NoticeVersion, + "incoming_issue_date": n.IssueDate, + "existing_notice_id": t.ContractNoticeID, + "existing_version": t.NoticeVersion, + "existing_issue_date": t.IssueDate, + "existing_status": string(t.Status), + }) + return t, nil + } + + var ( + preserveTenderID string + preserveProjectName string + preserveRelatedPubIDs []string + previousNoticePublicationID string + previousStatus tender.TenderStatus + ) + if !t.ID.IsZero() { preserveTenderID = t.TenderID preserveProjectName = t.ProjectName + preserveRelatedPubIDs = append(preserveRelatedPubIDs, t.RelatedNoticePublicationIDs...) + previousNoticePublicationID = t.NoticePublicationID + previousStatus = t.Status } // Buyer Organization @@ -374,47 +436,9 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) { } } - var title string - if w.GLM != nil { - title, err = w.GLM.Translate(context.Background(), n.Title, n.NoticeLanguageCode, "en") - if err != nil { - w.Logger.Error("Failed to translate title", map[string]interface{}{ - "notice_publication_id": n.NoticePublicationID, - "country_code": n.CountryCode, - "content": n.Title, - "error": err.Error(), - }) - title = n.Title - } - } else { - w.Logger.Warn("GLM service not available, using original title", map[string]interface{}{ - "notice_publication_id": n.NoticePublicationID, - "country_code": n.CountryCode, - }) - title = n.Title - } - - var description string - if w.GLM != nil { - description, err = w.GLM.Translate(context.Background(), n.Description, n.NoticeLanguageCode, "en") - if err != nil { - w.Logger.Error("Failed to translate description", map[string]interface{}{ - "notice_publication_id": n.NoticePublicationID, - "country_code": n.CountryCode, - "content": n.Description, - "error": err.Error(), - }) - description = n.Description - } - } else { - w.Logger.Warn("GLM service not available, using original description", map[string]interface{}{ - "notice_publication_id": n.NoticePublicationID, - "country_code": n.CountryCode, - }) - description = n.Description - } - t.Title = title - t.Description = description + // Title and description stay in the notice language; English translation is handled by the AI service (translation worker). + t.Title = n.Title + t.Description = n.Description t.NoticePublicationID = n.NoticePublicationID t.ProcurementTypeCode = n.ProcurementTypeCode t.ProcedureCode = n.ProcedureCode @@ -455,7 +479,7 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) { t.MainClassification = n.MainClassification t.AdditionalClassifications = n.AdditionalClassifications t.OfficialLanguages = n.OfficialLanguages - t.Status = tender.TenderStatus(n.Status) + t.Status = chooseTenderStatus(previousStatus, tender.TenderStatus(n.Status)) t.Source = tender.TenderSource(n.Source) t.ContractNoticeID = n.ContractNoticeID t.ContractFolderID = n.ContractFolderID @@ -491,9 +515,151 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) { t.ProjectName = w.GenerateProjectName(t) } + t.RelatedNoticePublicationIDs = mergeRelatedPublicationIDs( + preserveRelatedPubIDs, + previousNoticePublicationID, + n.NoticePublicationID, + ) + return t, nil } +// resolveExistingTender finds the tender that this notice should update, preferring the procedure-level +// ContractFolderID (so follow-up notices like a Result for an earlier Competition notice merge into the +// same tender) and falling back to NoticePublicationID for older rows or notices without a folder id. +func (w *NoticeWorker) resolveExistingTender(ctx context.Context, n *notice.Notice) *tender.Tender { + if strings.TrimSpace(n.ContractFolderID) != "" { + if t, err := w.TenderRepo.GetByContractFolderID(ctx, n.ContractFolderID); err == nil && t != nil { + return t + } + } + if strings.TrimSpace(n.NoticePublicationID) != "" { + if t, err := w.TenderRepo.GetByNoticePublicationID(ctx, n.NoticePublicationID); err == nil && t != nil { + return t + } + } + return nil +} + +// isStaleNotice reports whether an incoming notice is older than what the existing tender already +// reflects, and therefore should not overwrite it. Two cases qualify as stale: +// 1. Same TED contract notice id (a re-scrape of the same document) with an older BT-757 NoticeVersion. +// 2. A strictly earlier IssueDate than the tender's current IssueDate (an out-of-order publication on +// the same procedure, e.g. a stale CN arriving after a CAN has already been ingested). +func isStaleNotice(existing *tender.Tender, n *notice.Notice) bool { + if existing == nil || existing.ID.IsZero() { + return false + } + if existing.ContractNoticeID != "" && existing.ContractNoticeID == n.ContractNoticeID { + if compareNoticeVersions(n.NoticeVersion, existing.NoticeVersion) < 0 { + return true + } + } + if existing.IssueDate > 0 && n.IssueDate > 0 && n.IssueDate < existing.IssueDate { + return true + } + return false +} + +// parseNoticeVersion parses the BT-757 numeric notice version (e.g. "01", "02"). Empty or unparsable +// strings yield 0, which matches the convention used on the notice-ingest side. +func parseNoticeVersion(s string) int { + s = strings.TrimSpace(s) + if s == "" { + return 0 + } + v, err := strconv.Atoi(s) + if err != nil { + return 0 + } + return v +} + +// compareNoticeVersions returns -1/0/1 for incoming vs existing notice versions. +func compareNoticeVersions(incoming, existing string) int { + vi := parseNoticeVersion(incoming) + ve := parseNoticeVersion(existing) + switch { + case vi > ve: + return 1 + case vi < ve: + return -1 + default: + return 0 + } +} + +// chooseTenderStatus enforces lifecycle precedence so a stale earlier-stage notice cannot downgrade a +// tender that has already reached a later stage. The ranks follow the TED lifecycle: +// draft < active/published < modified < suspended < expired < closed < awarded < cancelled. +// Cancellation is the strongest terminal state so an Awarded tender can still be moved to Cancelled +// (a procedure can be cancelled post-award), but a stale Active CN cannot downgrade an Awarded tender. +// Equal-or-higher incoming ranks are accepted; strictly lower ranks are rejected. +func chooseTenderStatus(existing, incoming tender.TenderStatus) tender.TenderStatus { + if existing == "" { + return incoming + } + if incoming == "" { + return existing + } + if tenderStatusRank(existing) > tenderStatusRank(incoming) { + return existing + } + return incoming +} + +func tenderStatusRank(s tender.TenderStatus) int { + switch s { + case tender.TenderStatusCancelled: + return 7 + case tender.TenderStatusAwarded: + return 6 + case tender.TenderStatusClosed: + return 5 + case tender.TenderStatusExpired: + return 4 + case tender.TenderStatusSuspended: + return 3 + case tender.TenderStatusModified: + return 2 + case tender.TenderStatusActive, tender.TenderStatusPublished: + return 1 + case tender.TenderStatusDraft: + return 0 + default: + return 1 + } +} + +// mergeRelatedPublicationIDs appends previous and current publication ids to the existing history, +// preserving order and deduplicating. The latest publication id (passed via current) lives on +// Tender.NoticePublicationID; this list tracks every publication that contributed to this tender so +// historical TED detail URLs stay resolvable. +func mergeRelatedPublicationIDs(existing []string, previous, current string) []string { + seen := make(map[string]struct{}, len(existing)+2) + out := make([]string, 0, len(existing)+2) + add := func(id string) { + id = strings.TrimSpace(id) + if id == "" { + return + } + if _, ok := seen[id]; ok { + return + } + seen[id] = struct{}{} + out = append(out, id) + } + for _, id := range existing { + add(id) + } + add(previous) + add(current) + if len(out) == 0 { + return nil + } + return out +} + // GenerateTenderID generates a unique tender ID using the PBL naming convention: SCDYYNNN // SCD = Source Code (3 letters), YY = Year (2 digits), NNN = Sequential number (variable digits) func (w *NoticeWorker) GenerateTenderID(ctx context.Context, t *notice.Notice) string { @@ -575,16 +741,80 @@ func cleanForProjectName(s string) string { return clean } -// getNextSequentialNumber returns the next sequential number for the given source code and year +// getNextSequentialNumber atomically allocates the next sequential number for the given +// (sourceCode, year) pair via a Mongo counter document. The counter is seeded once per process +// per key from the current max tender_id in the tenders collection, then incremented atomically +// with $inc so two parallel goroutines (or processes) can never get the same value. func (w *NoticeWorker) getNextSequentialNumber(ctx context.Context, sourceCode, year string) int { - // Query tenders with tender IDs that match the pattern: {sourceCode}{year}* - // We need to find the highest NNN value for this source code and year + key := fmt.Sprintf("%s%s", sourceCode, year) - // Pattern to match: e.g., "PTD25" followed by one or more digits + if err := w.ensureTenderIDCounterSeeded(ctx, sourceCode, year, key); err != nil { + w.Logger.Error("Failed to seed tender id counter, falling back to live max query", map[string]interface{}{ + "error": err.Error(), + "key": key, + }) + return w.queryMaxTenderSeq(ctx, sourceCode, year) + 1 + } + + var res struct { + Seq int `bson:"seq"` + } + err := w.Mongo.GetCollection(tenderIDCountersCollection).FindOneAndUpdate( + ctx, + bson.M{"_id": key}, + bson.M{"$inc": bson.M{"seq": 1}}, + options.FindOneAndUpdate().SetUpsert(true).SetReturnDocument(options.After), + ).Decode(&res) + if err != nil { + w.Logger.Error("Failed to increment tender id counter, falling back to live max query", map[string]interface{}{ + "error": err.Error(), + "key": key, + }) + return w.queryMaxTenderSeq(ctx, sourceCode, year) + 1 + } + return res.Seq +} + +// ensureTenderIDCounterSeeded sets the counter to at least the current max sequential number +// observed in the tenders collection. It is idempotent (uses $max) and only runs once per key +// per process. The first call across all processes pays the aggregation cost; subsequent calls +// are no-ops at the process level and (worst case) one extra cheap $max operation at Mongo. +func (w *NoticeWorker) ensureTenderIDCounterSeeded(ctx context.Context, sourceCode, year, key string) error { + w.tenderIDSeedMu.Lock() + defer w.tenderIDSeedMu.Unlock() + if w.tenderIDSeeded[key] { + return nil + } + + currentMax := w.queryMaxTenderSeq(ctx, sourceCode, year) + _, err := w.Mongo.GetCollection(tenderIDCountersCollection).UpdateOne( + ctx, + bson.M{"_id": key}, + bson.M{ + "$max": bson.M{"seq": currentMax}, + "$setOnInsert": bson.M{"_id": key}, + }, + options.UpdateOne().SetUpsert(true), + ) + if err != nil { + return fmt.Errorf("seed counter %s: %w", key, err) + } + + w.tenderIDSeeded[key] = true + w.Logger.Info("Tender id counter seeded", map[string]interface{}{ + "key": key, + "seed_value": currentMax, + "next_id": fmt.Sprintf("%s%d", key, currentMax+1), + }) + return nil +} + +// queryMaxTenderSeq returns the highest NNN suffix found in the tenders collection for tender_ids +// matching {sourceCode}{year}NNN, or 0 if none exist. Used to seed the counter. +func (w *NoticeWorker) queryMaxTenderSeq(ctx context.Context, sourceCode, year string) int { prefix := fmt.Sprintf("%s%s", sourceCode, year) pattern := fmt.Sprintf("^%s\\d+$", prefix) - // Create aggregation pipeline to find the highest sequential number pipeline := mongodriver.Pipeline{ bson.D{{Key: "$match", Value: bson.M{ "tender_id": bson.M{"$regex": pattern}, @@ -598,7 +828,6 @@ func (w *NoticeWorker) getNextSequentialNumber(ctx context.Context, sourceCode, }}}, } - // Execute aggregation cursor, err := w.Mongo.GetCollection("tenders").Aggregate(ctx, pipeline) if err != nil { w.Logger.Error("Failed to aggregate tenders for sequential numbering", map[string]interface{}{ @@ -606,22 +835,80 @@ func (w *NoticeWorker) getNextSequentialNumber(ctx context.Context, sourceCode, "source_code": sourceCode, "year": year, }) - return 1 // Return 1 as fallback + return 0 } defer cursor.Close(ctx) - // Parse result var result struct { MaxNum int `bson:"max_num"` } - if cursor.Next(ctx) { if err := cursor.Decode(&result); err == nil { - return result.MaxNum + 1 + return result.MaxNum } } + return 0 +} - return 1 // No existing tenders found, start with 1 +// persistTender writes the given tender to Mongo. For new tenders (zero ID) it retries on +// duplicate-key errors against tender_id by allocating a fresh sequential number; this guards +// against a stale counter or external writes that bypass the counter. For existing tenders it +// is a straight update. +func (w *NoticeWorker) persistTender(ctx context.Context, t *tender.Tender, n *notice.Notice) error { + if !t.ID.IsZero() { + return w.TenderRepo.Update(ctx, t) + } + + var lastErr error + for attempt := 0; attempt < tenderIDMaxRetries; attempt++ { + err := w.TenderRepo.Create(ctx, t) + if err == nil { + return nil + } + lastErr = err + if !isDuplicateTenderIDError(err) { + return err + } + + // Re-allocate a tender id and refresh the project name. Reset Mongo's ObjectID so the + // repo treats this as a fresh insert on the next attempt. + t.ID = bson.NilObjectID + t.TenderID = w.GenerateTenderID(ctx, n) + t.ProjectName = w.GenerateProjectName(t) + + w.Logger.Warn("Duplicate tender_id on insert, retrying with new id", map[string]interface{}{ + "attempt": attempt + 1, + "new_tender_id": t.TenderID, + "notice_id": n.ID.Hex(), + "contract_folder": n.ContractFolderID, + }) + } + return fmt.Errorf("exhausted %d tender_id retries: %w", tenderIDMaxRetries, lastErr) +} + +// isDuplicateTenderIDError returns true if err is the Mongo "E11000 duplicate key" error on the +// tender_id_1 index. We retry only that specific case; other write errors propagate. +func isDuplicateTenderIDError(err error) bool { + if err == nil { + return false + } + // Cheap, driver-version-agnostic check: look for the error code and the specific index. + msg := err.Error() + if !strings.Contains(msg, "E11000") { + return false + } + if strings.Contains(msg, "tender_id_1") || strings.Contains(msg, "tender_id:") { + return true + } + var we mongodriver.WriteException + if errors.As(err, &we) { + for _, wErr := range we.WriteErrors { + if wErr.Code == 11000 { + return true + } + } + } + return false } // padOrTruncate pads a string with zeros or truncates it to the specified length diff --git a/cmd/worker/workers/summarizer.go b/cmd/worker/workers/summarizer.go index b75b385..8b3df6a 100644 --- a/cmd/worker/workers/summarizer.go +++ b/cmd/worker/workers/summarizer.go @@ -2,46 +2,53 @@ package workers import ( "context" - "fmt" - "io" "path/filepath" "strings" "time" "tm/internal/tender" - "tm/pkg/glm" + ai_summarizer "tm/pkg/ai_summarizer" "tm/pkg/logger" "tm/pkg/minio" "tm/pkg/mongo" ) -// DocumentSummarizationWorker handles document summarization using GLM AI +const presignedDocumentURLSeconds int64 = 7200 + +// DocumentSummarizationWorker asks the external AI service to summarize scraped tender documents. type DocumentSummarizationWorker struct { Mongo *mongo.ConnectionManager Logger logger.Logger TenderRepo tender.TenderRepository MinioSDK *minio.Service - GLM *glm.SDK + AIClient *ai_summarizer.Client } -// NewDocumentSummarizationWorker creates a new document summarization worker -func NewDocumentSummarizationWorker(mongo *mongo.ConnectionManager, logger logger.Logger, tenderRepo tender.TenderRepository, minioSDK *minio.Service, glmSDK *glm.SDK) *DocumentSummarizationWorker { +// 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 { return &DocumentSummarizationWorker{ Mongo: mongo, Logger: logger, TenderRepo: tenderRepo, MinioSDK: minioSDK, - GLM: glmSDK, + AIClient: aiClient, } } -// Run starts the document summarization process +// Run starts the document summarization process. func (w *DocumentSummarizationWorker) Run() { w.Logger.Info("Document summarization worker started", map[string]interface{}{}) - limit := 5 // Process fewer tenders at a time since summarization is resource-intensive + if w.AIClient == nil { + 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 + } + + limit := 5 for { - // Always use skip=0 because processed items are excluded by the filter, - // so incrementing skip would cause unprocessed items to be skipped. tenders, totalCount, err := w.TenderRepo.GetUnSummarizedTenders(context.Background(), limit, 0) if err != nil { w.Logger.Error("Failed to get un-summarized tenders", map[string]interface{}{ @@ -60,32 +67,29 @@ func (w *DocumentSummarizationWorker) Run() { break } - for _, t := range tenders { + for i := range tenders { + t := &tenders[i] 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.summarizeDocumentsForTender(t) } } w.Logger.Info("Document summarization worker completed", map[string]interface{}{}) } -// summarizeDocumentsForTender summarizes all documents for a single tender 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, }) - - // Mark as processed even if no documents (nothing to summarize) t.ProcessingMetadata.DocumentsSummarized = true t.ProcessingMetadata.DocumentsSummarizedAt = time.Now().Unix() - w.updateTenderSummarizationStatus(t) + _ = w.updateTenderSummarizationStatus(t) return } @@ -93,17 +97,15 @@ func (w *DocumentSummarizationWorker) summarizeDocumentsForTender(t *tender.Tend summarizedAt := time.Now().Unix() for _, doc := range t.ScrapedDocuments { - summary, err := w.summarizeDocument(doc, t.NoticeLanguageCode) - if err != nil { + 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": err.Error(), + "error": errStr, }) - - // Add summary with error summaries = append(summaries, tender.DocumentSummary{ DocumentName: doc.Filename, ObjectName: doc.ObjectName, @@ -112,8 +114,8 @@ func (w *DocumentSummarizationWorker) summarizeDocumentsForTender(t *tender.Tend SummaryLanguage: "en", DocumentType: w.getDocumentType(doc.Filename), SummarizedAt: summarizedAt, - SummaryModel: w.getGLMModelName(), - Error: err.Error(), + SummaryModel: model, + Error: errStr, }) continue } @@ -122,28 +124,26 @@ func (w *DocumentSummarizationWorker) summarizeDocumentsForTender(t *tender.Tend DocumentName: doc.Filename, ObjectName: doc.ObjectName, BucketName: doc.BucketName, - Summary: summary, + Summary: summaryText, SummaryLanguage: "en", DocumentType: w.getDocumentType(doc.Filename), SummarizedAt: summarizedAt, - SummaryModel: w.getGLMModelName(), + SummaryModel: model, }) w.Logger.Info("Document summarized successfully", map[string]interface{}{ "tender_id": t.ID.Hex(), "notice_publication_id": t.NoticePublicationID, "document_name": doc.Filename, - "summary_length": len(summary), + "summary_length": len(summaryText), }) } - // Update tender with summaries t.DocumentSummaries = summaries t.ProcessingMetadata.DocumentsSummarized = true t.ProcessingMetadata.DocumentsSummarizedAt = summarizedAt - err := w.updateTenderSummarizationStatus(t) - if err != nil { + if err := w.updateTenderSummarizationStatus(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, @@ -160,162 +160,74 @@ func (w *DocumentSummarizationWorker) summarizeDocumentsForTender(t *tender.Tend }) } -// summarizeDocument downloads a document from MinIO and summarizes it using GLM -func (w *DocumentSummarizationWorker) summarizeDocument(doc tender.ScrapedDocument, sourceLanguage string) (string, error) { - // Download document from MinIO - documentData, err := w.MinioSDK.Download(context.Background(), doc.BucketName, doc.ObjectName, minio.DownloadOptions{}) +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 "", fmt.Errorf("failed to download document from MinIO: %w", err) + return "", "", err.Error() } - defer documentData.Close() - // Read document content - documentBytes, err := io.ReadAll(documentData) + noticeID := strings.TrimSpace(t.NoticePublicationID) + if noticeID == "" { + return "", "", "tender has no notice_publication_id for AI summarize request" + } + + req := ai_summarizer.SummarizeRequest{ + NoticeID: noticeID, + DocumentURL: docURL, + Title: t.Title, + Description: t.Description, + Country: t.CountryCode, + Currency: t.Currency, + SubmissionDeadline: "", + AuthorityName: authorityNameFromTender(t), + } + if t.EstimatedValue > 0 { + v := t.EstimatedValue + req.EstimatedValue = &v + } + + resp, err := w.AIClient.FetchSummaryOnDemand(ctx, req) if err != nil { - return "", fmt.Errorf("failed to read document content: %w", err) + return "", "", err.Error() + } + if resp == nil { + return "", "", "AI summarize returned nil response" } - // Extract text from document based on type - textContent, err := w.extractTextFromDocument(documentBytes, doc.Filename) - if err != nil { - return "", fmt.Errorf("failed to extract text from document: %w", err) + 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 len(textContent) == 0 { - return "", fmt.Errorf("no text content extracted from document") + if summaryText == "" { + return "", model, "AI summarize returned no summary text for document" } - - // Limit text length for summarization (GLM has token limits) - maxLength := 10000 // Adjust based on GLM model limits - if len(textContent) > maxLength { - textContent = textContent[:maxLength] + "..." - } - - // Create summarization prompt - prompt := w.createSummarizationPrompt(textContent, sourceLanguage) - - // Use GLM to summarize - summary, err := w.GLM.ChatCompletion(context.Background(), &glm.ChatCompletionRequest{ - Messages: []glm.Message{ - { - Role: glm.MessageRoleUser, - Content: prompt, - }, - }, - Temperature: 0.3, // Lower temperature for more consistent summaries - MaxTokens: 1000, - }) - if err != nil { - // Check if GLM is configured (has API key) - if strings.Contains(err.Error(), "token expired") || strings.Contains(err.Error(), "AUTHENTICATION_ERROR") { - // GLM is configured but API key is invalid - return error instead of mock - return "", fmt.Errorf("GLM API authentication failed: %w", err) - } - - // GLM is not configured or other error - use mock summary for development - w.Logger.Warn("GLM API failed, using mock summary for development", map[string]interface{}{ - "error": err.Error(), - }) - return fmt.Sprintf("Mock summary for document: %s (GLM API not configured)", doc.Filename), nil - } - - if len(summary.Choices) == 0 || summary.Choices[0].Message.Content == "" { - return "", fmt.Errorf("GLM API returned empty response for document: %s", doc.Filename) - } - - return strings.TrimSpace(summary.Choices[0].Message.Content), nil + return summaryText, model, "" } -// extractTextFromDocument extracts text content from various document formats -func (w *DocumentSummarizationWorker) extractTextFromDocument(data []byte, filename string) (string, error) { - ext := strings.ToLower(filepath.Ext(filename)) - - switch ext { - case ".xlsx": - return w.extractTextFromExcel(data) - case ".pdf": - return w.extractTextFromPDF(data) - case ".docx": - return w.extractTextFromDOCX(data) - case ".txt": - return string(data), nil - case ".html", ".htm": - return w.extractTextFromHTML(data) - default: - // Try to extract as plain text for unknown formats - return string(data), nil +func authorityNameFromTender(t *tender.Tender) string { + if t == nil || t.BuyerOrganization == nil { + return "" } + return strings.TrimSpace(t.BuyerOrganization.Name) } -// extractTextFromPDF extracts text from PDF documents -func (w *DocumentSummarizationWorker) extractTextFromPDF(data []byte) (string, error) { - // For now, return a placeholder. In production, you'd use a PDF parsing library - // like github.com/unidoc/unipdf or github.com/rsc/pdf - return fmt.Sprintf("PDF content extraction not implemented. File size: %d bytes", len(data)), nil -} - -// extractTextFromDOCX extracts text from DOCX documents -func (w *DocumentSummarizationWorker) extractTextFromDOCX(data []byte) (string, error) { - // For now, return a placeholder. In production, you'd use a DOCX parsing library - // like github.com/lukasjarosch/go-docx or github.com/fumiama/go-docx - return fmt.Sprintf("DOCX content extraction not implemented. File size: %d bytes", len(data)), nil -} - -// extractTextFromExcel extracts text from Excel documents (.xlsx) -func (w *DocumentSummarizationWorker) extractTextFromExcel(data []byte) (string, error) { - // For now, return a placeholder. In production, you'd use an Excel parsing library - // like github.com/tealeg/xlsx/v3 or github.com/360EntSecGroup-Skylar/excelize - return fmt.Sprintf("Excel content extraction not implemented. File size: %d bytes", len(data)), nil -} - -// extractTextFromHTML extracts text from HTML documents -func (w *DocumentSummarizationWorker) extractTextFromHTML(data []byte) (string, error) { - // Simple HTML text extraction (remove tags) - htmlContent := string(data) - - // Basic HTML tag removal (very simple implementation) - // In production, use a proper HTML parser like goquery - htmlContent = strings.ReplaceAll(htmlContent, "") - htmlContent = strings.ReplaceAll(htmlContent, "") - - // Remove HTML tags using simple regex-like approach - var result strings.Builder - inTag := false - for _, r := range htmlContent { - switch r { - case '<': - inTag = true - case '>': - inTag = false - default: - if !inTag { - result.WriteRune(r) - } +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) } } - - return strings.TrimSpace(result.String()), nil + for _, s := range items { + if strings.TrimSpace(s.Summary) != "" { + return strings.TrimSpace(s.Summary), strings.TrimSpace(s.SummaryModel) + } + } + return "", "" } -// createSummarizationPrompt creates a prompt for GLM to summarize tender documents -func (w *DocumentSummarizationWorker) createSummarizationPrompt(textContent, sourceLanguage string) string { - return fmt.Sprintf(`Please provide a concise summary of the following tender document in English. Focus on key information such as: - -1. Project/Tender name and description -2. Contracting authority/organization details -3. Key requirements and specifications -4. Important dates (deadline, award date, etc.) -5. Estimated value and currency -6. Selection criteria -7. Contact information - -Document content: -%s - -Summary:`, textContent) -} - -// getDocumentType returns the document type based on file extension func (w *DocumentSummarizationWorker) getDocumentType(filename string) string { ext := strings.ToLower(filepath.Ext(filename)) switch ext { @@ -336,12 +248,6 @@ func (w *DocumentSummarizationWorker) getDocumentType(filename string) string { } } -// getGLMModelName returns the GLM model name being used -func (w *DocumentSummarizationWorker) getGLMModelName() string { - return "glm-4.5" -} - -// updateTenderSummarizationStatus updates the tender's document summarization metadata func (w *DocumentSummarizationWorker) updateTenderSummarizationStatus(tender *tender.Tender) error { err := w.TenderRepo.Update(context.Background(), tender) if err != nil { diff --git a/go.mod b/go.mod index 003756d..65bcfc1 100644 --- a/go.mod +++ b/go.mod @@ -68,7 +68,7 @@ require ( github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect go.uber.org/multierr v1.10.0 // indirect golang.org/x/net v0.42.0 // indirect - golang.org/x/sync v0.16.0 // indirect + golang.org/x/sync v0.16.0 golang.org/x/sys v0.34.0 // indirect golang.org/x/text v0.27.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/internal/customer/handler.go b/internal/customer/handler.go index 7a54374..132b238 100644 --- a/internal/customer/handler.go +++ b/internal/customer/handler.go @@ -403,7 +403,7 @@ func (h *Handler) GetProfile(c echo.Context) error { // UpdateProfileKeywords updates customer profile keywords (Mobile) // @Summary Update customer profile keywords -// @Description Update profile keywords manually and trigger asynchronous GLM enhancement based on customer company information (including country). +// @Description Update profile keywords manually; optional async keyword refresh is reserved for the AI service. // @Tags Authorization // @Accept json // @Produce json diff --git a/internal/customer/service.go b/internal/customer/service.go index 0e15209..ea54daa 100644 --- a/internal/customer/service.go +++ b/internal/customer/service.go @@ -11,7 +11,6 @@ import ( "time" "tm/internal/company" - "tm/pkg/glm" "tm/pkg/logger" "tm/pkg/notification" "tm/pkg/redis" @@ -63,11 +62,10 @@ type customerService struct { redisClient redis.Client notification notification.SDK validator ValidationService - glmSDK *glm.SDK } // New creates a new customer service -func NewService(repository Repository, logger logger.Logger, authService authorization.AuthorizationService, companyService company.Service, redisClient redis.Client, notificationSDK notification.SDK, validator ValidationService, glmSDK *glm.SDK) Service { +func NewService(repository Repository, logger logger.Logger, authService authorization.AuthorizationService, companyService company.Service, redisClient redis.Client, notificationSDK notification.SDK, validator ValidationService) Service { return &customerService{ repository: repository, logger: logger, @@ -76,7 +74,6 @@ func NewService(repository Repository, logger logger.Logger, authService authori redisClient: redisClient, notification: notificationSDK, validator: validator, - glmSDK: glmSDK, } } @@ -134,7 +131,7 @@ func (s *customerService) Register(ctx context.Context, form *CreateCustomerForm Phone: form.Phone, } - // Generate keywords using GLM AI + // Keyword auto-generation is handled by the AI service when integrated; backend leaves keywords empty on register. customer.KeywordsStatus = KeywordsStatusInProgress keywords, err := s.generateKeywords(ctx, customer) if err != nil { @@ -1349,99 +1346,11 @@ func (s *customerService) AssignRole(ctx context.Context, customerID string, for }, nil } -// generateKeywords generates keywords for a customer using GLM AI based on their information +// generateKeywords is reserved for future integration with the AI service; the backend does not generate keywords. func (s *customerService) generateKeywords(ctx context.Context, customer *Customer) ([]string, error) { - // If GLM SDK is not available, return empty keywords - if s.glmSDK == nil { - s.logger.Warn("GLM SDK not available, skipping keyword generation", map[string]interface{}{ - "customer_id": customer.ID, - }) - return []string{}, nil - } - - // Build customer information string for AI - var customerInfo strings.Builder - if customer.FullName != nil { - customerInfo.WriteString(fmt.Sprintf("Name: %s\n", *customer.FullName)) - } - customerInfo.WriteString(fmt.Sprintf("Type: %s\n", customer.Type)) - customerInfo.WriteString(fmt.Sprintf("Role: %s\n", customer.Role)) - if customer.Phone != nil { - customerInfo.WriteString(fmt.Sprintf("Phone: %s\n", *customer.Phone)) - } - - // Get company information if available - if len(customer.Companies) > 0 { - customerInfo.WriteString("Companies: ") - for i, companyID := range customer.Companies { - comp, err := s.companyService.GetByID(ctx, companyID) - if err == nil { - if i > 0 { - customerInfo.WriteString(", ") - } - customerInfo.WriteString(comp.Name) - if comp.Industry != "" { - customerInfo.WriteString(fmt.Sprintf(" (Industry: %s)", comp.Industry)) - } - if comp.Address != nil && comp.Address.Country != "" { - customerInfo.WriteString(fmt.Sprintf(" (Country: %s)", comp.Address.Country)) - } - if comp.Tags != nil && len(comp.Tags.Keywords) > 0 { - customerInfo.WriteString(fmt.Sprintf(" (Company Keywords: %s)", strings.Join(comp.Tags.Keywords, ", "))) - } - } - } - customerInfo.WriteString("\n") - } - - systemPrompt := `You are a keyword extraction assistant for a tender management system. -Analyze the customer information and generate 5-10 relevant keywords that would help match them with appropriate tenders. -Keywords should be: -- Relevant to tender/procurement categories -- Based on customer type, role, and company information -- Professional and specific -- Suitable for matching with tender descriptions - -Return ONLY a comma-separated list of keywords, no explanations or additional text.` - - userMessage := fmt.Sprintf("Generate keywords for this customer:\n\n%s", customerInfo.String()) - - s.logger.Info("Generating keywords using GLM AI", map[string]interface{}{ - "customer_id": customer.ID, - }) - - response, err := s.glmSDK.QuickChatWithSystem(ctx, systemPrompt, userMessage) - if err != nil { - s.logger.Error("Failed to generate keywords using GLM AI", map[string]interface{}{ - "error": err.Error(), - "customer_id": customer.ID, - }) - return []string{}, fmt.Errorf("failed to generate keywords: %w", err) - } - - // Parse comma-separated keywords - keywordsStr := strings.TrimSpace(response) - if keywordsStr == "" { - return []string{}, nil - } - - // Split by comma and clean up - keywordList := strings.Split(keywordsStr, ",") - keywords := make([]string, 0, len(keywordList)) - for _, kw := range keywordList { - kw = strings.TrimSpace(kw) - if kw != "" { - keywords = append(keywords, kw) - } - } - - s.logger.Info("Keywords generated successfully", map[string]interface{}{ - "customer_id": customer.ID, - "keywords": keywords, - "count": len(keywords), - }) - - return keywords, nil + _ = ctx + _ = customer + return []string{}, nil } func (s *customerService) processKeywordsAsync(customerID string) { @@ -1456,34 +1365,26 @@ func (s *customerService) processKeywordsAsync(customerID string) { return } - if s.glmSDK == nil { - s.logger.Info("GLM SDK not available, keeping manual keywords", map[string]interface{}{ + keywords, err := s.generateKeywords(ctx, cust) + if err != nil { + s.logger.Warn("Failed to process keywords after manual update", map[string]interface{}{ + "error": err.Error(), + "customer_id": customerID, + }) + cust.KeywordsStatus = KeywordsStatusFailed + } else if len(keywords) > 0 { + cust.Keywords = keywords + cust.KeywordsStatus = KeywordsStatusReady + } else { + s.logger.Debug("No auto-generated keywords from backend, keeping existing keywords", map[string]interface{}{ "customer_id": customerID, }) cust.KeywordsStatus = KeywordsStatusReady - } else { - keywords, err := s.generateKeywords(ctx, cust) - if err != nil { - s.logger.Warn("Failed to process keywords after manual update", map[string]interface{}{ - "error": err.Error(), - "customer_id": customerID, - }) - cust.KeywordsStatus = KeywordsStatusFailed - } else if len(keywords) > 0 { - cust.Keywords = keywords - cust.KeywordsStatus = KeywordsStatusReady - } else { - s.logger.Warn("GLM returned empty keywords, keeping manual keywords", map[string]interface{}{ - "customer_id": customerID, - }) - cust.KeywordsStatus = KeywordsStatusReady - } } - err = s.repository.Update(ctx, cust) - if err != nil { + if updateErr := s.repository.Update(ctx, cust); updateErr != nil { s.logger.Error("Failed to update customer keywords status", map[string]interface{}{ - "error": err.Error(), + "error": updateErr.Error(), "customer_id": customerID, }) } else { diff --git a/internal/tender/entity.go b/internal/tender/entity.go index 2b712e7..d332cac 100644 --- a/internal/tender/entity.go +++ b/internal/tender/entity.go @@ -32,56 +32,59 @@ type ScrapedDocument struct { // Tender represents a tender/contract notice entity type Tender struct { - mongo.Model `bson:",inline"` - ContractNoticeID string `bson:"contract_notice_id" json:"contract_notice_id"` - NoticePublicationID string `bson:"notice_publication_id" json:"notice_publication_id"` - ContractFolderID string `bson:"contract_folder_id" json:"contract_folder_id"` - NoticeTypeCode string `bson:"notice_type_code" json:"notice_type_code"` - FormType string `bson:"form_type,omitempty" json:"form_type,omitempty"` - NoticeSubTypeCode string `bson:"notice_sub_type_code" json:"notice_sub_type_code"` - NoticeIdentifier string `bson:"notice_identifier,omitempty" json:"notice_identifier,omitempty"` - NoticeVersion string `bson:"notice_version,omitempty" json:"notice_version,omitempty"` - NoticeDispatchAt int64 `bson:"notice_dispatch_at,omitempty" json:"notice_dispatch_at,omitempty"` - ESenderDispatchAt int64 `bson:"esender_dispatch_at,omitempty" json:"esender_dispatch_at,omitempty"` - NoticeLanguageCode string `bson:"notice_language_code" json:"notice_language_code"` - IssueDate int64 `bson:"issue_date" json:"issue_date"` - IssueTime int64 `bson:"issue_time" json:"issue_time"` - TenderDeadline int64 `bson:"tender_deadline" json:"tender_deadline"` - PublicationDate int64 `bson:"publication_date" json:"publication_date"` - SubmissionDeadline int64 `bson:"submission_deadline" json:"submission_deadline"` - ApplicationDeadline int64 `bson:"application_deadline" json:"application_deadline"` - GazetteID string `bson:"gazette_id" json:"gazette_id"` - Title string `bson:"title" json:"title"` - Description string `bson:"description" json:"description"` - ProcurementTypeCode string `bson:"procurement_type_code" json:"procurement_type_code"` - ProcedureCode string `bson:"procedure_code" json:"procedure_code"` - MainClassification string `bson:"main_classification" json:"main_classification"` - AdditionalClassifications []string `bson:"additional_classifications" json:"additional_classifications"` - EstimatedValue float64 `bson:"estimated_value" json:"estimated_value"` - Currency string `bson:"currency" json:"currency"` - Duration string `bson:"duration" json:"duration"` - DurationUnit string `bson:"duration_unit" json:"duration_unit"` - PlaceOfPerformance string `bson:"place_of_performance" json:"place_of_performance"` - CountryCode string `bson:"country_code" json:"country_code"` - RegionCode string `bson:"region_code" json:"region_code"` - CityName string `bson:"city_name" json:"city_name"` - PostalCode string `bson:"postal_code" json:"postal_code"` - DocumentURI string `bson:"document_uri" json:"document_uri"` - BuyerProfileURL string `bson:"buyer_profile_url,omitempty" json:"buyer_profile_url,omitempty"` - ProcurementLots []ProcurementLot `bson:"procurement_lots,omitempty" json:"procurement_lots,omitempty"` - TenderURL string `bson:"tender_url" json:"tender_url"` - SubmissionURL string `bson:"submission_url" json:"submission_url"` - BuyerOrganization *Organization `bson:"buyer_organization" json:"buyer_organization"` - ReviewOrganization *Organization `bson:"review_organization,omitempty" json:"review_organization,omitempty"` - Organizations []Organization `bson:"organizations" json:"organizations"` - SelectionCriteria []SelectionCriterion `bson:"selection_criteria" json:"selection_criteria"` - OfficialLanguages []string `bson:"official_languages" json:"official_languages"` - Status TenderStatus `bson:"status" json:"status"` - Source TenderSource `bson:"source" json:"source"` - SourceFileURL string `bson:"source_file_url" json:"source_file_url"` - SourceFileName string `bson:"source_file_name" json:"source_file_name"` - ContentXML string `bson:"content_xml,omitempty" json:"content_xml,omitempty"` - ProcessingMetadata ProcessingMetadata `bson:"processing_metadata" json:"processing_metadata"` + mongo.Model `bson:",inline"` + ContractNoticeID string `bson:"contract_notice_id" json:"contract_notice_id"` + NoticePublicationID string `bson:"notice_publication_id" json:"notice_publication_id"` + // RelatedNoticePublicationIDs lists every TED notice publication id that has contributed to this tender + // (same procedure / contract_folder_id). Ordered oldest-first; the latest is also reflected in NoticePublicationID. + RelatedNoticePublicationIDs []string `bson:"related_notice_publication_ids,omitempty" json:"related_notice_publication_ids,omitempty"` + ContractFolderID string `bson:"contract_folder_id" json:"contract_folder_id"` + NoticeTypeCode string `bson:"notice_type_code" json:"notice_type_code"` + FormType string `bson:"form_type,omitempty" json:"form_type,omitempty"` + NoticeSubTypeCode string `bson:"notice_sub_type_code" json:"notice_sub_type_code"` + NoticeIdentifier string `bson:"notice_identifier,omitempty" json:"notice_identifier,omitempty"` + NoticeVersion string `bson:"notice_version,omitempty" json:"notice_version,omitempty"` + NoticeDispatchAt int64 `bson:"notice_dispatch_at,omitempty" json:"notice_dispatch_at,omitempty"` + ESenderDispatchAt int64 `bson:"esender_dispatch_at,omitempty" json:"esender_dispatch_at,omitempty"` + NoticeLanguageCode string `bson:"notice_language_code" json:"notice_language_code"` + IssueDate int64 `bson:"issue_date" json:"issue_date"` + IssueTime int64 `bson:"issue_time" json:"issue_time"` + TenderDeadline int64 `bson:"tender_deadline" json:"tender_deadline"` + PublicationDate int64 `bson:"publication_date" json:"publication_date"` + SubmissionDeadline int64 `bson:"submission_deadline" json:"submission_deadline"` + ApplicationDeadline int64 `bson:"application_deadline" json:"application_deadline"` + GazetteID string `bson:"gazette_id" json:"gazette_id"` + Title string `bson:"title" json:"title"` + Description string `bson:"description" json:"description"` + ProcurementTypeCode string `bson:"procurement_type_code" json:"procurement_type_code"` + ProcedureCode string `bson:"procedure_code" json:"procedure_code"` + MainClassification string `bson:"main_classification" json:"main_classification"` + AdditionalClassifications []string `bson:"additional_classifications" json:"additional_classifications"` + EstimatedValue float64 `bson:"estimated_value" json:"estimated_value"` + Currency string `bson:"currency" json:"currency"` + Duration string `bson:"duration" json:"duration"` + DurationUnit string `bson:"duration_unit" json:"duration_unit"` + PlaceOfPerformance string `bson:"place_of_performance" json:"place_of_performance"` + CountryCode string `bson:"country_code" json:"country_code"` + RegionCode string `bson:"region_code" json:"region_code"` + CityName string `bson:"city_name" json:"city_name"` + PostalCode string `bson:"postal_code" json:"postal_code"` + DocumentURI string `bson:"document_uri" json:"document_uri"` + BuyerProfileURL string `bson:"buyer_profile_url,omitempty" json:"buyer_profile_url,omitempty"` + ProcurementLots []ProcurementLot `bson:"procurement_lots,omitempty" json:"procurement_lots,omitempty"` + TenderURL string `bson:"tender_url" json:"tender_url"` + SubmissionURL string `bson:"submission_url" json:"submission_url"` + BuyerOrganization *Organization `bson:"buyer_organization" json:"buyer_organization"` + ReviewOrganization *Organization `bson:"review_organization,omitempty" json:"review_organization,omitempty"` + Organizations []Organization `bson:"organizations" json:"organizations"` + SelectionCriteria []SelectionCriterion `bson:"selection_criteria" json:"selection_criteria"` + OfficialLanguages []string `bson:"official_languages" json:"official_languages"` + Status TenderStatus `bson:"status" json:"status"` + Source TenderSource `bson:"source" json:"source"` + SourceFileURL string `bson:"source_file_url" json:"source_file_url"` + SourceFileName string `bson:"source_file_name" json:"source_file_name"` + ContentXML string `bson:"content_xml,omitempty" json:"content_xml,omitempty"` + ProcessingMetadata ProcessingMetadata `bson:"processing_metadata" json:"processing_metadata"` // AIOverallSummary optional denormalized overall summary for search/list (e.g. synced from AI pipeline storage). AIOverallSummary string `bson:"ai_overall_summary,omitempty" json:"ai_overall_summary,omitempty"` TenderID string `bson:"tender_id" json:"tender_id"` // PBL Project Number (SCDYYNNN format) diff --git a/internal/tender/repository.go b/internal/tender/repository.go index fc40577..e2ef240 100644 --- a/internal/tender/repository.go +++ b/internal/tender/repository.go @@ -20,6 +20,7 @@ type TenderRepository interface { GetByID(ctx context.Context, id string) (*Tender, error) GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Tender, error) GetByNoticePublicationID(ctx context.Context, noticePublicationID string) (*Tender, error) + GetByContractFolderID(ctx context.Context, contractFolderID string) (*Tender, error) FindTendersWithContentXML(ctx context.Context, limit, skip int) ([]Tender, error) GetTenderCountByCountry(ctx context.Context) (map[string]int64, error) GetTenderCountByType(ctx context.Context) (map[string]int64, error) @@ -52,6 +53,7 @@ func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Te *orm.CreateUniqueIndex("tender_id_idx", bson.D{{Key: "tender_id", Value: 1}}), *orm.NewIndex("project_name_idx", bson.D{{Key: "project_name", Value: 1}}), *orm.NewIndex("contract_notice_id_idx", bson.D{{Key: "contract_notice_id", Value: 1}}), + *orm.NewIndex("contract_folder_id_idx", bson.D{{Key: "contract_folder_id", Value: 1}}), *orm.NewIndex("status_idx", bson.D{{Key: "status", Value: 1}}), *orm.NewIndex("source_idx", bson.D{{Key: "source", Value: 1}}), *orm.NewIndex("country_code_idx", bson.D{{Key: "country_code", Value: 1}}), @@ -158,9 +160,16 @@ func (r *tenderRepository) GetByContractNoticeID(ctx context.Context, contractNo return &result.Items[0], nil } -// GetByNoticePublicationID retrieves a tender by notice publication ID +// GetByNoticePublicationID retrieves a tender by notice publication ID. +// Matches both the canonical NoticePublicationID and any id retained in RelatedNoticePublicationIDs +// so historical TED detail links keep resolving after a tender update from a follow-up notice. func (r *tenderRepository) GetByNoticePublicationID(ctx context.Context, noticePublicationID string) (*Tender, error) { - filter := bson.M{"notice_publication_id": noticePublicationID} + filter := bson.M{ + "$or": []bson.M{ + {"notice_publication_id": noticePublicationID}, + {"related_notice_publication_ids": noticePublicationID}, + }, + } pagination := orm.Pagination{Limit: 1} result, err := r.ormRepo.FindAll(ctx, filter, pagination) @@ -179,6 +188,34 @@ func (r *tenderRepository) GetByNoticePublicationID(ctx context.Context, noticeP return &result.Items[0], nil } +// GetByContractFolderID retrieves a tender by its procedure-level identifier (ContractFolderID). +// TED groups multiple notice publications (e.g. competition notice + result notice) under the same +// ContractFolderID, so this lookup is the authoritative way to find the existing tender for a procedure +// when a follow-up notice arrives. Returns the most recently updated match if there are several. +func (r *tenderRepository) GetByContractFolderID(ctx context.Context, contractFolderID string) (*Tender, error) { + filter := bson.M{"contract_folder_id": contractFolderID} + pagination := orm.Pagination{ + Limit: 1, + SortField: "updated_at", + SortOrder: -1, + } + + result, err := r.ormRepo.FindAll(ctx, filter, pagination) + if err != nil { + r.logger.Error("Failed to get tender by contract folder ID", map[string]interface{}{ + "contract_folder_id": contractFolderID, + "error": err.Error(), + }) + return nil, err + } + + if len(result.Items) == 0 { + return nil, orm.ErrDocumentNotFound + } + + return &result.Items[0], nil +} + // FindTendersWithContentXML returns tenders that still have TED XML (for backfill after notices were deleted). func (r *tenderRepository) FindTendersWithContentXML(ctx context.Context, limit, skip int) ([]Tender, error) { filter := bson.M{"content_xml": bson.M{"$exists": true, "$ne": ""}} diff --git a/pkg/glm/README.md b/pkg/glm/README.md deleted file mode 100644 index 3f35d2c..0000000 --- a/pkg/glm/README.md +++ /dev/null @@ -1,205 +0,0 @@ -# GLM SDK - -A Go SDK for interacting with the GLM (Zhipu AI) API, specifically designed for chat completions and text generation tasks. - -## Features - -- **Chat Completions**: Perform conversational AI interactions -- **Streaming Support**: Real-time streaming responses -- **Translation Helper**: Built-in convenience methods for translation tasks -- **Model Management**: List available models -- **Health Checks**: Service availability monitoring -- **Retry Logic**: Automatic retry with configurable attempts -- **Structured Logging**: Comprehensive logging support -- **Error Handling**: Detailed error types and handling - -## Installation - -```bash -go get tm/pkg/glm -``` - -## Quick Start - -```go -package main - -import ( - "context" - "log" - "tm/pkg/glm" - "tm/pkg/logger" -) - -func main() { - // Create configuration - config := &glm.Config{ - APIKey: "your-api-key-here", - BaseURL: "https://api.z.ai", - } - - // Create logger (you can use your preferred logger) - logger := logger.NewLogger() - - // Create SDK instance - sdk, err := glm.New(config, logger) - if err != nil { - log.Fatal(err) - } - - // Simple chat - ctx := context.Background() - response, err := sdk.QuickChat(ctx, "Hello, how are you?") - if err != nil { - log.Fatal(err) - } - fmt.Println(response) - - // Translation example (matching the provided curl request) - translated, err := sdk.Translate(ctx, "IT-stöd för projekt- och portföljstyrning", "Swedish", "English") - if err != nil { - log.Fatal(err) - } - fmt.Println(translated) -} -``` - -## Configuration - -```go -config := &glm.Config{ - BaseURL: "https://api.z.ai", // API base URL - APIKey: "your-api-key", // Authentication key - Timeout: 300 * time.Second, // Request timeout - RetryAttempts: 3, // Number of retries - RetryDelay: 2 * time.Second, // Delay between retries - EnableLogging: true, // Enable request logging - UserAgent: "MyApp/1.0", // Custom user agent - DefaultModel: "glm-4.5", // Default model - DefaultMaxTokens: 4096, // Default max tokens - DefaultTopP: 1.0, // Default top_p - DefaultTemperature: 0.7, // Default temperature - EnableStreaming: false, // Enable streaming by default - EnableThinking: false, // Enable thinking mode -} -``` - -## API Methods - -### Chat Completion - -```go -req := &glm.ChatCompletionRequest{ - Model: "glm-4.5", - Messages: []glm.Message{ - { - Role: glm.MessageRoleSystem, - Content: "You are a helpful assistant.", - }, - { - Role: glm.MessageRoleUser, - Content: "Explain quantum computing.", - }, - }, - MaxTokens: 4096, - TopP: 1.0, - Temperature: 0.7, - FrequencyPenalty: 0, - PresencePenalty: 0, - Thinking: &glm.Thinking{ - Type: "false", - }, -} - -resp, err := sdk.ChatCompletion(ctx, req) -if err != nil { - log.Fatal(err) -} - -fmt.Println(resp.Choices[0].Message.Content) -``` - -### Streaming Chat Completion - -```go -req := &glm.ChatCompletionRequest{ - Model: "glm-4.5", - Messages: []glm.Message{ - { - Role: glm.MessageRoleUser, - Content: "Tell me a long story.", - }, - }, - Stream: true, -} - -err := sdk.StreamChatCompletion(ctx, req, func(chunk *glm.StreamChatCompletionResponse) error { - if len(chunk.Choices) > 0 { - fmt.Print(chunk.Choices[0].Delta.Content) - } - return nil -}) -``` - -### Convenience Methods - -```go -// Simple chat -response, err := sdk.QuickChat(ctx, "Hello!") - -// Chat with system prompt -response, err := sdk.QuickChatWithSystem(ctx, "You are a poet.", "Write a haiku about coding.") - -// Translation (matching the curl example) -translated, err := sdk.Translate(ctx, "IT-stöd för projekt- och portföljstyrning", "Swedish", "English") - -// Streaming chat -err := sdk.StreamQuickChat(ctx, "Tell me a story", func(chunk string) error { - fmt.Print(chunk) - return nil -}) -``` - -## Error Handling - -The SDK provides detailed error types: - -```go -resp, err := sdk.ChatCompletion(ctx, req) -if err != nil { - var glmErr *glm.Error - if errors.As(err, &glmErr) { - switch glmErr.Code { - case glm.ErrCodeAuthentication: - // Handle auth errors - case glm.ErrCodeRateLimit: - // Handle rate limits - case glm.ErrCodeServerError: - // Handle server errors - } - } -} -``` - -## Environment Variables - -You can configure the SDK using environment variables: - -```bash -export GLM_BASE_URL="https://api.z.ai" -export GLM_API_KEY="your-api-key" -export GLM_TIMEOUT="300s" -export GLM_RETRY_ATTEMPTS="3" -export GLM_DEFAULT_MODEL="glm-4.5" -export GLM_DEFAULT_MAX_TOKENS="4096" -``` - -## Testing - -```bash -go test ./pkg/glm/... -``` - -## License - -This SDK is part of the Tender Management system and follows the same licensing terms. diff --git a/pkg/glm/USAGE.md b/pkg/glm/USAGE.md deleted file mode 100644 index 8da739b..0000000 --- a/pkg/glm/USAGE.md +++ /dev/null @@ -1,280 +0,0 @@ -# GLM SDK Usage Guide - -## Matching the Provided Curl Request - -The SDK can replicate the exact curl request provided: - -```go -package main - -import ( - "context" - "fmt" - "log" - "tm/pkg/glm" - "tm/pkg/logger" -) - -func main() { - config := &glm.Config{ - BaseURL: "https://api.z.ai", - APIKey: "4e267ac7897f4a789cacb3dbee15f312.coJtK3397qFv16x5", - } - - logger := logger.NewLogger() - sdk, err := glm.New(config, logger) - if err != nil { - log.Fatal(err) - } - - req := &glm.ChatCompletionRequest{ - Model: "glm-4.5", - Messages: []glm.Message{ - { - Role: glm.MessageRoleSystem, - Content: "You are a high-precision translator specializing in professional tender documents. Always respond with only the translated English text — no explanations, notes, or markdown formatting.", - }, - { - Role: glm.MessageRoleUser, - Content: "Translate the following text from Swedish to English accurately and professionally:\n\"IT-stöd för projekt- och portföljstyrning\"", - }, - }, - Thinking: &glm.Thinking{ - Type: "false", - }, - MaxTokens: 4096, - TopP: 1, - FrequencyPenalty: 0, - PresencePenalty: 0, - } - - ctx := context.Background() - resp, err := sdk.ChatCompletion(ctx, req) - if err != nil { - log.Fatal(err) - } - - if len(resp.Choices) > 0 { - fmt.Println(resp.Choices[0].Message.Content) - } -} -``` - -## Using the Translate Convenience Method - -For the specific translation use case: - -```go -translated, err := sdk.Translate(ctx, "IT-stöd för projekt- och portföljstyrning", "Swedish", "English") -if err != nil { - log.Fatal(err) -} -fmt.Println(translated) // Output: "IT support for project and portfolio management" -``` - -### Custom Translation Options - -For more control over translation parameters: - -```go -opts := &glm.TranslateOptions{ - Model: "glm-4.5", - MaxTokens: 1024, - Temperature: 0.1, // Very low for consistent translations - FrequencyPenalty: 0.2, - PresencePenalty: 0.2, - EnableThinking: &[]bool{false}[0], -} - -translated, err := sdk.TranslateWithOptions(ctx, "IT-stöd för projekt- och portföljstyrning", "Swedish", "English", opts) -if err != nil { - log.Fatal(err) -} -fmt.Println(translated) -``` - -## Advanced Configuration - -### GLM SDK Configuration - -```go -config := &glm.Config{ - BaseURL: "https://api.z.ai", - APIKey: "your-api-key", - Timeout: 300 * time.Second, - RetryAttempts: 3, - RetryDelay: 2 * time.Second, - EnableLogging: true, - DefaultModel: "glm-4.5", - DefaultMaxTokens: 4096, - DefaultTopP: 1.0, - DefaultTemperature: 0.7, - DefaultFrequencyPenalty: 0.0, - DefaultPresencePenalty: 0.0, - EnableStreaming: false, - EnableThinking: false, - - // Translation-specific configuration - TranslationMaxTokens: 2048, - TranslationTemperature: 0.3, - TranslationFrequencyPenalty: 0.1, - TranslationPresencePenalty: 0.1, - TranslationEnableThinking: false, -} -``` - -### Bootstrap Configuration (Application Level) - -Configure translation parameters via environment variables in your application bootstrap: - -```go -// cmd/worker/bootstrap/config.go -type GLMConfig struct { - // ... basic GLM settings ... - - // Translation options configured via environment variables - TranslateOptions TranslateOptions -} - -type TranslateOptions struct { - MaxTokens int `env:"GLM_TRANSLATE_MAX_TOKENS" default:"2048"` - Temperature float64 `env:"GLM_TRANSLATE_TEMPERATURE" default:"0.3"` - FrequencyPenalty float64 `env:"GLM_TRANSLATE_FREQUENCY_PENALTY" default:"0.1"` - PresencePenalty float64 `env:"GLM_TRANSLATE_PRESENCE_PENALTY" default:"0.1"` - EnableThinking bool `env:"GLM_TRANSLATE_ENABLE_THINKING" default:"false"` -} -``` - -Set these environment variables to customize translation behavior: - -```bash -export GLM_TRANSLATE_MAX_TOKENS=1024 -export GLM_TRANSLATE_TEMPERATURE=0.2 -export GLM_TRANSLATE_FREQUENCY_PENALTY=0.2 -export GLM_TRANSLATE_PRESENCE_PENALTY=0.2 -export GLM_TRANSLATE_ENABLE_THINKING=false -``` - -## Error Handling Examples - -```go -resp, err := sdk.ChatCompletion(ctx, req) -if err != nil { - var glmErr *glm.Error - if errors.As(err, &glmErr) { - switch glmErr.Code { - case glm.ErrCodeAuthentication: - log.Printf("Authentication failed: %s", glmErr.Message) - // Handle authentication error (e.g., refresh token) - case glm.ErrCodeRateLimit: - log.Printf("Rate limit exceeded: %s", glmErr.Message) - // Implement backoff strategy - case glm.ErrCodeInvalidRequest: - log.Printf("Invalid request: %s", glmErr.Message) - // Fix request parameters - case glm.ErrCodeServerError: - log.Printf("Server error: %s", glmErr.Message) - // Retry or alert administrators - default: - log.Printf("Unknown error: %s", glmErr.Error()) - } - } else { - log.Printf("Non-GLM error: %v", err) - } - return -} -``` - -## Streaming with Progress Tracking - -```go -var fullResponse strings.Builder - -err := sdk.StreamChatCompletion(ctx, req, func(chunk *glm.StreamChatCompletionResponse) error { - if len(chunk.Choices) > 0 { - content := chunk.Choices[0].Delta.Content - fmt.Print(content) // Print as it comes - fullResponse.WriteString(content) - } - - // Check for completion - if len(chunk.Choices) > 0 && chunk.Choices[0].FinishReason != nil { - fmt.Printf("\n\nFinished with reason: %s\n", *chunk.Choices[0].FinishReason) - } - - return nil -}) - -if err != nil { - log.Fatal(err) -} - -fmt.Printf("Complete response: %s\n", fullResponse.String()) -``` - -## Integration with Tender Management System - -```go -// In your service layer -type TranslationService struct { - glmSDK *glm.SDK - logger logger.Logger -} - -func NewTranslationService(glmSDK *glm.SDK, logger logger.Logger) *TranslationService { - return &TranslationService{ - glmSDK: glmSDK, - logger: logger, - } -} - -func (s *TranslationService) TranslateTenderDocument(ctx context.Context, text, sourceLang, targetLang string) (string, error) { - s.logger.Info("Translating tender document", map[string]interface{}{ - "text_length": len(text), - "source_lang": sourceLang, - "target_lang": targetLang, - }) - - translated, err := s.glmSDK.Translate(ctx, text, sourceLang, targetLang) - if err != nil { - s.logger.Error("Translation failed", map[string]interface{}{ - "error": err.Error(), - }) - return "", err - } - - s.logger.Info("Translation completed", map[string]interface{}{ - "translated_length": len(translated), - }) - - return translated, nil -} -``` - -## Health Monitoring - -```go -// Check service health -err := sdk.Health(ctx) -if err != nil { - log.Printf("GLM service is unhealthy: %v", err) - // Alert monitoring system -} else { - log.Println("GLM service is healthy") -} -``` - -## Model Management - -```go -// List available models -models, err := sdk.ListModels(ctx) -if err != nil { - log.Fatal(err) -} - -fmt.Println("Available models:") -for _, model := range models.Data { - fmt.Printf("- %s (owned by %s)\n", model.ID, model.OwnedBy) -} -``` diff --git a/pkg/glm/client.go b/pkg/glm/client.go deleted file mode 100644 index d19a01f..0000000 --- a/pkg/glm/client.go +++ /dev/null @@ -1,226 +0,0 @@ -package glm - -import ( - "bytes" - "context" - "encoding/json" - "errors" - "fmt" - "io" - "net/http" - "strings" - "time" - "tm/pkg/logger" -) - -// Client implements the HTTPClient interface for GLM API -type Client struct { - config *Config - httpClient *http.Client - logger logger.Logger -} - -// NewClient creates a new GLM HTTP client -func NewClient(config *Config, logger logger.Logger) (*Client, error) { - if config == nil { - config = DefaultConfig() - } - - if err := config.Validate(); err != nil { - return nil, err - } - - httpClient := &http.Client{ - Timeout: config.Timeout, - } - - return &Client{ - config: config, - httpClient: httpClient, - logger: logger, - }, nil -} - -// Post performs a POST request to the GLM API -func (c *Client) Post(ctx context.Context, endpoint string, body interface{}, result interface{}) error { - return c.makeRequest(ctx, "POST", endpoint, body, result) -} - -// Get performs a GET request to the GLM API -func (c *Client) Get(ctx context.Context, endpoint string, result interface{}) error { - return c.makeRequest(ctx, "GET", endpoint, nil, result) -} - -// makeRequest performs an HTTP request with retry logic and logging -func (c *Client) makeRequest(ctx context.Context, method, endpoint string, body interface{}, result interface{}) error { - url := c.config.BaseURL + endpoint - - var bodyReader io.Reader - var bodyBytes []byte - if body != nil { - var err error - bodyBytes, err = json.Marshal(body) - if err != nil { - return fmt.Errorf("failed to marshal request body: %w", err) - } - bodyReader = bytes.NewReader(bodyBytes) - } - - var lastErr error - for attempt := 0; attempt <= c.config.RetryAttempts; attempt++ { - if attempt > 0 { - // Wait before retry - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(c.config.RetryDelay): - } - } - - err := c.doRequest(ctx, method, url, bodyReader, bodyBytes, result) - if err == nil { - return nil - } - - lastErr = err - - // Check if error is retryable - if !c.isRetryableError(err) { - break - } - - if c.logger != nil { - c.logger.Warn("Request failed, retrying", map[string]interface{}{ - "attempt": attempt + 1, - "max_attempts": c.config.RetryAttempts + 1, - "error": err.Error(), - "url": url, - }) - } - } - - return lastErr -} - -// doRequest performs the actual HTTP request -func (c *Client) doRequest(ctx context.Context, method, url string, bodyReader io.Reader, bodyBytes []byte, result interface{}) error { - // Reset body reader for retry - if bodyReader != nil { - bodyReader = bytes.NewReader(bodyBytes) - } - - req, err := http.NewRequestWithContext(ctx, method, url, bodyReader) - if err != nil { - return fmt.Errorf("failed to create request: %w", err) - } - - // Set headers - req.Header.Set("Content-Type", "application/json") - req.Header.Set("User-Agent", c.config.UserAgent) - if c.config.APIKey != "" { - req.Header.Set("Authorization", "Bearer "+c.config.APIKey) - } - - // Log request - if c.logger != nil && c.config.EnableLogging { - c.logger.Debug("Making GLM API request", map[string]interface{}{ - "method": method, - "url": url, - "headers": map[string]string{ - "Content-Type": req.Header.Get("Content-Type"), - "User-Agent": req.Header.Get("User-Agent"), - "Authorization": strings.Repeat("*", len(req.Header.Get("Authorization"))), - }, - }) - } - - resp, err := c.httpClient.Do(req) - if err != nil { - return fmt.Errorf("failed to execute request: %w", err) - } - defer resp.Body.Close() - - // Read response body - respBody, err := io.ReadAll(resp.Body) - if err != nil { - return fmt.Errorf("failed to read response body: %w", err) - } - - // Log response - if c.logger != nil && c.config.EnableLogging { - c.logger.Debug("GLM API response received", map[string]interface{}{ - "status_code": resp.StatusCode, - "headers": resp.Header, - "body_size": len(respBody), - }) - } - - // Check for HTTP errors - if resp.StatusCode >= 400 { - return c.handleHTTPError(resp.StatusCode, respBody) - } - - // Parse response - if result != nil { - if err := json.Unmarshal(respBody, result); err != nil { - return ErrInvalidResponse(fmt.Sprintf("failed to unmarshal response: %v", err)) - } - } - - return nil -} - -// handleHTTPError handles HTTP error responses -func (c *Client) handleHTTPError(statusCode int, body []byte) error { - var errorResp struct { - Error struct { - Message string `json:"message"` - Type string `json:"type"` - Code string `json:"code"` - } `json:"error"` - } - - if err := json.Unmarshal(body, &errorResp); err != nil { - // If we can't parse the error response, return a generic error - return ErrHTTPError(statusCode, string(body)) - } - - message := errorResp.Error.Message - if message == "" { - message = "Unknown error" - } - - switch statusCode { - case 400: - return ErrInvalidRequest(message) - case 401: - return ErrAuthentication(message) - case 403: - return ErrAuthentication(message) - case 404: - return ErrModelNotFound(message) - case 429: - return ErrRateLimit(message) - case 500, 502, 503, 504: - return ErrServerError(message) - default: - return ErrHTTPError(statusCode, message) - } -} - -// isRetryableError determines if an error should be retried -func (c *Client) isRetryableError(err error) bool { - var glmErr *Error - if errors.As(err, &glmErr) { - // Retry on server errors and rate limits - switch glmErr.Code { - case ErrCodeServerError, ErrCodeRateLimit: - return true - default: - return false - } - } - - // Retry on network errors, timeouts, etc. - return true -} diff --git a/pkg/glm/config.go b/pkg/glm/config.go deleted file mode 100644 index 242504a..0000000 --- a/pkg/glm/config.go +++ /dev/null @@ -1,138 +0,0 @@ -package glm - -import ( - "time" -) - -// Config represents the configuration for the GLM SDK -type Config struct { - // BaseURL is the base URL of the GLM API server - BaseURL string - - // APIKey is the authorization token for GLM API - APIKey string - - // Timeout is the timeout for HTTP requests - Timeout time.Duration - - // RetryAttempts is the number of retry attempts for failed requests - RetryAttempts int - - // RetryDelay is the delay between retry attempts - RetryDelay time.Duration - - // EnableLogging enables request/response logging - EnableLogging bool - - // UserAgent is the user agent string for HTTP requests - UserAgent string - - // DefaultModel is the default model to use if none is specified - DefaultModel string - - // DefaultMaxTokens is the default maximum number of tokens to generate - DefaultMaxTokens int - - // DefaultTopP is the default top_p value for text generation - DefaultTopP float64 - - // DefaultTemperature is the default temperature for text generation - DefaultTemperature float64 - - // DefaultFrequencyPenalty is the default frequency penalty - DefaultFrequencyPenalty float64 - - // DefaultPresencePenalty is the default presence penalty - DefaultPresencePenalty float64 - - // EnableStreaming enables streaming responses by default - EnableStreaming bool - - // EnableThinking enables thinking mode by default - EnableThinking bool - - // TranslationConfig is the translation configuration - TranslationOptions TranslationOption -} - -// TranslationOption represents the translation options -type TranslationOption struct { - Model string - MaxTokens int - Temperature float64 - FrequencyPenalty float64 - PresencePenalty float64 - TopP float64 - EnableThinking bool -} - -// DefaultConfig returns a default configuration for the GLM SDK -func DefaultConfig() *Config { - return &Config{ - BaseURL: "https://api.z.ai", - APIKey: "", - Timeout: 300 * time.Second, - RetryAttempts: 3, - RetryDelay: 2 * time.Second, - EnableLogging: true, - UserAgent: "Opplens-GLMSDK/1.0", - DefaultModel: "glm-4.5", - DefaultMaxTokens: 256, - DefaultTopP: 1.0, - DefaultTemperature: 0.5, - DefaultFrequencyPenalty: 0, - DefaultPresencePenalty: 0, - EnableStreaming: false, - EnableThinking: false, - TranslationOptions: TranslationOption{ - Model: "glm-4.5", - MaxTokens: 256, - Temperature: 0.5, - FrequencyPenalty: 0, - PresencePenalty: 0, - TopP: 1.0, - EnableThinking: false, - }, - } -} - -// Validate validates the configuration -func (c *Config) Validate() error { - if c.BaseURL == "" { - return ErrInvalidConfig("BaseURL cannot be empty") - } - - if c.Timeout <= 0 { - return ErrInvalidConfig("Timeout must be positive") - } - - if c.RetryAttempts < 0 { - return ErrInvalidConfig("RetryAttempts cannot be negative") - } - - if c.RetryDelay < 0 { - return ErrInvalidConfig("RetryDelay cannot be negative") - } - - if c.DefaultMaxTokens < 1 || c.DefaultMaxTokens > 32768 { - return ErrInvalidConfig("DefaultMaxTokens must be between 1 and 32768") - } - - if c.DefaultTopP < 0 || c.DefaultTopP > 1 { - return ErrInvalidConfig("DefaultTopP must be between 0 and 1") - } - - if c.DefaultTemperature < 0 || c.DefaultTemperature > 2 { - return ErrInvalidConfig("DefaultTemperature must be between 0 and 2") - } - - if c.DefaultFrequencyPenalty < -2 || c.DefaultFrequencyPenalty > 2 { - return ErrInvalidConfig("DefaultFrequencyPenalty must be between -2 and 2") - } - - if c.DefaultPresencePenalty < -2 || c.DefaultPresencePenalty > 2 { - return ErrInvalidConfig("DefaultPresencePenalty must be between -2 and 2") - } - - return nil -} diff --git a/pkg/glm/entities.go b/pkg/glm/entities.go deleted file mode 100644 index 0feb732..0000000 --- a/pkg/glm/entities.go +++ /dev/null @@ -1,92 +0,0 @@ -package glm - -// ChatCompletionRequest represents a request for chat completion -type ChatCompletionRequest struct { - Model string `json:"model" validate:"required"` - Messages []Message `json:"messages" validate:"required,min=1"` - Thinking *Thinking `json:"thinking,omitempty"` - MaxTokens int `json:"max_tokens,omitempty" validate:"min=1,max=32768"` - TopP float64 `json:"top_p,omitempty" validate:"min=0,max=1"` - FrequencyPenalty float64 `json:"frequency_penalty,omitempty" validate:"min=-2,max=2"` - PresencePenalty float64 `json:"presence_penalty,omitempty" validate:"min=-2,max=2"` - Temperature float64 `json:"temperature,omitempty" validate:"min=0,max=2"` - Stop []string `json:"stop,omitempty"` - Stream bool `json:"stream,omitempty"` -} - -// Message represents a message in a chat conversation -type Message struct { - Role MessageRole `json:"role" validate:"required,oneof=system user assistant"` - Content string `json:"content" validate:"required"` -} - -// MessageRole defines the role of a message -type MessageRole string - -const ( - MessageRoleSystem MessageRole = "system" - MessageRoleUser MessageRole = "user" - MessageRoleAssistant MessageRole = "assistant" -) - -// Thinking represents the thinking configuration for GLM models -type Thinking struct { - Type string `json:"type" validate:"required,oneof=true false"` -} - -// ChatCompletionResponse represents the response from a chat completion request -type ChatCompletionResponse struct { - ID string `json:"id"` - Object string `json:"object"` - Created int64 `json:"created"` - Model string `json:"model"` - Choices []Choice `json:"choices"` - Usage Usage `json:"usage"` - SystemFingerprint string `json:"system_fingerprint,omitempty"` -} - -// Choice represents a single choice in the chat completion response -type Choice struct { - Index int `json:"index"` - Message Message `json:"message"` - FinishReason string `json:"finish_reason"` -} - -// Usage represents token usage information -type Usage struct { - PromptTokens int `json:"prompt_tokens"` - CompletionTokens int `json:"completion_tokens"` - TotalTokens int `json:"total_tokens"` -} - -// StreamChatCompletionResponse represents a streaming response chunk -type StreamChatCompletionResponse struct { - ID string `json:"id"` - Object string `json:"object"` - Created int64 `json:"created"` - Model string `json:"model"` - Choices []StreamChoice `json:"choices"` - Usage *Usage `json:"usage,omitempty"` - SystemFingerprint string `json:"system_fingerprint,omitempty"` -} - -// StreamChoice represents a single choice in a streaming response -type StreamChoice struct { - Index int `json:"index"` - Delta Message `json:"delta"` - FinishReason *string `json:"finish_reason,omitempty"` -} - -// Model represents information about available models -type Model struct { - ID string `json:"id"` - Object string `json:"object"` - Created int64 `json:"created"` - OwnedBy string `json:"owned_by"` -} - -// ModelsResponse represents the response from listing available models -type ModelsResponse struct { - Object string `json:"object"` - Data []Model `json:"data"` -} diff --git a/pkg/glm/errors.go b/pkg/glm/errors.go deleted file mode 100644 index 7e3183e..0000000 --- a/pkg/glm/errors.go +++ /dev/null @@ -1,94 +0,0 @@ -package glm - -import ( - "fmt" -) - -// Error represents a GLM SDK error -type Error struct { - Code string - Message string - Cause error -} - -// Error implements the error interface -func (e *Error) Error() string { - if e.Cause != nil { - return fmt.Sprintf("glm: %s: %s: %v", e.Code, e.Message, e.Cause) - } - return fmt.Sprintf("glm: %s: %s", e.Code, e.Message) -} - -// Unwrap returns the underlying cause of the error -func (e *Error) Unwrap() error { - return e.Cause -} - -// NewError creates a new GLM error -func NewError(code, message string, cause error) *Error { - return &Error{ - Code: code, - Message: message, - Cause: cause, - } -} - -// Error codes -const ( - ErrCodeInvalidConfig = "INVALID_CONFIG" - ErrCodeHTTPError = "HTTP_ERROR" - ErrCodeInvalidRequest = "INVALID_REQUEST" - ErrCodeAuthentication = "AUTHENTICATION_ERROR" - ErrCodeRateLimit = "RATE_LIMIT_ERROR" - ErrCodeServerError = "SERVER_ERROR" - ErrCodeInvalidResponse = "INVALID_RESPONSE" - ErrCodeChatCompletion = "CHAT_COMPLETION_ERROR" - ErrCodeModelNotFound = "MODEL_NOT_FOUND" - ErrCodeTokenLimit = "TOKEN_LIMIT_ERROR" - ErrCodeStreamError = "STREAM_ERROR" -) - -// Error constructors -func ErrInvalidConfig(message string) *Error { - return NewError(ErrCodeInvalidConfig, message, nil) -} - -func ErrHTTPError(statusCode int, message string) *Error { - return NewError(ErrCodeHTTPError, fmt.Sprintf("HTTP %d: %s", statusCode, message), nil) -} - -func ErrInvalidRequest(message string) *Error { - return NewError(ErrCodeInvalidRequest, message, nil) -} - -func ErrAuthentication(message string) *Error { - return NewError(ErrCodeAuthentication, message, nil) -} - -func ErrRateLimit(message string) *Error { - return NewError(ErrCodeRateLimit, message, nil) -} - -func ErrServerError(message string) *Error { - return NewError(ErrCodeServerError, message, nil) -} - -func ErrInvalidResponse(message string) *Error { - return NewError(ErrCodeInvalidResponse, message, nil) -} - -func ErrChatCompletion(message string, cause error) *Error { - return NewError(ErrCodeChatCompletion, message, cause) -} - -func ErrModelNotFound(model string) *Error { - return NewError(ErrCodeModelNotFound, fmt.Sprintf("model '%s' not found", model), nil) -} - -func ErrTokenLimit(message string) *Error { - return NewError(ErrCodeTokenLimit, message, nil) -} - -func ErrStreamError(message string, cause error) *Error { - return NewError(ErrCodeStreamError, message, cause) -} diff --git a/pkg/glm/interfaces.go b/pkg/glm/interfaces.go deleted file mode 100644 index 8cc7590..0000000 --- a/pkg/glm/interfaces.go +++ /dev/null @@ -1,29 +0,0 @@ -package glm - -import ( - "context" -) - -// GLMService defines the interface for GLM service operations -type GLMService interface { - // ChatCompletion performs a chat completion request - ChatCompletion(ctx context.Context, req *ChatCompletionRequest) (*ChatCompletionResponse, error) - - // StreamChatCompletion performs a streaming chat completion request - StreamChatCompletion(ctx context.Context, req *ChatCompletionRequest, onChunk func(*StreamChatCompletionResponse) error) error - - // ListModels lists available models - ListModels(ctx context.Context) (*ModelsResponse, error) - - // Health checks if the GLM service is available - Health(ctx context.Context) error -} - -// HTTPClient defines the interface for HTTP operations -type HTTPClient interface { - // Post performs a POST request - Post(ctx context.Context, url string, body interface{}, result interface{}) error - - // Get performs a GET request - Get(ctx context.Context, url string, result interface{}) error -} diff --git a/pkg/glm/sdk.go b/pkg/glm/sdk.go deleted file mode 100644 index 6f98aa4..0000000 --- a/pkg/glm/sdk.go +++ /dev/null @@ -1,277 +0,0 @@ -package glm - -import ( - "context" - "fmt" - "tm/pkg/logger" -) - -// SDK provides a high-level interface for the GLM service -type SDK struct { - service GLMService - client *Client - config *Config -} - -// New creates a new GLM SDK with the provided configuration and logger -func New(config *Config, logger logger.Logger) (*SDK, error) { - cfg := DefaultConfig() - if config.BaseURL != "" { - cfg.BaseURL = config.BaseURL - } - if config.APIKey != "" { - cfg.APIKey = config.APIKey - } - if config.Timeout != 0 { - cfg.Timeout = config.Timeout - } - if config.RetryAttempts != 0 { - cfg.RetryAttempts = config.RetryAttempts - } - if config.RetryDelay != 0 { - cfg.RetryDelay = config.RetryDelay - } - if config.EnableLogging { - cfg.EnableLogging = config.EnableLogging - } - if config.UserAgent != "" { - cfg.UserAgent = config.UserAgent - } - if config.DefaultModel != "" { - cfg.DefaultModel = config.DefaultModel - } - if config.DefaultMaxTokens != 0 { - cfg.DefaultMaxTokens = config.DefaultMaxTokens - } - if config.DefaultTopP != 0 { - cfg.DefaultTopP = config.DefaultTopP - } - if config.DefaultTemperature != 0 { - cfg.DefaultTemperature = config.DefaultTemperature - } - if config.DefaultFrequencyPenalty != 0 { - cfg.DefaultFrequencyPenalty = config.DefaultFrequencyPenalty - } - if config.DefaultPresencePenalty != 0 { - cfg.DefaultPresencePenalty = config.DefaultPresencePenalty - } - if config.EnableStreaming { - cfg.EnableStreaming = config.EnableStreaming - } - if config.EnableThinking { - cfg.EnableThinking = config.EnableThinking - } - if config.TranslationOptions.Model != "" { - cfg.TranslationOptions.Model = config.TranslationOptions.Model - } - if config.TranslationOptions.MaxTokens != 0 { - cfg.TranslationOptions.MaxTokens = config.TranslationOptions.MaxTokens - } - if config.TranslationOptions.Temperature != 0 { - cfg.TranslationOptions.Temperature = config.TranslationOptions.Temperature - } - if config.TranslationOptions.FrequencyPenalty != 0 { - cfg.TranslationOptions.FrequencyPenalty = config.TranslationOptions.FrequencyPenalty - } - if config.TranslationOptions.PresencePenalty != 0 { - cfg.TranslationOptions.PresencePenalty = config.TranslationOptions.PresencePenalty - } - if config.TranslationOptions.TopP != 0 { - cfg.TranslationOptions.TopP = config.TranslationOptions.TopP - } - if config.TranslationOptions.EnableThinking { - cfg.TranslationOptions.EnableThinking = config.TranslationOptions.EnableThinking - } - - client, err := NewClient(cfg, logger) - if err != nil { - return nil, err - } - - service := NewService(client) - - return &SDK{ - service: service, - client: client, - config: cfg, - }, nil -} - -// ChatCompletion performs a chat completion request -func (s *SDK) ChatCompletion(ctx context.Context, req *ChatCompletionRequest) (*ChatCompletionResponse, error) { - return s.service.ChatCompletion(ctx, req) -} - -// StreamChatCompletion performs a streaming chat completion request -func (s *SDK) StreamChatCompletion(ctx context.Context, req *ChatCompletionRequest, onChunk func(*StreamChatCompletionResponse) error) error { - return s.service.StreamChatCompletion(ctx, req, onChunk) -} - -// ListModels lists available models -func (s *SDK) ListModels(ctx context.Context) (*ModelsResponse, error) { - return s.service.ListModels(ctx) -} - -// Health checks if the GLM service is available -func (s *SDK) Health(ctx context.Context) error { - return s.service.Health(ctx) -} - -// QuickChat is a convenience method for simple chat interactions -func (s *SDK) QuickChat(ctx context.Context, userMessage string) (string, error) { - req := &ChatCompletionRequest{ - Model: s.config.DefaultModel, - Messages: []Message{ - { - Role: MessageRoleUser, - Content: userMessage, - }, - }, - MaxTokens: s.config.DefaultMaxTokens, - TopP: s.config.DefaultTopP, - Temperature: s.config.DefaultTemperature, - FrequencyPenalty: s.config.DefaultFrequencyPenalty, - PresencePenalty: s.config.DefaultPresencePenalty, - } - - if s.config.EnableThinking { - req.Thinking = &Thinking{Type: "true"} - } - - resp, err := s.ChatCompletion(ctx, req) - if err != nil { - return "", err - } - - if len(resp.Choices) == 0 { - return "", ErrInvalidResponse("no choices returned") - } - - return resp.Choices[0].Message.Content, nil -} - -// QuickChatWithSystem is a convenience method for chat with system prompt -func (s *SDK) QuickChatWithSystem(ctx context.Context, systemPrompt, userMessage string) (string, error) { - req := &ChatCompletionRequest{ - Model: s.config.DefaultModel, - Messages: []Message{ - { - Role: MessageRoleSystem, - Content: systemPrompt, - }, - { - Role: MessageRoleUser, - Content: userMessage, - }, - }, - MaxTokens: s.config.DefaultMaxTokens, - TopP: s.config.DefaultTopP, - Temperature: s.config.DefaultTemperature, - FrequencyPenalty: s.config.DefaultFrequencyPenalty, - PresencePenalty: s.config.DefaultPresencePenalty, - } - - if s.config.EnableThinking { - req.Thinking = &Thinking{Type: "true"} - } - - resp, err := s.ChatCompletion(ctx, req) - if err != nil { - return "", err - } - - if len(resp.Choices) == 0 { - return "", ErrInvalidResponse("no choices returned") - } - - return resp.Choices[0].Message.Content, nil -} - -// TranslateOptions represents options for translation -type TranslateOptions struct { - Model string `json:"model,omitempty"` - MaxTokens int `json:"max_tokens,omitempty"` - Temperature float64 `json:"temperature,omitempty"` - FrequencyPenalty float64 `json:"frequency_penalty,omitempty"` - PresencePenalty float64 `json:"presence_penalty,omitempty"` - TopP float64 `json:"top_p,omitempty"` - EnableThinking *bool `json:"enable_thinking,omitempty"` -} - -// Translate is a convenience method for translation tasks using dynamic configuration parameters optimized for translation -func (s *SDK) Translate(ctx context.Context, text, fromLang, toLang string) (string, error) { - systemPrompt := "You are a high-precision translator specializing in professional tender documents. Always respond with only the translated text — no explanations, notes, or markdown formatting." - - userMessage := fmt.Sprintf("Translate the following text from %s to %s accurately and professionally:\n\"%s\"", fromLang, toLang, text) - - // Build request with custom options or defaults - req := &ChatCompletionRequest{ - Model: s.config.TranslationOptions.Model, - Messages: []Message{ - { - Role: MessageRoleSystem, - Content: systemPrompt, - }, - { - Role: MessageRoleUser, - Content: userMessage, - }, - }, - MaxTokens: s.config.TranslationOptions.MaxTokens, - TopP: s.config.TranslationOptions.TopP, - Temperature: s.config.TranslationOptions.Temperature, - FrequencyPenalty: s.config.TranslationOptions.FrequencyPenalty, - PresencePenalty: s.config.TranslationOptions.PresencePenalty, - } - - resp, err := s.ChatCompletion(ctx, req) - if err != nil { - return "", err - } - - if len(resp.Choices) == 0 { - return "", ErrInvalidResponse("no choices returned") - } - - return resp.Choices[0].Message.Content, nil -} - -// StreamQuickChat is a convenience method for streaming chat interactions -func (s *SDK) StreamQuickChat(ctx context.Context, userMessage string, onChunk func(string) error) error { - req := &ChatCompletionRequest{ - Model: s.config.DefaultModel, - Messages: []Message{ - { - Role: MessageRoleUser, - Content: userMessage, - }, - }, - MaxTokens: s.config.DefaultMaxTokens, - TopP: s.config.DefaultTopP, - Temperature: s.config.DefaultTemperature, - FrequencyPenalty: s.config.DefaultFrequencyPenalty, - PresencePenalty: s.config.DefaultPresencePenalty, - Stream: true, - } - - if s.config.EnableThinking { - req.Thinking = &Thinking{Type: "true"} - } - - return s.StreamChatCompletion(ctx, req, func(chunk *StreamChatCompletionResponse) error { - if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" { - return onChunk(chunk.Choices[0].Delta.Content) - } - return nil - }) -} - -// GetService returns the underlying service for advanced usage -func (s *SDK) GetService() GLMService { - return s.service -} - -// GetConfig returns the current configuration -func (s *SDK) GetConfig() *Config { - return s.config -} diff --git a/pkg/glm/service.go b/pkg/glm/service.go deleted file mode 100644 index 31a5f39..0000000 --- a/pkg/glm/service.go +++ /dev/null @@ -1,218 +0,0 @@ -package glm - -import ( - "bufio" - "bytes" - "context" - "encoding/json" - "fmt" - "io" - "net/http" - "strings" - "tm/pkg/logger" -) - -// Service implements the GLMService interface -type Service struct { - client *Client - config *Config - logger logger.Logger -} - -// NewService creates a new GLM service -func NewService(client *Client) GLMService { - return &Service{ - client: client, - config: client.config, - logger: client.logger, - } -} - -// ChatCompletion performs a chat completion request -func (s *Service) ChatCompletion(ctx context.Context, req *ChatCompletionRequest) (*ChatCompletionResponse, error) { - if s.logger != nil { - s.logger.Info("Performing chat completion", map[string]interface{}{ - "model": req.Model, - "message_count": len(req.Messages), - "max_tokens": req.MaxTokens, - "stream": req.Stream, - }) - } - - var resp ChatCompletionResponse - err := s.client.Post(ctx, "/api/coding/paas/v4/chat/completions", req, &resp) - if err != nil { - if s.logger != nil { - s.logger.Error("Chat completion failed", map[string]interface{}{ - "model": req.Model, - "error": err.Error(), - }) - } - return nil, ErrChatCompletion("chat completion request failed", err) - } - - if s.logger != nil { - s.logger.Info("Chat completion completed", map[string]interface{}{ - "model": req.Model, - "choice_count": len(resp.Choices), - "usage": resp.Usage, - }) - } - - return &resp, nil -} - -// StreamChatCompletion performs a streaming chat completion request -func (s *Service) StreamChatCompletion(ctx context.Context, req *ChatCompletionRequest, onChunk func(*StreamChatCompletionResponse) error) error { - if s.logger != nil { - s.logger.Info("Starting streaming chat completion", map[string]interface{}{ - "model": req.Model, - "message_count": len(req.Messages), - "max_tokens": req.MaxTokens, - }) - } - - // Force streaming - req.Stream = true - - return s.streamRequest(ctx, req, onChunk) -} - -// streamRequest handles streaming requests -func (s *Service) streamRequest(ctx context.Context, req *ChatCompletionRequest, onChunk func(*StreamChatCompletionResponse) error) error { - // Create a custom HTTP request for streaming - httpReq, err := s.client.createStreamingRequest(ctx, req) - if err != nil { - return ErrStreamError("failed to create streaming request", err) - } - - resp, err := s.client.httpClient.Do(httpReq) - if err != nil { - return ErrStreamError("failed to execute streaming request", err) - } - defer resp.Body.Close() - - if resp.StatusCode != 200 { - body, _ := io.ReadAll(resp.Body) - return s.client.handleHTTPError(resp.StatusCode, body) - } - - scanner := bufio.NewScanner(resp.Body) - for scanner.Scan() { - line := scanner.Text() - - // Skip empty lines - if strings.TrimSpace(line) == "" { - continue - } - - // Parse SSE format - if strings.HasPrefix(line, "data: ") { - data := strings.TrimPrefix(line, "data: ") - - // Check for end of stream - if data == "[DONE]" { - break - } - - var chunk StreamChatCompletionResponse - if err := json.Unmarshal([]byte(data), &chunk); err != nil { - if s.logger != nil { - s.logger.Warn("Failed to parse streaming chunk", map[string]interface{}{ - "error": err.Error(), - "data": data, - }) - } - continue - } - - if err := onChunk(&chunk); err != nil { - return ErrStreamError("chunk callback failed", err) - } - } - } - - if err := scanner.Err(); err != nil { - return ErrStreamError("streaming scanner error", err) - } - - if s.logger != nil { - s.logger.Info("Streaming chat completion finished", map[string]interface{}{}) - } - - return nil -} - -// ListModels lists available models -func (s *Service) ListModels(ctx context.Context) (*ModelsResponse, error) { - if s.logger != nil { - s.logger.Info("Listing available models", map[string]interface{}{}) - } - - var resp ModelsResponse - err := s.client.Get(ctx, "/api/coding/paas/v4/models", &resp) - if err != nil { - if s.logger != nil { - s.logger.Error("Failed to list models", map[string]interface{}{ - "error": err.Error(), - }) - } - return nil, fmt.Errorf("failed to list models: %w", err) - } - - if s.logger != nil { - s.logger.Info("Models listed successfully", map[string]interface{}{ - "model_count": len(resp.Data), - }) - } - - return &resp, nil -} - -// Health checks if the GLM service is available -func (s *Service) Health(ctx context.Context) error { - if s.logger != nil { - s.logger.Debug("Performing health check", map[string]interface{}{}) - } - - // Simple health check by listing models - _, err := s.ListModels(ctx) - if err != nil { - if s.logger != nil { - s.logger.Error("Health check failed", map[string]interface{}{ - "error": err.Error(), - }) - } - return fmt.Errorf("health check failed: %w", err) - } - - if s.logger != nil { - s.logger.Debug("Health check passed", map[string]interface{}{}) - } - - return nil -} - -// createStreamingRequest creates an HTTP request for streaming -func (c *Client) createStreamingRequest(ctx context.Context, req *ChatCompletionRequest) (*http.Request, error) { - url := c.config.BaseURL + "/api/coding/paas/v4/chat/completions" - - bodyBytes, err := json.Marshal(req) - if err != nil { - return nil, fmt.Errorf("failed to marshal request body: %w", err) - } - - httpReq, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(bodyBytes)) - if err != nil { - return nil, fmt.Errorf("failed to create request: %w", err) - } - - // Set headers - httpReq.Header.Set("Content-Type", "application/json") - httpReq.Header.Set("User-Agent", c.config.UserAgent) - if c.config.APIKey != "" { - httpReq.Header.Set("Authorization", "Bearer "+c.config.APIKey) - } - - return httpReq, nil -}