




"""simple docstring"""






import multiprocessing
import os
from typing import BinaryIO, Optional, Union

import fsspec

from .. import Dataset, Features, NamedSplit, config
from ..formatting import query_table
from ..packaged_modules.json.json import Json
from ..utils import logging
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader




class 	a__      (							__lowercase	):





						def __init__(				self   :						Union[str, Any]  ,a__   :						NestedDataStructureLike[PathLike]  ,a__   :						Optional[NamedSplit] = None  ,a__   :						Optional[Features] = None  ,a__   :						str = None  ,a__   :						bool = False  ,a__   :						bool = False  ,a__   :						Optional[str] = None  ,a__   :						Optional[int] = None  ,**a__   :						int  ,)    ->							List[str]:




											"""simple docstring"""







											super().__init__(
											    a__  ,split=a__  ,features=a__  ,cache_dir=a__  ,keep_in_memory=a__  ,streaming=a__  ,num_proc=a__  ,**a__  ,)
											_lowerCAmelCase:Union[str, Any]						=	field
											_lowerCAmelCase:Dict						=	path_or_paths if isinstance(a__  ,a__) else {self.split: path_or_paths}
											_lowerCAmelCase:Any						=	Json(
											    cache_dir=a__  ,data_files=a__  ,features=a__  ,field=a__  ,**a__  ,)







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




											"""simple docstring"""







											if self.streaming:
																_lowerCAmelCase:int						=	self.builder.as_streaming_dataset(split=self.split)
											# Build regular (map-style) dataset
											else:
																_lowerCAmelCase:List[str]						=	None
																_lowerCAmelCase:str						=	None
																_lowerCAmelCase:List[Any]						=	None
																_lowerCAmelCase:int						=	None

																self.builder.download_and_prepare(
																    download_config=a__  ,download_mode=a__  ,verification_mode=a__  ,base_path=a__  ,num_proc=self.num_proc  ,)
																_lowerCAmelCase:Tuple						=	self.builder.as_dataset(
																    split=self.split  ,verification_mode=a__  ,in_memory=self.keep_in_memory)
											return dataset




