Mazyar
236 lines
7.0 KiB
Go
236 lines
7.0 KiB
Go
package ai_summarizer
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"strings"
|
|
|
|
"github.com/minio/minio-go/v7"
|
|
"github.com/minio/minio-go/v7/pkg/credentials"
|
|
|
|
"tm/pkg/logger"
|
|
)
|
|
|
|
// 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 {
|
|
NoticeID string `json:"notice_id"`
|
|
OverallSummary string `json:"overall_summary"`
|
|
Done bool `json:"done"`
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
|
|
minioClient, err := minio.New(config.MinioEndpoint, &minio.Options{
|
|
Creds: credentials.NewStaticV4(config.MinioAccessKey, config.MinioSecretKey, ""),
|
|
Secure: config.MinioUseSSL,
|
|
Region: config.MinioRegion,
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("ai_summarizer: failed to create MinIO client: %w", err)
|
|
}
|
|
|
|
return &StorageClient{
|
|
client: minioClient,
|
|
config: config,
|
|
logger: log,
|
|
}, nil
|
|
}
|
|
|
|
// GetSummaryFromStorage fetches the <noticeID>/tender.json file from the AI
|
|
// service's MinIO bucket, parses it, and returns the overall_summary string.
|
|
//
|
|
// 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 overall_summary is empty/null
|
|
// - ("", error) for any other failure
|
|
func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, noticeID string) (string, error) {
|
|
objectPath := fmt.Sprintf("%s/tender.json", noticeID)
|
|
|
|
s.logger.Debug("Fetching tender.json from MinIO", map[string]interface{}{
|
|
"bucket": s.config.MinioBucket,
|
|
"path": objectPath,
|
|
})
|
|
|
|
// Attempt to get the object
|
|
object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectPath, minio.GetObjectOptions{})
|
|
if err != nil {
|
|
// minio-go returns an error response that we can inspect
|
|
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)
|
|
}
|
|
}
|
|
return "", fmt.Errorf("ai_summarizer: failed to get object: %w", err)
|
|
}
|
|
defer object.Close()
|
|
|
|
// Read the object content
|
|
data, err := io.ReadAll(object)
|
|
if err != nil {
|
|
return "", fmt.Errorf("ai_summarizer: failed to read tender.json: %w", err)
|
|
}
|
|
|
|
// Parse the JSON
|
|
var tenderJSON TenderJSON
|
|
if err := json.Unmarshal(data, &tenderJSON); err != nil {
|
|
return "", fmt.Errorf("ai_summarizer: failed to parse tender.json: %w", err)
|
|
}
|
|
|
|
// Check if summarization is complete
|
|
if !tenderJSON.SummarizeStatus.Done {
|
|
return "", ErrSummaryNotReady
|
|
}
|
|
|
|
// Check if overall_summary is available
|
|
if tenderJSON.OverallSummary == nil || *tenderJSON.OverallSummary == "" {
|
|
return "", ErrNoOverallSummary
|
|
}
|
|
|
|
s.logger.Info("Retrieved overall_summary from storage", map[string]interface{}{
|
|
"notice_id": noticeID,
|
|
"summary_length": len(*tenderJSON.OverallSummary),
|
|
})
|
|
|
|
return *tenderJSON.OverallSummary, 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, noticeID string) (bool, error) {
|
|
objectPath := fmt.Sprintf("%s/tender.json", noticeID)
|
|
|
|
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
|
|
}
|
|
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)
|
|
}
|
|
|
|
var tenderJSON TenderJSON
|
|
if err := json.Unmarshal(data, &tenderJSON); err != nil {
|
|
return false, fmt.Errorf("ai_summarizer: failed to parse tender.json: %w", err)
|
|
}
|
|
|
|
return tenderJSON.SummarizeStatus.Done, 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
|
|
// already processed.
|
|
func (s *StorageClient) GetAllSummaries(ctx context.Context) ([]TenderSummaryItem, error) {
|
|
s.logger.Debug("Listing all tender.json files from MinIO", map[string]interface{}{
|
|
"bucket": s.config.MinioBucket,
|
|
})
|
|
|
|
// List all objects ending with /tender.json
|
|
objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{
|
|
Prefix: "",
|
|
Recursive: true,
|
|
})
|
|
|
|
var results []TenderSummaryItem
|
|
|
|
for object := range objectCh {
|
|
if object.Err != nil {
|
|
s.logger.Error("Error listing MinIO objects", map[string]interface{}{
|
|
"error": object.Err.Error(),
|
|
})
|
|
continue
|
|
}
|
|
|
|
// Only process tender.json files (path pattern: <notice_id>/tender.json)
|
|
if !strings.HasSuffix(object.Key, "/tender.json") {
|
|
continue
|
|
}
|
|
|
|
// Extract notice_id from the path
|
|
noticeID := strings.TrimSuffix(object.Key, "/tender.json")
|
|
if noticeID == "" {
|
|
continue
|
|
}
|
|
|
|
// Fetch and parse the tender.json
|
|
summary, err := s.GetSummaryFromStorage(ctx, noticeID)
|
|
if err != nil {
|
|
// Log but skip tenders that aren't ready or have errors
|
|
s.logger.Debug("Skipping tender summary", map[string]interface{}{
|
|
"notice_id": noticeID,
|
|
"error": err.Error(),
|
|
})
|
|
|
|
// Still include items that aren't done so the caller knows they exist
|
|
if errors.Is(err, ErrSummaryNotReady) {
|
|
results = append(results, TenderSummaryItem{
|
|
NoticeID: noticeID,
|
|
Done: false,
|
|
})
|
|
}
|
|
continue
|
|
}
|
|
|
|
results = append(results, TenderSummaryItem{
|
|
NoticeID: noticeID,
|
|
OverallSummary: summary,
|
|
Done: true,
|
|
})
|
|
}
|
|
|
|
s.logger.Info("Retrieved all summaries from storage", map[string]interface{}{
|
|
"total_count": len(results),
|
|
"completed_count": countDone(results),
|
|
})
|
|
|
|
return results, nil
|
|
}
|
|
|
|
// 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
|
|
}
|