<commit_before>package main

import (
	"bufio"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"sync"

	"code.google.com/p/go.text/encoding/japanese"
	"code.google.com/p/go.text/transform"
)

// FIXME: brash up regexp
var ImgUrlRegexp *regexp.Regexp = regexp.MustCompile("h?ttp://[0-9a-zA-Z/\\-.%]+?\\.(jpg|jpeg|gif|png)")

var Fetched = struct {
	sync.RWMutex
	m map[string]bool
}{m: make(map[string]bool)}

func Img(url string) {
	Fetched.Lock()
	if Fetched.m[url] {
		Fetched.Unlock()
		return
	}
	Fetched.m[url] = true
	Fetched.Unlock()
	log.Printf("downloading %v...\n", url)

	out, err := os.Create(filepath.Base(url))
	if err != nil {
		log.Printf("failed to create download file: %v\n", err)
		return
	}
	defer out.Close()

	resp, err := http.Get(url)
	if err != nil {
		log.Printf("failed to download img: %v, err: \n", url, err)
		return
	}
	defer resp.Body.Close()

	n, err := io.Copy(out, resp.Body)
	if err != nil {
		log.Printf("failed to copy img: %v\n", err)
		return
	}
	log.Printf("saved %v (%v bytes)\n", url, n)
}

func ImgQueue(ch chan string, done chan bool) {
	for {
		url, more := <-ch
		if more && url != "" {
			Img(url)
		} else {
			done <- true
		}
	}
}

func Dat(url string) {
	Fetched.Lock()
	if Fetched.m[url] {
		Fetched.Unlock()
		return
	}
	Fetched.m[url] = true
	Fetched.Unlock()
	log.Printf("reading %v...\n", url)

	resp, err := http.Get(url)
	if err != nil {
		log.Printf("failed to fetch: %v, err: %v\n", url, err)
		return
	}
	defer resp.Body.Close()

	imgCh := make(chan string)
	done := make(chan bool)
	go ImgQueue(imgCh, done)

	reader := transform.NewReader(resp.Body, japanese.ShiftJIS.NewDecoder())
	bufr := bufio.NewReader(reader)
	for {
		lineb, err := bufr.ReadString('\n')
		if err == io.EOF {
			return
		}
		if err != nil {
			log.Printf("failed to read content. err: %v\n", err)
			return
		}
		line := strings.Split(string(lineb), "<>")
		if len(line) >= 4 {
			body := line[3]
			matched := ImgUrlRegexp.FindAllString(body, -1)
			for i := 0; i < len(matched); i++ {
				imgCh <- matched[i]
			}
		}
	}
	close(imgCh)

	<-done
}

func DatQueue(ch chan string, done chan bool) {
	for {
		url, more := <-ch
		if more && url != "" {
			Dat(url)
		} else {
			done <- true
		}
	}
}

func main() {
	datCh := make(chan string)
	done := make(chan bool)

	go DatQueue(datCh, done)

	if len(os.Args) > 1 {
		for i := 1; i < len(os.Args); i++ {
			datCh <- os.Args[i]
		}
	} else {
		in := bufio.NewReader(os.Stdin)
		for {
			input, err := in.ReadString('\n')
			if err != nil {
				break
			}
			datCh <- input
		}
	}
	close(datCh)

	<-done

	fmt.Println("OK")
}<commit_msg>multi-thread, use WaitGroup<commit_after>package main

import (
	"bufio"
	"io"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"runtime"
	"strings"
	"sync"

	"code.google.com/p/go.text/encoding/japanese"
	"code.google.com/p/go.text/transform"
)

var wg sync.WaitGroup

const NumProc = 16

// FIXME: brash up regexp
var ImgUrlRegexp *regexp.Regexp = regexp.MustCompile("h?ttp://[0-9a-zA-Z/\\-.%]+?\\.(jpg|jpeg|gif|png)")

var Fetched = struct {
	sync.RWMutex
	m map[string]bool
}{m: make(map[string]bool)}

func Img(url string) {
	defer wg.Done()

	Fetched.Lock()
	if Fetched.m[url] {
		Fetched.Unlock()
		return
	}
	Fetched.m[url] = true
	Fetched.Unlock()
	log.Printf("downloading %v...\n", url)

	out, err := os.Create(filepath.Base(url))
	if err != nil {
		log.Printf("failed to create download file: %v\n", err)
		return
	}
	defer out.Close()

	resp, err := http.Get(url)
	if err != nil {
		log.Printf("failed to download img: %v, err: \n", url, err)
		return
	}
	defer resp.Body.Close()

	n, err := io.Copy(out, resp.Body)
	if err != nil {
		log.Printf("failed to copy img: %v\n", err)
		return
	}
	log.Printf("saved %v (%v bytes)\n", url, n)
}

func ImgQueue(ch chan string) {
	for {
		url, more := <-ch
		if more && url != "" {
			wg.Add(1)
			go Img(url)
		} else {
			break
		}
	}
}

func Dat(url string) {
	defer wg.Done()

	Fetched.Lock()
	if Fetched.m[url] {
		Fetched.Unlock()
		return
	}
	Fetched.m[url] = true
	Fetched.Unlock()
	log.Printf("reading %v...\n", url)

	resp, err := http.Get(url)
	if err != nil {
		log.Printf("failed to fetch: %v, err: %v\n", url, err)
		return
	}
	defer resp.Body.Close()

	imgCh := make(chan string, NumProc)
	go ImgQueue(imgCh)

	reader := transform.NewReader(resp.Body, japanese.ShiftJIS.NewDecoder())
	bufr := bufio.NewReader(reader)
	for {
		lineb, err := bufr.ReadString('\n')
		if err == io.EOF {
			return
		}
		if err != nil {
			log.Printf("failed to read content. err: %v\n", err)
			return
		}
		line := strings.Split(string(lineb), "<>")
		if len(line) >= 4 {
			body := line[3]
			matched := ImgUrlRegexp.FindAllString(body, -1)
			for i := 0; i < len(matched); i++ {
				imgCh <- matched[i]
			}
		}
	}
	close(imgCh)
}

func DatQueue(ch chan string, done chan bool) {
	for {
		url, more := <-ch
		if more && url != "" {
			wg.Add(1)
			go Dat(url)
		} else {
			break
		}
	}

	wg.Wait()
	done <- true
}

func main() {
	runtime.GOMAXPROCS(runtime.NumCPU())

	datCh := make(chan string, NumProc)
	done := make(chan bool)

	go DatQueue(datCh, done)

	if len(os.Args) > 1 {
		for i := 1; i < len(os.Args); i++ {
			datCh <- os.Args[i]
		}
	} else {
		in := bufio.NewReader(os.Stdin)
		for {
			input, err := in.ReadString('\n')
			if err != nil {
				break
			}
			datCh <- input
		}
	}
	close(datCh)

	<-done
}
{'fraction_non_alphanumeric': 0.1862094395280236, 'fraction_numerical': 0.003687315634218289, 'mean_word_length': 2.9512017479970867, 'pattern_counts': {'":': 0, '<': 25, '<?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': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7066053', 'n_tokens_mistral': 2517, 'n_tokens_neox': 2324, 'n_words': 525}