






import bza
import gzip
import lzma
import os
import shutil
import struct
import tarfile
import warnings
import zipfile
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Dict, List, Optional, Type, Union

from .. import config
from .filelock import FileLock
from .logging import get_logger


__A      =				get_logger(__name__)





class 						SCREAMING_SNAKE_CASE				:





							"""simple docstring"""

							def __init__(  self:   Dict		,     __A:   Optional[str] = None   )    ->				List[Any]:
												_A								= (
												    os.path.join(__A		,     config.EXTRACTED_DATASETS_DIR   ) if cache_dir else config.EXTRACTED_DATASETS_PATH
												)
												_A								= Extractor

							def       __A (  self:   Union[str, Any]		,     __A:   str   )    ->				str:
												from .file_utils import hash_url_to_filename

												# Path where we extract compressed archives
												# We extract in the cache dir, and get the extracted path name by hashing the original path"
												_A								= os.path.abspath(__A   )
												return os.path.join(self.extract_dir		,     hash_url_to_filename(__A   )   )

							def       __A (  self:   str		,     __A:   str		,     __A:   bool   )    ->				bool:
												return force_extract or (
												    not os.path.isfile(__A   ) and not (os.path.isdir(__A   ) and os.listdir(__A   ))
												)



							def       __A (  self:   List[str]		,     __A:   str		,     __A:   bool = False   )    ->				str:
												_A								= self.extractor.infer_extractor_format(__A   )
												if not extractor_format:
																	return input_path
												_A								= self._get_output_path(__A   )
												if self._do_extract(__A		,     __A   ):
																	self.extractor.extract(__A		,     __A		,     __A   )
												return output_path





class 						SCREAMING_SNAKE_CASE				( snake_case					):





							"""simple docstring"""

							@classmethod
							@abstractmethod
							def       __A (  cls:   str		,     __A:   Union[Path, str]		,     **__A:   Any   )    ->				bool:
												...



							@staticmethod
							@abstractmethod
							def       __A (  __A:   Union[Path, str]		,     __A:   Union[Path, str]   )    ->				None:
												...





class 						SCREAMING_SNAKE_CASE				( snake_case  ,	snake_case					):





							"""simple docstring"""

							A_						  =					[]

							@staticmethod
							def       __A (  __A:   Union[Path, str]		,     __A:   int   )    ->				List[Any]:
												with open(__A		,     '''rb'''   ) as f:
																	return f.read(__A   )



							@classmethod
							def       __A (  cls:   Any		,     __A:   Union[Path, str]		,     __A:   bytes = b""   )    ->				bool:
												if not magic_number:
																	_A								= max(len(__A   ) for cls_magic_number in cls.magic_numbers   )
																	try:
																						_A								= cls.read_magic_number(__A		,     __A   )
																	except OSError:
																						return False
												return any(magic_number.startswith(__A   ) for cls_magic_number in cls.magic_numbers   )





