Mazyar
541d08a014
continuous-integration/drone/push Build is passing
- Added new field `Days` and `ScrapedTED` to `SummaryResponse` for tracking daily TED scrape counts. - Updated `SummaryQuery` to include `Days` parameter for querying scraped TED data. - Modified `Summary` handler to accept and process the new `Days` parameter. - Refactored `Summary` method in the repository to support the new `Days` parameter and improved aggregation logic. - Enhanced caching logic in the service layer to utilize a composite cache key based on `closingWindowSec` and `days`, improving cache management and retrieval efficiency. This update improves the dashboard's functionality by providing more detailed insights into TED scraping activities and optimizing the caching strategy for better performance.
693 lines
17 KiB
Go
693 lines
17 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, days int) (*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, days int) (*SummaryResponse, error) {
|
|
now := time.Now().Unix()
|
|
windowEnd := now + closingWindowSec
|
|
|
|
nowUTC := time.Now().UTC()
|
|
endDay := time.Date(nowUTC.Year(), nowUTC.Month(), nowUTC.Day(), 0, 0, 0, 0, time.UTC)
|
|
startDay := endDay.AddDate(0, 0, -(days - 1))
|
|
|
|
var (
|
|
statusCounts summaryStatusCounts
|
|
valueCurrency string
|
|
totalValue float64
|
|
scrapedTED map[string]int64
|
|
wg sync.WaitGroup
|
|
mu sync.Mutex
|
|
firstErr error
|
|
)
|
|
|
|
recordErr := func(section string, err error) {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
if firstErr == nil {
|
|
firstErr = fmt.Errorf("%s: %w", section, err)
|
|
}
|
|
}
|
|
|
|
wg.Add(3)
|
|
go func() {
|
|
defer wg.Done()
|
|
counts, err := r.summaryStatusCounts(ctx, now, windowEnd)
|
|
if err != nil {
|
|
recordErr("status_counts", err)
|
|
return
|
|
}
|
|
statusCounts = counts
|
|
}()
|
|
|
|
go func() {
|
|
defer wg.Done()
|
|
currency, value, err := r.summaryCurrencyValue(ctx)
|
|
if err != nil {
|
|
recordErr("currency_value", err)
|
|
return
|
|
}
|
|
valueCurrency, totalValue = currency, value
|
|
}()
|
|
|
|
go func() {
|
|
defer wg.Done()
|
|
qctx, cancel := context.WithTimeout(ctx, statisticsMongoQueryTimeout)
|
|
defer cancel()
|
|
|
|
counts, err := r.metricsCounter.GetTEDNoticeScrapedDailyCounts(qctx, startDay, endDay)
|
|
if err != nil {
|
|
if r.logger != nil {
|
|
r.logger.Warn("Dashboard summary scraped TED daily counts failed", map[string]interface{}{
|
|
"error": err.Error(),
|
|
})
|
|
}
|
|
scrapedTED = map[string]int64{}
|
|
return
|
|
}
|
|
scrapedTED = counts
|
|
}()
|
|
|
|
wg.Wait()
|
|
if firstErr != nil {
|
|
return nil, fmt.Errorf("dashboard summary aggregation: %w", firstErr)
|
|
}
|
|
|
|
return &SummaryResponse{
|
|
TotalTenders: statusCounts.total,
|
|
ActiveTenders: statusCounts.active,
|
|
AwardedTenders: statusCounts.awarded,
|
|
ExpiredTenders: statusCounts.expired,
|
|
CancelledTenders: statusCounts.cancelled,
|
|
ClosingSoon: statusCounts.closingSoon,
|
|
TotalEstimatedValue: totalValue,
|
|
ValueCurrency: valueCurrency,
|
|
Days: days,
|
|
ScrapedTED: fillTrendSeries(days, scrapedTED),
|
|
GeneratedAt: now,
|
|
}, nil
|
|
}
|
|
|
|
type summaryStatusCounts struct {
|
|
total int64
|
|
active int64
|
|
awarded int64
|
|
expired int64
|
|
cancelled int64
|
|
closingSoon int64
|
|
}
|
|
|
|
func (r *repository) summaryStatusCounts(ctx context.Context, now, windowEnd int64) (summaryStatusCounts, error) {
|
|
pipeline := mongo.Pipeline{
|
|
{{Key: "$addFields", Value: bson.M{
|
|
"expiry_deadline": expiryDeadlineExpr(),
|
|
"effective_deadline": effectiveDeadlineExpr(),
|
|
}}},
|
|
{{Key: "$group", Value: bson.M{
|
|
"_id": nil,
|
|
"total": bson.M{"$sum": 1},
|
|
"active": bson.M{"$sum": bson.M{"$cond": bson.A{
|
|
bson.M{"$eq": bson.A{"$status", tender.TenderStatusActive}}, 1, 0,
|
|
}}},
|
|
"awarded": bson.M{"$sum": bson.M{"$cond": bson.A{
|
|
bson.M{"$eq": bson.A{"$status", tender.TenderStatusAwarded}}, 1, 0,
|
|
}}},
|
|
"cancelled": bson.M{"$sum": bson.M{"$cond": bson.A{
|
|
bson.M{"$eq": bson.A{"$status", tender.TenderStatusCancelled}}, 1, 0,
|
|
}}},
|
|
"expired": bson.M{"$sum": bson.M{"$cond": bson.A{
|
|
bson.M{"$and": bson.A{
|
|
bson.M{"$not": bson.M{"$in": bson.A{
|
|
"$status",
|
|
bson.A{tender.TenderStatusAwarded, tender.TenderStatusCancelled},
|
|
}}},
|
|
bson.M{"$or": bson.A{
|
|
bson.M{"$eq": bson.A{"$status", tender.TenderStatusExpired}},
|
|
bson.M{"$and": bson.A{
|
|
bson.M{"$gt": bson.A{"$expiry_deadline", 0}},
|
|
bson.M{"$lte": bson.A{"$expiry_deadline", now}},
|
|
}},
|
|
}},
|
|
}}, 1, 0,
|
|
}}},
|
|
"closing_soon": bson.M{"$sum": bson.M{"$cond": bson.A{
|
|
bson.M{"$and": bson.A{
|
|
bson.M{"$gt": bson.A{"$effective_deadline", now}},
|
|
bson.M{"$lte": bson.A{"$effective_deadline", windowEnd}},
|
|
}}, 1, 0,
|
|
}}},
|
|
}}},
|
|
}
|
|
|
|
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
|
if err != nil {
|
|
return summaryStatusCounts{}, err
|
|
}
|
|
if len(results) == 0 {
|
|
return summaryStatusCounts{}, nil
|
|
}
|
|
|
|
row := results[0]
|
|
return summaryStatusCounts{
|
|
total: aggregateCount(row, "total"),
|
|
active: aggregateCount(row, "active"),
|
|
awarded: aggregateCount(row, "awarded"),
|
|
expired: aggregateCount(row, "expired"),
|
|
cancelled: aggregateCount(row, "cancelled"),
|
|
closingSoon: aggregateCount(row, "closing_soon"),
|
|
}, nil
|
|
}
|
|
|
|
func (r *repository) summaryCurrencyValue(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, err
|
|
}
|
|
if len(results) == 0 {
|
|
return defaultValueCurrency, 0, nil
|
|
}
|
|
|
|
currency, total := facetCurrencyValue(bson.M{"currency_value": bson.A{results[0]}})
|
|
return currency, total, 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
|
|
}
|