Add scrapedDocumentsScanner interface and enhance document scanning logic
continuous-integration/drone/push Build is passing
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.
This commit is contained in:
@@ -22,6 +22,11 @@ type ProcedureDocumentsLister interface {
|
|||||||
ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error)
|
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"
|
const defaultValueCurrency = "EUR"
|
||||||
|
|
||||||
// Repository defines dashboard data access.
|
// Repository defines dashboard data access.
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"tm/internal/notice"
|
"tm/internal/notice"
|
||||||
|
"tm/pkg/ai_summarizer"
|
||||||
orm "tm/pkg/mongo"
|
orm "tm/pkg/mongo"
|
||||||
|
|
||||||
"go.mongodb.org/mongo-driver/v2/bson"
|
"go.mongodb.org/mongo-driver/v2/bson"
|
||||||
@@ -31,6 +32,8 @@ const (
|
|||||||
type scrapedTendersScope struct {
|
type scrapedTendersScope struct {
|
||||||
match bson.M
|
match bson.M
|
||||||
zero bool
|
zero bool
|
||||||
|
fromMinIO bool
|
||||||
|
dailyCounts map[string]int64 // non-nil when scope was resolved from MinIO
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) (*StatisticsReportResponse, error) {
|
func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) (*StatisticsReportResponse, error) {
|
||||||
@@ -158,12 +161,42 @@ func (r *repository) scrapedDocumentsPerDay(ctx context.Context, startUnix int64
|
|||||||
return map[string]int64{}, nil
|
return map[string]int64{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Daily buckets use documents_scraped_at only. Tenders present in MinIO but not yet
|
if scope.dailyCounts != nil {
|
||||||
// synced to Mongo are included in the lifetime total but omitted from the daily series.
|
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) {
|
||||||
pipeline := mongodriver.Pipeline{
|
pipeline := mongodriver.Pipeline{
|
||||||
{{Key: "$match", Value: scope.match}},
|
{{Key: "$match", Value: match}},
|
||||||
|
{{Key: "$unwind", Value: "$scraped_documents"}},
|
||||||
{{Key: "$addFields", Value: bson.M{
|
{{Key: "$addFields", Value: bson.M{
|
||||||
"metric_ts": normalizeTimestampExpr("processing_metadata.documents_scraped_at"),
|
"metric_ts": bson.M{
|
||||||
|
"$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},
|
||||||
@@ -257,10 +290,12 @@ func (r *repository) cachedScrapedTendersScope(ctx context.Context) (scrapedTend
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (r *repository) resolveScrapedTendersScope(ctx context.Context) (scrapedTendersScope, error) {
|
func (r *repository) resolveScrapedTendersScope(ctx context.Context) (scrapedTendersScope, error) {
|
||||||
folderIDs, err := r.scrapedDocumentsFolderIDs(ctx)
|
procedures, dailyCounts, err := r.listScrapedProcedures(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return scrapedTendersScope{}, err
|
return scrapedTendersScope{}, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
folderIDs := folderIDsFromProcedures(procedures)
|
||||||
if len(folderIDs) > 0 {
|
if len(folderIDs) > 0 {
|
||||||
if len(folderIDs) > maxScrapedFolderInClause {
|
if len(folderIDs) > maxScrapedFolderInClause {
|
||||||
r.logger.Warn("Too many MinIO procedure folders for dashboard statistics, using Mongo fallback", map[string]interface{}{
|
r.logger.Warn("Too many MinIO procedure folders for dashboard statistics, using Mongo fallback", map[string]interface{}{
|
||||||
@@ -271,21 +306,30 @@ func (r *repository) resolveScrapedTendersScope(ctx context.Context) (scrapedTen
|
|||||||
}
|
}
|
||||||
return scrapedTendersScope{
|
return scrapedTendersScope{
|
||||||
match: bson.M{"contract_folder_id": bson.M{"$in": folderIDs}},
|
match: bson.M{"contract_folder_id": bson.M{"$in": folderIDs}},
|
||||||
|
fromMinIO: true,
|
||||||
|
dailyCounts: dailyCounts,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
return mongoScrapedTendersScope(), nil
|
return mongoScrapedTendersScope(), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *repository) scrapedDocumentsFolderIDs(ctx context.Context) ([]string, error) {
|
func (r *repository) listScrapedProcedures(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, map[string]int64, error) {
|
||||||
if r.procedureLister == nil {
|
if r.procedureLister == nil {
|
||||||
return nil, nil
|
return nil, nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if scanner, ok := r.procedureLister.(scrapedDocumentsScanner); ok {
|
||||||
|
return scanner.ScanScrapedDocuments(ctx)
|
||||||
}
|
}
|
||||||
|
|
||||||
procedures, err := r.procedureLister.ListProceduresWithDocuments(ctx)
|
procedures, err := r.procedureLister.ListProceduresWithDocuments(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, nil, err
|
||||||
}
|
}
|
||||||
|
return procedures, nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func folderIDsFromProcedures(procedures []ai_summarizer.ProcedureDocumentsSummary) []string {
|
||||||
folderIDs := make([]string, 0, len(procedures))
|
folderIDs := make([]string, 0, len(procedures))
|
||||||
seen := make(map[string]struct{}, len(procedures))
|
seen := make(map[string]struct{}, len(procedures))
|
||||||
for _, proc := range procedures {
|
for _, proc := range procedures {
|
||||||
@@ -302,8 +346,19 @@ func (r *repository) scrapedDocumentsFolderIDs(ctx context.Context) ([]string, e
|
|||||||
seen[id] = struct{}{}
|
seen[id] = struct{}{}
|
||||||
folderIDs = append(folderIDs, id)
|
folderIDs = append(folderIDs, id)
|
||||||
}
|
}
|
||||||
|
return folderIDs
|
||||||
|
}
|
||||||
|
|
||||||
return folderIDs, nil
|
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) {
|
func decodeDailyCounts(ctx context.Context, cursor *mongodriver.Cursor) (map[string]int64, error) {
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"tm/pkg/ai_summarizer"
|
"tm/pkg/ai_summarizer"
|
||||||
|
|
||||||
@@ -12,6 +13,7 @@ import (
|
|||||||
|
|
||||||
type mockProcedureDocumentsLister struct {
|
type mockProcedureDocumentsLister struct {
|
||||||
procedures []ai_summarizer.ProcedureDocumentsSummary
|
procedures []ai_summarizer.ProcedureDocumentsSummary
|
||||||
|
dailyCounts map[string]int64
|
||||||
err error
|
err error
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -22,6 +24,13 @@ func (m *mockProcedureDocumentsLister) ListProceduresWithDocuments(context.Conte
|
|||||||
return m.procedures, nil
|
return m.procedures, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (m *mockProcedureDocumentsLister) ScanScrapedDocuments(context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, map[string]int64, error) {
|
||||||
|
if m.err != nil {
|
||||||
|
return nil, nil, m.err
|
||||||
|
}
|
||||||
|
return m.procedures, m.dailyCounts, nil
|
||||||
|
}
|
||||||
|
|
||||||
func TestScrapedDocumentsFolderIDsUsesMinIOProcedures(t *testing.T) {
|
func TestScrapedDocumentsFolderIDsUsesMinIOProcedures(t *testing.T) {
|
||||||
repo := &repository{
|
repo := &repository{
|
||||||
procedureLister: &mockProcedureDocumentsLister{
|
procedureLister: &mockProcedureDocumentsLister{
|
||||||
@@ -91,6 +100,9 @@ func TestResolveScrapedTendersScopeUsesMinIOFolders(t *testing.T) {
|
|||||||
if scope.zero {
|
if scope.zero {
|
||||||
t.Fatal("expected non-zero scope")
|
t.Fatal("expected non-zero scope")
|
||||||
}
|
}
|
||||||
|
if !scope.fromMinIO {
|
||||||
|
t.Fatal("expected MinIO-backed scope")
|
||||||
|
}
|
||||||
in, ok := scope.match["contract_folder_id"].(bson.M)
|
in, ok := scope.match["contract_folder_id"].(bson.M)
|
||||||
if !ok {
|
if !ok {
|
||||||
t.Fatalf("expected contract_folder_id filter, got %#v", scope.match)
|
t.Fatalf("expected contract_folder_id filter, got %#v", scope.match)
|
||||||
@@ -133,6 +145,52 @@ func TestResolveScrapedTendersScopeMongoFallbackWithoutLister(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestFilterDailyCountsSince(t *testing.T) {
|
||||||
|
counts := map[string]int64{
|
||||||
|
"2026-06-01": 2,
|
||||||
|
"2026-06-15": 5,
|
||||||
|
"2026-06-28": 1,
|
||||||
|
}
|
||||||
|
|
||||||
|
startUnix := time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC).Unix()
|
||||||
|
got := filterDailyCountsSince(counts, startUnix)
|
||||||
|
|
||||||
|
want := map[string]int64{
|
||||||
|
"2026-06-15": 5,
|
||||||
|
"2026-06-28": 1,
|
||||||
|
}
|
||||||
|
if len(got) != len(want) {
|
||||||
|
t.Fatalf("expected %d dates, got %d: %v", len(want), len(got), got)
|
||||||
|
}
|
||||||
|
for date, count := range want {
|
||||||
|
if got[date] != count {
|
||||||
|
t.Fatalf("expected %d on %s, got %d", count, date, got[date])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestScrapedDocumentsPerDayUsesMinIODailyCounts(t *testing.T) {
|
||||||
|
repo := &repository{}
|
||||||
|
scope := scrapedTendersScope{
|
||||||
|
dailyCounts: map[string]int64{
|
||||||
|
"2026-06-01": 2,
|
||||||
|
"2026-06-28": 3,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
startUnix := time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC).Unix()
|
||||||
|
|
||||||
|
got, err := repo.scrapedDocumentsPerDay(context.Background(), startUnix, scope)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("unexpected error: %v", err)
|
||||||
|
}
|
||||||
|
if got["2026-06-01"] != 0 {
|
||||||
|
t.Fatalf("expected old date filtered out, got %v", got)
|
||||||
|
}
|
||||||
|
if got["2026-06-28"] != 3 {
|
||||||
|
t.Fatalf("expected 3 on 2026-06-28, got %v", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestMongoScrapedTendersScopeMatchesStatisticsFilter(t *testing.T) {
|
func TestMongoScrapedTendersScopeMatchesStatisticsFilter(t *testing.T) {
|
||||||
scope := mongoScrapedTendersScope()
|
scope := mongoScrapedTendersScope()
|
||||||
if scope.zero {
|
if scope.zero {
|
||||||
|
|||||||
@@ -570,12 +570,20 @@ func (s *StorageClient) ListDocuments(ctx context.Context, contractFolderID, not
|
|||||||
// ListProceduresWithDocuments scans MinIO for PROC_<contractFolderID>/…/documents/* keys
|
// ListProceduresWithDocuments scans MinIO for PROC_<contractFolderID>/…/documents/* keys
|
||||||
// and returns one entry per contract folder id that has at least one document file.
|
// and returns one entry per contract folder id that has at least one document file.
|
||||||
func (s *StorageClient) ListProceduresWithDocuments(ctx context.Context) ([]ProcedureDocumentsSummary, error) {
|
func (s *StorageClient) ListProceduresWithDocuments(ctx context.Context) ([]ProcedureDocumentsSummary, error) {
|
||||||
s.logger.Debug("Listing procedure folders with documents from MinIO", map[string]interface{}{
|
procedures, _, err := s.ScanScrapedDocuments(ctx)
|
||||||
|
return procedures, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// ScanScrapedDocuments scans MinIO once and returns per-procedure summaries plus
|
||||||
|
// daily document counts keyed by UTC date (YYYY-MM-DD).
|
||||||
|
func (s *StorageClient) ScanScrapedDocuments(ctx context.Context) ([]ProcedureDocumentsSummary, map[string]int64, error) {
|
||||||
|
s.logger.Debug("Scanning scraped documents from MinIO", map[string]interface{}{
|
||||||
"bucket": s.config.MinioBucket,
|
"bucket": s.config.MinioBucket,
|
||||||
"prefix": procedurePrefix,
|
"prefix": procedurePrefix,
|
||||||
})
|
})
|
||||||
|
|
||||||
byFolder := make(map[string]*ProcedureDocumentsSummary)
|
byFolder := make(map[string]*ProcedureDocumentsSummary)
|
||||||
|
dailyCounts := make(map[string]int64)
|
||||||
var fatalErr error
|
var fatalErr error
|
||||||
|
|
||||||
appendFromPrefix := func(prefix string) bool {
|
appendFromPrefix := func(prefix string) bool {
|
||||||
@@ -603,24 +611,28 @@ func (s *StorageClient) ListProceduresWithDocuments(ctx context.Context) ([]Proc
|
|||||||
byFolder[contractFolderID] = entry
|
byFolder[contractFolderID] = entry
|
||||||
}
|
}
|
||||||
entry.DocumentCount++
|
entry.DocumentCount++
|
||||||
modified := object.LastModified.Unix()
|
modified := normalizeUnixSeconds(object.LastModified.Unix())
|
||||||
if modified > entry.LatestModified {
|
if modified > entry.LatestModified {
|
||||||
entry.LatestModified = modified
|
entry.LatestModified = modified
|
||||||
}
|
}
|
||||||
|
if modified > 0 {
|
||||||
|
date := time.Unix(modified, 0).UTC().Format("2006-01-02")
|
||||||
|
dailyCounts[date]++
|
||||||
|
}
|
||||||
}
|
}
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
if !appendFromPrefix(procedurePrefix) {
|
if !appendFromPrefix(procedurePrefix) {
|
||||||
return nil, fatalErr
|
return nil, nil, fatalErr
|
||||||
}
|
}
|
||||||
if len(byFolder) == 0 {
|
if len(byFolder) == 0 {
|
||||||
if !appendFromPrefix("") {
|
if !appendFromPrefix("") {
|
||||||
return nil, fatalErr
|
return nil, nil, fatalErr
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if fatalErr != nil {
|
if fatalErr != nil {
|
||||||
return nil, fatalErr
|
return nil, nil, fatalErr
|
||||||
}
|
}
|
||||||
|
|
||||||
results := make([]ProcedureDocumentsSummary, 0, len(byFolder))
|
results := make([]ProcedureDocumentsSummary, 0, len(byFolder))
|
||||||
@@ -634,12 +646,19 @@ func (s *StorageClient) ListProceduresWithDocuments(ctx context.Context) ([]Proc
|
|||||||
return results[i].LatestModified > results[j].LatestModified
|
return results[i].LatestModified > results[j].LatestModified
|
||||||
})
|
})
|
||||||
|
|
||||||
s.logger.Info("Listed procedure folders with documents from storage", map[string]interface{}{
|
s.logger.Info("Scanned scraped documents from storage", map[string]interface{}{
|
||||||
"bucket": s.config.MinioBucket,
|
"bucket": s.config.MinioBucket,
|
||||||
"procedure_count": len(results),
|
"procedure_count": len(results),
|
||||||
})
|
})
|
||||||
|
|
||||||
return results, nil
|
return results, dailyCounts, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func normalizeUnixSeconds(ts int64) int64 {
|
||||||
|
if ts > 1_000_000_000_000 {
|
||||||
|
return ts / 1000
|
||||||
|
}
|
||||||
|
return ts
|
||||||
}
|
}
|
||||||
|
|
||||||
// contractFolderIDFromDocumentKey parses PROC_<contractFolderID>/<notice>/documents/<file>.
|
// contractFolderIDFromDocumentKey parses PROC_<contractFolderID>/<notice>/documents/<file>.
|
||||||
|
|||||||
Reference in New Issue
Block a user