Mazyar
39ac76e7b0
continuous-integration/drone/push Build is passing
- Introduced the `scrapedDocumentsScanner` interface to facilitate scanning of scraped documents from MinIO, returning both procedure summaries and daily document counts. - Updated the `ListProceduresWithDocuments` method to utilize the new scanning functionality, improving data retrieval efficiency. - Enhanced the `scrapedDocumentsPerDay` method to filter daily counts based on a specified start date, ensuring accurate reporting of document statistics. - Added unit tests for the new scanning logic and daily counts filtering, ensuring robust functionality and error handling. This update enhances the dashboard's document management capabilities, providing better insights into scraped documents and their daily counts.
596 lines
15 KiB
Go
596 lines
15 KiB
Go
package dashboard
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
"tm/internal/tender"
|
|
"tm/pkg/ai_summarizer"
|
|
"tm/pkg/logger"
|
|
orm "tm/pkg/mongo"
|
|
|
|
"go.mongodb.org/mongo-driver/v2/bson"
|
|
"go.mongodb.org/mongo-driver/v2/mongo"
|
|
"golang.org/x/sync/singleflight"
|
|
)
|
|
|
|
// ProcedureDocumentsLister lists contract folders that have scraped documents in MinIO.
|
|
// This is the source of truth for document-scrape statistics, matching tender panel search.
|
|
type ProcedureDocumentsLister interface {
|
|
ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error)
|
|
}
|
|
|
|
// scrapedDocumentsScanner optionally scans MinIO for per-day document counts in the same pass.
|
|
type scrapedDocumentsScanner interface {
|
|
ScanScrapedDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, map[string]int64, error)
|
|
}
|
|
|
|
const defaultValueCurrency = "EUR"
|
|
|
|
// Repository defines dashboard data access.
|
|
type Repository interface {
|
|
Summary(ctx context.Context, closingWindowSec int64) (*SummaryResponse, error)
|
|
Trend(ctx context.Context, days int, metric string, startUnix int64) (map[string]int64, error)
|
|
Countries(ctx context.Context, limit int) (*CountriesResponse, error)
|
|
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]
|
|
mongoManager *orm.ConnectionManager
|
|
metricsCounter *orm.Counter
|
|
procedureLister ProcedureDocumentsLister
|
|
logger logger.Logger
|
|
scrapedScopeMu sync.Mutex
|
|
scrapedScope scrapedTendersScope
|
|
scrapedScopeExpiry time.Time
|
|
scrapedScopeCached bool
|
|
scrapedScopeGroup singleflight.Group
|
|
}
|
|
|
|
// NewRepository creates a dashboard repository backed by the tenders collection.
|
|
func NewRepository(mongoManager *orm.ConnectionManager, log logger.Logger, procedureLister ProcedureDocumentsLister) Repository {
|
|
return &repository{
|
|
ormRepo: orm.NewRepository[tender.Tender](mongoManager.GetCollection(tenderCollectionName()), log),
|
|
mongoManager: mongoManager,
|
|
metricsCounter: orm.NewCounter(mongoManager),
|
|
procedureLister: procedureLister,
|
|
logger: log,
|
|
}
|
|
}
|
|
|
|
func tenderCollectionName() string {
|
|
return "tenders"
|
|
}
|
|
|
|
func (r *repository) Summary(ctx context.Context, closingWindowSec int64) (*SummaryResponse, error) {
|
|
now := time.Now().Unix()
|
|
windowEnd := now + closingWindowSec
|
|
|
|
pipeline := mongo.Pipeline{
|
|
{{Key: "$facet", Value: bson.M{
|
|
"total": bson.A{
|
|
bson.M{"$count": "count"},
|
|
},
|
|
"active": bson.A{
|
|
bson.M{"$match": bson.M{"status": tender.TenderStatusActive}},
|
|
bson.M{"$count": "count"},
|
|
},
|
|
"awarded": bson.A{
|
|
bson.M{"$match": bson.M{"status": tender.TenderStatusAwarded}},
|
|
bson.M{"$count": "count"},
|
|
},
|
|
"cancelled": bson.A{
|
|
bson.M{"$match": bson.M{"status": tender.TenderStatusCancelled}},
|
|
bson.M{"$count": "count"},
|
|
},
|
|
"expired": bson.A{
|
|
bson.M{"$addFields": bson.M{"expiry_deadline": expiryDeadlineExpr()}},
|
|
bson.M{"$match": bson.M{
|
|
"status": bson.M{"$nin": []tender.TenderStatus{
|
|
tender.TenderStatusAwarded,
|
|
tender.TenderStatusCancelled,
|
|
}},
|
|
"$or": []bson.M{
|
|
{"status": tender.TenderStatusExpired},
|
|
{"expiry_deadline": bson.M{"$gt": 0, "$lte": now}},
|
|
},
|
|
}},
|
|
bson.M{"$count": "count"},
|
|
},
|
|
"closing_soon": bson.A{
|
|
bson.M{"$addFields": bson.M{"effective_deadline": effectiveDeadlineExpr()}},
|
|
bson.M{"$match": bson.M{
|
|
"effective_deadline": bson.M{"$gt": now, "$lte": windowEnd},
|
|
}},
|
|
bson.M{"$count": "count"},
|
|
},
|
|
"currency_value": bson.A{
|
|
bson.M{"$match": bson.M{
|
|
"estimated_value": bson.M{"$gt": 0},
|
|
"currency": bson.M{"$type": "string", "$ne": ""},
|
|
}},
|
|
bson.M{"$group": bson.M{
|
|
"_id": "$currency",
|
|
"count": bson.M{"$sum": 1},
|
|
"total": bson.M{"$sum": "$estimated_value"},
|
|
}},
|
|
bson.M{"$sort": bson.M{"count": -1}},
|
|
bson.M{"$limit": 1},
|
|
},
|
|
}}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("dashboard summary aggregation: %w", err)
|
|
}
|
|
if len(results) == 0 {
|
|
return &SummaryResponse{
|
|
ValueCurrency: defaultValueCurrency,
|
|
GeneratedAt: now,
|
|
}, nil
|
|
}
|
|
|
|
facet := results[0]
|
|
valueCurrency, totalValue := facetCurrencyValue(facet)
|
|
|
|
return &SummaryResponse{
|
|
TotalTenders: facetCount(facet, "total"),
|
|
ActiveTenders: facetCount(facet, "active"),
|
|
AwardedTenders: facetCount(facet, "awarded"),
|
|
ExpiredTenders: facetCount(facet, "expired"),
|
|
CancelledTenders: facetCount(facet, "cancelled"),
|
|
ClosingSoon: facetCount(facet, "closing_soon"),
|
|
TotalEstimatedValue: totalValue,
|
|
ValueCurrency: valueCurrency,
|
|
GeneratedAt: now,
|
|
}, nil
|
|
}
|
|
|
|
func facetCount(facet bson.M, key string) int64 {
|
|
rows := facetRows(facet, key)
|
|
if len(rows) == 0 {
|
|
return 0
|
|
}
|
|
return aggregateCount(asBSONMap(rows[0]), "count")
|
|
}
|
|
|
|
func facetCurrencyValue(facet bson.M) (string, float64) {
|
|
rows := facetRows(facet, "currency_value")
|
|
if len(rows) == 0 {
|
|
return defaultValueCurrency, 0
|
|
}
|
|
row := asBSONMap(rows[0])
|
|
if row == nil {
|
|
return defaultValueCurrency, 0
|
|
}
|
|
|
|
currency := defaultValueCurrency
|
|
if c, ok := row["_id"].(string); ok && c != "" {
|
|
currency = strings.ToUpper(c)
|
|
}
|
|
return currency, toFloat64(row["total"])
|
|
}
|
|
|
|
func facetRows(facet bson.M, key string) []interface{} {
|
|
raw, ok := facet[key]
|
|
if !ok {
|
|
return nil
|
|
}
|
|
switch rows := raw.(type) {
|
|
case bson.A:
|
|
return []interface{}(rows)
|
|
case []interface{}:
|
|
return rows
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func asBSONMap(v interface{}) bson.M {
|
|
switch doc := v.(type) {
|
|
case bson.M:
|
|
return doc
|
|
case bson.D:
|
|
out := make(bson.M, len(doc))
|
|
for _, elem := range doc {
|
|
out[elem.Key] = elem.Value
|
|
}
|
|
return out
|
|
case map[string]interface{}:
|
|
return doc
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func (r *repository) Trend(ctx context.Context, days int, metric string, startUnix int64) (map[string]int64, error) {
|
|
field := trendTimestampField(metric)
|
|
|
|
pipeline := mongo.Pipeline{
|
|
{{Key: "$match", Value: bson.M{
|
|
field: bson.M{"$gte": startUnix, "$gt": 0},
|
|
}}},
|
|
{{Key: "$addFields", Value: bson.M{
|
|
"metric_ts": normalizeTimestampExpr(field),
|
|
}}},
|
|
{{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},
|
|
}}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("trend aggregation: %w", err)
|
|
}
|
|
|
|
counts := make(map[string]int64, len(results))
|
|
for _, row := range results {
|
|
date, ok := row["_id"].(string)
|
|
if !ok || date == "" {
|
|
continue
|
|
}
|
|
counts[date] = aggregateCount(row, "count")
|
|
}
|
|
|
|
return counts, nil
|
|
}
|
|
|
|
func (r *repository) Countries(ctx context.Context, limit int) (*CountriesResponse, error) {
|
|
pipeline := mongo.Pipeline{
|
|
{{Key: "$match", Value: bson.M{
|
|
"country_code": bson.M{"$type": "string", "$ne": ""},
|
|
}}},
|
|
{{Key: "$group", Value: bson.M{
|
|
"_id": bson.M{"$toUpper": "$country_code"},
|
|
"count": bson.M{"$sum": 1},
|
|
}}},
|
|
{{Key: "$sort", Value: bson.M{"count": -1}}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("countries aggregation: %w", err)
|
|
}
|
|
|
|
var total int64
|
|
items := make([]CountryItem, 0, limit)
|
|
var other int64
|
|
|
|
for i, row := range results {
|
|
code, _ := row["_id"].(string)
|
|
count := aggregateCount(row, "count")
|
|
total += count
|
|
|
|
if code == "" {
|
|
other += count
|
|
continue
|
|
}
|
|
|
|
if i < limit {
|
|
items = append(items, CountryItem{CountryCode: code, Count: count})
|
|
} else {
|
|
other += count
|
|
}
|
|
}
|
|
|
|
return &CountriesResponse{
|
|
Total: total,
|
|
Items: items,
|
|
OtherCount: other,
|
|
}, nil
|
|
}
|
|
|
|
func (r *repository) NoticeTypes(ctx context.Context) (*NoticeTypesResponse, error) {
|
|
pipeline := mongo.Pipeline{
|
|
{{Key: "$group", Value: bson.M{
|
|
"_id": bson.M{
|
|
"$cond": bson.A{
|
|
bson.M{"$and": bson.A{
|
|
bson.M{"$ne": bson.A{"$notice_type_code", ""}},
|
|
bson.M{"$ne": bson.A{"$notice_type_code", nil}},
|
|
}},
|
|
"$notice_type_code",
|
|
"UNKNOWN",
|
|
},
|
|
},
|
|
"count": bson.M{"$sum": 1},
|
|
}}},
|
|
{{Key: "$sort", Value: bson.M{"count": -1}}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("notice types aggregation: %w", err)
|
|
}
|
|
|
|
items := make([]NoticeTypeItem, 0, len(results))
|
|
var total int64
|
|
for _, row := range results {
|
|
noticeType, _ := row["_id"].(string)
|
|
count := aggregateCount(row, "count")
|
|
total += count
|
|
items = append(items, NoticeTypeItem{Type: noticeType, Count: count})
|
|
}
|
|
|
|
return &NoticeTypesResponse{Total: total, Items: items}, nil
|
|
}
|
|
|
|
func (r *repository) ClosingSoon(ctx context.Context, limit int, windowSec int64) ([]ClosingSoonItem, error) {
|
|
now := time.Now().Unix()
|
|
windowEnd := now + windowSec
|
|
|
|
pipeline := mongo.Pipeline{
|
|
{{Key: "$addFields", Value: bson.M{"effective_deadline": effectiveDeadlineExpr()}}},
|
|
{{Key: "$match", Value: bson.M{
|
|
"effective_deadline": bson.M{"$gt": now, "$lte": windowEnd},
|
|
}}},
|
|
{{Key: "$sort", Value: bson.M{"effective_deadline": 1}}},
|
|
{{Key: "$limit", Value: limit}},
|
|
{{Key: "$project", Value: bson.M{
|
|
"_id": 1,
|
|
"title": 1,
|
|
"country_code": 1,
|
|
"buyer_organization": 1,
|
|
"submission_deadline": 1,
|
|
"tender_deadline": 1,
|
|
"status": 1,
|
|
"estimated_value": 1,
|
|
"currency": 1,
|
|
"effective_deadline": 1,
|
|
}}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("closing soon list: %w", err)
|
|
}
|
|
|
|
items := make([]ClosingSoonItem, 0, len(results))
|
|
for _, row := range results {
|
|
items = append(items, mapClosingSoonRow(row))
|
|
}
|
|
|
|
return items, nil
|
|
}
|
|
|
|
func (r *repository) Recent(ctx context.Context, limit int, cursor string) ([]RecentItem, *string, error) {
|
|
filter := bson.M{}
|
|
|
|
mongoPagination, err := orm.BuildListPagination(
|
|
limit,
|
|
0,
|
|
cursor,
|
|
"created_at",
|
|
"desc",
|
|
filter,
|
|
orm.ListPaginationOptions{
|
|
Projection: bson.M{
|
|
"title": 1,
|
|
"country_code": 1,
|
|
"status": 1,
|
|
"buyer_organization": 1,
|
|
"created_at": 1,
|
|
"publication_date": 1,
|
|
},
|
|
},
|
|
)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("recent pagination: %w", err)
|
|
}
|
|
|
|
result, err := r.ormRepo.FindAll(ctx, filter, mongoPagination)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("recent list: %w", err)
|
|
}
|
|
|
|
items := make([]RecentItem, 0, len(result.Items))
|
|
for _, t := range result.Items {
|
|
items = append(items, mapRecentTender(&t))
|
|
}
|
|
|
|
var nextCursor *string
|
|
if result.NextCursor != "" {
|
|
nextCursor = &result.NextCursor
|
|
}
|
|
|
|
return items, nextCursor, nil
|
|
}
|
|
|
|
func mapClosingSoonRow(row bson.M) ClosingSoonItem {
|
|
item := ClosingSoonItem{
|
|
Title: stringField(row, "title"),
|
|
CountryCode: strings.ToUpper(stringField(row, "country_code")),
|
|
Status: stringField(row, "status"),
|
|
EstimatedValue: toFloat64(row["estimated_value"]),
|
|
Currency: strings.ToUpper(stringField(row, "currency")),
|
|
}
|
|
|
|
if id, ok := row["_id"]; ok {
|
|
item.ID = objectIDString(id)
|
|
}
|
|
|
|
item.SubmissionDeadline = normalizeUnixTimestamp(int64Field(row, "submission_deadline"))
|
|
item.TenderDeadline = normalizeUnixTimestamp(int64Field(row, "tender_deadline"))
|
|
|
|
if effective := int64Field(row, "effective_deadline"); effective > 0 && item.SubmissionDeadline == 0 {
|
|
item.SubmissionDeadline = effective
|
|
}
|
|
|
|
if buyer, ok := row["buyer_organization"].(bson.M); ok {
|
|
item.BuyerName = stringField(buyer, "name")
|
|
} else if buyer, ok := row["buyer_organization"].(map[string]interface{}); ok {
|
|
if name, ok := buyer["name"].(string); ok {
|
|
item.BuyerName = name
|
|
}
|
|
}
|
|
|
|
return item
|
|
}
|
|
|
|
func mapRecentTender(t *tender.Tender) RecentItem {
|
|
item := RecentItem{
|
|
ID: t.GetID(),
|
|
Title: t.Title,
|
|
CountryCode: strings.ToUpper(t.CountryCode),
|
|
Status: string(t.Status),
|
|
CreatedAt: normalizeUnixTimestamp(t.CreatedAt),
|
|
PublicationDate: normalizeUnixTimestamp(t.PublicationDate),
|
|
}
|
|
|
|
if t.BuyerOrganization != nil {
|
|
item.BuyerName = t.BuyerOrganization.Name
|
|
}
|
|
|
|
if item.CreatedAt == 0 && item.PublicationDate > 0 {
|
|
item.CreatedAt = item.PublicationDate
|
|
}
|
|
|
|
return item
|
|
}
|
|
|
|
func trendTimestampField(metric string) string {
|
|
switch strings.ToLower(strings.TrimSpace(metric)) {
|
|
case "published":
|
|
return "publication_date"
|
|
case "awarded":
|
|
return "award_date"
|
|
default:
|
|
return "created_at"
|
|
}
|
|
}
|
|
|
|
// effectiveDeadlineExpr is used for closing-soon (submission first, per dashboard-api.md).
|
|
func effectiveDeadlineExpr() bson.M {
|
|
return normalizeDeadlineFieldExpr(
|
|
"$submission_deadline",
|
|
"$tender_deadline",
|
|
)
|
|
}
|
|
|
|
// expiryDeadlineExpr is used for expired counts (tender deadline takes precedence).
|
|
func expiryDeadlineExpr() bson.M {
|
|
return normalizeDeadlineFieldExpr(
|
|
"$tender_deadline",
|
|
"$submission_deadline",
|
|
)
|
|
}
|
|
|
|
func normalizeDeadlineFieldExpr(primaryField, fallbackField string) bson.M {
|
|
return bson.M{
|
|
"$let": bson.M{
|
|
"vars": bson.M{
|
|
"raw": bson.M{
|
|
"$cond": bson.A{
|
|
bson.M{"$gt": bson.A{primaryField, 0}},
|
|
primaryField,
|
|
fallbackField,
|
|
},
|
|
},
|
|
},
|
|
"in": normalizeTimestampExpr("$$raw"),
|
|
},
|
|
}
|
|
}
|
|
|
|
func normalizeTimestampExpr(field string) bson.M {
|
|
ref := field
|
|
if !strings.HasPrefix(ref, "$") {
|
|
ref = "$" + ref
|
|
}
|
|
return bson.M{
|
|
"$cond": bson.A{
|
|
bson.M{"$gt": bson.A{ref, int64(1_000_000_000_000)}},
|
|
bson.M{"$divide": bson.A{ref, 1000}},
|
|
ref,
|
|
},
|
|
}
|
|
}
|
|
|
|
func aggregateCount(row bson.M, field string) int64 {
|
|
if row == nil {
|
|
return 0
|
|
}
|
|
if field != "" {
|
|
if v, ok := row[field]; ok {
|
|
return toInt64(v)
|
|
}
|
|
}
|
|
if v, ok := row["count"]; ok {
|
|
return toInt64(v)
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func toInt64(v interface{}) int64 {
|
|
switch n := v.(type) {
|
|
case int32:
|
|
return int64(n)
|
|
case int64:
|
|
return n
|
|
case float64:
|
|
return int64(n)
|
|
case int:
|
|
return int64(n)
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
|
|
func toFloat64(v interface{}) float64 {
|
|
switch n := v.(type) {
|
|
case float64:
|
|
return n
|
|
case float32:
|
|
return float64(n)
|
|
case int32:
|
|
return float64(n)
|
|
case int64:
|
|
return float64(n)
|
|
case int:
|
|
return float64(n)
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
|
|
func int64Field(row bson.M, key string) int64 {
|
|
return toInt64(row[key])
|
|
}
|
|
|
|
func stringField(row bson.M, key string) string {
|
|
if v, ok := row[key].(string); ok {
|
|
return v
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func objectIDString(v interface{}) string {
|
|
switch id := v.(type) {
|
|
case bson.ObjectID:
|
|
return id.Hex()
|
|
default:
|
|
return fmt.Sprint(v)
|
|
}
|
|
}
|
|
|
|
func normalizeUnixTimestamp(value int64) int64 {
|
|
if value > 1_000_000_000_000 {
|
|
return value / 1000
|
|
}
|
|
return value
|
|
}
|