



'''simple docstring'''





import json
import multiprocessing as mp
import re
from collections import defaultdict
from functools import partial
from typing import Dict, List, Optional, Set, Tuple, Type

from datasets import Dataset
from datasketch import MinHash, MinHashLSH
from dpu_utils.utils.iterators import ThreadedIterator
from tqdm import tqdm


SCREAMING_SNAKE_CASE__      					=     re.compile("[^A-Za-z_0-9]")
# parameters used in DuplicationIndex
SCREAMING_SNAKE_CASE__      					=     10
SCREAMING_SNAKE_CASE__      					=     256




def        lowerCamelCase  (				_snake_case  :						List[str]       ):





    '''simple docstring'''


    if len(_snake_case       ) < MIN_NUM_TOKENS:
        return None
    lowercase__		       =						MinHash(num_perm=_snake_case       )
    for token in set(_snake_case       ):
        min_hash.update(token.encode()       )
    return min_hash




def        lowerCamelCase  (				_snake_case  :						str       ):





    '''simple docstring'''


    return {t for t in NON_ALPHA.split(_snake_case       ) if len(t.strip()       ) > 0}


class 				snake_case :



       def __init__(				self     ,*,
           UpperCAmelCase_ = 0.85     ,)  ->							Optional[int]:
           lowercase__		       =						duplication_jaccard_threshold
           lowercase__		       =						NUM_PERM
           lowercase__		       =						MinHashLSH(threshold=self._duplication_jaccard_threshold     ,num_perm=self._num_perm      )

           lowercase__		       =						defaultdict(UpperCAmelCase_      )



       def 							_a      (				self     ,UpperCAmelCase_     ,UpperCAmelCase_      )  ->							None:
           lowercase__		       =						self._index.query(UpperCAmelCase_      )
           if code_key in self._index.keys:
               print(F'''Duplicate key {code_key}'''      )
               return

           self._index.insert(UpperCAmelCase_     ,UpperCAmelCase_      )
           if len(UpperCAmelCase_      ) > 0:
               for base_duplicate in close_duplicates:
                   if base_duplicate in self._duplicate_clusters:
                       self._duplicate_clusters[base_duplicate].add(UpperCAmelCase_      )
                       break
               else:
                   self._duplicate_clusters[close_duplicates[0]].add(UpperCAmelCase_      )



       def 							_a      (				self      )  ->							List[List[Dict]]:
           lowercase__		       =						[]
           for base, duplicates in self._duplicate_clusters.items():
               lowercase__		       =						[base] + list(UpperCAmelCase_      )
               # reformat the cluster to be a list of dict
               lowercase__		       =						[{"base_index": el[0], "repo_name": el[1], "path": el[2]} for el in cluster]
               duplicate_clusters.append(UpperCAmelCase_      )
           return duplicate_clusters







       def 							_a      (				self     ,UpperCAmelCase_      )  ->							None:
           lowercase__		       =						self.get_duplicate_clusters()
           with open(UpperCAmelCase_     ,"w"      ) as f:
               json.dump(UpperCAmelCase_     ,UpperCAmelCase_      )





def        lowerCamelCase  (				_snake_case  :						Optional[int]       ):





    '''simple docstring'''


    lowercase__       ,       lowercase__		       =						element
    lowercase__		       =						get_min_hash([t for t in NON_ALPHA.split(data["content"]       ) if len(t.strip()       ) > 0]       )
    if min_hash is not None:
        return (index, data["repo_name"], data["path"]), min_hash




def        lowerCamelCase  (				_snake_case  :						Type[Dataset]       ):





    '''simple docstring'''


    with mp.Pool() as pool:
        for data in pool.imap_unordered(
            _compute_min_hash     ,ThreadedIterator(_snake_case     ,max_queue_size=10_000       )     ,chunksize=100     ,):
            if data is not None:
                yield data




def        lowerCamelCase  (				_snake_case  :						Type[Dataset]     ,_snake_case  :						float       ):





    '''simple docstring'''


    lowercase__		       =						DuplicationIndex(duplication_jaccard_threshold=_snake_case       )

    for filename, min_hash in tqdm(ThreadedIterator(minhash_iter(enumerate(_snake_case       )       )     ,max_queue_size=100       )       ):
        di.add(_snake_case     ,_snake_case       )

    # Returns a List[Cluster] where Cluster is List[str] with the filenames.
    return di.get_duplicate_clusters()




