<commit_before>package main

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"path/filepath"
	"strconv"
	"strings"
	"github.com/nutrun/lentil"
)

type Client struct {
	q       *lentil.Beanstalkd
	verbose bool
}

func NewClient(verbose bool) (*Client, error) {
	this := new(Client)
	q, err := lentil.Dial(Config.QueueAddr)
	if err != nil {
		return nil, err
	}
	this.q = q
	this.verbose = verbose
	return this, nil
}

func isValidMessage(msg map[string]interface{}) error {
    _, found_executable := msg["executable"]
    _, found_cmd := msg["cmd"]
    if found_executable && found_cmd {
        return errors.New("Found both executable and cmd in message. Don't know which one to use")
    }
    if !found_executable && !found_cmd {
        return errors.New("Neither executable nor cmd field provided in message")
    }
    return nil
}

func (this *Client) put(cmd, executable string, arguments []string, mailto, workdir, out, tube string, priority, delay int) error {
    msg := make(map[string]interface{})
    msg["cmd"] = cmd

	msg["executable"] = executable

    argument_json, e := json.Marshal(arguments)
    if e != nil {
        panic(e)
    }
	msg["arguments"] = string(argument_json)

    msg["tube"] = tube
	msg["pri"] = fmt.Sprintf("%d", priority) // Not used except for debugging
	if tube == "" {
		return errors.New("Missing required param -tube")
	}
	msg["delay"] = fmt.Sprintf("%d", delay) // Not used except for debugging
	workdir, e = filepath.Abs(workdir)
	if e != nil {
		return e
	}
	msg["workdir"] = workdir
	msg["out"] = out

    if e = isValidMessage(msg); e != nil {
        return e
    }

	message, e := json.Marshal(msg)
	if this.verbose {
		log.Printf("QUEUEING UP: %s\n", message)
	}
	if e != nil {
		return e
	}
	if tube != "default" {
		e = this.q.Use(tube)
		if e != nil {
			return e
		}
	}
	_, e = this.q.Put(priority, delay, 60*60, message) // An hour TTR?
	return e
}

func (this *Client) putMany(input []byte) error {
	jobs := make([]map[string]interface{}, 0)
	e := json.Unmarshal(input, &jobs)
	if e != nil {
		return e
	}
	for _, job := range jobs {
        if e := isValidMessage(job); e != nil {
            panic(e)
        }
		priority := 0
		if priorityStr, exists := job["pri"]; exists {
			priority, e = strconv.Atoi(priorityStr.(string))
			if e != nil {
				return e
			}
		}
		delay := 0
		if delayStr, exists := job["delay"]; exists {
			delay, e = strconv.Atoi(delayStr.(string))
			if e != nil {
				return e
			}
		}
		out, exists := job["out"]
		if !exists {
			out = "/dev/null"
		}
		workdir := "/tmp"
		dir, exists := job["workdir"]
		if exists {
			workdir = dir.(string)
		}

		e = this.put(job["cmd"].(string), job["executable"].(string), job["arguments"].([]string),
                     job["mailto"].(string), workdir, out.(string), job["tube"].(string),
                     priority, delay)
		if e != nil {
			return e
		}
	}
	return nil
}

func (this *Client) stats() error {
	q := NewJobQueue(this.q, false, make([]string, 0))
	stats, err := json.Marshal(q)
	if err != nil {
		return err
	}
	buffer := bytes.NewBufferString("")
	err = json.Indent(buffer, stats, "", "\t")
	if err != nil {
		return err
	}
	fmt.Printf("%s\n", buffer.String())
	return nil
}<commit_msg>created message struct. got it to compile. WIP<commit_after>package main

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"path/filepath"
	"strings"
	"github.com/nutrun/lentil"
)

type Client struct {
	q       *lentil.Beanstalkd
	verbose bool
}

func NewClient(verbose bool) (*Client, error) {
	this := new(Client)
	q, err := lentil.Dial(Config.QueueAddr)
	if err != nil {
		return nil, err
	}
	this.q = q
	this.verbose = verbose
	return this, nil
}

type Message struct {
    Command     string      `json:command`
    Executable  string      `json:executable`
    Arguments   []string    `json:arguments`
    Mailto      string      `json:mailto`
    Workdir     string      `json:workdir`
    Out         string      `json:out`
    Tube        string      `json:tube`
    Priority    int         `json:priority`
    Delay       int         `json:delay`
}

func isValidMessage(msg *Message) error {
    if msg.Command != "" && msg.Executable != "" {
        return errors.New("Found both executable and cmd in message. Don't know which one to use")
    }
    if msg.Command == "" && msg.Executable == "" {
        return errors.New("Neither executable nor cmd field provided in message")
    }
    if msg.Tube == "" {
		return errors.New("Missing required param -tube")
    }
    return nil
}

func (this *Client) put_message(msg *Message) error {
    if e := isValidMessage(msg); e != nil {
        return e
    }

	message, e := json.Marshal(msg)
	if this.verbose {
		log.Printf("QUEUEING UP: %s\n", message)
	}
	if e != nil {
		return e
	}
	if msg.Tube != "default" {
		e = this.q.Use(msg.Tube)
		if e != nil {
			return e
		}
	}
	_, e = this.q.Put(msg.Priority, msg.Delay, 60*60, message) // An hour TTR?
	return e
}

func (this *Client) put(cmd, executable string, arguments []string, mailto string, workdir, out, tube string, priority, delay int) error {
    workdir, e := filepath.Abs(workdir)
	if e != nil {
		return e
	}
    msg := &Message{cmd, executable, arguments, mailto, workdir, out, tube, priority, delay}
    return this.put_message(msg)
}

func (this *Client) putMany(input []byte) error {
	jobs := make([]*Message, 0)
	e := json.Unmarshal(input, &jobs)
	if e != nil {
		return e
	}
	for _, job := range jobs {
        if e := isValidMessage(job); e != nil {
            panic(e)
        }
        if job.Out == "" {
            job.Out = "/dev/null"
        }
		if job.Workdir == "" {
            job.Workdir = "/tmp"
        }

		e = this.put_message(job)
		if e != nil {
			return e
		}
	}
	return nil
}

func (this *Client) stats() error {
	q := NewJobQueue(this.q, false, make([]string, 0))
	stats, err := json.Marshal(q)
	if err != nil {
		return err
	}
	buffer := bytes.NewBufferString("")
	err = json.Indent(buffer, stats, "", "\t")
	if err != nil {
		return err
	}
	fmt.Printf("%s\n", buffer.String())
	return nil
}
{'fraction_non_alphanumeric': 0.15256179407431658, 'fraction_numerical': 0.0022917007693566867, 'mean_word_length': 2.727882855399634, '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': '2031204', 'n_tokens_mistral': 2256, 'n_tokens_neox': 2126, 'n_words': 596}