






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


__lowerCamelCase :      List[str]										=				re.compile(R"""\s+""")







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







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







def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							List[str]       ):
	snake_case__			:				Tuple       =  np.mean([c.isalnum() for c in example["content"]]       )
	return {"alpha_frac": alpha_frac}







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







def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							List[Any]					,  snake_case_ :							Dict=5       ):
	snake_case__			:				Dict       =  ["auto-generated", "autogenerated", "automatically generated"]
	snake_case__			:				str       =  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 					SCREAMING_SNAKE_CASE    (			snake_case_ :							Tuple					,  snake_case_ :							List[str]=5					,  snake_case_ :							Dict=0.05       ):
	snake_case__			:				int       =  ["unit tests", "test file", "configuration file"]
	snake_case__			:				Dict       =  example["content"].splitlines()
	snake_case__			:				List[str]       =  0
	snake_case__			:				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__			:				Optional[int]       =  example["content"].count("\n"       )
	snake_case__			:				int       =  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    (			snake_case_ :							Optional[int]       ):
	snake_case__			:				Any       =  ["def ", "class ", "for ", "while "]
	snake_case__			:				Union[str, Any]       =  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    (			snake_case_ :							int					,  snake_case_ :							int=4       ):
	snake_case__			:				Optional[int]       =  example["content"].splitlines()
	snake_case__			:				Any       =  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    (			snake_case_ :							List[str]       ):
	snake_case__			:				str       =  tokenizer(example["content"]					,  truncation=snake_case_       )["input_ids"]
	snake_case__			:				int       =  len(example["content"]       ) / len(snake_case_       )
	return {"ratio": ratio}







def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							Dict       ):
	snake_case__			:				Dict       =  {}
	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 					SCREAMING_SNAKE_CASE    (			snake_case_ :							List[Any]					,  snake_case_ :							Union[str, Any]					,  snake_case_ :							Dict       ):
	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 					SCREAMING_SNAKE_CASE    (			snake_case_ :							Tuple       ):
	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
__lowerCamelCase :      Optional[Any]										=				HfArgumentParser(PreprocessingArguments)
__lowerCamelCase :      List[str]										=				parser.parse_args()
if args.num_workers is None:
	__lowerCamelCase :      Optional[int]										=				multiprocessing.cpu_count()
__lowerCamelCase :      Dict										=				AutoTokenizer.from_pretrained(args.tokenizer_dir)

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

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

# Deduplicate hashes
__lowerCamelCase :      Dict										=				set(ds.unique("""hash"""))
__lowerCamelCase :      int										=				len(uniques) / len(ds)
print(f"Fraction of duplicates: {1-frac:.2%}")

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

__lowerCamelCase :      Tuple										=				output_dir / """data"""
data_dir.mkdir(exist_ok=True)

__lowerCamelCase :      Optional[Any]										=				time.time()
for file_number, index in enumerate(range(0, len(ds_filter), args.samples_per_file)):
	__lowerCamelCase :      Optional[Any]										=				str(data_dir / f"file-{file_number+1:012}.json")
	__lowerCamelCase :      List[str]										=				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 tempfile import TemporaryDirectory
from unittest import TestCase
from unittest.mock import MagicMock, patch

from transformers import AutoModel, TFAutoModel
from transformers.onnx import FeaturesManager
from transformers.testing_utils import SMALL_MODEL_IDENTIFIER, require_tf, require_torch



@require_torch
@require_tf
class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					def 	_lowercase					( self					:  Optional[Any]	):
						snake_case__			:				Optional[Any]       =  SMALL_MODEL_IDENTIFIER
						snake_case__			:				Any       =  "pt"
						snake_case__			:				Any       =  "tf"



					def 	_lowercase					( self					:  Union[str, Any]							,		__A					:  List[Any]	):
						snake_case__			:				int       =  AutoModel.from_pretrained(self.test_model	)
						model_pt.save_pretrained(__A	)



					def 	_lowercase					( self					:  Optional[int]							,		__A					:  Tuple	):
						snake_case__			:				List[Any]       =  TFAutoModel.from_pretrained(self.test_model							,		from_pt=__A	)
						model_tf.save_pretrained(__A	)



					def 	_lowercase					( self					:  str	):
						snake_case__			:				Optional[Any]       =  "mock_framework"

						# Framework provided - return whatever the user provides
						snake_case__			:				Optional[Any]       =  FeaturesManager.determine_framework(self.test_model							,		__A	)
						self.assertEqual(__A							,		__A	)

						# Local checkpoint and framework provided - return provided framework
						# PyTorch checkpoint
						with TemporaryDirectory() as local_pt_ckpt:
							self._setup_pt_ckpt(__A	)
							snake_case__			:				Optional[int]       =  FeaturesManager.determine_framework(__A							,		__A	)
							self.assertEqual(__A							,		__A	)

						# TensorFlow checkpoint
						with TemporaryDirectory() as local_tf_ckpt:
							self._setup_tf_ckpt(__A	)
							snake_case__			:				int       =  FeaturesManager.determine_framework(__A							,		__A	)
							self.assertEqual(__A							,		__A	)



					def 	_lowercase					( self					:  Dict	):
						# PyTorch checkpoint
						with TemporaryDirectory() as local_pt_ckpt:
							self._setup_pt_ckpt(__A	)
							snake_case__			:				List[str]       =  FeaturesManager.determine_framework(__A	)
							self.assertEqual(__A							,		self.framework_pt	)

						# TensorFlow checkpoint
						with TemporaryDirectory() as local_tf_ckpt:
							self._setup_tf_ckpt(__A	)
							snake_case__			:				Tuple       =  FeaturesManager.determine_framework(__A	)
							self.assertEqual(__A							,		self.framework_tf	)

						# Invalid local checkpoint
						with TemporaryDirectory() as local_invalid_ckpt:
							with self.assertRaises(__A	):
								snake_case__			:				int       =  FeaturesManager.determine_framework(__A	)






					def 	_lowercase					( self					:  Dict	):
						snake_case__			:				Dict       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_tf_available"							,		__A	):
							snake_case__			:				List[Any]       =  FeaturesManager.determine_framework(self.test_model	)
							self.assertEqual(__A							,		self.framework_pt	)

						# PyTorch not in environment -> use TensorFlow
						snake_case__			:				Tuple       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_torch_available"							,		__A	):
							snake_case__			:				int       =  FeaturesManager.determine_framework(self.test_model	)
							self.assertEqual(__A							,		self.framework_tf	)

						# Both in environment -> use PyTorch
						snake_case__			:				Dict       =  MagicMock(return_value=__A	)
						snake_case__			:				Optional[int]       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_tf_available"							,		__A	), patch(
						    "transformers.onnx.features.is_torch_available"							,		__A	):
							snake_case__			:				Optional[Any]       =  FeaturesManager.determine_framework(self.test_model	)
							self.assertEqual(__A							,		self.framework_pt	)

						# Both not in environment -> raise error
						snake_case__			:				List[str]       =  MagicMock(return_value=__A	)
						snake_case__			:				Optional[Any]       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_tf_available"							,		__A	), patch(
						    "transformers.onnx.features.is_torch_available"							,		__A	):
							with self.assertRaises(__A	):
								snake_case__			:				Optional[Any]       =  FeaturesManager.determine_framework(self.test_model	)