def        lowerCamelCase  (				_snake_case  :						str     ,_snake_case  :						str       ):





    '''simple docstring'''


    lowercase__		       =						get_tokens(_snake_case       )
    lowercase__		       =						get_tokens(_snake_case       )
    return len(tokensa & tokensa       ) / len(tokensa | tokensa       )


SCREAMING_SNAKE_CASE__      					=     None




def        lowerCamelCase  (				_snake_case  :						Optional[int]     ,_snake_case  :						Tuple       ):





    '''simple docstring'''


    lowercase__		       =						[]
    for elementa in cluster:
        lowercase__		       =						_shared_dataset[elementa["base_index"]]["content"]
        for elementa in extremes:
            lowercase__		       =						_shared_dataset[elementa["base_index"]]["content"]
            if jaccard_similarity(_snake_case     ,_snake_case       ) >= jaccard_threshold:
                elementa["copies"] += 1
                break
        else:
            lowercase__		       =						1
            extremes.append(_snake_case       )
    return extremes




def        lowerCamelCase  (				_snake_case  :						Optional[Any]     ,_snake_case  :						int     ,_snake_case  :						Tuple       ):





    '''simple docstring'''


    global _shared_dataset
    lowercase__		       =						dataset
    lowercase__		       =						[]
    lowercase__		       =						partial(_find_cluster_extremes_shared     ,jaccard_threshold=_snake_case       )
    with mp.Pool() as pool:
        for extremes in tqdm(
            pool.imap_unordered(
                _snake_case     ,_snake_case     ,)     ,total=len(_snake_case       )     ,):
            extremes_list.append(_snake_case       )
    return extremes_list




def        lowerCamelCase  (				_snake_case  :						Type[Dataset]     ,_snake_case  :						float = 0.85       ):





    '''simple docstring'''


    lowercase__		       =						make_duplicate_clusters(_snake_case     ,_snake_case       )
    lowercase__		       =						{x["base_index"] for cluster in duplicate_clusters for x in cluster}
    lowercase__		       =						{}
    lowercase__		       =						find_extremes(_snake_case     ,_snake_case     ,_snake_case       )
    for extremes in extremes_clusters:
        for element in extremes:
            lowercase__		       =						element
    lowercase__		       =						duplicate_indices - set(extreme_dict.keys()       )
    lowercase__		       =						dataset.filter(lambda _snake_case     ,_snake_case       : idx not in remove_indices     ,with_indices=_snake_case       )

    # update duplicate_clusters
    for cluster in duplicate_clusters:
        for element in cluster:
            lowercase__		       =						element["base_index"] in extreme_dict
            if element["is_extreme"]:
                lowercase__		       =						extreme_dict[element["base_index"]]["copies"]

    print(f'''Original dataset size: {len(_snake_case       )}'''       )
    print(f'''Number of duplicate clusters: {len(_snake_case       )}'''       )
    print(f'''Files in duplicate cluster: {len(_snake_case       )}'''       )
    print(f'''Unique files in duplicate cluster: {len(_snake_case       )}'''       )
    print(f'''Filtered dataset size: {len(_snake_case       )}'''       )

    return ds_filter, duplicate_clusters









'''simple docstring'''





import numpy as np
import torch
from torch.utils.data import Dataset

from utils import logger


