2e112fe08c
continuous-integration/drone/push Build is passing
- Changed the default MinIO bucket name from "opplens-documents" to "opplens" across multiple configuration files. - Introduced caching for dashboard statistics in the service layer to improve performance and reduce redundant data fetching. - Implemented a mutex for thread-safe access to cached statistics, ensuring data integrity during concurrent requests. This update streamlines the configuration for the AI summarizer and optimizes the dashboard service, enhancing overall system efficiency.
880 lines
28 KiB
Go
880 lines
28 KiB
Go
package ai_summarizer
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/minio/minio-go/v7"
|
|
"github.com/minio/minio-go/v7/pkg/credentials"
|
|
|
|
"tm/pkg/logger"
|
|
)
|
|
|
|
// procedurePrefix is prepended to every contract folder id in the MinIO key.
|
|
// The AI team's storage layout in bucket opplens is:
|
|
//
|
|
// PROC_<contractFolderID>/<noticePublicationID>/tender.json
|
|
// PROC_<contractFolderID>/<noticePublicationID>/documents/{files}
|
|
const procedurePrefix = "PROC_"
|
|
|
|
// StorageClient wraps a MinIO client configured to connect to the AI service's
|
|
// MinIO bucket and retrieve tender.json files.
|
|
type StorageClient struct {
|
|
client *minio.Client
|
|
config *Config
|
|
logger logger.Logger
|
|
}
|
|
|
|
// TenderSummaryItem represents a single tender's summary retrieved from storage.
|
|
type TenderSummaryItem struct {
|
|
ContractFolderID string `json:"contract_folder_id"`
|
|
NoticeID string `json:"notice_id"`
|
|
OverallSummary string `json:"overall_summary"`
|
|
Done bool `json:"done"`
|
|
}
|
|
|
|
// 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,
|
|
normalizeContractFolderID(contractFolderID),
|
|
strings.TrimSpace(noticeKey),
|
|
)
|
|
}
|
|
|
|
// 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
|
|
// notice publication id (e.g. "00322263-2026") so it matches the AI team's
|
|
// folder convention ("322263-2026"). Empty input is preserved.
|
|
func normalizeNoticeKey(noticePublicationID string) string {
|
|
id := strings.TrimSpace(noticePublicationID)
|
|
if id == "" {
|
|
return ""
|
|
}
|
|
// Strip leading zeros from the numeric portion only; keep the year suffix
|
|
// (and any non-numeric tail) intact.
|
|
out := strings.TrimLeft(id, "0")
|
|
if out == "" || strings.HasPrefix(out, "-") {
|
|
// All-zero numeric portion: keep a single "0" so the key is never empty
|
|
// and never starts with "-".
|
|
return "0" + out
|
|
}
|
|
return out
|
|
}
|
|
|
|
// NewStorageClient creates a new MinIO storage client for the AI service.
|
|
// It establishes a connection to the external MinIO instance and verifies
|
|
// that the configured bucket exists.
|
|
func NewStorageClient(config *Config, log logger.Logger) (*StorageClient, error) {
|
|
if config == nil {
|
|
config = DefaultConfig()
|
|
}
|
|
if err := config.ValidateStorage(); err != nil {
|
|
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)
|
|
}
|
|
|
|
storage := &StorageClient{
|
|
client: minioClient,
|
|
config: config,
|
|
logger: log,
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
// BucketName returns the configured MinIO bucket used for AI pipeline artefacts.
|
|
func (s *StorageClient) BucketName() string {
|
|
if s == nil || s.config == nil {
|
|
return ""
|
|
}
|
|
return s.config.MinioBucket
|
|
}
|
|
|
|
// 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 s.logMinIOFailuref("bucket_exists", err, nil, "bucket exists check for %q", s.config.MinioBucket)
|
|
}
|
|
if !exists {
|
|
err := fmt.Errorf("%w: %q", ErrBucketNotFound, s.config.MinioBucket)
|
|
return s.logMinIOFailure("bucket_exists", err, map[string]interface{}{
|
|
"bucket": 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_<contractFolderID>/<notice>/ 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 "", s.logMinIOFailure("stat_tender_json", err, map[string]interface{}{
|
|
"prefix": prefix,
|
|
"contract_folder_id": contractFolderID,
|
|
"notice_id": noticePublicationID,
|
|
})
|
|
}
|
|
}
|
|
|
|
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 "", s.logMinIOFailure("list_notice_folder", object.Err, map[string]interface{}{
|
|
"notice_id": noticePublicationID,
|
|
"contract_folder_id": contractFolderID,
|
|
"notice_key": noticeKey,
|
|
})
|
|
}
|
|
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
|
|
// PROC_<contractFolderID>/<noticePublicationID>/tender.json file from the AI
|
|
// service's MinIO bucket, parses it, and returns the tender summary text.
|
|
//
|
|
// Summary text is read from summary (current) or overall_summary (legacy).
|
|
//
|
|
// It returns:
|
|
// - (summary, nil) on success
|
|
// - ("", ErrObjectNotFound) if the tender.json does not exist (404)
|
|
// - ("", ErrSummaryNotReady) if the file exists but summarize_status.done is false
|
|
// - ("", ErrNoOverallSummary) if done is true but summary text is empty
|
|
// - ("", error) for any other failure
|
|
func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) (string, error) {
|
|
if strings.TrimSpace(contractFolderID) == "" {
|
|
return "", fmt.Errorf("ai_summarizer: contractFolderID is required")
|
|
}
|
|
if strings.TrimSpace(noticePublicationID) == "" {
|
|
return "", fmt.Errorf("ai_summarizer: noticePublicationID is required")
|
|
}
|
|
s.logger.Debug("Fetching tender.json from MinIO", map[string]interface{}{
|
|
"bucket": s.config.MinioBucket,
|
|
"contract_folder_id": contractFolderID,
|
|
"notice_id": noticePublicationID,
|
|
})
|
|
|
|
prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID)
|
|
if err != nil {
|
|
if !errors.Is(err, ErrObjectNotFound) {
|
|
s.logger.Warn("Failed to resolve MinIO path for tender summary", map[string]interface{}{
|
|
"endpoint": s.config.MinioEndpoint,
|
|
"bucket": s.config.MinioBucket,
|
|
"contract_folder_id": contractFolderID,
|
|
"notice_id": noticePublicationID,
|
|
"error": err.Error(),
|
|
})
|
|
}
|
|
return "", err
|
|
}
|
|
return s.getSummaryFromObjectKey(ctx, prefix+"tender.json", contractFolderID, noticePublicationID)
|
|
}
|
|
|
|
// GetTranslationFromStorage reads tender.json and returns the translation for a language.
|
|
// It prefers entries marked done in translation_status; if missing, it falls back to
|
|
// translations[<lang>] when title is non-empty (on-demand API may lag status updates).
|
|
func (s *StorageClient) GetTranslationFromStorage(ctx context.Context, contractFolderID, noticePublicationID, language string) (StoredTranslation, error) {
|
|
if strings.TrimSpace(contractFolderID) == "" {
|
|
return StoredTranslation{}, fmt.Errorf("ai_summarizer: contractFolderID is required")
|
|
}
|
|
if strings.TrimSpace(noticePublicationID) == "" {
|
|
return StoredTranslation{}, fmt.Errorf("ai_summarizer: noticePublicationID is required")
|
|
}
|
|
if strings.TrimSpace(language) == "" {
|
|
return StoredTranslation{}, fmt.Errorf("ai_summarizer: language is required")
|
|
}
|
|
|
|
prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID)
|
|
if err != nil {
|
|
return StoredTranslation{}, err
|
|
}
|
|
return s.getTranslationFromObjectKey(ctx, prefix+"tender.json", language)
|
|
}
|
|
|
|
// getTranslationFromObjectKey reads tender.json at objectKey and returns translation for language.
|
|
func (s *StorageClient) getTranslationFromObjectKey(ctx context.Context, objectKey, language string) (StoredTranslation, error) {
|
|
tenderJSON, err := s.readTenderJSONAtKey(ctx, objectKey)
|
|
if err != nil {
|
|
return StoredTranslation{}, err
|
|
}
|
|
|
|
if entry, ok := tenderJSON.translationForLanguage(language); ok {
|
|
return entry, nil
|
|
}
|
|
if entry, ok := tenderJSON.storedTranslationText(language); ok {
|
|
return entry, nil
|
|
}
|
|
if !languageInDone(tenderJSON.translationsDone(), language) {
|
|
return StoredTranslation{}, ErrTranslationNotReady
|
|
}
|
|
return StoredTranslation{}, ErrNoTranslation
|
|
}
|
|
|
|
// getSummaryFromObjectKey reads tender.json at objectKey and returns summary text when ready.
|
|
// contractFolderID and noticePublicationID are used for logging only (may be empty when unknown).
|
|
func (s *StorageClient) readTenderJSONAtKey(ctx context.Context, objectKey string) (*TenderJSON, 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 nil, ErrObjectNotFound
|
|
}
|
|
var minioErr minio.ErrorResponse
|
|
if errors.As(err, &minioErr) && minioErr.StatusCode == http.StatusForbidden {
|
|
return nil, s.logMinIOFailure("get_object", err, map[string]interface{}{
|
|
"object_key": objectKey,
|
|
})
|
|
}
|
|
return nil, s.logMinIOFailure("get_object", err, map[string]interface{}{
|
|
"object_key": objectKey,
|
|
})
|
|
}
|
|
defer object.Close()
|
|
|
|
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 nil, ErrObjectNotFound
|
|
}
|
|
return nil, s.logMinIOFailure("read_tender_json", err, map[string]interface{}{
|
|
"object_key": objectKey,
|
|
})
|
|
}
|
|
|
|
var tenderJSON TenderJSON
|
|
if err := json.Unmarshal(data, &tenderJSON); err != nil {
|
|
return nil, fmt.Errorf("ai_summarizer: failed to parse tender.json: %w", err)
|
|
}
|
|
return &tenderJSON, nil
|
|
}
|
|
|
|
func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey, contractFolderID, noticePublicationID string) (string, error) {
|
|
tenderJSON, err := s.readTenderJSONAtKey(ctx, objectKey)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
|
|
done, summaryText := tenderJSON.resolved()
|
|
if !done {
|
|
return "", ErrSummaryNotReady
|
|
}
|
|
if summaryText == "" {
|
|
return "", ErrNoOverallSummary
|
|
}
|
|
|
|
s.logger.Info("Retrieved tender summary from storage", map[string]interface{}{
|
|
"contract_folder_id": contractFolderID,
|
|
"notice_id": noticePublicationID,
|
|
"object_key": objectKey,
|
|
"summary_length": len(summaryText),
|
|
})
|
|
|
|
return summaryText, nil
|
|
}
|
|
|
|
// GetDocumentSummariesFromStorage reads per-document summaries from tender.json
|
|
// when summarize_status.done is true.
|
|
func (s *StorageClient) GetDocumentSummariesFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) ([]DocumentSummary, error) {
|
|
if strings.TrimSpace(contractFolderID) == "" {
|
|
return nil, fmt.Errorf("ai_summarizer: contractFolderID is required")
|
|
}
|
|
if strings.TrimSpace(noticePublicationID) == "" {
|
|
return nil, fmt.Errorf("ai_summarizer: noticePublicationID is required")
|
|
}
|
|
|
|
prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
tenderJSON, err := s.readTenderJSONAtKey(ctx, prefix+"tender.json")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
done, _ := tenderJSON.resolved()
|
|
if !done {
|
|
return nil, ErrSummaryNotReady
|
|
}
|
|
if len(tenderJSON.Summaries) == 0 {
|
|
return []DocumentSummary{}, nil
|
|
}
|
|
|
|
out := make([]DocumentSummary, len(tenderJSON.Summaries))
|
|
copy(out, tenderJSON.Summaries)
|
|
return out, nil
|
|
}
|
|
|
|
// IsSummaryReady is a convenience method that checks whether the tender.json
|
|
// exists and the summarize_status.done flag is true, without returning the
|
|
// actual summary text.
|
|
func (s *StorageClient) IsSummaryReady(ctx context.Context, contractFolderID, noticePublicationID string) (bool, error) {
|
|
if strings.TrimSpace(contractFolderID) == "" || strings.TrimSpace(noticePublicationID) == "" {
|
|
return false, fmt.Errorf("ai_summarizer: contractFolderID and noticePublicationID are required")
|
|
}
|
|
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"
|
|
tenderJSON, err := s.readTenderJSONAtKey(ctx, objectPath)
|
|
if err != nil {
|
|
if errors.Is(err, ErrObjectNotFound) {
|
|
return false, nil
|
|
}
|
|
return false, err
|
|
}
|
|
|
|
done, _ := tenderJSON.resolved()
|
|
return done, nil
|
|
}
|
|
|
|
// ListDocuments lists document files stored under
|
|
// PROC_<contractFolderID>/<noticePublicationID>/. Non-document artefacts
|
|
// (tender.json, AI cache files, etc.) are filtered out.
|
|
func (s *StorageClient) ListDocuments(ctx context.Context, contractFolderID, noticePublicationID string) ([]StoredDocument, error) {
|
|
if strings.TrimSpace(contractFolderID) == "" {
|
|
return nil, fmt.Errorf("ai_summarizer: contractFolderID is required")
|
|
}
|
|
if strings.TrimSpace(noticePublicationID) == "" {
|
|
return nil, fmt.Errorf("ai_summarizer: noticePublicationID is required")
|
|
}
|
|
|
|
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)
|
|
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,
|
|
"prefixes": prefixes,
|
|
"documents_count": len(documents),
|
|
})
|
|
|
|
return documents, nil
|
|
}
|
|
|
|
// ListProceduresWithDocuments scans MinIO for PROC_<contractFolderID>/…/documents/* keys
|
|
// and returns one entry per contract folder id that has at least one document file.
|
|
func (s *StorageClient) ListProceduresWithDocuments(ctx context.Context) ([]ProcedureDocumentsSummary, error) {
|
|
s.logger.Debug("Listing procedure folders with documents from MinIO", map[string]interface{}{
|
|
"bucket": s.config.MinioBucket,
|
|
"prefix": procedurePrefix,
|
|
})
|
|
|
|
byFolder := make(map[string]*ProcedureDocumentsSummary)
|
|
var fatalErr error
|
|
|
|
appendFromPrefix := func(prefix string) bool {
|
|
objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{
|
|
Prefix: prefix,
|
|
Recursive: true,
|
|
})
|
|
for object := range objectCh {
|
|
if object.Err != nil {
|
|
fatalErr = s.logMinIOFailure("list_procedure_documents", object.Err, map[string]interface{}{
|
|
"prefix": prefix,
|
|
})
|
|
return false
|
|
}
|
|
contractFolderID, noticePrefix, ok := contractFolderIDFromDocumentKey(object.Key)
|
|
if !ok {
|
|
continue
|
|
}
|
|
if !isTenderDocumentObject(noticePrefix, object.Key) {
|
|
continue
|
|
}
|
|
entry := byFolder[contractFolderID]
|
|
if entry == nil {
|
|
entry = &ProcedureDocumentsSummary{ContractFolderID: contractFolderID}
|
|
byFolder[contractFolderID] = entry
|
|
}
|
|
entry.DocumentCount++
|
|
modified := object.LastModified.Unix()
|
|
if modified > entry.LatestModified {
|
|
entry.LatestModified = modified
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
if !appendFromPrefix(procedurePrefix) {
|
|
return nil, fatalErr
|
|
}
|
|
if len(byFolder) == 0 {
|
|
if !appendFromPrefix("") {
|
|
return nil, fatalErr
|
|
}
|
|
}
|
|
if fatalErr != nil {
|
|
return nil, fatalErr
|
|
}
|
|
|
|
results := make([]ProcedureDocumentsSummary, 0, len(byFolder))
|
|
for _, entry := range byFolder {
|
|
results = append(results, *entry)
|
|
}
|
|
sort.Slice(results, func(i, j int) bool {
|
|
if results[i].LatestModified == results[j].LatestModified {
|
|
return results[i].ContractFolderID < results[j].ContractFolderID
|
|
}
|
|
return results[i].LatestModified > results[j].LatestModified
|
|
})
|
|
|
|
s.logger.Info("Listed procedure folders with documents from storage", map[string]interface{}{
|
|
"bucket": s.config.MinioBucket,
|
|
"procedure_count": len(results),
|
|
})
|
|
|
|
return results, nil
|
|
}
|
|
|
|
// contractFolderIDFromDocumentKey parses PROC_<contractFolderID>/<notice>/documents/<file>.
|
|
func contractFolderIDFromDocumentKey(key string) (contractFolderID, noticePrefix string, ok bool) {
|
|
key = strings.TrimPrefix(strings.TrimSpace(key), "/")
|
|
parts := strings.Split(key, "/")
|
|
if len(parts) < 4 {
|
|
return "", "", false
|
|
}
|
|
if !strings.EqualFold(parts[2], "documents") {
|
|
return "", "", false
|
|
}
|
|
contractFolderID = stripLeadingProcPrefix(parts[0])
|
|
if contractFolderID == "" {
|
|
return "", "", false
|
|
}
|
|
noticePrefix = strings.Join(parts[:3], "/") + "/"
|
|
return contractFolderID, noticePrefix, true
|
|
}
|
|
|
|
// isTenderDocumentObject reports whether an object key under the given prefix
|
|
// represents a downloadable tender document (PDF/DOCX/...) rather than an
|
|
// internal artefact such as tender.json or a cache file.
|
|
func isTenderDocumentObject(prefix, objectName string) bool {
|
|
if objectName == "" || strings.HasSuffix(objectName, "/") {
|
|
return false
|
|
}
|
|
|
|
relativePath := strings.TrimPrefix(objectName, prefix)
|
|
if relativePath == "" || relativePath == objectName {
|
|
// objectName did not start with prefix; ignore.
|
|
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 walks the MinIO bucket for tender.json objects. It first lists
|
|
// under the PROC_ prefix; if none match, it falls back to a full-bucket recursive
|
|
// scan (slower on large buckets) so alternate key layouts still work.
|
|
func (s *StorageClient) GetAllSummaries(ctx context.Context) ([]TenderSummaryItem, error) {
|
|
s.logger.Info("Listing all tender summaries from MinIO", map[string]interface{}{
|
|
"endpoint": s.config.MinioEndpoint,
|
|
"bucket": s.config.MinioBucket,
|
|
})
|
|
|
|
results := make([]TenderSummaryItem, 0)
|
|
seen := make(map[string]struct{})
|
|
var fatalErr error
|
|
|
|
appendFromPrefix := func(prefix string) bool {
|
|
s.logger.Debug("Listing tender.json files from MinIO", map[string]interface{}{
|
|
"bucket": s.config.MinioBucket,
|
|
"prefix": prefix,
|
|
})
|
|
|
|
objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{
|
|
Prefix: prefix,
|
|
Recursive: true,
|
|
})
|
|
|
|
for object := range objectCh {
|
|
if object.Err != nil {
|
|
fatalErr = s.logMinIOFailure("list_tender_json", object.Err, map[string]interface{}{
|
|
"prefix": prefix,
|
|
})
|
|
return false
|
|
}
|
|
key := object.Key
|
|
if !strings.HasSuffix(key, "/tender.json") {
|
|
continue
|
|
}
|
|
if _, dup := seen[key]; dup {
|
|
continue
|
|
}
|
|
|
|
contractFolderID, noticePublicationID, ok := parseTenderJSONKey(key)
|
|
if !ok {
|
|
s.logger.Debug("Skipping unparseable tender.json key", map[string]interface{}{
|
|
"key": key,
|
|
})
|
|
continue
|
|
}
|
|
seen[key] = struct{}{}
|
|
|
|
summary, err := s.GetSummaryFromStorage(ctx, contractFolderID, noticePublicationID)
|
|
if errors.Is(err, ErrObjectNotFound) {
|
|
summary, err = s.getSummaryFromObjectKey(ctx, key, contractFolderID, noticePublicationID)
|
|
}
|
|
if err != nil {
|
|
switch {
|
|
case errors.Is(err, ErrSummaryNotReady):
|
|
results = append(results, TenderSummaryItem{
|
|
ContractFolderID: contractFolderID,
|
|
NoticeID: noticePublicationID,
|
|
Done: false,
|
|
})
|
|
case errors.Is(err, ErrNoOverallSummary):
|
|
// File exists and pipeline reports done, but no summary text yet (or empty).
|
|
results = append(results, TenderSummaryItem{
|
|
ContractFolderID: contractFolderID,
|
|
NoticeID: noticePublicationID,
|
|
OverallSummary: "",
|
|
Done: true,
|
|
})
|
|
case errors.Is(err, ErrMinIOUnavailable):
|
|
fatalErr = err
|
|
return false
|
|
default:
|
|
s.logger.Debug("Skipping tender summary", map[string]interface{}{
|
|
"object_key": key,
|
|
"contract_folder_id": contractFolderID,
|
|
"notice_id": noticePublicationID,
|
|
"error": err.Error(),
|
|
})
|
|
}
|
|
continue
|
|
}
|
|
|
|
results = append(results, TenderSummaryItem{
|
|
ContractFolderID: contractFolderID,
|
|
NoticeID: noticePublicationID,
|
|
OverallSummary: summary,
|
|
Done: true,
|
|
})
|
|
}
|
|
return true
|
|
}
|
|
|
|
if !appendFromPrefix(procedurePrefix) {
|
|
return nil, fatalErr
|
|
}
|
|
if len(seen) == 0 {
|
|
s.logger.Info("No tender.json keys under PROC_; listing entire bucket for */tender.json (may be slower on large buckets)", map[string]interface{}{
|
|
"bucket": s.config.MinioBucket,
|
|
})
|
|
if !appendFromPrefix("") {
|
|
return nil, fatalErr
|
|
}
|
|
}
|
|
|
|
if fatalErr != nil {
|
|
return nil, fatalErr
|
|
}
|
|
|
|
s.logger.Info("Retrieved all summaries from storage", map[string]interface{}{
|
|
"endpoint": s.config.MinioEndpoint,
|
|
"bucket": s.config.MinioBucket,
|
|
"total_count": len(results),
|
|
"completed_count": countDone(results),
|
|
})
|
|
|
|
return results, nil
|
|
}
|
|
|
|
// parseTenderJSONKey splits an object key ending in /tender.json into contract folder
|
|
// id and notice publication id. Accepts PROC_/proc_ prefix or plain <folder>/<notice>/tender.json.
|
|
func parseTenderJSONKey(key string) (contractFolderID, noticePublicationID string, ok bool) {
|
|
key = strings.TrimPrefix(strings.TrimSpace(key), "/")
|
|
if !strings.HasSuffix(key, "/tender.json") {
|
|
return "", "", false
|
|
}
|
|
trimmed := strings.TrimSuffix(key, "/tender.json")
|
|
if trimmed == "" {
|
|
return "", "", false
|
|
}
|
|
last := strings.LastIndex(trimmed, "/")
|
|
if last < 0 {
|
|
return "", "", false
|
|
}
|
|
noticePublicationID = strings.TrimSpace(trimmed[last+1:])
|
|
parent := strings.TrimSpace(trimmed[:last])
|
|
if noticePublicationID == "" || parent == "" {
|
|
return "", "", false
|
|
}
|
|
contractFolderID = stripLeadingProcPrefix(parent)
|
|
if contractFolderID == "" {
|
|
return "", "", false
|
|
}
|
|
return contractFolderID, noticePublicationID, true
|
|
}
|
|
|
|
func stripLeadingProcPrefix(parent string) string {
|
|
if len(parent) >= 5 && strings.EqualFold(parent[:5], "proc_") {
|
|
return strings.TrimSpace(parent[5:])
|
|
}
|
|
return parent
|
|
}
|
|
|
|
// countDone counts how many items have Done=true
|
|
func countDone(items []TenderSummaryItem) int {
|
|
count := 0
|
|
for _, item := range items {
|
|
if item.Done {
|
|
count++
|
|
}
|
|
}
|
|
return count
|
|
}
|
|
|
|
func documentType(filename string) string {
|
|
extension := strings.TrimPrefix(strings.ToLower(filepath.Ext(filename)), ".")
|
|
if extension == "" {
|
|
return "unknown"
|
|
}
|
|
return extension
|
|
}
|