Mazyar
327 lines
9.0 KiB
Go
327 lines
9.0 KiB
Go
package dashboard
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"tm/internal/notice"
|
|
orm "tm/pkg/mongo"
|
|
|
|
"go.mongodb.org/mongo-driver/v2/bson"
|
|
mongodriver "go.mongodb.org/mongo-driver/v2/mongo"
|
|
"golang.org/x/sync/errgroup"
|
|
)
|
|
|
|
const noticesCollectionName = "notices"
|
|
|
|
// scrapedScopeCacheTTL avoids listing every MinIO object on each statistics request while
|
|
// keeping counts aligned with the tender panel (MinIO is the source of truth for scraped docs).
|
|
const scrapedScopeCacheTTL = 5 * time.Minute
|
|
|
|
const (
|
|
scrapedScopeResolveTimeout = 2 * time.Minute
|
|
statisticsQueryTimeout = 2 * time.Minute
|
|
maxScrapedFolderInClause = 5000
|
|
)
|
|
|
|
// scrapedTendersScope is the resolved Mongo filter for tenders with scraped documents.
|
|
// zero is set when MinIO reports no procedure folders; callers must return zero without querying.
|
|
type scrapedTendersScope struct {
|
|
match bson.M
|
|
zero bool
|
|
}
|
|
|
|
func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) (*StatisticsReportResponse, error) {
|
|
queryCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), statisticsQueryTimeout)
|
|
defer cancel()
|
|
|
|
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))
|
|
|
|
// Resolve scraped-document scope from MinIO (cached); falls back to Mongo flags on error.
|
|
scrapedScope, err := r.cachedScrapedTendersScope(queryCtx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var (
|
|
scrapedTED map[string]int64
|
|
scrapedDocuments map[string]int64
|
|
translatedNotices map[string]int64
|
|
totalDocuments int64
|
|
totalTranslated int64
|
|
)
|
|
|
|
g, gctx := errgroup.WithContext(queryCtx)
|
|
|
|
g.Go(func() error {
|
|
// Use created_at (first ingest time). processing_metadata.scraped_at is reset on
|
|
// notice refresh merges, which would collapse historical scrapes onto the latest day.
|
|
counts, err := r.dailyCountByTimestamp(gctx, noticesCollectionName, bson.M{
|
|
"source": notice.TenderSourceTEDScraper,
|
|
"created_at": bson.M{"$gte": startUnix, "$gt": 0},
|
|
}, "created_at")
|
|
if err != nil {
|
|
return fmt.Errorf("scraped ted per day: %w", err)
|
|
}
|
|
scrapedTED = counts
|
|
return nil
|
|
})
|
|
|
|
g.Go(func() error {
|
|
counts, err := r.scrapedDocumentsPerDay(gctx, startUnix, scrapedScope)
|
|
if err != nil {
|
|
return fmt.Errorf("scraped documents per day: %w", err)
|
|
}
|
|
scrapedDocuments = counts
|
|
return nil
|
|
})
|
|
|
|
g.Go(func() error {
|
|
counts, err := r.translatedNoticesPerDay(gctx, startDay, endDay)
|
|
if err != nil {
|
|
return fmt.Errorf("translated notices per day: %w", err)
|
|
}
|
|
translatedNotices = counts
|
|
return nil
|
|
})
|
|
|
|
g.Go(func() error {
|
|
total, err := r.totalScrapedTenders(gctx, scrapedScope)
|
|
if err != nil {
|
|
return fmt.Errorf("total scraped tenders: %w", err)
|
|
}
|
|
totalDocuments = total
|
|
return nil
|
|
})
|
|
|
|
g.Go(func() error {
|
|
total, err := r.metricsCounter.Get(gctx, orm.AITranslationSuccessCounterKey)
|
|
if err != nil {
|
|
return fmt.Errorf("total translated tenders: %w", err)
|
|
}
|
|
totalTranslated = total
|
|
return nil
|
|
})
|
|
|
|
if err := g.Wait(); err != nil {
|
|
return nil, 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, scope scrapedTendersScope) (map[string]int64, error) {
|
|
if scope.zero {
|
|
return map[string]int64{}, nil
|
|
}
|
|
|
|
// Daily buckets use documents_scraped_at only. Tenders present in MinIO but not yet
|
|
// synced to Mongo are included in the lifetime total but omitted from the daily series.
|
|
pipeline := mongodriver.Pipeline{
|
|
{{Key: "$match", Value: scope.match}},
|
|
{{Key: "$addFields", Value: bson.M{
|
|
"metric_ts": normalizeTimestampExpr("processing_metadata.documents_scraped_at"),
|
|
}}},
|
|
{{Key: "$match", Value: bson.M{
|
|
"metric_ts": bson.M{"$gte": startUnix, "$gt": 0},
|
|
}}},
|
|
{{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) totalScrapedTenders(ctx context.Context, scope scrapedTendersScope) (int64, error) {
|
|
if scope.zero {
|
|
return 0, nil
|
|
}
|
|
|
|
return r.ormRepo.Count(ctx, scope.match)
|
|
}
|
|
|
|
func mongoScrapedTendersScope() scrapedTendersScope {
|
|
return scrapedTendersScope{
|
|
match: bson.M{
|
|
"processing_metadata.documents_scraped": true,
|
|
"scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}},
|
|
},
|
|
}
|
|
}
|
|
|
|
func (r *repository) cachedScrapedTendersScope(ctx context.Context) (scrapedTendersScope, error) {
|
|
now := time.Now()
|
|
|
|
r.scrapedScopeMu.Lock()
|
|
if r.scrapedScopeCached && now.Before(r.scrapedScopeExpiry) {
|
|
scope := r.scrapedScope
|
|
r.scrapedScopeMu.Unlock()
|
|
return scope, nil
|
|
}
|
|
r.scrapedScopeMu.Unlock()
|
|
|
|
v, err, _ := r.scrapedScopeGroup.Do("scraped-scope", func() (interface{}, error) {
|
|
r.scrapedScopeMu.Lock()
|
|
if r.scrapedScopeCached && time.Now().Before(r.scrapedScopeExpiry) {
|
|
scope := r.scrapedScope
|
|
r.scrapedScopeMu.Unlock()
|
|
return scope, nil
|
|
}
|
|
r.scrapedScopeMu.Unlock()
|
|
|
|
resolveCtx, cancel := context.WithTimeout(context.Background(), scrapedScopeResolveTimeout)
|
|
defer cancel()
|
|
|
|
scope, resolveErr := r.resolveScrapedTendersScope(resolveCtx)
|
|
if resolveErr != nil {
|
|
r.logger.Warn("MinIO scraped scope unavailable, using Mongo fallback for dashboard statistics", map[string]interface{}{
|
|
"error": resolveErr.Error(),
|
|
})
|
|
return mongoScrapedTendersScope(), nil
|
|
}
|
|
|
|
r.scrapedScopeMu.Lock()
|
|
r.scrapedScope = scope
|
|
r.scrapedScopeExpiry = time.Now().Add(scrapedScopeCacheTTL)
|
|
r.scrapedScopeCached = true
|
|
r.scrapedScopeMu.Unlock()
|
|
|
|
return scope, nil
|
|
})
|
|
if err != nil {
|
|
return scrapedTendersScope{}, err
|
|
}
|
|
|
|
return v.(scrapedTendersScope), nil
|
|
}
|
|
|
|
func (r *repository) resolveScrapedTendersScope(ctx context.Context) (scrapedTendersScope, error) {
|
|
folderIDs, err := r.scrapedDocumentsFolderIDs(ctx)
|
|
if err != nil {
|
|
return scrapedTendersScope{}, err
|
|
}
|
|
if len(folderIDs) > 0 {
|
|
if len(folderIDs) > maxScrapedFolderInClause {
|
|
r.logger.Warn("Too many MinIO procedure folders for dashboard statistics, using Mongo fallback", map[string]interface{}{
|
|
"folder_count": len(folderIDs),
|
|
"max_in_clause": maxScrapedFolderInClause,
|
|
})
|
|
return mongoScrapedTendersScope(), nil
|
|
}
|
|
return scrapedTendersScope{
|
|
match: bson.M{"contract_folder_id": bson.M{"$in": folderIDs}},
|
|
}, nil
|
|
}
|
|
return mongoScrapedTendersScope(), nil
|
|
}
|
|
|
|
func (r *repository) scrapedDocumentsFolderIDs(ctx context.Context) ([]string, error) {
|
|
if r.procedureLister == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
procedures, err := r.procedureLister.ListProceduresWithDocuments(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
folderIDs := make([]string, 0, len(procedures))
|
|
seen := make(map[string]struct{}, len(procedures))
|
|
for _, proc := range procedures {
|
|
if proc.DocumentCount <= 0 {
|
|
continue
|
|
}
|
|
id := strings.TrimSpace(proc.ContractFolderID)
|
|
if id == "" {
|
|
continue
|
|
}
|
|
if _, ok := seen[id]; ok {
|
|
continue
|
|
}
|
|
seen[id] = struct{}{}
|
|
folderIDs = append(folderIDs, id)
|
|
}
|
|
|
|
return folderIDs, 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
|
|
}
|