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