<commit_before>package synco

import "code.google.com/p/go-imap/go1/imap"
import (
	"bytes"
	"fmt"
	"encoding/json"
	"log"
	"net/mail"
	"os"
	"time"
)

type IMAPServer struct {
	Host string
	Port uint16
}

type IMAPAccount struct {
	Username string
	Password string
	Server *IMAPServer
}

type UIDFetchJob struct {
	LoUID, HiUID uint32
}

var	mbox string = "[Gmail]/All Mail"

func PrintMail(acct *IMAPAccount, query string) {
	imap.DefaultLogger = log.New(os.Stdout, "", 0)
//	imap.DefaultLogMask = imap.LogConn | imap.LogRaw

	log.Printf("Running for user '%s' on IMAP server '%s:%d'", acct.Username, acct.Server.Host, acct.Server.Port)

	conns := 1
	partsize := 250

	// Fetch UIDs.
	c := Dial(acct.Server)
	login(c, acct.Username, acct.Password)
	c.Select(mbox, true)
	uids, _ := SearchUIDs(c, query)
	c.Close(false)
	c.Logout(-1)

	timestarted := time.Now()

	// Fetch message MIME bodies.
	nparts := (len(uids) + partsize - 1) / partsize
	jobs := make([]UIDFetchJob, nparts)
	for i := 0; i < nparts; i++ {
		lo := i * partsize
		hi_exclusive := (i + 1) * partsize
		if hi_exclusive >= len(uids) {
			hi_exclusive = len(uids) - 1
			for uids[hi_exclusive] == 0 { // hacky
				hi_exclusive--
			}
		}
		loUID := uids[lo]
		hiUID := uids[hi_exclusive]
		job := UIDFetchJob{ loUID, hiUID }
		jobs[i] = job
	}

	log.Printf("%d UIDs total, %d jobs of size <= %d\n", len(uids), len(jobs), partsize)

	// open conns
	if conns > len(jobs) {
		conns = len(jobs)
	}
	jobchan := make(chan UIDFetchJob)
	for i := 0; i < conns; i++ {
		go StartWorker(i, acct, jobchan)
	}

	nextjobindex := 0
	quits := 0
	for {
		if quits == conns {
			// have sent quits to everyone
			break
		}

		wantjob := <-jobchan
		if wantjob.LoUID == 0 {
			if nextjobindex < len(jobs) {
				jobchan <- jobs[nextjobindex]
				nextjobindex++
			} else {
				jobchan <- UIDFetchJob{ 0, 0 }
				quits++
			}
		} else {
			panic("unknown wantjob value")
		}
	}

	timeelapsed := time.Since(timestarted)
	msecpermessage := timeelapsed.Seconds() / float64(len(uids)) * 1000
	messagespersec := float64(len(uids)) / timeelapsed.Seconds()
	log.Printf("Finished fetching %d messages in %.2fs (%.1fms per message; %.1f messages per second)\n", len(uids), timeelapsed.Seconds(), msecpermessage, messagespersec)

}

func SearchUIDs(c *imap.Client, query string) (uids []uint32, err error) {
	cmd, err := c.UIDSearch("X-GM-RAW", fmt.Sprint("\"", query, "\""))
	
	for cmd.InProgress() {
		c.Recv(-1)
		for _, rsp := range cmd.Data {
			uids = rsp.SearchResults()
		}
		cmd.Data = nil
	}
	return
}

func FetchAllUIDs(c *imap.Client) (uids []uint32, err error) {
	maxmessages := 150000
	uids = make([]uint32, maxmessages)

	set, _ := imap.NewSeqSet("1:*")
	cmd, err := c.UIDFetch(set, "RFC822.SIZE")
	
	messagenum := uint32(0)
	for cmd.InProgress() {
		c.Recv(-1)
		for _, rsp := range cmd.Data {
			uid := imap.AsNumber(rsp.MessageInfo().Attrs["UID"])
			uids[messagenum] = uid
		}
		cmd.Data = nil
		messagenum++
	}

	uids = uids[:messagenum]
	return
}

func StartWorker(workerid int, acct *IMAPAccount, jobchan chan UIDFetchJob) {
	c := Dial(acct.Server)
	login(c, acct.Username, acct.Password)
	c.Select(mbox, true)

	for {
		jobchan <- UIDFetchJob{ 0, 0 }
		job := <-jobchan
		if job.LoUID != 0 {
			FetchMessages(c, job.LoUID, job.HiUID)
			log.Printf("Worker %d finished one job\n", workerid)
		} else {
			break
		}
	}
}

