
'''simple docstring'''






import os
import re
import warnings
from shutil import copyfile
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer


if TYPE_CHECKING:
 from ...tokenization_utils_base import TextInput
from ...utils import logging


UpperCAmelCase_       :   List[str]   =       logging.get_logger(__name__)

UpperCAmelCase_       :   Tuple   =       {'vocab_file': 'spiece.model'}

UpperCAmelCase_       :   List[Any]   =       {
    'vocab_file': {
        't5-small': 'https://huggingface.co/t5-small/resolve/main/spiece.model',
        't5-base': 'https://huggingface.co/t5-base/resolve/main/spiece.model',
        't5-large': 'https://huggingface.co/t5-large/resolve/main/spiece.model',
        't5-3b': 'https://huggingface.co/t5-3b/resolve/main/spiece.model',
        't5-11b': 'https://huggingface.co/t5-11b/resolve/main/spiece.model',
    }
}


# TODO(PVP) - this should be removed in Transformers v5
UpperCAmelCase_       :   Dict   =       {
    't5-small': 512,
    't5-base': 512,
    't5-large': 512,
    't5-3b': 512,
    't5-11b': 512,
}

UpperCAmelCase_       :   Optional[int]   =       '▁'

class  UpperCAmelCase__						(       lowercase__ ):
    lowerCAmelCase_				    =		VOCAB_FILES_NAMES
    lowerCAmelCase_				    =		PRETRAINED_VOCAB_FILES_MAP
    lowerCAmelCase_				    =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    lowerCAmelCase_				    =		["""input_ids""", """attention_mask"""]



    def __init__(	self      :       Optional[Any],__A      :       List[Any],__A      :       List[str]="</s>",__A      :       str="<unk>",__A      :       List[Any]="<pad>",__A      :       Optional[Any]=1_0_0,__A      :       List[Any]=None,__A      :       Optional[Dict[str, Any]] = None,__A      :       Optional[Any]=True,**__A      :       Optional[Any],):
          # Add extra_ids to the special token list
          if extra_ids > 0 and additional_special_tokens is None:
                _lowerCamelCase				:   Any								=   [f'<extra_id_{i}>' for i in range(__lowercase     )]
          elif extra_ids > 0 and additional_special_tokens is not None:
                # Check that we have the right number of extra_id special tokens
                _lowerCamelCase				:   Dict								=   len(set(filter(lambda __A     : bool("extra_id" in str(__lowercase     )     ),__lowercase     )     )     )
                if extra_tokens != extra_ids:
                      raise ValueError(
                          f'Both extra_ids ({extra_ids}) and additional_special_tokens ({additional_special_tokens}) are'
                          " provided to T5Tokenizer. In this case the additional_special_tokens must include the extra_ids"
                          " tokens"     )
          if legacy:
                logger.warning_once(
                    f'You are using the legacy behaviour of the {self.__class__}. This means that tokens that come after special tokens will not be properly handled. We recommend you to'
                    " read the related pull request available at https://github.com/huggingface/transformers/pull/24565"     )

          _lowerCamelCase				:   int								=   legacy
          _lowerCamelCase				:   Any								=   {} if sp_model_kwargs is None else sp_model_kwargs

          super().__init__(
              eos_token=__lowercase,unk_token=__lowercase,pad_token=__lowercase,extra_ids=__lowercase,additional_special_tokens=__lowercase,sp_model_kwargs=self.sp_model_kwargs,legacy=__lowercase,**__lowercase,)

          _lowerCamelCase				:   str								=   vocab_file
          _lowerCamelCase				:   Optional[int]								=   extra_ids

          _lowerCamelCase				:   str								=   spm.SentencePieceProcessor(**self.sp_model_kwargs     )
          self.sp_model.Load(__lowercase     )



    @staticmethod
    def   lowerCamelCase_  (	__A      :       List[Any],__A      :       Union[str, Any],__A      :       str     ):
          if pretrained_model_name_or_path in TaTokenizer.max_model_input_sizes:
                _lowerCamelCase				:   int								=   TaTokenizer.max_model_input_sizes[pretrained_model_name_or_path]
                if init_max_model_length is not None and init_max_model_length != max_model_length:
                      return init_max_model_length
                elif init_max_model_length is None:
                      warnings.warn(
                          "This tokenizer was incorrectly instantiated with a model max length of"
                          f' {deprecated_max_model_length} which will be corrected in Transformers v5.\nFor now, this'
                          " behavior is kept to avoid breaking backwards compatibility when padding/encoding with"
                          " `truncation is True`.\n- Be aware that you SHOULD NOT rely on"
                          f' {pretrained_model_name_or_path} automatically truncating your input to'
                          f' {deprecated_max_model_length} when padding/encoding.\n- If you want to encode/pad to sequences'
                          f' longer than {deprecated_max_model_length} you can either instantiate this tokenizer with'
                          " `model_max_length` or pass `max_length` when encoding/padding.\n- To avoid this warning, please"
                          " instantiate this tokenizer with `model_max_length` set to your preferred value.",__lowercase,)

          return max_model_length



    @property
    def   lowerCamelCase_  (	self      :       List[str]     ):
          return self.sp_model.get_piece_size() + self._extra_ids



    def   lowerCamelCase_  (	self      :       Tuple     ):
          _lowerCamelCase				:   Optional[int]								=   {self.convert_ids_to_tokens(__lowercase     ): i for i in range(self.vocab_size     )}
          vocab.update(self.added_tokens_encoder     )
          return vocab



    def   lowerCamelCase_  (	self      :       Optional[Any],__A      :       List[int],__A      :       Optional[List[int]] = None,__A      :       bool = False     ):
          if already_has_special_tokens:
                return super().get_special_tokens_mask(
                    token_ids_a=__lowercase,token_ids_a=__lowercase,already_has_special_tokens=__lowercase     )

          # normal case: some special tokens
          if token_ids_a is None:
                return ([0] * len(__lowercase     )) + [1]
          return ([0] * len(__lowercase     )) + [1] + ([0] * len(__lowercase     )) + [1]



    def   lowerCamelCase_  (	self      :       Union[str, Any]     ):
          return list(
              set(filter(lambda __A     : bool(re.search(r"<extra_id_\d+>",__lowercase     )     ) is not None,self.additional_special_tokens     )     )     )



    def   lowerCamelCase_  (	self      :       str     ):
          return [self._convert_token_to_id(__lowercase     ) for token in self.get_sentinel_tokens()]



    def   lowerCamelCase_  (	self      :       Any,__A      :       List[int]     ):
          if len(__lowercase     ) > 0 and token_ids[-1] == self.eos_token_id:
                warnings.warn(
                    f'This sequence already has {self.eos_token}. In future versions this behavior may lead to duplicated'
                    " eos tokens being added."     )
                return token_ids
          else:
                return token_ids + [self.eos_token_id]



    def   lowerCamelCase_  (	self      :       Union[str, Any],__A      :       List[int],__A      :       Optional[List[int]] = None     ):
          _lowerCamelCase				:   Dict								=   [self.eos_token_id]

          if token_ids_a is None:
                return len(token_ids_a + eos     ) * [0]
          return len(token_ids_a + eos + token_ids_a + eos     ) * [0]



    def   lowerCamelCase_  (	self      :       str,__A      :       List[int],__A      :       Optional[List[int]] = None     ):
          _lowerCamelCase				:   List[Any]								=   self._add_eos_if_not_present(__lowercase     )
          if token_ids_a is None:
                return token_ids_a
          else:
                _lowerCamelCase				:   str								=   self._add_eos_if_not_present(__lowercase     )
                return token_ids_a + token_ids_a



    def __getstate__(	self      :       Tuple     ):
          _lowerCamelCase				:   Union[str, Any]								=   self.__dict__.copy()
          _lowerCamelCase				:   Optional[int]								=   None
          return state



    def __setstate__(	self      :       Dict,__A      :       List[str]     ):
          _lowerCamelCase				:   Any								=   d

          # for backward compatibility
          if not hasattr(self,"sp_model_kwargs"     ):
                _lowerCamelCase				:   Dict								=   {}

          _lowerCamelCase				:   Any								=   spm.SentencePieceProcessor(**self.sp_model_kwargs     )
          self.sp_model.Load(self.vocab_file     )



    def   lowerCamelCase_  (	self      :       List[str],__A      :       "TextInput",**__A      :       str     ):
          # Replace the SPIECE_UNDERLINE with a space to make sure SPIECE_UNDERLINE is only used at
          # the beginning of the text
          if not self.legacy:
                _lowerCamelCase				:   int								=   SPIECE_UNDERLINE + text.replace(__lowercase," "     )
          return super().tokenize(__lowercase,**__lowercase     )



    def   lowerCamelCase_  (	self      :       Union[str, Any],__A      :       Optional[Any],**__A      :       List[Any]     ):
          if not self.legacy:
                _lowerCamelCase				:   Any								=   text.startswith(__lowercase     )
                if is_first:
                      _lowerCamelCase				:   Union[str, Any]								=   text[1:]

          _lowerCamelCase				:   Optional[Any]								=   self.sp_model.encode(__lowercase,out_type=__lowercase     )

          if not self.legacy and not is_first and not text.startswith(" "     ) and tokens[0].startswith(__lowercase     ):
                _lowerCamelCase				:   str								=   ([tokens[0][1:]] if len(tokens[0]     ) > 1 else []) + tokens[1:]
          return tokens



    def   lowerCamelCase_  (	self      :       Tuple,__A      :       Tuple     ):
          if token.startswith("<extra_id_"     ):
                _lowerCamelCase				:   List[Any]								=   re.match(r"<extra_id_(\d+)>",__lowercase     )
                _lowerCamelCase				:   Dict								=   int(match.group(1     )     )
                return self.vocab_size - num - 1
          return self.sp_model.piece_to_id(__lowercase     )



    def   lowerCamelCase_  (	self      :       Optional[Any],__A      :       Optional[int]     ):
          if index < self.sp_model.get_piece_size():
                _lowerCamelCase				:   Optional[Any]								=   self.sp_model.IdToPiece(__lowercase     )
          else:
                _lowerCamelCase				:   List[str]								=   f'<extra_id_{self.vocab_size - 1 - index}>'
          return token



    def   lowerCamelCase_  (	self      :       Optional[int],__A      :       Tuple     ):
          _lowerCamelCase				:   Tuple								=   []
          _lowerCamelCase				:   List[Any]								=   ""
          _lowerCamelCase				:   List[str]								=   False
          for token in tokens:
                # make sure that special tokens are not decoded using sentencepiece model
                if token in self.all_special_tokens:
                      if not prev_is_special:
                            out_string += " "
                      out_string += self.sp_model.decode(__lowercase     ) + token
                      _lowerCamelCase				:   List[str]								=   True
                      _lowerCamelCase				:   str								=   []
                else:
                      current_sub_tokens.append(__lowercase     )
                      _lowerCamelCase				:   Tuple								=   False
          out_string += self.sp_model.decode(__lowercase     )
          return out_string.strip()



    def   lowerCamelCase_  (	self      :       List[str],__A      :       str,__A      :       Optional[str] = None     ):
          if not os.path.isdir(__lowercase     ):
                logger.error(f'Vocabulary path ({save_directory}) should be a directory'     )
                return
          _lowerCamelCase				:   List[Any]								=   os.path.join(
              __lowercase,(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]     )

          if os.path.abspath(self.vocab_file     ) != os.path.abspath(__lowercase     ) and os.path.isfile(self.vocab_file     ):
                copyfile(self.vocab_file,__lowercase     )
          elif not os.path.isfile(self.vocab_file     ):
                with open(__lowercase,"wb"     ) as fi:
                      _lowerCamelCase				:   Optional[Any]								=   self.sp_model.serialized_model_proto()
                      fi.write(__lowercase     )

          return (out_vocab_file,)

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


