Mazyar
1df44ec8ca
continuous-integration/drone/push Build is passing
- Introduced a new `translatedNoticesScanner` interface for scanning MinIO for daily translated notice counts. - Enhanced the `repository` struct to include fields for managing translated notice scope and caching. - Implemented the `ScanTranslatedNotices` method in the `StorageClient` to retrieve daily counts and total from MinIO. - Updated the `Statistics` method in the repository to utilize the new translated notices scope, improving data accuracy. - Added unit tests for the translated notices functionality, ensuring robust validation of the new features. This update significantly enhances the dashboard's ability to handle translated notice statistics, improving overall data management and reporting capabilities.
482 lines
13 KiB
Go
482 lines
13 KiB
Go
package dashboard
|
|
|
|
import (
|
|
"context"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"tm/pkg/ai_summarizer"
|
|
orm "tm/pkg/mongo"
|
|
|
|
"go.mongodb.org/mongo-driver/v2/bson"
|
|
mongodriver "go.mongodb.org/mongo-driver/v2/mongo"
|
|
)
|
|
|
|
// 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 = 15 * time.Minute
|
|
|
|
const (
|
|
scrapedScopeResolveTimeout = 3 * time.Minute
|
|
statisticsMongoQueryTimeout = 90 * time.Second
|
|
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
|
|
fromMinIO bool
|
|
dailyCounts map[string]int64 // non-nil when scope was resolved from MinIO
|
|
totalTenderCount int64 // >0 when MinIO procedure count should be used instead of Mongo Count
|
|
}
|
|
|
|
// translatedNoticesScope holds MinIO-backed translation statistics when available.
|
|
type translatedNoticesScope struct {
|
|
fromMinIO bool
|
|
dailyCounts map[string]int64
|
|
totalCount int64
|
|
}
|
|
|
|
func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) (*StatisticsReportResponse, error) {
|
|
_ = ctx
|
|
|
|
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))
|
|
|
|
var (
|
|
scrapedDocuments map[string]int64
|
|
translatedNotices map[string]int64
|
|
totalDocuments int64
|
|
totalTranslated int64
|
|
totalScrapedTED int64
|
|
scrapedScope scrapedTendersScope
|
|
translatedScope translatedNoticesScope
|
|
)
|
|
|
|
var wg sync.WaitGroup
|
|
var mu sync.Mutex
|
|
|
|
recordFailure := func(section string, err error) {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
r.logger.Warn("Dashboard statistics section failed", map[string]interface{}{
|
|
"section": section,
|
|
"error": err.Error(),
|
|
})
|
|
}
|
|
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout)
|
|
defer cancel()
|
|
|
|
total, err := r.metricsCounter.Get(qctx, orm.TEDNoticeScrapedCounterKey)
|
|
if err != nil {
|
|
recordFailure("scraped_ted_total", err)
|
|
return
|
|
}
|
|
totalScrapedTED = total
|
|
}()
|
|
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
scope, err := r.cachedTranslatedNoticesScope(context.Background())
|
|
if err != nil {
|
|
recordFailure("translated_scope", err)
|
|
return
|
|
}
|
|
translatedScope = scope
|
|
if scope.fromMinIO {
|
|
return
|
|
}
|
|
|
|
qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout)
|
|
defer cancel()
|
|
|
|
counts, err := r.translatedNoticesPerDayFromMetrics(qctx, startDay, endDay)
|
|
if err != nil {
|
|
recordFailure("translated_notices", err)
|
|
translatedNotices = map[string]int64{}
|
|
} else {
|
|
translatedNotices = counts
|
|
}
|
|
|
|
total, err := r.metricsCounter.Get(qctx, orm.AITranslationSuccessCounterKey)
|
|
if err != nil {
|
|
recordFailure("translated_total", err)
|
|
return
|
|
}
|
|
totalTranslated = total
|
|
}()
|
|
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
scope, err := r.cachedScrapedTendersScope(context.Background())
|
|
if err != nil {
|
|
recordFailure("scraped_scope", err)
|
|
scrapedScope = mongoScrapedTendersScope()
|
|
return
|
|
}
|
|
scrapedScope = scope
|
|
}()
|
|
|
|
wg.Wait()
|
|
|
|
if translatedScope.fromMinIO {
|
|
translatedNotices = filterDailyCountsSince(translatedScope.dailyCounts, startUnix)
|
|
totalTranslated = translatedScope.totalCount
|
|
}
|
|
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout)
|
|
defer cancel()
|
|
|
|
counts, err := r.scrapedDocumentsPerDay(qctx, startUnix, scrapedScope)
|
|
if err != nil {
|
|
recordFailure("scraped_documents", err)
|
|
scrapedDocuments = map[string]int64{}
|
|
return
|
|
}
|
|
scrapedDocuments = counts
|
|
}()
|
|
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout)
|
|
defer cancel()
|
|
|
|
total, err := r.totalScrapedTenders(qctx, scrapedScope)
|
|
if err != nil {
|
|
recordFailure("scraped_documents_total", err)
|
|
return
|
|
}
|
|
totalDocuments = total
|
|
}()
|
|
|
|
wg.Wait()
|
|
|
|
return &StatisticsReportResponse{
|
|
Days: days,
|
|
GeneratedAt: now.Unix(),
|
|
Daily: StatisticsDailySeries{
|
|
ScrapedDocuments: fillTrendSeries(days, scrapedDocuments),
|
|
TranslatedNotices: fillTrendSeries(days, translatedNotices),
|
|
},
|
|
Totals: StatisticsLifetimeTotals{
|
|
ScrapedDocuments: totalDocuments,
|
|
TranslatedTenders: totalTranslated,
|
|
ScrapedTEDNotices: totalScrapedTED,
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
func (r *repository) scrapedDocumentsPerDay(ctx context.Context, startUnix int64, scope scrapedTendersScope) (map[string]int64, error) {
|
|
if scope.zero {
|
|
return map[string]int64{}, nil
|
|
}
|
|
|
|
if scope.dailyCounts != nil {
|
|
return filterDailyCountsSince(scope.dailyCounts, startUnix), nil
|
|
}
|
|
|
|
return r.scrapedDocumentsPerDayFromMongo(ctx, startUnix, scope.match)
|
|
}
|
|
|
|
func filterDailyCountsSince(counts map[string]int64, startUnix int64) map[string]int64 {
|
|
if len(counts) == 0 {
|
|
return map[string]int64{}
|
|
}
|
|
startDay := time.Unix(startUnix, 0).UTC().Format("2006-01-02")
|
|
filtered := make(map[string]int64)
|
|
for date, count := range counts {
|
|
if date >= startDay {
|
|
filtered[date] = count
|
|
}
|
|
}
|
|
return filtered
|
|
}
|
|
|
|
func (r *repository) scrapedDocumentsPerDayFromMongo(ctx context.Context, startUnix int64, match bson.M) (map[string]int64, error) {
|
|
combinedMatch := bson.M{
|
|
"processing_metadata.documents_scraped_at": bson.M{"$gte": startUnix, "$gt": 0},
|
|
}
|
|
for key, value := range match {
|
|
combinedMatch[key] = value
|
|
}
|
|
|
|
pipeline := mongodriver.Pipeline{
|
|
{{Key: "$match", Value: combinedMatch}},
|
|
{{Key: "$addFields", Value: bson.M{
|
|
"metric_ts": normalizeTimestampExpr("processing_metadata.documents_scraped_at"),
|
|
}}},
|
|
{{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) translatedNoticesPerDayFromMetrics(ctx context.Context, startDay, endDay time.Time) (map[string]int64, error) {
|
|
return r.metricsCounter.GetDailyCounts(ctx, startDay, endDay)
|
|
}
|
|
|
|
func (r *repository) cachedTranslatedNoticesScope(_ context.Context) (translatedNoticesScope, error) {
|
|
now := time.Now()
|
|
|
|
r.translatedScopeMu.Lock()
|
|
if r.translatedScopeCached && now.Before(r.translatedScopeExpiry) {
|
|
scope := r.translatedScope
|
|
r.translatedScopeMu.Unlock()
|
|
return scope, nil
|
|
}
|
|
r.translatedScopeMu.Unlock()
|
|
|
|
v, err, _ := r.translatedScopeGroup.Do("translated-scope", func() (interface{}, error) {
|
|
r.translatedScopeMu.Lock()
|
|
if r.translatedScopeCached && time.Now().Before(r.translatedScopeExpiry) {
|
|
scope := r.translatedScope
|
|
r.translatedScopeMu.Unlock()
|
|
return scope, nil
|
|
}
|
|
r.translatedScopeMu.Unlock()
|
|
|
|
resolveCtx, cancel := context.WithTimeout(context.Background(), scrapedScopeResolveTimeout)
|
|
defer cancel()
|
|
|
|
scope, resolveErr := r.resolveTranslatedNoticesScope(resolveCtx)
|
|
if resolveErr != nil {
|
|
if r.logger != nil {
|
|
r.logger.Warn("MinIO translation scope unavailable, using metrics counter fallback for dashboard statistics", map[string]interface{}{
|
|
"error": resolveErr.Error(),
|
|
})
|
|
}
|
|
return translatedNoticesScope{}, nil
|
|
}
|
|
|
|
r.translatedScopeMu.Lock()
|
|
r.translatedScope = scope
|
|
r.translatedScopeExpiry = time.Now().Add(scrapedScopeCacheTTL)
|
|
r.translatedScopeCached = true
|
|
r.translatedScopeMu.Unlock()
|
|
|
|
return scope, nil
|
|
})
|
|
if err != nil {
|
|
return translatedNoticesScope{}, err
|
|
}
|
|
|
|
return v.(translatedNoticesScope), nil
|
|
}
|
|
|
|
func (r *repository) resolveTranslatedNoticesScope(ctx context.Context) (translatedNoticesScope, error) {
|
|
if r.procedureLister == nil {
|
|
return translatedNoticesScope{}, nil
|
|
}
|
|
scanner, ok := r.procedureLister.(translatedNoticesScanner)
|
|
if !ok {
|
|
return translatedNoticesScope{}, nil
|
|
}
|
|
|
|
dailyCounts, total, err := scanner.ScanTranslatedNotices(ctx)
|
|
if err != nil {
|
|
return translatedNoticesScope{}, err
|
|
}
|
|
|
|
return translatedNoticesScope{
|
|
fromMinIO: true,
|
|
dailyCounts: dailyCounts,
|
|
totalCount: total,
|
|
}, nil
|
|
}
|
|
|
|
func (r *repository) totalScrapedTenders(ctx context.Context, scope scrapedTendersScope) (int64, error) {
|
|
if scope.zero {
|
|
return 0, nil
|
|
}
|
|
if scope.totalTenderCount > 0 {
|
|
return scope.totalTenderCount, 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(_ 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 {
|
|
if r.logger != 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) {
|
|
procedures, dailyCounts, err := r.listScrapedProcedures(ctx)
|
|
if err != nil {
|
|
return scrapedTendersScope{}, err
|
|
}
|
|
|
|
folderIDs := folderIDsFromProcedures(procedures)
|
|
if len(folderIDs) > 0 {
|
|
totalTenders := int64(len(folderIDs))
|
|
if len(folderIDs) > maxScrapedFolderInClause {
|
|
if r.logger != nil {
|
|
r.logger.Warn("Too many MinIO procedure folders for dashboard statistics, using MinIO counts without Mongo $in filter", map[string]interface{}{
|
|
"folder_count": len(folderIDs),
|
|
"max_in_clause": maxScrapedFolderInClause,
|
|
})
|
|
}
|
|
return scrapedTendersScope{
|
|
fromMinIO: true,
|
|
dailyCounts: dailyCounts,
|
|
totalTenderCount: totalTenders,
|
|
}, nil
|
|
}
|
|
return scrapedTendersScope{
|
|
match: bson.M{"contract_folder_id": bson.M{"$in": folderIDs}},
|
|
fromMinIO: true,
|
|
dailyCounts: dailyCounts,
|
|
totalTenderCount: totalTenders,
|
|
}, nil
|
|
}
|
|
return mongoScrapedTendersScope(), nil
|
|
}
|
|
|
|
func (r *repository) listScrapedProcedures(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, map[string]int64, error) {
|
|
if r.procedureLister == nil {
|
|
return nil, nil, nil
|
|
}
|
|
|
|
if scanner, ok := r.procedureLister.(scrapedDocumentsScanner); ok {
|
|
return scanner.ScanScrapedDocuments(ctx)
|
|
}
|
|
|
|
procedures, err := r.procedureLister.ListProceduresWithDocuments(ctx)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
return procedures, nil, nil
|
|
}
|
|
|
|
func folderIDsFromProcedures(procedures []ai_summarizer.ProcedureDocumentsSummary) []string {
|
|
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
|
|
}
|
|
|
|
func (r *repository) scrapedDocumentsFolderIDs(ctx context.Context) ([]string, error) {
|
|
if r.procedureLister == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
procedures, _, err := r.listScrapedProcedures(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return folderIDsFromProcedures(procedures), 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
|
|
}
|