<commit_before>import argparse
import os
import sys
import threading
import time
import Queue

from boto.s3.connection import S3Connection

import humanize

DEBUG = False

parser = argparse.ArgumentParser(description = "Download a bucket from S3")
parser.add_argument('--workers', '-n', metavar='N', type=int, help="Number of worker threads", default = 100)
parser.add_argument('bucket', metavar='<s3 bucket>', type=str, help="S3 bucket to download")

args = parser.parse_args()

print "Workers", args.workers

conn = S3Connection()
bucket = conn.get_bucket(args.bucket)
items = bucket.list()
#items = [(i.name, i.size, i) for i in items]

def download(key):
    #if os.path.dirname(key.name) and not os.path.exists(os.path.dirname(key.name)):
    # Try/except avoids threading conflict issues. Errors will still
    # be caught if this fails at the time of the attempted download.
    # Without this, occasionally two workers will contend, try to
    # create the same directory, and fail
    try:
        os.makedirs(os.path.dirname(key.name))
    except: 
        pass 
    if not os.path.exists(key.name):
        key.get_contents_to_filename(key.name)

task_queue = Queue.PriorityQueue()
lock = threading.Lock()
size = 0

def worker():
    while True:
        item = task_queue.get()
        if DEBUG:
            print "Grabbing", item
        try: 
            download(item)
            with lock:
                global size
                size = size - item.size
        except:
            print "Download of ", item, "failed"
            print sys.exc_info()
        task_queue.task_done()

print "Queuing bucket..."

directories = set()

for item in items:
    directories.add(os.path.dirname(item.name))

count  = 0
for item in items:
    if item.name not in directories:
        task_queue.put(item, -item.size)
        count = count + 1
        size = size + item.size
    
total_size = size

print "Running workers..."

for i in range(args.workers):
    t = threading.Thread(target = worker)
    t.daemon = True
    t.start()<commit_msg>Skip directories<commit_after>import argparse
import os
import sys
import threading
import time
import Queue

from boto.s3.connection import S3Connection

import humanize

DEBUG = False

parser = argparse.ArgumentParser(description = "Download a bucket from S3")
parser.add_argument('--workers', '-n', metavar='N', type=int, help="Number of worker threads", default = 100)
parser.add_argument('bucket', metavar='<s3 bucket>', type=str, help="S3 bucket to download")

args = parser.parse_args()

print "Workers", args.workers

conn = S3Connection()
bucket = conn.get_bucket(args.bucket)
items = bucket.list()
#items = [(i.name, i.size, i) for i in items]

def download(key):
    #if os.path.dirname(key.name) and not os.path.exists(os.path.dirname(key.name)):
    # Try/except avoids threading conflict issues. Errors will still
    # be caught if this fails at the time of the attempted download.
    # Without this, occasionally two workers will contend, try to
    # create the same directory, and fail
    try:
        os.makedirs(os.path.dirname(key.name))
    except: 
        pass 
    if not os.path.exists(key.name):
        key.get_contents_to_filename(key.name)

task_queue = Queue.PriorityQueue()
lock = threading.Lock()
size = 0

def worker():
    while True:
        item = task_queue.get()
        if DEBUG:
            print "Grabbing", item
        try: 
            download(item)
            with lock:
                global size
                size = size - item.size
        except:
            print "Download of ", item, "failed"
            print sys.exc_info()
        task_queue.task_done()

print "Queuing bucket..."

directories = set()

for item in items:
    directories.add(os.path.dirname(item.name))

count  = 0
for item in items:
    # S3 can't tell me if something is a bucket. 
    # I used 3 heuristics. I think the first two might be sufficient. 
    if item.name in directories:
        continue
    if item.name.endswith('/'):
        continue
    if item.content_type == 'application/x-directory':
        continue
    task_queue.put(item, -item.size)
    count = count + 1
    size = size + item.size
    
total_size = size

print "Running workers..."

for i in range(args.workers):
    t = threading.Thread(target = worker)
    t.daemon = True
    t.start()
{'fraction_non_alphanumeric': 0.10835629017447199, 'fraction_numerical': 0.005968778696051423, 'mean_word_length': 2.7560344827586207, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6286524', 'n_tokens_mistral': 1355, 'n_tokens_neox': 1318, 'n_words': 414}