From e136f0eaa7d1c811c201c580dad91208fd4e06c1 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Sun, 10 May 2026 02:14:53 +0330 Subject: [PATCH] List and download tender documents - worker auto translation to english job --- cmd/web/router/routes.go | 4 + cmd/worker/bootstrap/bootstrap.go | 60 +++++++++- cmd/worker/bootstrap/config.go | 12 ++ cmd/worker/main.go | 5 +- cmd/worker/workers/translation.go | 149 +++++++++++++++++++++++ internal/tender/form.go | 14 +++ internal/tender/handler.go | 69 +++++++++++ internal/tender/repository.go | 47 ++++++++ internal/tender/service.go | 193 ++++++++++++++++++++++++++++++ pkg/ai_summarizer/entities.go | 9 ++ pkg/ai_summarizer/storage.go | 92 ++++++++++++++ 11 files changed, 652 insertions(+), 2 deletions(-) create mode 100644 cmd/worker/workers/translation.go diff --git a/cmd/web/router/routes.go b/cmd/web/router/routes.go index f7f9de7..d51bc29 100644 --- a/cmd/web/router/routes.go +++ b/cmd/web/router/routes.go @@ -109,6 +109,8 @@ func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler tendersGP.GET("/:id", tenderHandler.Get) tendersGP.PUT("/:id", tenderHandler.Update) tendersGP.DELETE("/:id", tenderHandler.Delete) + tendersGP.GET("/:id/documents", tenderHandler.GetDocuments) + tendersGP.GET("/:id/documents/download", tenderHandler.DownloadDocuments) tendersGP.GET("/:id/ai-summary", tenderHandler.GetAISummary) tendersGP.POST("/:id/ai-summarize", tenderHandler.TriggerAISummarize) tendersGP.POST("/:id/ai-translate", tenderHandler.TriggerAITranslate) @@ -236,6 +238,8 @@ func RegisterPublicRoutes(e *echo.Echo, customerHandler *customer.Handler, tende tendersGP.GET("", tenderHandler.GetPublicTenders) tendersGP.GET("/recommend", tenderHandler.RecommendTenders) tendersGP.GET("/details/:id", tenderHandler.GetPublicTenderDetails) + tendersGP.GET("/:id/documents", tenderHandler.GetDocuments) + tendersGP.GET("/:id/documents/download", tenderHandler.DownloadDocuments) tendersGP.GET("/:id/document-summaries", tenderHandler.GetDocumentSummaries) tendersGP.GET("/:id/ai-summary", tenderHandler.GetPublicAISummary) tendersGP.POST("/:id/ai-translate", tenderHandler.TriggerPublicAITranslate) diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index f098b54..302e48e 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -6,6 +6,7 @@ import ( "tm/cmd/worker/workers" "tm/internal/notice" "tm/internal/tender" + ai_summarizer "tm/pkg/ai_summarizer" "tm/pkg/config" "tm/pkg/glm" "tm/pkg/logger" @@ -77,7 +78,7 @@ func InitMongoDB(conf config.MongoConfig, log logger.Logger) *mongo.ConnectionMa } // Init Worker -func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, glmService *glm.SDK, minioService *minio.Service) *schedule.CronScheduler { +func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, glmService *glm.SDK, minioService *minio.Service, aiClient *ai_summarizer.Client) *schedule.CronScheduler { // Debug: Log worker config appLogger.Info("Worker configuration", map[string]interface{}{ "worker_interval": config.Worker.Interval, @@ -186,6 +187,33 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger Expr: workerInterval, }) + // Initialize translation backfill worker for default language + if aiClient != nil { + targetLanguage := config.AISummarizer.DefaultLanguage + scheduler.AddJob(schedule.Job{ + Name: "Tender Translation Worker Job", + Func: func() { + worker := workers.NewTranslationWorker( + mongoManager, + appLogger, + tenderRepo, + aiClient, + targetLanguage, + config.Worker.TranslationBatchSize, + ) + worker.Run() + }, + Expr: config.Worker.TranslationInterval, + }) + appLogger.Info("Scheduled tender translation worker", map[string]interface{}{ + "target_language": targetLanguage, + "interval": config.Worker.TranslationInterval, + "batch_size": config.Worker.TranslationBatchSize, + }) + } else { + appLogger.Warn("AI summarizer client not available, tender translation worker is disabled", map[string]interface{}{}) + } + // Start the cron scheduler so all recurring jobs actually run scheduler.Start() appLogger.Info("Cron scheduler started with all worker jobs", map[string]interface{}{}) @@ -193,6 +221,36 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger return scheduler } +// InitAISummarizerClient initializes the AI summarizer HTTP client used by worker jobs. +func InitAISummarizerClient(conf AISummarizerConfig, log logger.Logger) *ai_summarizer.Client { + if conf.APIBaseURL == "" { + log.Warn("AI Summarizer API base URL not configured, translation worker will be unavailable", map[string]interface{}{}) + return nil + } + + cfg := &ai_summarizer.Config{ + APIBaseURL: conf.APIBaseURL, + APITimeout: conf.APITimeout, + APIRetryCount: conf.APIRetryCount, + APIRetryDelay: conf.APIRetryDelay, + } + + client, err := ai_summarizer.NewClient(cfg, log) + if err != nil { + log.Error("Failed to initialize AI Summarizer client for worker", map[string]interface{}{ + "error": err.Error(), + }) + return nil + } + + log.Info("AI Summarizer client initialized for worker", map[string]interface{}{ + "base_url": conf.APIBaseURL, + "timeout": conf.APITimeout, + }) + + return client +} + // Init Notification Service func InitNotificationService(conf config.NotificationConfig, log logger.Logger) notification.SDK { notificationConfig := ¬ification.Config{ diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 3e8c27a..3fc8abc 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -15,6 +15,7 @@ type Config struct { GLM GLMConfig Scraper config.ScraperConfig MinIO config.MinIOConfig + AISummarizer AISummarizerConfig Queue QueueConfig `env:"QUEUE_" envPrefix:"QUEUE_"` AlertMail string `env:"ALERT_MAIL" envDefault:"alerts@opplens.com"` } @@ -37,6 +38,17 @@ type QueueConfig struct { type WorkerConfig struct { Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"` NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"0"` + TranslationInterval string `env:"WORKER_TRANSLATION_INTERVAL" envDefault:"0 */30 * * * *"` + TranslationBatchSize int `env:"WORKER_TRANSLATION_BATCH_SIZE" envDefault:"20"` +} + +// AISummarizerConfig holds configuration for the external AI summarizer service. +type AISummarizerConfig struct { + APIBaseURL string `env:"AI_SUMMARIZER_API_BASE_URL" envDefault:""` + APITimeout time.Duration `env:"AI_SUMMARIZER_API_TIMEOUT" envDefault:"120s"` + APIRetryCount int `env:"AI_SUMMARIZER_API_RETRY_COUNT" envDefault:"2"` + APIRetryDelay time.Duration `env:"AI_SUMMARIZER_API_RETRY_DELAY" envDefault:"3s"` + DefaultLanguage string `env:"AI_SUMMARIZER_DEFAULT_LANGUAGE" envDefault:"en"` } type GLMConfig struct { diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 84825d4..e0a1682 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -54,8 +54,11 @@ func main() { appLogger.Info("MinIO service initialized successfully", map[string]interface{}{}) } + // Initialize AI summarizer client (used by translation worker) + aiSummarizerClient := bootstrap.InitAISummarizerClient(config.AISummarizer, appLogger) + // Initialize Worker - scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, glmService, minioService) + scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, glmService, minioService, aiSummarizerClient) // Set up signal handling for graceful shutdown signalChan := make(chan os.Signal, 1) diff --git a/cmd/worker/workers/translation.go b/cmd/worker/workers/translation.go new file mode 100644 index 0000000..69fb155 --- /dev/null +++ b/cmd/worker/workers/translation.go @@ -0,0 +1,149 @@ +package workers + +import ( + "context" + "strings" + "time" + "tm/internal/tender" + "tm/pkg/ai_summarizer" + "tm/pkg/logger" + "tm/pkg/mongo" +) + +// TranslationWorker backfills missing tender translations using AI summarizer API. +type TranslationWorker struct { + Mongo *mongo.ConnectionManager + Logger logger.Logger + TenderRepo tender.TenderRepository + AIClient *ai_summarizer.Client + TargetLanguage string + BatchSize int +} + +// NewTranslationWorker creates a translation worker for a target language. +func NewTranslationWorker(mongo *mongo.ConnectionManager, logger logger.Logger, tenderRepo tender.TenderRepository, aiClient *ai_summarizer.Client, targetLanguage string, batchSize int) *TranslationWorker { + if strings.TrimSpace(targetLanguage) == "" { + targetLanguage = "en" + } + if batchSize <= 0 { + batchSize = 20 + } + + return &TranslationWorker{ + Mongo: mongo, + Logger: logger, + TenderRepo: tenderRepo, + AIClient: aiClient, + TargetLanguage: strings.ToLower(strings.TrimSpace(targetLanguage)), + BatchSize: batchSize, + } +} + +// Run starts backfilling translations for tenders missing target-language content. +func (w *TranslationWorker) Run() { + if w.AIClient == nil { + w.Logger.Warn("Translation worker skipped: AI client is not configured", map[string]interface{}{ + "target_language": w.TargetLanguage, + }) + return + } + + w.Logger.Info("Translation worker started", map[string]interface{}{ + "target_language": w.TargetLanguage, + "batch_size": w.BatchSize, + }) + + for { + // Always use skip=0 because successfully translated tenders are excluded by filter. + tenders, totalCount, err := w.TenderRepo.GetUnTranslatedTenders(context.Background(), w.TargetLanguage, w.BatchSize, 0) + if err != nil { + w.Logger.Error("Failed to fetch untranslated tenders", map[string]interface{}{ + "target_language": w.TargetLanguage, + "error": err.Error(), + }) + break + } + + w.Logger.Info("Fetched untranslated tenders batch", map[string]interface{}{ + "target_language": w.TargetLanguage, + "batch_count": len(tenders), + "total_count": totalCount, + }) + + if len(tenders) == 0 { + break + } + + for _, t := range tenders { + w.translateTender(&t) + } + } + + w.Logger.Info("Translation worker completed", map[string]interface{}{ + "target_language": w.TargetLanguage, + }) +} + +func (w *TranslationWorker) translateTender(t *tender.Tender) { + if t.NoticePublicationID == "" { + return + } + if existing, ok := t.Translations[w.TargetLanguage]; ok && strings.TrimSpace(existing.Title) != "" { + w.Logger.Debug("Skipping tender translation because target language already exists", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "notice_id": t.NoticePublicationID, + "target_language": w.TargetLanguage, + }) + return + } + + resp, err := w.AIClient.FetchTranslationOnDemand(context.Background(), ai_summarizer.TranslateRequest{ + NoticeID: t.NoticePublicationID, + Title: t.Title, + Description: t.Description, + Language: w.TargetLanguage, + }) + if err != nil { + w.Logger.Error("Failed to translate tender", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "notice_id": t.NoticePublicationID, + "target_language": w.TargetLanguage, + "translation_error": err.Error(), + }) + return + } + + if t.Translations == nil { + t.Translations = make(map[string]tender.TranslationEntry) + } + now := time.Now().Unix() + // Only update the target-language slot. Existing translations in other + // languages (e.g., "de", "fr") are preserved and never modified. + t.Translations[w.TargetLanguage] = tender.TranslationEntry{ + Title: resp.TranslatedTitle, + Description: resp.TranslatedDescription, + Language: w.TargetLanguage, + TranslatedAt: now, + } + if t.ProcessingMetadata.TranslatedData == nil { + t.ProcessingMetadata.TranslatedData = make(map[string]string) + } + t.ProcessingMetadata.TranslatedData[w.TargetLanguage] = "done" + t.ProcessingMetadata.TranslatedAt = now + + if err := w.TenderRepo.Update(context.Background(), t); err != nil { + w.Logger.Error("Failed to persist translated tender", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "notice_id": t.NoticePublicationID, + "target_language": w.TargetLanguage, + "persistence_error": err.Error(), + }) + return + } + + w.Logger.Info("Tender translated successfully", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "notice_id": t.NoticePublicationID, + "target_language": w.TargetLanguage, + }) +} diff --git a/internal/tender/form.go b/internal/tender/form.go index ff6373f..0a5e7b6 100644 --- a/internal/tender/form.go +++ b/internal/tender/form.go @@ -70,6 +70,20 @@ type TenderResponse struct { Language string `json:"language,omitempty"` } +// TenderDocumentResponse represents a scraped tender document available for download. +type TenderDocumentResponse struct { + Filename string `json:"filename"` + Size int64 `json:"size"` + LastModified int64 `json:"last_modified"` + DocumentType string `json:"document_type,omitempty"` +} + +// TenderDocumentsResponse represents scraped documents for a tender. +type TenderDocumentsResponse struct { + TenderID string `json:"tender_id"` + Documents []TenderDocumentResponse `json:"documents"` +} + type OrganizationResponse struct { Name string `json:"name"` } diff --git a/internal/tender/handler.go b/internal/tender/handler.go index be1e80a..15f1aaa 100644 --- a/internal/tender/handler.go +++ b/internal/tender/handler.go @@ -1,6 +1,10 @@ package tender import ( + "errors" + "mime" + "net/http" + "strconv" "time" "tm/pkg/logger" "tm/pkg/response" @@ -361,6 +365,71 @@ func (h *TenderHandler) GetDocumentSummaries(c echo.Context) error { return response.Success(c, summaries, "Document summaries retrieved successfully") } +// GetDocuments retrieves scraped documents for a tender +// @Summary Get scraped documents for a tender +// @Description Retrieve scraped document metadata for a specific tender +// @Tags Tenders +// @Produce json +// @Param id path string true "Tender ID" +// @Success 200 {object} response.APIResponse{data=TenderDocumentsResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 404 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Router /api/v1/tenders/{id}/documents [get] +func (h *TenderHandler) GetDocuments(c echo.Context) error { + id := c.Param("id") + if id == "" { + return response.BadRequest(c, "Tender ID is required", "ID parameter cannot be empty") + } + + documents, err := h.service.GetDocuments(c.Request().Context(), id) + if err != nil { + return response.NotFound(c, "Tender documents not found") + } + + return response.Success(c, documents, "Tender documents retrieved successfully") +} + +// DownloadDocuments downloads all scraped documents for a tender +// @Summary Download scraped documents for a tender +// @Description Stream all scraped tender documents from storage as a ZIP archive +// @Tags Tenders +// @Produce application/zip +// @Param id path string true "Tender ID" +// @Success 200 {file} file +// @Failure 400 {object} response.APIResponse +// @Failure 404 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Router /api/v1/tenders/{id}/documents/download [get] +func (h *TenderHandler) DownloadDocuments(c echo.Context) error { + id := c.Param("id") + if id == "" { + return response.BadRequest(c, "Tender ID is required", "ID parameter cannot be empty") + } + + download, err := h.service.DownloadDocuments(c.Request().Context(), id) + if err != nil { + if errors.Is(err, ErrTenderDocumentNotFound) { + return response.NotFound(c, "Tender documents not found") + } + if errors.Is(err, ErrTenderDocumentStorageUnavailable) { + return response.InternalServerError(c, "Tender document storage is unavailable") + } + return response.InternalServerError(c, "Failed to download tender documents") + } + defer download.Reader.Close() + + headers := c.Response().Header() + headers.Set(echo.HeaderContentDisposition, mime.FormatMediaType("attachment", map[string]string{ + "filename": download.Filename, + })) + if download.Size > 0 { + headers.Set(echo.HeaderContentLength, strconv.FormatInt(download.Size, 10)) + } + + return c.Stream(http.StatusOK, download.ContentType, download.Reader) +} + // GetAISummary retrieves the AI-generated overall summary for a tender // @Summary Get AI-generated tender summary // @Description Retrieve the AI-generated overall summary for a tender from the AI pipeline storage diff --git a/internal/tender/repository.go b/internal/tender/repository.go index 6109c0f..b46cd57 100644 --- a/internal/tender/repository.go +++ b/internal/tender/repository.go @@ -25,6 +25,7 @@ type TenderRepository interface { GetExpiredTenders(ctx context.Context, beforeDate int64) ([]Tender, error) GetUnScrapedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error) GetUnSummarizedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error) + GetUnTranslatedTenders(ctx context.Context, language string, limit, skip int) ([]Tender, int64, error) Update(ctx context.Context, tender *Tender) error Delete(ctx context.Context, id string) error GetStatistics(ctx context.Context) (*TenderStatistics, error) @@ -337,6 +338,52 @@ func (r *tenderRepository) GetUnSummarizedTenders(ctx context.Context, limit, sk return result.Items, result.TotalCount, nil } +// GetUnTranslatedTenders retrieves tenders that don't have a translation for the given language yet. +func (r *tenderRepository) GetUnTranslatedTenders(ctx context.Context, language string, limit, skip int) ([]Tender, int64, error) { + translationKey := "translations." + language + metadataKey := "processing_metadata.translated_data." + language + + filter := bson.M{ + "notice_publication_id": bson.M{"$ne": ""}, + "title": bson.M{"$ne": ""}, + "description": bson.M{"$ne": ""}, + "$and": []bson.M{ + { + "$or": []bson.M{ + {translationKey: bson.M{"$exists": false}}, + {translationKey: nil}, + }, + }, + { + "$or": []bson.M{ + {metadataKey: bson.M{"$exists": false}}, + {metadataKey: bson.M{"$ne": "done"}}, + }, + }, + }, + } + + pagination := orm.Pagination{ + Limit: limit, + Skip: skip, + SortField: "created_at", + SortOrder: 1, // Oldest first + } + + result, err := r.ormRepo.FindAll(ctx, filter, pagination) + if err != nil { + r.logger.Error("Failed to get untranslated tenders", map[string]interface{}{ + "language": language, + "limit": limit, + "skip": skip, + "error": err.Error(), + }) + return nil, 0, err + } + + return result.Items, result.TotalCount, nil +} + // GetStatistics calculates tender statistics using aggregation func (r *tenderRepository) GetStatistics(ctx context.Context) (*TenderStatistics, error) { stats := &TenderStatistics{ diff --git a/internal/tender/service.go b/internal/tender/service.go index fa3f559..f4d9e87 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -1,9 +1,13 @@ package tender import ( + "archive/zip" "context" "errors" "fmt" + "io" + "path" + "path/filepath" "strings" "time" "tm/internal/company" @@ -22,6 +26,16 @@ type AISummarizerClient interface { // AISummarizerStorage defines the interface for fetching summaries from MinIO storage type AISummarizerStorage interface { GetSummaryFromStorage(ctx context.Context, noticeID string) (string, error) + ListDocuments(ctx context.Context, noticeID string) ([]ai_summarizer.StoredDocument, error) + DownloadDocument(ctx context.Context, objectName string) (io.ReadCloser, error) +} + +// TenderDocumentDownload contains an authorized document stream. +type TenderDocumentDownload struct { + Reader io.ReadCloser + Filename string + ContentType string + Size int64 } // TenderService interface defines tender business logic operations @@ -33,6 +47,8 @@ type Service interface { Search(ctx context.Context, req *SearchForm, pagination *response.Pagination) (*SearchResponse, error) Recommend(ctx context.Context, form *SearchForm, pagination *response.Pagination) (*SearchResponse, error) GetDocumentSummaries(ctx context.Context, id string) ([]DocumentSummary, error) + GetDocuments(ctx context.Context, id string) (*TenderDocumentsResponse, error) + DownloadDocuments(ctx context.Context, id string) (*TenderDocumentDownload, error) // AI summarization operations GetAISummary(ctx context.Context, id string) (*AISummaryResponse, error) @@ -58,6 +74,11 @@ type tenderService struct { const aiSummaryEnrichmentTimeout = 2 * time.Second +var ( + ErrTenderDocumentStorageUnavailable = errors.New("tender document storage is unavailable") + ErrTenderDocumentNotFound = errors.New("tender document not found") +) + // NewService creates a new tender service func NewService(repository TenderRepository, companyService company.Service, customerService customer.Service, logger logger.Logger, aiClient AISummarizerClient, aiStorage AISummarizerStorage, defaultLanguage string) Service { if strings.TrimSpace(defaultLanguage) == "" { @@ -241,6 +262,178 @@ func (s *tenderService) GetDocumentSummaries(ctx context.Context, id string) ([] return tender.DocumentSummaries, nil } +// GetDocuments retrieves scraped document metadata for a tender. +func (s *tenderService) GetDocuments(ctx context.Context, id string) (*TenderDocumentsResponse, error) { + if id == "" { + return nil, fmt.Errorf("tender ID cannot be empty") + } + if s.aiSummarizerStorage == nil { + return nil, ErrTenderDocumentStorageUnavailable + } + + tender, err := s.repository.GetByID(ctx, id) + if err != nil { + s.logger.Error("Failed to get tender for documents", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to get tender: %w", err) + } + if tender == nil { + return nil, fmt.Errorf("tender not found") + } + if strings.TrimSpace(tender.NoticePublicationID) == "" { + return nil, fmt.Errorf("tender notice publication ID is empty") + } + + storedDocuments, err := s.aiSummarizerStorage.ListDocuments(ctx, tender.NoticePublicationID) + if err != nil { + s.logger.Error("Failed to list tender documents from storage", map[string]interface{}{ + "tender_id": id, + "notice_id": tender.NoticePublicationID, + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to list tender documents: %w", err) + } + + documents := make([]TenderDocumentResponse, 0, len(storedDocuments)) + for _, doc := range storedDocuments { + documents = append(documents, TenderDocumentResponse{ + Filename: doc.DocumentName, + Size: doc.Size, + LastModified: doc.LastModified, + DocumentType: doc.DocumentType, + }) + } + + s.logger.Info("Retrieved scraped documents for tender", map[string]interface{}{ + "tender_id": id, + "notice_id": tender.NoticePublicationID, + "documents_count": len(documents), + }) + + return &TenderDocumentsResponse{ + TenderID: id, + Documents: documents, + }, nil +} + +// DownloadDocuments returns a ZIP stream containing all documents for a tender. +func (s *tenderService) DownloadDocuments(ctx context.Context, id string) (*TenderDocumentDownload, error) { + if id == "" { + return nil, fmt.Errorf("tender ID cannot be empty") + } + if s.aiSummarizerStorage == nil { + return nil, ErrTenderDocumentStorageUnavailable + } + + tender, err := s.repository.GetByID(ctx, id) + if err != nil { + s.logger.Error("Failed to get tender for document download", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to get tender: %w", err) + } + if tender == nil { + return nil, fmt.Errorf("tender not found") + } + if strings.TrimSpace(tender.NoticePublicationID) == "" { + return nil, ErrTenderDocumentNotFound + } + + documents, err := s.aiSummarizerStorage.ListDocuments(ctx, tender.NoticePublicationID) + if err != nil { + s.logger.Error("Failed to list tender documents for archive", map[string]interface{}{ + "tender_id": id, + "notice_id": tender.NoticePublicationID, + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to list tender documents: %w", err) + } + if len(documents) == 0 { + return nil, ErrTenderDocumentNotFound + } + + reader, writer := io.Pipe() + go s.writeDocumentsArchive(ctx, writer, documents) + + filename := fmt.Sprintf("%s-documents.zip", tender.NoticePublicationID) + s.logger.Info("Tender documents archive download prepared", map[string]interface{}{ + "tender_id": id, + "notice_id": tender.NoticePublicationID, + "documents_count": len(documents), + }) + + return &TenderDocumentDownload{ + Reader: reader, + Filename: filename, + ContentType: "application/zip", + }, nil +} + +func (s *tenderService) writeDocumentsArchive(ctx context.Context, writer *io.PipeWriter, documents []ai_summarizer.StoredDocument) { + zipWriter := zip.NewWriter(writer) + usedNames := make(map[string]int, len(documents)) + + for _, doc := range documents { + select { + case <-ctx.Done(): + zipWriter.Close() + writer.CloseWithError(ctx.Err()) + return + default: + } + + if err := s.addDocumentToArchive(ctx, zipWriter, usedNames, doc); err != nil { + zipWriter.Close() + writer.CloseWithError(err) + return + } + } + + if err := zipWriter.Close(); err != nil { + writer.CloseWithError(err) + return + } + writer.Close() +} + +func (s *tenderService) addDocumentToArchive(ctx context.Context, zipWriter *zip.Writer, usedNames map[string]int, doc ai_summarizer.StoredDocument) error { + reader, err := s.aiSummarizerStorage.DownloadDocument(ctx, doc.ObjectName) + if err != nil { + return fmt.Errorf("failed to open document %s: %w", doc.DocumentName, err) + } + defer reader.Close() + + entry, err := zipWriter.Create(uniqueArchiveName(usedNames, doc.DocumentName)) + if err != nil { + return fmt.Errorf("failed to create archive entry for %s: %w", doc.DocumentName, err) + } + + if _, err := io.Copy(entry, reader); err != nil { + return fmt.Errorf("failed to write document %s to archive: %w", doc.DocumentName, err) + } + return nil +} + +func uniqueArchiveName(usedNames map[string]int, filename string) string { + cleanName := path.Base(filename) + if cleanName == "." || cleanName == "/" || cleanName == "" { + cleanName = "document" + } + + count := usedNames[cleanName] + usedNames[cleanName] = count + 1 + if count == 0 { + return cleanName + } + + extension := filepath.Ext(cleanName) + base := strings.TrimSuffix(cleanName, extension) + return fmt.Sprintf("%s-%d%s", base, count+1, extension) +} + // ListTenders retrieves tenders with pagination and filtering func (s *tenderService) Search(ctx context.Context, form *SearchForm, pagination *response.Pagination) (*SearchResponse, error) { s.logger.Info("Listing tenders", map[string]interface{}{ diff --git a/pkg/ai_summarizer/entities.go b/pkg/ai_summarizer/entities.go index 9dcd4a1..94b4baa 100644 --- a/pkg/ai_summarizer/entities.go +++ b/pkg/ai_summarizer/entities.go @@ -30,6 +30,15 @@ type DocumentSummary struct { Error string `json:"error"` // empty string when no error } +// StoredDocument represents a document object stored for a tender in MinIO. +type StoredDocument struct { + ObjectName string `json:"-"` + DocumentName string `json:"document_name"` + Size int64 `json:"size"` + LastModified int64 `json:"last_modified"` + DocumentType string `json:"document_type,omitempty"` +} + // TranslateRequest represents the request payload sent to the AI service // POST /ai/translate endpoint. type TranslateRequest struct { diff --git a/pkg/ai_summarizer/storage.go b/pkg/ai_summarizer/storage.go index fa79531..8b8464b 100644 --- a/pkg/ai_summarizer/storage.go +++ b/pkg/ai_summarizer/storage.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "net/http" + "path/filepath" "strings" "github.com/minio/minio-go/v7" @@ -153,6 +154,89 @@ func (s *StorageClient) IsSummaryReady(ctx context.Context, noticeID string) (bo return tenderJSON.SummarizeStatus.Done, nil } +// ListDocuments lists document files stored under /. +func (s *StorageClient) ListDocuments(ctx context.Context, noticeID string) ([]StoredDocument, error) { + noticeID = strings.TrimSpace(noticeID) + if noticeID == "" { + return nil, fmt.Errorf("ai_summarizer: noticeID is required") + } + + prefix := fmt.Sprintf("%s/", noticeID) + objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{ + Prefix: prefix, + Recursive: true, + }) + + documents := make([]StoredDocument, 0) + for object := range objectCh { + if object.Err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to list documents for notice %s: %w", noticeID, object.Err) + } + if !isTenderDocumentObject(noticeID, object.Key) { + continue + } + + documentName := filepath.Base(object.Key) + documents = append(documents, StoredDocument{ + ObjectName: object.Key, + DocumentName: documentName, + Size: object.Size, + LastModified: object.LastModified.Unix(), + DocumentType: documentType(documentName), + }) + } + + s.logger.Info("Listed tender documents from storage", map[string]interface{}{ + "notice_id": noticeID, + "documents_count": len(documents), + }) + + return documents, nil +} + +func isTenderDocumentObject(noticeID, objectName string) bool { + if objectName == "" || strings.HasSuffix(objectName, "/") { + return false + } + + relativePath := strings.TrimPrefix(objectName, fmt.Sprintf("%s/", noticeID)) + if relativePath == "" { + return false + } + + filename := strings.ToLower(filepath.Base(relativePath)) + switch filename { + case "tender.json", "ai_cache.db": + return false + } + + switch strings.ToLower(filepath.Ext(filename)) { + case ".json", ".db", ".sqlite", ".sqlite3": + return false + default: + return true + } +} + +// DownloadDocument opens a document object from the configured MinIO bucket. +func (s *StorageClient) DownloadDocument(ctx context.Context, objectName string) (io.ReadCloser, error) { + objectName = strings.TrimSpace(objectName) + if objectName == "" { + return nil, fmt.Errorf("ai_summarizer: objectName is required") + } + + object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectName, minio.GetObjectOptions{}) + if err != nil { + var minioErr minio.ErrorResponse + if errors.As(err, &minioErr) && minioErr.StatusCode == http.StatusNotFound { + return nil, ErrObjectNotFound + } + return nil, fmt.Errorf("ai_summarizer: failed to get document object: %w", err) + } + + return object, nil +} + // GetAllSummaries lists all tender.json files in the MinIO bucket and returns // the overall_summary for each tender where summarize_status.done is true. // This is useful for batch-retrieving all summaries that the AI pipeline has @@ -233,3 +317,11 @@ func countDone(items []TenderSummaryItem) int { } return count } + +func documentType(filename string) string { + extension := strings.TrimPrefix(strings.ToLower(filepath.Ext(filename)), ".") + if extension == "" { + return "unknown" + } + return extension +}