






"""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


A_    =							re.compile(r'''\s+''')




def      _lowerCAmelCase		(							UpperCAmelCase__       :					List[Any]      )		->str:
						return {"hash": hashlib.mda(re.sub(lowercase__,						"""""",						example["""content"""]      ).encode("""utf-8"""      )      ).hexdigest()}




def      _lowerCAmelCase		(							UpperCAmelCase__       :					str      )		->Tuple:
						A__						:  Optional[int]		  =  [len(lowercase__      ) for line in example["""content"""].splitlines()]
						return {"line_mean": np.mean(lowercase__      ), "line_max": max(lowercase__      )}




def      _lowerCAmelCase		(							UpperCAmelCase__       :					List[Any]      )		->Optional[Any]:
						A__						:  Optional[int]		  =  np.mean([c.isalnum() for c in example["""content"""]]      )
						return {"alpha_frac": alpha_frac}




def      _lowerCAmelCase		(							UpperCAmelCase__       :					Any,						UpperCAmelCase__       :					int      )		->Tuple:
						if example["hash"] in uniques:
												uniques.remove(example["""hash"""]      )
												return True
						else:
												return False




def      _lowerCAmelCase		(							UpperCAmelCase__       :					int,						UpperCAmelCase__       :					Tuple=5      )		->Union[str, Any]:
						A__						:  Tuple		  =  ["""auto-generated""", """autogenerated""", """automatically generated"""]
						A__						:  List[str]		  =  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      _lowerCAmelCase		(							UpperCAmelCase__       :					Dict,						UpperCAmelCase__       :					Any=5,						UpperCAmelCase__       :					Any=0.05      )		->Optional[Any]:
						A__						:  Optional[Any]		  =  ["""unit tests""", """test file""", """configuration file"""]
						A__						:  Any		  =  example["""content"""].splitlines()
						A__						:  Tuple		  =  0
						A__						:  str		  =  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
						A__						:  Optional[int]		  =  example["""content"""].count("""\n"""      )
						A__						:  Optional[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      _lowerCAmelCase		(							UpperCAmelCase__       :					Optional[int]      )		->Optional[Any]:
						A__						:  Optional[int]		  =  ["""def """, """class """, """for """, """while """]
						A__						:  Dict		  =  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      _lowerCAmelCase		(							UpperCAmelCase__       :					List[Any],						UpperCAmelCase__       :					Optional[Any]=4      )		->List[Any]:
						A__						:  Optional[int]		  =  example["""content"""].splitlines()
						A__						:  List[str]		  =  0
						for line in lines:
												counter += line.lower().count("""="""      )
												if counter > minimum:
																		return {"has_few_assignments": False}
						return {"has_few_assignments": True}




def      _lowerCAmelCase		(							UpperCAmelCase__       :					List[str]      )		->List[Any]:
						A__						:  List[Any]		  =  tokenizer(example["""content"""],						truncation=lowercase__      )["""input_ids"""]
						A__						:  Union[str, Any]		  =  len(example["""content"""]      ) / len(lowercase__      )
						return {"ratio": ratio}




def      _lowerCAmelCase		(							UpperCAmelCase__       :					Optional[Any]      )		->Optional[int]:
						A__						:  Optional[Any]		  =  {}
						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      _lowerCAmelCase		(							UpperCAmelCase__       :					List[str],						UpperCAmelCase__       :					List[Any],						UpperCAmelCase__       :					Optional[int]      )		->str:
						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      _lowerCAmelCase		(							UpperCAmelCase__       :					List[Any]      )		->str:
						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
A_    =							HfArgumentParser(PreprocessingArguments)
A_    =							parser.parse_args()
if args.num_workers is None:
						A_    =							multiprocessing.cpu_count()
A_    =							AutoTokenizer.from_pretrained(args.tokenizer_dir)

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

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

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

# Deduplicate data and apply heuristics
A_    =							time.time()
A_    =							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:
						A_    =							time.time()
						A_    ,	A_    =							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
A_    =							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)

A_    =							output_dir / '''data'''
data_dir.mkdir(exist_ok=True)

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



import unittest

from transformers import (
    MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING,
    TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING,
    TextClassificationPipeline,
    pipeline,
)
from transformers.testing_utils import is_pipeline_test, nested_simplify, require_tf, require_torch, slow

from .test_pipelines_common import ANY


# These 2 model types require different inputs than those of the usual text models.
A_    =							{'''LayoutLMv2Config''', '''LayoutLMv3Config'''}






@is_pipeline_test
class 							__SCREAMING_SNAKE_CASE						(				unittest.TestCase		):
							snake_case_     =							MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING
							snake_case_     =							TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING

							if model_mapping is not None:
														snake_case_     =							{config: model for config, model in model_mapping.items() if config.__name__ not in _TO_SKIP}
							if tf_model_mapping is not None:
														snake_case_     =							{
														    config: model for config, model in tf_model_mapping.items() if config.__name__ not in _TO_SKIP
														}


							@require_torch
							def 						_UpperCamelCase       (			self      :						Tuple							):
													'''simple docstring'''





													A__						:  Tuple		  =  pipeline(
													    task="""text-classification"""			,    model="""hf-internal-testing/tiny-random-distilbert"""			,    framework="""pt"""							)

													A__						:  Dict		  =  text_classifier("""This is great !"""							)
													self.assertEqual(nested_simplify(snake_case							)			,    [{"""label""": """LABEL_0""", """score""": 0.504}]							)

													A__						:  Optional[int]		  =  text_classifier("""This is great !"""			,    top_k=2							)
													self.assertEqual(
													    nested_simplify(snake_case							)			,    [{"""label""": """LABEL_0""", """score""": 0.504}, {"""label""": """LABEL_1""", """score""": 0.496}]							)

													A__						:  Any		  =  text_classifier(["""This is great !""", """This is bad"""]			,    top_k=2							)
													self.assertEqual(
													    nested_simplify(snake_case							)			,    [
													        [{"""label""": """LABEL_0""", """score""": 0.504}, {"""label""": """LABEL_1""", """score""": 0.496}],
													        [{"""label""": """LABEL_0""", """score""": 0.504}, {"""label""": """LABEL_1""", """score""": 0.496}],
													    ]			,    )

													A__						:  Optional[Any]		  =  text_classifier("""This is great !"""			,    top_k=1							)
													self.assertEqual(nested_simplify(snake_case							)			,    [{"""label""": """LABEL_0""", """score""": 0.504}]							)

													# Legacy behavior
													A__						:  str		  =  text_classifier("""This is great !"""			,    return_all_scores=snake_case							)
													self.assertEqual(nested_simplify(snake_case							)			,    [{"""label""": """LABEL_0""", """score""": 0.504}]							)

													A__						:  int		  =  text_classifier("""This is great !"""			,    return_all_scores=snake_case							)
													self.assertEqual(
													    nested_simplify(snake_case							)			,    [[{"""label""": """LABEL_0""", """score""": 0.504}, {"""label""": """LABEL_1""", """score""": 0.496}]]							)

													A__						:  str		  =  text_classifier(["""This is great !""", """Something else"""]			,    return_all_scores=snake_case							)
													self.assertEqual(
													    nested_simplify(snake_case							)			,    [
													        [{"""label""": """LABEL_0""", """score""": 0.504}, {"""label""": """LABEL_1""", """score""": 0.496}],
													        [{"""label""": """LABEL_0""", """score""": 0.504}, {"""label""": """LABEL_1""", """score""": 0.496}],
													    ]			,    )

													A__						:  Optional[int]		  =  text_classifier(["""This is great !""", """Something else"""]			,    return_all_scores=snake_case							)
													self.assertEqual(
													    nested_simplify(snake_case							)			,    [
													        {"""label""": """LABEL_0""", """score""": 0.504},
													        {"""label""": """LABEL_0""", """score""": 0.504},
													    ]			,    )


							@require_torch
							def 						_UpperCamelCase       (			self      :						int							):
													'''simple docstring'''





													import torch

													A__						:  Tuple		  =  pipeline(
													    task="""text-classification"""			,    model="""hf-internal-testing/tiny-random-distilbert"""			,    framework="""pt"""			,    device=torch.device("""cpu"""							)			,    )

													A__						:  List[str]		  =  text_classifier("""This is great !"""							)
													self.assertEqual(nested_simplify(snake_case							)			,    [{"""label""": """LABEL_0""", """score""": 0.504}]							)


							@require_tf
							def 						_UpperCamelCase       (			self      :						List[str]							):
													'''simple docstring'''





													A__						:  str		  =  pipeline(
													    task="""text-classification"""			,    model="""hf-internal-testing/tiny-random-distilbert"""			,    framework="""tf"""							)

													A__						:  Optional[Any]		  =  text_classifier("""This is great !"""							)
													self.assertEqual(nested_simplify(snake_case							)			,    [{"""label""": """LABEL_0""", """score""": 0.504}]							)


							@slow
							@require_torch
							def 						_UpperCamelCase       (			self      :						List[Any]							):
													'''simple docstring'''





													A__						:  Optional[int]		  =  pipeline("""text-classification"""							)

													A__						:  Optional[int]		  =  text_classifier("""This is great !"""							)
													self.assertEqual(nested_simplify(snake_case							)			,    [{"""label""": """POSITIVE""", """score""": 1.0}]							)
													A__						:  Optional[Any]		  =  text_classifier("""This is bad !"""							)
													self.assertEqual(nested_simplify(snake_case							)			,    [{"""label""": """NEGATIVE""", """score""": 1.0}]							)
													A__						:  Dict		  =  text_classifier("""Birds are a type of animal"""							)
													self.assertEqual(nested_simplify(snake_case							)			,    [{"""label""": """POSITIVE""", """score""": 0.988}]							)


							@slow
							@require_tf
							def 						_UpperCamelCase       (			self      :						Any							):
													'''simple docstring'''





													A__						:  Union[str, Any]		  =  pipeline("""text-classification"""			,    framework="""tf"""							)

													A__						:  Dict		  =  text_classifier("""This is great !"""							)
													self.assertEqual(nested_simplify(snake_case							)			,    [{"""label""": """POSITIVE""", """score""": 1.0}]							)
													A__						:  Optional[int]		  =  text_classifier("""This is bad !"""							)
													self.assertEqual(nested_simplify(snake_case							)			,    [{"""label""": """NEGATIVE""", """score""": 1.0}]							)
													A__						:  Tuple		  =  text_classifier("""Birds are a type of animal"""							)
													self.assertEqual(nested_simplify(snake_case							)			,    [{"""label""": """POSITIVE""", """score""": 0.988}]							)


							def 						_UpperCamelCase       (			self      :						Optional[Any]			,    snake_case      :						Any			,    snake_case      :						List[Any]			,    snake_case      :						Any							):
													'''simple docstring'''





													A__						:  Union[str, Any]		  =  TextClassificationPipeline(model=snake_case			,    tokenizer=snake_case							)
													return text_classifier, ["HuggingFace is in", "This is another test"]


							def 						_UpperCamelCase       (			self      :						int			,    snake_case      :						Union[str, Any]			,    snake_case      :						Dict							):
													'''simple docstring'''





													A__						:  List[str]		  =  text_classifier.model
													# Small inputs because BartTokenizer tiny has maximum position embeddings = 22
													A__						:  int		  =  """HuggingFace is in"""
													A__						:  List[Any]		  =  text_classifier(snake_case							)

													self.assertEqual(nested_simplify(snake_case							)			,    [{"""label""": ANY(snake_case							), """score""": ANY(snake_case							)}]							)
													self.assertTrue(outputs[0]["""label"""] in model.config.idalabel.values()							)

													A__						:  Any		  =  ["""HuggingFace is in """, """Paris is in France"""]
													A__						:  Any		  =  text_classifier(snake_case							)
													self.assertEqual(
													    nested_simplify(snake_case							)			,    [{"""label""": ANY(snake_case							), """score""": ANY(snake_case							)}, {"""label""": ANY(snake_case							), """score""": ANY(snake_case							)}]			,    )
													self.assertTrue(outputs[0]["""label"""] in model.config.idalabel.values()							)
													self.assertTrue(outputs[1]["""label"""] in model.config.idalabel.values()							)

													# Forcing to get all results with `top_k=None`
													# This is NOT the legacy format
													A__						:  List[Any]		  =  text_classifier(snake_case			,    top_k=snake_case							)
													A__						:  Tuple		  =  len(model.config.idalabel.values()							)
													self.assertEqual(
													    nested_simplify(snake_case							)			,    [[{"""label""": ANY(snake_case							), """score""": ANY(snake_case							)}] * N, [{"""label""": ANY(snake_case							), """score""": ANY(snake_case							)}] * N]			,    )

													A__						:  Optional[Any]		  =  {"""text""": """HuggingFace is in """, """text_pair""": """Paris is in France"""}
													A__						:  Union[str, Any]		  =  text_classifier(snake_case							)
													self.assertEqual(
													    nested_simplify(snake_case							)			,    {"""label""": ANY(snake_case							), """score""": ANY(snake_case							)}			,    )
													self.assertTrue(outputs["""label"""] in model.config.idalabel.values()							)

													# This might be used a text pair, but tokenizer + pipe interaction
													# makes it hard to understand that it's not using the pair properly
													# https://github.com/huggingface/transformers/issues/17305
													# We disabled this usage instead as it was outputting wrong outputs.
													A__						:  Tuple		  =  [["""HuggingFace is in """, """Paris is in France"""]]
													with self.assertRaises(snake_case							):
																			text_classifier(snake_case							)

													# This used to be valid for doing text pairs
													# We're keeping it working because of backward compatibility
													A__						:  List[Any]		  =  text_classifier([[["""HuggingFace is in """, """Paris is in France"""]]]							)
													self.assertEqual(
													    nested_simplify(snake_case							)			,    [{"""label""": ANY(snake_case							), """score""": ANY(snake_case							)}]			,    )
													self.assertTrue(outputs[0]["""label"""] in model.config.idalabel.values()							)