func FetchMessages(c *imap.Client, loUID, hiUID uint32) (err error) {
	set, _ := imap.NewSeqSet(fmt.Sprintf("%d:%d", loUID, hiUID))
	cmd, err := c.UIDFetch(set, "RFC822")

	for cmd.InProgress() {
		c.Recv(-1)
		for _, rsp := range cmd.Data {
			uid := imap.AsNumber(rsp.MessageInfo().Attrs["UID"])
			mime := imap.AsBytes(rsp.MessageInfo().Attrs["RFC822"])
			if msg, _ := mail.ReadMessage(bytes.NewReader(mime)); msg != nil {
				PrintMessageAsJSON(msg, uid)<commit_msg>removed parallelism<commit_after>package synco

import "code.google.com/p/go-imap/go1/imap"
import (
	"bytes"
	"fmt"
	"encoding/json"
	"log"
	"net/mail"
	"os"
	"time"
)

type IMAPServer struct {
	Host string
	Port uint16
}

type IMAPAccount struct {
	Username string
	Password string
	Server *IMAPServer
}

type UIDFetchJob struct {
	uids []uint32
}

var	mbox string = "[Gmail]/All Mail"

func PrintMail(acct *IMAPAccount, query string) {
	imap.DefaultLogger = log.New(os.Stdout, "", 0)
//	imap.DefaultLogMask = imap.LogConn | imap.LogRaw

	log.Printf("Running for user '%s' on IMAP server '%s:%d'", acct.Username, acct.Server.Host, acct.Server.Port)

	jobsize := 250

	// Fetch UIDs.
	c := Dial(acct.Server)
	login(c, acct.Username, acct.Password)
	c.Select(mbox, true)
	uids, _ := SearchUIDs(c, query)

	timestarted := time.Now()

	nparts := (len(uids) + jobsize - 1) / jobsize
	jobs := make([]*imap.SeqSet, nparts)
	for i := 0; i < nparts; i++ {
		lo := i * jobsize
		hi_exclusive := (i + 1) * jobsize
		if hi_exclusive >= len(uids) {
			hi_exclusive = len(uids) - 1
			for uids[hi_exclusive] == 0 { // hacky
				hi_exclusive--
			}
		}
		set, _ := imap.NewSeqSet("")
		set.AddNum(uids[lo:hi_exclusive]...)
		jobs[i] = set
	}

	log.Printf("%d UIDs total, %d jobs of size <= %d\n", len(uids), len(jobs), jobsize)

	for _, jobUIDs := range jobs {
		FetchMessages(c, jobUIDs)
	}

	timeelapsed := time.Since(timestarted)
	msecpermessage := timeelapsed.Seconds() / float64(len(uids)) * 1000
	messagespersec := float64(len(uids)) / timeelapsed.Seconds()
	log.Printf("Finished fetching %d messages in %.2fs (%.1fms per message; %.1f messages per second)\n", len(uids), timeelapsed.Seconds(), msecpermessage, messagespersec)

	c.Close(false)
	c.Logout(-1)
}

func SearchUIDs(c *imap.Client, query string) (uids []uint32, err error) {
	cmd, err := c.UIDSearch("X-GM-RAW", fmt.Sprint("\"", query, "\""))
	
	for cmd.InProgress() {
		c.Recv(-1)
		for _, rsp := range cmd.Data {
			uids = rsp.SearchResults()
		}
		cmd.Data = nil
	}
	return
}

func FetchAllUIDs(c *imap.Client) (uids []uint32, err error) {
	maxmessages := 150000
	uids = make([]uint32, maxmessages)

	set, _ := imap.NewSeqSet("1:*")
	cmd, err := c.UIDFetch(set, "RFC822.SIZE")
	
	messagenum := uint32(0)
	for cmd.InProgress() {
		c.Recv(-1)
		for _, rsp := range cmd.Data {
			uid := imap.AsNumber(rsp.MessageInfo().Attrs["UID"])
			uids[messagenum] = uid
		}
		cmd.Data = nil
		messagenum++
	}

	uids = uids[:messagenum]
	return
}


func FetchMessages(c *imap.Client, uidSet *imap.SeqSet) (err error) {
	cmd, err := c.UIDFetch(uidSet, "RFC822")

	for cmd.InProgress() {
		c.Recv(-1)
		for _, rsp := range cmd.Data {
			uid := imap.AsNumber(rsp.MessageInfo().Attrs["UID"])
			mime := imap.AsBytes(rsp.MessageInfo().Attrs["RFC822"])
			if msg, _ := mail.ReadMessage(bytes.NewReader(mime)); msg != nil {
				PrintMessageAsJSON(msg, uid)
{'fraction_non_alphanumeric': 0.1662958018098205, 'fraction_numerical': 0.017801513128615932, 'mean_word_length': 3.6304945054945055, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 6, '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': '7339883', 'n_tokens_mistral': 2967, 'n_tokens_neox': 2719, 'n_words': 602}