<commit_before>package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"os"
	"regexp"
	"strings"
	"time"

	"github.com/realzeitmedia/bubbles"
)

const (
	glog = `^\S+ \S+ \S+\s+\d+ (?P<file>.+):\d+\] (?P<msg>(?P<msgcore>.*?:)?.*)`
)

var (
	indexTmpl  = flag.String("index", "logstash-20060102", "index name")
	hosts      = flag.String("hosts", "localhost:9200", "ES hosts (,)")
	lineRegexp = flag.String("regexp", glog, "regexp with capture groups")
	verbose    = flag.Bool("verbose", false, "verbose")
)

func main() {
	flag.Parse()

	lineReg, err := regexp.Compile(*lineRegexp)
	if err != nil {
		fmt.Fprintf(os.Stderr, "invalid -regexp: %s\n", err)
		os.Exit(1)
	}
	if *verbose {
		fmt.Printf("using regexp: %q\n", *lineRegexp)
	}

	es := bubbles.New(strings.Split(*hosts, ","),
		bubbles.OptFlush(1*time.Second),
	)

	defer func() {
		if *verbose {
			fmt.Printf("shutting down...\n")
		}
		time.Sleep(2 * time.Second) // more than the ES flush time
	}()

	var (
		r       = bufio.NewScanner(bufio.NewReader(os.Stdin))
		subexps = lineReg.SubexpNames()
	)
	for r.Scan() {
		l := r.Text()
		if *verbose {
			fmt.Printf("line: %q\n", l)
		}

		t := time.Now()
		fields := map[string]string{
			"@timestamp": t.Format(time.RFC3339Nano),
			"message":    l,
		}
		if m := lineReg.FindStringSubmatch(l); m != nil {
			for i, v := range m {
				if i == 0 {
					continue
				}
				fields[subexps[i]] = v<commit_msg>add -copy flag to make piping easier<commit_after>package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"os"
	"regexp"
	"strings"
	"time"

	"github.com/realzeitmedia/bubbles"
)

const (
	glog = `^\S+ \S+ \S+\s+\d+ (?P<file>.+):\d+\] (?P<msg>(?P<msgcore>.*?:)?.*)`
)

var (
	indexTmpl  = flag.String("index", "logstash-20060102", "index name")
	hosts      = flag.String("hosts", "localhost:9200", "ES hosts (,)")
	lineRegexp = flag.String("regexp", glog, "regexp with capture groups")
	verbose    = flag.Bool("verbose", false, "verbose")
	copy       = flag.Bool("copy", false, "copy all lines to STDOUT")
)

func main() {
	flag.Parse()

	if *copy && *verbose {
		fmt.Fprintf(os.Stderr, "can't enable both -verbose and -copy\n")
		os.Exit(1)
	}

	lineReg, err := regexp.Compile(*lineRegexp)
	if err != nil {
		fmt.Fprintf(os.Stderr, "invalid -regexp: %s\n", err)
		os.Exit(2)
	}
	if *verbose {
		fmt.Printf("using regexp: %q\n", *lineRegexp)
	}

	es := bubbles.New(strings.Split(*hosts, ","),
		bubbles.OptFlush(1*time.Second),
	)

	defer func() {
		if *verbose {
			fmt.Printf("shutting down...\n")
		}
		time.Sleep(2 * time.Second) // more than the ES flush time
	}()

	var (
		r       = bufio.NewScanner(bufio.NewReader(os.Stdin))
		subexps = lineReg.SubexpNames()
	)
	for r.Scan() {
		l := r.Text()
		if *verbose {
			fmt.Printf("line: %q\n", l)
		}
		if *copy {
			fmt.Println(l)
		}

		t := time.Now()
		fields := map[string]string{
			"@timestamp": t.Format(time.RFC3339Nano),
			"message":    l,
		}
		if m := lineReg.FindStringSubmatch(l); m != nil {
			for i, v := range m {
				if i == 0 {
					continue
				}
				fields[subexps[i]] = v
{'fraction_non_alphanumeric': 0.20298798311139982, 'fraction_numerical': 0.013316011692107827, 'mean_word_length': 3.1791044776119404, 'pattern_counts': {'":': 4, '<': 9, '<?xml version=': 0, '>': 9, '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': '384339', 'n_tokens_mistral': 1420, 'n_tokens_neox': 1287, 'n_words': 261}