internal/parser/parser.go

// Package parser provides streaming CSV parsing for GDELT files.
// It parses data line-by-line using channels for pipeline processing.
package parser

import (
	"archive/zip"
	"bufio"
	"bytes"
	"context"
	"fmt"
	"io"
	"strconv"
	"strings"
	"time"

	"gdelt-engine/internal/constants"
	"gdelt-engine/internal/schema"

	"go.uber.org/zap"
)

// ParseResult represents a parsed record sent through channel
type ParseResult struct {
	// Record is the parsed data (Event, Mention, or GKG)
	Record interface{}

	// FileType indicates which type of record this is
	FileType string

	// LineNumber is the line number in the source file
	LineNumber int

	// Err is any parsing error (nil for successful parse)
	Err error
}

// StreamParser defines the interface for streaming CSV parsers
type StreamParser interface {
	// ParseStream reads from io.Reader and sends parsed records to channel
	ParseStream(ctx context.Context, r io.Reader, fileType, timestamp string) <-chan ParseResult

	// ParseZipStream extracts and parses a ZIP stream
	ParseZipStream(ctx context.Context, zipData []byte, fileType, timestamp string) <-chan ParseResult
}

// Compile-time interface verification
var _ StreamParser = (*CSVStreamParser)(nil)

// CSVStreamParser implements StreamParser for GDELT CSV files
type CSVStreamParser struct {
	logger     *zap.Logger
	bufferSize int
	maxLine    int
}

// ParserOption is a functional option for parser configuration
type ParserOption func(*CSVStreamParser)

// WithParserLogger sets the logger
func WithParserLogger(logger *zap.Logger) ParserOption {
	return func(p *CSVStreamParser) {
		p.logger = logger
	}
}

// WithBufferSize sets the scanner buffer size
func WithBufferSize(size int) ParserOption {
	return func(p *CSVStreamParser) {
		p.bufferSize = size
	}
}

// NewCSVStreamParser creates a new streaming CSV parser
func NewCSVStreamParser(opts ...ParserOption) *CSVStreamParser {
	p := &CSVStreamParser{
		logger:     zap.NewNop(),
		bufferSize: constants.InitialBufferSize,
		maxLine:    constants.MaxLineSize,
	}

	for _, opt := range opts {
		opt(p)
	}

	return p
}

// ParseZipStream extracts the first file from ZIP data and parses it as a stream.
// Note: ZIP requires random access, so we need the full data here.
// The parsing itself is still streamed line-by-line.
func (p *CSVStreamParser) ParseZipStream(ctx context.Context, zipData []byte, fileType, timestamp string) <-chan ParseResult {
	out := make(chan ParseResult, constants.ChannelBufferSize)

	go func() {
		defer close(out)

		// Open ZIP from memory
		zipReader, err := zip.NewReader(bytes.NewReader(zipData), int64(len(zipData)))
		if err != nil {
			out <- ParseResult{Err: fmt.Errorf("zip open failed: %w", err)}
			return
		}

		if len(zipReader.File) == 0 {
			out <- ParseResult{Err: fmt.Errorf("empty zip file")}
			return
		}

		// Open first file in ZIP
		csvFile, err := zipReader.File[0].Open()
		if err != nil {
			out <- ParseResult{Err: fmt.Errorf("csv extract failed: %w", err)}
			return
		}
		defer csvFile.Close()

		// Parse the CSV stream
		for result := range p.ParseStream(ctx, csvFile, fileType, timestamp) {
			select {
			case <-ctx.Done():
				out <- ParseResult{Err: ctx.Err()}
				return
			case out <- result:
			}
		}
	}()

	return out
}

