






'''simple docstring'''



# Lint as: python3
import os
import re
import urllib.parse
from pathlib import Path
from typing import Callable, List, Optional, Union
from zipfile import ZipFile

from ..utils.file_utils import cached_path, hf_github_url
from ..utils.logging import get_logger
from ..utils.version import Version


__UpperCamelCase   :							Optional[Any]         =   get_logger(__name__)
class 				__SCREAMING_SNAKE_CASE							:
      __a	       ="dummy_data"
      __a	       ="datasets"
      __a	       =False




      def __init__(	self			, lowerCamelCase			, lowerCamelCase			, lowerCamelCase			, lowerCamelCase = None			, lowerCamelCase = False			, lowerCamelCase = True			, lowerCamelCase = None			, )				->Union[str, Any]:



         '''simple docstring'''
         __a					  =     0
         __a					  =     dataset_name
         __a					  =     cache_dir
         __a					  =     use_local_dummy_data
         __a					  =     config
         # download_callbacks take a single url as input
         __a					  =     download_callbacks or []
         # if False, it doesn't load existing files and it returns the paths of the dummy files relative
         # to the dummy_data zip file root
         __a					  =     load_existing_dummy_data

         # TODO(PVP, QL) might need to make this more general
         __a					  =     str(lowerCamelCase   )
         # to be downloaded
         __a					  =     None
         __a					  =     None




      @property
      def 			__UpperCamelCase				(	self   )				->List[Any]:



         '''simple docstring'''
         if self._dummy_file is None:
            __a					  =     self.download_dummy_data()
         return self._dummy_file




      @property
      def 			__UpperCamelCase				(	self   )				->Any:



         '''simple docstring'''
         if self.config is not None:
            # structure is dummy / config_name / version_name
            return os.path.join('dummy'			, self.config.name			, self.version_name   )
         # structure is dummy / version_name
         return os.path.join('dummy'			, self.version_name   )




      @property
      def 			__UpperCamelCase				(	self   )				->Any:



         '''simple docstring'''
         return os.path.join(self.dummy_data_folder			, 'dummy_data.zip'   )




      def 			__UpperCamelCase				(	self   )				->List[str]:



         '''simple docstring'''
         __a					  =     (
             self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
         )

         __a					  =     cached_path(
             lowerCamelCase			, cache_dir=self.cache_dir			, extract_compressed_file=lowerCamelCase			, force_extract=lowerCamelCase   )

         return os.path.join(lowerCamelCase			, self.dummy_file_name   )




      @property
      def 			__UpperCamelCase				(	self   )				->Dict:



         '''simple docstring'''
         return os.path.join(self.datasets_scripts_dir			, self.dataset_name			, self.dummy_zip_file   )




      @property
      def 			__UpperCamelCase				(	self   )				->Dict:



         '''simple docstring'''
         if self._bucket_url is None:
            __a					  =     hf_github_url(self.dataset_name			, self.dummy_zip_file.replace(os.sep			, '/'   )   )
         return self._bucket_url




      @property
      def 			__UpperCamelCase				(	self   )				->int:



         '''simple docstring'''
         # return full path if its a dir
         if os.path.isdir(self.dummy_file   ):
            return self.dummy_file
         # else cut off path to file -> example `xsum`.
         return "/".join(self.dummy_file.replace(os.sep			, '/'   ).split('/'   )[:-1]   )




      def 			__UpperCamelCase				(	self			, lowerCamelCase			, *lowerCamelCase   )				->Any:



         '''simple docstring'''
         if self.load_existing_dummy_data:
            # dummy data is downloaded and tested
            __a					  =     self.dummy_file
         else:
            # dummy data cannot be downloaded and only the path to dummy file is returned
            __a					  =     self.dummy_file_name

         # special case when data_url is a dict
         if isinstance(lowerCamelCase			, lowerCamelCase   ):
            return self.create_dummy_data_dict(lowerCamelCase			, lowerCamelCase   )
         elif isinstance(lowerCamelCase			, (list, tuple)   ):
            return self.create_dummy_data_list(lowerCamelCase			, lowerCamelCase   )
         else:
            return self.create_dummy_data_single(lowerCamelCase			, lowerCamelCase   )




      def 			__UpperCamelCase				(	self			, lowerCamelCase			, *lowerCamelCase   )				->Optional[Any]:



         '''simple docstring'''
         return self.download_and_extract(lowerCamelCase   )




      def 			__UpperCamelCase				(	self			, lowerCamelCase			, lowerCamelCase   )				->Tuple:



         '''simple docstring'''
         return self.download_and_extract(lowerCamelCase   )




      def 			__UpperCamelCase				(	self			, lowerCamelCase			, *lowerCamelCase			, **lowerCamelCase   )				->str:



         '''simple docstring'''
         return path




      def 			__UpperCamelCase				(	self   )				->List[Any]:



         '''simple docstring'''
         return {}




      def 			__UpperCamelCase				(	self			, lowerCamelCase			, lowerCamelCase   )				->Tuple:



         '''simple docstring'''
         __a					  =     {}
         for key, single_urls in data_url.items():
            for download_callback in self.download_callbacks:
               if isinstance(lowerCamelCase			, lowerCamelCase   ):
                  for single_url in single_urls:
                     download_callback(lowerCamelCase   )
               else:
                  __a					  =     single_urls
                  download_callback(lowerCamelCase   )
            # we force the name of each key to be the last file / folder name of the url path
            # if the url has arguments, we need to encode them with urllib.parse.quote_plus
            if isinstance(lowerCamelCase			, lowerCamelCase   ):
               __a					  =     [os.path.join(lowerCamelCase			, urllib.parse.quote_plus(Path(lowerCamelCase   ).name   )   ) for x in single_urls]
            else:
               __a					  =     single_urls
               __a					  =     os.path.join(lowerCamelCase			, urllib.parse.quote_plus(Path(lowerCamelCase   ).name   )   )
            __a					  =     value

         # make sure that values are unique
         if all(isinstance(lowerCamelCase			, lowerCamelCase   ) for i in dummy_data_dict.values()   ) and len(set(dummy_data_dict.values()   )   ) < len(
             dummy_data_dict.values()   ):
            # append key to value to make its name unique
            __a					  =     {key: value + key for key, value in dummy_data_dict.items()}

         return dummy_data_dict




      def 			__UpperCamelCase				(	self			, lowerCamelCase			, lowerCamelCase   )				->Union[str, Any]:



         '''simple docstring'''
         __a					  =     []
         # trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
         __a					  =     all(bool(re.findall('[0-9]{3,}-of-[0-9]{3,}'			, lowerCamelCase   )   ) for url in data_url   )
         __a					  =     all(
             url.startswith('https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/pubmed'   ) for url in data_url   )
         if data_url and (is_tf_records or is_pubmed_records):
            __a					  =     [data_url[0]] * len(lowerCamelCase   )
         for single_url in data_url:
            for download_callback in self.download_callbacks:
               download_callback(lowerCamelCase   )
            # we force the name of each key to be the last file / folder name of the url path
            # if the url has arguments, we need to encode them with urllib.parse.quote_plus
            __a					  =     os.path.join(lowerCamelCase			, urllib.parse.quote_plus(single_url.split('/'   )[-1]   )   )
            dummy_data_list.append(lowerCamelCase   )
         return dummy_data_list




      def 			__UpperCamelCase				(	self			, lowerCamelCase			, lowerCamelCase   )				->List[Any]:



         '''simple docstring'''
         for download_callback in self.download_callbacks:
            download_callback(lowerCamelCase   )
         # we force the name of each key to be the last file / folder name of the url path
         # if the url has arguments, we need to encode them with urllib.parse.quote_plus
         __a					  =     os.path.join(lowerCamelCase			, urllib.parse.quote_plus(data_url.split('/'   )[-1]   )   )
         if os.path.exists(lowerCamelCase   ) or not self.load_existing_dummy_data:
            return value
         else:
            # Backward compatibility, maybe deprecate at one point.
            # For many datasets with single url calls to dl_manager.download_and_extract,
            # the dummy_data.zip file is actually the zipped downloaded file
            # while now we expected the dummy_data.zip file to be a directory containing
            # the downloaded file.
            return path_to_dummy_data




      def 			__UpperCamelCase				(	self   )				->Any:



         '''simple docstring'''
         pass




      def 			__UpperCamelCase				(	self   )				->Tuple:



         '''simple docstring'''
         pass




      def 			__UpperCamelCase				(	self			, lowerCamelCase   )				->str:



         '''simple docstring'''
         def _iter_archive_members(lowerCamelCase   ):
            # this preserves the order of the members inside the ZIP archive
            __a					  =     Path(self.dummy_file   ).parent
            __a					  =     path.relative_to(lowerCamelCase   )
            with ZipFile(self.local_path_to_dummy_data   ) as zip_file:
               __a					  =     zip_file.namelist()
            for member in members:
               if member.startswith(relative_path.as_posix()   ):
                  yield dummy_parent_path.joinpath(lowerCamelCase   )

         __a					  =     Path(lowerCamelCase   )
         __a					  =     _iter_archive_members(lowerCamelCase   ) if self.use_local_dummy_data else path.rglob('*'   )
         for file_path in file_paths:
            if file_path.is_file() and not file_path.name.startswith(('.', '__')   ):
               yield file_path.relative_to(lowerCamelCase   ).as_posix(), file_path.open('rb'   )






      def 			__UpperCamelCase				(	self			, lowerCamelCase   )				->List[str]:



         '''simple docstring'''
         if not isinstance(lowerCamelCase			, lowerCamelCase   ):
            __a					  =     [paths]
         for path in paths:
            if os.path.isfile(lowerCamelCase   ):
               if os.path.basename(lowerCamelCase   ).startswith(('.', '__')   ):
                  return
               yield path
            else:
               for dirpath, dirnames, filenames in os.walk(lowerCamelCase   ):
                  if os.path.basename(lowerCamelCase   ).startswith(('.', '__')   ):
                     continue
                  dirnames.sort()
                  for filename in sorted(lowerCamelCase   ):
                     if filename.startswith(('.', '__')   ):
                        continue
                     yield os.path.join(lowerCamelCase			, lowerCamelCase   )







