






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


__UpperCAmelCase							:   Dict							=  re.compile(R"\s+")





def 					lowercase_ (    __snake_case     :							int      )     -> Dict:





				'''simple docstring'''


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





def 					lowercase_ (    __snake_case     :							Any      )     -> List[Any]:





				'''simple docstring'''


				snake_case__      :Union[str, Any]         =   [len(__snake_case      ) for line in example["content"].splitlines()]
				return {"line_mean": np.mean(__snake_case      ), "line_max": max(__snake_case      )}





def 					lowercase_ (    __snake_case     :							Union[str, Any]      )     -> List[str]:





				'''simple docstring'''


				snake_case__      :Union[str, Any]         =   np.mean([c.isalnum() for c in example["content"]]      )
				return {"alpha_frac": alpha_frac}





def 					lowercase_ (    __snake_case     :							List[str]	,	__snake_case     :							int      )     -> Dict:





				'''simple docstring'''


				if example["hash"] in uniques:
								uniques.remove(example["hash"]      )
								return True
				else:
								return False





def 					lowercase_ (    __snake_case     :							Dict	,	__snake_case     :							Optional[Any]=5      )     -> str:





				'''simple docstring'''


				snake_case__      :Dict         =   ["auto-generated", "autogenerated", "automatically generated"]
				snake_case__      :List[Any]         =   example["content"].splitlines()
				for _, line in zip(range(__snake_case      )	,	__snake_case      ):
								for keyword in keywords:
												if keyword in line.lower():
																return {"autogenerated": True}
				else:
								return {"autogenerated": False}





def 					lowercase_ (    __snake_case     :							List[str]	,	__snake_case     :							int=5	,	__snake_case     :							List[str]=0.0_5      )     -> Optional[int]:





				'''simple docstring'''


				snake_case__      :int         =   ["unit tests", "test file", "configuration file"]
				snake_case__      :Any         =   example["content"].splitlines()
				snake_case__      :List[Any]         =   0
				snake_case__      :Union[str, Any]         =   0
				# first test
				for _, line in zip(range(__snake_case      )	,	__snake_case      ):
								for keyword in keywords:
												if keyword in line.lower():
																return {"config_or_test": True}
    # second test
				snake_case__      :Dict         =   example["content"].count("\n"      )
				snake_case__      :Optional[Any]         =   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 					lowercase_ (    __snake_case     :							int      )     -> Tuple:





				'''simple docstring'''


				snake_case__      :str         =   ["def ", "class ", "for ", "while "]
				snake_case__      :int         =   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 					lowercase_ (    __snake_case     :							Tuple	,	__snake_case     :							Tuple=4      )     -> List[Any]:





				'''simple docstring'''


				snake_case__      :Tuple         =   example["content"].splitlines()
				snake_case__      :Optional[int]         =   0
				for line in lines:
								counter += line.lower().count("="      )
								if counter > minimum:
												return {"has_few_assignments": False}
				return {"has_few_assignments": True}





def 					lowercase_ (    __snake_case     :							str      )     -> Optional[int]:





				'''simple docstring'''


				snake_case__      :List[str]         =   tokenizer(example["content"]	,	truncation=__snake_case      )["input_ids"]
				snake_case__      :int         =   len(example["content"]      ) / len(__snake_case      )
				return {"ratio": ratio}





def 					lowercase_ (    __snake_case     :							Union[str, Any]      )     -> Union[str, Any]:





				'''simple docstring'''


				snake_case__      :str         =   {}
				results.update(get_hash(__snake_case      )      )
				results.update(line_stats(__snake_case      )      )
				results.update(alpha_stats(__snake_case      )      )
				results.update(char_token_ratio(__snake_case      )      )
				results.update(is_autogenerated(__snake_case      )      )
				results.update(is_config_or_test(__snake_case      )      )
				results.update(has_no_keywords(__snake_case      )      )
				results.update(has_few_assignments(__snake_case      )      )
				return results





