






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   snake_case (UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   )	->    int:
  UpperCamelCase_: Dict   				=    {
      '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_				,UpperCamelCase_: Union[str, Any]   				=    input_paths_and_base_extractors[compression_format]
  if input_path is None:
    UpperCamelCase_: Union[str, Any]   				=    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(UpperCAmelCase__  )
  assert base_extractor.is_extractable(UpperCAmelCase__  )
  UpperCamelCase_: List[Any]   				=    tmp_path / ('extracted' if is_archive else 'extracted.txt')
  base_extractor.extract(UpperCAmelCase__     ,   UpperCAmelCase__  )
  if is_archive:
    assert output_path.is_dir()
    for file_path in output_path.iterdir():
      assert file_path.name == text_file.name
      UpperCamelCase_: Any   				=    file_path.read_text(encoding='utf-8'  )
  else:
    UpperCamelCase_: Union[str, Any]   				=    output_path.read_text(encoding='utf-8'  )
  UpperCamelCase_: Optional[Any]   				=    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   snake_case (UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   )	->    str:
  UpperCamelCase_: Optional[int]   				=    {
      '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_: List[str]   				=    input_paths[compression_format]
  if input_path is None:
    UpperCamelCase_: str   				=    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(UpperCAmelCase__  )
  UpperCamelCase_: int   				=    Extractor.infer_extractor_format(UpperCAmelCase__  )
  assert extractor_format is not None
  UpperCamelCase_: Tuple   				=    tmp_path / ('extracted' if is_archive else 'extracted.txt')
  Extractor.extract(UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__  )
  if is_archive:
    assert output_path.is_dir()
    for file_path in output_path.iterdir():
      assert file_path.name == text_file.name
      UpperCamelCase_: Union[str, Any]   				=    file_path.read_text(encoding='utf-8'  )
  else:
    UpperCamelCase_: Optional[int]   				=    output_path.read_text(encoding='utf-8'  )
  UpperCamelCase_: List[Any]   				=    text_file.read_text(encoding='utf-8'  )
  assert extracted_file_content == expected_file_content

@pytest.fixture
def   snake_case (UpperCAmelCase__     ,   UpperCAmelCase__  )	->    int:
  import tarfile

  UpperCamelCase_: List[str]   				=    tmp_path / 'data_dot_dot'
  directory.mkdir()
  UpperCamelCase_: Dict   				=    directory / 'tar_file_with_dot_dot.tar'
  with tarfile.TarFile(UpperCAmelCase__     ,   'w'  ) as f:
    f.add(UpperCAmelCase__     ,   arcname=os.path.join('..'     ,   text_file.name  )  )
  return path

@pytest.fixture
def   snake_case (UpperCAmelCase__  )	->    str:
  import tarfile

  UpperCamelCase_: List[str]   				=    tmp_path / 'data_sym_link'
  directory.mkdir()
  UpperCamelCase_: Tuple   				=    directory / 'tar_file_with_sym_link.tar'
  os.symlink('..'     ,   directory / 'subdir'     ,   target_is_directory=UpperCAmelCase__  )
  with tarfile.TarFile(UpperCAmelCase__     ,   '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   snake_case (UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__  )	->    Optional[Any]:
  UpperCamelCase_: Optional[Any]   				=    {
      'tar_file_with_dot_dot': tar_file_with_dot_dot,
      'tar_file_with_sym_link': tar_file_with_sym_link,
  }
  UpperCamelCase_: Any   				=    insecure_tar_files[insecure_tar_file]
  UpperCamelCase_: List[str]   				=    tmp_path / 'extracted'
  TarExtractor.extract(UpperCAmelCase__     ,   UpperCAmelCase__  )
  assert caplog.text
  for record in caplog.records:
    assert record.levelname == "ERROR"
    assert error_log in record.msg

def   snake_case (UpperCAmelCase__  )	->    Union[str, Any]:
  # We should have less false positives than zipfile.is_zipfile
  # We do that by checking only the magic number
  UpperCamelCase_: Optional[Any]   				=    tmpdir / 'not_a_zip_file'
  # From: https://github.com/python/cpython/pull/5053
  UpperCamelCase_: Union[str, Any]   				=    (
      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(UpperCAmelCase__  )
  assert zipfile.is_zipfile(str(UpperCAmelCase__  )  )  # is a false positive for `zipfile`
  assert not ZipExtractor.is_extractable(UpperCAmelCase__  )  # but we're right







from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


A_		:		Tuple		     =			{
    'configuration_funnel': ['FUNNEL_PRETRAINED_CONFIG_ARCHIVE_MAP', 'FunnelConfig'],
    'convert_funnel_original_tf_checkpoint_to_pytorch': [],
    'tokenization_funnel': ['FunnelTokenizer'],
}

try:
   if not is_tokenizers_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   A_		:		List[str]		     =			['FunnelTokenizerFast']

try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   A_		:		List[str]		     =			[
       'FUNNEL_PRETRAINED_MODEL_ARCHIVE_LIST',
       'FunnelBaseModel',
       'FunnelForMaskedLM',
       'FunnelForMultipleChoice',
       'FunnelForPreTraining',
       'FunnelForQuestionAnswering',
       'FunnelForSequenceClassification',
       'FunnelForTokenClassification',
       'FunnelModel',
       'FunnelPreTrainedModel',
       'load_tf_weights_in_funnel',
   ]

try:
   if not is_tf_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   A_		:		str		     =			[
       'TF_FUNNEL_PRETRAINED_MODEL_ARCHIVE_LIST',
       'TFFunnelBaseModel',
       'TFFunnelForMaskedLM',
       'TFFunnelForMultipleChoice',
       'TFFunnelForPreTraining',
       'TFFunnelForQuestionAnswering',
       'TFFunnelForSequenceClassification',
       'TFFunnelForTokenClassification',
       'TFFunnelModel',
       'TFFunnelPreTrainedModel',
   ]


if TYPE_CHECKING:
   from .configuration_funnel import FUNNEL_PRETRAINED_CONFIG_ARCHIVE_MAP, FunnelConfig
   from .tokenization_funnel import FunnelTokenizer

   try:
      if not is_tokenizers_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .tokenization_funnel_fast import FunnelTokenizerFast

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_funnel import (
          FUNNEL_PRETRAINED_MODEL_ARCHIVE_LIST,
          FunnelBaseModel,
          FunnelForMaskedLM,
          FunnelForMultipleChoice,
          FunnelForPreTraining,
          FunnelForQuestionAnswering,
          FunnelForSequenceClassification,
          FunnelForTokenClassification,
          FunnelModel,
          FunnelPreTrainedModel,
          load_tf_weights_in_funnel,
      )

   try:
      if not is_tf_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_tf_funnel import (
          TF_FUNNEL_PRETRAINED_MODEL_ARCHIVE_LIST,
          TFFunnelBaseModel,
          TFFunnelForMaskedLM,
          TFFunnelForMultipleChoice,
          TFFunnelForPreTraining,
          TFFunnelForQuestionAnswering,
          TFFunnelForSequenceClassification,
          TFFunnelForTokenClassification,
          TFFunnelModel,
          TFFunnelPreTrainedModel,
      )

else:
   import sys

   A_		:		Any		     =			_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)