diff --git a/cmd/web/main.go b/cmd/web/main.go index cbace38..b74c131 100644 --- a/cmd/web/main.go +++ b/cmd/web/main.go @@ -104,9 +104,7 @@ import ( "tm/internal/feedback" "tm/internal/inquiry" "tm/internal/kanban" - "tm/internal/notice" "tm/internal/notification" - "tm/internal/scraper" "tm/internal/tender" "tm/internal/tender_approval" "tm/internal/user" @@ -118,56 +116,6 @@ import ( "tm/cmd/web/router" ) -func buildTEDConfig(c *bootstrap.TEDConfig) *scraper.TEDConfig { - if c == nil { - return &scraper.TEDConfig{ - BaseURL: "https://ted.europa.eu", - Timeout: 30 * time.Second, - MaxRetries: 3, - RetryDelay: 5 * time.Second, - UserAgent: "TM-TED-Scraper/1.0", - MaxConcurrency: 5, - DownloadDir: "./downloads", - CleanupAfter: 24 * time.Hour, - ScrapingInterval: "* * 10 * * *", - AlertMail: "alerts@opplens.com", - } - } - baseURL := c.BaseURL - if baseURL == "" { - baseURL = "https://ted.europa.eu" - } - userAgent := c.UserAgent - if userAgent == "" { - userAgent = "TM-TED-Scraper/1.0" - } - downloadDir := c.DownloadDir - if downloadDir == "" { - downloadDir = "./downloads" - } - scrapingInterval := c.ScrapingInterval - if scrapingInterval == "" { - scrapingInterval = "* * 10 * * *" - } - alertMail := c.AlertMail - if alertMail == "" { - alertMail = "alerts@opplens.com" - } - return &scraper.TEDConfig{ - BaseURL: baseURL, - Timeout: c.Timeout, - MaxRetries: c.MaxRetries, - RetryDelay: c.RetryDelay, - UserAgent: userAgent, - MaxConcurrency: c.MaxConcurrency, - DownloadDir: downloadDir, - CleanupAfter: c.CleanupAfter, - ScrapingInterval: scrapingInterval, - MaxNoticesPerDay: c.MaxNoticesPerDay, - AlertMail: alertMail, - } -} - func main() { conf := bootstrap.InitConfig() @@ -229,7 +177,7 @@ func main() { inquiryRepository := inquiry.NewRepository(mongoManager, logger) contactRepository := contact.NewContactRepository(mongoManager, logger) cmsRepository := cms.NewRepository(mongoManager, logger) - noticeRepository := notice.NewRepository(mongoManager, logger) + // noticeRepository := notice.NewRepository(mongoManager, logger) boardRepository := kanban.NewBoardRepository(mongoManager, logger) columnRepository := kanban.NewColumnRepository(mongoManager, logger) cardRepository := kanban.NewCardRepository(mongoManager, logger) @@ -256,8 +204,6 @@ func main() { notificationService := notification.NewService(notificationSDK, notificationRepository, userService, customerService, logger) contactService := contact.NewService(contactRepository, logger, notificationSDK) cmsService := cms.NewService(cmsRepository, logger) - tedConfig := buildTEDConfig(&conf.TED) - scraperService := scraper.NewServiceWithTED(conf.Scraper.BaseURL, conf.Scraper.Timeout, logger, mongoManager, noticeRepository, notificationSDK, tedConfig) kanbanService := kanban.NewService(boardRepository, columnRepository, cardRepository, logger) documentScraperService := document_scraper.NewService(tenderRepository, logger) logger.Info("Services initialized successfully", map[string]interface{}{ @@ -277,7 +223,6 @@ func main() { notificationHandler := notification.NewHandler(notificationService) contactHandler := contact.NewHandler(contactService, hcaptchaVerifier) cmsHandler := cms.NewHandler(cmsService) - scraperHandler := scraper.NewHandler(scraperService) kanbanHandler := kanban.NewHandler(kanbanService) fileStoreHandler := filestore.NewHandler(fileStoreService, logger) documentScraperHandler := document_scraper.NewHandler(documentScraperService, logger) @@ -289,7 +234,7 @@ func main() { e := bootstrap.InitHTTPServer(conf, logger) // Register routes - router.RegisterAdminRoutes(e, userHandler, companyHandler, customerHandler, tenderHandler, feedbackHandler, tenderApprovalHandler, inquiryHandler, categoryHandler, flagHandler, notificationHandler, contactHandler, cmsHandler, scraperHandler, kanbanHandler, fileStoreHandler, xssPolicy) + router.RegisterAdminRoutes(e, userHandler, companyHandler, customerHandler, tenderHandler, feedbackHandler, tenderApprovalHandler, inquiryHandler, categoryHandler, flagHandler, notificationHandler, contactHandler, cmsHandler, kanbanHandler, fileStoreHandler, xssPolicy) router.RegisterPublicRoutes(e, customerHandler, tenderHandler, feedbackHandler, tenderApprovalHandler, companyHandler, inquiryHandler, categoryHandler, flagHandler, notificationHandler, contactHandler, cmsHandler, fileStoreHandler, xssPolicy) router.RegisterDocumentScraperRoutes(e, documentScraperHandler, conf.DocumentScraper.APIKey) diff --git a/cmd/web/router/routes.go b/cmd/web/router/routes.go index 273ecd5..a00fd20 100644 --- a/cmd/web/router/routes.go +++ b/cmd/web/router/routes.go @@ -12,7 +12,6 @@ import ( "tm/internal/inquiry" "tm/internal/kanban" "tm/internal/notification" - "tm/internal/scraper" "tm/internal/tender" "tm/internal/tender_approval" "tm/internal/user" @@ -22,7 +21,7 @@ import ( "github.com/labstack/echo/v4" ) -func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler *company.Handler, customerHandler *customer.Handler, tenderHandler *tender.TenderHandler, feedbackHandler *feedback.Handler, tenderApprovalHandler *tender_approval.Handler, inquiryHandler *inquiry.Handler, categoryHandler *company_category.Handler, flagHandler *assets.Handler, notificationHandler *notification.NotificationHandler, contactHandler *contact.Handler, cmsHandler *cms.Handler, scraperHandler *scraper.Handler, kanbanHandler *kanban.Handler, fileStoreHandler *filestore.Handler, xssPolicy *security.XSSPolicy) { +func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler *company.Handler, customerHandler *customer.Handler, tenderHandler *tender.TenderHandler, feedbackHandler *feedback.Handler, tenderApprovalHandler *tender_approval.Handler, inquiryHandler *inquiry.Handler, categoryHandler *company_category.Handler, flagHandler *assets.Handler, notificationHandler *notification.NotificationHandler, contactHandler *contact.Handler, cmsHandler *cms.Handler, kanbanHandler *kanban.Handler, fileStoreHandler *filestore.Handler, xssPolicy *security.XSSPolicy) { adminV1 := e.Group("/admin/v1") // Add CSP middleware to admin routes (stricter policy) @@ -180,14 +179,6 @@ func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler cmsGP.DELETE("/:id", cmsHandler.Delete) } - // Scraper Routes - scraperGP := adminV1.Group("/scraper") - { - scraperGP.Use(userHandler.AuthMiddleware(), userHandler.AdminMiddleware()) - scraperGP.POST("/download", scraperHandler.ScrapeDocuments) - scraperGP.POST("/ted/one-time", scraperHandler.TriggerTEDOneTimeScraping) - } - // Kanban Routes kanbanGP := adminV1.Group("/kanban") { diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 9768096..0afab52 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -15,7 +15,6 @@ import ( "tm/pkg/ollama" "tm/pkg/queue" "tm/pkg/schedule" - "tm/pkg/scraper" ) // Init Application Configuration @@ -78,7 +77,7 @@ 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, scraperService scraper.SDK, minioService *minio.Service) { +func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, glmService *glm.SDK, minioService *minio.Service) { // Debug: Log worker config appLogger.Info("Worker configuration", map[string]interface{}{ "worker_interval": config.Worker.Interval, @@ -93,7 +92,7 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger appLogger.Info("Starting notice worker", map[string]interface{}{ "processing_limit": config.Worker.NoticeProcessingLimit, }) - noticeWorker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, scraperService, config.Worker.NoticeProcessingLimit) + noticeWorker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, config.Worker.NoticeProcessingLimit) noticeWorker.Run() // Initialize Queue-based scraping if enabled @@ -124,24 +123,6 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger } } - // Start Queue Consumer Worker (24/7 continuous processing) - if config.Queue.Enabled && (config.Queue.Mode == "consumer" || config.Queue.Mode == "both") { - appLogger.Info("Starting queue consumer worker", map[string]interface{}{ - "concurrency": config.Queue.Concurrency, - }) - consumerWorker := workers.NewDocumentScraperConsumerWorker( - mongoManager, - appLogger, - tenderRepo, - queueService, - scraperService, - config.Queue.Concurrency, - ) - go consumerWorker.Run() - } else if !config.Queue.Enabled { - appLogger.Info("Queue consumer worker disabled, using scheduled scraping instead", map[string]interface{}{}) - } - // Start Queue Producer Worker (enqueues unscraped tenders periodically) if config.Queue.Enabled && (config.Queue.Mode == "producer" || config.Queue.Mode == "both") { appLogger.Info("Scheduling queue producer worker", map[string]interface{}{ @@ -165,19 +146,6 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger }) } - // Fallback: Keep the old scheduled scraper for backward compatibility - if !config.Queue.Enabled { - appLogger.Info("Scheduling fallback document scraper worker", map[string]interface{}{}) - schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{ - Name: "Document Scraper Worker Job (Fallback)", - Func: func() { - worker := workers.NewDocumentScraperWorker(mongoManager, appLogger, tenderRepo, scraperService) - worker.Run() - }, - Expr: "0 10 * * * *", // Run at 10 AM every day - }) - } - // Initialize Document Summarization Worker (only if MinIO is available) if minioService != nil { appLogger.Info("Starting document summarization worker", map[string]interface{}{}) @@ -209,7 +177,7 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{ Name: "Notice Worker Job", Func: func() { - worker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, scraperService, config.Worker.NoticeProcessingLimit) + worker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, config.Worker.NoticeProcessingLimit) worker.Run() }, Expr: workerInterval, @@ -367,11 +335,3 @@ func InitMinIOService(conf config.MinIOConfig, log logger.Logger) *minio.Service return service } - -func InitScraperService(config Config, logger logger.Logger) scraper.SDK { - return scraper.NewClient( - config.Scraper.BaseURL, - config.Scraper.Timeout, - logger, - ) -} diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 3a61b09..6eb875a 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -46,7 +46,7 @@ func main() { } // Initialize scraper service - scraperService := bootstrap.InitScraperService(*config, appLogger) + // scraperService := bootstrap.InitScraperService(*config, appLogger) // Initialize MinIO service minioService := bootstrap.InitMinIOService(config.MinIO, appLogger) @@ -55,7 +55,7 @@ func main() { } // Initialize Worker - bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, glmService, scraperService, minioService) + bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, glmService, minioService) // Set up signal handling for graceful shutdown signalChan := make(chan os.Signal, 1) diff --git a/cmd/worker/workers/notice.go b/cmd/worker/workers/notice.go index ec3168f..9cb068c 100644 --- a/cmd/worker/workers/notice.go +++ b/cmd/worker/workers/notice.go @@ -11,7 +11,6 @@ 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" @@ -24,11 +23,10 @@ type NoticeWorker struct { NoticeRepo notice.Repository TenderRepo tender.TenderRepository GLM *glm.SDK - Scraper scraper.SDK ProcessingLimit int } -func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, glmService *glm.SDK, scraper scraper.SDK, processingLimit int) *NoticeWorker { +func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, glmService *glm.SDK, processingLimit int) *NoticeWorker { if processingLimit < 0 { processingLimit = 5 // Default limit } @@ -39,7 +37,6 @@ func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notif NoticeRepo: noticeRepo, TenderRepo: tenderRepo, GLM: glmService, - Scraper: scraper, ProcessingLimit: processingLimit, } } @@ -204,65 +201,6 @@ func (w *NoticeWorker) Run() { w.cleanupProcessedNotices() } -// 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) if err != nil { diff --git a/cmd/worker/workers/scraper.go b/cmd/worker/workers/scraper.go deleted file mode 100644 index 62b6887..0000000 --- a/cmd/worker/workers/scraper.go +++ /dev/null @@ -1,200 +0,0 @@ -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 -} diff --git a/cmd/worker/workers/scraper_consumer.go b/cmd/worker/workers/scraper_consumer.go deleted file mode 100644 index 14c6a8f..0000000 --- a/cmd/worker/workers/scraper_consumer.go +++ /dev/null @@ -1,422 +0,0 @@ -package workers - -import ( - "context" - "fmt" - "time" - tenderPkg "tm/internal/tender" - "tm/pkg/logger" - "tm/pkg/mongo" - "tm/pkg/queue" - "tm/pkg/scraper" -) - -// DocumentScraperConsumerWorker processes scraping jobs from the queue continuously -type DocumentScraperConsumerWorker struct { - Mongo *mongo.ConnectionManager - Logger logger.Logger - TenderRepo tenderPkg.TenderRepository - Queue queue.Queue - Scraper scraper.SDK - Concurrency int - StopChan chan struct{} -} - -// NewDocumentScraperConsumerWorker creates a new consumer worker -func NewDocumentScraperConsumerWorker( - mongo *mongo.ConnectionManager, - logger logger.Logger, - tenderRepo tenderPkg.TenderRepository, - queue queue.Queue, - scraperService scraper.SDK, - concurrency int, -) *DocumentScraperConsumerWorker { - if concurrency <= 0 { - concurrency = 5 // Default concurrency - } - - logger.Info("Creating new document scraper consumer worker", map[string]interface{}{ - "concurrency": concurrency, - }) - - return &DocumentScraperConsumerWorker{ - Mongo: mongo, - Logger: logger, - TenderRepo: tenderRepo, - Queue: queue, - Scraper: scraperService, - Concurrency: concurrency, - StopChan: make(chan struct{}), - } -} - -// Run starts consuming jobs from the queue with specified concurrency -func (w *DocumentScraperConsumerWorker) Run() { - w.Logger.Info("Document scraper consumer worker started", map[string]interface{}{ - "concurrency": w.Concurrency, - }) - - // Create worker pool channels - jobsChan := make(chan *queue.DocumentScraperJob, w.Concurrency) - resultsChan := make(chan *ProcessingResult, w.Concurrency) - - // Start worker goroutines - for i := 0; i < w.Concurrency; i++ { - go w.processJobs(i, jobsChan, resultsChan) - } - - // Start job fetcher - go w.fetchJobs(jobsChan) - - // Start result handler - go w.handleResults(resultsChan) - - // Wait for stop signal - <-w.StopChan - w.Logger.Info("Document scraper consumer worker stopping", map[string]interface{}{}) - - close(jobsChan) - close(resultsChan) -} - -// Stop signals the worker to stop -func (w *DocumentScraperConsumerWorker) Stop() { - w.Logger.Info("Stop signal sent to consumer worker", map[string]interface{}{}) - close(w.StopChan) -} - -// fetchJobs continuously fetches jobs from the queue and sends them to workers -func (w *DocumentScraperConsumerWorker) fetchJobs(jobsChan chan<- *queue.DocumentScraperJob) { - defer func() { - w.Logger.Info("Job fetcher stopped", map[string]interface{}{}) - }() - - for { - select { - case <-w.StopChan: - return - default: - // Try to dequeue a job - ctx, cancel := context.WithTimeout(context.Background(), 35*time.Second) - job, err := w.Queue.Dequeue(ctx) - cancel() - - if err != nil { - w.Logger.Error("Failed to dequeue job", map[string]interface{}{ - "error": err.Error(), - }) - time.Sleep(5 * time.Second) // Back off on error - continue - } - - if job == nil { - // Queue is empty, brief wait before retry - time.Sleep(2 * time.Second) - continue - } - - // Send to processing channel - select { - case jobsChan <- job: - // Job sent successfully - case <-w.StopChan: - return - } - } - } -} - -// processJobs processes jobs from the jobs channel -func (w *DocumentScraperConsumerWorker) processJobs( - workerID int, - jobsChan <-chan *queue.DocumentScraperJob, - resultsChan chan<- *ProcessingResult, -) { - w.Logger.Info("Worker started", map[string]interface{}{ - "worker_id": workerID, - }) - - defer func() { - w.Logger.Info("Worker stopped", map[string]interface{}{ - "worker_id": workerID, - }) - }() - - for { - select { - case job, ok := <-jobsChan: - if !ok { - return // Channel closed - } - - if job == nil { - continue - } - - // Process the job - result := w.processJob(job, workerID) - select { - case resultsChan <- result: - // Result sent successfully - case <-w.StopChan: - return - } - - case <-w.StopChan: - return - } - } -} - -// processJob executes the actual scraping task -func (w *DocumentScraperConsumerWorker) processJob( - job *queue.DocumentScraperJob, - workerID int, -) *ProcessingResult { - startTime := time.Now() - - w.Logger.Info("Processing job", map[string]interface{}{ - "worker_id": workerID, - "job_id": job.ID, - "notice_id": job.NoticeID, - "attempt": job.Attempts, - "max_retries": job.MaxRetries, - }) - - result := &ProcessingResult{ - JobID: job.ID, - WorkerID: workerID, - StartedAt: startTime, - Job: job, - } - - // Skip if missing required fields - if job.NoticeID == "" || job.NoticeURL == "" { - result.Success = false - result.Error = "missing required fields: notice_id or notice_url" - result.CompletedAt = time.Now() - return result - } - - // Call scraper SDK with timeout - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) - defer cancel() - - scrapeReq := &scraper.ScrapeRequest{ - NoticeID: job.NoticeID, - NoticeURL: job.NoticeURL, - } - - w.Logger.Info("Calling scraper service", map[string]interface{}{ - "job_id": job.ID, - "notice_id": job.NoticeID, - "notice_url": job.NoticeURL, - "worker_id": workerID, - }) - - response, err := w.Scraper.ScrapeDocuments(ctx, scrapeReq) - if err != nil { - result.Success = false - result.Error = err.Error() - result.CompletedAt = time.Now() - result.ShouldRetry = job.Attempts < job.MaxRetries - - w.Logger.Error("Scraping failed", map[string]interface{}{ - "job_id": job.ID, - "worker_id": workerID, - "attempt": job.Attempts, - "error": err.Error(), - "should_retry": result.ShouldRetry, - }) - - return result - } - - if !response.Success { - result.Success = false - result.Error = "scraper returned failure" - if len(response.Errors) > 0 { - result.Error += ": " + response.Errors[0].Error - } - result.CompletedAt = time.Now() - result.ShouldRetry = job.Attempts < job.MaxRetries - - w.Logger.Warn("Scraping reported failure", map[string]interface{}{ - "job_id": job.ID, - "worker_id": workerID, - "error_count": len(response.Errors), - "should_retry": result.ShouldRetry, - }) - - return result - } - - // Update tender with scraped documents - scrapedAt := time.Now().Unix() - tender, err := w.TenderRepo.GetByNoticePublicationID(context.Background(), job.NoticeID) - if err != nil { - result.Success = false - result.Error = "failed to fetch tender: " + err.Error() - result.CompletedAt = time.Now() - result.ShouldRetry = job.Attempts < job.MaxRetries - - w.Logger.Error("Failed to fetch tender for update", map[string]interface{}{ - "job_id": job.ID, - "notice_id": job.NoticeID, - "error": err.Error(), - "worker_id": workerID, - }) - - return result - } - - if tender == nil { - result.Success = false - result.Error = "tender not found" - result.CompletedAt = time.Now() - - w.Logger.Warn("Tender not found for scraping result", map[string]interface{}{ - "job_id": job.ID, - "notice_id": job.NoticeID, - "worker_id": workerID, - }) - - return result - } - - // Update tender with scraped documents - tender.ScrapedDocuments = make([]tenderPkg.ScrapedDocument, len(response.UploadedFiles)) - for i, uploadedFile := range response.UploadedFiles { - tender.ScrapedDocuments[i] = tenderPkg.ScrapedDocument{ - Filename: uploadedFile.Filename, - ObjectName: uploadedFile.ObjectName, - BucketName: "opplens", // Default bucket - Size: uploadedFile.Size, - ScrapedAt: scrapedAt, - } - } - - // Update metadata - tender.ProcessingMetadata.DocumentsScraped = true - tender.ProcessingMetadata.DocumentsScrapedAt = scrapedAt - - // Save to database - if err := w.TenderRepo.Update(context.Background(), tender); err != nil { - result.Success = false - result.Error = "failed to update tender: " + err.Error() - result.CompletedAt = time.Now() - result.ShouldRetry = job.Attempts < job.MaxRetries - - w.Logger.Error("Failed to update tender with scraped documents", map[string]interface{}{ - "job_id": job.ID, - "tender_id": tender.ID.Hex(), - "error": err.Error(), - "worker_id": workerID, - "should_retry": result.ShouldRetry, - }) - - return result - } - - result.Success = true - result.UploadedCount = response.UploadedCount - result.CompletedAt = time.Now() - - w.Logger.Info("Scraping job completed successfully", map[string]interface{}{ - "job_id": job.ID, - "tender_id": tender.ID.Hex(), - "uploaded_count": response.UploadedCount, - "worker_id": workerID, - "duration_seconds": time.Since(startTime).Seconds(), - }) - - return result -} - -// handleResults processes the results from worker jobs -func (w *DocumentScraperConsumerWorker) handleResults(resultsChan <-chan *ProcessingResult) { - defer func() { - w.Logger.Info("Result handler stopped", map[string]interface{}{}) - }() - - for { - select { - case result, ok := <-resultsChan: - if !ok { - return // Channel closed - } - - if result == nil { - continue - } - - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - - if result.Success { - // Acknowledge successful job - if err := w.Queue.AckJob(ctx, result.JobID); err != nil { - w.Logger.Error("Failed to acknowledge job", map[string]interface{}{ - "job_id": result.JobID, - "error": err.Error(), - }) - } - } else { - // Handle failure - potentially retry - if result.ShouldRetry { - w.Logger.Info("Job will be retried", map[string]interface{}{ - "job_id": result.JobID, - "attempt": result.Job.Attempts, - "error": result.Error, - "retry_in": "30 seconds", - }) - - // Re-enqueue the job for retry - job := result.Job - job.NextRetryTime = time.Now().Add(30 * time.Second).Unix() - job.Status = string(queue.StatusRetrying) - - if err := w.Queue.Enqueue(ctx, job); err != nil { - w.Logger.Error("Failed to re-enqueue job for retry", map[string]interface{}{ - "job_id": job.ID, - "error": err.Error(), - }) - // Still nack it if re-enqueue fails - w.Queue.NackJob(ctx, result.JobID, fmt.Errorf(result.Error)) - } - } else { - // Max retries exceeded, send to DLQ - w.Logger.Error("Job exceeded max retries, sending to DLQ", map[string]interface{}{ - "job_id": result.JobID, - "max_retry": result.Job.MaxRetries, - "error": result.Error, - }) - - if err := w.Queue.NackJob(ctx, result.JobID, fmt.Errorf(result.Error)); err != nil { - w.Logger.Error("Failed to nack job", map[string]interface{}{ - "job_id": result.JobID, - "error": err.Error(), - }) - } - } - } - - cancel() - - case <-w.StopChan: - return - } - } -} - -// ProcessingResult represents the result of processing a job -type ProcessingResult struct { - JobID string - WorkerID int - Success bool - Error string - ShouldRetry bool - UploadedCount int - StartedAt time.Time - CompletedAt time.Time - Job *queue.DocumentScraperJob -} diff --git a/internal/scraper/form.go b/internal/scraper/form.go deleted file mode 100644 index 7556db8..0000000 --- a/internal/scraper/form.go +++ /dev/null @@ -1,43 +0,0 @@ -package scraper - -// ScrapeRequestForm represents the request form for scraping documents -type ScrapeRequestForm struct { - NoticeID string `json:"notice_id" valid:"required,length(1|100)" example:"75896"` - NoticeURL string `json:"notice_url" valid:"required,url" example:"https://tendsign.com/supplier/s_meformsnotice.aspx?MeFormsNoticeId=75896"` -} - -// ScrapeResponse represents the response from scraping operation -type ScrapeResponse struct { - Success bool `json:"success"` - NoticeID string `json:"notice_id"` - UploadedCount int `json:"uploaded_count"` - UploadedFiles []UploadedFileInfo `json:"uploaded_files"` - Errors []FileError `json:"errors,omitempty"` - Buckets []BucketInfo `json:"buckets,omitempty"` -} - -// UploadedFileInfo represents information about an uploaded file -type UploadedFileInfo struct { - Filename string `json:"filename"` - ObjectName string `json:"object_name"` - Size int64 `json:"size"` -} - -// FileError represents an error for a specific file -type FileError struct { - Filename string `json:"filename"` - Error string `json:"error"` -} - -// BucketInfo represents information about a bucket -type BucketInfo struct { - Name string `json:"name"` - NoticeID string `json:"notice_id"` - Count int `json:"count"` -} - -// TEDOneTimeScrapeForm represents the request form for triggering TED one-time scraping -type TEDOneTimeScrapeForm struct { - FromDate string `json:"from_date" valid:"required" example:"01/01/2026"` - ToDate string `json:"to_date" valid:"required" example:"19/02/2026"` -} diff --git a/internal/scraper/handler.go b/internal/scraper/handler.go deleted file mode 100644 index 610ae84..0000000 --- a/internal/scraper/handler.go +++ /dev/null @@ -1,72 +0,0 @@ -package scraper - -import ( - "tm/pkg/response" - - "github.com/labstack/echo/v4" -) - -// Handler handles HTTP requests for scraper operations -type Handler struct { - service Service -} - -// NewHandler creates a new scraper handler -func NewHandler(service Service) *Handler { - return &Handler{ - service: service, - } -} - -// ScrapeDocuments handles document scraping requests -// @Summary Scrape documents from external source -// @Description Scrapes documents from TendSign and uploads them to MinIO storage -// @Tags Admin-Scraper -// @Accept json -// @Produce json -// @Param request body ScrapeRequestForm true "Scraping request information" -// @Success 200 {object} response.Response{data=ScrapeResponse} -// @Failure 400 {object} response.Response -// @Failure 500 {object} response.Response -// @Router /admin/v1/scraper/download [post] -func (h *Handler) ScrapeDocuments(c echo.Context) error { - form, err := response.Parse[ScrapeRequestForm](c) - if err != nil { - return response.ValidationError(c, "Invalid request data", err.Error()) - } - - result, err := h.service.ScrapeDocuments(c.Request().Context(), form) - if err != nil { - return response.InternalServerError(c, "Failed to scrape documents") - } - - return response.Success(c, result, "Documents scraped and uploaded successfully") -} - -// TriggerTEDOneTimeScraping handles TED one-time scraping requests -// @Summary Trigger TED one-time scraping -// @Description Starts TED one-time scraping for the specified date range. The operation runs asynchronously; check logs for progress. -// @Tags Admin-Scraper -// @Accept json -// @Produce json -// @Param request body TEDOneTimeScrapeForm true "Date range for scraping (DD/MM/YYYY)" -// @Success 202 {object} response.APIResponse "TED scraping started" -// @Failure 400 {object} response.APIResponse "Invalid date format or range" -// @Failure 500 {object} response.APIResponse "TED scraping not configured" -// @Security BearerAuth -// @Router /admin/v1/scraper/ted/one-time [post] -func (h *Handler) TriggerTEDOneTimeScraping(c echo.Context) error { - form, err := response.Parse[TEDOneTimeScrapeForm](c) - if err != nil { - return response.ValidationError(c, "Invalid request data", err.Error()) - } - - err = h.service.TriggerTEDOneTimeScraping(c.Request().Context(), form) - if err != nil { - return response.BadRequest(c, err.Error(), "") - } - - return response.Accepted(c, map[string]string{ - "message": "TED one-time scraping started. Check logs for progress.", - }, "TED one-time scraping started") -} diff --git a/internal/scraper/service.go b/internal/scraper/service.go deleted file mode 100644 index 5e47e03..0000000 --- a/internal/scraper/service.go +++ /dev/null @@ -1,213 +0,0 @@ -package scraper - -import ( - "context" - "fmt" - "time" - - "tm/internal/notice" - "tm/pkg/logger" - "tm/pkg/mongo" - "tm/pkg/notification" - "tm/pkg/scraper" - "tm/ted" -) - -// TEDConfig holds configuration for TED one-time scraping (used when creating service) -type TEDConfig struct { - BaseURL string - Timeout time.Duration - MaxRetries int - RetryDelay time.Duration - UserAgent string - MaxConcurrency int - DownloadDir string - CleanupAfter time.Duration - ScrapingInterval string - MaxNoticesPerDay int - AlertMail string -} - -// Service defines the interface for scraper business operations -type Service interface { - // ScrapeDocuments scrapes documents and uploads them to MinIO - ScrapeDocuments(ctx context.Context, form *ScrapeRequestForm) (*ScrapeResponse, error) - // TriggerTEDOneTimeScraping starts TED one-time scraping for the given date range (runs in background) - TriggerTEDOneTimeScraping(ctx context.Context, form *TEDOneTimeScrapeForm) error -} - -// scraperService implements Service interface -type scraperService struct { - sdk scraper.SDK - logger logger.Logger - mongoManager *mongo.ConnectionManager - noticeRepo notice.Repository - notificationSDK notification.SDK - tedConfig *TEDConfig -} - -// NewService creates a new scraper service with stored credentials -func NewService(baseURL string, timeout time.Duration, logger logger.Logger) Service { - sdk := scraper.NewClient(baseURL, timeout, logger) - return &scraperService{ - sdk: sdk, - logger: logger, - } -} - -// NewServiceWithTED creates a new scraper service with both TendSign SDK and TED scraping support -func NewServiceWithTED(baseURL string, timeout time.Duration, logger logger.Logger, mongoManager *mongo.ConnectionManager, noticeRepo notice.Repository, notificationSDK notification.SDK, tedConfig *TEDConfig) Service { - sdk := scraper.NewClient(baseURL, timeout, logger) - return &scraperService{ - sdk: sdk, - logger: logger, - mongoManager: mongoManager, - noticeRepo: noticeRepo, - notificationSDK: notificationSDK, - tedConfig: tedConfig, - } -} - -// ScrapeDocuments scrapes documents and uploads them to MinIO -func (s *scraperService) ScrapeDocuments(ctx context.Context, form *ScrapeRequestForm) (*ScrapeResponse, error) { - s.logger.Info("Starting document scraping", map[string]interface{}{ - "notice_id": form.NoticeID, - }) - - // Convert form to SDK request (only dynamic fields) - req := &scraper.ScrapeRequest{ - NoticeID: form.NoticeID, - NoticeURL: form.NoticeURL, - } - - // Call SDK (credentials are already configured) - response, err := s.sdk.ScrapeDocuments(ctx, req) - if err != nil { - s.logger.Error("Failed to scrape documents via SDK", map[string]interface{}{ - "error": err.Error(), - "notice_id": form.NoticeID, - }) - return nil, err - } - - // Convert SDK response to service response - serviceResp := &ScrapeResponse{ - Success: response.Success, - NoticeID: response.NoticeID, - UploadedCount: response.UploadedCount, - UploadedFiles: make([]UploadedFileInfo, len(response.UploadedFiles)), - Errors: make([]FileError, len(response.Errors)), - Buckets: make([]BucketInfo, len(response.Buckets)), - } - - // Convert uploaded files - for i, file := range response.UploadedFiles { - serviceResp.UploadedFiles[i] = UploadedFileInfo{ - Filename: file.Filename, - ObjectName: file.ObjectName, - Size: file.Size, - } - } - - // Convert errors - for i, err := range response.Errors { - serviceResp.Errors[i] = FileError{ - Filename: err.Filename, - Error: err.Error, - } - } - - // Convert buckets - for i, bucket := range response.Buckets { - serviceResp.Buckets[i] = BucketInfo{ - Name: bucket.Name, - NoticeID: bucket.NoticeID, - Count: bucket.Count, - } - } - - s.logger.Info("Document scraping completed", map[string]interface{}{ - "notice_id": serviceResp.NoticeID, - "uploaded_count": serviceResp.UploadedCount, - "success": serviceResp.Success, - }) - - return serviceResp, nil -} - -// TriggerTEDOneTimeScraping starts TED one-time scraping for the given date range. -// The scraping runs in a background goroutine; the handler returns 202 Accepted immediately. -func (s *scraperService) TriggerTEDOneTimeScraping(ctx context.Context, form *TEDOneTimeScrapeForm) error { - if s.tedConfig == nil || s.mongoManager == nil || s.noticeRepo == nil { - return fmt.Errorf("TED scraping is not configured") - } - - from, err := time.Parse("02/01/2006", form.FromDate) - if err != nil { - return fmt.Errorf("invalid from_date format: expected DD/MM/YYYY (e.g. 01/01/2026)") - } - - to, err := time.Parse("02/01/2006", form.ToDate) - if err != nil { - return fmt.Errorf("invalid to_date format: expected DD/MM/YYYY (e.g. 19/02/2026)") - } - - if from.After(to) { - return fmt.Errorf("from_date cannot be after to_date") - } - - tedConfig := s.tedConfig - mongoManager := s.mongoManager - noticeRepo := s.noticeRepo - notificationSDK := s.notificationSDK - log := s.logger - - go func() { - tedScraperConfig := &ted.Config{ - BaseURL: tedConfig.BaseURL, - UserAgent: tedConfig.UserAgent, - DownloadDir: tedConfig.DownloadDir, - MaxRetries: tedConfig.MaxRetries, - MaxConcurrency: tedConfig.MaxConcurrency, - Timeout: tedConfig.Timeout, - RetryDelay: tedConfig.RetryDelay, - CleanupAfter: tedConfig.CleanupAfter, - ScrapingInterval: tedConfig.ScrapingInterval, - AlertMail: tedConfig.AlertMail, - MaxNoticesPerDay: tedConfig.MaxNoticesPerDay, - } - - tedScraper := ted.NewTEDScraper( - tedScraperConfig, - log, - mongoManager, - noticeRepo, - notificationSDK, - ) - - runCtx, cancel := context.WithTimeout(context.Background(), 24*time.Hour) - defer cancel() - - log.Info("Starting TED one-time scraping (triggered via API)", map[string]interface{}{ - "from_date": from.Format("02/01/2006"), - "to_date": to.Format("02/01/2006"), - }) - - err := tedScraper.Run(runCtx, &from, &to) - if err != nil { - log.Error("TED one-time scraping failed", map[string]interface{}{ - "error": err.Error(), - "from_date": from.Format("02/01/2006"), - "to_date": to.Format("02/01/2006"), - }) - return - } - - log.Info("TED one-time scraping completed successfully", map[string]interface{}{ - "from_date": from.Format("02/01/2006"), - "to_date": to.Format("02/01/2006"), - }) - }() - - return nil -} diff --git a/pkg/scraper/sdk.go b/pkg/scraper/sdk.go deleted file mode 100644 index e7fb51c..0000000 --- a/pkg/scraper/sdk.go +++ /dev/null @@ -1,246 +0,0 @@ -package scraper - -import ( - "bytes" - "context" - "encoding/json" - "fmt" - "io" - "net/http" - "time" - - "tm/pkg/logger" -) - -// SDK represents the scraper SDK interface -type SDK interface { - // ScrapeDocuments scrapes documents for a given notice - ScrapeDocuments(ctx context.Context, req *ScrapeRequest) (*ScrapeResponse, error) - - // ListBuckets lists all buckets with notice IDs - ListBuckets(ctx context.Context) (*ListBucketsResponse, error) - - // HealthCheck checks if the scraper service is healthy - HealthCheck(ctx context.Context) error -} - -// ScrapeRequest represents a request to scrape documents (only dynamic fields) -type ScrapeRequest struct { - NoticeID string `json:"notice_id"` - NoticeURL string `json:"notice_url"` -} - -// ScrapeResponse represents the response from scraping operation -type ScrapeResponse struct { - Success bool `json:"success"` - NoticeID string `json:"notice_id"` - UploadedCount int `json:"uploaded_count"` - UploadedFiles []UploadedFile `json:"uploaded_files"` - Errors []FileError `json:"errors,omitempty"` - Buckets []BucketInfo `json:"buckets,omitempty"` -} - -// UploadedFile represents information about an uploaded file -type UploadedFile struct { - Filename string `json:"filename"` - ObjectName string `json:"object_name"` - Size int64 `json:"size"` -} - -// FileError represents an error for a specific file -type FileError struct { - Filename string `json:"filename"` - Error string `json:"error"` -} - -// BucketInfo represents information about a bucket -type BucketInfo struct { - Name string `json:"name"` - NoticeID string `json:"notice_id"` - Count int `json:"count"` -} - -// ListBucketsResponse represents the response from listing buckets -type ListBucketsResponse struct { - Buckets []BucketInfo `json:"buckets"` - Total int `json:"total"` -} - -// Client implements the SDK interface -type Client struct { - httpClient *http.Client - baseURL string - logger logger.Logger -} - -// NewClient creates a new scraper SDK client with stored credentials -func NewClient(baseURL string, timeout time.Duration, logger logger.Logger) SDK { - return &Client{ - httpClient: &http.Client{ - Timeout: timeout, - }, - baseURL: baseURL, - logger: logger, - } -} - -// ScrapeDocuments scrapes documents and uploads them to MinIO -func (c *Client) ScrapeDocuments(ctx context.Context, req *ScrapeRequest) (*ScrapeResponse, error) { - c.logger.Info("Starting document scraping via SDK", map[string]interface{}{ - "notice_id": req.NoticeID, - "base_url": c.baseURL, - }) - - // Prepare request body with stored credentials + dynamic fields - requestBody := map[string]interface{}{ - "notice_id": req.NoticeID, - "notice_url": req.NoticeURL, - } - - // Use unified handler API - var response ScrapeResponse - err := c.makeRequest(ctx, "POST", "/download", requestBody, &response) - if err != nil { - return nil, err - } - - // Get buckets list after successful scraping - bucketsResp, err := c.ListBuckets(ctx) - if err != nil { - c.logger.Warn("Failed to list buckets after scraping", map[string]interface{}{ - "error": err.Error(), - }) - } else { - response.Buckets = bucketsResp.Buckets - } - - c.logger.Info("Document scraping completed via SDK", map[string]interface{}{ - "notice_id": response.NoticeID, - "uploaded_count": response.UploadedCount, - "success": response.Success, - }) - - return &response, nil -} - -// ListBuckets lists all buckets with notice IDs -func (c *Client) ListBuckets(ctx context.Context) (*ListBucketsResponse, error) { - var response ListBucketsResponse - err := c.makeRequest(ctx, "GET", "/buckets", nil, &response) - return &response, err -} - -// HealthCheck checks if the scraper service is healthy -func (c *Client) HealthCheck(ctx context.Context) error { - var response map[string]interface{} - err := c.makeRequest(ctx, "GET", "/health", nil, &response) - if err != nil { - return err - } - - // Health check doesn't need to return data, just check if request succeeded - return nil -} - -// Unified HTTP Handler API -// makeRequest handles all HTTP requests with automatic JSON marshaling/unmarshaling -func (c *Client) makeRequest(ctx context.Context, method, endpoint string, requestBody interface{}, responseBody interface{}) error { - // Build full URL - url := fmt.Sprintf("%s%s", c.baseURL, endpoint) - - // Marshal request body to JSON if provided - var bodyReader io.Reader - if requestBody != nil { - jsonData, err := json.Marshal(requestBody) - if err != nil { - c.logger.Error("Failed to marshal request body", map[string]interface{}{ - "error": err.Error(), - "endpoint": endpoint, - }) - return fmt.Errorf("failed to marshal request: %w", err) - } - bodyReader = bytes.NewReader(jsonData) - } - - // Create HTTP request - req, err := http.NewRequestWithContext(ctx, method, url, bodyReader) - if err != nil { - c.logger.Error("Failed to create HTTP request", map[string]interface{}{ - "error": err.Error(), - "method": method, - "url": url, - }) - return fmt.Errorf("failed to create request: %w", err) - } - - // Set headers - if requestBody != nil { - req.Header.Set("Content-Type", "application/json") - } - req.Header.Set("Accept", "application/json") - - // Send request - resp, err := c.httpClient.Do(req) - if err != nil { - c.logger.Error("Failed to send HTTP request", map[string]interface{}{ - "error": err.Error(), - "method": method, - "url": url, - }) - return fmt.Errorf("failed to send request: %w", err) - } - defer resp.Body.Close() - - // Read response body - respBody, err := io.ReadAll(resp.Body) - if err != nil { - c.logger.Error("Failed to read response body", map[string]interface{}{ - "error": err.Error(), - "url": url, - }) - return fmt.Errorf("failed to read response: %w", err) - } - - // Handle error responses - if resp.StatusCode >= 400 { - // Try to parse error response - var errorResp map[string]interface{} - if err := json.Unmarshal(respBody, &errorResp); err == nil { - if errorMsg, ok := errorResp["error"].(string); ok { - c.logger.Error("Request failed with error response", map[string]interface{}{ - "status_code": resp.StatusCode, - "error": errorMsg, - "url": url, - }) - return fmt.Errorf("request failed: %s", errorMsg) - } - } - - c.logger.Error("Request failed with status code", map[string]interface{}{ - "status_code": resp.StatusCode, - "url": url, - "body": string(respBody), - }) - return fmt.Errorf("request failed with status %d", resp.StatusCode) - } - - // Handle successful responses - if responseBody != nil { - if err := json.Unmarshal(respBody, responseBody); err != nil { - c.logger.Error("Failed to unmarshal response", map[string]interface{}{ - "error": err.Error(), - "url": url, - "body": string(respBody), - }) - return fmt.Errorf("failed to unmarshal response: %w", err) - } - } - - c.logger.Debug("Request completed successfully", map[string]interface{}{ - "method": method, - "url": url, - "status_code": resp.StatusCode, - }) - - return nil -}