def 					lowercase_ (    __snake_case     :							Union[str, Any]	,	__snake_case     :							Union[str, Any]	,	__snake_case     :							Tuple      )     -> Any:





				'''simple docstring'''


				if not check_uniques(__snake_case	,	__snake_case      ):
								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 					lowercase_ (    __snake_case     :							Dict      )     -> Any:





				'''simple docstring'''


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


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

# Load dataset
__UpperCAmelCase							:   int							=  time.time()
__UpperCAmelCase							:   Dict							=  load_dataset(args.dataset_name, split="train")
print(F'''Time to load dataset: {time.time()-t_start:.2f}''')

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

# Deduplicate hashes
__UpperCAmelCase							:   Dict							=  set(ds.unique("hash"))
__UpperCAmelCase							:   List[Any]							=  len(uniques) / len(ds)
print(F'''Fraction of duplicates: {1-frac:.2%}''')

# Deduplicate data and apply heuristics
__UpperCAmelCase							:   List[str]							=  time.time()
__UpperCAmelCase							:   Tuple							=  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:
	__UpperCAmelCase							:   Tuple							=  time.time()
	__UpperCAmelCase						,						__UpperCAmelCase							:   Optional[Any]							=  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
__UpperCAmelCase							:   Optional[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)

__UpperCAmelCase							:   Any							=  output_dir / "data"
data_dir.mkdir(exist_ok=True)