UpperCAmelCase_       :   Any   =       re.compile(R'\s+')







def 		A_       (	_lowerCAmelCase   :    List[str]							):



      """simple docstring"""




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







def 		A_       (	_lowerCAmelCase   :    Union[str, Any]							):



      """simple docstring"""




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







def 		A_       (	_lowerCAmelCase   :    int							):



      """simple docstring"""




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







def 		A_       (	_lowerCAmelCase   :    str				,	_lowerCAmelCase   :    Tuple							):



      """simple docstring"""




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







def 		A_       (	_lowerCAmelCase   :    Optional[int]				,	_lowerCAmelCase   :    Dict=5							):



      """simple docstring"""




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







def 		A_       (	_lowerCAmelCase   :    List[str]				,	_lowerCAmelCase   :    Optional[Any]=5				,	_lowerCAmelCase   :    List[Any]=0.0_5							):



      """simple docstring"""




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



      """simple docstring"""




      _lowerCamelCase				:   str								=   ["def ", "class ", "for ", "while "]
      _lowerCamelCase				:   List[str]								=   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 		A_       (	_lowerCAmelCase   :    Dict				,	_lowerCAmelCase   :    int=4							):



      """simple docstring"""




      _lowerCamelCase				:   List[Any]								=   example["content"].splitlines()
      _lowerCamelCase				:   Union[str, Any]								=   0
      for line in lines:
            counter += line.lower().count("="							)
            if counter > minimum:
                  return {"has_few_assignments": False}
      return {"has_few_assignments": True}







