tm_back/ted/scraper.go

package ted

import (
	"archive/tar"
	"compress/gzip"
	"context"
	"fmt"
	"io"
	"net/http"
	"path/filepath"
	"strings"
	"time"
	"tm/internal/tender"
	"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"`
}

// 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   tender.TenderRepository
}

// NewTEDScraper creates a new TED scraper instance
func NewTEDScraper(
	config *Config,
	logger logger.Logger,
	mongoManager *mongo.ConnectionManager,
	tenderRepo tender.TenderRepository,
	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
		}
	}

	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: "nakhostin.nima1998@gmail.com",
			},
			UserID: "ted-scraper",
			Type:   "alert",
		})

	return result, nil
}

func (s *TEDScraper) findTenderByContractNoticeID(ctx context.Context, contractNoticeID string) (*tender.Tender, 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,
			"tender_id":          t.TenderID,
		})

		t.CreatedAt = time.Now().Unix()
		t.UpdatedAt = t.CreatedAt

		err = s.tenderRepo.Create(ctx, t)
		if err != nil {
			s.logger.Error("Failed to create new tender", map[string]interface{}{
				"contract_notice_id": t.ContractNoticeID,
				"tender_id":          t.TenderID,
				"error":              err.Error(),
			})
			return fmt.Errorf("failed to create tender: %w", err)
		}

		s.logger.Info("Successfully created new tender", map[string]interface{}{
			"tender_id":          t.TenderID,
			"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_id":   t.TenderID,
		"tender_url":  t.TenderURL,
	})

	result.SuccessCount++
	return nil
}

func (s *TEDScraper) Run(ctx context.Context) error {
	now := time.Now().Local()
	s.logger.Info("Running periodic scraping", map[string]interface{}{
		"time": now.Format(time.RFC3339),
	})

	// get ojs
	ojs, err := GetOJS(s.config.BaseURL, now.Year(), now.Format("02/01/2006"))
	if err != nil {
		s.logger.Error("Failed to get OJS", map[string]interface{}{
			"error": err.Error(),
		})
		return nil
	}

	// download daily package
	result, err := s.DownloadDailyPackage(ctx, ojs)
	if err != nil {
		s.logger.Error("Failed to download daily package", map[string]interface{}{
			"error": err.Error(),
		})
		return err
	}

	// process daily package
	s.logger.Info("Processed daily package successfully", map[string]interface{}{
		"result": result,
		"time":   now.Format(time.RFC3339),
		"ojs":    ojs,
	})

	return nil
}