class 						SCREAMING_SNAKE_CASE				( snake_case					):





							"""simple docstring"""

							@classmethod
							def       __A (  cls:   Union[str, Any]		,     __A:   Union[Path, str]		,     **__A:   str   )    ->				bool:
												return tarfile.is_tarfile(__A   )

							@staticmethod
							def       __A (  __A:   Optional[Any]		,     __A:   Any   )    ->				str:

												def resolved(__A:   str   ) -> str:
																	return os.path.realpath(os.path.abspath(__A   )   )

												def badpath(__A:   str		,     __A:   str   ) -> bool:
																	# joinpath will ignore base if path is absolute
																	return not resolved(os.path.join(__A		,     __A   )   ).startswith(__A   )

												def badlink(__A:   Optional[int]		,     __A:   str   ) -> bool:
																	# Links are interpreted relative to the directory containing the link
																	_A								= resolved(os.path.join(__A		,     os.path.dirname(info.name   )   )   )
																	return badpath(info.linkname		,     base=__A   )

												_A								= resolved(__A   )

												for finfo in members:
																	if badpath(finfo.name		,     __A   ):
																						logger.error(f"""Extraction of {finfo.name} is blocked (illegal path)"""   )
																	elif finfo.issym() and badlink(__A		,     __A   ):
																						logger.error(f"""Extraction of {finfo.name} is blocked: Symlink to {finfo.linkname}"""   )
																	elif finfo.islnk() and badlink(__A		,     __A   ):
																						logger.error(f"""Extraction of {finfo.name} is blocked: Hard link to {finfo.linkname}"""   )
																	else:
																						yield finfo



							@staticmethod
							def       __A (  __A:   Union[Path, str]		,     __A:   Union[Path, str]   )    ->				None:
												os.makedirs(__A		,     exist_ok=__A   )
												_A								= tarfile.open(__A   )
												tar_file.extractall(__A		,     members=TarExtractor.safemembers(__A		,     __A   )   )
												tar_file.close()





class 						SCREAMING_SNAKE_CASE				( snake_case					):





							"""simple docstring"""

							A_						  =					[B"\x1F\x8B"]

							@staticmethod
							def       __A (  __A:   Union[Path, str]		,     __A:   Union[Path, str]   )    ->				None:
												with gzip.open(__A		,     '''rb'''   ) as gzip_file:
																	with open(__A		,     '''wb'''   ) as extracted_file:
																						shutil.copyfileobj(__A		,     __A   )





class 						SCREAMING_SNAKE_CASE				( snake_case					):





							"""simple docstring"""

							A_						  =					[
							    B"PK\x03\x04",
							    B"PK\x05\x06",  # empty archive
							    B"PK\x07\x08",  # spanned archive
							]

							@classmethod
							def       __A (  cls:   List[str]		,     __A:   Union[Path, str]		,     __A:   bytes = b""   )    ->				bool:
												if super().is_extractable(__A		,     magic_number=__A   ):
																	return True
												try:
																	# Alternative version of zipfile.is_zipfile that has less false positives, but misses executable zip archives.
																	# From: https://github.com/python/cpython/pull/5053
																	from zipfile import (
																	    _CD_SIGNATURE,
																	    _ECD_DISK_NUMBER,
																	    _ECD_DISK_START,
																	    _ECD_ENTRIES_TOTAL,
																	    _ECD_OFFSET,
																	    _ECD_SIZE,
																	    _EndRecData,
																	    sizeCentralDir,
																	    stringCentralDir,
																	    structCentralDir,
																	)

																	with open(__A		,     '''rb'''   ) as fp:
																						_A								= _EndRecData(__A   )
																						if endrec:
																											if endrec[_ECD_ENTRIES_TOTAL] == 0 and endrec[_ECD_SIZE] == 0 and endrec[_ECD_OFFSET] == 0:
																																return True  # Empty zipfiles are still zipfiles
																											elif endrec[_ECD_DISK_NUMBER] == endrec[_ECD_DISK_START]:
																																fp.seek(endrec[_ECD_OFFSET]   )  # Central directory is on the same disk
																																if fp.tell() == endrec[_ECD_OFFSET] and endrec[_ECD_SIZE] >= sizeCentralDir:
																																					_A								= fp.read(__A   )  # CD is where we expect it to be
																																					if len(__A   ) == sizeCentralDir:
																																										_A								= struct.unpack(__A		,     __A   )  # CD is the right size
																																										if centdir[_CD_SIGNATURE] == stringCentralDir:
																																															return True  # First central directory entry  has correct magic number
																	return False
												except Exception:  # catch all errors in case future python versions change the zipfile internals
																	return False



							@staticmethod
							def       __A (  __A:   Union[Path, str]		,     __A:   Union[Path, str]   )    ->				None:
												os.makedirs(__A		,     exist_ok=__A   )
												with zipfile.ZipFile(__A		,     '''r'''   ) as zip_file:
																	zip_file.extractall(__A   )
																	zip_file.close()





