Add TED Scraper Configuration and Cron Scheduler Implementation
- Introduced a new configuration file for the TED scraper, defining database connection settings, logging preferences, and scraping parameters. - Implemented a CronScheduler to manage scheduled tasks for TED operations, utilizing the robfig/cron library for scheduling. - Added new entities and structures for handling TED XML data, including eForms and tender-related information, enhancing the scraper's functionality. - Updated the main application to initialize the TED scraper with the new configuration and set up graceful shutdown handling. - Removed the deprecated scraping implementation to streamline the codebase and focus on the new architecture.
This commit is contained in:
n.nakhostin
+439
@@ -0,0 +1,439 @@
|
||||
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(),
|
||||
¬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: "nakhostin.nima1998@gmail.com",
|
||||
},
|
||||
UserID: "ted-scraper",
|
||||
Type: "alert",
|
||||
})
|
||||
|
||||
return result, 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
|
||||
_, id, noticeType, err := s.xmlParser.Parse(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)
|
||||
}
|
||||
|
||||
// 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,
|
||||
})
|
||||
|
||||
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(now.Year(), now.Format("02/01/2006"))
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to get OJS", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
// parseDate parses various date formats used in TED XML
|
||||
func (s *TEDScraper) parseDate(dateStr string) (time.Time, error) {
|
||||
if dateStr == "" {
|
||||
return time.Time{}, fmt.Errorf("empty date string")
|
||||
}
|
||||
|
||||
// Common TED date formats
|
||||
formats := []string{
|
||||
"2006-01-02+07:00",
|
||||
"2006-01-02-07:00",
|
||||
"2006-01-02T15:04:05+07:00",
|
||||
"2006-01-02T15:04:05-07:00",
|
||||
"2006-01-02T15:04:05Z",
|
||||
"2006-01-02Z",
|
||||
"2006-01-02",
|
||||
time.RFC3339,
|
||||
time.RFC3339Nano,
|
||||
}
|
||||
|
||||
for _, format := range formats {
|
||||
if parsedTime, err := time.Parse(format, dateStr); err == nil {
|
||||
return parsedTime, nil
|
||||
}
|
||||
}
|
||||
|
||||
return time.Time{}, fmt.Errorf("unable to parse date: %s", dateStr)
|
||||
}
|
||||
|
||||
// parseDateToUnixMilli converts string date to Unix milliseconds (int64)
|
||||
func (s *TEDScraper) parseDateToUnixMilli(dateStr string) int64 {
|
||||
if dateStr == "" {
|
||||
return 0
|
||||
}
|
||||
|
||||
if parsedTime, err := s.parseDate(dateStr); err == nil {
|
||||
return parsedTime.UnixMilli()
|
||||
}
|
||||
|
||||
return 0
|
||||
}
|
||||
|
||||
// parseTimeToUnixMilli converts string time to Unix milliseconds (int64)
|
||||
func (s *TEDScraper) parseTimeToUnixMilli(timeStr string) int64 {
|
||||
if timeStr == "" {
|
||||
return 0
|
||||
}
|
||||
|
||||
// If time string is just time without date, use today's date
|
||||
if !strings.Contains(timeStr, "T") && !strings.Contains(timeStr, " ") {
|
||||
// Assume it's just a time like "15:04:05"
|
||||
today := time.Now().Format("2006-01-02")
|
||||
timeStr = fmt.Sprintf("%sT%s", today, timeStr)
|
||||
}
|
||||
|
||||
if parsedTime, err := s.parseDate(timeStr); err == nil {
|
||||
return parsedTime.UnixMilli()
|
||||
}
|
||||
|
||||
return 0
|
||||
}
|
||||
|
||||
// unixMilliToDateString converts Unix milliseconds to date string for TenderID generation
|
||||
func (s *TEDScraper) unixMilliToDateString(unixMilli int64) string {
|
||||
if unixMilli == 0 {
|
||||
return "00000000"
|
||||
}
|
||||
|
||||
return time.UnixMilli(unixMilli).Format("20060102")
|
||||
}
|
||||
|
||||
func (s *TEDScraper) calculateSubmissionDeadline(publicationDate time.Time) time.Time {
|
||||
deadline := publicationDate
|
||||
workingHoursAdded := 0
|
||||
|
||||
for workingHoursAdded < 48 {
|
||||
deadline = deadline.Add(1 * time.Hour)
|
||||
|
||||
if deadline.Weekday() != time.Saturday && deadline.Weekday() != time.Sunday {
|
||||
workingHoursAdded++
|
||||
}
|
||||
}
|
||||
|
||||
return deadline
|
||||
}
|
||||
|
||||
func (s *TEDScraper) calculateApplicationDeadline(tenderDeadline time.Time) time.Time {
|
||||
applicationDeadline := tenderDeadline
|
||||
workingDaysSubtracted := 0
|
||||
|
||||
for workingDaysSubtracted < 7 {
|
||||
applicationDeadline = applicationDeadline.AddDate(0, 0, -1)
|
||||
|
||||
if applicationDeadline.Weekday() != time.Saturday && applicationDeadline.Weekday() != time.Sunday {
|
||||
workingDaysSubtracted++
|
||||
}
|
||||
}
|
||||
|
||||
return applicationDeadline
|
||||
}
|
||||
Reference in New Issue
Block a user