class 	a__      :





						def __init__(				self   :						Any  ,a__   :						Dataset  ,a__   :						Union[PathLike, BinaryIO]  ,a__   :						Optional[int] = None  ,a__   :						Optional[int] = None  ,**a__   :						List[Any]  ,)    ->							List[str]:




											"""simple docstring"""







											if num_proc is not None and num_proc <= 0:
																raise ValueError(F'num_proc {num_proc} must be an integer > 0.')

											_lowerCAmelCase:List[str]						=	dataset
											_lowerCAmelCase:List[str]						=	path_or_buf
											_lowerCAmelCase:Optional[int]						=	batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
											_lowerCAmelCase:List[str]						=	num_proc
											_lowerCAmelCase:int						=	"""utf-8"""
											_lowerCAmelCase:Tuple						=	to_json_kwargs





						def    __UpperCamelCase  (				self   :						Tuple)    ->							int:




											"""simple docstring"""







											_lowerCAmelCase:Optional[int]						=	self.to_json_kwargs.pop('''path_or_buf'''  ,a__)
											_lowerCAmelCase:Optional[Any]						=	self.to_json_kwargs.pop('''orient'''  ,'''records''')
											_lowerCAmelCase:List[Any]						=	self.to_json_kwargs.pop('''lines'''  ,True if orient == '''records''' else False)
											_lowerCAmelCase:Optional[Any]						=	self.to_json_kwargs.pop('''index'''  ,False if orient in ['''split''', '''table'''] else True)
											_lowerCAmelCase:Dict						=	self.to_json_kwargs.pop('''compression'''  ,a__)

											if compression not in [None, "infer", "gzip", "bz2", "xz"]:
																raise NotImplementedError(F'`datasets` currently does not support {compression} compression')

											if isinstance(self.path_or_buf  ,(str, bytes, os.PathLike)):
																with fsspec.open(self.path_or_buf  ,'''wb'''  ,compression=a__) as buffer:
																					_lowerCAmelCase:Any						=	self._write(file_obj=a__  ,orient=a__  ,lines=a__  ,index=a__  ,**self.to_json_kwargs)
											else:
																if compression:
																					raise NotImplementedError(
																					    F'The compression parameter is not supported when writing to a buffer, but compression={compression}'
																					    ''' was passed. Please provide a local path instead.''')
																_lowerCAmelCase:Dict						=	self._write(
																    file_obj=self.path_or_buf  ,orient=a__  ,lines=a__  ,index=a__  ,**self.to_json_kwargs)
											return written





						def    __UpperCamelCase  (				self   :						Optional[Any]  ,a__   :						Tuple)    ->							Optional[int]:




											"""simple docstring"""







											_lowerCAmelCase:Union[str, Any]						=	args

											_lowerCAmelCase:List[Any]						=	query_table(
											    table=self.dataset.data  ,key=slice(a__  ,offset + self.batch_size)  ,indices=self.dataset._indices  ,)
											_lowerCAmelCase:str						=	batch.to_pandas().to_json(
											    path_or_buf=a__  ,orient=a__  ,lines=a__  ,index=a__  ,**a__)
											if not json_str.endswith('''\n'''):
																json_str += "\n"
											return json_str.encode(self.encoding)







						def    __UpperCamelCase  (				self   :						Tuple  ,a__   :						BinaryIO  ,a__   :						List[str]  ,a__   :						Union[str, Any]  ,a__   :						Optional[int]  ,**a__   :						int  ,)    ->							int:




											"""simple docstring"""







											_lowerCAmelCase:Tuple						=	0

											if self.num_proc is None or self.num_proc == 1:
																for offset in logging.tqdm(
																    range(0  ,len(self.dataset)  ,self.batch_size)  ,unit='''ba'''  ,disable=not logging.is_progress_bar_enabled()  ,desc='''Creating json from Arrow format'''  ,):
																					_lowerCAmelCase:Tuple						=	self._batch_json((offset, orient, lines, index, to_json_kwargs))
																					written += file_obj.write(a__)
											else:
																_lowerCAmelCase:Any						=	len(self.dataset), self.batch_size
																with multiprocessing.Pool(self.num_proc) as pool:
																					for json_str in logging.tqdm(
																					    pool.imap(
																					        self._batch_json  ,[(offset, orient, lines, index, to_json_kwargs) for offset in range(0  ,a__  ,a__)]  ,)  ,total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size  ,unit='''ba'''  ,disable=not logging.is_progress_bar_enabled()  ,desc='''Creating json from Arrow format'''  ,):
																										written += file_obj.write(a__)

											return written













'''simple docstring'''






import json
import os
from pathlib import Path

import pytest

from datasets.download.download_config import DownloadConfig
from datasets.download.download_manager import DownloadManager
from datasets.utils.file_utils import hash_url_to_filename


__UpperCamelCase					    =	"http://www.mocksite.com/file1.txt"
__UpperCamelCase					    =	"\"text\": [\"foo\", \"foo\"]"
__UpperCamelCase					    =	"6d8ce9aa78a471c7477201efbeabd3bb01ac2e7d100a6dc024ba1608361f90a8"




class  _A :
		lowercase__:   str				     =   200
		lowercase__:   List[str]				     =   {'''Content-Length''': '''100'''}
		lowercase__:   Union[str, Any]				     =   {}

		def 	lowercase__     (   self					:   Any  ,      **__magic_name__					:   List[Any]			)					->    Dict:


					"""simple docstring"""

					return [bytes(__magic_name__  ,      """utf-8"""			)]







def 					_a						(       *_lowerCamelCase    ,	**_lowerCamelCase ) ->				List[str]:



			"""simple docstring"""


			return MockResponse()







