Mazyar
9676f99304
- Introduced the ProcedureDocumentsLister interface to list contract folders with scraped documents, enhancing the accuracy of document-scrape statistics. - Updated the dashboard repository to accept ProcedureDocumentsLister as a dependency, allowing for improved data retrieval. - Implemented tests for the new functionality, ensuring proper handling of scraped document folder IDs and error propagation. This update enhances the dashboard's capability to manage and report on scraped documents, improving overall system efficiency and data integrity.
577 lines
14 KiB
Go
577 lines
14 KiB
Go
package dashboard
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"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"
|
|
)
|
|
|
|
// 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)
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
// 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
|
|
|
|
total, err := r.ormRepo.Count(ctx, bson.M{})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("count total tenders: %w", err)
|
|
}
|
|
|
|
active, err := r.ormRepo.Count(ctx, bson.M{"status": tender.TenderStatusActive})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("count active tenders: %w", err)
|
|
}
|
|
|
|
awarded, err := r.ormRepo.Count(ctx, bson.M{"status": tender.TenderStatusAwarded})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("count awarded tenders: %w", err)
|
|
}
|
|
|
|
expired, err := r.countExpired(ctx, now)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cancelled, err := r.ormRepo.Count(ctx, bson.M{"status": tender.TenderStatusCancelled})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("count cancelled tenders: %w", err)
|
|
}
|
|
|
|
closingSoon, err := r.countClosingSoon(ctx, now, windowEnd)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
valueCurrency, totalValue, err := r.dominantCurrencyValue(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &SummaryResponse{
|
|
TotalTenders: total,
|
|
ActiveTenders: active,
|
|
AwardedTenders: awarded,
|
|
ExpiredTenders: expired,
|
|
CancelledTenders: cancelled,
|
|
ClosingSoon: closingSoon,
|
|
TotalEstimatedValue: totalValue,
|
|
ValueCurrency: valueCurrency,
|
|
GeneratedAt: now,
|
|
}, nil
|
|
}
|
|
|
|
func (r *repository) countExpired(ctx context.Context, now int64) (int64, error) {
|
|
pipeline := mongo.Pipeline{
|
|
{{Key: "$addFields", Value: bson.M{"expiry_deadline": expiryDeadlineExpr()}}},
|
|
{{Key: "$match", Value: 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},
|
|
},
|
|
},
|
|
}}},
|
|
{{Key: "$count", Value: "count"}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("count expired tenders: %w", err)
|
|
}
|
|
|
|
if len(results) == 0 {
|
|
return 0, nil
|
|
}
|
|
|
|
return aggregateCount(results[0], "count"), nil
|
|
}
|
|
|
|
func (r *repository) countClosingSoon(ctx context.Context, now, windowEnd int64) (int64, error) {
|
|
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: "$count", Value: "count"}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("count closing soon: %w", err)
|
|
}
|
|
|
|
if len(results) == 0 {
|
|
return 0, nil
|
|
}
|
|
|
|
return aggregateCount(results[0], "count"), nil
|
|
}
|
|
|
|
func (r *repository) dominantCurrencyValue(ctx context.Context) (string, float64, error) {
|
|
pipeline := mongo.Pipeline{
|
|
{{Key: "$match", Value: bson.M{
|
|
"estimated_value": bson.M{"$gt": 0},
|
|
"currency": bson.M{"$type": "string", "$ne": ""},
|
|
}}},
|
|
{{Key: "$group", Value: bson.M{
|
|
"_id": "$currency",
|
|
"count": bson.M{"$sum": 1},
|
|
"total": bson.M{"$sum": "$estimated_value"},
|
|
}}},
|
|
{{Key: "$sort", Value: bson.M{"count": -1}}},
|
|
{{Key: "$limit", Value: 1}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return defaultValueCurrency, 0, fmt.Errorf("dominant currency value: %w", err)
|
|
}
|
|
|
|
if len(results) == 0 {
|
|
return defaultValueCurrency, 0, nil
|
|
}
|
|
|
|
currency := defaultValueCurrency
|
|
if c, ok := results[0]["_id"].(string); ok && c != "" {
|
|
currency = strings.ToUpper(c)
|
|
}
|
|
|
|
return currency, toFloat64(results[0]["total"]), 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
|
|
}
|