






'''simple docstring'''
import io
import itertools
import json
from dataclasses import dataclass
from typing import Optional

import pyarrow as pa
import pyarrow.json as paj

import datasets
from datasets.table import table_cast
from datasets.utils.file_utils import readline


UpperCamelCase__			=      datasets.utils.logging.get_logger(__name__)


@dataclass
class 							_UpperCAmelCase    (		datasets.BuilderConfig      ):
    __lowerCamelCase: int									=     None
    __lowerCamelCase: int									=     'utf-8'
    __lowerCamelCase: Dict									=     None
    __lowerCamelCase: List[Any]									=     None
    __lowerCamelCase: str									=     True  # deprecated
    __lowerCamelCase: Optional[int]									=     None  # deprecated
    __lowerCamelCase: Union[str, Any]									=     10 << 20  # 10MB
    __lowerCamelCase: Tuple									=     None


class 							_UpperCAmelCase    (		datasets.ArrowBasedBuilder      ):
    __lowerCamelCase: int									=     JsonConfig




    def 					lowerCAmelCase__					(      self	:							Optional[Any]      ):






       '''simple docstring'''






       if self.config.block_size is not None:
          logger.warning("The JSON loader parameter `block_size` is deprecated. Please use `chunksize` instead"      )
          lowercase_						:     Any							     =  self.config.block_size
       if self.config.use_threads is not True:
          logger.warning(
              "The JSON loader parameter `use_threads` is deprecated and doesn\'t have any effect anymore."      )
       if self.config.newlines_in_values is not None:
          raise ValueError("The JSON loader parameter `newlines_in_values` is no longer supported"      )
       return datasets.DatasetInfo(features=self.config.features      )




    def 					lowerCAmelCase__					(      self	:							Tuple	,  a	:							Union[str, Any]      ):






       '''simple docstring'''






       if not self.config.data_files:
          raise ValueError(f"""At least one data file must be specified, but got data_files={self.config.data_files}"""      )
       lowercase_						:     Dict							     =  dl_manager.download_and_extract(self.config.data_files      )
       if isinstance(snake_case__	,  (str, list, tuple)      ):
          lowercase_						:     Tuple							     =  data_files
          if isinstance(snake_case__	,  snake_case__      ):
             lowercase_						:     Any							     =  [files]
          lowercase_						:     str							     =  [dl_manager.iter_files(snake_case__      ) for file in files]
          return [datasets.SplitGenerator(name=datasets.Split.TRAIN	,  gen_kwargs={"files": files}      )]
       lowercase_						:     Union[str, Any]							     =  []
       for split_name, files in data_files.items():
          if isinstance(snake_case__	,  snake_case__      ):
             lowercase_						:     str							     =  [files]
          lowercase_						:     str							     =  [dl_manager.iter_files(snake_case__      ) for file in files]
          splits.append(datasets.SplitGenerator(name=snake_case__	,  gen_kwargs={"files": files}      )      )
       return splits




    def 					lowerCAmelCase__					(      self	:							Tuple	,  a	:							pa.Table      ):






       '''simple docstring'''






       if self.config.features is not None:
          # adding missing columns
          for column_name in set(self.config.features      ) - set(pa_table.column_names      ):
             lowercase_						:     Dict							     =  self.config.features.arrow_schema.field(snake_case__      ).type
             lowercase_						:     Tuple							     =  pa_table.append_column(snake_case__	,  pa.array([None] * len(snake_case__      )	,  type=snake_case__      )      )
          # more expensive cast to support nested structures with keys in a different order
          # allows str <-> int/float or str to Audio for example
          lowercase_						:     List[str]							     =  table_cast(snake_case__	,  self.config.features.arrow_schema      )
       return pa_table





    def 					lowerCAmelCase__					(      self	:							Any	,  a	:							Optional[int]      ):






       '''simple docstring'''






       for file_idx, file in enumerate(itertools.chain.from_iterable(snake_case__      )      ):
          # If the file is one json object and if we need to look at the list of items in one specific field
          if self.config.field is not None:
             with open(snake_case__	,  encoding=self.config.encoding	,  errors=self.config.encoding_errors      ) as f:
                lowercase_						:     Optional[Any]							     =  json.load(snake_case__      )

             # We keep only the field we are interested in
             lowercase_						:     Tuple							     =  dataset[self.config.field]

             # We accept two format: a list of dicts or a dict of lists
             if isinstance(snake_case__	,  (list, tuple)      ):
                lowercase_						:     List[str]							     =  set().union(*[row.keys() for row in dataset]      )
                lowercase_						:     Union[str, Any]							     =  {col: [row.get(snake_case__      ) for row in dataset] for col in keys}
             else:
                lowercase_						:     Any							     =  dataset
             lowercase_						:     Union[str, Any]							     =  pa.Table.from_pydict(snake_case__      )
             yield file_idx, self._cast_table(snake_case__      )

          # If the file has one json object per line
          else:
             with open(snake_case__	,  "rb"      ) as f:
                lowercase_						:     str							     =  0
                # Use block_size equal to the chunk size divided by 32 to leverage multithreading
                # Set a default minimum value of 16kB if the chunk size is really small
                lowercase_						:     Optional[int]							     =  max(self.config.chunksize // 3_2	,  1_6 << 1_0      )
                lowercase_						:     List[str]							     =  (
                    self.config.encoding_errors if self.config.encoding_errors is not None else "strict"
                )
                while True:
                   lowercase_						:     str							     =  f.read(self.config.chunksize      )
                   if not batch:
                      break
                   # Finish current line
                   try:
                      batch += f.readline()
                   except (AttributeError, io.UnsupportedOperation):
                      batch += readline(snake_case__      )
                   # PyArrow only accepts utf-8 encoded bytes
                   if self.config.encoding != "utf-8":
                      lowercase_						:     List[str]							     =  batch.decode(self.config.encoding	,  errors=snake_case__      ).encode("utf-8"      )
                   try:
                      while True:
                         try:
                            lowercase_						:     List[str]							     =  paj.read_json(
                                io.BytesIO(snake_case__      )	,  read_options=paj.ReadOptions(block_size=snake_case__      )      )
                            break
                         except (pa.ArrowInvalid, pa.ArrowNotImplementedError) as e:
                            if (
                                isinstance(snake_case__	,  pa.ArrowInvalid      )
                                and "straddling" not in str(snake_case__      )
                                or block_size > len(snake_case__      )
                            ):
                               raise
                            else:
                               # Increase the block size in case it was too small.
                               # The block size will be reset for the next file.
                               logger.debug(
                                   f"""Batch of {len(snake_case__      )} bytes couldn't be parsed with block_size={block_size}. Retrying with block_size={block_size * 2}."""      )
                               block_size *= 2
                   except pa.ArrowInvalid as e:
                      try:
                         with open(
                             snake_case__	,  encoding=self.config.encoding	,  errors=self.config.encoding_errors      ) as f:
                            lowercase_						:     str							     =  json.load(snake_case__      )
                      except json.JSONDecodeError:
                         logger.error(f"""Failed to read file '{file}' with error {type(snake_case__      )}: {e}"""      )
                         raise e
                      # If possible, parse the file as a list of json objects and exit the loop
                      if isinstance(snake_case__	,  snake_case__      ):  # list is the only sequence type supported in JSON
                         try:
                            lowercase_						:     str							     =  set().union(*[row.keys() for row in dataset]      )
                            lowercase_						:     Tuple							     =  {col: [row.get(snake_case__      ) for row in dataset] for col in keys}
                            lowercase_						:     int							     =  pa.Table.from_pydict(snake_case__      )
                         except (pa.ArrowInvalid, AttributeError) as e:
                            logger.error(f"""Failed to read file '{file}' with error {type(snake_case__      )}: {e}"""      )
                            raise ValueError(f"""Not able to read records in the JSON file at {file}."""      ) from None
                         yield file_idx, self._cast_table(snake_case__      )
                         break
                      else:
                         logger.error(f"""Failed to read file '{file}' with error {type(snake_case__      )}: {e}"""      )
                         raise ValueError(
                             f"""Not able to read records in the JSON file at {file}. """
                             f"""You should probably indicate the field of the JSON file containing your records. """
                             f"""This JSON file contain the following fields: {str(list(dataset.keys()      )      )}. """
                             f"""Select the correct one and provide it as `field='XXX'` to the dataset loading method. """      ) from None
                        # Uncomment for debugging (will print the Arrow table size and elements)
                        # logger.warning(f"pa_table: {pa_table} num rows: {pa_table.num_rows}")
                        # logger.warning('\n'.join(str(pa_table.slice(i, 1).to_pydict()) for i in range(pa_table.num_rows)))
                   yield (file_idx, batch_idx), self._cast_table(snake_case__      )
                   batch_idx += 1





from abc import ABC, abstractmethod
from argparse import ArgumentParser





class     lowerCAmelCase__     (				SCREAMING_SNAKE_CASE_			):


		'''simple docstring'''






		@staticmethod
		@abstractmethod
		def   _snake_case       (       snake_case__     :     ArgumentParser		)      ->	Tuple:
				raise NotImplementedError()



		@abstractmethod
		def   _snake_case       (       self     :     Optional[Any]		)      ->	Optional[Any]:
				raise NotImplementedError()