<commit_before>package elasticthought

import (
	"archive/tar"
	"bufio"
	"bytes"
	"compress/gzip"
	"io"
	"log"
	"net/http"
	"os"
	"path/filepath"
)

type tarFile struct {
	Name string
	Body string
}

// Opens tar.gz stream
func openTarGzStream(url string) (*tar.Reader, error) {

	resp, err := http.Get(url)
	if err != nil {
		return nil, err
	}
	gzipReader, err := gzip.NewReader(resp.Body)
	if err != nil {
		return nil, err
	}
	tarReader := tar.NewReader(gzipReader)

	return tarReader, nil

}

func untarWithToc(reader io.Reader, destDirectory string) ([]string, error) {

	toc := []string{}
	tr := tar.NewReader(reader)

	// Iterate through the files in the archive.
	for {
		hdr, err := tr.Next()
		if err == io.EOF {
			// end of tar archive
			break
		}
		if err != nil {
			return nil, err
		}

		if err := writeToDest(hdr, tr, destDirectory); err != nil {
			return nil, err
		}

		// add to toc
		toc = append(toc, hdr.Name)

	}

	return toc, nil

}

// Given a reader, wrap in a tar.gz reader and write all entries
// to destDirectory.  Also return a table of contents.
func untarGzWithToc(reader io.Reader, destDirectory string) ([]string, error) {

	gzipReader, err := gzip.NewReader(reader)
	if err != nil {
		return nil, err
	}
	return untarWithToc(gzipReader, destDirectory)
}

func writeToDest(hdr *tar.Header, tr *tar.Reader, destDirectory string) error {

	// write stream to file in work directory
	destPath := filepath.Join(destDirectory, hdr.Name)

	// does dir exist? if not, make it
	mkdir(filepath.Dir(destPath))

	f, err := os.Create(destPath)
	if err != nil {
		return err
	}
	w := bufio.NewWriter(f)
	defer w.Flush()
	_, err = io.Copy(w, tr)
	if err != nil {
		return err
	}
	return nil

}

func createArchive(buf *bytes.Buffer, tarFiles []tarFile) {

	// Create a new tar archive.
	tw := tar.NewWriter(buf)

	for _, file := range tarFiles {
		hdr := &tar.Header{
			Name: file.Name,
			Size: int64(len(file.Body)),
			Uid:  100,
			Gid:  101,
		}
		if err := tw.WriteHeader(hdr); err != nil {
			log.Fatalln(err)
		}
		if _, err := tw.Write([]byte(file.Body)); err != nil {
			log.Fatalln(err)
		}
	}
	// Make sure to check the error on Close.
	if err := tw.Close(); err != nil {
		log.Fatalln(err)
	}
<commit_msg>untar works now<commit_after>package elasticthought

import (
	"archive/tar"
	"bufio"
	"bytes"
	"compress/gzip"
	"io"
	"log"
	"net/http"
	"os"
	"path/filepath"

	"github.com/couchbaselabs/logg"
)

type tarFile struct {
	Name string
	Body string
}

// Opens tar.gz stream
func openTarGzStream(url string) (*tar.Reader, error) {

	resp, err := http.Get(url)
	if err != nil {
		return nil, err
	}
	gzipReader, err := gzip.NewReader(resp.Body)
	if err != nil {
		return nil, err
	}
	tarReader := tar.NewReader(gzipReader)

	return tarReader, nil

}

func untarWithToc(reader io.Reader, destDirectory string) ([]string, error) {

	toc := []string{}
	tr := tar.NewReader(reader)

	// Iterate through the files in the archive.
	for {
		hdr, err := tr.Next()
		if err == io.EOF {
			// end of tar archive
			break
		}
		if err != nil {
			return nil, err
		}

		if err := writeToDest(hdr, tr, destDirectory); err != nil {
			return nil, err
		}

		// add to toc
		toc = append(toc, hdr.Name)

	}

	return toc, nil

}

// Given a reader, wrap in a tar.gz reader and write all entries
// to destDirectory.  Also return a table of contents.
func untarGzWithToc(reader io.Reader, destDirectory string) ([]string, error) {

	gzipReader, err := gzip.NewReader(reader)
	if err != nil {
		return nil, err
	}
	return untarWithToc(gzipReader, destDirectory)
}

func writeToDest(hdr *tar.Header, tr *tar.Reader, destDirectory string) error {

	// write stream to file in work directory
	destPath := filepath.Join(destDirectory, hdr.Name)

	logg.LogTo("TRAINING_JOB", "writeToDest called with: %v hdr: %+v", destPath, hdr)

	switch hdr.Typeflag {
	case tar.TypeDir:
		logg.LogTo("TRAINING_JOB", "%v is a directory", destPath)
		// does dir exist? if not, make it
		logg.LogTo("TRAINING_JOB", "calling mkdir on %v", destPath)
		if err := mkdir(destPath); err != nil {
			logg.LogTo("TRAINING_JOB", "mkdir failed on %v", destPath)
			return err
		}

	default:

		logg.LogTo("TRAINING_JOB", "calling os.Create on %v", destPath)
		f, err := os.Create(destPath)
		if err != nil {
			logg.LogTo("TRAINING_JOB", "calling os.Create failed on %v", destPath)
			return err
		}
		w := bufio.NewWriter(f)
		defer w.Flush()
		_, err = io.Copy(w, tr)
		if err != nil {
			logg.LogTo("TRAINING_JOB", "io.Copy failed: %v", err)
			return err
		}

	}
	return nil

}

func createArchive(buf *bytes.Buffer, tarFiles []tarFile) {

	// Create a new tar archive.
	tw := tar.NewWriter(buf)

	for _, file := range tarFiles {
		hdr := &tar.Header{
			Name: file.Name,
			Size: int64(len(file.Body)),
			Uid:  100,
			Gid:  101,
		}
		if err := tw.WriteHeader(hdr); err != nil {
			log.Fatalln(err)
		}
		if _, err := tw.Write([]byte(file.Body)); err != nil {
			log.Fatalln(err)
		}
	}
	// Make sure to check the error on Close.
	if err := tw.Close(); err != nil {
		log.Fatalln(err)
	}

{'fraction_non_alphanumeric': 0.1415650806134487, 'fraction_numerical': 0.0031458906802988595, 'mean_word_length': 3.2391666666666667, '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': '4170759', 'n_tokens_mistral': 2096, 'n_tokens_neox': 1933, 'n_words': 547}