__UpperCAmelCase							:   Optional[Any]							=  time.time()
for file_number, index in enumerate(range(0, len(ds_filter), args.samples_per_file)):
	__UpperCAmelCase							:   Dict							=  str(data_dir / F'''file-{file_number+1:012}.json''')
	__UpperCAmelCase							:   List[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}''')







from typing import Any





def 					lowercase_ (    __snake_case     :							list	,	__snake_case     :							list	,	__snake_case     :							dict	,	__snake_case     :							dict	,	__snake_case     :							dict	,	)     -> list:





				'''simple docstring'''


				_validation(
				    __snake_case	,	__snake_case	,	__snake_case	,	__snake_case	,	__snake_case	,	)
				# Creates data structures and fill initial step
				snake_case__      :dict         =   {}
				snake_case__      :dict         =   {}
				for state in states_space:
								snake_case__      :List[Any]         =   observations_space[0]
								snake_case__      :str         =   (
								    initial_probabilities[state] * emission_probabilities[state][observation]
								)
								snake_case__      :str         =   None

				# Fills the data structure with the probabilities of
				# different transitions and pointers to previous states
				for o in range(1	,	len(__snake_case      )      ):
								snake_case__      :Any         =   observations_space[o]
								snake_case__      :Tuple         =   observations_space[o - 1]
								for state in states_space:
												# Calculates the argmax for probability function
												snake_case__      :Tuple         =   ""
												snake_case__      :Union[str, Any]         =   -1
												for k_state in states_space:
																snake_case__      :int         =   (
																    probabilities[(k_state, prior_observation)]
																    * transition_probabilities[k_state][state]
																    * emission_probabilities[state][observation]
																)
																if probability > max_probability:
																				snake_case__      :str         =   probability
																				snake_case__      :Tuple         =   k_state

            # Update probabilities and pointers dicts
												snake_case__      :List[str]         =   (
												    probabilities[(arg_max, prior_observation)]
												    * transition_probabilities[arg_max][state]
												    * emission_probabilities[state][observation]
												)

												snake_case__      :List[str]         =   arg_max

    # The final observation
				snake_case__      :str         =   observations_space[len(__snake_case      ) - 1]

				# argmax for given final observation
				snake_case__      :Optional[int]         =   ""
				snake_case__      :List[str]         =   -1
				for k_state in states_space:
								snake_case__      :List[str]         =   probabilities[(k_state, final_observation)]
								if probability > max_probability:
												snake_case__      :List[str]         =   probability
												snake_case__      :int         =   k_state
				snake_case__      :Any         =   arg_max

				# Process pointers backwards
				snake_case__      :int         =   last_state
				snake_case__      :List[str]         =   []
				for o in range(len(__snake_case      ) - 1	,	-1	,	-1      ):
								result.append(__snake_case      )
								snake_case__      :List[str]         =   pointers[previous, observations_space[o]]
				result.reverse()

				return result





def 					lowercase_ (    __snake_case     :							Any	,	__snake_case     :							Any	,	__snake_case     :							Any	,	__snake_case     :							Any	,	__snake_case     :							Any	,	)     -> None:





				'''simple docstring'''


				_validate_not_empty(
				    __snake_case	,	__snake_case	,	__snake_case	,	__snake_case	,	__snake_case	,	)
				_validate_lists(__snake_case	,	__snake_case      )
				_validate_dicts(
				    __snake_case	,	__snake_case	,	__snake_case      )





def 					lowercase_ (    __snake_case     :							Any	,	__snake_case     :							Any	,	__snake_case     :							Any	,	__snake_case     :							Any	,	__snake_case     :							Any	,	)     -> None:





				'''simple docstring'''


				if not all(
				    [
				        observations_space,
				        states_space,
				        initial_probabilities,
				        transition_probabilities,
				        emission_probabilities,
				    ]      ):
								raise ValueError("There's an empty parameter"      )





def 					lowercase_ (    __snake_case     :							Any	,	__snake_case     :							Any      )     -> None:





				'''simple docstring'''


				_validate_list(__snake_case	,	"observations_space"      )
				_validate_list(__snake_case	,	"states_space"      )





def 					lowercase_ (    __snake_case     :							Any	,	__snake_case     :							str      )     -> None:





				'''simple docstring'''


				if not isinstance(_object	,	__snake_case      ):
								snake_case__      :Optional[int]         =   F'{var_name} must be a list'
								raise ValueError(__snake_case      )
				else:
								for x in _object:
												if not isinstance(__snake_case	,	__snake_case      ):
																snake_case__      :Any         =   F'{var_name} must be a list of strings'
																raise ValueError(__snake_case      )





def 					lowercase_ (    __snake_case     :							Any	,	__snake_case     :							Any	,	__snake_case     :							Any	,	)     -> None:





				'''simple docstring'''


				_validate_dict(__snake_case	,	"initial_probabilities"	,	__snake_case      )
				_validate_nested_dict(__snake_case	,	"transition_probabilities"      )
				_validate_nested_dict(__snake_case	,	"emission_probabilities"      )





def 					lowercase_ (    __snake_case     :							Any	,	__snake_case     :							str      )     -> None:





				'''simple docstring'''


				_validate_dict(_object	,	__snake_case	,	__snake_case      )
				for x in _object.values():
								_validate_dict(__snake_case	,	__snake_case	,	__snake_case	,	__snake_case      )





def 					lowercase_ (    __snake_case     :							Any	,	__snake_case     :							str	,	__snake_case     :							type	,	__snake_case     :							bool = False      )     -> None:





				'''simple docstring'''


				if not isinstance(_object	,	__snake_case      ):
								snake_case__      :str         =   F'{var_name} must be a dict'
								raise ValueError(__snake_case      )
				if not all(isinstance(__snake_case	,	__snake_case      ) for x in _object      ):
								snake_case__      :List[Any]         =   F'{var_name} all keys must be strings'
								raise ValueError(__snake_case      )
				if not all(isinstance(__snake_case	,	__snake_case      ) for x in _object.values()      ):
								snake_case__      :Optional[int]         =   "nested dictionary " if nested else ""
								snake_case__      :int         =   F'{var_name} {nested_text}all values must be {value_type.__name__}'
								raise ValueError(__snake_case      )


if __name__ == "__main__":
	from doctest import testmod

	testmod()