



'''simple docstring'''


import os
import zipfile

import pytest

from datasets.utils.extract import (
    BzipaExtractor,
    Extractor,
    GzipExtractor,
    LzaExtractor,
    SevenZipExtractor,
    TarExtractor,
    XzExtractor,
    ZipExtractor,
    ZstdExtractor,
)

from .utils import require_lza, require_pyazr, require_zstandard







@pytest.mark.parametrize(
    'compression_format, is_archive'    ,						[
        ('7z', True),
        ('bz2', False),
        ('gzip', False),
        ('lz4', False),
        ('tar', True),
        ('xz', False),
        ('zip', True),
        ('zstd', False),
    ]    ,						)
def 				__lowerCamelCase				(   A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						)       ->	str:






   """simple docstring"""
   UpperCamelCase					=  {
       '7z': (seven_zip_file, SevenZipExtractor),
       'bz2': (bza_file, BzipaExtractor),
       'gzip': (gz_file, GzipExtractor),
       'lz4': (lza_file, LzaExtractor),
       'tar': (tar_file, TarExtractor),
       'xz': (xz_file, XzExtractor),
       'zip': (zip_file, ZipExtractor),
       'zstd': (zstd_file, ZstdExtractor),
   }
   UpperCamelCase					=  input_paths_and_base_extractors[compression_format]
   if input_path is None:
      UpperCamelCase					=  F"""for '{compression_format}' compression_format, """
      if compression_format == "7z":
         reason += require_pyazr.kwargs["reason"]
      elif compression_format == "lz4":
         reason += require_lza.kwargs["reason"]
      elif compression_format == "zstd":
         reason += require_zstandard.kwargs["reason"]
      pytest.skip(snake_case__    )
   assert base_extractor.is_extractable(snake_case__    )
   UpperCamelCase					=  tmp_path / ('extracted' if is_archive else 'extracted.txt')
   base_extractor.extract(snake_case__    ,						snake_case__    )
   if is_archive:
      assert output_path.is_dir()
      for file_path in output_path.iterdir():
         assert file_path.name == text_file.name
         UpperCamelCase					=  file_path.read_text(encoding='utf-8'    )
   else:
      UpperCamelCase					=  output_path.read_text(encoding='utf-8'    )
   UpperCamelCase					=  text_file.read_text(encoding='utf-8'    )
   assert extracted_file_content == expected_file_content







@pytest.mark.parametrize(
    'compression_format, is_archive'    ,						[
        ('7z', True),
        ('bz2', False),
        ('gzip', False),
        ('lz4', False),
        ('tar', True),
        ('xz', False),
        ('zip', True),
        ('zstd', False),
    ]    ,						)
def 				__lowerCamelCase				(   A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						)       ->	List[Any]:






   """simple docstring"""
   UpperCamelCase					=  {
       '7z': seven_zip_file,
       'bz2': bza_file,
       'gzip': gz_file,
       'lz4': lza_file,
       'tar': tar_file,
       'xz': xz_file,
       'zip': zip_file,
       'zstd': zstd_file,
   }
   UpperCamelCase					=  input_paths[compression_format]
   if input_path is None:
      UpperCamelCase					=  F"""for '{compression_format}' compression_format, """
      if compression_format == "7z":
         reason += require_pyazr.kwargs["reason"]
      elif compression_format == "lz4":
         reason += require_lza.kwargs["reason"]
      elif compression_format == "zstd":
         reason += require_zstandard.kwargs["reason"]
      pytest.skip(snake_case__    )
   UpperCamelCase					=  Extractor.infer_extractor_format(snake_case__    )
   assert extractor_format is not None
   UpperCamelCase					=  tmp_path / ('extracted' if is_archive else 'extracted.txt')
   Extractor.extract(snake_case__    ,						snake_case__    ,						snake_case__    )
   if is_archive:
      assert output_path.is_dir()
      for file_path in output_path.iterdir():
         assert file_path.name == text_file.name
         UpperCamelCase					=  file_path.read_text(encoding='utf-8'    )
   else:
      UpperCamelCase					=  output_path.read_text(encoding='utf-8'    )
   UpperCamelCase					=  text_file.read_text(encoding='utf-8'    )
   assert extracted_file_content == expected_file_content







@pytest.fixture
def 				__lowerCamelCase				(   A__    ,						A__    )       ->	List[str]:






   """simple docstring"""
   import tarfile

   UpperCamelCase					=  tmp_path / 'data_dot_dot'
   directory.mkdir()
   UpperCamelCase					=  directory / 'tar_file_with_dot_dot.tar'
   with tarfile.TarFile(snake_case__    ,						'w'    ) as f:
      f.add(snake_case__    ,						arcname=os.path.join('..'    ,						text_file.name    )    )
   return path







