diff --git a/cmd/web/bootstrap/bootstrap.go b/cmd/web/bootstrap/bootstrap.go index 401d725..90285bd 100644 --- a/cmd/web/bootstrap/bootstrap.go +++ b/cmd/web/bootstrap/bootstrap.go @@ -290,23 +290,25 @@ func InitFileStore(mongoManager *mongo.ConnectionManager, log logger.Logger) (*f } // InitAISummarizerClient initializes the AI summarizer HTTP client for on-demand summarization -func InitAISummarizerClient(conf AISummarizerConfig, log logger.Logger) *ai_summarizer.Client { +func InitAISummarizerClient(conf AISummarizerConfig, mongoManager *mongo.ConnectionManager, 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 } + translationCounter := mongo.NewCounter(mongoManager) 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, + 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, + OnSuccessfulTranslation: mongo.AITranslationSuccessCallback(translationCounter), } client, err := ai_summarizer.NewClient(cfg, log) diff --git a/cmd/web/main.go b/cmd/web/main.go index de5256f..f7ac0ac 100644 --- a/cmd/web/main.go +++ b/cmd/web/main.go @@ -168,7 +168,7 @@ func main() { var aiSummarizerClient tender.AISummarizerClient var aiSummarizerStorage tender.AISummarizerStorage - if c := bootstrap.InitAISummarizerClient(conf.AISummarizer, logger); c != nil { + if c := bootstrap.InitAISummarizerClient(conf.AISummarizer, mongoManager, logger); c != nil { aiSummarizerClient = c } if s := bootstrap.InitAISummarizerStorage(conf.AISummarizer, logger); s != nil { diff --git a/cmd/web/router/routes.go b/cmd/web/router/routes.go index 44439d2..f76826a 100644 --- a/cmd/web/router/routes.go +++ b/cmd/web/router/routes.go @@ -222,6 +222,7 @@ func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler dashboardGP.GET("/notice-types", dashboardHandler.NoticeTypes) dashboardGP.GET("/closing-soon", dashboardHandler.ClosingSoon) dashboardGP.GET("/recent", dashboardHandler.Recent) + dashboardGP.GET("/statistics", dashboardHandler.Statistics) } } diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 1bb88df..eda46b9 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -196,6 +196,7 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger // Initialize translation backfill worker for configured languages if aiClient != nil { targetLanguages := parseTranslationLanguages(config.AISummarizer) + translationSuccessCounter := mongo.NewCounter(mongoManager) scheduler.AddJob(schedule.Job{ Name: "Tender Translation Worker Job", Func: func() { @@ -205,6 +206,7 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger tenderRepo, aiClient, aiStorage, + translationSuccessCounter, targetLanguages, config.Worker.TranslationBatchSize, ) @@ -247,17 +249,19 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger } // InitAISummarizerClient initializes the AI summarizer HTTP client used by worker jobs. -func InitAISummarizerClient(conf AISummarizerConfig, log logger.Logger) *ai_summarizer.Client { +func InitAISummarizerClient(conf AISummarizerConfig, mongoManager *mongo.ConnectionManager, log logger.Logger) *ai_summarizer.Client { if conf.APIBaseURL == "" { log.Warn("AI Summarizer API base URL not configured, translation worker will be unavailable", map[string]interface{}{}) return nil } + translationCounter := mongo.NewCounter(mongoManager) cfg := &ai_summarizer.Config{ - APIBaseURL: conf.APIBaseURL, - APITimeout: conf.APITimeout, - APIRetryCount: conf.APIRetryCount, - APIRetryDelay: conf.APIRetryDelay, + APIBaseURL: conf.APIBaseURL, + APITimeout: conf.APITimeout, + APIRetryCount: conf.APIRetryCount, + APIRetryDelay: conf.APIRetryDelay, + OnSuccessfulTranslation: mongo.AITranslationSuccessCallback(translationCounter), } client, err := ai_summarizer.NewClient(cfg, log) diff --git a/cmd/worker/main.go b/cmd/worker/main.go index e312f1b..a401a5e 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -44,7 +44,7 @@ func main() { } // Initialize AI summarizer client (translation + document summarization workers) - aiSummarizerClient := bootstrap.InitAISummarizerClient(config.AISummarizer, appLogger) + aiSummarizerClient := bootstrap.InitAISummarizerClient(config.AISummarizer, mongoManager, appLogger) aiSummarizerStorage := bootstrap.InitAISummarizerStorage(config.AISummarizer, appLogger) // Initialize Worker diff --git a/cmd/worker/workers/translation.go b/cmd/worker/workers/translation.go index dc4fdcf..bb8e6a5 100644 --- a/cmd/worker/workers/translation.go +++ b/cmd/worker/workers/translation.go @@ -13,13 +13,14 @@ import ( // TranslationWorker backfills missing tender translations using the AI pipeline // and on-demand translate API. Results are persisted by the AI service in MinIO only. type TranslationWorker struct { - Mongo *mongo.ConnectionManager - Logger logger.Logger - TenderRepo tender.TenderRepository - AIClient *ai_summarizer.Client - AIStorage *ai_summarizer.StorageClient - TargetLanguages []string - BatchSize int + Mongo *mongo.ConnectionManager + Logger logger.Logger + TenderRepo tender.TenderRepository + AIClient *ai_summarizer.Client + AIStorage *ai_summarizer.StorageClient + TranslationSuccessCounter *mongo.Counter + TargetLanguages []string + BatchSize int } // NewTranslationWorker creates a translation worker for the given target languages. @@ -29,6 +30,7 @@ func NewTranslationWorker( tenderRepo tender.TenderRepository, aiClient *ai_summarizer.Client, aiStorage *ai_summarizer.StorageClient, + translationSuccessCounter *mongo.Counter, targetLanguages []string, batchSize int, ) *TranslationWorker { @@ -53,13 +55,14 @@ func NewTranslationWorker( } return &TranslationWorker{ - Mongo: mongo, - Logger: logger, - TenderRepo: tenderRepo, - AIClient: aiClient, - AIStorage: aiStorage, - TargetLanguages: langs, - BatchSize: batchSize, + Mongo: mongo, + Logger: logger, + TenderRepo: tenderRepo, + AIClient: aiClient, + AIStorage: aiStorage, + TranslationSuccessCounter: translationSuccessCounter, + TargetLanguages: langs, + BatchSize: batchSize, } } @@ -77,6 +80,8 @@ func (w *TranslationWorker) Run() { "batch_size": w.BatchSize, }) + startCount := w.getSuccessfulTranslationCount(mongo.AITranslationSuccessCounterKeyDailyJob) + if _, err := w.AIClient.TriggerPipelineTranslate(context.Background(), w.TargetLanguages); err != nil { w.Logger.Warn("Failed to trigger AI pipeline translate (continuing with per-tender sync)", map[string]interface{}{ "target_languages": w.TargetLanguages, @@ -88,11 +93,36 @@ func (w *TranslationWorker) Run() { w.runForLanguage(language) } + endCount := w.getSuccessfulTranslationCount(mongo.AITranslationSuccessCounterKeyDailyJob) + totalCount := w.getSuccessfulTranslationCount(mongo.AITranslationSuccessCounterKey) + runCount := endCount - startCount + if runCount < 0 { + runCount = 0 + } + w.Logger.Info("Translation worker completed", map[string]interface{}{ - "target_languages": w.TargetLanguages, + "target_languages": w.TargetLanguages, + "successful_ai_translation_requests_run": runCount, + "successful_ai_translation_requests_daily_job": endCount, + "successful_ai_translation_requests_total": totalCount, }) } +func (w *TranslationWorker) getSuccessfulTranslationCount(key string) int64 { + if w.TranslationSuccessCounter == nil { + return 0 + } + count, err := w.TranslationSuccessCounter.Get(context.Background(), key) + if err != nil { + w.Logger.Warn("Failed to read AI translation success counter", map[string]interface{}{ + "counter_key": key, + "error": err.Error(), + }) + return 0 + } + return count +} + func (w *TranslationWorker) runForLanguage(language string) { skip := 0 for { @@ -210,6 +240,7 @@ func (w *TranslationWorker) resolveTranslation(t *tender.Tender, language string Title: t.Title, Description: t.Description, Language: language, + RequestSource: ai_summarizer.TranslationRequestSourceDailyJob, }) if apiErr != nil { return "", "", "api", apiErr diff --git a/internal/dashboard/entity.go b/internal/dashboard/entity.go index 47e7d18..670107b 100644 --- a/internal/dashboard/entity.go +++ b/internal/dashboard/entity.go @@ -85,3 +85,24 @@ type RecentItem struct { CreatedAt int64 `json:"created_at"` PublicationDate int64 `json:"publication_date"` } + +// StatisticsReportResponse powers the statistics and charts report. +type StatisticsReportResponse struct { + Days int `json:"days"` + GeneratedAt int64 `json:"generated_at"` + Daily StatisticsDailySeries `json:"daily"` + Totals StatisticsLifetimeTotals `json:"totals"` +} + +// StatisticsDailySeries contains per-day chart series. +type StatisticsDailySeries struct { + ScrapedTED []TrendPoint `json:"scraped_ted"` + ScrapedDocuments []TrendPoint `json:"scraped_documents"` + TranslatedNotices []TrendPoint `json:"translated_notices"` +} + +// StatisticsLifetimeTotals contains all-time totals. +type StatisticsLifetimeTotals struct { + ScrapedDocuments int64 `json:"scraped_documents"` + TranslatedTenders int64 `json:"translated_tenders"` +} diff --git a/internal/dashboard/form.go b/internal/dashboard/form.go index 7587a4e..753113e 100644 --- a/internal/dashboard/form.go +++ b/internal/dashboard/form.go @@ -27,3 +27,8 @@ type RecentQuery struct { Limit int `query:"limit"` Cursor string `query:"cursor"` } + +// StatisticsQuery binds query params for GET /dashboard/statistics. +type StatisticsQuery struct { + Days int `query:"days"` +} diff --git a/internal/dashboard/handler.go b/internal/dashboard/handler.go index 642790b..6a1b9ea 100644 --- a/internal/dashboard/handler.go +++ b/internal/dashboard/handler.go @@ -161,6 +161,30 @@ func (h *Handler) Recent(c echo.Context) error { return response.Success(c, out, "Recent tenders retrieved successfully") } +// Statistics returns scraping and translation statistics for charts. +// @Summary Dashboard statistics report +// @Description Daily and lifetime statistics for TED scraping, document scraping, and AI translation +// @Tags Admin-Dashboard +// @Produce json +// @Param days query int false "Days back for daily series (default 14, max 90)" +// @Success 200 {object} response.APIResponse{data=StatisticsReportResponse} +// @Failure 500 {object} response.APIResponse +// @Router /admin/v1/dashboard/statistics [get] +// @Security BearerAuth +func (h *Handler) Statistics(c echo.Context) error { + query := StatisticsQuery{ + Days: parseIntQuery(c, "days", 0), + } + + out, err := h.service.Statistics(c.Request().Context(), query) + if err != nil { + return response.InternalServerError(c, "Failed to load dashboard statistics") + } + + setPrivateCache(c, 60) + return response.Success(c, out, "Dashboard statistics retrieved successfully") +} + func parseIntQuery(c echo.Context, name string, fallback int) int { raw := c.QueryParam(name) if raw == "" { diff --git a/internal/dashboard/repository.go b/internal/dashboard/repository.go index dcf7bfc..7587af3 100644 --- a/internal/dashboard/repository.go +++ b/internal/dashboard/repository.go @@ -23,18 +23,23 @@ type Repository interface { NoticeTypes(ctx context.Context) (*NoticeTypesResponse, error) ClosingSoon(ctx context.Context, limit int, windowSec int64) ([]ClosingSoonItem, error) Recent(ctx context.Context, limit int, cursor string) ([]RecentItem, *string, error) + Statistics(ctx context.Context, days int, startUnix int64) (*StatisticsReportResponse, error) } type repository struct { - ormRepo orm.Repository[tender.Tender] - logger logger.Logger + ormRepo orm.Repository[tender.Tender] + mongoManager *orm.ConnectionManager + metricsCounter *orm.Counter + logger logger.Logger } // NewRepository creates a dashboard repository backed by the tenders collection. func NewRepository(mongoManager *orm.ConnectionManager, log logger.Logger) Repository { return &repository{ - ormRepo: orm.NewRepository[tender.Tender](mongoManager.GetCollection(tenderCollectionName()), log), - logger: log, + ormRepo: orm.NewRepository[tender.Tender](mongoManager.GetCollection(tenderCollectionName()), log), + mongoManager: mongoManager, + metricsCounter: orm.NewCounter(mongoManager), + logger: log, } } diff --git a/internal/dashboard/service.go b/internal/dashboard/service.go index e9841a8..73b5ecd 100644 --- a/internal/dashboard/service.go +++ b/internal/dashboard/service.go @@ -25,6 +25,7 @@ type Service interface { NoticeTypes(ctx context.Context) (*NoticeTypesResponse, error) ClosingSoon(ctx context.Context, query ClosingSoonQuery) (*ClosingSoonResponse, error) Recent(ctx context.Context, query RecentQuery) (*RecentResponse, error) + Statistics(ctx context.Context, query StatisticsQuery) (*StatisticsReportResponse, error) } type service struct { @@ -157,6 +158,25 @@ func (s *service) Recent(ctx context.Context, query RecentQuery) (*RecentRespons }, nil } +func (s *service) Statistics(ctx context.Context, query StatisticsQuery) (*StatisticsReportResponse, error) { + days := trendDays(query.Days) + startUnix := trendStartUnix(days) + + s.logger.Info("Fetching dashboard statistics report", map[string]interface{}{ + "days": days, + }) + + out, err := s.repo.Statistics(ctx, days, startUnix) + if err != nil { + s.logger.Error("Failed to fetch dashboard statistics report", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("dashboard statistics: %w", err) + } + + return out, nil +} + func closingWindowHours(hours int) int { if hours <= 0 { return defaultClosingWindowHours diff --git a/internal/dashboard/statistics_repository.go b/internal/dashboard/statistics_repository.go new file mode 100644 index 0000000..f1e0ca6 --- /dev/null +++ b/internal/dashboard/statistics_repository.go @@ -0,0 +1,189 @@ +package dashboard + +import ( + "context" + "fmt" + "time" + + "tm/internal/notice" + orm "tm/pkg/mongo" + + "go.mongodb.org/mongo-driver/v2/bson" + mongodriver "go.mongodb.org/mongo-driver/v2/mongo" +) + +const noticesCollectionName = "notices" + +func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) (*StatisticsReportResponse, error) { + now := time.Now().UTC() + endDay := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC) + startDay := endDay.AddDate(0, 0, -(days - 1)) + + scrapedTED, err := r.dailyCountByTimestamp(ctx, noticesCollectionName, bson.M{ + "source": notice.TenderSourceTEDScraper, + "processing_metadata.scraped_at": bson.M{"$gte": startUnix, "$gt": 0}, + }, "processing_metadata.scraped_at") + if err != nil { + return nil, fmt.Errorf("scraped ted per day: %w", err) + } + + scrapedDocuments, err := r.scrapedDocumentsPerDay(ctx, startUnix) + if err != nil { + return nil, fmt.Errorf("scraped documents per day: %w", err) + } + + translatedNotices, err := r.translatedNoticesPerDay(ctx, startDay, endDay) + if err != nil { + return nil, fmt.Errorf("translated notices per day: %w", err) + } + + totalDocuments, err := r.totalScrapedDocuments(ctx) + if err != nil { + return nil, fmt.Errorf("total scraped documents: %w", err) + } + + totalTranslated, err := r.metricsCounter.Get(ctx, orm.AITranslationSuccessCounterKey) + if err != nil { + return nil, fmt.Errorf("total translated tenders: %w", err) + } + + return &StatisticsReportResponse{ + Days: days, + GeneratedAt: now.Unix(), + Daily: StatisticsDailySeries{ + ScrapedTED: fillTrendSeries(days, scrapedTED), + ScrapedDocuments: fillTrendSeries(days, scrapedDocuments), + TranslatedNotices: fillTrendSeries(days, translatedNotices), + }, + Totals: StatisticsLifetimeTotals{ + ScrapedDocuments: totalDocuments, + TranslatedTenders: totalTranslated, + }, + }, nil +} + +func (r *repository) dailyCountByTimestamp(ctx context.Context, collection string, match bson.M, timestampField string) (map[string]int64, error) { + pipeline := mongodriver.Pipeline{ + {{Key: "$match", Value: match}}, + {{Key: "$addFields", Value: bson.M{ + "metric_ts": normalizeTimestampExpr(timestampField), + }}}, + {{Key: "$group", Value: bson.M{ + "_id": bson.M{ + "$dateToString": bson.M{ + "format": "%Y-%m-%d", + "date": bson.M{"$toDate": bson.M{"$multiply": bson.A{"$metric_ts", 1000}}}, + "timezone": "UTC", + }, + }, + "count": bson.M{"$sum": 1}, + }}}, + } + + cursor, err := r.mongoManager.GetCollection(collection).Aggregate(ctx, pipeline) + if err != nil { + return nil, err + } + defer cursor.Close(ctx) + + return decodeDailyCounts(ctx, cursor) +} + +func (r *repository) scrapedDocumentsPerDay(ctx context.Context, startUnix int64) (map[string]int64, error) { + pipeline := mongodriver.Pipeline{ + {{Key: "$match", Value: bson.M{ + "processing_metadata.documents_scraped": true, + "scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}}, + }}}, + {{Key: "$unwind", Value: "$scraped_documents"}}, + {{Key: "$addFields", Value: bson.M{ + "metric_ts": bson.M{ + "$cond": bson.A{ + bson.M{"$gt": bson.A{"$scraped_documents.scraped_at", 0}}, + "$scraped_documents.scraped_at", + "$processing_metadata.documents_scraped_at", + }, + }, + }}}, + {{Key: "$match", Value: bson.M{ + "metric_ts": bson.M{"$gte": startUnix, "$gt": 0}, + }}}, + {{Key: "$addFields", Value: bson.M{ + "metric_ts": normalizeTimestampExpr("metric_ts"), + }}}, + {{Key: "$group", Value: bson.M{ + "_id": bson.M{ + "$dateToString": bson.M{ + "format": "%Y-%m-%d", + "date": bson.M{"$toDate": bson.M{"$multiply": bson.A{"$metric_ts", 1000}}}, + "timezone": "UTC", + }, + }, + "count": bson.M{"$sum": 1}, + }}}, + } + + cursor, err := r.mongoManager.GetCollection(tenderCollectionName()).Aggregate(ctx, pipeline) + if err != nil { + return nil, err + } + defer cursor.Close(ctx) + + return decodeDailyCounts(ctx, cursor) +} + +func (r *repository) translatedNoticesPerDay(ctx context.Context, startDay, endDay time.Time) (map[string]int64, error) { + return r.metricsCounter.GetDailyCounts(ctx, startDay, endDay) +} + +func (r *repository) totalScrapedDocuments(ctx context.Context) (int64, error) { + pipeline := mongodriver.Pipeline{ + {{Key: "$match", Value: bson.M{ + "processing_metadata.documents_scraped": true, + "scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}}, + }}}, + {{Key: "$project", Value: bson.M{ + "doc_count": bson.M{"$size": "$scraped_documents"}, + }}}, + {{Key: "$group", Value: bson.M{ + "_id": nil, + "total": bson.M{"$sum": "$doc_count"}, + }}}, + } + + cursor, err := r.mongoManager.GetCollection(tenderCollectionName()).Aggregate(ctx, pipeline) + if err != nil { + return 0, err + } + defer cursor.Close(ctx) + + if !cursor.Next(ctx) { + return 0, nil + } + + var row bson.M + if err := cursor.Decode(&row); err != nil { + return 0, err + } + + return aggregateCount(row, "total"), nil +} + +func decodeDailyCounts(ctx context.Context, cursor *mongodriver.Cursor) (map[string]int64, error) { + counts := make(map[string]int64) + for cursor.Next(ctx) { + var row bson.M + if err := cursor.Decode(&row); err != nil { + return nil, err + } + date, ok := row["_id"].(string) + if !ok || date == "" { + continue + } + counts[date] = aggregateCount(row, "count") + } + if err := cursor.Err(); err != nil { + return nil, err + } + return counts, nil +} diff --git a/internal/notice/repository.go b/internal/notice/repository.go index 8e9e2b2..2d10bcf 100644 --- a/internal/notice/repository.go +++ b/internal/notice/repository.go @@ -43,7 +43,7 @@ func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Re *orm.NewIndex("contract_notice_id_idx", bson.D{{Key: "contract_notice_id", Value: 1}}), // One row per TED ContractNoticeID when present (parallel scrape races). *orm.CreateUniqueIndex("contract_notice_id_unique", bson.D{{Key: "contract_notice_id", Value: 1}}). - WithPartialFilterExpression(bson.M{"contract_notice_id": bson.M{"$type": "string", "$ne": ""}}), + WithPartialFilterExpression(orm.NonEmptyStringPartialFilter("contract_notice_id")), } // Create indexes diff --git a/internal/tender/repository.go b/internal/tender/repository.go index b8a12a3..3b03226 100644 --- a/internal/tender/repository.go +++ b/internal/tender/repository.go @@ -94,16 +94,16 @@ func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Te *orm.NewIndex("contract_notice_id_idx", bson.D{{Key: "contract_notice_id", Value: 1}}), // One tender row per TED ContractNoticeID when present (parallel ingest / lookup gaps). *orm.CreateUniqueIndex("contract_notice_id_unique", bson.D{{Key: "contract_notice_id", Value: 1}}). - WithPartialFilterExpression(bson.M{"contract_notice_id": bson.M{"$type": "string", "$ne": ""}}), + WithPartialFilterExpression(orm.NonEmptyStringPartialFilter("contract_notice_id")), *orm.NewIndex("contract_folder_id_idx", bson.D{{Key: "contract_folder_id", Value: 1}}), // One tender row per TED publication / procedure when ids are present (prevents parallel worker races). *orm.CreateUniqueIndex("notice_publication_id_unique", bson.D{{Key: "notice_publication_id", Value: 1}}). - WithPartialFilterExpression(bson.M{"notice_publication_id": bson.M{"$type": "string", "$ne": ""}}), + WithPartialFilterExpression(orm.NonEmptyStringPartialFilter("notice_publication_id")), *orm.CreateUniqueIndex("contract_folder_id_unique", bson.D{{Key: "contract_folder_id", Value: 1}}). - WithPartialFilterExpression(bson.M{"contract_folder_id": bson.M{"$type": "string", "$ne": ""}}), + WithPartialFilterExpression(orm.NonEmptyStringPartialFilter("contract_folder_id")), // One tender per UBL procurement project (merges per-lot notices that share ProcurementProject/ID). *orm.CreateUniqueIndex("procurement_project_id_unique", bson.D{{Key: "procurement_project_id", Value: 1}}). - WithPartialFilterExpression(bson.M{"procurement_project_id": bson.M{"$type": "string", "$ne": ""}}), + WithPartialFilterExpression(orm.NonEmptyStringPartialFilter("procurement_project_id")), *orm.NewIndex("status_idx", bson.D{{Key: "status", Value: 1}}), *orm.NewIndex("source_idx", bson.D{{Key: "source", Value: 1}}), *orm.NewIndex("country_code_idx", bson.D{{Key: "country_code", Value: 1}}), diff --git a/internal/tender/service.go b/internal/tender/service.go index 8ea38e6..a3e7c02 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -1476,7 +1476,7 @@ func (s *tenderService) TriggerAITranslate(ctx context.Context, id, language str return nil, fmt.Errorf("tender has no contract folder ID") } - title, description, err := s.resolveTranslation(ctx, t, targetLanguage) + title, description, fromAPI, err := s.resolveTranslation(ctx, t, targetLanguage, ai_summarizer.TranslationRequestSourceManualTrigger) if err != nil { s.logger.Error("AI translation failed", map[string]interface{}{ "tender_id": id, @@ -1487,6 +1487,14 @@ func (s *tenderService) TriggerAITranslate(ctx context.Context, id, language str return nil, fmt.Errorf("AI translation request failed: %w", err) } + if fromAPI { + s.logger.Info("Manual AI translation request succeeded", map[string]interface{}{ + "tender_id": id, + "notice_id": t.NoticePublicationID, + "language": targetLanguage, + }) + } + return s.buildAITranslateResponse(t, targetLanguage, title, description), nil } @@ -1500,11 +1508,11 @@ func (s *tenderService) buildAITranslateResponse(t *Tender, language, title, des } } -func (s *tenderService) resolveTranslation(ctx context.Context, t *Tender, language string) (string, string, error) { +func (s *tenderService) resolveTranslation(ctx context.Context, t *Tender, language, requestSource string) (string, string, bool, error) { if s.aiSummarizerStorage != nil { stored, err := s.aiSummarizerStorage.GetTranslationFromStorage(ctx, t.ContractFolderID, t.NoticePublicationID, language) if err == nil { - return stored.Title, stored.Description, nil + return stored.Title, stored.Description, false, nil } if !errors.Is(err, ai_summarizer.ErrTranslationNotReady) && !errors.Is(err, ai_summarizer.ErrObjectNotFound) && @@ -1523,11 +1531,12 @@ func (s *tenderService) resolveTranslation(ctx context.Context, t *Tender, langu Title: t.Title, Description: t.Description, Language: language, + RequestSource: requestSource, }) if err != nil { - return "", "", err + return "", "", false, err } - return resp.TranslatedTitle, resp.TranslatedDescription, nil + return resp.TranslatedTitle, resp.TranslatedDescription, true, nil } func (s *tenderService) pickResponseLanguage(language *string) string { diff --git a/pkg/ai_summarizer/client.go b/pkg/ai_summarizer/client.go index 590e715..27ed873 100644 --- a/pkg/ai_summarizer/client.go +++ b/pkg/ai_summarizer/client.go @@ -107,7 +107,7 @@ func (c *Client) FetchTranslationOnDemand(ctx context.Context, reqBody Translate } } - resp, err := c.doTranslatePost(ctx, url, jsonBody) + resp, err := c.doTranslatePost(ctx, url, jsonBody, reqBody.RequestSource) if err != nil { lastErr = err c.logger.Error("AI translate request failed", map[string]interface{}{ @@ -216,7 +216,7 @@ func (c *Client) doPost(ctx context.Context, url string, jsonBody []byte) (*Summ } // doTranslatePost performs a single translation POST request and parses the response. -func (c *Client) doTranslatePost(ctx context.Context, url string, jsonBody []byte) (*TranslateResponse, error) { +func (c *Client) doTranslatePost(ctx context.Context, url string, jsonBody []byte, requestSource string) (*TranslateResponse, error) { req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(jsonBody)) if err != nil { return nil, fmt.Errorf("ai_summarizer: failed to create translation request: %w", err) @@ -253,6 +253,17 @@ func (c *Client) doTranslatePost(ctx context.Context, url string, jsonBody []byt "language": result.Language, }) + if c.config.OnSuccessfulTranslation != nil { + if err := c.config.OnSuccessfulTranslation(ctx, requestSource); err != nil { + c.logger.Warn("Failed to increment AI translation success counter", map[string]interface{}{ + "notice_publication_id": result.NoticePublicationID, + "language": result.Language, + "request_source": requestSource, + "error": err.Error(), + }) + } + } + return result, nil } diff --git a/pkg/ai_summarizer/config.go b/pkg/ai_summarizer/config.go index fa43f57..6b42e2d 100644 --- a/pkg/ai_summarizer/config.go +++ b/pkg/ai_summarizer/config.go @@ -1,6 +1,7 @@ package ai_summarizer import ( + "context" "errors" "fmt" "time" @@ -15,6 +16,10 @@ type Config struct { APIRetryCount int // Number of retry attempts for failed API requests APIRetryDelay time.Duration // Delay between retry attempts + // OnSuccessfulTranslation is called after a successful POST /ai/translate request. + // source identifies the caller (daily_job or manual_trigger). + OnSuccessfulTranslation func(ctx context.Context, source string) error + // MinIO storage settings MinioEndpoint string // MinIO endpoint, e.g. "ai-host:9000" MinioAccessKey string // MinIO access key ID diff --git a/pkg/ai_summarizer/entities.go b/pkg/ai_summarizer/entities.go index 56cd7f3..9b04ee0 100644 --- a/pkg/ai_summarizer/entities.go +++ b/pkg/ai_summarizer/entities.go @@ -109,8 +109,14 @@ type TranslateRequest struct { Title string `json:"title"` Description string `json:"description"` Language string `json:"language"` + RequestSource string `json:"-"` // internal: daily_job | manual_trigger } +const ( + TranslationRequestSourceDailyJob = "daily_job" + TranslationRequestSourceManualTrigger = "manual_trigger" +) + // TranslateResponse represents the translation payload from POST /ai/translate. type TranslateResponse struct { ContractFolderID string `json:"contract_folder_id"` diff --git a/pkg/mongo/counter.go b/pkg/mongo/counter.go new file mode 100644 index 0000000..3a5dc69 --- /dev/null +++ b/pkg/mongo/counter.go @@ -0,0 +1,130 @@ +package mongo + +import ( + "context" + "errors" + "fmt" + "time" + + "go.mongodb.org/mongo-driver/v2/bson" + mongodriver "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" +) + +const ( + metricsCountersCollection = "metrics_counters" + AITranslationSuccessCounterKey = "ai_translation_successful_requests" + AITranslationSuccessCounterKeyDailyJob = "ai_translation_successful_requests_daily_job" + AITranslationSuccessCounterKeyManualTrigger = "ai_translation_successful_requests_manual_trigger" +) + +// Counter provides atomic increment/read operations for named metrics stored in MongoDB. +type Counter struct { + mongo *ConnectionManager +} + +// NewCounter creates a counter backed by the metrics_counters collection. +func NewCounter(mongo *ConnectionManager) *Counter { + return &Counter{mongo: mongo} +} + +// Increment atomically increments the counter identified by key and returns the new value. +func (c *Counter) Increment(ctx context.Context, key string) (int64, error) { + if key == "" { + return 0, fmt.Errorf("mongo counter: key is required") + } + + var res struct { + Count int64 `bson:"count"` + } + err := c.mongo.GetCollection(metricsCountersCollection).FindOneAndUpdate( + ctx, + bson.M{"_id": key}, + bson.M{ + "$inc": bson.M{"count": 1}, + "$set": bson.M{"updated_at": time.Now().Unix()}, + "$setOnInsert": bson.M{"_id": key}, + }, + options.FindOneAndUpdate().SetUpsert(true).SetReturnDocument(options.After), + ).Decode(&res) + if err != nil { + return 0, fmt.Errorf("mongo counter increment %q: %w", key, err) + } + + return res.Count, nil +} + +// Get returns the current value of the counter identified by key (0 when missing). +func (c *Counter) Get(ctx context.Context, key string) (int64, error) { + if key == "" { + return 0, fmt.Errorf("mongo counter: key is required") + } + + var doc struct { + Count int64 `bson:"count"` + } + err := c.mongo.GetCollection(metricsCountersCollection).FindOne(ctx, bson.M{"_id": key}).Decode(&doc) + if err != nil { + if errors.Is(err, mongodriver.ErrNoDocuments) { + return 0, nil + } + return 0, fmt.Errorf("mongo counter get %q: %w", key, err) + } + + return doc.Count, nil +} + +// AITranslationSuccessCallback returns a callback that increments successful AI translation counters. +func AITranslationSuccessCallback(counter *Counter) func(ctx context.Context, source string) error { + return func(ctx context.Context, source string) error { + if _, err := counter.Increment(ctx, AITranslationSuccessCounterKey); err != nil { + return err + } + if _, err := counter.Increment(ctx, AITranslationSuccessDailyCounterKey(time.Now().UTC())); err != nil { + return err + } + if key := AITranslationSuccessCounterKeyForSource(source); key != "" { + _, err := counter.Increment(ctx, key) + return err + } + return nil + } +} + +// AITranslationSuccessDailyCounterKey returns the metrics key for successful translations on a given UTC day. +func AITranslationSuccessDailyCounterKey(day time.Time) string { + return AITranslationSuccessCounterKey + "_day:" + day.UTC().Format("2006-01-02") +} + +// GetDailyCounts returns counter values for each UTC day from start through end inclusive. +func (c *Counter) GetDailyCounts(ctx context.Context, startDay, endDay time.Time) (map[string]int64, error) { + startDay = time.Date(startDay.Year(), startDay.Month(), startDay.Day(), 0, 0, 0, 0, time.UTC) + endDay = time.Date(endDay.Year(), endDay.Month(), endDay.Day(), 0, 0, 0, 0, time.UTC) + if endDay.Before(startDay) { + return map[string]int64{}, nil + } + + counts := make(map[string]int64) + for day := startDay; !day.After(endDay); day = day.AddDate(0, 0, 1) { + date := day.Format("2006-01-02") + value, err := c.Get(ctx, AITranslationSuccessDailyCounterKey(day)) + if err != nil { + return nil, err + } + counts[date] = value + } + + return counts, nil +} + +// AITranslationSuccessCounterKeyForSource maps a request source to its metrics counter key. +func AITranslationSuccessCounterKeyForSource(source string) string { + switch source { + case "daily_job": + return AITranslationSuccessCounterKeyDailyJob + case "manual_trigger": + return AITranslationSuccessCounterKeyManualTrigger + default: + return "" + } +} \ No newline at end of file diff --git a/pkg/mongo/utils.go b/pkg/mongo/utils.go index d52a62a..f57b170 100644 --- a/pkg/mongo/utils.go +++ b/pkg/mongo/utils.go @@ -17,7 +17,7 @@ func NewIndex(name string, keys bson.D) *Index { return &Index{ Name: name, Keys: keys, - Options: options.Index(), + Options: options.Index().SetName(name), } } @@ -39,6 +39,17 @@ func (i *Index) WithExpireAfterSeconds(seconds int32) *Index { return i } +// NonEmptyStringPartialFilter returns a partial-index filter for non-empty string field values. +// MongoDB partial indexes do not support $ne; use $gt "" instead. +func NonEmptyStringPartialFilter(field string) bson.M { + return bson.M{ + field: bson.M{ + "$type": "string", + "$gt": "", + }, + } +} + // WithPartialFilterExpression sets the partial filter expression for the index func (i *Index) WithPartialFilterExpression(filter bson.M) *Index { i.Options.SetPartialFilterExpression(filter) diff --git a/ted/calendar.go b/ted/calendar.go index 0015204..ffc4d61 100644 --- a/ted/calendar.go +++ b/ted/calendar.go @@ -2,58 +2,183 @@ package ted import ( "bytes" + "context" "encoding/csv" + "encoding/json" "fmt" "io" "net/http" "strings" + "time" ) const ( - calendarUrl = "%s/en/release-calendar/-/download/file/CSV/%d" + calendarURL = "%s/en/release-calendar/-/download/file/CSV/%d" + searchAPIURL = "https://api.ted.europa.eu/v3/notices/search" ) -// Get OJS Number -func GetOJS(baseURL string, year int, date string, client *http.Client) (string, error) { - var ( - ojs string - ) - url := fmt.Sprintf(calendarUrl, baseURL, year) +type searchAPIRequest struct { + Query string `json:"query"` + Fields []string `json:"fields"` + Limit int `json:"limit"` +} - resp, err := client.Get(url) +type searchAPIResponse struct { + Notices []struct { + OJSNumber string `json:"ojs-number"` + } `json:"notices"` +} + +// GetOJS resolves the OJS package id for a publication date. +// It tries the TED release calendar CSV first, then falls back to the TED Search API +// when the calendar download is blocked (e.g. AWS WAF HTTP 202). +func GetOJS(ctx context.Context, baseURL string, year int, date, userAgent string, client *http.Client) (string, error) { + if ojs, err := getOJSFromCalendarCSV(ctx, baseURL, year, date, userAgent, client); err == nil { + return ojs, nil + } else if !isCalendarUnavailable(err) { + return "", err + } + + return getOJSFromSearchAPIAt(ctx, date, userAgent, client, searchAPIURL) +} + +func getOJSFromCalendarCSV(ctx context.Context, baseURL string, year int, date, userAgent string, client *http.Client) (string, error) { + url := fmt.Sprintf(calendarURL, baseURL, year) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) if err != nil { - return ojs, err + return "", fmt.Errorf("create calendar request: %w", err) + } + if strings.TrimSpace(userAgent) != "" { + req.Header.Set("User-Agent", userAgent) + } + + resp, err := client.Do(req) + if err != nil { + return "", err } defer resp.Body.Close() - // read CSV body, err := io.ReadAll(resp.Body) if err != nil { - return ojs, err + return "", err + } + + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("calendar HTTP error: %d %s (body: %s)", resp.StatusCode, resp.Status, previewBody(body)) } reader := csv.NewReader(bytes.NewReader(body)) records, err := reader.ReadAll() if err != nil { - return ojs, err + return "", fmt.Errorf("parse calendar CSV: %w", err) } if len(records) < 2 { - return ojs, fmt.Errorf("ojs calendar is empty") + return "", fmt.Errorf("ojs calendar is empty (rows=%d, body=%s)", len(records), previewBody(body)) } + targetDate := strings.TrimSpace(date) for _, row := range records[1:] { + if len(row) < 2 { + continue + } rowDate := strings.TrimSpace(row[1]) - if rowDate == strings.TrimSpace(date) { - // Format: YEAR + zero-padded OJS number to make total length 9 digits - // Examples: 202500004, 202500014, 202500196 - ojs = fmt.Sprintf("%d%05s", year, strings.TrimSpace(fmt.Sprint(row[0]))) - break + if rowDate == targetDate { + return fmt.Sprintf("%d%05s", year, strings.TrimSpace(row[0])), nil } } - if ojs == "" { - return ojs, fmt.Errorf("can not find ojs number") + + return "", fmt.Errorf("can not find ojs number for date %s", date) +} + +func getOJSFromSearchAPIAt(ctx context.Context, date, userAgent string, client *http.Client, apiURL string) (string, error) { + parsed, err := time.Parse("02/01/2006", strings.TrimSpace(date)) + if err != nil { + return "", fmt.Errorf("parse publication date %s: %w", date, err) } - return ojs, nil + payload, err := json.Marshal(searchAPIRequest{ + Query: fmt.Sprintf("publication-date=%s", parsed.Format("20060102")), + Fields: []string{"ojs-number"}, + Limit: 1, + }) + if err != nil { + return "", fmt.Errorf("marshal search request: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(payload)) + if err != nil { + return "", fmt.Errorf("create search request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + if strings.TrimSpace(userAgent) != "" { + req.Header.Set("User-Agent", userAgent) + } + + resp, err := client.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return "", err + } + + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("search API HTTP error: %d %s (body: %s)", resp.StatusCode, resp.Status, previewBody(body)) + } + + var result searchAPIResponse + if err := json.Unmarshal(body, &result); err != nil { + return "", fmt.Errorf("decode search API response: %w", err) + } + if len(result.Notices) == 0 || strings.TrimSpace(result.Notices[0].OJSNumber) == "" { + return "", fmt.Errorf("can not find ojs number for date %s", date) + } + + return formatOJSPackageID(parsed.Year(), result.Notices[0].OJSNumber) +} + +func formatOJSPackageID(fallbackYear int, ojsNumber string) (string, error) { + parts := strings.Split(strings.TrimSpace(ojsNumber), "/") + if len(parts) != 2 { + return "", fmt.Errorf("unexpected ojs-number format %q", ojsNumber) + } + + seq := strings.TrimSpace(parts[0]) + yearStr := strings.TrimSpace(parts[1]) + if seq == "" || yearStr == "" { + return "", fmt.Errorf("unexpected ojs-number format %q", ojsNumber) + } + + year := fallbackYear + if parsedYear, err := time.Parse("2006", yearStr); err == nil { + year = parsedYear.Year() + } + + return fmt.Sprintf("%d%05s", year, seq), nil +} + +func isCalendarUnavailable(err error) bool { + if err == nil { + return false + } + msg := err.Error() + return strings.Contains(msg, "calendar HTTP error: 202") || + strings.Contains(msg, "ojs calendar is empty") +} + +func previewBody(body []byte) string { + const maxLen = 160 + s := strings.TrimSpace(string(body)) + s = strings.ReplaceAll(s, "\n", " ") + s = strings.ReplaceAll(s, "\r", " ") + if len(s) <= maxLen { + return s + } + return s[:maxLen] + "..." } diff --git a/ted/scraper.go b/ted/scraper.go index 4bbb054..4224b4e 100644 --- a/ted/scraper.go +++ b/ted/scraper.go @@ -525,7 +525,7 @@ func (s *TEDScraper) processSingleDate(ctx context.Context, date time.Time) erro }) // get ojs - ojs, err := GetOJS(s.config.BaseURL, date.Year(), date.Format("02/01/2006"), s.httpClient) + ojs, err := GetOJS(ctx, s.config.BaseURL, date.Year(), date.Format("02/01/2006"), s.config.UserAgent, s.httpClient) if err != nil { s.logger.Error("Failed to get OJS", map[string]interface{}{ "date": date.Format("02/01/2006"),