tm_back/ted/scraper.go

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"`
}

// 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
		}

		// 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(),
			&notification.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)

	// existingTender, err := s.findTenderByContractNoticeID(ctx, t.ContractNoticeID)
	// if err != nil {
	// 	s.logger.Warn("Error checking for existing tender", map[string]interface{}{
	// 		"contract_notice_id": t.ContractNoticeID,
	// 		"error":              err.Error(),
	// 	})
	// }

	// if err == nil && existingTender != nil {
	// 	// Update existing tender
	// 	s.logger.Info("Found existing tender, updating", map[string]interface{}{
	// 		"existing_tender_id": existingTender.ID.Hex(),
	// 		"contract_notice_id": t.ContractNoticeID,
	// 	})

	// 	t.ID = existingTender.ID
	// 	t.UpdatedAt = time.Now().Unix()
	// 	err = s.tenderRepo.Update(ctx, t)
	// 	if err != nil {
	// 		s.logger.Error("Failed to update existing tender", map[string]interface{}{
	// 			"tender_id": t.ID.Hex(),
	// 			"error":     err.Error(),
	// 		})
	// 		return fmt.Errorf("failed to update existing tender: %w", err)
	// 	}

	// 	s.logger.Info("Successfully updated tender", map[string]interface{}{
	// 		"tender_id": t.ID.Hex(),
	// 	})
	// } else {
	// 	// 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"))
	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
}