@pytest.fixture
def 				__lowerCamelCase				(   A__    )       ->	Tuple:






   """simple docstring"""
   import tarfile

   UpperCamelCase					=  tmp_path / 'data_sym_link'
   directory.mkdir()
   UpperCamelCase					=  directory / 'tar_file_with_sym_link.tar'
   os.symlink('..'    ,						directory / 'subdir'    ,						target_is_directory=snake_case__    )
   with tarfile.TarFile(snake_case__    ,						'w'    ) as f:
      f.add(str(directory / 'subdir'    )    ,						arcname='subdir'    )  # str required by os.readlink on Windows and Python < 3.8
   return path







@pytest.mark.parametrize(
    'insecure_tar_file, error_log'    ,						[('tar_file_with_dot_dot', 'illegal path'), ('tar_file_with_sym_link', 'Symlink')]    ,						)
def 				__lowerCamelCase				(   A__    ,						A__    ,						A__    ,						A__    ,						A__    ,						A__    )       ->	Tuple:






   """simple docstring"""
   UpperCamelCase					=  {
       'tar_file_with_dot_dot': tar_file_with_dot_dot,
       'tar_file_with_sym_link': tar_file_with_sym_link,
   }
   UpperCamelCase					=  insecure_tar_files[insecure_tar_file]
   UpperCamelCase					=  tmp_path / 'extracted'
   TarExtractor.extract(snake_case__    ,						snake_case__    )
   assert caplog.text
   for record in caplog.records:
      assert record.levelname == "ERROR"
      assert error_log in record.msg







def 				__lowerCamelCase				(   A__    )       ->	str:






   """simple docstring"""
   # We should have less false positives than zipfile.is_zipfile
   # We do that by checking only the magic number
   UpperCamelCase					=  tmpdir / 'not_a_zip_file'
   # From: https://github.com/python/cpython/pull/5053
   UpperCamelCase					=  (
       B'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00'
       B'\x00\x02\x08\x06\x00\x00\x00\x99\x81\xb6\'\x00\x00\x00\x15I'
       B'DATx\x01\x01\n\x00\xf5\xff\x00PK\x05\x06\x00PK\x06\x06\x07'
       B'\xac\x01N\xc6|a\r\x00\x00\x00\x00IEND\xaeB`\x82'
   )
   with not_a_zip_file.open('wb'    ) as f:
      f.write(snake_case__    )
   assert zipfile.is_zipfile(str(snake_case__    )    )  # is a false positive for `zipfile`
   assert not ZipExtractor.is_extractable(snake_case__    )  # but we're right






from typing import Optional

import pyspark

from .. import Features, NamedSplit
from ..download import DownloadMode
from ..packaged_modules.spark.spark import Spark
from .abc import AbstractDatasetReader



class 				lowerCAmelCase_		(  a__							):
       def __init__(			self,			SCREAMING_SNAKE_CASE_,			SCREAMING_SNAKE_CASE_ = None,			SCREAMING_SNAKE_CASE_ = None,			SCREAMING_SNAKE_CASE_ = True,			SCREAMING_SNAKE_CASE_ = None,			SCREAMING_SNAKE_CASE_ = False,			SCREAMING_SNAKE_CASE_ = None,			SCREAMING_SNAKE_CASE_ = True,			SCREAMING_SNAKE_CASE_ = "arrow",			**SCREAMING_SNAKE_CASE_,			)     ->		Optional[int]:
          super().__init__(
              split=SCREAMING_SNAKE_CASE_,			features=SCREAMING_SNAKE_CASE_,			cache_dir=SCREAMING_SNAKE_CASE_,			keep_in_memory=SCREAMING_SNAKE_CASE_,			streaming=SCREAMING_SNAKE_CASE_,			**SCREAMING_SNAKE_CASE_,			)
          UpperCamelCase    :  List[str]							=	load_from_cache_file
          UpperCamelCase    :  List[str]							=	file_format
          UpperCamelCase    :  Optional[int]							=	Spark(
              df=SCREAMING_SNAKE_CASE_,			features=SCREAMING_SNAKE_CASE_,			cache_dir=SCREAMING_SNAKE_CASE_,			working_dir=SCREAMING_SNAKE_CASE_,			**SCREAMING_SNAKE_CASE_,			)






       def 					snake_case_       (			self       )     ->		int:
          if self.streaming:
             return self.builder.as_streaming_dataset(split=self.split       )
          UpperCamelCase    :  Union[str, Any]							=	None if self._load_from_cache_file else DownloadMode.FORCE_REDOWNLOAD
          self.builder.download_and_prepare(
              download_mode=SCREAMING_SNAKE_CASE_,			file_format=self._file_format,			)
          return self.builder.as_dataset(split=self.split       )



