add csv parsing

service_a/internal/domains/meteo/file.go

@@ -0,0 +1,170 @@
+package meteo
+
+import (
+	"encoding/csv"
+	"fmt"
+	"io"
+	"strconv"
+	"strings"
+	"time"
+)
+
+func (mt *MeteoData) validate() error {
+	if mt.MaxTemp > 60 {
+		return ErrMaxTempOutOfRange
+	}
+
+	if mt.MinTemp < -20 {
+		return ErrMinTempOutOfRange
+	}
+
+	if mt.Rainfall < 0 || mt.Rainfall > 80 {
+		return ErrRainfallOutOfRange
+	}
+
+	if mt.Cloudiness < 0 || mt.Cloudiness > 100 {
+		return ErrCloudinessOutOfRange
+	}
+
+	return nil
+}
+
+type FileIngest interface {
+	Parse(io io.Reader, fs *FileStats) ([]MeteoData, []RejectedMeteoData, error)
+}
+
+type CSV struct{}
+
+var _ FileIngest = (*CSV)(nil)
+
+func (c *CSV) Parse(r io.Reader, fs *FileStats) ([]MeteoData, []RejectedMeteoData, error) {
+	reader := csv.NewReader(r)
+	reader.Comma = ';'
+	reader.TrimLeadingSpace = true
+
+	header, err := reader.Read()
+	if err != nil {
+		return nil, nil, fmt.Errorf("%w: %v", ErrReadingCSVHeader, err)
+	}
+
+	var meteoDataList []MeteoData
+	var rejectedDataList []RejectedMeteoData
+
+	for {
+		row, err := reader.Read()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			return nil, nil, fmt.Errorf("%w: %v", ErrReadingCSVRow, err)
+		}
+
+		if len(row) == 0 || (len(row) == 1 && row[0] == "") {
+			continue
+		}
+
+		rowValue := strings.Join(row, ";")
+
+		record := make(H)
+		for i, value := range row {
+			if i < len(header) {
+				record[header[i]] = value
+			}
+		}
+
+		meteoData, err := normalize(record)
+		if err != nil {
+			fs.RowsRejected++
+			rejectedDataList = append(rejectedDataList, RejectedMeteoData{
+				RowValue: rowValue,
+				Reason:   err.Error(),
+			})
+			continue
+		}
+
+		if err := meteoData.validate(); err != nil {
+			fs.RowsRejected++
+			rejectedDataList = append(rejectedDataList, RejectedMeteoData{
+				RowValue: rowValue,
+				Reason:   err.Error(),
+			})
+			continue
+		}
+
+		meteoDataList = append(meteoDataList, *meteoData)
+		fs.RowsInserted++
+	}
+
+	return meteoDataList, rejectedDataList, nil
+}
+
+func normalize(record H) (*MeteoData, error) {
+	meteoData := &MeteoData{}
+
+	if dateStr, ok := record["Fecha"].(string); ok {
+		t, err := time.Parse("2006/01/02", dateStr)
+		if err != nil {
+			return nil, fmt.Errorf("%w: %v", ErrInvalidDateFormat, err)
+		}
+		meteoData.Timestamp = t
+	} else {
+		return nil, ErrMissingDateField
+	}
+
+	if location, ok := record["Ciudad"].(string); ok {
+		meteoData.Location = location
+	} else {
+		return nil, ErrMissingCityField
+	}
+
+	if maxTempStr, ok := record["Temperatura Máxima (C)"].(string); ok {
+		maxTemp, err := parseFloat(maxTempStr)
+		if err != nil {
+			return nil, fmt.Errorf("%w: %v", ErrInvalidMaxTemp, err)
+		}
+		meteoData.MaxTemp = maxTemp
+	} else {
+		return nil, ErrMissingMaxTempField
+	}
+
+	if minTempStr, ok := record["Temperatura Mínima (C)"].(string); ok {
+		minTemp, err := parseFloat(minTempStr)
+		if err != nil {
+			return nil, fmt.Errorf("%w: %v", ErrInvalidMinTemp, err)
+		}
+		meteoData.MinTemp = minTemp
+	} else {
+		return nil, ErrMissingMinTempField
+	}
+
+	if rainfallStr, ok := record["Precipitación (mm)"].(string); ok {
+		rainfall, err := parseFloat(rainfallStr)
+		if err != nil {
+			return nil, fmt.Errorf("%w: %v", ErrInvalidRainfall, err)
+		}
+		meteoData.Rainfall = rainfall
+	} else {
+		return nil, ErrMissingRainfallField
+	}
+
+	if cloudinessStr, ok := record["Nubosidad (%)"].(string); ok {
+		cloudiness, err := parseFloat(cloudinessStr)
+		if err != nil {
+			return nil, fmt.Errorf("%w: %v", ErrInvalidCloudiness, err)
+		}
+		meteoData.Cloudiness = cloudiness
+	} else {
+		return nil, ErrMissingCloudinessField
+	}
+
+	return meteoData, nil
+}
+
+func parseFloat(s string) (float32, error) {
+	s = strings.Replace(s, ",", ".", 1)
+	f, err := strconv.ParseFloat(s, 32)
+	if err != nil {
+		return 0, err
+	}
+	return float32(f), nil
+}