Merge pull request 'Implemented the integration with scraper python server, tenders summarise, and some fixes.' (#14) from integration into develop
Reviewed-on: https://repo.ravanertebat.com/TM/tm_back/pulls/14 Reviewed-by: Nima Nakhsotin <n.nakhostin@ravanertebat.com>
This commit is contained in:
@@ -24,9 +24,6 @@ type Config struct {
|
||||
type ScraperConfig struct {
|
||||
BaseURL string `env:"SCRAPER_BASE_URL"`
|
||||
Timeout time.Duration `env:"SCRAPER_TIMEOUT"`
|
||||
Username string `env:"SCRAPER_USERNAME"`
|
||||
Password string `env:"SCRAPER_PASSWORD"`
|
||||
LoginURL string `env:"SCRAPER_LOGIN_URL"`
|
||||
}
|
||||
|
||||
type AuthConfig struct {
|
||||
|
||||
+1
-1
@@ -185,7 +185,7 @@ func main() {
|
||||
notificationService := notification.NewService(notificationSDK, userService, customerService, logger)
|
||||
contactService := contact.NewService(contactRepository, logger, notificationSDK)
|
||||
cmsService := cms.NewService(cmsRepository, logger)
|
||||
scraperService := scraper.NewService(conf.Scraper.BaseURL, conf.Scraper.Timeout, logger, conf.Scraper.Username, conf.Scraper.Password, conf.Scraper.LoginURL)
|
||||
scraperService := scraper.NewService(conf.Scraper.BaseURL, conf.Scraper.Timeout, logger)
|
||||
logger.Info("Services initialized successfully", map[string]interface{}{
|
||||
"services": []string{"customer", "user", "company", "category", "tender", "feedback", "tender_approval", "inquiry", "flag", "notification", "contact", "cms", "scraper"},
|
||||
})
|
||||
|
||||
@@ -218,6 +218,7 @@ func RegisterPublicRoutes(e *echo.Echo, customerHandler *customer.Handler, tende
|
||||
tendersGP.GET("", tenderHandler.GetPublicTenders)
|
||||
tendersGP.GET("/recommend", tenderHandler.RecommendTenders)
|
||||
tendersGP.GET("/details/:id", tenderHandler.GetPublicTenderDetails)
|
||||
tendersGP.GET("/:id/document-summaries", tenderHandler.GetDocumentSummaries)
|
||||
}
|
||||
|
||||
// Public Feedback Routes
|
||||
|
||||
@@ -9,10 +9,12 @@ import (
|
||||
"tm/pkg/config"
|
||||
"tm/pkg/glm"
|
||||
"tm/pkg/logger"
|
||||
"tm/pkg/minio"
|
||||
"tm/pkg/mongo"
|
||||
"tm/pkg/notification"
|
||||
"tm/pkg/ollama"
|
||||
"tm/pkg/schedule"
|
||||
"tm/pkg/scraper"
|
||||
)
|
||||
|
||||
// Init Application Configuration
|
||||
@@ -75,24 +77,61 @@ func InitMongoDB(conf config.MongoConfig, log logger.Logger) *mongo.ConnectionMa
|
||||
}
|
||||
|
||||
// Init Worker
|
||||
func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, glmService *glm.SDK) {
|
||||
func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, glmService *glm.SDK, scraperService scraper.SDK, minioService *minio.Service) {
|
||||
// Initialize repositories
|
||||
noticeRepo := notice.NewRepository(mongoManager, appLogger)
|
||||
tenderRepo := tender.NewRepository(mongoManager, appLogger)
|
||||
|
||||
worker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService)
|
||||
worker.Run()
|
||||
// Initialize Notice Worker
|
||||
appLogger.Info("Starting notice worker", map[string]interface{}{})
|
||||
noticeWorker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, scraperService)
|
||||
noticeWorker.Run()
|
||||
|
||||
// start Worker job
|
||||
// Initialize Document Scraper Worker
|
||||
// appLogger.Info("Initializing document scraper worker", map[string]interface{}{})
|
||||
// scraperWorker := workers.NewDocumentScraperWorker(mongoManager, appLogger, tenderRepo, scraperService)
|
||||
// appLogger.Info("Starting document scraper worker", map[string]interface{}{})
|
||||
// scraperWorker.Run()
|
||||
|
||||
// Initialize Document Summarization Worker (only if MinIO is available)
|
||||
if minioService != nil {
|
||||
appLogger.Info("Starting document summarization worker", map[string]interface{}{})
|
||||
summarizerWorker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, minioService, glmService)
|
||||
summarizerWorker.Run()
|
||||
|
||||
// Schedule Document Summarization Worker job
|
||||
schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{
|
||||
Name: "Worker Job",
|
||||
Name: "Document Summarization Worker Job",
|
||||
Func: func() {
|
||||
worker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService)
|
||||
worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, minioService, glmService)
|
||||
worker.Run()
|
||||
},
|
||||
Expr: "0 11 * * * *", // Run at 11 AM every day (1 hour after document scraping)
|
||||
})
|
||||
} else {
|
||||
appLogger.Warn("MinIO service not available, document summarization will be skipped", map[string]interface{}{})
|
||||
}
|
||||
|
||||
// Schedule Notice Worker job
|
||||
schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{
|
||||
Name: "Notice Worker Job",
|
||||
Func: func() {
|
||||
worker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, scraperService)
|
||||
worker.Run()
|
||||
},
|
||||
Expr: config.Worker.Interval,
|
||||
})
|
||||
|
||||
// Schedule Document Scraper Worker job (runs after notice worker)
|
||||
schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{
|
||||
Name: "Document Scraper Worker Job",
|
||||
Func: func() {
|
||||
worker := workers.NewDocumentScraperWorker(mongoManager, appLogger, tenderRepo, scraperService)
|
||||
worker.Run()
|
||||
},
|
||||
Expr: "0 10 * * * *", // Run at 10 AM every day (1 hour after notice worker)
|
||||
})
|
||||
|
||||
}
|
||||
|
||||
// Init Notification Service
|
||||
@@ -194,3 +233,62 @@ func InitGLMService(conf GLMConfig, log logger.Logger) *glm.SDK {
|
||||
|
||||
return glmSDK
|
||||
}
|
||||
|
||||
// Init MinIO Service
|
||||
func InitMinIOService(conf config.MinIOConfig, log logger.Logger) *minio.Service {
|
||||
// Convert bootstrap config to minio config
|
||||
minioConfig := minio.Config{
|
||||
Endpoint: conf.Endpoint,
|
||||
AccessKeyID: conf.AccessKeyID,
|
||||
SecretAccessKey: conf.SecretAccessKey,
|
||||
UseSSL: conf.UseSSL,
|
||||
Region: conf.Region,
|
||||
DefaultBucket: conf.DefaultBucket,
|
||||
HierarchicalBucket: conf.HierarchicalBucket,
|
||||
}
|
||||
|
||||
// Validate config
|
||||
if err := minioConfig.Validate(); err != nil {
|
||||
log.Warn("Invalid MinIO configuration, MinIO features will be disabled", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil // Return nil instead of panicking
|
||||
}
|
||||
|
||||
// Create connection manager
|
||||
connManager, err := minio.NewConnectionManager(minioConfig, log)
|
||||
if err != nil {
|
||||
log.Error("Failed to create MinIO connection manager", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
panic(fmt.Sprintf("Failed to create MinIO connection manager: %v", err))
|
||||
}
|
||||
|
||||
// Connect to MinIO
|
||||
if err := connManager.Connect(); err != nil {
|
||||
log.Warn("Failed to connect to MinIO, MinIO features will be disabled", map[string]interface{}{
|
||||
"endpoint": minioConfig.Endpoint,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil // Return nil instead of panicking
|
||||
}
|
||||
|
||||
// Create service
|
||||
service := minio.NewService(connManager, minioConfig, log)
|
||||
|
||||
log.Info("MinIO service initialized successfully", map[string]interface{}{
|
||||
"endpoint": conf.Endpoint,
|
||||
"default_bucket": conf.DefaultBucket,
|
||||
"hierarchical_bucket": conf.HierarchicalBucket,
|
||||
})
|
||||
|
||||
return service
|
||||
}
|
||||
|
||||
func InitScraperService(config Config, logger logger.Logger) scraper.SDK {
|
||||
return scraper.NewClient(
|
||||
config.Scraper.BaseURL,
|
||||
config.Scraper.Timeout,
|
||||
logger,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -5,7 +5,7 @@ import (
|
||||
"tm/pkg/config"
|
||||
)
|
||||
|
||||
// Config defines the scraper application configuration
|
||||
// Config defines the worker application configuration
|
||||
type Config struct {
|
||||
Database config.DatabaseConfig
|
||||
Logging config.LoggingConfig
|
||||
@@ -13,6 +13,8 @@ type Config struct {
|
||||
Ollama config.OllamaConfig
|
||||
Worker WorkerConfig
|
||||
GLM GLMConfig
|
||||
Scraper config.ScraperConfig
|
||||
MinIO config.MinIOConfig
|
||||
AlertMail string `env:"ALERT_MAIL" envDefault:"alerts@opplens.com"`
|
||||
}
|
||||
|
||||
|
||||
+10
-1
@@ -40,8 +40,17 @@ func main() {
|
||||
// Initialize GLM service
|
||||
glmService := bootstrap.InitGLMService(config.GLM, appLogger)
|
||||
|
||||
// Initialize scraper service
|
||||
scraperService := bootstrap.InitScraperService(*config, appLogger)
|
||||
|
||||
// Initialize MinIO service
|
||||
minioService := bootstrap.InitMinIOService(config.MinIO, appLogger)
|
||||
if minioService != nil {
|
||||
appLogger.Info("MinIO service initialized successfully", map[string]interface{}{})
|
||||
}
|
||||
|
||||
// Initialize Worker
|
||||
bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, glmService)
|
||||
bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, glmService, scraperService, minioService)
|
||||
|
||||
// Set up signal handling for graceful shutdown
|
||||
signalChan := make(chan os.Signal, 1)
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"tm/pkg/logger"
|
||||
"tm/pkg/mongo"
|
||||
"tm/pkg/notification"
|
||||
"tm/pkg/scraper"
|
||||
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
mongodriver "go.mongodb.org/mongo-driver/v2/mongo"
|
||||
@@ -23,9 +24,10 @@ type NoticeWorker struct {
|
||||
NoticeRepo notice.Repository
|
||||
TenderRepo tender.TenderRepository
|
||||
GLM *glm.SDK
|
||||
Scraper scraper.SDK
|
||||
}
|
||||
|
||||
func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, glmService *glm.SDK) *NoticeWorker {
|
||||
func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, glmService *glm.SDK, scraper scraper.SDK) *NoticeWorker {
|
||||
return &NoticeWorker{
|
||||
Mongo: mongo,
|
||||
Logger: logger,
|
||||
@@ -33,15 +35,74 @@ func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notif
|
||||
NoticeRepo: noticeRepo,
|
||||
TenderRepo: tenderRepo,
|
||||
GLM: glmService,
|
||||
Scraper: scraper,
|
||||
}
|
||||
}
|
||||
|
||||
// cleanupProcessedNotices removes notices that have been successfully processed
|
||||
func (w *NoticeWorker) cleanupProcessedNotices() {
|
||||
w.Logger.Info("Starting cleanup of processed notices", map[string]interface{}{})
|
||||
|
||||
// Get all processed notices (limit to avoid memory issues)
|
||||
limit := 100
|
||||
skip := 0
|
||||
|
||||
for {
|
||||
notices, _, err := w.NoticeRepo.GetProcessedNotices(context.Background(), limit, skip)
|
||||
if err != nil {
|
||||
w.Logger.Error("Failed to get processed notices for cleanup", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
break
|
||||
}
|
||||
|
||||
if len(notices) == 0 {
|
||||
break
|
||||
}
|
||||
|
||||
deletedCount := 0
|
||||
for _, notice := range notices {
|
||||
if err := w.NoticeRepo.Delete(context.Background(), notice.ID.Hex()); err != nil {
|
||||
w.Logger.Warn("Failed to delete processed notice during cleanup", map[string]interface{}{
|
||||
"notice_id": notice.ID.Hex(),
|
||||
"error": err.Error(),
|
||||
})
|
||||
} else {
|
||||
deletedCount++
|
||||
}
|
||||
}
|
||||
|
||||
w.Logger.Info("Cleanup batch completed", map[string]interface{}{
|
||||
"processed": len(notices),
|
||||
"deleted": deletedCount,
|
||||
"skip": skip,
|
||||
})
|
||||
|
||||
skip += limit
|
||||
|
||||
// Safety limit to avoid infinite loops
|
||||
if skip > 10000 {
|
||||
w.Logger.Warn("Cleanup reached safety limit, stopping", map[string]interface{}{
|
||||
"skip": skip,
|
||||
})
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
w.Logger.Info("Notice cleanup completed", map[string]interface{}{})
|
||||
}
|
||||
|
||||
func (w *NoticeWorker) Run() {
|
||||
w.Logger.Info("Notice worker started", map[string]interface{}{})
|
||||
|
||||
w.cleanupProcessedNotices()
|
||||
|
||||
limit := 10
|
||||
skip := 0
|
||||
for {
|
||||
processedCount := 0
|
||||
maxToProcess := 5 // Limit processing to allow scraper worker to run
|
||||
|
||||
for processedCount < maxToProcess {
|
||||
notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), limit, skip)
|
||||
if err != nil {
|
||||
w.Logger.Error("Failed to get notices", map[string]interface{}{
|
||||
@@ -51,6 +112,9 @@ func (w *NoticeWorker) Run() {
|
||||
}
|
||||
|
||||
if len(notices) == 0 {
|
||||
w.Logger.Info("No more unprocessed notices found", map[string]interface{}{
|
||||
"processed_count": processedCount,
|
||||
})
|
||||
break
|
||||
}
|
||||
|
||||
@@ -89,15 +153,98 @@ func (w *NoticeWorker) Run() {
|
||||
w.Logger.Error("Failed to update notice", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
w.Logger.Info("Notice processed", map[string]interface{}{
|
||||
} else {
|
||||
processedCount++
|
||||
w.Logger.Info("Notice processed successfully", map[string]interface{}{
|
||||
"notice": n.ID.Hex(),
|
||||
"processed_count": processedCount,
|
||||
})
|
||||
|
||||
// Clean up: Delete processed notice from MongoDB
|
||||
if err := w.NoticeRepo.Delete(context.Background(), n.ID.Hex()); err != nil {
|
||||
w.Logger.Warn("Failed to delete processed notice", map[string]interface{}{
|
||||
"notice_id": n.ID.Hex(),
|
||||
"error": err.Error(),
|
||||
})
|
||||
} else {
|
||||
w.Logger.Debug("Successfully deleted processed notice", map[string]interface{}{
|
||||
"notice_id": n.ID.Hex(),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
skip += limit
|
||||
|
||||
// Check if we've reached the processing limit
|
||||
if processedCount >= maxToProcess {
|
||||
w.Logger.Info("Reached maximum processing limit, stopping", map[string]interface{}{
|
||||
"processed_count": processedCount,
|
||||
"max_to_process": maxToProcess,
|
||||
})
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// scrapeDocumentsForNotice calls the scraper SDK to scrape documents for a processed notice
|
||||
func (w *NoticeWorker) scrapeDocumentsForNotice(notice *notice.Notice) {
|
||||
// Skip if no NoticePublicationID
|
||||
if notice.NoticePublicationID == "" {
|
||||
w.Logger.Warn("Skipping document scraping - no NoticePublicationID", map[string]interface{}{
|
||||
"notice_id": notice.ID.Hex(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// Get the tender URL
|
||||
noticeURL := notice.GetTenderURL()
|
||||
if noticeURL == "" {
|
||||
w.Logger.Warn("Skipping document scraping - no tender URL available", map[string]interface{}{
|
||||
"notice_id": notice.ID.Hex(),
|
||||
"notice_publication_id": notice.NoticePublicationID,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// Call scraper SDK
|
||||
scrapeReq := &scraper.ScrapeRequest{
|
||||
NoticeID: notice.NoticePublicationID, // Use NoticePublicationID as notice_id
|
||||
NoticeURL: noticeURL,
|
||||
}
|
||||
|
||||
w.Logger.Info("Starting document scraping for processed notice", map[string]interface{}{
|
||||
"notice_id": notice.ID.Hex(),
|
||||
"notice_publication_id": notice.NoticePublicationID,
|
||||
"notice_url": noticeURL,
|
||||
})
|
||||
|
||||
response, err := w.Scraper.ScrapeDocuments(context.Background(), scrapeReq)
|
||||
if err != nil {
|
||||
w.Logger.Error("Failed to scrape documents", map[string]interface{}{
|
||||
"notice_id": notice.ID.Hex(),
|
||||
"notice_publication_id": notice.NoticePublicationID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
if !response.Success {
|
||||
w.Logger.Warn("Document scraping failed", map[string]interface{}{
|
||||
"notice_id": notice.ID.Hex(),
|
||||
"notice_publication_id": notice.NoticePublicationID,
|
||||
"errors": response.Errors,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
w.Logger.Info("Document scraping completed successfully", map[string]interface{}{
|
||||
"notice_id": notice.ID.Hex(),
|
||||
"notice_publication_id": notice.NoticePublicationID,
|
||||
"uploaded_count": response.UploadedCount,
|
||||
"bucket": response.Buckets,
|
||||
})
|
||||
}
|
||||
|
||||
func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) {
|
||||
t, err := w.TenderRepo.GetByNoticePublicationID(context.Background(), n.NoticePublicationID)
|
||||
|
||||
@@ -0,0 +1,200 @@
|
||||
package workers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
"tm/internal/tender"
|
||||
"tm/pkg/logger"
|
||||
"tm/pkg/mongo"
|
||||
"tm/pkg/scraper"
|
||||
)
|
||||
|
||||
// DocumentScraperWorker handles document scraping for tenders
|
||||
type DocumentScraperWorker struct {
|
||||
Mongo *mongo.ConnectionManager
|
||||
Logger logger.Logger
|
||||
TenderRepo tender.TenderRepository
|
||||
Scraper scraper.SDK
|
||||
}
|
||||
|
||||
// NewDocumentScraperWorker creates a new document scraper worker
|
||||
func NewDocumentScraperWorker(mongo *mongo.ConnectionManager, logger logger.Logger, tenderRepo tender.TenderRepository, scraper scraper.SDK) *DocumentScraperWorker {
|
||||
logger.Info("Creating new document scraper worker", map[string]interface{}{})
|
||||
return &DocumentScraperWorker{
|
||||
Mongo: mongo,
|
||||
Logger: logger,
|
||||
TenderRepo: tenderRepo,
|
||||
Scraper: scraper,
|
||||
}
|
||||
}
|
||||
|
||||
// Run starts the document scraping process
|
||||
func (w *DocumentScraperWorker) Run() {
|
||||
w.Logger.Info("Document scraper worker started", map[string]interface{}{})
|
||||
|
||||
limit := 10
|
||||
skip := 0
|
||||
for {
|
||||
w.Logger.Info("Document scraper worker checking for tenders", map[string]interface{}{
|
||||
"skip": skip,
|
||||
"limit": limit,
|
||||
})
|
||||
tenders, totalCount, err := w.TenderRepo.GetUnScrapedTenders(context.Background(), limit, skip)
|
||||
if err != nil {
|
||||
w.Logger.Error("Failed to get un-scraped tenders", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
w.Logger.Info("Found tenders for document scraping", map[string]interface{}{
|
||||
"count": len(tenders),
|
||||
"total_count": totalCount,
|
||||
"skip": skip,
|
||||
})
|
||||
|
||||
if len(tenders) == 0 {
|
||||
w.Logger.Info("No more tenders to process for document scraping", map[string]interface{}{})
|
||||
break
|
||||
}
|
||||
|
||||
w.Logger.Info("Processing tenders for document scraping", map[string]interface{}{
|
||||
"tender_count": len(tenders),
|
||||
})
|
||||
|
||||
for _, t := range tenders {
|
||||
w.Logger.Info("Processing tender for document scraping", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
})
|
||||
|
||||
w.scrapeDocumentsForTender(&t)
|
||||
}
|
||||
|
||||
skip += limit
|
||||
}
|
||||
|
||||
w.Logger.Info("Document scraper worker completed", map[string]interface{}{})
|
||||
}
|
||||
|
||||
// scrapeDocumentsForTender scrapes documents for a single tender
|
||||
func (w *DocumentScraperWorker) scrapeDocumentsForTender(t *tender.Tender) {
|
||||
w.Logger.Info("Starting document scraping for tender", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
"document_uri": t.DocumentURI,
|
||||
})
|
||||
|
||||
// Skip if no NoticePublicationID
|
||||
if t.NoticePublicationID == "" {
|
||||
w.Logger.Warn("Skipping document scraping - no NoticePublicationID", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// Use DocumentURI as NoticeURL for the scraper
|
||||
noticeURL := t.DocumentURI
|
||||
if noticeURL == "" {
|
||||
w.Logger.Warn("Skipping document scraping - no DocumentURI available", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// Call scraper SDK
|
||||
scrapeReq := &scraper.ScrapeRequest{
|
||||
NoticeID: t.NoticePublicationID, // Use NoticePublicationID as notice_id
|
||||
NoticeURL: noticeURL, // Use DocumentURI as notice_url
|
||||
}
|
||||
|
||||
w.Logger.Info("Starting document scraping for tender", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
"document_uri": noticeURL,
|
||||
})
|
||||
|
||||
response, err := w.Scraper.ScrapeDocuments(context.Background(), scrapeReq)
|
||||
if err != nil {
|
||||
w.Logger.Error("Failed to scrape documents for tender", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
|
||||
// Mark as processed with error (don't retry indefinitely)
|
||||
t.ProcessingMetadata.DocumentsScraped = false
|
||||
t.ProcessingMetadata.DocumentsScrapedAt = time.Now().Unix()
|
||||
w.updateTenderScrapingStatus(t)
|
||||
return
|
||||
}
|
||||
|
||||
if !response.Success {
|
||||
w.Logger.Warn("Document scraping failed for tender", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
"errors": response.Errors,
|
||||
})
|
||||
|
||||
// Mark as processed with failure
|
||||
t.ProcessingMetadata.DocumentsScraped = false
|
||||
t.ProcessingMetadata.DocumentsScrapedAt = time.Now().Unix()
|
||||
w.updateTenderScrapingStatus(t)
|
||||
return
|
||||
}
|
||||
|
||||
// Store scraped documents information
|
||||
scrapedAt := time.Now().Unix()
|
||||
t.ScrapedDocuments = make([]tender.ScrapedDocument, len(response.UploadedFiles))
|
||||
for i, uploadedFile := range response.UploadedFiles {
|
||||
t.ScrapedDocuments[i] = tender.ScrapedDocument{
|
||||
Filename: uploadedFile.Filename,
|
||||
ObjectName: uploadedFile.ObjectName,
|
||||
BucketName: "opplens",
|
||||
Size: uploadedFile.Size,
|
||||
ScrapedAt: scrapedAt,
|
||||
}
|
||||
}
|
||||
|
||||
// Mark as successfully scraped
|
||||
t.ProcessingMetadata.DocumentsScraped = true
|
||||
t.ProcessingMetadata.DocumentsScrapedAt = scrapedAt
|
||||
|
||||
err = w.updateTenderScrapingStatus(t)
|
||||
if err != nil {
|
||||
w.Logger.Error("Failed to update tender scraping status", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
w.Logger.Info("Document scraping completed successfully for tender", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
"uploaded_count": response.UploadedCount,
|
||||
"bucket_count": len(response.Buckets),
|
||||
})
|
||||
}
|
||||
|
||||
// updateTenderScrapingStatus updates the tender's document scraping metadata
|
||||
func (w *DocumentScraperWorker) updateTenderScrapingStatus(t *tender.Tender) error {
|
||||
err := w.TenderRepo.Update(context.Background(), t)
|
||||
if err != nil {
|
||||
w.Logger.Error("Failed to update tender scraping status", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"error": err.Error(),
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
w.Logger.Debug("Updated tender scraping status", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"documents_scraped": t.ProcessingMetadata.DocumentsScraped,
|
||||
"documents_scraped_at": t.ProcessingMetadata.DocumentsScrapedAt,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,365 @@
|
||||
package workers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
"tm/internal/tender"
|
||||
"tm/pkg/glm"
|
||||
"tm/pkg/logger"
|
||||
"tm/pkg/minio"
|
||||
"tm/pkg/mongo"
|
||||
)
|
||||
|
||||
// DocumentSummarizationWorker handles document summarization using GLM AI
|
||||
type DocumentSummarizationWorker struct {
|
||||
Mongo *mongo.ConnectionManager
|
||||
Logger logger.Logger
|
||||
TenderRepo tender.TenderRepository
|
||||
MinioSDK *minio.Service
|
||||
GLM *glm.SDK
|
||||
}
|
||||
|
||||
// NewDocumentSummarizationWorker creates a new document summarization worker
|
||||
func NewDocumentSummarizationWorker(mongo *mongo.ConnectionManager, logger logger.Logger, tenderRepo tender.TenderRepository, minioSDK *minio.Service, glmSDK *glm.SDK) *DocumentSummarizationWorker {
|
||||
return &DocumentSummarizationWorker{
|
||||
Mongo: mongo,
|
||||
Logger: logger,
|
||||
TenderRepo: tenderRepo,
|
||||
MinioSDK: minioSDK,
|
||||
GLM: glmSDK,
|
||||
}
|
||||
}
|
||||
|
||||
// Run starts the document summarization process
|
||||
func (w *DocumentSummarizationWorker) Run() {
|
||||
w.Logger.Info("Document summarization worker started", map[string]interface{}{})
|
||||
|
||||
limit := 5 // Process fewer tenders at a time since summarization is resource-intensive
|
||||
skip := 0
|
||||
for {
|
||||
tenders, totalCount, err := w.TenderRepo.GetUnSummarizedTenders(context.Background(), limit, skip)
|
||||
if err != nil {
|
||||
w.Logger.Error("Failed to get un-summarized tenders", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
w.Logger.Info("Found tenders for document summarization", map[string]interface{}{
|
||||
"count": len(tenders),
|
||||
"total_count": totalCount,
|
||||
"skip": skip,
|
||||
})
|
||||
|
||||
if len(tenders) == 0 {
|
||||
w.Logger.Info("No more tenders to process for document summarization", map[string]interface{}{})
|
||||
break
|
||||
}
|
||||
|
||||
for _, t := range tenders {
|
||||
w.Logger.Info("Processing tender for document summarization", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
"document_count": len(t.ScrapedDocuments),
|
||||
})
|
||||
|
||||
w.summarizeDocumentsForTender(&t)
|
||||
}
|
||||
|
||||
skip += limit
|
||||
}
|
||||
|
||||
w.Logger.Info("Document summarization worker completed", map[string]interface{}{})
|
||||
}
|
||||
|
||||
// summarizeDocumentsForTender summarizes all documents for a single tender
|
||||
func (w *DocumentSummarizationWorker) summarizeDocumentsForTender(t *tender.Tender) {
|
||||
if len(t.ScrapedDocuments) == 0 {
|
||||
w.Logger.Warn("No scraped documents found for tender", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
})
|
||||
|
||||
// Mark as processed even if no documents (nothing to summarize)
|
||||
t.ProcessingMetadata.DocumentsSummarized = true
|
||||
t.ProcessingMetadata.DocumentsSummarizedAt = time.Now().Unix()
|
||||
w.updateTenderSummarizationStatus(t)
|
||||
return
|
||||
}
|
||||
|
||||
summaries := make([]tender.DocumentSummary, 0, len(t.ScrapedDocuments))
|
||||
summarizedAt := time.Now().Unix()
|
||||
|
||||
for _, doc := range t.ScrapedDocuments {
|
||||
summary, err := w.summarizeDocument(doc, t.NoticeLanguageCode)
|
||||
if err != nil {
|
||||
w.Logger.Error("Failed to summarize document", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
"document_name": doc.Filename,
|
||||
"object_name": doc.ObjectName,
|
||||
"error": err.Error(),
|
||||
})
|
||||
|
||||
// Add summary with error
|
||||
summaries = append(summaries, tender.DocumentSummary{
|
||||
DocumentName: doc.Filename,
|
||||
ObjectName: doc.ObjectName,
|
||||
BucketName: doc.BucketName,
|
||||
Summary: "",
|
||||
SummaryLanguage: "en",
|
||||
DocumentType: w.getDocumentType(doc.Filename),
|
||||
SummarizedAt: summarizedAt,
|
||||
SummaryModel: w.getGLMModelName(),
|
||||
Error: err.Error(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
summaries = append(summaries, tender.DocumentSummary{
|
||||
DocumentName: doc.Filename,
|
||||
ObjectName: doc.ObjectName,
|
||||
BucketName: doc.BucketName,
|
||||
Summary: summary,
|
||||
SummaryLanguage: "en",
|
||||
DocumentType: w.getDocumentType(doc.Filename),
|
||||
SummarizedAt: summarizedAt,
|
||||
SummaryModel: w.getGLMModelName(),
|
||||
})
|
||||
|
||||
w.Logger.Info("Document summarized successfully", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
"document_name": doc.Filename,
|
||||
"summary_length": len(summary),
|
||||
})
|
||||
}
|
||||
|
||||
// Update tender with summaries
|
||||
t.DocumentSummaries = summaries
|
||||
t.ProcessingMetadata.DocumentsSummarized = true
|
||||
t.ProcessingMetadata.DocumentsSummarizedAt = summarizedAt
|
||||
|
||||
err := w.updateTenderSummarizationStatus(t)
|
||||
if err != nil {
|
||||
w.Logger.Error("Failed to update tender summarization status", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
w.Logger.Info("Document summarization completed for tender", map[string]interface{}{
|
||||
"tender_id": t.ID.Hex(),
|
||||
"notice_publication_id": t.NoticePublicationID,
|
||||
"summarized_count": len(summaries),
|
||||
"total_documents": len(t.ScrapedDocuments),
|
||||
})
|
||||
}
|
||||
|
||||
// summarizeDocument downloads a document from MinIO and summarizes it using GLM
|
||||
func (w *DocumentSummarizationWorker) summarizeDocument(doc tender.ScrapedDocument, sourceLanguage string) (string, error) {
|
||||
// Download document from MinIO
|
||||
documentData, err := w.MinioSDK.Download(context.Background(), doc.BucketName, doc.ObjectName, minio.DownloadOptions{})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to download document from MinIO: %w", err)
|
||||
}
|
||||
defer documentData.Close()
|
||||
|
||||
// Read document content
|
||||
documentBytes, err := io.ReadAll(documentData)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to read document content: %w", err)
|
||||
}
|
||||
|
||||
// Extract text from document based on type
|
||||
textContent, err := w.extractTextFromDocument(documentBytes, doc.Filename)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to extract text from document: %w", err)
|
||||
}
|
||||
|
||||
if len(textContent) == 0 {
|
||||
return "", fmt.Errorf("no text content extracted from document")
|
||||
}
|
||||
|
||||
// Limit text length for summarization (GLM has token limits)
|
||||
maxLength := 10000 // Adjust based on GLM model limits
|
||||
if len(textContent) > maxLength {
|
||||
textContent = textContent[:maxLength] + "..."
|
||||
}
|
||||
|
||||
// Create summarization prompt
|
||||
prompt := w.createSummarizationPrompt(textContent, sourceLanguage)
|
||||
|
||||
// Use GLM to summarize
|
||||
summary, err := w.GLM.ChatCompletion(context.Background(), &glm.ChatCompletionRequest{
|
||||
Messages: []glm.Message{
|
||||
{
|
||||
Role: glm.MessageRoleUser,
|
||||
Content: prompt,
|
||||
},
|
||||
},
|
||||
Temperature: 0.3, // Lower temperature for more consistent summaries
|
||||
MaxTokens: 1000,
|
||||
})
|
||||
if err != nil {
|
||||
// Check if GLM is configured (has API key)
|
||||
if strings.Contains(err.Error(), "token expired") || strings.Contains(err.Error(), "AUTHENTICATION_ERROR") {
|
||||
// GLM is configured but API key is invalid - return error instead of mock
|
||||
return "", fmt.Errorf("GLM API authentication failed: %w", err)
|
||||
}
|
||||
|
||||
// GLM is not configured or other error - use mock summary for development
|
||||
w.Logger.Warn("GLM API failed, using mock summary for development", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return fmt.Sprintf("Mock summary for document: %s (GLM API not configured)", doc.Filename), nil
|
||||
}
|
||||
|
||||
if len(summary.Choices) == 0 || summary.Choices[0].Message.Content == "" {
|
||||
return "", fmt.Errorf("GLM API returned empty response for document: %s", doc.Filename)
|
||||
}
|
||||
|
||||
return strings.TrimSpace(summary.Choices[0].Message.Content), nil
|
||||
}
|
||||
|
||||
// extractTextFromDocument extracts text content from various document formats
|
||||
func (w *DocumentSummarizationWorker) extractTextFromDocument(data []byte, filename string) (string, error) {
|
||||
ext := strings.ToLower(filepath.Ext(filename))
|
||||
|
||||
switch ext {
|
||||
case ".xlsx":
|
||||
return w.extractTextFromExcel(data)
|
||||
case ".pdf":
|
||||
return w.extractTextFromPDF(data)
|
||||
case ".docx":
|
||||
return w.extractTextFromDOCX(data)
|
||||
case ".txt":
|
||||
return string(data), nil
|
||||
case ".html", ".htm":
|
||||
return w.extractTextFromHTML(data)
|
||||
default:
|
||||
// Try to extract as plain text for unknown formats
|
||||
return string(data), nil
|
||||
}
|
||||
}
|
||||
|
||||
// extractTextFromPDF extracts text from PDF documents
|
||||
func (w *DocumentSummarizationWorker) extractTextFromPDF(data []byte) (string, error) {
|
||||
// For now, return a placeholder. In production, you'd use a PDF parsing library
|
||||
// like github.com/unidoc/unipdf or github.com/rsc/pdf
|
||||
return fmt.Sprintf("PDF content extraction not implemented. File size: %d bytes", len(data)), nil
|
||||
}
|
||||
|
||||
// extractTextFromDOCX extracts text from DOCX documents
|
||||
func (w *DocumentSummarizationWorker) extractTextFromDOCX(data []byte) (string, error) {
|
||||
// For now, return a placeholder. In production, you'd use a DOCX parsing library
|
||||
// like github.com/lukasjarosch/go-docx or github.com/fumiama/go-docx
|
||||
return fmt.Sprintf("DOCX content extraction not implemented. File size: %d bytes", len(data)), nil
|
||||
}
|
||||
|
||||
// extractTextFromExcel extracts text from Excel documents (.xlsx)
|
||||
func (w *DocumentSummarizationWorker) extractTextFromExcel(data []byte) (string, error) {
|
||||
// For now, return a placeholder. In production, you'd use an Excel parsing library
|
||||
// like github.com/tealeg/xlsx/v3 or github.com/360EntSecGroup-Skylar/excelize
|
||||
return fmt.Sprintf("Excel content extraction not implemented. File size: %d bytes", len(data)), nil
|
||||
}
|
||||
|
||||
// extractTextFromHTML extracts text from HTML documents
|
||||
func (w *DocumentSummarizationWorker) extractTextFromHTML(data []byte) (string, error) {
|
||||
// Simple HTML text extraction (remove tags)
|
||||
htmlContent := string(data)
|
||||
|
||||
// Basic HTML tag removal (very simple implementation)
|
||||
// In production, use a proper HTML parser like goquery
|
||||
htmlContent = strings.ReplaceAll(htmlContent, "<script", "</script>")
|
||||
htmlContent = strings.ReplaceAll(htmlContent, "<style", "</style>")
|
||||
|
||||
// Remove HTML tags using simple regex-like approach
|
||||
var result strings.Builder
|
||||
inTag := false
|
||||
for _, r := range htmlContent {
|
||||
switch r {
|
||||
case '<':
|
||||
inTag = true
|
||||
case '>':
|
||||
inTag = false
|
||||
default:
|
||||
if !inTag {
|
||||
result.WriteRune(r)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return strings.TrimSpace(result.String()), nil
|
||||
}
|
||||
|
||||
// createSummarizationPrompt creates a prompt for GLM to summarize tender documents
|
||||
func (w *DocumentSummarizationWorker) createSummarizationPrompt(textContent, sourceLanguage string) string {
|
||||
return fmt.Sprintf(`Please provide a concise summary of the following tender document in English. Focus on key information such as:
|
||||
|
||||
1. Project/Tender name and description
|
||||
2. Contracting authority/organization details
|
||||
3. Key requirements and specifications
|
||||
4. Important dates (deadline, award date, etc.)
|
||||
5. Estimated value and currency
|
||||
6. Selection criteria
|
||||
7. Contact information
|
||||
|
||||
Document content:
|
||||
%s
|
||||
|
||||
Summary:`, textContent)
|
||||
}
|
||||
|
||||
// getDocumentType returns the document type based on file extension
|
||||
func (w *DocumentSummarizationWorker) getDocumentType(filename string) string {
|
||||
ext := strings.ToLower(filepath.Ext(filename))
|
||||
switch ext {
|
||||
case ".xlsx":
|
||||
return "xlsx"
|
||||
case ".pdf":
|
||||
return "pdf"
|
||||
case ".docx":
|
||||
return "docx"
|
||||
case ".doc":
|
||||
return "doc"
|
||||
case ".txt":
|
||||
return "txt"
|
||||
case ".html", ".htm":
|
||||
return "html"
|
||||
default:
|
||||
return "unknown"
|
||||
}
|
||||
}
|
||||
|
||||
// getGLMModelName returns the GLM model name being used
|
||||
func (w *DocumentSummarizationWorker) getGLMModelName() string {
|
||||
return "glm-4.5"
|
||||
}
|
||||
|
||||
// updateTenderSummarizationStatus updates the tender's document summarization metadata
|
||||
func (w *DocumentSummarizationWorker) updateTenderSummarizationStatus(tender *tender.Tender) error {
|
||||
err := w.TenderRepo.Update(context.Background(), tender)
|
||||
if err != nil {
|
||||
w.Logger.Error("Failed to update tender summarization status", map[string]interface{}{
|
||||
"tender_id": tender.ID.Hex(),
|
||||
"error": err.Error(),
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
w.Logger.Debug("Updated tender summarization status", map[string]interface{}{
|
||||
"tender_id": tender.ID.Hex(),
|
||||
"documents_summarized": tender.ProcessingMetadata.DocumentsSummarized,
|
||||
"documents_summarized_at": tender.ProcessingMetadata.DocumentsSummarizedAt,
|
||||
"summary_count": len(tender.DocumentSummaries),
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -17,6 +17,7 @@ type Repository interface {
|
||||
GetByID(ctx context.Context, id string) (*Notice, error)
|
||||
GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Notice, error)
|
||||
GetUnProcessedNotices(ctx context.Context, limit int, skip int) ([]Notice, int64, error)
|
||||
GetProcessedNotices(ctx context.Context, limit int, skip int) ([]Notice, int64, error)
|
||||
Delete(ctx context.Context, id string) error
|
||||
}
|
||||
|
||||
@@ -166,6 +167,28 @@ func (r *noticeRepository) GetUnProcessedNotices(ctx context.Context, limit int,
|
||||
return result.Items, result.TotalCount, nil
|
||||
}
|
||||
|
||||
// GetProcessedNotices retrieves notices that have been successfully processed
|
||||
func (r *noticeRepository) GetProcessedNotices(ctx context.Context, limit int, skip int) ([]Notice, int64, error) {
|
||||
filter := bson.M{"processing_metadata.processed": true}
|
||||
|
||||
pagination := orm.Pagination{
|
||||
Limit: limit,
|
||||
Skip: skip,
|
||||
SortField: "updated_at",
|
||||
SortOrder: -1,
|
||||
}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get processed notices", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
return result.Items, result.TotalCount, nil
|
||||
}
|
||||
|
||||
// Delete deletes a notice by ID
|
||||
func (r *noticeRepository) Delete(ctx context.Context, id string) error {
|
||||
err := r.ormRepo.Delete(ctx, id)
|
||||
|
||||
@@ -21,8 +21,8 @@ type scraperService struct {
|
||||
}
|
||||
|
||||
// NewService creates a new scraper service with stored credentials
|
||||
func NewService(baseURL string, timeout time.Duration, logger logger.Logger, username, password, loginURL string) Service {
|
||||
sdk := scraper.NewClient(baseURL, timeout, logger, username, password, loginURL)
|
||||
func NewService(baseURL string, timeout time.Duration, logger logger.Logger) Service {
|
||||
sdk := scraper.NewClient(baseURL, timeout, logger)
|
||||
return &scraperService{
|
||||
sdk: sdk,
|
||||
logger: logger,
|
||||
|
||||
@@ -8,6 +8,28 @@ import (
|
||||
"tm/pkg/mongo"
|
||||
)
|
||||
|
||||
// DocumentSummary represents a summary of a tender document
|
||||
type DocumentSummary struct {
|
||||
DocumentName string `bson:"document_name" json:"document_name"`
|
||||
ObjectName string `bson:"object_name" json:"object_name"`
|
||||
BucketName string `bson:"bucket_name" json:"bucket_name"`
|
||||
Summary string `bson:"summary" json:"summary"`
|
||||
SummaryLanguage string `bson:"summary_language" json:"summary_language"`
|
||||
DocumentType string `bson:"document_type" json:"document_type"`
|
||||
SummarizedAt int64 `bson:"summarized_at" json:"summarized_at"`
|
||||
SummaryModel string `bson:"summary_model" json:"summary_model"`
|
||||
Error string `bson:"error,omitempty" json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// ScrapedDocument represents information about a document that was scraped and uploaded to MinIO
|
||||
type ScrapedDocument struct {
|
||||
Filename string `bson:"filename" json:"filename"`
|
||||
ObjectName string `bson:"object_name" json:"object_name"`
|
||||
BucketName string `bson:"bucket_name" json:"bucket_name"`
|
||||
Size int64 `bson:"size" json:"size"`
|
||||
ScrapedAt int64 `bson:"scraped_at" json:"scraped_at"`
|
||||
}
|
||||
|
||||
// Tender represents a tender/contract notice entity
|
||||
type Tender struct {
|
||||
mongo.Model `bson:",inline"`
|
||||
@@ -66,6 +88,8 @@ type Tender struct {
|
||||
SuspensionDate int64 `bson:"suspension_date,omitempty" json:"suspension_date,omitempty"` // Unix milliseconds
|
||||
Modifications []TenderModification `bson:"modifications,omitempty" json:"modifications,omitempty"`
|
||||
AwardedEntities []Awarded `bson:"awarded,omitempty" json:"awarded,omitempty"`
|
||||
ScrapedDocuments []ScrapedDocument `bson:"scraped_documents,omitempty" json:"scraped_documents,omitempty"`
|
||||
DocumentSummaries []DocumentSummary `bson:"document_summaries,omitempty" json:"document_summaries,omitempty"`
|
||||
}
|
||||
|
||||
// Organization represents organization information
|
||||
@@ -108,6 +132,13 @@ type ProcessingMetadata struct {
|
||||
ParsingErrors []string `bson:"parsing_errors,omitempty" json:"parsing_errors,omitempty"`
|
||||
ValidationErrors []string `bson:"validation_errors,omitempty" json:"validation_errors,omitempty"`
|
||||
EnrichmentData map[string]string `bson:"enrichment_data,omitempty" json:"enrichment_data,omitempty"`
|
||||
TranslatedData map[string]string `bson:"translated_data,omitempty" json:"translated_data,omitempty"`
|
||||
TranslatedAt int64 `bson:"translated_at,omitempty" json:"translated_at,omitempty"`
|
||||
Processed bool `bson:"processed" json:"processed"`
|
||||
DocumentsScraped bool `bson:"documents_scraped" json:"documents_scraped"` // Whether documents have been scraped
|
||||
DocumentsScrapedAt int64 `bson:"documents_scraped_at" json:"documents_scraped_at"` // When documents were scraped
|
||||
DocumentsSummarized bool `bson:"documents_summarized" json:"documents_summarized"` // Whether documents have been summarized
|
||||
DocumentsSummarizedAt int64 `bson:"documents_summarized_at" json:"documents_summarized_at"` // When documents were summarized
|
||||
}
|
||||
|
||||
// TenderModification represents a modification made to a tender
|
||||
|
||||
@@ -312,3 +312,32 @@ func (h *TenderHandler) GetPublicTenderDetails(c echo.Context) error {
|
||||
|
||||
return response.Success(c, tender, "Tender details retrieved successfully")
|
||||
}
|
||||
|
||||
// GetDocumentSummaries retrieves document summaries for a tender
|
||||
// @Summary Get document summaries for a tender
|
||||
// @Description Retrieve AI-generated summaries of documents for a specific tender
|
||||
// @Tags Tenders
|
||||
// @Produce json
|
||||
// @Param id path string true "Tender ID"
|
||||
// @Success 200 {object} response.APIResponse{data=[]DocumentSummary}
|
||||
// @Failure 400 {object} response.APIResponse
|
||||
// @Failure 404 {object} response.APIResponse
|
||||
// @Failure 500 {object} response.APIResponse
|
||||
// @Router /api/v1/tenders/{id}/document-summaries [get]
|
||||
func (h *TenderHandler) GetDocumentSummaries(c echo.Context) error {
|
||||
id := c.Param("id")
|
||||
if id == "" {
|
||||
return response.BadRequest(c, "Tender ID is required", "ID parameter cannot be empty")
|
||||
}
|
||||
|
||||
summaries, err := h.service.GetDocumentSummaries(c.Request().Context(), id)
|
||||
if err != nil {
|
||||
h.logger.Error("Failed to get document summaries", map[string]interface{}{
|
||||
"tender_id": id,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return response.NotFound(c, "Document summaries not found")
|
||||
}
|
||||
|
||||
return response.Success(c, summaries, "Document summaries retrieved successfully")
|
||||
}
|
||||
|
||||
@@ -23,6 +23,8 @@ type TenderRepository interface {
|
||||
GetTenderCountByClassification(ctx context.Context) (map[string]int64, error)
|
||||
Search(ctx context.Context, form *SearchForm, pagination *response.Pagination) ([]Tender, int64, error)
|
||||
GetExpiredTenders(ctx context.Context, beforeDate int64) ([]Tender, error)
|
||||
GetUnScrapedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error)
|
||||
GetUnSummarizedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error)
|
||||
Update(ctx context.Context, tender *Tender) error
|
||||
Delete(ctx context.Context, id string) error
|
||||
GetStatistics(ctx context.Context) (*TenderStatistics, error)
|
||||
@@ -275,6 +277,66 @@ func (r *tenderRepository) GetExpiredTenders(ctx context.Context, beforeDate int
|
||||
return result.Items, nil
|
||||
}
|
||||
|
||||
// GetUnScrapedTenders retrieves tenders that haven't been processed for document scraping
|
||||
func (r *tenderRepository) GetUnScrapedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error) {
|
||||
filter := bson.M{
|
||||
"$or": []bson.M{
|
||||
{"processing_metadata.documents_scraped": bson.M{"$exists": false}},
|
||||
{"processing_metadata.documents_scraped": false},
|
||||
{"processing_metadata.documents_scraped": nil},
|
||||
},
|
||||
}
|
||||
|
||||
pagination := orm.Pagination{
|
||||
Limit: limit,
|
||||
Skip: skip,
|
||||
SortField: "created_at",
|
||||
SortOrder: 1, // Oldest first
|
||||
}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get un-scraped tenders", map[string]interface{}{
|
||||
"limit": limit,
|
||||
"skip": skip,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
return result.Items, result.TotalCount, nil
|
||||
}
|
||||
|
||||
// GetUnSummarizedTenders retrieves tenders that have documents scraped but not summarized
|
||||
func (r *tenderRepository) GetUnSummarizedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error) {
|
||||
filter := bson.M{
|
||||
"processing_metadata.documents_scraped": true,
|
||||
"$or": []bson.M{
|
||||
{"processing_metadata.documents_summarized": bson.M{"$exists": false}},
|
||||
{"processing_metadata.documents_summarized": false},
|
||||
},
|
||||
}
|
||||
|
||||
pagination := orm.Pagination{
|
||||
Limit: limit,
|
||||
Skip: skip,
|
||||
SortField: "processing_metadata.documents_scraped_at",
|
||||
SortOrder: 1, // Oldest first (process oldest scraped documents first)
|
||||
}
|
||||
|
||||
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
|
||||
if err != nil {
|
||||
r.logger.Error("Failed to get un-summarized tenders", map[string]interface{}{
|
||||
"limit": limit,
|
||||
"skip": skip,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
return result.Items, result.TotalCount, nil
|
||||
}
|
||||
|
||||
// GetStatistics calculates tender statistics using aggregation
|
||||
func (r *tenderRepository) GetStatistics(ctx context.Context) (*TenderStatistics, error) {
|
||||
stats := &TenderStatistics{
|
||||
|
||||
@@ -17,6 +17,7 @@ type Service interface {
|
||||
Delete(ctx context.Context, id string) error
|
||||
Search(ctx context.Context, req *SearchForm, pagination *response.Pagination) (*SearchResponse, error)
|
||||
Recommend(ctx context.Context, form *SearchForm, pagination *response.Pagination) (*SearchResponse, error)
|
||||
GetDocumentSummaries(ctx context.Context, id string) ([]DocumentSummary, error)
|
||||
|
||||
// Maintenance operations
|
||||
UpdateExpiredTenders(ctx context.Context) error
|
||||
@@ -129,6 +130,39 @@ func (s *tenderService) Delete(ctx context.Context, id string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetDocumentSummaries retrieves document summaries for a specific tender
|
||||
func (s *tenderService) GetDocumentSummaries(ctx context.Context, id string) ([]DocumentSummary, error) {
|
||||
if id == "" {
|
||||
return nil, fmt.Errorf("tender ID cannot be empty")
|
||||
}
|
||||
|
||||
// Get tender from repository
|
||||
tender, err := s.repository.GetByID(ctx, id)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to get tender for document summaries", map[string]interface{}{
|
||||
"tender_id": id,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return nil, fmt.Errorf("failed to get tender: %w", err)
|
||||
}
|
||||
|
||||
if tender == nil {
|
||||
return nil, fmt.Errorf("tender not found")
|
||||
}
|
||||
|
||||
// Return document summaries (empty slice if none exist)
|
||||
if tender.DocumentSummaries == nil {
|
||||
return []DocumentSummary{}, nil
|
||||
}
|
||||
|
||||
s.logger.Info("Retrieved document summaries for tender", map[string]interface{}{
|
||||
"tender_id": id,
|
||||
"summaries_count": len(tender.DocumentSummaries),
|
||||
})
|
||||
|
||||
return tender.DocumentSummaries, nil
|
||||
}
|
||||
|
||||
// ListTenders retrieves tenders with pagination and filtering
|
||||
func (s *tenderService) Search(ctx context.Context, form *SearchForm, pagination *response.Pagination) (*SearchResponse, error) {
|
||||
s.logger.Info("Listing tenders", map[string]interface{}{
|
||||
|
||||
@@ -95,6 +95,21 @@ type HCaptchaConfig struct {
|
||||
Timeout time.Duration `env:"HCAPTCHA_TIMEOUT"`
|
||||
}
|
||||
|
||||
type ScraperConfig struct {
|
||||
BaseURL string `env:"SCRAPER_BASE_URL"`
|
||||
Timeout time.Duration `env:"SCRAPER_TIMEOUT"`
|
||||
}
|
||||
|
||||
type MinIOConfig struct {
|
||||
Endpoint string `env:"MINIO_ENDPOINT" default:"localhost:9000"`
|
||||
AccessKeyID string `env:"MINIO_ACCESS_KEY_ID" default:""`
|
||||
SecretAccessKey string `env:"MINIO_SECRET_ACCESS_KEY" default:""`
|
||||
UseSSL bool `env:"MINIO_USE_SSL" default:"false"`
|
||||
Region string `env:"MINIO_REGION" default:"us-east-1"`
|
||||
DefaultBucket string `env:"MINIO_DEFAULT_BUCKET" default:"files"`
|
||||
HierarchicalBucket string `env:"MINIO_HIERARCHICAL_BUCKET" default:"documents"`
|
||||
}
|
||||
|
||||
// LoadConfig loads configuration with priority: OS environment variables > .env file
|
||||
// OS environment variables take precedence over .env file values
|
||||
func LoadConfig[T any](path string, config T) (T, error) {
|
||||
|
||||
+10
-4
@@ -3,6 +3,7 @@ package minio
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/minio/minio-go/v7"
|
||||
"github.com/minio/minio-go/v7/pkg/credentials"
|
||||
@@ -36,6 +37,8 @@ func (cm *ConnectionManager) Connect() error {
|
||||
Region: cm.config.Region,
|
||||
Transport: &http.Transport{
|
||||
ResponseHeaderTimeout: cm.config.RequestTimeout,
|
||||
TLSHandshakeTimeout: 10 * time.Second,
|
||||
DisableKeepAlives: false,
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
@@ -48,14 +51,16 @@ func (cm *ConnectionManager) Connect() error {
|
||||
|
||||
cm.client = minioClient
|
||||
|
||||
// Test connection by listing buckets
|
||||
ctx, cancel := context.WithTimeout(context.Background(), cm.config.ConnectTimeout)
|
||||
// Test connection by checking if we can reach the server
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
_, err = cm.client.ListBuckets(ctx)
|
||||
// Try bucket exists check with shorter timeout
|
||||
exists, err := cm.client.BucketExists(ctx, cm.config.DefaultBucket)
|
||||
if err != nil {
|
||||
cm.logger.Error("Failed to connect to MinIO", map[string]interface{}{
|
||||
"endpoint": cm.config.Endpoint,
|
||||
"bucket": cm.config.DefaultBucket,
|
||||
"error": err.Error(),
|
||||
})
|
||||
return NewErrorWithCause(ErrCodeConnectionFailed, "failed to connect to MinIO", err)
|
||||
@@ -64,8 +69,9 @@ func (cm *ConnectionManager) Connect() error {
|
||||
cm.logger.Info("Successfully connected to MinIO", map[string]interface{}{
|
||||
"endpoint": cm.config.Endpoint,
|
||||
"region": cm.config.Region,
|
||||
"bucket": cm.config.DefaultBucket,
|
||||
"exists": exists,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
+1
-11
@@ -71,23 +71,16 @@ type Client struct {
|
||||
httpClient *http.Client
|
||||
baseURL string
|
||||
logger logger.Logger
|
||||
// Stored credentials
|
||||
username string
|
||||
password string
|
||||
loginURL string
|
||||
}
|
||||
|
||||
// NewClient creates a new scraper SDK client with stored credentials
|
||||
func NewClient(baseURL string, timeout time.Duration, logger logger.Logger, username, password, loginURL string) SDK {
|
||||
func NewClient(baseURL string, timeout time.Duration, logger logger.Logger) SDK {
|
||||
return &Client{
|
||||
httpClient: &http.Client{
|
||||
Timeout: timeout,
|
||||
},
|
||||
baseURL: baseURL,
|
||||
logger: logger,
|
||||
username: username,
|
||||
password: password,
|
||||
loginURL: loginURL,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -101,9 +94,6 @@ func (c *Client) ScrapeDocuments(ctx context.Context, req *ScrapeRequest) (*Scra
|
||||
// Prepare request body with stored credentials + dynamic fields
|
||||
requestBody := map[string]interface{}{
|
||||
"notice_id": req.NoticeID,
|
||||
"username": c.username,
|
||||
"password": c.password,
|
||||
"login_url": c.loginURL,
|
||||
"notice_url": req.NoticeURL,
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user