<commit_before>package recording

import (
	"bufio"
	"context"
	"errors"
	"log"
	"net/http"
	"os"
	"time"
)

var (
	ErrMismatchWrite = errors.New("recording: did not write the same number of bytes that were read")
)

// Recording ...
type Recording struct {
	ctx     context.Context
	url     string
	fname   string
	fout    *os.File
	cancel  context.CancelFunc
	started time.Time

	Debug bool
	Err   error
}

// New creates a new Recording of the given URL to the given filename for output.
func New(url, fname string) (*Recording, error) {
	fout, err := os.Create(fname)
	if err != nil {
		return nil, err
	}

	ctx, cancel := context.WithTimeout(context.Background(), 4*time.Hour)

	r := &Recording{
		ctx:     ctx,
		url:     url,
		fname:   fname,
		fout:    fout,
		cancel:  cancel,
		started: time.Now(),
	}

	return r, nil
}

func (r *Recording) Cancel() {
	r.cancel()
}

func (r *Recording) Done() <-chan struct{} {
	return r.ctx.Done()
}

// OutputFilename gets the output filename originally passed into New.
func (r *Recording) OutputFilename() string {
	return r.fname
}

// StartTime gets start time
func (r *Recording) StartTime() time.Time {
	return r.started
}

// Start blockingly starts the recording and returns the error if one is encountered while streaming.
// This should be stopped in another goroutine.
func (r *Recording) Start() error {
	resp, err := http.Get(r.url)
	if err != nil {
		return err
	}

	defer resp.Body.Close()
	defer r.fout.Close()
	defer r.cancel()

	reader := bufio.NewReader(resp.Body)

	c := time.NewTicker(5 * time.Second)
	defer c.Stop()

	buf := make([]byte, 65536)

	for {
		time.Sleep(250 * time.Millisecond)

		select {
		case <-r.ctx.Done():
			r.fout.Sync()
			return nil
		case <-c.C:
			if r.Debug {
				log.Println("Syncing file")
			}
			err := r.fout.Sync()
			if err != nil {
				r.Err = err
				return err
			}
		default:
			nr, err := reader.Read(buf)
			if err != nil {
				r.Err = err
				return err
			}

			if r.Debug {
				log.Printf("%d bytes read", nr)
			}

			buf = buf[:nr]

			_, err = r.fout.Write(buf)
			if err != nil {
				r.Err = err
				return err
			}
		}
	}
}<commit_msg>recording: append 1's to every additional download due to connections cutting out<commit_after>package recording

import (
	"bufio"
	"context"
	"errors"
	"io"
	"log"
	"net/http"
	"os"
	"time"
)

var (
	ErrMismatchWrite = errors.New("recording: did not write the same number of bytes that were read")
)

// Recording ...
type Recording struct {
	ctx      context.Context
	url      string
	fname    string
	fout     *os.File
	cancel   context.CancelFunc
	started  time.Time
	restarts int

	Debug bool
	Err   error
}

// New creates a new Recording of the given URL to the given filename for output.
func New(url, fname string) (*Recording, error) {
	fout, err := os.Create(fname)
	if err != nil {
		return nil, err
	}

	ctx, cancel := context.WithTimeout(context.Background(), 4*time.Hour)

	r := &Recording{
		ctx:     ctx,
		url:     url,
		fname:   fname,
		fout:    fout,
		cancel:  cancel,
		started: time.Now(),
	}

	return r, nil
}

func (r *Recording) Cancel() {
	r.cancel()
}

func (r *Recording) Done() <-chan struct{} {
	return r.ctx.Done()
}

// OutputFilename gets the output filename originally passed into New.
func (r *Recording) OutputFilename() string {
	return r.fname
}

// StartTime gets start time
func (r *Recording) StartTime() time.Time {
	return r.started
}

// Start blockingly starts the recording and returns the error if one is encountered while streaming.
// This should be stopped in another goroutine.
func (r *Recording) Start() error {
	resp, err := http.Get(r.url)
	if err != nil {
		return err
	}

	defer resp.Body.Close()
	defer r.fout.Close()
	defer r.cancel()

	reader := bufio.NewReader(resp.Body)

	c := time.NewTicker(5 * time.Second)
	defer c.Stop()

	buf := make([]byte, 65536)

	for {
		time.Sleep(65 * time.Millisecond)

		select {
		case <-r.ctx.Done():
			r.fout.Sync()
			return nil
		case <-c.C:
			if r.Debug {
				log.Println("Syncing file")
			}
			err := r.fout.Sync()
			if err != nil {
				r.Err = err
				return err
			}
		default:
			nr, err := reader.Read(buf)
			if err != nil {
				if err == io.EOF {
					log.Println("restarting download")
					r.fout, err = os.Create(r.fname + "1")
					if err != nil {
						return err
					}

					resp, err := http.Get(r.url)
					if err != nil {
						return err
					}

					reader = bufio.NewReader(resp.Body)
					defer resp.Body.Close()
					defer r.fout.Close()
					r.restarts++
					continue
				}
				r.Err = err
				return err
			}

			if r.Debug {
				log.Printf("%d bytes read", nr)
			}

			buf = buf[:nr]

			_, err = r.fout.Write(buf)
			if err != nil {
				r.Err = err
				return err
			}
		}
	}
}
{'fraction_non_alphanumeric': 0.13119227539882453, 'fraction_numerical': 0.004408060453400504, 'mean_word_length': 2.6429663608562692, 'pattern_counts': {'":': 0, '<': 9, '<?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': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5073223', 'n_tokens_mistral': 1985, 'n_tokens_neox': 1720, 'n_words': 520}