@pytest.mark.parametrize("""urls_type"""    ,	[str, list, dict] )
def 					_a						(       _lowerCamelCase    ,	_lowerCamelCase    ,	_lowerCamelCase ) ->				List[str]:



			"""simple docstring"""


			import requests

			monkeypatch.setattr(_lowerCamelCase    ,	"""request"""    ,	_lowerCamelCase )

			__snake_case					:      Union[str, Any]							      =     URL
			if issubclass(_lowerCamelCase    ,	_lowerCamelCase ):
						__snake_case					:      str							      =     url
			elif issubclass(_lowerCamelCase    ,	_lowerCamelCase ):
						__snake_case					:      Dict							      =     [url]
			elif issubclass(_lowerCamelCase    ,	_lowerCamelCase ):
						__snake_case					:      Union[str, Any]							      =     {"""train""": url}
			__snake_case					:      Dict							      =     """dummy"""
			__snake_case					:      List[str]							      =     """downloads"""
			__snake_case					:      List[Any]							      =     tmp_path
			__snake_case					:      List[Any]							      =     DownloadConfig(
			    cache_dir=os.path.join(_lowerCamelCase    ,	_lowerCamelCase )    ,	use_etag=_lowerCamelCase    ,	)
			__snake_case					:      List[str]							      =     DownloadManager(dataset_name=_lowerCamelCase    ,	download_config=_lowerCamelCase )
			__snake_case					:      int							      =     dl_manager.download(_lowerCamelCase )
			__snake_case					:      Tuple							      =     urls
			for downloaded_paths in [downloaded_paths]:
						if isinstance(_lowerCamelCase    ,	_lowerCamelCase ):
									__snake_case					:      Any							      =     [downloaded_paths]
									__snake_case					:      List[Any]							      =     [urls]
						elif isinstance(_lowerCamelCase    ,	_lowerCamelCase ):
									assert "train" in downloaded_paths.keys()
									__snake_case					:      Tuple							      =     downloaded_paths.values()
									__snake_case					:      Optional[int]							      =     urls.values()
						assert downloaded_paths
						for downloaded_path, input_url in zip(_lowerCamelCase    ,	_lowerCamelCase ):
									assert downloaded_path == dl_manager.downloaded_paths[input_url]
									__snake_case					:      List[str]							      =     Path(_lowerCamelCase )
									__snake_case					:      Any							      =     downloaded_path.parts
									assert parts[-1] == HASH
									assert parts[-2] == cache_subdir
									assert downloaded_path.exists()
									__snake_case					:      Union[str, Any]							      =     downloaded_path.read_text()
									assert content == CONTENT
									__snake_case					:      List[str]							      =     downloaded_path.with_suffix(""".json""" )
									assert metadata_downloaded_path.exists()
									__snake_case					:      Union[str, Any]							      =     json.loads(metadata_downloaded_path.read_text() )
									assert metadata_content == {"url": URL, "etag": None}