'''simple docstring'''



import os
import re
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


__UpperCamelCase   :							int         =   logging.get_logger(__name__)

__UpperCamelCase   :							List[str]         =   {
    """vocab_file""": """vocab.txt""",
    """merges_file""": """bpe.codes""",
}

__UpperCamelCase   :							List[str]         =   {
    """vocab_file""": {
        """vinai/phobert-base""": """https://huggingface.co/vinai/phobert-base/resolve/main/vocab.txt""",
        """vinai/phobert-large""": """https://huggingface.co/vinai/phobert-large/resolve/main/vocab.txt""",
    },
    """merges_file""": {
        """vinai/phobert-base""": """https://huggingface.co/vinai/phobert-base/resolve/main/bpe.codes""",
        """vinai/phobert-large""": """https://huggingface.co/vinai/phobert-large/resolve/main/bpe.codes""",
    },
}

__UpperCamelCase   :							Union[str, Any]         =   {
    """vinai/phobert-base""": 256,
    """vinai/phobert-large""": 256,
}



def  __UpperCAmelCase						(					SCREAMING_SNAKE_CASE__:						int     )	->      List[str]:

   """simple docstring"""
   __a					  =     set()
   __a					  =     word[0]
   for char in word[1:]:
      pairs.add((prev_char, char)     )
      __a					  =     char

   __a					  =     set(SCREAMING_SNAKE_CASE__     )
   return pairs



