



'''simple docstring'''


import gzip
import hashlib
import json
import multiprocessing
import os
import re
import shutil
import time
from pathlib import Path

import numpy as np
from arguments import PreprocessingArguments
from datasets import load_dataset
from minhash_deduplication import deduplicate_dataset

from transformers import AutoTokenizer, HfArgumentParser


lowercase_      :				Optional[int]   					=    re.compile(r'''\s+''')







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      str						):
						return {"hash": hashlib.mda(re.sub(lowercase_ ,  """""" ,  example["""content"""]						).encode("""utf-8"""						)						).hexdigest()}







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      Any						):
						lowercase        =							[len(lowercase_						) for line in example["""content"""].splitlines()]
						return {"line_mean": np.mean(lowercase_						), "line_max": max(lowercase_						)}







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      List[Any]						):
						lowercase        =							np.mean([c.isalnum() for c in example["""content"""]]						)
						return {"alpha_frac": alpha_frac}







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      List[str] ,  lowercase_    :      Optional[Any]						):
						if example["hash"] in uniques:
												uniques.remove(example["""hash"""]						)
												return True
						else:
												return False







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      List[str] ,  lowercase_    :      Optional[Any]=5						):
						lowercase        =							["""auto-generated""", """autogenerated""", """automatically generated"""]
						lowercase        =							example["""content"""].splitlines()
						for _, line in zip(range(lowercase_						) ,  lowercase_						):
												for keyword in keywords:
																		if keyword in line.lower():
																								return {"autogenerated": True}
						else:
												return {"autogenerated": False}







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      int ,  lowercase_    :      int=5 ,  lowercase_    :      Optional[Any]=0.05						):
						lowercase        =							["""unit tests""", """test file""", """configuration file"""]
						lowercase        =							example["""content"""].splitlines()
						lowercase        =							0
						lowercase        =							0
						# first test
						for _, line in zip(range(lowercase_						) ,  lowercase_						):
												for keyword in keywords:
																		if keyword in line.lower():
																								return {"config_or_test": True}
    # second test
						lowercase        =							example["""content"""].count("""\n"""						)
						lowercase        =							int(coeff * nlines						)
						for line in lines:
												count_config += line.lower().count("""config"""						)
												count_test += line.lower().count("""test"""						)
												if count_config > threshold or count_test > threshold:
																		return {"config_or_test": True}
						return {"config_or_test": False}







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      List[str]						):
						lowercase        =							["""def """, """class """, """for """, """while """]
						lowercase        =							example["""content"""].splitlines()
						for line in lines:
												for keyword in keywords:
																		if keyword in line.lower():
																								return {"has_no_keywords": False}
						return {"has_no_keywords": True}







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      int ,  lowercase_    :      Tuple=4						):
						lowercase        =							example["""content"""].splitlines()
						lowercase        =							0
						for line in lines:
												counter += line.lower().count("""="""						)
												if counter > minimum:
																		return {"has_few_assignments": False}
						return {"has_few_assignments": True}







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      List[str]						):
						lowercase        =							tokenizer(example["""content"""] ,  truncation=lowercase_						)["""input_ids"""]
						lowercase        =							len(example["""content"""]						) / len(lowercase_						)
						return {"ratio": ratio}







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      List[Any]						):
						lowercase        =							{}
						results.update(get_hash(lowercase_						)						)
						results.update(line_stats(lowercase_						)						)
						results.update(alpha_stats(lowercase_						)						)
						results.update(char_token_ratio(lowercase_						)						)
						results.update(is_autogenerated(lowercase_						)						)
						results.update(is_config_or_test(lowercase_						)						)
						results.update(has_no_keywords(lowercase_						)						)
						results.update(has_few_assignments(lowercase_						)						)
						return results







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      Tuple ,  lowercase_    :      str ,  lowercase_    :      Optional[int]						):
						if not check_uniques(lowercase_ ,  lowercase_						):
												return False
						elif example["autogenerated"]:
												return False
						elif example["line_max"] > args.line_max:
												return False
						elif example["line_mean"] > args.line_mean:
												return False
						elif example["alpha_frac"] < args.alpha_frac:
												return False
						elif example["ratio"] < args.min_token_ratio:
												return False
						elif example["config_or_test"] and np.random.rand() <= args.filter_proba:
												return False
						elif example["has_no_keywords"] and np.random.rand() <= args.filter_proba:
												return False
						elif example["has_few_assignments"]:
												return False
						else:
												return True







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      Any						):
						with open(lowercase_ ,  """rb"""						) as f_in:
												with gzip.open(str(lowercase_						) + """.gz""" ,  """wb""" ,  compresslevel=6						) as f_out:
																		shutil.copyfileobj(lowercase_ ,  lowercase_						)
						os.unlink(lowercase_						)


# Settings
lowercase_      :				Optional[int]   					=    HfArgumentParser(PreprocessingArguments)
lowercase_      :				Union[str, Any]   					=    parser.parse_args()
if args.num_workers is None:
						lowercase_      :				Union[str, Any]   					=    multiprocessing.cpu_count()
lowercase_      :				Optional[Any]   					=    AutoTokenizer.from_pretrained(args.tokenizer_dir)

