diff --git a/.env b/.env index 5cdd5ed..0ce64b6 100644 --- a/.env +++ b/.env @@ -8,9 +8,5 @@ DB_SSLMODE=disable SERVER_PORT=8080 SERVER_HOST=localhost -# LinkedIn parsing settings -LINKEDIN_BASE_URL=https://www.linkedin.com -REQUEST_TIMEOUT=30s -RATE_LIMIT_DELAY=2s CV_AI_MODEL=your_ai_model OPENROUTER_API_KEY=your_openrouter_api_key \ No newline at end of file diff --git a/internal/pipeline/test.go b/internal/pipeline/job_pipeline-workers.go similarity index 91% rename from internal/pipeline/test.go rename to internal/pipeline/job_pipeline-workers.go index bee3db1..c4323af 100644 --- a/internal/pipeline/test.go +++ b/internal/pipeline/job_pipeline-workers.go @@ -25,13 +25,12 @@ func GetJobs(context context.Context, scraperService *Scraper) <-chan models.Job return jobChan } -func GetJobDescription(context context.Context, scraperService *Scraper, jobChan <-chan models.Job) <-chan models.JobWithDescription { +func GetJobDescription(context context.Context, scraperService *Scraper, jobChan <-chan models.Job, numWorkers int) <-chan models.JobWithDescription { jobDescriptionChan := make(chan models.JobWithDescription, 100) - const numWorkers = 3 var wg sync.WaitGroup - // Start 3 worker goroutines + // Start worker goroutines for range numWorkers { wg.Add(1) go func() { diff --git a/internal/pipeline/job_pipeline.go b/internal/pipeline/job_pipeline.go index 5e10e20..eff384e 100644 --- a/internal/pipeline/job_pipeline.go +++ b/internal/pipeline/job_pipeline.go @@ -11,7 +11,6 @@ import ( "github.com/jobs-scraper/internal/models" "github.com/jobs-scraper/internal/repo" - "golang.org/x/time/rate" ) // JobDescriptionResult represents the result of job description scraping @@ -43,10 +42,9 @@ func (p *JobPipeline) ProcessJobsStreaming(ctx context.Context, numPages int, jo // Create channels for the pipeline allJobs := make([]models.Job, 0, 100) allJobDescriptions := make([]models.JobDescription, 0, 100) - // var mu sync.Mutex var jbMu sync.Mutex jobsChan := GetJobs(ctx, p.scraperService) - jobWithDescriptionChan := GetJobDescription(ctx, p.scraperService, jobsChan) + jobWithDescriptionChan := GetJobDescription(ctx, p.scraperService, jobsChan, 3) for jobWithDescription := range jobWithDescriptionChan { fmt.Printf("Received job description for job : %d\n", jobWithDescription.Job.ID) @@ -66,42 +64,3 @@ func (p *JobPipeline) ProcessJobsStreaming(ctx context.Context, numPages int, jo return nil } - -// jobDescriptionWorker processes jobs from jobChan and sends results to jobDescriptionChan -func (p *JobPipeline) jobDescriptionWorker(ctx context.Context, jobChan <-chan models.Job, resultChan chan<- JobDescriptionResult) { - - limiter := rate.NewLimiter(rate.Every(2*time.Second), 1) // 1 request per second - - for { - select { - case <-ctx.Done(): - return - case job, ok := <-jobChan: - if !ok { - return - } - - if err := limiter.Wait(ctx); err != nil { - return - } - - fmt.Printf("Processing job: %s\n", job.Title) - - description, criteria, err := p.scraperService.ScrapeJobDescriptionWithContext(ctx, job) - - result := JobDescriptionResult{ - Job: job, - Description: description, - Criteria: criteria, - Error: err, - } - - select { - case resultChan <- result: - case <-ctx.Done(): - return - } - - } - } -} diff --git a/internal/pipeline/scraper.go b/internal/pipeline/scraper.go index d33108d..63bea35 100644 --- a/internal/pipeline/scraper.go +++ b/internal/pipeline/scraper.go @@ -88,7 +88,13 @@ func (s *Scraper) ScrapeJobsWithContext(ctx context.Context, page int, params mo jobs := make([]models.Job, 0, 10) url := s.buildSearchURL(params, page) - res, err := utils.RetryableHTTPRequest(ctx, url) + retryableRequest := utils.NewRetryableHTTPRequest(utils.RetryConfig{ + BaseDelay: s.config.BaseDelay, + MaxDelay: s.config.MaxDelay, + MaxRetries: s.config.MaxRetries, + }) + + res, err := retryableRequest.RetryableHTTPRequest(ctx, url, "GET", nil, nil) if err != nil { fmt.Printf("Error fetching URL after retries: %v\n", err) return jobs, err @@ -132,7 +138,13 @@ func (s *Scraper) ScrapeJobDescriptionWithContext(ctx context.Context, job model var jobDescription string url := s.buildJobDescriptionSearchURL(job.JobLink) - res, err := utils.RetryableHTTPRequest(ctx, url) + retryableRequest := utils.NewRetryableHTTPRequest(utils.RetryConfig{ + MaxRetries: s.config.MaxRetries, + BaseDelay: s.config.BaseDelay, + MaxDelay: s.config.MaxDelay, + }) + + res, err := retryableRequest.RetryableHTTPRequest(ctx, url, "GET", nil, nil) if err != nil { fmt.Printf("Error fetching job description URL after retries: %v\n", err) return "", map[string]string{}, err diff --git a/internal/utils/retryable-http-request.go b/internal/utils/retryable-http-request.go index f459605..3868113 100644 --- a/internal/utils/retryable-http-request.go +++ b/internal/utils/retryable-http-request.go @@ -3,7 +3,7 @@ package utils import ( "context" "fmt" - "math" + "io" "net/http" "time" ) @@ -14,23 +14,39 @@ type RetryConfig struct { MaxDelay time.Duration } -func RetryableHTTPRequest(ctx context.Context, url string) (*http.Response, error) { +type RetryableHTTPRequestImpl struct { + client *http.Client + config RetryConfig +} + +func NewRetryableHTTPRequest(config RetryConfig) *RetryableHTTPRequestImpl { client := &http.Client{ Timeout: 30 * time.Second, } + return &RetryableHTTPRequestImpl{ + client: client, + config: config, + } +} + +func (s *RetryableHTTPRequestImpl) RetryableHTTPRequest(ctx context.Context, url, method string, body io.Reader, headers []http.Header) (*http.Response, error) { + var lastErr error - for attempt := 0; attempt <= 3; attempt++ { - // Create request with context - req, err := http.NewRequestWithContext(ctx, "GET", url, nil) + for attempt := 0; attempt <= s.config.MaxRetries; attempt++ { + req, err := http.NewRequestWithContext(ctx, method, url, body) + if err != nil { return nil, fmt.Errorf("failed to create request: %w", err) } req.Header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36") + for _, header := range headers { + req.Header.Set(header.Get("Key"), header.Get("Value")) + } - resp, err := client.Do(req) + resp, err := s.client.Do(req) if err != nil { lastErr = err fmt.Printf("Request attempt %d failed: %v\n", attempt+1, err) @@ -49,10 +65,10 @@ func RetryableHTTPRequest(ctx context.Context, url string) (*http.Response, erro } // max attempts for now = 3 - if attempt < 3 { - delay := min(time.Duration(math.Pow(2, float64(attempt)))*1*time.Second, 30*time.Second) + if attempt < s.config.MaxRetries { + delay := time.Second * 2 - fmt.Printf("Retrying in %v... (attempt %d/%d)\n", delay, attempt+1, 3) + fmt.Printf("Retrying in %v... (attempt %d/%d)\n", delay, attempt+1, s.config.MaxRetries) select { case <-ctx.Done():