class 				__SCREAMING_SNAKE_CASE							(				_lowerCAmelCase  ):
      __a	       =VOCAB_FILES_NAMES
      __a	       =PRETRAINED_VOCAB_FILES_MAP
      __a	       =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES




      def __init__(	self			, lowerCamelCase			, lowerCamelCase			, lowerCamelCase="<s>"			, lowerCamelCase="</s>"			, lowerCamelCase="</s>"			, lowerCamelCase="<s>"			, lowerCamelCase="<unk>"			, lowerCamelCase="<pad>"			, lowerCamelCase="<mask>"			, **lowerCamelCase			, )				->str:



         '''simple docstring'''
         super().__init__(
             bos_token=lowerCamelCase			, eos_token=lowerCamelCase			, unk_token=lowerCamelCase			, sep_token=lowerCamelCase			, cls_token=lowerCamelCase			, pad_token=lowerCamelCase			, mask_token=lowerCamelCase			, **lowerCamelCase			, )

         __a					  =     vocab_file
         __a					  =     merges_file

         __a					  =     {}
         __a					  =     0
         __a					  =     1
         __a					  =     2
         __a					  =     3

         self.add_from_file(lowerCamelCase   )

         __a					  =     {v: k for k, v in self.encoder.items()}

         with open(lowerCamelCase			, encoding='utf-8'   ) as merges_handle:
            __a					  =     merges_handle.read().split('\n'   )[:-1]
         __a					  =     [tuple(merge.split()[:-1]   ) for merge in merges]
         __a					  =     dict(zip(lowerCamelCase			, range(len(lowerCamelCase   )   )   )   )
         __a					  =     {}




      def 			__UpperCamelCase				(	self			, lowerCamelCase			, lowerCamelCase = None   )				->List[int]:



         '''simple docstring'''

         if token_ids_a is None:
            return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
         __a					  =     [self.cls_token_id]
         __a					  =     [self.sep_token_id]
         return cls + token_ids_a + sep + sep + token_ids_a + sep




      def 			__UpperCamelCase				(	self			, lowerCamelCase			, lowerCamelCase = None			, lowerCamelCase = False   )				->List[int]:



         '''simple docstring'''

         if already_has_special_tokens:
            return super().get_special_tokens_mask(
                token_ids_a=lowerCamelCase			, token_ids_a=lowerCamelCase			, already_has_special_tokens=lowerCamelCase   )

         if token_ids_a is None:
            return [1] + ([0] * len(lowerCamelCase   )) + [1]
         return [1] + ([0] * len(lowerCamelCase   )) + [1, 1] + ([0] * len(lowerCamelCase   )) + [1]




      def 			__UpperCamelCase				(	self			, lowerCamelCase			, lowerCamelCase = None   )				->List[int]:



         '''simple docstring'''
         __a					  =     [self.sep_token_id]
         __a					  =     [self.cls_token_id]

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




      @property
      def 			__UpperCamelCase				(	self   )				->Union[str, Any]:



         '''simple docstring'''
         return len(self.encoder   )




      def 			__UpperCamelCase				(	self   )				->Optional[int]:



         '''simple docstring'''
         return dict(self.encoder			, **self.added_tokens_encoder   )




      def 			__UpperCamelCase				(	self			, lowerCamelCase   )				->List[str]:



         '''simple docstring'''
         if token in self.cache:
            return self.cache[token]
         __a					  =     tuple(lowerCamelCase   )
         __a					  =     tuple(list(word[:-1]   ) + [word[-1] + '</w>']   )
         __a					  =     get_pairs(lowerCamelCase   )

         if not pairs:
            return token

         while True:
            __a					  =     min(lowerCamelCase			, key=lambda lowerCamelCase   : self.bpe_ranks.get(lowerCamelCase			, float('inf'   )   )   )
            if bigram not in self.bpe_ranks:
               break
            __a  ,      __a					  =     bigram
            __a					  =     []
            __a					  =     0
            while i < len(lowerCamelCase   ):
               try:
                  __a					  =     word.index(lowerCamelCase			, lowerCamelCase   )
               except ValueError:
                  new_word.extend(word[i:]   )
                  break
               else:
                  new_word.extend(word[i:j]   )
                  __a					  =     j

               if word[i] == first and i < len(lowerCamelCase   ) - 1 and word[i + 1] == second:
                  new_word.append(first + second   )
                  i += 2
               else:
                  new_word.append(word[i]   )
                  i += 1
            __a					  =     tuple(lowerCamelCase   )
            __a					  =     new_word
            if len(lowerCamelCase   ) == 1:
               break
            else:
               __a					  =     get_pairs(lowerCamelCase   )
         __a					  =     '@@ '.join(lowerCamelCase   )
         __a					  =     word[:-4]
         __a					  =     word
         return word




      def 			__UpperCamelCase				(	self			, lowerCamelCase   )				->Dict:



         '''simple docstring'''
         __a					  =     []

         __a					  =     re.findall(r'\S+\n?'			, lowerCamelCase   )

         for token in words:
            split_tokens.extend(list(self.bpe(lowerCamelCase   ).split(' '   )   )   )
         return split_tokens




      def 			__UpperCamelCase				(	self			, lowerCamelCase   )				->List[Any]:



         '''simple docstring'''
         return self.encoder.get(lowerCamelCase			, self.encoder.get(self.unk_token   )   )




      def 			__UpperCamelCase				(	self			, lowerCamelCase   )				->List[Any]:



         '''simple docstring'''
         return self.decoder.get(lowerCamelCase			, self.unk_token   )




      def 			__UpperCamelCase				(	self			, lowerCamelCase   )				->Dict:



         '''simple docstring'''
         __a					  =     ' '.join(lowerCamelCase   ).replace('@@ '			, ''   ).strip()
         return out_string




      def 			__UpperCamelCase				(	self			, lowerCamelCase			, lowerCamelCase = None   )				->Tuple[str]:



         '''simple docstring'''
         if not os.path.isdir(lowerCamelCase   ):
            logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""   )
            return
         __a					  =     os.path.join(
             lowerCamelCase			, (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']   )
         __a					  =     os.path.join(
             lowerCamelCase			, (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']   )

         if os.path.abspath(self.vocab_file   ) != os.path.abspath(lowerCamelCase   ):
            copyfile(self.vocab_file			, lowerCamelCase   )

         if os.path.abspath(self.merges_file   ) != os.path.abspath(lowerCamelCase   ):
            copyfile(self.merges_file			, lowerCamelCase   )

         return out_vocab_file, out_merge_file






      def 			__UpperCamelCase				(	self			, lowerCamelCase   )				->str:



         '''simple docstring'''
         if isinstance(lowerCamelCase			, lowerCamelCase   ):
            try:
               with open(lowerCamelCase			, 'r'			, encoding='utf-8'   ) as fd:
                  self.add_from_file(lowerCamelCase   )
            except FileNotFoundError as fnfe:
               raise fnfe
            except UnicodeError:
               raise Exception(F"""Incorrect encoding detected in {f}, please rebuild the dataset"""   )
            return

         __a					  =     f.readlines()
         for lineTmp in lines:
            __a					  =     lineTmp.strip()
            __a					  =     line.rfind(' '   )
            if idx == -1:
               raise ValueError('Incorrect dictionary format, expected \'<token> <cnt>\''   )
            __a					  =     line[:idx]
            __a					  =     len(self.encoder   )