class 						SCREAMING_SNAKE_CASE				( snake_case					):





							"""simple docstring"""

							A_						  =					[B"\xFD\x37\x7A\x58\x5A\x00"]

							@staticmethod
							def       __A (  __A:   Union[Path, str]		,     __A:   Union[Path, str]   )    ->				None:
												with lzma.open(__A   ) as compressed_file:
																	with open(__A		,     '''wb'''   ) as extracted_file:
																						shutil.copyfileobj(__A		,     __A   )





class 						SCREAMING_SNAKE_CASE				( snake_case					):





							"""simple docstring"""

							A_						  =					[B"Rar!\x1a\x07\x00", B"Rar!\x1a\x07\x01\x00"]  # RAR_ID  # RAR5_ID

							@staticmethod
							def       __A (  __A:   Union[Path, str]		,     __A:   Union[Path, str]   )    ->				None:
												if not config.RARFILE_AVAILABLE:
																	raise ImportError('''Please pip install rarfile'''   )
												import rarfile

												os.makedirs(__A		,     exist_ok=__A   )
												_A								= rarfile.RarFile(__A   )
												rf.extractall(__A   )
												rf.close()





class 						SCREAMING_SNAKE_CASE				( snake_case					):





							"""simple docstring"""

							A_						  =					[B"\x28\xb5\x2F\xFD"]

							@staticmethod
							def       __A (  __A:   Union[Path, str]		,     __A:   Union[Path, str]   )    ->				None:
												if not config.ZSTANDARD_AVAILABLE:
																	raise ImportError('''Please pip install zstandard'''   )
												import zstandard as zstd

												_A								= zstd.ZstdDecompressor()
												with open(__A		,     '''rb'''   ) as ifh, open(__A		,     '''wb'''   ) as ofh:
																	dctx.copy_stream(__A		,     __A   )





class 						SCREAMING_SNAKE_CASE				( snake_case					):





							"""simple docstring"""

							A_						  =					[B"\x42\x5A\x68"]

							@staticmethod
							def       __A (  __A:   Union[Path, str]		,     __A:   Union[Path, str]   )    ->				None:
												with bza.open(__A		,     '''rb'''   ) as compressed_file:
																	with open(__A		,     '''wb'''   ) as extracted_file:
																						shutil.copyfileobj(__A		,     __A   )





class 						SCREAMING_SNAKE_CASE				( snake_case					):





							"""simple docstring"""

							A_						  =					[B"\x37\x7A\xBC\xAF\x27\x1C"]

							@staticmethod
							def       __A (  __A:   Union[Path, str]		,     __A:   Union[Path, str]   )    ->				None:
												if not config.PY7ZR_AVAILABLE:
																	raise ImportError('''Please pip install py7zr'''   )
												import pyazr

												os.makedirs(__A		,     exist_ok=__A   )
												with pyazr.SevenZipFile(__A		,     '''r'''   ) as archive:
																	archive.extractall(__A   )





class 						SCREAMING_SNAKE_CASE				( snake_case					):





							"""simple docstring"""

							A_						  =					[B"\x04\x22\x4D\x18"]

							@staticmethod
							def       __A (  __A:   Union[Path, str]		,     __A:   Union[Path, str]   )    ->				None:
												if not config.LZ4_AVAILABLE:
																	raise ImportError('''Please pip install lz4'''   )
												import lza.frame

												with lza.frame.open(__A		,     '''rb'''   ) as compressed_file:
																	with open(__A		,     '''wb'''   ) as extracted_file:
																						shutil.copyfileobj(__A		,     __A   )





