diff --git a/cmd/web/bootstrap/bootstrap.go b/cmd/web/bootstrap/bootstrap.go index 5a1cc77..17d1dc0 100644 --- a/cmd/web/bootstrap/bootstrap.go +++ b/cmd/web/bootstrap/bootstrap.go @@ -3,6 +3,7 @@ package bootstrap import ( "fmt" "time" + ai_summarizer "tm/pkg/ai_summarizer" "tm/pkg/authorization" "tm/pkg/config" "tm/pkg/filestore" @@ -333,3 +334,75 @@ func InitFileStore(mongoManager *mongo.ConnectionManager, log logger.Logger) (*f return fileStoreService, nil } + +// InitAISummarizerClient initializes the AI summarizer HTTP client for on-demand summarization +func InitAISummarizerClient(conf AISummarizerConfig, log logger.Logger) *ai_summarizer.Client { + if conf.APIBaseURL == "" { + log.Warn("AI Summarizer API base URL not configured, on-demand summarization will be unavailable", map[string]interface{}{}) + return nil + } + + cfg := &ai_summarizer.Config{ + APIBaseURL: conf.APIBaseURL, + APITimeout: conf.APITimeout, + APIRetryCount: conf.APIRetryCount, + APIRetryDelay: conf.APIRetryDelay, + MinioEndpoint: conf.MinioEndpoint, + MinioAccessKey: conf.MinioAccessKey, + MinioSecretKey: conf.MinioSecretKey, + MinioUseSSL: conf.MinioUseSSL, + MinioRegion: conf.MinioRegion, + MinioBucket: conf.MinioBucket, + } + + client, err := ai_summarizer.NewClient(cfg, log) + if err != nil { + log.Error("Failed to initialize AI Summarizer client", map[string]interface{}{ + "error": err.Error(), + }) + return nil + } + + log.Info("AI Summarizer HTTP client initialized successfully", map[string]interface{}{ + "base_url": conf.APIBaseURL, + "timeout": conf.APITimeout, + }) + + return client +} + +// InitAISummarizerStorage initializes the AI summarizer MinIO storage client +func InitAISummarizerStorage(conf AISummarizerConfig, log logger.Logger) *ai_summarizer.StorageClient { + if conf.MinioEndpoint == "" { + log.Warn("AI Summarizer MinIO endpoint not configured, storage-based summaries will be unavailable", map[string]interface{}{}) + return nil + } + + cfg := &ai_summarizer.Config{ + APIBaseURL: conf.APIBaseURL, + APITimeout: conf.APITimeout, + APIRetryCount: conf.APIRetryCount, + APIRetryDelay: conf.APIRetryDelay, + MinioEndpoint: conf.MinioEndpoint, + MinioAccessKey: conf.MinioAccessKey, + MinioSecretKey: conf.MinioSecretKey, + MinioUseSSL: conf.MinioUseSSL, + MinioRegion: conf.MinioRegion, + MinioBucket: conf.MinioBucket, + } + + storage, err := ai_summarizer.NewStorageClient(cfg, log) + if err != nil { + log.Error("Failed to initialize AI Summarizer storage client", map[string]interface{}{ + "error": err.Error(), + }) + return nil + } + + log.Info("AI Summarizer storage client initialized successfully", map[string]interface{}{ + "endpoint": conf.MinioEndpoint, + "bucket": conf.MinioBucket, + }) + + return storage +} diff --git a/cmd/web/bootstrap/config.go b/cmd/web/bootstrap/config.go index 9727d78..90175fe 100644 --- a/cmd/web/bootstrap/config.go +++ b/cmd/web/bootstrap/config.go @@ -22,6 +22,7 @@ type Config struct { UserAuth AuthConfig CustomerAuth AuthConfig DocumentScraper DocumentScraperConfig + AISummarizer AISummarizerConfig } type GLMConfig struct { @@ -88,3 +89,20 @@ type JWTConfig struct { type DocumentScraperConfig struct { APIKey string `env:"DOCUMENT_SCRAPER_API_KEY" envDefault:""` } + +// AISummarizerConfig holds configuration for the external AI summarizer service +type AISummarizerConfig struct { + // HTTP API settings + APIBaseURL string `env:"AI_SUMMARIZER_API_BASE_URL" envDefault:""` + APITimeout time.Duration `env:"AI_SUMMARIZER_API_TIMEOUT" envDefault:"120s"` + APIRetryCount int `env:"AI_SUMMARIZER_API_RETRY_COUNT" envDefault:"2"` + APIRetryDelay time.Duration `env:"AI_SUMMARIZER_API_RETRY_DELAY" envDefault:"3s"` + + // MinIO storage settings + MinioEndpoint string `env:"AI_SUMMARIZER_MINIO_ENDPOINT" envDefault:""` + MinioAccessKey string `env:"AI_SUMMARIZER_MINIO_ACCESS_KEY" envDefault:""` + MinioSecretKey string `env:"AI_SUMMARIZER_MINIO_SECRET_KEY" envDefault:""` + MinioUseSSL bool `env:"AI_SUMMARIZER_MINIO_USE_SSL" envDefault:"false"` + MinioRegion string `env:"AI_SUMMARIZER_MINIO_REGION" envDefault:"us-east-1"` + MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens-documents"` +} diff --git a/cmd/web/main.go b/cmd/web/main.go index b74c131..cb3acd2 100644 --- a/cmd/web/main.go +++ b/cmd/web/main.go @@ -162,6 +162,17 @@ func main() { // Initialize GLM service glmSDK := bootstrap.InitGLMService(conf.GLM, logger) + // Initialize AI Summarizer service + var aiSummarizerClient tender.AISummarizerClient + var aiSummarizerStorage tender.AISummarizerStorage + + if c := bootstrap.InitAISummarizerClient(conf.AISummarizer, logger); c != nil { + aiSummarizerClient = c + } + if s := bootstrap.InitAISummarizerStorage(conf.AISummarizer, logger); s != nil { + aiSummarizerStorage = s + } + // Initialize authorization service userAuthService := bootstrap.InitAuthorizationService(conf.UserAuth.JWT, redisClient, logger) customerAuthService := bootstrap.InitAuthorizationService(conf.CustomerAuth.JWT, redisClient, logger) @@ -196,7 +207,7 @@ func main() { categoryService := company_category.NewService(categoryRepository, logger) companyService := company.NewService(companyRepository, categoryService, logger) customerService := customer.NewService(customerRepository, logger, customerAuthService, companyService, redisClient, notificationSDK, customerValidator, glmSDK) - tenderService := tender.NewService(tenderRepository, companyService, customerService, logger) + tenderService := tender.NewService(tenderRepository, companyService, customerService, logger, aiSummarizerClient, aiSummarizerStorage) feedbackService := feedback.NewService(feedbackRepo, tenderService, companyService, logger) tenderApprovalService := tender_approval.NewService(tenderApprovalRepository, tenderService, logger) inquiryService := inquiry.NewService(inquiryRepository, notificationSDK, logger, xssPolicy) diff --git a/cmd/web/router/routes.go b/cmd/web/router/routes.go index a00fd20..212ff81 100644 --- a/cmd/web/router/routes.go +++ b/cmd/web/router/routes.go @@ -105,9 +105,12 @@ func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler { tendersGP.Use(userHandler.AuthMiddleware()) tendersGP.GET("", tenderHandler.Search) + tendersGP.GET("/ai-summaries", tenderHandler.GetAllAISummaries) tendersGP.GET("/:id", tenderHandler.Get) tendersGP.PUT("/:id", tenderHandler.Update) tendersGP.DELETE("/:id", tenderHandler.Delete) + tendersGP.GET("/:id/ai-summary", tenderHandler.GetAISummary) + tendersGP.POST("/:id/ai-summarize", tenderHandler.TriggerAISummarize) } // Admin Feedback Routes @@ -233,6 +236,7 @@ func RegisterPublicRoutes(e *echo.Echo, customerHandler *customer.Handler, tende tendersGP.GET("/recommend", tenderHandler.RecommendTenders) tendersGP.GET("/details/:id", tenderHandler.GetPublicTenderDetails) tendersGP.GET("/:id/document-summaries", tenderHandler.GetDocumentSummaries) + tendersGP.GET("/:id/ai-summary", tenderHandler.GetPublicAISummary) } // Public Feedback Routes diff --git a/internal/tender/form.go b/internal/tender/form.go index 0abd08d..8ea4a9d 100644 --- a/internal/tender/form.go +++ b/internal/tender/form.go @@ -65,6 +65,7 @@ type TenderResponse struct { Status TenderStatus `json:"status"` TenderID string `json:"tender_id"` CreatedAt int64 `json:"created_at"` + OverallSummary string `json:"overall_summary,omitempty"` // AI-generated summary from the pipeline } type OrganizationResponse struct { @@ -103,3 +104,26 @@ func (t *Tender) ToResponse() *TenderResponse { return response } + +// AISummaryResponse represents the response for getting an AI-generated summary +type AISummaryResponse struct { + NoticeID string `json:"notice_id"` + OverallSummary string `json:"overall_summary"` + Source string `json:"source"` // "storage" or "on_demand" +} + +// AISummarizeResponse represents the response for triggering on-demand AI summarization +type AISummarizeResponse struct { + NoticeID string `json:"notice_id"` + Summaries []AIDocumentSummary `json:"summaries"` + OverallSummary *string `json:"overall_summary"` +} + +// AIDocumentSummary represents a single document summary from the AI service +type AIDocumentSummary struct { + DocumentName string `json:"document_name"` + DocumentType string `json:"document_type"` + Summary string `json:"summary"` + SummaryModel string `json:"summary_model"` + Error string `json:"error,omitempty"` +} diff --git a/internal/tender/handler.go b/internal/tender/handler.go index 1966088..3253377 100644 --- a/internal/tender/handler.go +++ b/internal/tender/handler.go @@ -360,3 +360,114 @@ func (h *TenderHandler) GetDocumentSummaries(c echo.Context) error { return response.Success(c, summaries, "Document summaries retrieved successfully") } + +// GetAISummary retrieves the AI-generated overall summary for a tender +// @Summary Get AI-generated tender summary +// @Description Retrieve the AI-generated overall summary for a tender from the AI pipeline storage +// @Tags Admin-Tenders +// @Produce json +// @Param id path string true "Tender ID" +// @Success 200 {object} response.APIResponse{data=AISummaryResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 404 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Router /admin/v1/tenders/{id}/ai-summary [get] +// @Security BearerAuth +func (h *TenderHandler) GetAISummary(c echo.Context) error { + id := c.Param("id") + if id == "" { + return response.BadRequest(c, "Tender ID is required", "ID parameter cannot be empty") + } + + summary, err := h.service.GetAISummary(c.Request().Context(), id) + if err != nil { + h.logger.Error("Failed to get AI summary", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return response.NotFound(c, err.Error()) + } + + return response.Success(c, summary, "AI summary retrieved successfully") +} + +// TriggerAISummarize triggers on-demand AI summarization for a tender +// @Summary Trigger AI summarization +// @Description Trigger on-demand AI summarization for a tender by calling the external AI service +// @Tags Admin-Tenders +// @Produce json +// @Param id path string true "Tender ID" +// @Success 200 {object} response.APIResponse{data=AISummarizeResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 404 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Router /admin/v1/tenders/{id}/ai-summarize [post] +// @Security BearerAuth +func (h *TenderHandler) TriggerAISummarize(c echo.Context) error { + id := c.Param("id") + if id == "" { + return response.BadRequest(c, "Tender ID is required", "ID parameter cannot be empty") + } + + result, err := h.service.TriggerAISummarize(c.Request().Context(), id) + if err != nil { + h.logger.Error("Failed to trigger AI summarization", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return response.InternalServerError(c, err.Error()) + } + + return response.Success(c, result, "AI summarization completed successfully") +} + +// GetPublicAISummary retrieves the AI-generated overall summary for a tender (public endpoint) +// @Summary Get AI-generated tender summary (public) +// @Description Retrieve the AI-generated overall summary for a tender from the AI pipeline storage +// @Tags Tenders +// @Produce json +// @Param id path string true "Tender ID" +// @Success 200 {object} response.APIResponse{data=AISummaryResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 404 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Router /api/v1/tenders/{id}/ai-summary [get] +// @Security BearerAuth +func (h *TenderHandler) GetPublicAISummary(c echo.Context) error { + id := c.Param("id") + if id == "" { + return response.BadRequest(c, "Tender ID is required", "ID parameter cannot be empty") + } + + summary, err := h.service.GetAISummary(c.Request().Context(), id) + if err != nil { + h.logger.Error("Failed to get AI summary", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return response.NotFound(c, err.Error()) + } + + return response.Success(c, summary, "AI summary retrieved successfully") +} + +// GetAllAISummaries retrieves all AI-generated summaries from storage +// @Summary Get all AI-generated tender summaries +// @Description Retrieve all AI-generated summaries from the AI pipeline storage, including pending and completed ones +// @Tags Admin-Tenders +// @Produce json +// @Success 200 {object} response.APIResponse{data=[]ai_summarizer.TenderSummaryItem} +// @Failure 500 {object} response.APIResponse +// @Router /admin/v1/tenders/ai-summaries [get] +// @Security BearerAuth +func (h *TenderHandler) GetAllAISummaries(c echo.Context) error { + summaries, err := h.service.GetAllAISummaries(c.Request().Context()) + if err != nil { + h.logger.Error("Failed to get all AI summaries", map[string]interface{}{ + "error": err.Error(), + }) + return response.InternalServerError(c, err.Error()) + } + + return response.Success(c, summaries, "AI summaries retrieved successfully") +} diff --git a/internal/tender/service.go b/internal/tender/service.go index b1302a6..70be4ef 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -2,15 +2,27 @@ package tender import ( "context" + "errors" "fmt" "strings" "time" "tm/internal/company" "tm/internal/customer" + "tm/pkg/ai_summarizer" "tm/pkg/logger" "tm/pkg/response" ) +// AISummarizerClient defines the interface for on-demand AI summarization +type AISummarizerClient interface { + FetchSummaryOnDemand(ctx context.Context, reqBody ai_summarizer.SummarizeRequest) (*ai_summarizer.SummarizeResponse, error) +} + +// AISummarizerStorage defines the interface for fetching summaries from MinIO storage +type AISummarizerStorage interface { + GetSummaryFromStorage(ctx context.Context, noticeID string) (string, error) +} + // TenderService interface defines tender business logic operations type Service interface { // Tender operations @@ -21,6 +33,11 @@ type Service interface { Recommend(ctx context.Context, form *SearchForm, pagination *response.Pagination) (*SearchResponse, error) GetDocumentSummaries(ctx context.Context, id string) ([]DocumentSummary, error) + // AI summarization operations + GetAISummary(ctx context.Context, id string) (*AISummaryResponse, error) + TriggerAISummarize(ctx context.Context, id string) (*AISummarizeResponse, error) + GetAllAISummaries(ctx context.Context) ([]ai_summarizer.TenderSummaryItem, error) + // Maintenance operations UpdateExpiredTenders(ctx context.Context) error CleanupOldData(ctx context.Context, olderThan time.Duration, pagination *response.Pagination) error @@ -28,19 +45,23 @@ type Service interface { // tenderService implements TenderService interface type tenderService struct { - repository TenderRepository - companyService company.Service - customerService customer.Service - logger logger.Logger + repository TenderRepository + companyService company.Service + customerService customer.Service + logger logger.Logger + aiSummarizerClient AISummarizerClient + aiSummarizerStorage AISummarizerStorage } // NewService creates a new tender service -func NewService(repository TenderRepository, companyService company.Service, customerService customer.Service, logger logger.Logger) Service { +func NewService(repository TenderRepository, companyService company.Service, customerService customer.Service, logger logger.Logger, aiClient AISummarizerClient, aiStorage AISummarizerStorage) Service { return &tenderService{ - repository: repository, - companyService: companyService, - customerService: customerService, - logger: logger, + repository: repository, + companyService: companyService, + customerService: customerService, + logger: logger, + aiSummarizerClient: aiClient, + aiSummarizerStorage: aiStorage, } } @@ -59,7 +80,33 @@ func (s *tenderService) GetByID(ctx context.Context, id string) (*TenderResponse return nil, fmt.Errorf("failed to retrieve tender: %w", err) } - return tender.ToResponse(), nil + resp := tender.ToResponse() + + // Enrich with AI summary from MinIO storage (best-effort, non-blocking) + s.enrichWithAISummary(ctx, resp) + + return resp, nil +} + +// enrichWithAISummary attempts to fetch the AI-generated overall summary from +// MinIO storage and attaches it to the response. Failures are logged but do not +// cause the request to fail — the field will simply be omitted. +func (s *tenderService) enrichWithAISummary(ctx context.Context, resp *TenderResponse) { + if s.aiSummarizerStorage == nil || resp.NoticePublicationID == "" { + return + } + + summary, err := s.aiSummarizerStorage.GetSummaryFromStorage(ctx, resp.NoticePublicationID) + if err != nil { + s.logger.Debug("AI summary not available for tender", map[string]interface{}{ + "tender_id": resp.ID, + "notice_id": resp.NoticePublicationID, + "error": err.Error(), + }) + return + } + + resp.OverallSummary = summary } // UpdateTender updates an existing tender @@ -428,3 +475,190 @@ func (s *tenderService) CleanupOldData(ctx context.Context, olderThan time.Durat return nil } + +// GetAISummary retrieves the AI-generated summary for a tender. +// It first tries to fetch from MinIO storage (async pipeline result). +// If not found or not ready, it returns an appropriate error. +func (s *tenderService) GetAISummary(ctx context.Context, id string) (*AISummaryResponse, error) { + if id == "" { + return nil, fmt.Errorf("tender ID cannot be empty") + } + + // Get tender to find the notice_publication_id + t, err := s.repository.GetByID(ctx, id) + if err != nil { + s.logger.Error("Failed to get tender for AI summary", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to get tender: %w", err) + } + + if t == nil { + return nil, fmt.Errorf("tender not found") + } + + noticeID := t.NoticePublicationID + if noticeID == "" { + return nil, fmt.Errorf("tender has no notice publication ID") + } + + // Try to get summary from MinIO storage (async pipeline) + if s.aiSummarizerStorage != nil { + summary, err := s.aiSummarizerStorage.GetSummaryFromStorage(ctx, noticeID) + if err == nil { + return &AISummaryResponse{ + NoticeID: noticeID, + OverallSummary: summary, + Source: "storage", + }, nil + } + + // Log but don't fail — the summary might just not be ready yet + s.logger.Info("AI summary not available from storage", map[string]interface{}{ + "tender_id": id, + "notice_id": noticeID, + "error": err.Error(), + }) + + // Return specific error if summary not ready + if errors.Is(err, ai_summarizer.ErrSummaryNotReady) { + return nil, fmt.Errorf("AI summary is not ready yet, the pipeline is still processing") + } + if errors.Is(err, ai_summarizer.ErrObjectNotFound) { + return nil, fmt.Errorf("AI summary not found — this tender has not been processed by the AI pipeline yet") + } + + return nil, fmt.Errorf("failed to get AI summary: %w", err) + } + + return nil, fmt.Errorf("AI summarizer storage is not configured") +} + +// TriggerAISummarize triggers on-demand AI summarization for a tender. +// It calls the external AI service HTTP API to generate summaries. +func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AISummarizeResponse, error) { + if id == "" { + return nil, fmt.Errorf("tender ID cannot be empty") + } + + if s.aiSummarizerClient == nil { + return nil, fmt.Errorf("AI summarizer service is not configured") + } + + // Get tender to build the request + t, err := s.repository.GetByID(ctx, id) + if err != nil { + s.logger.Error("Failed to get tender for on-demand summarization", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to get tender: %w", err) + } + + if t == nil { + return nil, fmt.Errorf("tender not found") + } + + noticeID := t.NoticePublicationID + if noticeID == "" { + return nil, fmt.Errorf("tender has no notice publication ID") + } + + // Build the AI summarizer request + reqBody := ai_summarizer.SummarizeRequest{ + NoticeID: noticeID, + DocumentURL: t.DocumentURI, + Title: t.Title, + Description: t.Description, + Country: t.CountryCode, + SubmissionDeadline: time.UnixMilli(t.SubmissionDeadline).Format(time.RFC3339), + } + + if t.EstimatedValue > 0 { + ev := t.EstimatedValue + reqBody.EstimatedValue = &ev + } + if t.Currency != "" { + reqBody.Currency = t.Currency + } + if t.BuyerOrganization != nil { + reqBody.AuthorityName = t.BuyerOrganization.Name + } + + s.logger.Info("Triggering on-demand AI summarization", map[string]interface{}{ + "tender_id": id, + "notice_id": noticeID, + }) + + // Call the AI service + resp, err := s.aiSummarizerClient.FetchSummaryOnDemand(ctx, reqBody) + if err != nil { + s.logger.Error("On-demand AI summarization failed", map[string]interface{}{ + "tender_id": id, + "notice_id": noticeID, + "error": err.Error(), + }) + return nil, fmt.Errorf("AI summarization request failed: %w", err) + } + + // Map the response to our response type + summaries := make([]AIDocumentSummary, 0, len(resp.Summaries)) + for _, s := range resp.Summaries { + summaries = append(summaries, AIDocumentSummary{ + DocumentName: s.DocumentName, + DocumentType: s.DocumentType, + Summary: s.Summary, + SummaryModel: s.SummaryModel, + Error: s.Error, + }) + } + + s.logger.Info("On-demand AI summarization completed", map[string]interface{}{ + "tender_id": id, + "notice_id": noticeID, + "summaries_count": len(summaries), + }) + + return &AISummarizeResponse{ + NoticeID: resp.NoticeID, + Summaries: summaries, + OverallSummary: resp.OverallSummary, + }, nil +} + +// GetAllAISummaries retrieves all AI-generated summaries from the MinIO storage. +// It lists all tender.json files in the bucket and returns summaries for tenders +// where the AI pipeline has completed processing. +func (s *tenderService) GetAllAISummaries(ctx context.Context) ([]ai_summarizer.TenderSummaryItem, error) { + if s.aiSummarizerStorage == nil { + return nil, fmt.Errorf("AI summarizer storage is not configured") + } + + s.logger.Info("Retrieving all AI summaries from storage", map[string]interface{}{}) + + // We need to access the underlying StorageClient to call GetAllSummaries. + // Since aiSummarizerStorage is an interface, we type-assert to access GetAllSummaries. + type allSummaries interface { + GetAllSummaries(ctx context.Context) ([]ai_summarizer.TenderSummaryItem, error) + } + + storage, ok := s.aiSummarizerStorage.(allSummaries) + if !ok { + return nil, fmt.Errorf("AI summarizer storage does not support batch retrieval") + } + + results, err := storage.GetAllSummaries(ctx) + if err != nil { + s.logger.Error("Failed to get all AI summaries", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to get all AI summaries: %w", err) + } + + s.logger.Info("Retrieved all AI summaries", map[string]interface{}{ + "total_count": len(results), + }) + + return results, nil +} diff --git a/pkg/ai_summarizer/client.go b/pkg/ai_summarizer/client.go new file mode 100644 index 0000000..0cb76b5 --- /dev/null +++ b/pkg/ai_summarizer/client.go @@ -0,0 +1,125 @@ +package ai_summarizer + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "time" + + "tm/pkg/logger" +) + +// Client is the HTTP client for the AI summarization on-demand API. +type Client struct { + config *Config + httpClient *http.Client + logger logger.Logger +} + +// NewClient creates a new AI summarizer HTTP client. +func NewClient(config *Config, log logger.Logger) (*Client, error) { + if config == nil { + config = DefaultConfig() + } + if err := config.ValidateAPI(); err != nil { + return nil, fmt.Errorf("ai_summarizer: invalid config: %w", err) + } + + return &Client{ + config: config, + httpClient: &http.Client{ + Timeout: config.APITimeout, + }, + logger: log, + }, nil +} + +// FetchSummaryOnDemand calls the external AI service POST /ai/summarize endpoint +// to generate a summary on demand. It retries on transient failures up to the +// configured number of attempts. +func (c *Client) FetchSummaryOnDemand(ctx context.Context, reqBody SummarizeRequest) (*SummarizeResponse, error) { + jsonBody, err := json.Marshal(reqBody) + if err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to marshal request body: %w", err) + } + + url := c.config.APIBaseURL + "/ai/summarize" + + var lastErr error + for attempt := 0; attempt <= c.config.APIRetryCount; attempt++ { + if attempt > 0 { + c.logger.Warn("Retrying AI summarize request", map[string]interface{}{ + "notice_id": reqBody.NoticeID, + "attempt": attempt, + }) + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(c.config.APIRetryDelay): + } + } + + resp, err := c.doPost(ctx, url, jsonBody) + if err != nil { + lastErr = err + c.logger.Error("AI summarize request failed", map[string]interface{}{ + "notice_id": reqBody.NoticeID, + "attempt": attempt, + "error": err.Error(), + }) + continue + } + + return resp, nil + } + + return nil, fmt.Errorf("ai_summarizer: all %d attempts failed, last error: %w", c.config.APIRetryCount+1, lastErr) +} + +// doPost performs a single POST request and parses the response. +func (c *Client) doPost(ctx context.Context, url string, jsonBody []byte) (*SummarizeResponse, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(jsonBody)) + if err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + + c.logger.Debug("Sending AI summarize request", map[string]interface{}{ + "url": url, + }) + + httpResp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("ai_summarizer: request error: %w", err) + } + defer httpResp.Body.Close() + + bodyBytes, err := io.ReadAll(httpResp.Body) + if err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to read response body: %w", err) + } + + if httpResp.StatusCode == http.StatusNotFound { + return nil, fmt.Errorf("%w: status 404, body: %s", ErrObjectNotFound, string(bodyBytes)) + } + + if httpResp.StatusCode >= 400 { + return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + } + + var result SummarizeResponse + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode response JSON: %w", err) + } + + c.logger.Info("AI summarize request succeeded", map[string]interface{}{ + "notice_id": result.NoticeID, + "summaries_count": len(result.Summaries), + }) + + return &result, nil +} diff --git a/pkg/ai_summarizer/config.go b/pkg/ai_summarizer/config.go new file mode 100644 index 0000000..fa43f57 --- /dev/null +++ b/pkg/ai_summarizer/config.go @@ -0,0 +1,76 @@ +package ai_summarizer + +import ( + "errors" + "fmt" + "time" +) + +// Config holds the configuration for the AI Summarizer service. +// It contains settings for both the HTTP API and the MinIO storage integration. +type Config struct { + // HTTP API settings + APIBaseURL string // Base URL of the AI service, e.g. "http://ai-host:8000" + APITimeout time.Duration // Timeout for HTTP requests to the AI API + APIRetryCount int // Number of retry attempts for failed API requests + APIRetryDelay time.Duration // Delay between retry attempts + + // MinIO storage settings + MinioEndpoint string // MinIO endpoint, e.g. "ai-host:9000" + MinioAccessKey string // MinIO access key ID + MinioSecretKey string // MinIO secret access key + MinioUseSSL bool // Whether to use SSL for MinIO connection + MinioRegion string // MinIO region + MinioBucket string // MinIO bucket name (default: "opplens-documents") + MinioTimeout time.Duration // Timeout for MinIO operations +} + +// DefaultConfig returns a Config with sensible defaults. +func DefaultConfig() *Config { + return &Config{ + APITimeout: 120 * time.Second, + APIRetryCount: 2, + APIRetryDelay: 3 * time.Second, + + MinioBucket: "opplens-documents", + MinioRegion: "us-east-1", + MinioUseSSL: false, + MinioTimeout: 30 * time.Second, + } +} + +// Validate checks that all required configuration fields are present. +// Use this when both HTTP API and MinIO storage are needed. +func (c *Config) Validate() error { + if c.APIBaseURL == "" { + return errors.New("ai_summarizer: APIBaseURL is required") + } + return c.ValidateStorage() +} + +// ValidateStorage checks only the MinIO storage-related configuration fields. +// Use this when only the storage client is needed (no HTTP API). +func (c *Config) ValidateStorage() error { + if c.MinioEndpoint == "" { + return errors.New("ai_summarizer: MinioEndpoint is required") + } + if c.MinioAccessKey == "" { + return errors.New("ai_summarizer: MinioAccessKey is required") + } + if c.MinioSecretKey == "" { + return errors.New("ai_summarizer: MinioSecretKey is required") + } + if c.MinioBucket == "" { + return fmt.Errorf("ai_summarizer: MinioBucket is required") + } + return nil +} + +// ValidateAPI checks only the HTTP API-related configuration fields. +// Use this when only the HTTP client is needed (no MinIO storage). +func (c *Config) ValidateAPI() error { + if c.APIBaseURL == "" { + return errors.New("ai_summarizer: APIBaseURL is required") + } + return nil +} diff --git a/pkg/ai_summarizer/entities.go b/pkg/ai_summarizer/entities.go new file mode 100644 index 0000000..220c1ed --- /dev/null +++ b/pkg/ai_summarizer/entities.go @@ -0,0 +1,43 @@ +package ai_summarizer + +// SummarizeRequest represents the request payload sent to the AI service +// POST /ai/summarize endpoint for on-demand summarization. +type SummarizeRequest struct { + NoticeID string `json:"notice_id"` // required + DocumentURL string `json:"document_url"` // required + Title string `json:"title"` // required + Description string `json:"description"` // required + Country string `json:"country,omitempty"` // optional + EstimatedValue *float64 `json:"estimated_value,omitempty"` // optional (pointer to distinguish 0 from absent) + Currency string `json:"currency,omitempty"` // optional + SubmissionDeadline string `json:"submission_deadline,omitempty"` // optional + AuthorityName string `json:"authority_name,omitempty"` // optional +} + +// SummarizeResponse represents the response from the AI summarization API. +type SummarizeResponse struct { + NoticeID string `json:"notice_id"` + Summaries []DocumentSummary `json:"summaries"` + OverallSummary *string `json:"overall_summary"` // nil when not available +} + +// DocumentSummary represents a summary for a single document within the AI response. +type DocumentSummary struct { + DocumentName string `json:"document_name"` + DocumentType string `json:"document_type"` + Summary string `json:"summary"` + SummaryModel string `json:"summary_model"` + Error string `json:"error"` // empty string when no error +} + +// TenderJSON represents the structure of the /tender.json file +// stored in the AI service's MinIO bucket. +type TenderJSON struct { + OverallSummary *string `json:"overall_summary"` // nil if not yet generated + SummarizeStatus SummarizeStatus `json:"summarize_status"` +} + +// SummarizeStatus represents the summarization status inside tender.json. +type SummarizeStatus struct { + Done bool `json:"done"` // true when the pipeline has finished +} diff --git a/pkg/ai_summarizer/errors.go b/pkg/ai_summarizer/errors.go new file mode 100644 index 0000000..0570212 --- /dev/null +++ b/pkg/ai_summarizer/errors.go @@ -0,0 +1,24 @@ +package ai_summarizer + +import "errors" + +// Common sentinel errors for the AI summarizer package. +var ( + // ErrSummaryNotReady is returned when the tender.json exists but + // summarize_status.done is still false. + ErrSummaryNotReady = errors.New("ai_summarizer: summary not ready yet") + + // ErrNoOverallSummary is returned when the pipeline is done but + // overall_summary is empty or null. + ErrNoOverallSummary = errors.New("ai_summarizer: overall_summary is empty") + + // ErrObjectNotFound is returned when the tender.json object does not + // exist in the MinIO bucket (HTTP 404 / NoSuchKey). + ErrObjectNotFound = errors.New("ai_summarizer: tender.json not found in storage") + + // ErrBucketNotFound is returned when the MinIO bucket does not exist. + ErrBucketNotFound = errors.New("ai_summarizer: bucket not found") + + // ErrAPINonSuccess is returned when the AI API returns a non-2xx status code. + ErrAPINonSuccess = errors.New("ai_summarizer: API returned non-success status") +) diff --git a/pkg/ai_summarizer/storage.go b/pkg/ai_summarizer/storage.go new file mode 100644 index 0000000..fa79531 --- /dev/null +++ b/pkg/ai_summarizer/storage.go @@ -0,0 +1,235 @@ +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 /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: /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 +}