# Load dataset
lowercase_      :				Union[str, Any]   					=    time.time()
lowercase_      :				Optional[int]   					=    load_dataset(args.dataset_name, split='''train''')
print(f'''Time to load dataset: {time.time()-t_start:.2f}''')

# Run preprocessing
lowercase_      :				Optional[Any]   					=    time.time()
lowercase_      :				Dict   					=    ds.map(preprocess, num_proc=args.num_workers)
print(f'''Time to preprocess dataset: {time.time()-t_start:.2f}''')

# Deduplicate hashes
lowercase_      :				Any   					=    set(ds.unique('''hash'''))
lowercase_      :				str   					=    len(uniques) / len(ds)
print(f'''Fraction of duplicates: {1-frac:.2%}''')

# Deduplicate data and apply heuristics
lowercase_      :				int   					=    time.time()
lowercase_      :				List[Any]   					=    ds.filter(filter, fn_kwargs={'''uniques''': uniques, '''args''': args})
print(f'''Time to filter dataset: {time.time()-t_start:.2f}''')
print(f'''Size of filtered dataset: {len(ds_filter)}''')

# Deduplicate with minhash and jaccard similarity
if args.near_deduplication:
						lowercase_      :				List[str]   					=    time.time()
						lowercase_					,       lowercase_      :				int   					=    deduplicate_dataset(ds_filter, args.jaccard_threshold)
						print(f'''Time to deduplicate dataset: {time.time()-t_start:.2f}''')
						print(f'''Size of deduplicate dataset: {len(ds_filter)}''')

# Save data in batches of samples_per_file
lowercase_      :				int   					=    Path(args.output_dir)
output_dir.mkdir(exist_ok=True)

# save duplicate_clusters in the output_dir as artifacts
# not sure it is the right place the save it
if args.near_deduplication:
						with open(output_dir / '''duplicate_clusters.json''', '''w''') as f:
												json.dump(duplicate_clusters, f)

lowercase_      :				List[str]   					=    output_dir / '''data'''
data_dir.mkdir(exist_ok=True)

lowercase_      :				List[Any]   					=    time.time()
for file_number, index in enumerate(range(0, len(ds_filter), args.samples_per_file)):
						lowercase_      :				Optional[int]   					=    str(data_dir / f'''file-{file_number+1:012}.json''')
						lowercase_      :				Any   					=    min(len(ds_filter), index + args.samples_per_file)
						ds_filter.select(list(range(index, end_index))).to_json(file_path)
						compress_file(file_path)
print(f'''Time to save dataset: {time.time()-t_start:.2f}''')







'''simple docstring'''


from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowercase_      :				Optional[Any]   					=    logging.get_logger(__name__)

lowercase_      :				int   					=    {
    '''bigcode/gpt_bigcode-santacoder''': '''https://huggingface.co/bigcode/gpt_bigcode-santacoder/resolve/main/config.json''',
}




class      __UpperCamelCase       (_UpperCAmelCase			):
	__A			  =					'''gpt_bigcode'''
	__A			  =					['''past_key_values''']
	__A			  =					{
	    '''hidden_size''': '''n_embd''',
	    '''max_position_embeddings''': '''n_positions''',
	    '''num_attention_heads''': '''n_head''',
	    '''num_hidden_layers''': '''n_layer''',
	}







	def __init__(    self ,				_lowerCAmelCase=5_0257 ,				_lowerCAmelCase=1024 ,				_lowerCAmelCase=768 ,				_lowerCAmelCase=12 ,				_lowerCAmelCase=12 ,				_lowerCAmelCase=None ,				_lowerCAmelCase="gelu_pytorch_tanh" ,				_lowerCAmelCase=0.1 ,				_lowerCAmelCase=0.1 ,				_lowerCAmelCase=0.1 ,				_lowerCAmelCase=1E-5 ,				_lowerCAmelCase=0.02 ,				_lowerCAmelCase=True ,				_lowerCAmelCase=True ,				_lowerCAmelCase=5_0256 ,				_lowerCAmelCase=5_0256 ,				_lowerCAmelCase=True ,				_lowerCAmelCase=True ,				_lowerCAmelCase=True ,				**_lowerCAmelCase ,				)    ->    Optional[int]:
							'''simple docstring'''






							lowercase        =							vocab_size
							lowercase        =							n_positions
							lowercase        =							n_embd
							lowercase        =							n_layer
							lowercase        =							n_head
							lowercase        =							n_inner
							lowercase        =							activation_function
							lowercase        =							resid_pdrop
							lowercase        =							embd_pdrop
							lowercase        =							attn_pdrop
							lowercase        =							layer_norm_epsilon
							lowercase        =							initializer_range
							lowercase        =							scale_attn_weights
							lowercase        =							use_cache
							lowercase        =							attention_softmax_in_fpaa
							lowercase        =							scale_attention_softmax_in_fpaa
							lowercase        =							multi_query

							lowercase        =							bos_token_id
							lowercase        =							eos_token_id

							super().__init__(bos_token_id=_lowerCAmelCase ,				eos_token_id=_lowerCAmelCase ,				**_lowerCAmelCase						)


