package ted import ( "archive/tar" "compress/gzip" "context" "fmt" "io" "net/http" "path/filepath" "strings" "time" "tm/internal/notice" "tm/pkg/logger" "tm/pkg/mongo" "tm/pkg/notification" ) // ScraperConfig holds configuration for the TED scraper type Config struct { BaseURL string `mapstructure:"base_url"` UserAgent string `mapstructure:"user_agent"` DownloadDir string `mapstructure:"download_dir"` MaxRetries int `mapstructure:"max_retries"` MaxConcurrency int `mapstructure:"max_concurrency"` Timeout time.Duration `mapstructure:"timeout"` RetryDelay time.Duration `mapstructure:"retry_delay"` CleanupAfter time.Duration `mapstructure:"cleanup_after"` ScrapingInterval string `mapstructure:"scraping_interval"` AlertMail string `mapstructure:"ALERT_MAIL"` MaxNoticesPerDay int `mapstructure:"max_notices_per_day"` } // TEDScraper handles downloading and parsing TED XML files type TEDScraper struct { notify notification.SDK config *Config logger logger.Logger httpClient *http.Client xmlParser *TEDParser mongoManager *mongo.ConnectionManager tenderRepo notice.Repository } // NewTEDScraper creates a new TED scraper instance func NewTEDScraper( config *Config, logger logger.Logger, mongoManager *mongo.ConnectionManager, tenderRepo notice.Repository, notify notification.SDK, ) *TEDScraper { httpClient := &http.Client{ Timeout: config.Timeout, } return &TEDScraper{ notify: notify, config: config, logger: logger, httpClient: httpClient, xmlParser: NewTEDParser(), mongoManager: mongoManager, tenderRepo: tenderRepo, } } // downloadDailyPackage downloads a file from a URL func (s *TEDScraper) DownloadDailyPackage(ctx context.Context, ojs string) (*FileProcessingResult, error) { var ( result *FileProcessingResult err error ) url := fmt.Sprintf("%s/packages/daily/%s", s.config.BaseURL, ojs) s.logger.Info("Downloading TED file", map[string]interface{}{ "url": url, "ojs": ojs, }) for attempt := 1; attempt <= s.config.MaxRetries; attempt++ { result, err = s.downloadWithRetry(ctx, url, ojs) if err == nil { break } if attempt < s.config.MaxRetries { s.logger.Warn("Download attempt failed, retrying", map[string]interface{}{ "attempt": attempt, "error": err.Error(), "url": url, }) select { case <-ctx.Done(): return result, ctx.Err() case <-time.After(s.config.RetryDelay): // Continue to next attempt } } } return result, err } // FileProcessingResult represents the result of processing a single file type FileProcessingResult struct { OJS string `json:"ojs"` ProcessedCount int `json:"processed_count"` SuccessCount int `json:"success_count"` Errors []string `json:"errors"` ProcessedAt int64 `json:"processed_at"` } func (s *TEDScraper) downloadWithRetry(ctx context.Context, url string, ojs string) (*FileProcessingResult, error) { // Create a longer timeout context for processing large files processCtx, cancel := context.WithTimeout(ctx, s.config.Timeout) defer cancel() // Create HTTP request req, err := http.NewRequestWithContext(processCtx, "GET", url, nil) if err != nil { return &FileProcessingResult{}, fmt.Errorf("failed to create request: %w", err) } req.Header.Set("User-Agent", s.config.UserAgent) s.logger.Info("Starting download", map[string]interface{}{ "url": url, "ojs": ojs, }) // Make request resp, err := s.httpClient.Do(req) if err != nil { return &FileProcessingResult{}, fmt.Errorf("failed to download file: %w", err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { return &FileProcessingResult{}, fmt.Errorf("HTTP error: %d %s", resp.StatusCode, resp.Status) } s.logger.Info("Download completed, starting processing", map[string]interface{}{ "url": url, "ojs": ojs, "content_length": resp.Header.Get("Content-Length"), }) return s.tarGzFileProcessor(processCtx, resp.Body, ojs) } // tarGzFileProcessor extracts and processes XML files from a tar.gz archive func (s *TEDScraper) tarGzFileProcessor(ctx context.Context, reader io.Reader, ojs string) (*FileProcessingResult, error) { // Create gzip reader gzipReader, err := gzip.NewReader(reader) if err != nil { return &FileProcessingResult{}, fmt.Errorf("failed to create gzip reader: %w", err) } defer gzipReader.Close() // Create tar reader tarReader := tar.NewReader(gzipReader) // Process each file in the tar archive fileCount := 0 result := &FileProcessingResult{ OJS: ojs, ProcessedCount: 0, SuccessCount: 0, Errors: make([]string, 0), ProcessedAt: time.Now().Unix(), } for { // Check context for cancellation before each tar entry select { case <-ctx.Done(): s.logger.Warn("Context cancelled during tar processing", map[string]interface{}{ "processed_files": fileCount, "error": ctx.Err().Error(), }) return &FileProcessingResult{}, ctx.Err() default: } header, err := tarReader.Next() if err == io.EOF { break // End of archive } if err != nil { s.logger.Warn("Error reading tar entry", map[string]interface{}{ "error": err.Error(), "processed_files": fileCount, }) // If it's a timeout error, return it immediately if ctx.Err() != nil { return &FileProcessingResult{}, fmt.Errorf("timeout during tar processing: %w", err) } continue } fileCount++ // Skip directories and non-XML files if header.Typeflag != tar.TypeReg { continue } fileName := filepath.Base(header.Name) if !strings.HasSuffix(strings.ToLower(fileName), ".xml") { continue } // Log progress every 10 files or for first file if result.ProcessedCount == 0 || result.ProcessedCount%10 == 0 { s.logger.Info("Processing XML file from tar.gz", map[string]interface{}{ "file_name": fileName, "size": header.Size, "processed_count": result.ProcessedCount, "total_files_seen": fileCount, "ojs": ojs, }) } else { s.logger.Debug("Processing XML file from tar.gz", map[string]interface{}{ "file_name": fileName, "processed_count": result.ProcessedCount, "ojs": ojs, }) } // Read file content content, err := io.ReadAll(tarReader) if err != nil { errMsg := fmt.Sprintf("failed to read XML file %s: %v", fileName, err) result.Errors = append(result.Errors, errMsg) continue } // Skip processing if per-day limit reached (for UAT/low storage) if s.config.MaxNoticesPerDay > 0 && result.SuccessCount >= s.config.MaxNoticesPerDay { s.logger.Info("Reached max notices per day limit, stopping processing", map[string]interface{}{ "limit": s.config.MaxNoticesPerDay, "ojs": ojs, }) break } // Parse and process XML content if err := s.processXMLContent(ctx, content, fileName, result); err != nil { errMsg := fmt.Sprintf("failed to process XML file %s: %v", fileName, err) result.Errors = append(result.Errors, errMsg) } result.ProcessedCount++ // Check context for cancellation select { case <-ctx.Done(): return result, ctx.Err() default: // Continue processing } } if s.config.AlertMail != "" { go s.notify.SendNotification( context.Background(), ¬ification.NotificationRequest{ EventType: notification.EventTypeEmail, Title: "TED scraper completed", Message: GenerateFileProcessingEmailTemplate( &FileProcessingEmailTemplateData{ OJS: ojs, ProcessedCount: result.ProcessedCount, SuccessCount: result.SuccessCount, Errors: result.Errors, ProcessedAt: result.ProcessedAt, }), Priority: "important", Methods: notification.NotificationMethods{ Email: s.config.AlertMail, }, UserID: "ted-scraper", Type: "alert", }) } return result, nil } func (s *TEDScraper) findTenderByContractNoticeID(ctx context.Context, contractNoticeID string) (*notice.Notice, error) { existingTender, err := s.tenderRepo.GetByContractNoticeID(ctx, contractNoticeID) if err != nil { if err.Error() == "document not found" { return nil, nil } return nil, err } return existingTender, nil } func (s *TEDScraper) processXMLContent(ctx context.Context, content []byte, fileName string, result *FileProcessingResult) error { s.logger.Debug("Processing XML content", map[string]interface{}{ "file_name": fileName, "content_size": len(content), "first_1000_chars": string(content[:min(1000, len(content))]), }) // Parse XML using the flexible TED parser that detects notice type notice, id, noticeType, err := s.xmlParser.ParseXML(content) if err != nil { s.logger.Error("XML parsing failed", map[string]interface{}{ "file_name": fileName, "error": err.Error(), "content_preview": string(content[:min(500, len(content))]), }) return fmt.Errorf("failed to parse XML: %w", err) } // Map to tender entity t := s.mapToTenderFromParsedDoc(notice, "", fileName) if t == nil { s.logger.Warn("Skipping document - mapper returned nil (unsupported or incomplete document)", map[string]interface{}{ "file_name": fileName, "notice_id": id, "notice_type": noticeType, }) return nil } t.ContentXML = string(content) // Check if tender already exists to prevent duplicates existingTender, err := s.findTenderByContractNoticeID(ctx, t.ContractNoticeID) if err != nil { s.logger.Error("Failed to check existing tender", map[string]interface{}{ "contract_notice_id": t.ContractNoticeID, "error": err.Error(), }) return fmt.Errorf("failed to check existing tender: %w", err) } if existingTender != nil { s.logger.Info("Tender already exists, skipping duplicate", map[string]interface{}{ "contract_notice_id": t.ContractNoticeID, "existing_id": existingTender.ID, }) return nil } // Create new tender s.logger.Info("Creating new tender", map[string]interface{}{ "contract_notice_id": t.ContractNoticeID, }) t.CreatedAt = time.Now().Unix() t.UpdatedAt = t.CreatedAt err = s.tenderRepo.Import(ctx, t) if err != nil { s.logger.Error("Failed to create new tender", map[string]interface{}{ "contract_notice_id": t.ContractNoticeID, "error": err.Error(), }) return fmt.Errorf("failed to create tender: %w", err) } s.logger.Info("Successfully created new tender", map[string]interface{}{ "contract_notice_id": t.ContractNoticeID, }) // } // Log detailed parsed information based on notice type s.logger.Info("Successfully parsed notice", map[string]interface{}{ "notice_id": id, "notice_type": noticeType, "file_name": fileName, "tender_url": t.TenderURL, }) result.SuccessCount++ return nil } func (s *TEDScraper) Run(ctx context.Context, from, to *time.Time) error { // If no date range is provided, use current date if from == nil && to == nil { now := time.Now().Local() s.logger.Info("Running periodic scraping for current date", map[string]interface{}{ "date": now.Format("02/01/2006"), }) return s.processSingleDate(ctx, now) } // Validate date range if from == nil || to == nil { return fmt.Errorf("both from and to dates must be provided for date range scraping") } if from.After(*to) { return fmt.Errorf("from date cannot be after to date") } s.logger.Info("Running date range scraping", map[string]interface{}{ "from_date": from.Format("02/01/2006"), "to_date": to.Format("02/01/2006"), }) // Process each date in the range current := *from totalDays := 0 successfulDays := 0 var errors []string for !current.After(*to) { totalDays++ s.logger.Info("Processing date", map[string]interface{}{ "date": current.Format("02/01/2006"), "progress": fmt.Sprintf("%d/%d", totalDays, int(to.Sub(*from).Hours()/24)+1), }) err := s.processSingleDate(ctx, current) if err != nil { errorMsg := fmt.Sprintf("failed to process date %s: %v", current.Format("02/01/2006"), err) errors = append(errors, errorMsg) s.logger.Error("Failed to process date", map[string]interface{}{ "date": current.Format("02/01/2006"), "error": err.Error(), }) } else { successfulDays++ } // Check for context cancellation select { case <-ctx.Done(): s.logger.Warn("Date range scraping cancelled", map[string]interface{}{ "processed_days": totalDays, "successful_days": successfulDays, "cancelled_at": current.Format("02/01/2006"), }) return ctx.Err() default: } // Move to next day current = current.AddDate(0, 0, 1) } s.logger.Info("Date range scraping completed", map[string]interface{}{ "total_days": totalDays, "successful_days": successfulDays, "failed_days": len(errors), "errors": errors, }) // Return error if no days were processed successfully if successfulDays == 0 && len(errors) > 0 { return fmt.Errorf("failed to process any dates in range: %v", errors) } return nil } // processSingleDate processes data for a single date func (s *TEDScraper) processSingleDate(ctx context.Context, date time.Time) error { s.logger.Info("Processing single date", map[string]interface{}{ "date": date.Format("02/01/2006"), }) // get ojs ojs, err := GetOJS(s.config.BaseURL, date.Year(), date.Format("02/01/2006"), s.httpClient) if err != nil { s.logger.Error("Failed to get OJS", map[string]interface{}{ "date": date.Format("02/01/2006"), "error": err.Error(), }) return fmt.Errorf("failed to get OJS for date %s: %w", date.Format("02/01/2006"), err) } // download daily package result, err := s.DownloadDailyPackage(ctx, ojs) if err != nil { s.logger.Error("Failed to download daily package", map[string]interface{}{ "date": date.Format("02/01/2006"), "ojs": ojs, "error": err.Error(), }) return fmt.Errorf("failed to download daily package for date %s: %w", date.Format("02/01/2006"), err) } // process daily package s.logger.Info("Processed daily package successfully", map[string]interface{}{ "result": result, "date": date.Format("02/01/2006"), "ojs": ojs, }) return nil }