def 		A_       (	_lowerCAmelCase   :    List[str]							):



      """simple docstring"""




      _lowerCamelCase				:   List[Any]								=   tokenizer(example["content"]				,	truncation=_lowerCAmelCase							)["input_ids"]
      _lowerCamelCase				:   str								=   len(example["content"]							) / len(_lowerCAmelCase							)
      return {"ratio": ratio}







def 		A_       (	_lowerCAmelCase   :    Optional[Any]							):



      """simple docstring"""




      _lowerCamelCase				:   Optional[Any]								=   {}
      results.update(get_hash(_lowerCAmelCase							)							)
      results.update(line_stats(_lowerCAmelCase							)							)
      results.update(alpha_stats(_lowerCAmelCase							)							)
      results.update(char_token_ratio(_lowerCAmelCase							)							)
      results.update(is_autogenerated(_lowerCAmelCase							)							)
      results.update(is_config_or_test(_lowerCAmelCase							)							)
      results.update(has_no_keywords(_lowerCAmelCase							)							)
      results.update(has_few_assignments(_lowerCAmelCase							)							)
      return results







def 		A_       (	_lowerCAmelCase   :    Union[str, Any]				,	_lowerCAmelCase   :    Optional[Any]				,	_lowerCAmelCase   :    str							):



      """simple docstring"""




      if not check_uniques(_lowerCAmelCase				,	_lowerCAmelCase							):
            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 		A_       (	_lowerCAmelCase   :    int							):



      """simple docstring"""




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


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

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

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

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

# Deduplicate data and apply heuristics
UpperCAmelCase_       :   Optional[int]   =       time.time()
UpperCAmelCase_       :   int   =       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_       :   List[str]   =       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_       :   List[str]   =       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_       :   List[Any]   =       output_dir / 'data'
data_dir.mkdir(exist_ok=True)

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