Implement TED Scraper and Configuration Management
- Introduced a new TED scraper in `cmd/scraper` to handle downloading and parsing TED XML files. - Added configuration management for the scraper, including a new `Config` struct and YAML configuration file. - Created necessary handler, service, and repository layers for tender management, adhering to Clean Architecture principles. - Implemented comprehensive logging and error handling throughout the scraper functionality. - Updated API routes to include tender management operations, enhancing the overall system capabilities.
This commit is contained in:
n.nakhostin
@@ -0,0 +1,828 @@
|
||||
package tender
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
"tm/pkg/logger"
|
||||
mongopkg "tm/pkg/mongo"
|
||||
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/bson/primitive"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
)
|
||||
|
||||
// TenderRepository interface defines tender data access methods
|
||||
type TenderRepository interface {
|
||||
// Tender CRUD operations
|
||||
Create(ctx context.Context, tender *Tender) error
|
||||
GetByID(ctx context.Context, id string) (*Tender, error)
|
||||
GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Tender, error)
|
||||
GetByNoticePublicationID(ctx context.Context, noticePublicationID string) (*Tender, error)
|
||||
Update(ctx context.Context, tender *Tender) error
|
||||
Delete(ctx context.Context, id string) error
|
||||
List(ctx context.Context, criteria TenderSearchCriteria, limit, offset int) ([]Tender, int64, error)
|
||||
|
||||
// Bulk operations
|
||||
CreateBatch(ctx context.Context, tenders []Tender) error
|
||||
UpdateBatch(ctx context.Context, tenders []Tender) error
|
||||
|
||||
// Search and filtering
|
||||
Search(ctx context.Context, criteria TenderSearchCriteria) ([]Tender, error)
|
||||
GetByCountryCode(ctx context.Context, countryCode string, limit, offset int) ([]Tender, error)
|
||||
GetByStatus(ctx context.Context, status TenderStatus, limit, offset int) ([]Tender, error)
|
||||
GetExpiredTenders(ctx context.Context, beforeDate int64) ([]Tender, error)
|
||||
GetActiveTenders(ctx context.Context, limit, offset int) ([]Tender, error)
|
||||
|
||||
// Statistics and aggregations
|
||||
GetStatistics(ctx context.Context) (*TenderStatistics, error)
|
||||
GetTenderCountByCountry(ctx context.Context) (map[string]int64, error)
|
||||
GetTenderCountByType(ctx context.Context) (map[string]int64, error)
|
||||
GetTenderCountByClassification(ctx context.Context) (map[string]int64, error)
|
||||
|
||||
// Scraping state management
|
||||
GetScrapingState(ctx context.Context) (*ScrapingState, error)
|
||||
SaveScrapingState(ctx context.Context, state *ScrapingState) error
|
||||
|
||||
// Scraping jobs management
|
||||
CreateScrapingJob(ctx context.Context, job *ScrapingJob) error
|
||||
GetScrapingJob(ctx context.Context, id string) (*ScrapingJob, error)
|
||||
UpdateScrapingJob(ctx context.Context, job *ScrapingJob) error
|
||||
ListScrapingJobs(ctx context.Context, limit, offset int) ([]ScrapingJob, int64, error)
|
||||
GetActiveScrapingJobs(ctx context.Context) ([]ScrapingJob, error)
|
||||
}
|
||||
|
||||
// tenderRepository implements TenderRepository interface using MongoDB ORM
|
||||
type tenderRepository struct {
|
||||
ormRepo mongopkg.Repository[Tender]
|
||||
stateOrmRepo mongopkg.Repository[ScrapingState]
|
||||
jobOrmRepo mongopkg.Repository[ScrapingJob]
|
||||
mongoManager *mongopkg.ConnectionManager
|
||||
logger logger.Logger
|
||||
}
|
||||
|
||||
// NewTenderRepository creates a new tender repository
|
||||
func NewTenderRepository(mongoManager *mongopkg.ConnectionManager, logger logger.Logger) TenderRepository {
|
||||
// Create indexes for tenders collection
|
||||
tenderIndexes := []mongopkg.Index{
|
||||
*mongopkg.CreateUniqueIndex("notice_publication_id_idx", bson.D{{Key: "notice_publication_id", Value: 1}}),
|
||||
*mongopkg.NewIndex("contract_notice_id_idx", bson.D{{Key: "contract_notice_id", Value: 1}}),
|
||||
*mongopkg.NewIndex("status_idx", bson.D{{Key: "status", Value: 1}}),
|
||||
*mongopkg.NewIndex("source_idx", bson.D{{Key: "source", Value: 1}}),
|
||||
*mongopkg.NewIndex("country_code_idx", bson.D{{Key: "country_code", Value: 1}}),
|
||||
*mongopkg.NewIndex("notice_type_code_idx", bson.D{{Key: "notice_type_code", Value: 1}}),
|
||||
*mongopkg.NewIndex("procurement_type_code_idx", bson.D{{Key: "procurement_type_code", Value: 1}}),
|
||||
*mongopkg.NewIndex("main_classification_idx", bson.D{{Key: "main_classification", Value: 1}}),
|
||||
*mongopkg.NewIndex("publication_date_idx", bson.D{{Key: "publication_date", Value: -1}}),
|
||||
*mongopkg.NewIndex("tender_deadline_idx", bson.D{{Key: "tender_deadline", Value: 1}}),
|
||||
*mongopkg.NewIndex("estimated_value_idx", bson.D{{Key: "estimated_value", Value: -1}}),
|
||||
*mongopkg.NewIndex("created_at_idx", bson.D{{Key: "created_at", Value: -1}}),
|
||||
*mongopkg.NewIndex("updated_at_idx", bson.D{{Key: "updated_at", Value: -1}}),
|
||||
// Compound indexes for common queries
|
||||
*mongopkg.NewIndex("status_country_idx", bson.D{
|
||||
{Key: "status", Value: 1},
|
||||
{Key: "country_code", Value: 1},
|
||||
}),
|
||||
*mongopkg.NewIndex("notice_type_status_idx", bson.D{
|
||||
{Key: "notice_type_code", Value: 1},
|
||||
{Key: "status", Value: 1},
|
||||
}),
|
||||
// Text index for search
|
||||
*mongopkg.CreateTextIndex("search_idx", "title", "description"),
|
||||
}
|
||||
|
||||
// Create indexes for scraping jobs collection
|
||||
jobIndexes := []mongopkg.Index{
|
||||
*mongopkg.NewIndex("type_idx", bson.D{{Key: "type", Value: 1}}),
|
||||
*mongopkg.NewIndex("status_idx", bson.D{{Key: "status", Value: 1}}),
|
||||
*mongopkg.NewIndex("started_at_idx", bson.D{{Key: "started_at", Value: -1}}),
|
||||
*mongopkg.NewIndex("type_status_idx", bson.D{
|
||||
{Key: "type", Value: 1},
|
||||
{Key: "status", Value: 1},
|
||||
}),
|
||||
}
|
||||
|
||||
// Create indexes
|
||||
if err := mongoManager.CreateIndexes("tenders", tenderIndexes); err != nil {
|
||||
logger.Warn("Failed to create tender indexes", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
|
||||
if err := mongoManager.CreateIndexes("scraping_jobs", jobIndexes); err != nil {
|
||||
logger.Warn("Failed to create scraping job indexes", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
|
||||
// Create ORM repositories
|
||||
tenderOrmRepo := mongopkg.NewRepository[Tender](mongoManager.GetCollection("tenders"), logger)
|
||||
stateOrmRepo := mongopkg.NewRepository[ScrapingState](mongoManager.GetCollection("scraping_state"), logger)
|
||||
jobOrmRepo := mongopkg.NewRepository[ScrapingJob](mongoManager.GetCollection("scraping_jobs"), logger)
|
||||
|
||||
return &tenderRepository{
|
||||
ormRepo: tenderOrmRepo,
|
||||
stateOrmRepo: stateOrmRepo,
|
||||
jobOrmRepo: jobOrmRepo,
|
||||
mongoManager: mongoManager,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// Create creates a new tender
|
||||
func (r *tenderRepository) Create(ctx context.Context, tender *Tender) error {
|
||||
err := r.ormRepo.Create(ctx, tender)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to create tender", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
"notice_publication_id": tender.NoticePublicationID,
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
r.logger.Info("Tender created successfully", map[string]interface{}{
|
||||
"tender_id": tender.ID,
|
||||
"notice_publication_id": tender.NoticePublicationID,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetByID retrieves a tender by ID
|
||||
func (r *tenderRepository) GetByID(ctx context.Context, id string) (*Tender, error) {
|
||||
tender, err := r.ormRepo.FindByID(ctx, id)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get tender by ID", map[string]interface{}{
|
||||
"tender_id": id,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return tender, nil
|
||||
}
|
||||
|
||||
// GetByContractNoticeID retrieves a tender by contract notice ID
|
||||
func (r *tenderRepository) GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Tender, error) {
|
||||
filter := bson.M{"contract_notice_id": contractNoticeID}
|
||||
pagination := mongopkg.Pagination{Limit: 1}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get tender by contract notice ID", map[string]interface{}{
|
||||
"contract_notice_id": contractNoticeID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if len(result.Items) == 0 {
|
||||
return nil, mongopkg.ErrDocumentNotFound
|
||||
}
|
||||
|
||||
return &result.Items[0], nil
|
||||
}
|
||||
|
||||
// GetByNoticePublicationID retrieves a tender by notice publication ID
|
||||
func (r *tenderRepository) GetByNoticePublicationID(ctx context.Context, noticePublicationID string) (*Tender, error) {
|
||||
filter := bson.M{"notice_publication_id": noticePublicationID}
|
||||
pagination := mongopkg.Pagination{Limit: 1}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get tender by notice publication ID", map[string]interface{}{
|
||||
"notice_publication_id": noticePublicationID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if len(result.Items) == 0 {
|
||||
return nil, mongopkg.ErrDocumentNotFound
|
||||
}
|
||||
|
||||
return &result.Items[0], nil
|
||||
}
|
||||
|
||||
// Update updates an existing tender
|
||||
func (r *tenderRepository) Update(ctx context.Context, tender *Tender) error {
|
||||
tender.UpdatedAt = time.Now().Unix()
|
||||
|
||||
err := r.ormRepo.Update(ctx, tender)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to update tender", map[string]interface{}{
|
||||
"tender_id": tender.ID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
r.logger.Info("Tender updated successfully", map[string]interface{}{
|
||||
"tender_id": tender.ID,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Delete deletes a tender by ID
|
||||
func (r *tenderRepository) Delete(ctx context.Context, id string) error {
|
||||
err := r.ormRepo.Delete(ctx, id)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to delete tender", map[string]interface{}{
|
||||
"tender_id": id,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
r.logger.Info("Tender deleted successfully", map[string]interface{}{
|
||||
"tender_id": id,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// List retrieves tenders with pagination and filtering
|
||||
func (r *tenderRepository) List(ctx context.Context, criteria TenderSearchCriteria, limit, offset int) ([]Tender, int64, error) {
|
||||
filter := r.buildSearchFilter(criteria)
|
||||
|
||||
pagination := mongopkg.Pagination{
|
||||
Limit: limit,
|
||||
Skip: offset,
|
||||
SortField: "created_at",
|
||||
SortOrder: -1,
|
||||
}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to list tenders", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
return result.Items, result.TotalCount, nil
|
||||
}
|
||||
|
||||
// CreateBatch creates multiple tenders in a single operation
|
||||
func (r *tenderRepository) CreateBatch(ctx context.Context, tenders []Tender) error {
|
||||
if len(tenders) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
for i := range tenders {
|
||||
if err := r.ormRepo.Create(ctx, &tenders[i]); err != nil {
|
||||
r.logger.Error("Failed to create tender in batch", map[string]interface{}{
|
||||
"tender_id": tenders[i].ID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
r.logger.Info("Tender batch created successfully", map[string]interface{}{
|
||||
"count": len(tenders),
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// UpdateBatch updates multiple tenders
|
||||
func (r *tenderRepository) UpdateBatch(ctx context.Context, tenders []Tender) error {
|
||||
if len(tenders) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
now := time.Now().Unix()
|
||||
|
||||
for i := range tenders {
|
||||
tenders[i].UpdatedAt = now
|
||||
|
||||
if err := r.ormRepo.Update(ctx, &tenders[i]); err != nil {
|
||||
r.logger.Error("Failed to update tender in batch", map[string]interface{}{
|
||||
"tender_id": tenders[i].ID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
r.logger.Info("Tender batch updated successfully", map[string]interface{}{
|
||||
"count": len(tenders),
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Search searches tenders based on criteria
|
||||
func (r *tenderRepository) Search(ctx context.Context, criteria TenderSearchCriteria) ([]Tender, error) {
|
||||
filter := r.buildSearchFilter(criteria)
|
||||
|
||||
pagination := mongopkg.Pagination{
|
||||
Limit: 100,
|
||||
SortField: "created_at",
|
||||
SortOrder: -1,
|
||||
}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to search tenders", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return result.Items, nil
|
||||
}
|
||||
|
||||
// GetByCountryCode retrieves tenders by country code
|
||||
func (r *tenderRepository) GetByCountryCode(ctx context.Context, countryCode string, limit, offset int) ([]Tender, error) {
|
||||
filter := bson.M{"country_code": countryCode}
|
||||
|
||||
pagination := mongopkg.Pagination{
|
||||
Limit: limit,
|
||||
Skip: offset,
|
||||
SortField: "created_at",
|
||||
SortOrder: -1,
|
||||
}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get tenders by country code", map[string]interface{}{
|
||||
"country_code": countryCode,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return result.Items, nil
|
||||
}
|
||||
|
||||
// GetByStatus retrieves tenders by status
|
||||
func (r *tenderRepository) GetByStatus(ctx context.Context, status TenderStatus, limit, offset int) ([]Tender, error) {
|
||||
filter := bson.M{"status": status}
|
||||
|
||||
pagination := mongopkg.Pagination{
|
||||
Limit: limit,
|
||||
Skip: offset,
|
||||
SortField: "created_at",
|
||||
SortOrder: -1,
|
||||
}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get tenders by status", map[string]interface{}{
|
||||
"status": status,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return result.Items, nil
|
||||
}
|
||||
|
||||
// GetExpiredTenders retrieves tenders that are expired
|
||||
func (r *tenderRepository) GetExpiredTenders(ctx context.Context, beforeDate int64) ([]Tender, error) {
|
||||
filter := bson.M{
|
||||
"$or": []bson.M{
|
||||
{"status": TenderStatusExpired},
|
||||
{
|
||||
"tender_deadline": bson.M{
|
||||
"$lt": beforeDate,
|
||||
},
|
||||
"status": bson.M{"$ne": TenderStatusExpired},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
pagination := mongopkg.Pagination{
|
||||
Limit: 1000, // Large limit for maintenance operations
|
||||
SortField: "tender_deadline",
|
||||
SortOrder: 1,
|
||||
}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get expired tenders", map[string]interface{}{
|
||||
"before_date": beforeDate,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return result.Items, nil
|
||||
}
|
||||
|
||||
// GetActiveTenders retrieves active tenders
|
||||
func (r *tenderRepository) GetActiveTenders(ctx context.Context, limit, offset int) ([]Tender, error) {
|
||||
filter := bson.M{"status": TenderStatusActive}
|
||||
|
||||
pagination := mongopkg.Pagination{
|
||||
Limit: limit,
|
||||
Skip: offset,
|
||||
SortField: "tender_deadline",
|
||||
SortOrder: 1,
|
||||
}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get active tenders", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return result.Items, nil
|
||||
}
|
||||
|
||||
// GetStatistics calculates tender statistics using aggregation
|
||||
func (r *tenderRepository) GetStatistics(ctx context.Context) (*TenderStatistics, error) {
|
||||
stats := &TenderStatistics{
|
||||
LastUpdated: time.Now().Unix(),
|
||||
}
|
||||
|
||||
// Get total count
|
||||
total, err := r.ormRepo.Count(ctx, bson.M{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stats.TotalTenders = total
|
||||
|
||||
// Get active count
|
||||
active, err := r.ormRepo.Count(ctx, bson.M{"status": TenderStatusActive})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stats.ActiveTenders = active
|
||||
|
||||
// Get expired count
|
||||
expired, err := r.ormRepo.Count(ctx, bson.M{"status": TenderStatusExpired})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stats.ExpiredTenders = expired
|
||||
|
||||
// Get statistics by country, type, and classification
|
||||
stats.TendersByCountry, _ = r.GetTenderCountByCountry(ctx)
|
||||
stats.TendersByType, _ = r.GetTenderCountByType(ctx)
|
||||
stats.TendersByClassification, _ = r.GetTenderCountByClassification(ctx)
|
||||
|
||||
// Calculate value statistics using aggregation
|
||||
pipeline := mongo.Pipeline{
|
||||
{
|
||||
{Key: "$group", Value: bson.M{
|
||||
"_id": nil,
|
||||
"total_estimated_value": bson.M{"$sum": "$estimated_value"},
|
||||
"count": bson.M{"$sum": 1},
|
||||
}},
|
||||
},
|
||||
}
|
||||
|
||||
result, err := r.ormRepo.Aggregate(ctx, pipeline)
|
||||
if err == nil && len(result) > 0 {
|
||||
if totalValue, ok := result[0]["total_estimated_value"].(float64); ok {
|
||||
stats.TotalEstimatedValue = totalValue
|
||||
if count, ok := result[0]["count"].(int32); ok && count > 0 {
|
||||
stats.AverageEstimatedValue = totalValue / float64(count)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
// GetTenderCountByCountry gets tender count grouped by country
|
||||
func (r *tenderRepository) GetTenderCountByCountry(ctx context.Context) (map[string]int64, error) {
|
||||
pipeline := mongo.Pipeline{
|
||||
{
|
||||
{Key: "$group", Value: bson.M{
|
||||
"_id": "$country_code",
|
||||
"count": bson.M{"$sum": 1},
|
||||
}},
|
||||
},
|
||||
}
|
||||
|
||||
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
counts := make(map[string]int64)
|
||||
for _, result := range results {
|
||||
if country, ok := result["_id"].(string); ok {
|
||||
if count, ok := result["count"].(int32); ok {
|
||||
counts[country] = int64(count)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return counts, nil
|
||||
}
|
||||
|
||||
// GetTenderCountByType gets tender count grouped by type
|
||||
func (r *tenderRepository) GetTenderCountByType(ctx context.Context) (map[string]int64, error) {
|
||||
pipeline := mongo.Pipeline{
|
||||
{
|
||||
{Key: "$group", Value: bson.M{
|
||||
"_id": "$notice_type_code",
|
||||
"count": bson.M{"$sum": 1},
|
||||
}},
|
||||
},
|
||||
}
|
||||
|
||||
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
counts := make(map[string]int64)
|
||||
for _, result := range results {
|
||||
if noticeType, ok := result["_id"].(string); ok {
|
||||
if count, ok := result["count"].(int32); ok {
|
||||
counts[noticeType] = int64(count)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return counts, nil
|
||||
}
|
||||
|
||||
// GetTenderCountByClassification gets tender count grouped by classification
|
||||
func (r *tenderRepository) GetTenderCountByClassification(ctx context.Context) (map[string]int64, error) {
|
||||
pipeline := mongo.Pipeline{
|
||||
{
|
||||
{Key: "$group", Value: bson.M{
|
||||
"_id": "$main_classification",
|
||||
"count": bson.M{"$sum": 1},
|
||||
}},
|
||||
},
|
||||
{
|
||||
{Key: "$limit", Value: 20},
|
||||
},
|
||||
}
|
||||
|
||||
results, err := r.ormRepo.Aggregate(ctx, pipeline)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
counts := make(map[string]int64)
|
||||
for _, result := range results {
|
||||
if classification, ok := result["_id"].(string); ok {
|
||||
if count, ok := result["count"].(int32); ok {
|
||||
counts[classification] = int64(count)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return counts, nil
|
||||
}
|
||||
|
||||
// GetScrapingState retrieves the current scraping state
|
||||
func (r *tenderRepository) GetScrapingState(ctx context.Context) (*ScrapingState, error) {
|
||||
// Try to find existing state
|
||||
filter := bson.M{}
|
||||
pagination := mongopkg.Pagination{Limit: 1}
|
||||
|
||||
result, err := r.stateOrmRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if len(result.Items) == 0 {
|
||||
// Return default state if none exists
|
||||
return &ScrapingState{
|
||||
ID: "default",
|
||||
LastProcessedYear: time.Now().Year(),
|
||||
LastProcessedNumber: 0,
|
||||
LastRunAt: 0,
|
||||
NextRunAt: time.Now().Add(4 * time.Hour).Unix(),
|
||||
IsRunning: false,
|
||||
CreatedAt: time.Now().Unix(),
|
||||
UpdatedAt: time.Now().Unix(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
return &result.Items[0], nil
|
||||
}
|
||||
|
||||
// SaveScrapingState saves the scraping state
|
||||
func (r *tenderRepository) SaveScrapingState(ctx context.Context, state *ScrapingState) error {
|
||||
if state.ID == "" {
|
||||
state.ID = "default"
|
||||
}
|
||||
|
||||
now := time.Now().Unix()
|
||||
if state.CreatedAt == 0 {
|
||||
state.CreatedAt = now
|
||||
}
|
||||
state.UpdatedAt = now
|
||||
|
||||
// Try to update existing state first
|
||||
existing, err := r.GetScrapingState(ctx)
|
||||
if err == nil && existing.ID != "" {
|
||||
state.ID = existing.ID
|
||||
state.CreatedAt = existing.CreatedAt
|
||||
return r.stateOrmRepo.Update(ctx, state)
|
||||
}
|
||||
|
||||
// Create new state if none exists
|
||||
return r.stateOrmRepo.Create(ctx, state)
|
||||
}
|
||||
|
||||
// CreateScrapingJob creates a new scraping job
|
||||
func (r *tenderRepository) CreateScrapingJob(ctx context.Context, job *ScrapingJob) error {
|
||||
if job.ID == "" {
|
||||
job.ID = primitive.NewObjectID().Hex()
|
||||
}
|
||||
|
||||
now := time.Now().Unix()
|
||||
job.CreatedAt = now
|
||||
job.UpdatedAt = now
|
||||
|
||||
err := r.jobOrmRepo.Create(ctx, job)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to create scraping job", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
"type": job.Type,
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
r.logger.Info("Scraping job created successfully", map[string]interface{}{
|
||||
"job_id": job.ID,
|
||||
"type": job.Type,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetScrapingJob retrieves a scraping job by ID
|
||||
func (r *tenderRepository) GetScrapingJob(ctx context.Context, id string) (*ScrapingJob, error) {
|
||||
job, err := r.jobOrmRepo.FindByID(ctx, id)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get scraping job", map[string]interface{}{
|
||||
"job_id": id,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return job, nil
|
||||
}
|
||||
|
||||
// UpdateScrapingJob updates a scraping job
|
||||
func (r *tenderRepository) UpdateScrapingJob(ctx context.Context, job *ScrapingJob) error {
|
||||
job.UpdatedAt = time.Now().Unix()
|
||||
|
||||
err := r.jobOrmRepo.Update(ctx, job)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to update scraping job", map[string]interface{}{
|
||||
"job_id": job.ID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
r.logger.Info("Scraping job updated successfully", map[string]interface{}{
|
||||
"job_id": job.ID,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// ListScrapingJobs retrieves scraping jobs with pagination
|
||||
func (r *tenderRepository) ListScrapingJobs(ctx context.Context, limit, offset int) ([]ScrapingJob, int64, error) {
|
||||
filter := bson.M{}
|
||||
|
||||
pagination := mongopkg.Pagination{
|
||||
Limit: limit,
|
||||
Skip: offset,
|
||||
SortField: "started_at",
|
||||
SortOrder: -1,
|
||||
}
|
||||
|
||||
result, err := r.jobOrmRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to list scraping jobs", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
return result.Items, result.TotalCount, nil
|
||||
}
|
||||
|
||||
// GetActiveScrapingJobs retrieves currently active scraping jobs
|
||||
func (r *tenderRepository) GetActiveScrapingJobs(ctx context.Context) ([]ScrapingJob, error) {
|
||||
filter := bson.M{
|
||||
"status": bson.M{
|
||||
"$in": []ScrapingJobStatus{
|
||||
ScrapingJobStatusPending,
|
||||
ScrapingJobStatusRunning,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
pagination := mongopkg.Pagination{
|
||||
Limit: 100,
|
||||
SortField: "started_at",
|
||||
SortOrder: -1,
|
||||
}
|
||||
|
||||
result, err := r.jobOrmRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get active scraping jobs", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return result.Items, nil
|
||||
}
|
||||
|
||||
// buildSearchFilter builds MongoDB filter from search criteria
|
||||
func (r *tenderRepository) buildSearchFilter(criteria TenderSearchCriteria) bson.M {
|
||||
filter := bson.M{}
|
||||
|
||||
if criteria.Query != "" {
|
||||
filter["$text"] = bson.M{"$search": criteria.Query}
|
||||
}
|
||||
|
||||
if criteria.NoticeType != "" {
|
||||
filter["notice_type_code"] = criteria.NoticeType
|
||||
}
|
||||
|
||||
if criteria.ProcurementType != "" {
|
||||
filter["procurement_type_code"] = criteria.ProcurementType
|
||||
}
|
||||
|
||||
if len(criteria.CountryCodes) > 0 {
|
||||
filter["country_code"] = bson.M{"$in": criteria.CountryCodes}
|
||||
}
|
||||
|
||||
if len(criteria.RegionCodes) > 0 {
|
||||
filter["region_code"] = bson.M{"$in": criteria.RegionCodes}
|
||||
}
|
||||
|
||||
if len(criteria.Classifications) > 0 {
|
||||
filter["$or"] = []bson.M{
|
||||
{"main_classification": bson.M{"$in": criteria.Classifications}},
|
||||
{"additional_classifications": bson.M{"$in": criteria.Classifications}},
|
||||
}
|
||||
}
|
||||
|
||||
if criteria.MinEstimatedValue != nil || criteria.MaxEstimatedValue != nil {
|
||||
valueFilter := bson.M{}
|
||||
if criteria.MinEstimatedValue != nil {
|
||||
valueFilter["$gte"] = *criteria.MinEstimatedValue
|
||||
}
|
||||
if criteria.MaxEstimatedValue != nil {
|
||||
valueFilter["$lte"] = *criteria.MaxEstimatedValue
|
||||
}
|
||||
filter["estimated_value"] = valueFilter
|
||||
}
|
||||
|
||||
if criteria.Currency != "" {
|
||||
filter["currency"] = criteria.Currency
|
||||
}
|
||||
|
||||
if criteria.DeadlineFrom != nil || criteria.DeadlineTo != nil {
|
||||
deadlineFilter := bson.M{}
|
||||
if criteria.DeadlineFrom != nil {
|
||||
deadlineFilter["$gte"] = *criteria.DeadlineFrom
|
||||
}
|
||||
if criteria.DeadlineTo != nil {
|
||||
deadlineFilter["$lte"] = *criteria.DeadlineTo
|
||||
}
|
||||
filter["tender_deadline"] = deadlineFilter
|
||||
}
|
||||
|
||||
if criteria.PublicationDateFrom != nil || criteria.PublicationDateTo != nil {
|
||||
pubDateFilter := bson.M{}
|
||||
if criteria.PublicationDateFrom != nil {
|
||||
pubDateFilter["$gte"] = *criteria.PublicationDateFrom
|
||||
}
|
||||
if criteria.PublicationDateTo != nil {
|
||||
pubDateFilter["$lte"] = *criteria.PublicationDateTo
|
||||
}
|
||||
filter["publication_date"] = pubDateFilter
|
||||
}
|
||||
|
||||
if len(criteria.Status) > 0 {
|
||||
filter["status"] = bson.M{"$in": criteria.Status}
|
||||
}
|
||||
|
||||
if len(criteria.Source) > 0 {
|
||||
filter["source"] = bson.M{"$in": criteria.Source}
|
||||
}
|
||||
|
||||
if criteria.BuyerOrganizationID != "" {
|
||||
filter["buyer_organization.id"] = criteria.BuyerOrganizationID
|
||||
}
|
||||
|
||||
if len(criteria.Languages) > 0 {
|
||||
filter["notice_language_code"] = bson.M{"$in": criteria.Languages}
|
||||
}
|
||||
|
||||
return filter
|
||||
}
|
||||
Reference in New Issue
Block a user