// ParseStream creates a pipeline that parses CSV line-by-line using a goroutine.
// It sends parsed records through the returned channel.
func (p *CSVStreamParser) ParseStream(ctx context.Context, r io.Reader, fileType, timestamp string) <-chan ParseResult {
	out := make(chan ParseResult, constants.ChannelBufferSize)

	go func() {
		defer close(out)

		scanner := bufio.NewScanner(r)

		// Set buffer for large lines (especially for GKG)
		buf := make([]byte, 0, p.bufferSize)
		scanner.Buffer(buf, p.maxLine)

		lineNum := 0
		skipped := 0
		now := time.Now()

		for scanner.Scan() {
			lineNum++

			select {
			case <-ctx.Done():
				out <- ParseResult{Err: ctx.Err()}
				return
			default:
			}

			line := scanner.Text()
			if line == "" {
				continue
			}

			record, err := p.parseLine(line, fileType, timestamp, now)
			if err != nil {
				skipped++
				continue // Skip malformed lines
			}

			out <- ParseResult{
				Record:     record,
				FileType:   fileType,
				LineNumber: lineNum,
			}
		}

		if err := scanner.Err(); err != nil {
			out <- ParseResult{Err: fmt.Errorf("scanner error: %w", err)}
		}

		if skipped > 0 {
			p.logger.Debug("Skipped malformed rows",
				zap.Int("count", skipped),
				zap.String("type", fileType),
			)
		}
	}()

	return out
}

// parseLine parses a single CSV line based on file type
func (p *CSVStreamParser) parseLine(line, fileType, timestamp string, now time.Time) (interface{}, error) {
	fields := strings.Split(line, "\t")

	switch fileType {
	case constants.FileTypeExport:
		return p.parseEvent(fields, timestamp, now)
	case constants.FileTypeMentions:
		return p.parseMention(fields, timestamp, now)
	case constants.FileTypeGKG:
		return p.parseGKG(fields, timestamp, now)
	default:
		return nil, fmt.Errorf("unknown file type: %s", fileType)
	}
}

// parseEvent parses an export CSV line into an Event
func (p *CSVStreamParser) parseEvent(fields []string, timestamp string, now time.Time) (*schema.Event, error) {
	if len(fields) < constants.EventColumns {
		return nil, fmt.Errorf("insufficient columns: %d < %d", len(fields), constants.EventColumns)
	}

	event := &schema.Event{
		GlobalEventID:        parseInt64(fields[0]),
		Day:                  parseInt(fields[1]),
		Actor1Name:           fields[6],
		Actor1CountryCode:    fields[7],
		Actor1Type1Code:      fields[12],
		Actor2Name:           fields[16],
		Actor2CountryCode:    fields[17],
		EventCode:            fields[26],
		EventBaseCode:        fields[27],
		EventRootCode:        fields[28],
		QuadClass:            parseInt(fields[29]),
		GoldsteinScale:       parseFloat(fields[30]),
		NumMentions:          parseInt(fields[31]),
		NumSources:           parseInt(fields[32]),
		NumArticles:          parseInt(fields[33]),
		AvgTone:              parseFloat(fields[34]),
		ActionGeoType:        parseInt(fields[51]),
		ActionGeoFullName:    fields[52],
		ActionGeoCountryCode: fields[53],
		ActionGeoADM1Code:    fields[54],
		ActionGeoLat:         parseFloat(fields[56]),
		ActionGeoLong:        parseFloat(fields[57]),
		SourceURL:            fields[60],
		Timestamp:            timestamp,
		ProcessedAt:          now,
	}

	return event, nil
}

// parseMention parses a mentions CSV line into a Mention
func (p *CSVStreamParser) parseMention(fields []string, timestamp string, now time.Time) (*schema.Mention, error) {
	if len(fields) < constants.MentionColumns {
		return nil, fmt.Errorf("insufficient columns: %d < %d", len(fields), constants.MentionColumns)
	}

	mention := &schema.Mention{
		GlobalEventID:         parseInt64(fields[0]),
		EventTimeDate:         parseInt64(fields[1]),
		MentionTimeDate:       parseInt64(fields[2]),
		MentionType:           parseInt(fields[3]),
		MentionSourceName:     fields[4],
		MentionIdentifier:     fields[5],
		SentenceID:            parseInt(fields[6]),
		Actor1CharOffset:      parseInt(fields[7]),
		Actor2CharOffset:      parseInt(fields[8]),
		ActionCharOffset:      parseInt(fields[9]),
		InRawText:             parseInt(fields[10]),
		Confidence:            parseInt(fields[11]),
		MentionDocLen:         parseInt(fields[12]),
		MentionDocTone:        parseFloat(fields[13]),
		MentionDocTranslation: fields[14],
		Timestamp:             timestamp,
		ProcessedAt:           now,
	}

	return mention, nil
}

