844 lines
27 KiB
Go
844 lines
27 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-documents 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
|
|
// when translation_status.done includes that language.
|
|
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 !languageInDone(tenderJSON.translationsDone(), language) {
|
|
return StoredTranslation{}, ErrTranslationNotReady
|
|
}
|
|
entry, ok := tenderJSON.translationForLanguage(language)
|
|
if !ok {
|
|
return StoredTranslation{}, ErrNoTranslation
|
|
}
|
|
return entry, nil
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// 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
|
|
}
|