class 						SCREAMING_SNAKE_CASE				:





							"""simple docstring"""

							A_						  =					{
							    "tar": TarExtractor,
							    "gzip": GzipExtractor,
							    "zip": ZipExtractor,
							    "xz": XzExtractor,
							    "rar": RarExtractor,
							    "zstd": ZstdExtractor,
							    "bz2": BzipaExtractor,
							    "7z": SevenZipExtractor,  # <Added version="2.4.0"/>
							    "lz4": LzaExtractor,  # <Added version="2.4.0"/>
							}

							@classmethod
							def       __A (  cls:   Any   )    ->				List[Any]:
												return max(
												    len(__A   )
												    for extractor in cls.extractors.values()
												    if issubclass(__A		,     __A   )
												    for extractor_magic_number in extractor.magic_numbers   )

							@staticmethod
							def       __A (  __A:   Union[Path, str]		,     __A:   int   )    ->				Tuple:
												try:
																	return MagicNumberBaseExtractor.read_magic_number(__A		,     magic_number_length=__A   )
												except OSError:
																	return b""

							@classmethod
							def       __A (  cls:   int		,     __A:   Union[Path, str]		,     __A:   bool = False   )    ->				bool:
												warnings.warn(
												    '''Method \'is_extractable\' was deprecated in version 2.4.0 and will be removed in 3.0.0. '''
												    '''Use \'infer_extractor_format\' instead.'''		,     category=__A		,     )
												_A								= cls.infer_extractor_format(__A   )
												if extractor_format:
																	return True if not return_extractor else (True, cls.extractors[extractor_format])
												return False if not return_extractor else (False, None)

							@classmethod
							def       __A (  cls:   Any		,     __A:   Union[Path, str]   )    ->				str:  # <Added version="2.4.0"/>
												_A								= cls._get_magic_number_max_length()
												_A								= cls._read_magic_number(__A		,     __A   )
												for extractor_format, extractor in cls.extractors.items():
																	if extractor.is_extractable(__A		,     magic_number=__A   ):
																						return extractor_format



							@classmethod
							def       __A (  cls:   Any		,     __A:   Union[Path, str]		,     __A:   Union[Path, str]		,     __A:   Optional[str] = None		,     __A:   Optional[BaseExtractor] = "deprecated"		,     )    ->				None:
												os.makedirs(os.path.dirname(__A   )		,     exist_ok=__A   )
												# Prevent parallel extractions
												_A								= str(Path(__A   ).with_suffix('''.lock'''   )   )
												with FileLock(__A   ):
																	shutil.rmtree(__A		,     ignore_errors=__A   )
																	if extractor_format or extractor != "deprecated":
																						if extractor != "deprecated" or not isinstance(__A		,     __A   ):  # passed as positional arg
																											warnings.warn(
																											    '''Parameter \'extractor\' was deprecated in version 2.4.0 and will be removed in 3.0.0. '''
																											    '''Use \'extractor_format\' instead.'''		,     category=__A		,     )
																											_A								= extractor if extractor != '''deprecated''' else extractor_format
																						else:
																											_A								= cls.extractors[extractor_format]
																						return extractor.extract(__A		,     __A   )
																	else:
																						warnings.warn(
																						    '''Parameter \'extractor_format\' was made required in version 2.4.0 and not passing it will raise an '''
																						    '''exception in 3.0.0.'''		,     category=__A		,     )
																						for extractor in cls.extractors.values():
																											if extractor.is_extractable(__A   ):
																																return extractor.extract(__A		,     __A   )











import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


__A      =				'\\n    Text data.\n    Second line of data.'

__A      =				'file'




@pytest.fixture(scope='''session''' )
def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= tmp_path_factory.mktemp('''data''' ) / (FILE_PATH + '''.zstd''')
					_A								= bytes(_lowercase					,			'''utf-8''' )
					with zstd.open(_lowercase					,			'''wb''' ) as f:
										f.write(_lowercase )
					return path




@pytest.fixture
def      __A					(     _lowercase ):

					'''simple docstring'''







					with open(os.path.join(tmpfs.local_root_dir					,			_lowercase )					,			'''w''' ) as f:
										f.write(_lowercase )
					return FILE_PATH