// parseGKG parses a GKG CSV line into a GKG record
func (p *CSVStreamParser) parseGKG(fields []string, timestamp string, now time.Time) (*schema.GKG, error) {
	if len(fields) < constants.GKGColumns {
		return nil, fmt.Errorf("insufficient columns: %d < %d", len(fields), constants.GKGColumns)
	}

	gkg := &schema.GKG{
		GKGRECORDID:        fields[0],
		Date:               parseInt64(fields[1]),
		SourceCollectionID: parseInt(fields[2]),
		SourceCommonName:   fields[3],
		DocumentIdentifier: fields[4],
		Timestamp:          timestamp,
		ProcessedAt:        now,
	}

	// Parse optional fields
	if len(fields) > 5 {
		gkg.Counts = fields[5]
	}
	if len(fields) > 6 {
		gkg.V2Counts = fields[6]
	}
	if len(fields) > 7 {
		gkg.Themes = splitSemicolon(fields[7])
	}
	if len(fields) > 8 {
		gkg.V2Themes = splitSemicolon(fields[8])
	}
	if len(fields) > 9 {
		gkg.Locations = splitSemicolon(fields[9])
	}
	if len(fields) > 10 {
		gkg.V2Locations = splitSemicolon(fields[10])
	}
	if len(fields) > 11 {
		gkg.Persons = splitSemicolon(fields[11])
	}
	if len(fields) > 12 {
		gkg.V2Persons = splitSemicolon(fields[12])
	}
	if len(fields) > 13 {
		gkg.Organizations = splitSemicolon(fields[13])
	}
	if len(fields) > 14 {
		gkg.V2Organizations = splitSemicolon(fields[14])
	}
	if len(fields) > 15 {
		gkg.V2Tone = fields[15]
	}
	if len(fields) > 16 {
		gkg.Dates = fields[16]
	}
	if len(fields) > 17 {
		gkg.GCAM = fields[17]
	}
	if len(fields) > 18 {
		gkg.SharingImage = fields[18]
	}
	if len(fields) > 19 {
		gkg.RelatedImages = splitSemicolon(fields[19])
	}
	if len(fields) > 20 {
		gkg.SocialImageEmbeds = splitSemicolon(fields[20])
	}
	if len(fields) > 21 {
		gkg.SocialVideoEmbeds = splitSemicolon(fields[21])
	}
	if len(fields) > 22 {
		gkg.Quotations = splitSemicolon(fields[22])
	}
	if len(fields) > 23 {
		gkg.AllNames = splitSemicolon(fields[23])
	}
	if len(fields) > 24 {
		gkg.Amounts = splitSemicolon(fields[24])
	}
	if len(fields) > 25 {
		gkg.TranslationInfo = fields[25]
	}
	if len(fields) > 26 {
		gkg.Extras = fields[26]
	}

	return gkg, nil
}

// Helper functions

func parseInt(s string) int {
	s = strings.TrimSpace(s)
	if s == "" {
		return 0
	}
	v, _ := strconv.Atoi(s)
	return v
}

func parseInt64(s string) int64 {
	s = strings.TrimSpace(s)
	if s == "" {
		return 0
	}
	v, _ := strconv.ParseInt(s, 10, 64)
	return v
}

func parseFloat(s string) *float64 {
	s = strings.TrimSpace(s)
	if s == "" {
		return nil
	}
	v, err := strconv.ParseFloat(s, 64)
	if err != nil {
		return nil
	}
	return &v
}

func splitSemicolon(s string) []string {
	s = strings.TrimSpace(s)
	if s == "" {
		return nil
	}
	parts := strings.Split(s, ";")
	var result []string
	for _, p := range parts {
		p = strings.TrimSpace(p)
		if p != "" {
			result = append(result, p)
		}
	}
	return result
}