mazyar/tm_back
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Removed old document scraper service

Browse Source
This commit is contained in:
m.nazemi
2026-05-03 13:46:38 +03:30
parent 0bd5e41b70
commit 446c11dfbf
11 changed files with 9 additions and 1371 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/worker/workers/notice.go
+1 -63
View File
@@ -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 {
cmd/worker/workers/scraper.go
-200
View File
@@ -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
}
cmd/worker/workers/scraper_consumer.go
-422
View File
@@ -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
}