@pytest.mark.parametrize('''compression_format'''					,			['''gzip''', '''xz''', '''zstd'''] )
def      __A					(     _lowercase					,			_lowercase					,			_lowercase					,			_lowercase					,			_lowercase					,			_lowercase ):

					'''simple docstring'''







					_A								= {'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
					_A								= input_paths[compression_format]
					_A								= tmp_path / '''cache'''
					_A								= DownloadConfig(cache_dir=_lowercase					,			extract_compressed_file=_lowercase )
					_A								= cached_path(_lowercase					,			download_config=_lowercase )
					with open(_lowercase ) as f:
										_A								= f.read()
					with open(_lowercase ) as f:
										_A								= f.read()
					assert extracted_file_content == expected_file_content




@pytest.mark.parametrize('''default_extracted'''					,			[True, False] )
@pytest.mark.parametrize('''default_cache_dir'''					,			[True, False] )
def      __A					(     _lowercase					,			_lowercase					,			_lowercase					,			_lowercase					,			_lowercase ):

					'''simple docstring'''







					_A								= '''custom_cache'''
					_A								= '''custom_extracted_dir'''
					_A								= tmp_path / '''custom_extracted_path'''
					if default_extracted:
										_A								= ('''downloads''' if default_cache_dir else custom_cache_dir, '''extracted''')
					else:
										monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_DIR'''					,			_lowercase )
										monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''					,			str(_lowercase ) )
										_A								= custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

					_A								= xz_file
					_A								= (
					    DownloadConfig(extract_compressed_file=_lowercase )
					    if default_cache_dir
					    else DownloadConfig(cache_dir=tmp_path / custom_cache_dir					,			extract_compressed_file=_lowercase )
					)
					_A								= cached_path(_lowercase					,			download_config=_lowercase )
					assert Path(_lowercase ).parent.parts[-2:] == expected




def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= str(Path(_lowercase ).resolve() )
					assert cached_path(_lowercase ) == text_file
					# relative path
					_A								= str(Path(_lowercase ).resolve().relative_to(Path(os.getcwd() ) ) )
					assert cached_path(_lowercase ) == text_file




def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= str(tmp_path.resolve() / '''__missing_file__.txt''' )
					with pytest.raises(_lowercase ):
										cached_path(_lowercase )
					# relative path
					_A								= '''./__missing_file__.txt'''
					with pytest.raises(_lowercase ):
										cached_path(_lowercase )




def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= get_from_cache(f"""tmp://{tmpfs_file}""" )
					with open(_lowercase ) as f:
										_A								= f.read()
					assert output_file_content == FILE_CONTENT




@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,			_lowercase )
def      __A					(     ):

					'''simple docstring'''







					with pytest.raises(_lowercase ):
										cached_path('''https://huggingface.co''' )




@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,			_lowercase )
def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= tmp_path_factory.mktemp('''data''' ) / '''file.html'''
					with pytest.raises(_lowercase ):
										http_get('''https://huggingface.co'''					,			temp_file=_lowercase )
					with pytest.raises(_lowercase ):
										http_head('''https://huggingface.co''' )




@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,			_lowercase )
def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= tmp_path_factory.mktemp('''data''' ) / '''file.html'''
					with pytest.raises(_lowercase ):
										ftp_get('''ftp://huggingface.co'''					,			temp_file=_lowercase )
					with pytest.raises(_lowercase ):
										ftp_head('''ftp://huggingface.co''' )




@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,			_lowercase )
def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= tmp_path_factory.mktemp('''data''' ) / '''file.html'''
					with pytest.raises(_lowercase ):
										fsspec_get('''s3://huggingface.co'''					,			temp_file=_lowercase )
					with pytest.raises(_lowercase ):
										fsspec_head('''s3://huggingface.co''' )



