Refactor dashboard repository to integrate ProcedureDocumentsLister

- 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.
This commit is contained in:
Mazyar
2026-06-17 14:17:41 +03:30
parent 7120a954b6
commit 9676f99304
4 changed files with 172 additions and 65 deletions
+3 -1
View File
@@ -171,6 +171,7 @@ func main() {
// Initialize AI Summarizer service // Initialize AI Summarizer service
var aiSummarizerClient tender.AISummarizerClient var aiSummarizerClient tender.AISummarizerClient
var aiSummarizerStorage tender.AISummarizerStorage var aiSummarizerStorage tender.AISummarizerStorage
var procedureDocumentsLister dashboard.ProcedureDocumentsLister
var aiRecommendationClient company.AIRecommendationClient var aiRecommendationClient company.AIRecommendationClient
var aiPipelineClient ai_pipeline.Client var aiPipelineClient ai_pipeline.Client
@@ -181,6 +182,7 @@ func main() {
} }
if s := bootstrap.InitAISummarizerStorage(conf.AISummarizer, logger); s != nil { if s := bootstrap.InitAISummarizerStorage(conf.AISummarizer, logger); s != nil {
aiSummarizerStorage = s aiSummarizerStorage = s
procedureDocumentsLister = s
} }
// Initialize authorization service // Initialize authorization service
@@ -230,7 +232,7 @@ func main() {
cmsService := cms.NewService(cmsRepository, logger) cmsService := cms.NewService(cmsRepository, logger)
kanbanService := kanban.NewService(boardRepository, columnRepository, cardRepository, goRulesClient, conf.GoRules.RuleID, logger) kanbanService := kanban.NewService(boardRepository, columnRepository, cardRepository, goRulesClient, conf.GoRules.RuleID, logger)
documentScraperService := document_scraper.NewService(tenderRepository, logger) documentScraperService := document_scraper.NewService(tenderRepository, logger)
dashboardRepository := dashboard.NewRepository(mongoManager, logger) dashboardRepository := dashboard.NewRepository(mongoManager, logger, procedureDocumentsLister)
dashboardService := dashboard.NewService(dashboardRepository, logger) dashboardService := dashboard.NewService(dashboardRepository, logger)
aiPipelineService := ai_pipeline.NewService(aiPipelineClient, logger, tenderService) aiPipelineService := ai_pipeline.NewService(aiPipelineClient, logger, tenderService)
logger.Info("Services initialized successfully", map[string]interface{}{ logger.Info("Services initialized successfully", map[string]interface{}{
+33 -24
View File
@@ -6,6 +6,7 @@ import (
"strings" "strings"
"time" "time"
"tm/internal/tender" "tm/internal/tender"
"tm/pkg/ai_summarizer"
"tm/pkg/logger" "tm/pkg/logger"
orm "tm/pkg/mongo" orm "tm/pkg/mongo"
@@ -13,6 +14,12 @@ import (
"go.mongodb.org/mongo-driver/v2/mongo" "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" const defaultValueCurrency = "EUR"
// Repository defines dashboard data access. // Repository defines dashboard data access.
@@ -27,19 +34,21 @@ type Repository interface {
} }
type repository struct { type repository struct {
ormRepo orm.Repository[tender.Tender] ormRepo orm.Repository[tender.Tender]
mongoManager *orm.ConnectionManager mongoManager *orm.ConnectionManager
metricsCounter *orm.Counter metricsCounter *orm.Counter
logger logger.Logger procedureLister ProcedureDocumentsLister
logger logger.Logger
} }
// NewRepository creates a dashboard repository backed by the tenders collection. // NewRepository creates a dashboard repository backed by the tenders collection.
func NewRepository(mongoManager *orm.ConnectionManager, log logger.Logger) Repository { func NewRepository(mongoManager *orm.ConnectionManager, log logger.Logger, procedureLister ProcedureDocumentsLister) Repository {
return &repository{ return &repository{
ormRepo: orm.NewRepository[tender.Tender](mongoManager.GetCollection(tenderCollectionName()), log), ormRepo: orm.NewRepository[tender.Tender](mongoManager.GetCollection(tenderCollectionName()), log),
mongoManager: mongoManager, mongoManager: mongoManager,
metricsCounter: orm.NewCounter(mongoManager), metricsCounter: orm.NewCounter(mongoManager),
logger: log, procedureLister: procedureLister,
logger: log,
} }
} }
@@ -313,16 +322,16 @@ func (r *repository) ClosingSoon(ctx context.Context, limit int, windowSec int64
{{Key: "$sort", Value: bson.M{"effective_deadline": 1}}}, {{Key: "$sort", Value: bson.M{"effective_deadline": 1}}},
{{Key: "$limit", Value: limit}}, {{Key: "$limit", Value: limit}},
{{Key: "$project", Value: bson.M{ {{Key: "$project", Value: bson.M{
"_id": 1, "_id": 1,
"title": 1, "title": 1,
"country_code": 1, "country_code": 1,
"buyer_organization": 1, "buyer_organization": 1,
"submission_deadline": 1, "submission_deadline": 1,
"tender_deadline": 1, "tender_deadline": 1,
"status": 1, "status": 1,
"estimated_value": 1, "estimated_value": 1,
"currency": 1, "currency": 1,
"effective_deadline": 1, "effective_deadline": 1,
}}}, }}},
} }
@@ -351,12 +360,12 @@ func (r *repository) Recent(ctx context.Context, limit int, cursor string) ([]Re
filter, filter,
orm.ListPaginationOptions{ orm.ListPaginationOptions{
Projection: bson.M{ Projection: bson.M{
"title": 1, "title": 1,
"country_code": 1, "country_code": 1,
"status": 1, "status": 1,
"buyer_organization": 1, "buyer_organization": 1,
"created_at": 1, "created_at": 1,
"publication_date": 1, "publication_date": 1,
}, },
}, },
) )
+62 -40
View File
@@ -3,6 +3,7 @@ package dashboard
import ( import (
"context" "context"
"fmt" "fmt"
"strings"
"time" "time"
"tm/internal/notice" "tm/internal/notice"
@@ -63,9 +64,9 @@ func (r *repository) Statistics(ctx context.Context, days int, startUnix int64)
}) })
g.Go(func() error { g.Go(func() error {
total, err := r.totalScrapedDocuments(gctx) total, err := r.totalScrapedTenders(gctx)
if err != nil { if err != nil {
return fmt.Errorf("total scraped documents: %w", err) return fmt.Errorf("total scraped tenders: %w", err)
} }
totalDocuments = total totalDocuments = total
return nil return nil
@@ -127,23 +128,18 @@ func (r *repository) dailyCountByTimestamp(ctx context.Context, collection strin
} }
func (r *repository) scrapedDocumentsPerDay(ctx context.Context, startUnix int64) (map[string]int64, error) { func (r *repository) scrapedDocumentsPerDay(ctx context.Context, startUnix int64) (map[string]int64, error) {
match, err := r.scrapedTendersMatchFilter(ctx)
if err != nil {
return nil, err
}
if len(match) == 0 {
return map[string]int64{}, nil
}
pipeline := mongodriver.Pipeline{ pipeline := mongodriver.Pipeline{
{{Key: "$match", Value: bson.M{ {{Key: "$match", Value: match}},
"processing_metadata.documents_scraped": true,
"scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}},
}}},
{{Key: "$unwind", Value: "$scraped_documents"}},
{{Key: "$addFields", Value: bson.M{ {{Key: "$addFields", Value: bson.M{
"metric_ts": bson.M{ "metric_ts": normalizeTimestampExpr("processing_metadata.documents_scraped_at"),
"$cond": bson.A{
bson.M{"$gt": bson.A{"$scraped_documents.scraped_at", 0}},
"$scraped_documents.scraped_at",
"$processing_metadata.documents_scraped_at",
},
},
}}},
{{Key: "$addFields", Value: bson.M{
"metric_ts": normalizeTimestampExpr("metric_ts"),
}}}, }}},
{{Key: "$match", Value: bson.M{ {{Key: "$match", Value: bson.M{
"metric_ts": bson.M{"$gte": startUnix, "$gt": 0}, "metric_ts": bson.M{"$gte": startUnix, "$gt": 0},
@@ -173,37 +169,63 @@ func (r *repository) translatedNoticesPerDay(ctx context.Context, startDay, endD
return r.metricsCounter.GetDailyCounts(ctx, startDay, endDay) return r.metricsCounter.GetDailyCounts(ctx, startDay, endDay)
} }
func (r *repository) totalScrapedDocuments(ctx context.Context) (int64, error) { func (r *repository) totalScrapedTenders(ctx context.Context) (int64, error) {
pipeline := mongodriver.Pipeline{ match, err := r.scrapedTendersMatchFilter(ctx)
{{Key: "$match", Value: bson.M{
"processing_metadata.documents_scraped": true,
"scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}},
}}},
{{Key: "$project", Value: bson.M{
"doc_count": bson.M{"$size": "$scraped_documents"},
}}},
{{Key: "$group", Value: bson.M{
"_id": nil,
"total": bson.M{"$sum": "$doc_count"},
}}},
}
cursor, err := r.mongoManager.GetCollection(tenderCollectionName()).Aggregate(ctx, pipeline)
if err != nil { if err != nil {
return 0, err return 0, err
} }
defer cursor.Close(ctx) if len(match) == 0 {
if !cursor.Next(ctx) {
return 0, nil return 0, nil
} }
var row bson.M return r.ormRepo.Count(ctx, match)
if err := cursor.Decode(&row); err != nil { }
return 0, err
func (r *repository) scrapedTendersMatchFilter(ctx context.Context) (bson.M, error) {
folderIDs, err := r.scrapedDocumentsFolderIDs(ctx)
if err != nil {
return nil, err
}
if len(folderIDs) > 0 {
return bson.M{"contract_folder_id": bson.M{"$in": folderIDs}}, nil
}
if r.procedureLister != nil {
return bson.M{}, nil
}
return bson.M{
"processing_metadata.documents_scraped": true,
"scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}},
}, nil
}
func (r *repository) scrapedDocumentsFolderIDs(ctx context.Context) ([]string, error) {
if r.procedureLister == nil {
return nil, nil
} }
return aggregateCount(row, "total"), 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) { func decodeDailyCounts(ctx context.Context, cursor *mongodriver.Cursor) (map[string]int64, error) {
@@ -0,0 +1,74 @@
package dashboard
import (
"context"
"errors"
"testing"
"tm/pkg/ai_summarizer"
)
type mockProcedureDocumentsLister struct {
procedures []ai_summarizer.ProcedureDocumentsSummary
err error
}
func (m *mockProcedureDocumentsLister) ListProceduresWithDocuments(context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) {
if m.err != nil {
return nil, m.err
}
return m.procedures, nil
}
func TestScrapedDocumentsFolderIDsUsesMinIOProcedures(t *testing.T) {
repo := &repository{
procedureLister: &mockProcedureDocumentsLister{
procedures: []ai_summarizer.ProcedureDocumentsSummary{
{ContractFolderID: "PROC-1", DocumentCount: 3},
{ContractFolderID: "PROC-2", DocumentCount: 0},
{ContractFolderID: " PROC-3 ", DocumentCount: 1},
{ContractFolderID: "PROC-1", DocumentCount: 2},
},
},
}
got, err := repo.scrapedDocumentsFolderIDs(context.Background())
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
want := []string{"PROC-1", "PROC-3"}
if len(got) != len(want) {
t.Fatalf("expected %d folder IDs, got %d: %v", len(want), len(got), got)
}
for i, id := range want {
if got[i] != id {
t.Fatalf("expected folder ID %q at index %d, got %q", id, i, got[i])
}
}
}
func TestScrapedDocumentsFolderIDsWithoutLister(t *testing.T) {
repo := &repository{}
got, err := repo.scrapedDocumentsFolderIDs(context.Background())
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got != nil {
t.Fatalf("expected nil folder IDs without lister, got %v", got)
}
}
func TestScrapedDocumentsFolderIDsPropagatesError(t *testing.T) {
repo := &repository{
procedureLister: &mockProcedureDocumentsLister{
err: errors.New("minio unavailable"),
},
}
_, err := repo.scrapedDocumentsFolderIDs(context.Background())
if err == nil {
t.Fatal("expected error")
}
}