@pytest.mark.parametrize("""paths_type"""    ,	[str, list, dict] )
def 					_a						(       _lowerCamelCase    ,	_lowerCamelCase    ,	_lowerCamelCase ) ->				List[Any]:



			"""simple docstring"""


			__snake_case					:      Any							      =     str(_lowerCamelCase )
			if issubclass(_lowerCamelCase    ,	_lowerCamelCase ):
						__snake_case					:      Optional[int]							      =     filename
			elif issubclass(_lowerCamelCase    ,	_lowerCamelCase ):
						__snake_case					:      Tuple							      =     [filename]
			elif issubclass(_lowerCamelCase    ,	_lowerCamelCase ):
						__snake_case					:      Dict							      =     {"""train""": filename}
			__snake_case					:      Optional[Any]							      =     """dummy"""
			__snake_case					:      List[Any]							      =     xz_file.parent
			__snake_case					:      int							      =     """extracted"""
			__snake_case					:      Dict							      =     DownloadConfig(
			    cache_dir=_lowerCamelCase    ,	use_etag=_lowerCamelCase    ,	)
			__snake_case					:      List[str]							      =     DownloadManager(dataset_name=_lowerCamelCase    ,	download_config=_lowerCamelCase )
			__snake_case					:      Optional[Any]							      =     dl_manager.extract(_lowerCamelCase )
			__snake_case					:      Union[str, Any]							      =     paths
			for extracted_paths in [extracted_paths]:
						if isinstance(_lowerCamelCase    ,	_lowerCamelCase ):
									__snake_case					:      Dict							      =     [extracted_paths]
									__snake_case					:      int							      =     [paths]
						elif isinstance(_lowerCamelCase    ,	_lowerCamelCase ):
									assert "train" in extracted_paths.keys()
									__snake_case					:      int							      =     extracted_paths.values()
									__snake_case					:      int							      =     paths.values()
						assert extracted_paths
						for extracted_path, input_path in zip(_lowerCamelCase    ,	_lowerCamelCase ):
									assert extracted_path == dl_manager.extracted_paths[input_path]
									__snake_case					:      Any							      =     Path(_lowerCamelCase )
									__snake_case					:      str							      =     extracted_path.parts
									assert parts[-1] == hash_url_to_filename(_lowerCamelCase    ,	etag=_lowerCamelCase )
									assert parts[-2] == extracted_subdir
									assert extracted_path.exists()
									__snake_case					:      Optional[int]							      =     extracted_path.read_text()
									__snake_case					:      str							      =     text_file.read_text()
									assert extracted_file_content == expected_file_content







def 					_a						(       _lowerCamelCase    ,	_lowerCamelCase ) ->				Optional[int]:



			"""simple docstring"""


			assert path.endswith(""".jsonl""" )
			for num_items, line in enumerate(_lowerCamelCase    ,	start=1 ):
						__snake_case					:      Tuple							      =     json.loads(line.decode("""utf-8""" ) )
						assert item.keys() == {"col_1", "col_2", "col_3"}
			assert num_items == 4







@pytest.mark.parametrize("""archive_jsonl"""    ,	["""tar_jsonl_path""", """zip_jsonl_path"""] )
def 					_a						(       _lowerCamelCase    ,	_lowerCamelCase ) ->				Optional[int]:



			"""simple docstring"""


			__snake_case					:      Any							      =     request.getfixturevalue(_lowerCamelCase )
			__snake_case					:      str							      =     DownloadManager()
			for num_jsonl, (path, file) in enumerate(dl_manager.iter_archive(_lowerCamelCase )    ,	start=1 ):
						_test_jsonl(_lowerCamelCase    ,	_lowerCamelCase )
			assert num_jsonl == 2







@pytest.mark.parametrize("""archive_nested_jsonl"""    ,	["""tar_nested_jsonl_path""", """zip_nested_jsonl_path"""] )
def 					_a						(       _lowerCamelCase    ,	_lowerCamelCase ) ->				List[str]:



			"""simple docstring"""


			__snake_case					:      int							      =     request.getfixturevalue(_lowerCamelCase )
			__snake_case					:      List[str]							      =     DownloadManager()
			for num_tar, (path, file) in enumerate(dl_manager.iter_archive(_lowerCamelCase )    ,	start=1 ):
						for num_jsonl, (subpath, subfile) in enumerate(dl_manager.iter_archive(_lowerCamelCase )    ,	start=1 ):
									_test_jsonl(_lowerCamelCase    ,	_lowerCamelCase )
			assert num_tar == 1
			assert num_jsonl == 2







def 					_a						(       _lowerCamelCase ) ->				Any:



			"""simple docstring"""


			__snake_case					:      List[str]							      =     DownloadManager()
			for num_file, file in enumerate(dl_manager.iter_files(_lowerCamelCase )    ,	start=1 ):
						assert os.path.basename(_lowerCamelCase ) == ("test.txt" if num_file == 1 else "train.txt")
			assert num_file == 2


