From ca490f3acf341db5485ed3d67bf737d8e1372bc6 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Sat, 16 May 2026 11:01:13 +0330 Subject: [PATCH] get aiSummary and documents refactor --- internal/document_scraper/form.go | 3 +- pkg/ai_summarizer/minio_errors.go | 31 +++ pkg/ai_summarizer/storage.go | 326 ++++++++++++++++++++++++------ 3 files changed, 302 insertions(+), 58 deletions(-) create mode 100644 pkg/ai_summarizer/minio_errors.go diff --git a/internal/document_scraper/form.go b/internal/document_scraper/form.go index 23ee630..65c91bc 100644 --- a/internal/document_scraper/form.go +++ b/internal/document_scraper/form.go @@ -56,7 +56,8 @@ type DocumentScraperWinnerResponse struct { // ContractFolderID is the TED procedure identifier (BT-04 / ContractFolderID), // included so the scraper can build the MinIO upload path for this notice: // -// PROC_// +// PROC_//documents/ +// PROC_//tender.json type DocumentScraperTenderResponse struct { ContractFolderID string `json:"contract_folder_id"` NoticePublicationID string `json:"notice_publication_id"` diff --git a/pkg/ai_summarizer/minio_errors.go b/pkg/ai_summarizer/minio_errors.go new file mode 100644 index 0000000..6dafef6 --- /dev/null +++ b/pkg/ai_summarizer/minio_errors.go @@ -0,0 +1,31 @@ +package ai_summarizer + +import ( + "errors" + "net/http" + "strings" + + "github.com/minio/minio-go/v7" +) + +// isMinioNotFound reports whether err indicates the object or bucket key is missing. +// The MinIO Go SDK often returns success from GetObject and surfaces NoSuchKey on Read. +func isMinioNotFound(err error) bool { + if err == nil { + return false + } + var minioErr minio.ErrorResponse + if errors.As(err, &minioErr) { + if minioErr.StatusCode == http.StatusNotFound { + return true + } + switch strings.ToLower(minioErr.Code) { + case "nosuchkey", "notfound": + return true + } + } + msg := strings.ToLower(err.Error()) + return strings.Contains(msg, "does not exist") || + strings.Contains(msg, "nosuchkey") || + strings.Contains(msg, "not found") +} diff --git a/pkg/ai_summarizer/storage.go b/pkg/ai_summarizer/storage.go index 1eda914..448dadf 100644 --- a/pkg/ai_summarizer/storage.go +++ b/pkg/ai_summarizer/storage.go @@ -9,6 +9,7 @@ import ( "net/http" "path/filepath" "strings" + "time" "github.com/minio/minio-go/v7" "github.com/minio/minio-go/v7/pkg/credentials" @@ -17,9 +18,10 @@ import ( ) // procedurePrefix is prepended to every contract folder id in the MinIO key. -// The AI team's storage layout is: +// The AI team's storage layout in bucket opplens-documents is: // -// PROC_//{tender.json, *.pdf, *.docx, ...} +// PROC_//tender.json +// PROC_//documents/{files} const procedurePrefix = "PROC_" // StorageClient wraps a MinIO client configured to connect to the AI service's @@ -38,22 +40,65 @@ type TenderSummaryItem struct { Done bool `json:"done"` } -// noticeObjectPrefix returns the MinIO key prefix that contains all artefacts -// (tender.json + scraped documents) for a single notice within a procedure. -// The trailing slash is intentional so it can be passed directly to -// minio.ListObjectsOptions.Prefix and also concatenated with a filename. -func noticeObjectPrefix(contractFolderID, noticePublicationID string) string { +// noticeKeyVariants returns MinIO notice folder names to try. The pipeline may +// store either the raw TED id (e.g. "00175735-2026") or a zero-stripped form +// ("175735-2026"); we try the raw value first, then the normalized variant. +func noticeKeyVariants(noticePublicationID string) []string { + raw := strings.TrimSpace(noticePublicationID) + if raw == "" { + return nil + } + normalized := normalizeNoticeKey(raw) + if normalized == raw { + return []string{raw} + } + return []string{raw, normalized} +} + +// noticeObjectPrefix returns the MinIO key prefix for one notice folder name. +func noticeObjectPrefix(contractFolderID, noticeKey string) string { return fmt.Sprintf( "%s%s/%s/", procedurePrefix, - strings.TrimSpace(contractFolderID), - normalizeNoticeKey(noticePublicationID), + normalizeContractFolderID(contractFolderID), + strings.TrimSpace(noticeKey), ) } -// tenderJSONKey returns the MinIO key for a notice's tender.json (overall summary + status). -func tenderJSONKey(contractFolderID, noticePublicationID string) string { - return noticeObjectPrefix(contractFolderID, noticePublicationID) + "tender.json" +// noticeObjectPrefixes returns every prefix variant for a notice publication id. +func noticeObjectPrefixes(contractFolderID, noticePublicationID string) []string { + folderID := normalizeContractFolderID(contractFolderID) + keys := noticeKeyVariants(noticePublicationID) + prefixes := make([]string, 0, len(keys)) + seen := make(map[string]struct{}, len(keys)) + for _, key := range keys { + prefix := noticeObjectPrefix(folderID, key) + if _, dup := seen[prefix]; dup { + continue + } + seen[prefix] = struct{}{} + prefixes = append(prefixes, prefix) + } + return prefixes +} + +// tenderJSONKeyVariants returns object keys to try for tender.json. +func tenderJSONKeyVariants(contractFolderID, noticePublicationID string) []string { + prefixes := noticeObjectPrefixes(contractFolderID, noticePublicationID) + keys := make([]string, 0, len(prefixes)) + for _, prefix := range prefixes { + keys = append(keys, prefix+"tender.json") + } + return keys +} + +// normalizeContractFolderID strips an accidental PROC_ prefix from stored ids. +func normalizeContractFolderID(contractFolderID string) string { + id := strings.TrimSpace(contractFolderID) + if len(id) >= 5 && strings.EqualFold(id[:5], "proc_") { + return strings.TrimSpace(id[5:]) + } + return id } // normalizeNoticeKey trims the leading zeros that we add when persisting the @@ -86,20 +131,138 @@ func NewStorageClient(config *Config, log logger.Logger) (*StorageClient, error) return nil, fmt.Errorf("ai_summarizer: invalid storage config: %w", err) } + log.Info("Connecting to AI summarizer MinIO", map[string]interface{}{ + "endpoint": config.MinioEndpoint, + "use_ssl": config.MinioUseSSL, + "region": config.MinioRegion, + "bucket": config.MinioBucket, + }) + minioClient, err := minio.New(config.MinioEndpoint, &minio.Options{ Creds: credentials.NewStaticV4(config.MinioAccessKey, config.MinioSecretKey, ""), Secure: config.MinioUseSSL, Region: config.MinioRegion, }) if err != nil { + log.Error("Failed to create AI summarizer MinIO client", map[string]interface{}{ + "endpoint": config.MinioEndpoint, + "error": err.Error(), + }) return nil, fmt.Errorf("ai_summarizer: failed to create MinIO client: %w", err) } - return &StorageClient{ + storage := &StorageClient{ client: minioClient, config: config, logger: log, - }, nil + } + + timeout := config.MinioTimeout + if timeout <= 0 { + timeout = 30 * time.Second + } + verifyCtx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + if err := storage.verifyConnection(verifyCtx); err != nil { + log.Error("AI summarizer MinIO connection check failed", map[string]interface{}{ + "endpoint": config.MinioEndpoint, + "bucket": config.MinioBucket, + "error": err.Error(), + }) + return nil, fmt.Errorf("ai_summarizer: MinIO connection check failed: %w", err) + } + + log.Info("AI summarizer MinIO connection verified", map[string]interface{}{ + "endpoint": config.MinioEndpoint, + "bucket": config.MinioBucket, + }) + + return storage, nil +} + +// verifyConnection checks that MinIO is reachable and the configured bucket exists. +func (s *StorageClient) verifyConnection(ctx context.Context) error { + exists, err := s.client.BucketExists(ctx, s.config.MinioBucket) + if err != nil { + return fmt.Errorf("bucket exists check for %q: %w", s.config.MinioBucket, err) + } + if !exists { + return fmt.Errorf("%w: %q", ErrBucketNotFound, s.config.MinioBucket) + } + + buckets, err := s.client.ListBuckets(ctx) + if err != nil { + s.logger.Warn("MinIO bucket exists but ListBuckets failed (credentials may be bucket-scoped)", map[string]interface{}{ + "endpoint": s.config.MinioEndpoint, + "bucket": s.config.MinioBucket, + "error": err.Error(), + }) + return nil + } + + names := make([]string, 0, len(buckets)) + for _, b := range buckets { + names = append(names, b.Name) + } + s.logger.Debug("MinIO ListBuckets succeeded", map[string]interface{}{ + "endpoint": s.config.MinioEndpoint, + "bucket_count": len(names), + "target_bucket": s.config.MinioBucket, + }) + return nil +} + +// Ping checks that MinIO is still reachable and the configured bucket exists. +func (s *StorageClient) Ping(ctx context.Context) error { + return s.verifyConnection(ctx) +} + +// resolveNoticeStoragePrefix locates the MinIO folder for a notice. It first +// checks PROC_// using known id variants, then falls +// back to a bucket search by notice publication id when the stored contract +// folder id does not match the AI pipeline folder (common in local testing). +func (s *StorageClient) resolveNoticeStoragePrefix(ctx context.Context, contractFolderID, noticePublicationID string) (string, error) { + for _, prefix := range noticeObjectPrefixes(contractFolderID, noticePublicationID) { + _, err := s.client.StatObject(ctx, s.config.MinioBucket, prefix+"tender.json", minio.StatObjectOptions{}) + if err == nil { + return prefix, nil + } + if !isMinioNotFound(err) { + return "", fmt.Errorf("ai_summarizer: stat tender.json under %s: %w", prefix, err) + } + } + + for _, noticeKey := range noticeKeyVariants(noticePublicationID) { + marker := "/" + noticeKey + "/" + var resolved string + objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{ + Prefix: procedurePrefix, + Recursive: true, + }) + for object := range objectCh { + if object.Err != nil { + return "", fmt.Errorf("ai_summarizer: search notice folder: %w", object.Err) + } + if idx := strings.Index(object.Key, marker); idx >= 0 { + resolved = object.Key[:idx+len(marker)] + break + } + } + if resolved != "" { + s.logger.Info("Resolved MinIO notice folder by notice publication id", map[string]interface{}{ + "notice_id": noticePublicationID, + "requested_contract_folder_id": contractFolderID, + "resolved_prefix": resolved, + }) + return resolved, nil + } + } + + s.logger.Debug("No MinIO folder found for notice", map[string]interface{}{ + "contract_folder_id": contractFolderID, + "notice_id": noticePublicationID, + }) + return "", ErrObjectNotFound } // GetSummaryFromStorage fetches the @@ -119,14 +282,17 @@ func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolde if strings.TrimSpace(noticePublicationID) == "" { return "", fmt.Errorf("ai_summarizer: noticePublicationID is required") } - objectPath := tenderJSONKey(contractFolderID, noticePublicationID) - s.logger.Debug("Fetching tender.json from MinIO", map[string]interface{}{ - "bucket": s.config.MinioBucket, - "path": objectPath, + "bucket": s.config.MinioBucket, + "contract_folder_id": contractFolderID, + "notice_id": noticePublicationID, }) - return s.getSummaryFromObjectKey(ctx, objectPath, contractFolderID, noticePublicationID) + prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID) + if err != nil { + return "", err + } + return s.getSummaryFromObjectKey(ctx, prefix+"tender.json", contractFolderID, noticePublicationID) } // getSummaryFromObjectKey reads tender.json at objectKey and returns overall_summary when ready. @@ -134,17 +300,16 @@ func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolde func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey, contractFolderID, noticePublicationID string) (string, error) { object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectKey, minio.GetObjectOptions{}) if err != nil { + if isMinioNotFound(err) { + s.logger.Debug("MinIO object not found", map[string]interface{}{ + "bucket": s.config.MinioBucket, + "object_key": objectKey, + }) + return "", ErrObjectNotFound + } var minioErr minio.ErrorResponse - if errors.As(err, &minioErr) { - switch minioErr.StatusCode { - case http.StatusNotFound: - return "", ErrObjectNotFound - case http.StatusForbidden: - return "", fmt.Errorf("ai_summarizer: access denied to bucket %s: %w", s.config.MinioBucket, err) - default: - return "", fmt.Errorf("ai_summarizer: MinIO error (code=%s, status=%d): %w", - minioErr.Code, minioErr.StatusCode, err) - } + if errors.As(err, &minioErr) && minioErr.StatusCode == http.StatusForbidden { + return "", fmt.Errorf("ai_summarizer: access denied to bucket %s: %w", s.config.MinioBucket, err) } return "", fmt.Errorf("ai_summarizer: failed to get object: %w", err) } @@ -152,6 +317,13 @@ func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey, data, err := io.ReadAll(object) if err != nil { + if isMinioNotFound(err) { + s.logger.Debug("MinIO object not found on read", map[string]interface{}{ + "bucket": s.config.MinioBucket, + "object_key": objectKey, + }) + return "", ErrObjectNotFound + } return "", fmt.Errorf("ai_summarizer: failed to read tender.json: %w", err) } @@ -187,22 +359,37 @@ func (s *StorageClient) IsSummaryReady(ctx context.Context, contractFolderID, no if strings.TrimSpace(contractFolderID) == "" || strings.TrimSpace(noticePublicationID) == "" { return false, fmt.Errorf("ai_summarizer: contractFolderID and noticePublicationID are required") } - objectPath := tenderJSONKey(contractFolderID, noticePublicationID) + prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID) + if errors.Is(err, ErrObjectNotFound) { + return false, nil + } + if err != nil { + return false, err + } + objectPath := prefix + "tender.json" object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectPath, minio.GetObjectOptions{}) if err != nil { - var minioErr minio.ErrorResponse - if errors.As(err, &minioErr) && minioErr.StatusCode == http.StatusNotFound { - return false, nil // not found means not ready, but not an error + if isMinioNotFound(err) { + return false, nil } return false, fmt.Errorf("ai_summarizer: failed to check summary readiness: %w", err) } - defer object.Close() - data, err := io.ReadAll(object) - if err != nil { - return false, fmt.Errorf("ai_summarizer: failed to read tender.json: %w", err) + data, readErr := io.ReadAll(object) + closeErr := object.Close() + if readErr != nil { + if isMinioNotFound(readErr) { + return false, nil + } + return false, fmt.Errorf("ai_summarizer: failed to read tender.json: %w", readErr) } + if closeErr != nil { + s.logger.Warn("Failed to close tender.json object after readiness check", map[string]interface{}{ + "object_key": objectPath, + "error": closeErr.Error(), + }) + } var tenderJSON TenderJSON if err := json.Unmarshal(data, &tenderJSON); err != nil { @@ -224,35 +411,60 @@ func (s *StorageClient) ListDocuments(ctx context.Context, contractFolderID, not return nil, fmt.Errorf("ai_summarizer: noticePublicationID is required") } - prefix := noticeObjectPrefix(contractFolderID, noticePublicationID) - objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{ - Prefix: prefix, - Recursive: true, + prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID) + if errors.Is(err, ErrObjectNotFound) { + s.logger.Info("Listed tender documents from storage", map[string]interface{}{ + "contract_folder_id": contractFolderID, + "notice_id": noticePublicationID, + "documents_count": 0, + }) + return nil, nil + } + if err != nil { + return nil, err + } + prefixes := []string{prefix} + + s.logger.Debug("Listing tender documents from MinIO", map[string]interface{}{ + "bucket": s.config.MinioBucket, + "prefixes": prefixes, }) documents := make([]StoredDocument, 0) - for object := range objectCh { - if object.Err != nil { - return nil, fmt.Errorf("ai_summarizer: failed to list documents under %s: %w", prefix, object.Err) - } - if !isTenderDocumentObject(prefix, 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), + seenObjects := make(map[string]struct{}) + for _, prefix := range prefixes { + objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{ + Prefix: prefix, + Recursive: true, }) + + for object := range objectCh { + if object.Err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to list documents under %s: %w", prefix, object.Err) + } + if !isTenderDocumentObject(prefix, object.Key) { + continue + } + if _, dup := seenObjects[object.Key]; dup { + continue + } + seenObjects[object.Key] = struct{}{} + + 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{}{ "contract_folder_id": contractFolderID, "notice_id": noticePublicationID, - "prefix": prefix, + "prefixes": prefixes, "documents_count": len(documents), })