Mazyar
460 lines
15 KiB
Go
460 lines
15 KiB
Go
package ai_summarizer
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"path/filepath"
|
|
"strings"
|
|
|
|
"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 is:
|
|
//
|
|
// PROC_<contractFolderID>/<noticePublicationID>/{tender.json, *.pdf, *.docx, ...}
|
|
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"`
|
|
}
|
|
|
|
// 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 {
|
|
return fmt.Sprintf(
|
|
"%s%s/%s/",
|
|
procedurePrefix,
|
|
strings.TrimSpace(contractFolderID),
|
|
normalizeNoticeKey(noticePublicationID),
|
|
)
|
|
}
|
|
|
|
// 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"
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
|
|
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
|
|
// PROC_<contractFolderID>/<noticePublicationID>/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, 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")
|
|
}
|
|
objectPath := tenderJSONKey(contractFolderID, noticePublicationID)
|
|
|
|
s.logger.Debug("Fetching tender.json from MinIO", map[string]interface{}{
|
|
"bucket": s.config.MinioBucket,
|
|
"path": objectPath,
|
|
})
|
|
|
|
return s.getSummaryFromObjectKey(ctx, objectPath, contractFolderID, noticePublicationID)
|
|
}
|
|
|
|
// getSummaryFromObjectKey reads tender.json at objectKey and returns overall_summary when ready.
|
|
// contractFolderID and noticePublicationID are used for logging only (may be empty when unknown).
|
|
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 {
|
|
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()
|
|
|
|
data, err := io.ReadAll(object)
|
|
if err != nil {
|
|
return "", fmt.Errorf("ai_summarizer: failed to read tender.json: %w", err)
|
|
}
|
|
|
|
var tenderJSON TenderJSON
|
|
if err := json.Unmarshal(data, &tenderJSON); err != nil {
|
|
return "", fmt.Errorf("ai_summarizer: failed to parse tender.json: %w", err)
|
|
}
|
|
|
|
done, overall := tenderJSON.resolved()
|
|
if !done {
|
|
return "", ErrSummaryNotReady
|
|
}
|
|
if overall == nil || strings.TrimSpace(*overall) == "" {
|
|
return "", ErrNoOverallSummary
|
|
}
|
|
|
|
summaryText := strings.TrimSpace(*overall)
|
|
|
|
s.logger.Info("Retrieved overall_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")
|
|
}
|
|
objectPath := tenderJSONKey(contractFolderID, noticePublicationID)
|
|
|
|
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)
|
|
}
|
|
|
|
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 := noticeObjectPrefix(contractFolderID, noticePublicationID)
|
|
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 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),
|
|
})
|
|
}
|
|
|
|
s.logger.Info("Listed tender documents from storage", map[string]interface{}{
|
|
"contract_folder_id": contractFolderID,
|
|
"notice_id": noticePublicationID,
|
|
"prefix": prefix,
|
|
"documents_count": len(documents),
|
|
})
|
|
|
|
return documents, nil
|
|
}
|
|
|
|
// 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) {
|
|
results := make([]TenderSummaryItem, 0)
|
|
seen := make(map[string]struct{})
|
|
|
|
appendFromPrefix := func(prefix string) {
|
|
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 {
|
|
s.logger.Error("Error listing MinIO objects", map[string]interface{}{
|
|
"prefix": prefix,
|
|
"error": object.Err.Error(),
|
|
})
|
|
continue
|
|
}
|
|
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,
|
|
})
|
|
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,
|
|
})
|
|
}
|
|
}
|
|
|
|
appendFromPrefix(procedurePrefix)
|
|
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,
|
|
})
|
|
appendFromPrefix("")
|
|
}
|
|
|
|
s.logger.Info("Retrieved all summaries from storage", map[string]interface{}{
|
|
"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
|
|
}
|