class 				snake_case (UpperCamelCase   ):



       def __init__(				self     ,UpperCAmelCase_     ,UpperCAmelCase_      )  ->							int:
           lowercase__		       =						params

           lowercase__		       =						np.array(UpperCAmelCase_      )
           lowercase__		       =						np.array([len(UpperCAmelCase_      ) for t in data]      )

           self.check()
           self.remove_long_sequences()
           self.remove_empty_sequences()
           self.remove_unknown_sequences()
           self.check()
           self.print_statistics()



       def __getitem__(				self     ,UpperCAmelCase_      )  ->							Optional[int]:
           return (self.token_ids[index], self.lengths[index])



       def __len__(				self      )  ->							Union[str, Any]:
           return len(self.lengths      )



       def 							_a      (				self      )  ->							int:
           assert len(self.token_ids      ) == len(self.lengths      )
           assert all(self.lengths[i] == len(self.token_ids[i]      ) for i in range(len(self.lengths      )      )      )



       def 							_a      (				self      )  ->							Optional[int]:
           lowercase__		       =						self.params.max_model_input_size
           lowercase__		       =						self.lengths > max_len
           logger.info(F'''Splitting {sum(UpperCAmelCase_      )} too long sequences.'''      )

           def divide_chunks(UpperCAmelCase_     ,UpperCAmelCase_      ):
               return [l[i : i + n] for i in range(0     ,len(UpperCAmelCase_      )     ,UpperCAmelCase_      )]

           lowercase__		       =						[]
           lowercase__		       =						[]
           if self.params.mlm:
               lowercase__       ,       lowercase__		       =						self.params.special_tok_ids["cls_token"], self.params.special_tok_ids["sep_token"]
           else:
               lowercase__       ,       lowercase__		       =						self.params.special_tok_ids["bos_token"], self.params.special_tok_ids["eos_token"]

           for seq_, len_ in zip(self.token_ids     ,self.lengths      ):
               assert (seq_[0] == cls_id) and (seq_[-1] == sep_id), seq_
               if len_ <= max_len:
                   new_tok_ids.append(seq_      )
                   new_lengths.append(len_      )
               else:
                   lowercase__		       =						[]
                   for sub_s in divide_chunks(seq_     ,max_len - 2      ):
                       if sub_s[0] != cls_id:
                           lowercase__		       =						np.insert(UpperCAmelCase_     ,0     ,UpperCAmelCase_      )
                       if sub_s[-1] != sep_id:
                           lowercase__		       =						np.insert(UpperCAmelCase_     ,len(UpperCAmelCase_      )     ,UpperCAmelCase_      )
                       assert len(UpperCAmelCase_      ) <= max_len
                       assert (sub_s[0] == cls_id) and (sub_s[-1] == sep_id), sub_s
                       sub_seqs.append(UpperCAmelCase_      )

                   new_tok_ids.extend(UpperCAmelCase_      )
                   new_lengths.extend([len(UpperCAmelCase_      ) for l in sub_seqs]      )

           lowercase__		       =						np.array(UpperCAmelCase_      )
           lowercase__		       =						np.array(UpperCAmelCase_      )



       def 							_a      (				self      )  ->							Any:
           lowercase__		       =						len(self      )
           lowercase__		       =						self.lengths > 11
           lowercase__		       =						self.token_ids[indices]
           lowercase__		       =						self.lengths[indices]
           lowercase__		       =						len(self      )
           logger.info(F'''Remove {init_size - new_size} too short (<=11 tokens) sequences.'''      )



       def 							_a      (				self      )  ->							List[Any]:
           if "unk_token" not in self.params.special_tok_ids:
               return
           else:
               lowercase__		       =						self.params.special_tok_ids["unk_token"]
           lowercase__		       =						len(self      )
           lowercase__		       =						np.array([np.count_nonzero(a == unk_token_id      ) for a in self.token_ids]      )
           lowercase__		       =						(unk_occs / self.lengths) < 0.5
           lowercase__		       =						self.token_ids[indices]
           lowercase__		       =						self.lengths[indices]
           lowercase__		       =						len(self      )
           logger.info(F'''Remove {init_size - new_size} sequences with a high level of unknown tokens (50%).'''      )



       def 							_a      (				self      )  ->							Optional[int]:
           if not self.params.is_master:
               return
           logger.info(F'''{len(self      )} sequences'''      )
           # data_len = sum(self.lengths)
           # nb_unique_tokens = len(Counter(list(chain(*self.token_ids))))
           # logger.info(f'{data_len} tokens ({nb_unique_tokens} unique)')

           # unk_idx = self.params.special_tok_ids['unk_token']
           # nb_unknown = sum([(t==unk_idx).sum() for t in self.token_ids])
           # logger.info(f'{nb_unknown} unknown tokens (covering {100*nb_unknown/data_len:.2f}% of the data)')







       def 							_a      (				self     ,UpperCAmelCase_      )  ->							List[str]:
           lowercase__		       =						[t[0] for t in batch]
           lowercase__		       =						[t[1] for t in batch]
           assert len(UpperCAmelCase_      ) == len(UpperCAmelCase_      )

           # Max for paddings
           lowercase__		       =						max(UpperCAmelCase_      )

           # Pad token ids
           if self.params.mlm:
               lowercase__		       =						self.params.special_tok_ids["pad_token"]
           else:
               lowercase__		       =						self.params.special_tok_ids["unk_token"]
           lowercase__		       =						[list(t.astype(UpperCAmelCase_      )      ) + [pad_idx] * (max_seq_len_ - len(UpperCAmelCase_      )) for t in token_ids]
           assert len(tk_      ) == len(UpperCAmelCase_      )
           assert all(len(UpperCAmelCase_      ) == max_seq_len_ for t in tk_      )

           lowercase__		       =						torch.tensor(tk_      )  # (bs, max_seq_len_)
           lowercase__		       =						torch.tensor(UpperCAmelCase_      )  # (bs)
           return tk_t, lg_t




