<commit_before>// Package logs provides structs for working with AWS CloudWatch Logs records.
package logs

import (
	"bytes"
	"compress/gzip"
	"encoding/json"

	"github.com/apex/go-apex"
	"github.com/apex/go-apex/kinesis"
)

// LogEvent represents a single log event.
type LogEvent struct {
	ID        string `json:"id"`
	Timestamp int64  `json:"timestamp"`
	Message   string `json:"message"`
}

// Event represents a Kinesis event with one or more records.
type Event struct {
	Records []*Record `json:"Records"`
}

// Record represents a single Kinesis record.
type Record struct {
	kinesis.Record
	Logs struct {
		Owner               string      `json:"owner"`
		LogGroup            string      `json:"logGroup"`
		LogStream           string      `json:"logStream"`
		SubscriptionFilters []string    `json:"subscriptionFilters"`
		MessageType         string      `json:"messageType"`
		LogEvents           []*LogEvent `json:"logEvents"`
	}
}

// Handler handles Logs events.
type Handler interface {
	HandleLogs(*Event, *apex.Context) error
}

// HandlerFunc unmarshals Logs events before passing control.
type HandlerFunc func(*Event, *apex.Context) error

// Handle implements apex.Handler.
func (h HandlerFunc) Handle(data json.RawMessage, ctx *apex.Context) (interface{}, error) {
	var event Event

	if err := json.Unmarshal(data, &event); err != nil {
		return nil, err
	}

	for _, record := range event.Records {
		r, err := gzip.NewReader(bytes.NewReader(record.Data()))
		if err != nil {
			return nil, err
		}

		if err = json.NewDecoder(r).Decode(&record.Logs); err != nil {
			return nil, err
		}

		if err := r.Close(); err != nil {
			return nil, err
		}
	}

	if err := h(&event, ctx); err != nil {
		return nil, err
	}

	return event, nil
}

// HandleFunc handles Logs events with callback function.
func HandleFunc(h HandlerFunc) {
	apex.Handle(h)
}

// Handle Logs events with handler.
func Handle(h Handler) {
	HandleFunc(HandlerFunc(h.HandleLogs))
}<commit_msg>refactor logs with decode helper<commit_after>// Package logs provides structs for working with AWS CloudWatch Logs records.
package logs

import (
	"bytes"
	"compress/gzip"
	"encoding/json"

	"github.com/apex/go-apex"
	"github.com/apex/go-apex/kinesis"
)

// LogEvent represents a single log event.
type LogEvent struct {
	ID        string `json:"id"`
	Timestamp int64  `json:"timestamp"`
	Message   string `json:"message"`
}

// Event represents a Kinesis event with one or more records.
type Event struct {
	Records []*Record `json:"Records"`
}

// Record represents a single Kinesis record.
type Record struct {
	kinesis.Record
	Logs struct {
		Owner               string      `json:"owner"`
		LogGroup            string      `json:"logGroup"`
		LogStream           string      `json:"logStream"`
		SubscriptionFilters []string    `json:"subscriptionFilters"`
		MessageType         string      `json:"messageType"`
		LogEvents           []*LogEvent `json:"logEvents"`
	}
}

// Handler handles Logs events.
type Handler interface {
	HandleLogs(*Event, *apex.Context) error
}

// HandlerFunc unmarshals Logs events before passing control.
type HandlerFunc func(*Event, *apex.Context) error

// Handle implements apex.Handler.
func (h HandlerFunc) Handle(data json.RawMessage, ctx *apex.Context) (interface{}, error) {
	event := new(Event)

	if err := json.Unmarshal(data, event); err != nil {
		return nil, err
	}

	if err := decode(event); err != nil {
		return nil, err
	}

	if err := h(event, ctx); err != nil {
		return nil, err
	}

	return event, nil
}

// HandleFunc handles Logs events with callback function.
func HandleFunc(h HandlerFunc) {
	apex.Handle(h)
}

// Handle Logs events with handler.
func Handle(h Handler) {
	HandleFunc(HandlerFunc(h.HandleLogs))
}

// decode decodes the log payload which is gzipped.
func decode(event *Event) error {
	for _, record := range event.Records {
		r, err := gzip.NewReader(bytes.NewReader(record.Data()))
		if err != nil {
			return err
		}

		if err = json.NewDecoder(r).Decode(&record.Logs); err != nil {
			return err
		}

		if err := r.Close(); err != nil {
			return err
		}
	}

	return nil
}
{'fraction_non_alphanumeric': 0.12070633768746976, 'fraction_numerical': 0.0009675858732462506, 'mean_word_length': 3.3117831074035453, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3653163', 'n_tokens_mistral': 1375, 'n_tokens_neox': 1299, 'n_words': 404}