"""simple docstring"""






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

import sentencepiece as spm

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


__A       =					'▁'

__A       =					{'vocab_file': 'spiece.model'}

__A       =					{
    'vocab_file': {'google/pegasus-xsum': 'https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model'}
}

__A       =					{
    'google/pegasus-xsum': 5_1_2,
}


__A       =					logging.get_logger(__name__)


class        UpperCAmelCase		(_UpperCAmelCase    ):



    """simple docstring"""

    _UpperCAmelCase      :List[Any]		       =					VOCAB_FILES_NAMES

    _UpperCAmelCase      :Any		       =					VOCAB_FILES_NAMES
    _UpperCAmelCase      :List[Any]		       =					PRETRAINED_VOCAB_FILES_MAP
    _UpperCAmelCase      :Dict		       =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    _UpperCAmelCase      :Dict		       =					['''input_ids''', '''attention_mask''']






    def __init__(  self       ,     _UpperCAmelCase       ,     _UpperCAmelCase="<pad>"       ,     _UpperCAmelCase="</s>"       ,     _UpperCAmelCase="<unk>"       ,     _UpperCAmelCase="<mask_2>"       ,     _UpperCAmelCase="<mask_1>"       ,     _UpperCAmelCase=None       ,     _UpperCAmelCase=103       ,     _UpperCAmelCase = None       ,     **_UpperCAmelCase       ,     ):
       lowercase__:  str   = offset
       if additional_special_tokens is not None:
          if not isinstance(A__       ,     A__				):
             raise TypeError(
                 F"""additional_special_tokens should be of type {type(A__				)}, but is"""
                 F""" {type(A__				)}"""				)

          lowercase__:  Optional[int]   = (
              ([mask_token_sent] + additional_special_tokens)
              if mask_token_sent not in additional_special_tokens and mask_token_sent is not None
              else additional_special_tokens
          )
          # fill additional tokens with ..., <unk_token_102> in case not all additional tokens are already taken
          additional_special_tokens_extended += [
              F"""<unk_{i}>""" for i in range(len(A__				)       ,     self.offset - 1				)
          ]

          if len(set(A__				)				) != len(A__				):
             raise ValueError(
                 '''Please make sure that the provided additional_special_tokens do not contain an incorrectly'''
                 F""" shifted list of <unk_x> tokens. Found {additional_special_tokens_extended}."""				)
          lowercase__:  List[str]   = additional_special_tokens_extended
       else:
          lowercase__:  Optional[int]   = [mask_token_sent] if mask_token_sent is not None else []
          additional_special_tokens += [F"""<unk_{i}>""" for i in range(2       ,     self.offset				)]

       lowercase__:  Optional[Any]   = {} if sp_model_kwargs is None else sp_model_kwargs

       super().__init__(
           eos_token=A__       ,     unk_token=A__       ,     mask_token=A__       ,     pad_token=A__       ,     mask_token_sent=A__       ,     offset=A__       ,     additional_special_tokens=A__       ,     sp_model_kwargs=self.sp_model_kwargs       ,     **A__       ,     )
       lowercase__:  Any   = mask_token_sent
       lowercase__:  int   = vocab_file
       lowercase__:  List[Any]   = spm.SentencePieceProcessor(**self.sp_model_kwargs				)
       self.sp_model.Load(A__				)

       # add special tokens to encoder dict
       lowercase__:  Dict[int, str]   = {
           0: self.pad_token,
           1: self.eos_token,
       }

       if self.mask_token_sent is not None:
          self.encoder.update(
              {
                  2: self.mask_token_sent,
                  3: self.mask_token,
              }				)

       if self.offset > 0:
          # entries 2-104 are only used for pretraining and called <mask_1>, <mask_2>, unk_2, ...unk_102
          # mask_token_sent is already added to list -> so start at 1
          self.encoder.update({i + 3: additional_special_tokens[i] for i in range(1       ,     self.offset - 1				)}				)

       lowercase__:  Dict[str, int]   = {v: k for k, v in self.encoder.items()}






    @property
    def _snake_case     (  self				):
       return len(self.sp_model				) + self.offset






    def _snake_case     (  self				):
       lowercase__:  Tuple   = {self.convert_ids_to_tokens(A__				): i for i in range(self.vocab_size				)}
       vocab.update(self.added_tokens_encoder				)
       return vocab






    def __getstate__(  self				):
       lowercase__:  Any   = self.__dict__.copy()
       lowercase__:  Union[str, Any]   = None
       return state






    def __setstate__(  self       ,     _UpperCAmelCase				):
       lowercase__:  Union[str, Any]   = d

       # for backward compatibility
       if not hasattr(self       ,     '''sp_model_kwargs'''				):
          lowercase__:  List[str]   = {}

       lowercase__:  Union[str, Any]   = spm.SentencePieceProcessor(**self.sp_model_kwargs				)
       self.sp_model.Load(self.vocab_file				)






    def _snake_case     (  self       ,     _UpperCAmelCase				):
       return self.sp_model.encode(A__       ,     out_type=A__				)






    def _snake_case     (  self       ,     _UpperCAmelCase				):
       if token in self.decoder:
          return self.decoder[token]
       elif token in self.added_tokens_decoder:
          return self.added_tokens_decoder[token]
       lowercase__:  str   = self.sp_model.piece_to_id(A__				)
       return sp_id + self.offset






    def _snake_case     (  self       ,     _UpperCAmelCase				):
       if index in self.encoder:
          return self.encoder[index]
       elif index in self.added_tokens_encoder:
          return self.added_tokens_encoder[index]
       else:
          lowercase__:  List[Any]   = self.sp_model.IdToPiece(index - self.offset				)
       return token






    def _snake_case     (  self       ,     _UpperCAmelCase				):
       lowercase__:  int   = []
       lowercase__:  Optional[Any]   = """"""
       for token in tokens:
          # make sure that special tokens are not decoded using sentencepiece model
          if token in self.all_special_tokens:
             out_string += self.sp_model.decode(A__				) + token
             lowercase__:  Dict   = []
          else:
             current_sub_tokens.append(A__				)
       out_string += self.sp_model.decode(A__				)
       return out_string.strip()






    def _snake_case     (  self       ,     _UpperCAmelCase=False				):
       return 1






    def _snake_case     (  self       ,     _UpperCAmelCase				):
       lowercase__:  List[str]   = set(self.all_special_ids				)  # call it once instead of inside list comp
       all_special_ids.remove(self.unk_token_id				)  # <unk> is only sometimes special

       return [1 if x in all_special_ids else 0 for x in seq]






    def _snake_case     (  self       ,     _UpperCAmelCase       ,     _UpperCAmelCase = None       ,     _UpperCAmelCase = False				):
       if already_has_special_tokens:
          return self._special_token_mask(A__				)
       elif token_ids_a is None:
          return self._special_token_mask(A__				) + [1]
       else:
          return self._special_token_mask(token_ids_a + token_ids_a				) + [1]






    def _snake_case     (  self       ,     _UpperCAmelCase       ,     _UpperCAmelCase=None				):
       if token_ids_a is None:
          return token_ids_a + [self.eos_token_id]
       # We don't expect to process pairs, but leave the pair logic for API consistency
       return token_ids_a + token_ids_a + [self.eos_token_id]






    def _snake_case     (  self       ,     _UpperCAmelCase       ,     _UpperCAmelCase = None				):
       if not os.path.isdir(A__				):
          logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""				)
          return
       lowercase__:  Any   = os.path.join(
           A__       ,     (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']				)

       if os.path.abspath(self.vocab_file				) != os.path.abspath(A__				) and os.path.isfile(self.vocab_file				):
          copyfile(self.vocab_file       ,     A__				)
       elif not os.path.isfile(self.vocab_file				):
          with open(A__       ,     '''wb'''				) as fi:
             lowercase__:  str   = self.sp_model.serialized_model_proto()
             fi.write(A__				)

       return (out_vocab_file,)



import os
import posixpath
import uuid
from dataclasses import dataclass
from typing import TYPE_CHECKING, Iterable, List, Optional, Tuple, Union

import numpy as np
import pyarrow as pa

import datasets
from datasets.arrow_writer import ArrowWriter, ParquetWriter
from datasets.config import MAX_SHARD_SIZE
from datasets.filesystems import (
    is_remote_filesystem,
    rename,
)
from datasets.iterable_dataset import _BaseExamplesIterable
from datasets.utils.py_utils import convert_file_size_to_int


A_			:					Any    			=     datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
	import pyspark



@dataclass
class        _a		(datasets.BuilderConfig							):


					'''simple docstring'''






					UpperCAmelCase__:  Optional[datasets.Features]    				=			None



def 		UpperCamelCase   (lowercase_:					"pyspark.sql.DataFrame"       , lowercase_:					List[int]       , )	-> Dict:
	import pyspark

	def generate_fn():
		A__				:       List[str]	  =						df.select("""*"""       , pyspark.sql.functions.spark_partition_id().alias("""part_id"""   )   )
		for partition_id in partition_order:
			A__				:       List[Any]	  =						df_with_partition_id.select("""*"""   ).where(f"""part_id = {partition_id}"""   ).drop("""part_id"""   )
			A__				:       Optional[int]	  =						partition_df.collect()
			A__				:       Any	  =						0
			for row in rows:
				yield f"""{partition_id}_{row_id}""", row.asDict()
				row_id += 1

	return generate_fn



class        _a		(_BaseExamplesIterable							):


					'''simple docstring'''






					def __init__(					self       ,			A__       ,			A__=None       ,			):
						A__				:       List[str]	  =						df
						A__				:       Optional[int]	  =						partition_order or range(self.df.rdd.getNumPartitions()     )
						A__				:       Tuple	  =						_generate_iterable_examples(self.df       ,			self.partition_order     )






					def __iter__(					self     ):
						yield from self.generate_examples_fn()






					def     __A     (					self       ,			A__     ):
						A__				:       str	  =						list(range(self.df.rdd.getNumPartitions()     )     )
						generator.shuffle(A__     )
						return SparkExamplesIterable(self.df       ,			partition_order=A__     )






					def     __A     (					self       ,			A__       ,			A__     ):
						A__				:       Optional[int]	  =						self.split_shard_indices_by_worker(A__       ,			A__     )
						return SparkExamplesIterable(self.df       ,			partition_order=A__     )






					@property
					def     __A     (					self     ):
						return len(self.partition_order     )



class        _a		(datasets.DatasetBuilder							):


					'''simple docstring'''






					UpperCAmelCase__:  Union[str, Any]    				=			SparkConfig






					def __init__(					self       ,			A__       ,			A__ = None       ,			A__ = None       ,			**A__       ,			):
						import pyspark

						A__				:       Dict	  =						pyspark.sql.SparkSession.builder.getOrCreate()
						A__				:       int	  =						df
						A__				:       Any	  =						working_dir

						super().__init__(
						    cache_dir=A__       ,			config_name=str(self.df.semanticHash()     )       ,			**A__       ,			)






					def     __A     (					self     ):
						# Returns the path of the created file.
						def create_cache_and_write_probe(A__     ):
							# makedirs with exist_ok will recursively create the directory. It will not throw an error if directories
							# already exist.
							os.makedirs(self._cache_dir       ,			exist_ok=A__     )
							A__				:       List[str]	  =						os.path.join(self._cache_dir       ,			"""fs_test""" + uuid.uuida().hex     )
							# Opening the file in append mode will create a new file unless it already exists, in which case it will not
							# change the file contents.
							open(A__       ,			"""a"""     )
							return [probe_file]

						if self._spark.conf.get("""spark.master"""       ,			""""""     ).startswith("""local"""     ):
							return

						# If the cluster is multi-node, make sure that the user provided a cache_dir and that it is on an NFS
						# accessible to the driver.
						# TODO: Stream batches to the driver using ArrowCollectSerializer instead of throwing an error.
						if self._cache_dir:
							A__				:       int	  =						(
							    self._spark.sparkContext.parallelize(range(1     )       ,			1     ).mapPartitions(A__     ).collect()
							)
							if os.path.isfile(probe[0]     ):
								return

						raise ValueError(
						    """When using Dataset.from_spark on a multi-node cluster, the driver and all workers should be able to access cache_dir"""     )






					def     __A     (					self     ):
						return datasets.DatasetInfo(features=self.config.features     )






					def     __A     (					self       ,			A__     ):
						return [datasets.SplitGenerator(name=datasets.Split.TRAIN     )]






					def     __A     (					self       ,			A__     ):
						import pyspark

						def get_arrow_batch_size(A__     ):
							for batch in it:
								yield pa.RecordBatch.from_pydict({"""batch_bytes""": [batch.nbytes]}     )

						A__				:       Dict	  =						self.df.count()
						A__				:       List[Any]	  =						df_num_rows if df_num_rows <= 100 else 100
						# Approximate the size of each row (in Arrow format) by averaging over a max-100-row sample.
						A__				:       Union[str, Any]	  =						(
						    self.df.limit(A__     )
						    .repartition(1     )
						    .mapInArrow(A__       ,			"""batch_bytes: long"""     )
						    .agg(pyspark.sql.functions.sum("""batch_bytes"""     ).alias("""sample_bytes"""     )     )
						    .collect()[0]
						    .sample_bytes
						    / sample_num_rows
						)
						A__				:       str	  =						approx_bytes_per_row * df_num_rows
						if approx_total_size > max_shard_size:
							# Make sure there is at least one row per partition.
							A__				:       Dict	  =						min(A__       ,			int(approx_total_size / max_shard_size     )     )
							A__				:       int	  =						self.df.repartition(A__     )






					def     __A     (					self       ,			A__       ,			A__       ,			A__       ,			):
						import pyspark

						A__				:       Optional[int]	  =						ParquetWriter if file_format == """parquet""" else ArrowWriter
						A__				:       Any	  =						os.path.join(self._working_dir       ,			os.path.basename(A__     )     ) if self._working_dir else fpath
						A__				:       Union[str, Any]	  =						file_format == """parquet"""

						# Define these so that we don't reference self in write_arrow, which will result in a pickling error due to
						# pickling the SparkContext.
						A__				:       str	  =						self.config.features
						A__				:       Union[str, Any]	  =						self._writer_batch_size
						A__				:       Tuple	  =						self._fs.storage_options

						def write_arrow(A__     ):
							# Within the same SparkContext, no two task attempts will share the same attempt ID.
							A__				:       str	  =						pyspark.TaskContext().taskAttemptId()
							A__				:       str	  =						next(A__       ,			A__     )
							if first_batch is None:
								# Some partitions might not receive any data.
								return pa.RecordBatch.from_arrays(
								    [[task_id], [0], [0]]       ,			names=["""task_id""", """num_examples""", """num_bytes"""]       ,			)
							A__				:       List[Any]	  =						0
							A__				:       Optional[Any]	  =						writer_class(
							    features=A__       ,			path=working_fpath.replace("""SSSSS"""       ,			F"""{shard_id:05d}"""     ).replace("""TTTTT"""       ,			F"""{task_id:05d}"""     )       ,			writer_batch_size=A__       ,			storage_options=A__       ,			embed_local_files=A__       ,			)
							A__				:       Tuple	  =						pa.Table.from_batches([first_batch]     )
							writer.write_table(A__     )
							for batch in it:
								if max_shard_size is not None and writer._num_bytes >= max_shard_size:
									A__  ,  A__				:       Union[str, Any]	  =						writer.finalize()
									writer.close()
									yield pa.RecordBatch.from_arrays(
									    [[task_id], [num_examples], [num_bytes]]       ,			names=["""task_id""", """num_examples""", """num_bytes"""]       ,			)
									shard_id += 1
									A__				:       Tuple	  =						writer_class(
									    features=writer._features       ,			path=working_fpath.replace("""SSSSS"""       ,			F"""{shard_id:05d}"""     ).replace("""TTTTT"""       ,			F"""{task_id:05d}"""     )       ,			writer_batch_size=A__       ,			storage_options=A__       ,			embed_local_files=A__       ,			)
								A__				:       Optional[int]	  =						pa.Table.from_batches([batch]     )
								writer.write_table(A__     )

							if writer._num_bytes > 0:
								A__  ,  A__				:       Any	  =						writer.finalize()
								writer.close()
								yield pa.RecordBatch.from_arrays(
								    [[task_id], [num_examples], [num_bytes]]       ,			names=["""task_id""", """num_examples""", """num_bytes"""]       ,			)

							if working_fpath != fpath:
								for file in os.listdir(os.path.dirname(A__     )     ):
									A__				:       Optional[Any]	  =						os.path.join(os.path.dirname(A__     )       ,			os.path.basename(A__     )     )
									shutil.move(A__       ,			A__     )

						A__				:       Tuple	  =						(
						    self.df.mapInArrow(A__       ,			"""task_id: long, num_examples: long, num_bytes: long"""     )
						    .groupBy("""task_id"""     )
						    .agg(
						        pyspark.sql.functions.sum("""num_examples"""     ).alias("""total_num_examples"""     )       ,			pyspark.sql.functions.sum("""num_bytes"""     ).alias("""total_num_bytes"""     )       ,			pyspark.sql.functions.count("""num_bytes"""     ).alias("""num_shards"""     )       ,			pyspark.sql.functions.collect_list("""num_examples"""     ).alias("""shard_lengths"""     )       ,			)
						    .collect()
						)
						for row in stats:
							yield row.task_id, (row.total_num_examples, row.total_num_bytes, row.num_shards, row.shard_lengths)






					def     __A     (					self       ,			A__       ,			A__ = "arrow"       ,			A__ = None       ,			A__ = None       ,			**A__       ,			):
						self._validate_cache_dir()

						A__				:       Union[str, Any]	  =						convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE     )
						self._repartition_df_if_needed(A__     )
						A__				:       Any	  =						not is_remote_filesystem(self._fs     )
						A__				:       Optional[int]	  =						os.path.join if is_local else posixpath.join

						A__				:       Dict	  =						"""-TTTTT-SSSSS-of-NNNNN"""
						A__				:       Any	  =						F"""{self.name}-{split_generator.name}{SUFFIX}.{file_format}"""
						A__				:       Any	  =						path_join(self._output_dir       ,			A__     )

						A__				:       Tuple	  =						0
						A__				:       str	  =						0
						A__				:       List[Any]	  =						0
						A__				:       List[Any]	  =						[]
						A__				:       Optional[Any]	  =						[]

						for task_id, content in self._prepare_split_single(A__       ,			A__       ,			A__     ):
							(
							    (
							    A__
							)  ,  (
							    A__
							)  ,  (
							    A__
							)  ,  (
							    A__
							)  ,  
							)				:       List[Any]	  =						content
							if num_bytes > 0:
								total_num_examples += num_examples
								total_num_bytes += num_bytes
								total_shards += num_shards
								task_id_and_num_shards.append((task_id, num_shards)     )
								all_shard_lengths.extend(A__     )

						A__				:       Optional[int]	  =						total_num_examples
						A__				:       Union[str, Any]	  =						total_num_bytes

						# should rename everything at the end
						logger.debug(F"""Renaming {total_shards} shards."""     )
						if total_shards > 1:
							A__				:       int	  =						all_shard_lengths

							# Define fs outside of _rename_shard so that we don't reference self in the function, which will result in a
							# pickling error due to pickling the SparkContext.
							A__				:       Dict	  =						self._fs

							# use the -SSSSS-of-NNNNN pattern
							def _rename_shard(
							    A__       ,			A__       ,			A__       ,			):
								rename(
								    A__       ,			fpath.replace("""SSSSS"""       ,			F"""{shard_id:05d}"""     ).replace("""TTTTT"""       ,			F"""{task_id:05d}"""     )       ,			fpath.replace("""TTTTT-SSSSS"""       ,			F"""{global_shard_id:05d}"""     ).replace("""NNNNN"""       ,			F"""{total_shards:05d}"""     )       ,			)

							A__				:       List[Any]	  =						[]
							A__				:       Union[str, Any]	  =						0
							for i in range(len(A__     )     ):
								A__  ,  A__				:       Optional[int]	  =						task_id_and_num_shards[i]
								for shard_id in range(A__     ):
									args.append([task_id, shard_id, global_shard_id]     )
									global_shard_id += 1
							self._spark.sparkContext.parallelize(A__       ,			len(A__     )     ).map(lambda A__     : _rename_shard(*A__     )     ).collect()
						else:
							# don't use any pattern
							A__				:       List[Any]	  =						0
							A__				:       List[str]	  =						task_id_and_num_shards[0][0]
							self._rename(
							    fpath.replace("""SSSSS"""       ,			F"""{shard_id:05d}"""     ).replace("""TTTTT"""       ,			F"""{task_id:05d}"""     )       ,			fpath.replace(A__       ,			""""""     )       ,			)






					def     __A     (					self       ,			A__       ,			):
						return SparkExamplesIterable(self.df     )



