from torchtext.data.datasets_utils import (
    _RawTextIterableDataset,
    _wrap_split_argument,
    _add_docstring_header,
    _download_extract_validate,
    _create_dataset_directory,
    _create_data_from_csv,
)
import os
import logging

URL = 'https://drive.google.com/uc?export=download&id=0Bz8a_Dbh9QhbZVhsUnRWRDhETzA'

MD5 = '57d28bd5d930e772930baddf36641c7c'

NUM_LINES = {
    'train': 3000000,
    'test': 650000,
}

_PATH = 'amazon_review_full_csv.tar.gz'

_EXTRACTED_FILES = {
    'train': f'{os.sep}'.join(['amazon_review_full_csv', 'train.csv']),
    'test': f'{os.sep}'.join(['amazon_review_full_csv', 'test.csv']),
}

_EXTRACTED_FILES_MD5 = {
    'train': "31b268b09fd794e0ca5a1f59a0358677",
    'test': "0f1e78ab60f625f2a30eab6810ef987c"
}

DATASET_NAME = "AmazonReviewFull"


@_add_docstring_header(num_lines=NUM_LINES, num_classes=5)
@_create_dataset_directory(dataset_name=DATASET_NAME)
@_wrap_split_argument(('train', 'test'))
def AmazonReviewFull(root, split):
    path = _download_extract_validate(root, URL, MD5, os.path.join(root, _PATH), os.path.join(root, _EXTRACTED_FILES[split]),
                                      _EXTRACTED_FILES_MD5[split], hash_type="md5")
    logging.info('Creating {} data'.format(split))
    return _RawTextIterableDataset(DATASET_NAME, NUM_LINES[split],
                                   _create_data_from_csv(path))

@@ -1,12 +1,231 @@
+from torchtext._internal.module_utils import is_module_available%0Afrom typing import Union, Tuple%0A%0Aif is_module_available(%22torchdata%22):%0A    from torchdata.datapipes.iter import FileOpener, GDriveReader, IterableWrapper%0A%0A
 from torchte
@@ -260,37 +260,8 @@
 t (%0A
-    _RawTextIterableDataset,%0A
     
@@ -313,40 +313,8 @@
 er,%0A
-    _download_extract_validate,%0A
     
@@ -344,37 +344,11 @@
 ry,%0A
-    _create_data_from_csv,%0A
 )%0A
+%0A
 impo
@@ -356,23 +356,8 @@
 t os
-%0Aimport logging
 %0A%0AUR
@@ -1047,23 +1047,23 @@
 nt((
-'
+%22
 train
-', '
+%22, %22
 test
-'
+%22
 ))%0Ad
@@ -1090,406 +1090,805 @@
 root
+: str
 , split
-):%0A    path = _download_extract_validate(root, URL, MD5, os.path.join(root, _PATH), os.path.join(root, _EXTRACTED_FILES%5Bsplit%5D),
+: Union%5BTuple%5Bstr%5D, str%5D):%0A    if not is_module_available(%22torchdata%22):%0A        raise ModuleNotFoundError(%22Package %60torchdata%60 not found. Please install following instructions at %60https://github.com/pytorch/data%60%22)%0A%0A    url_dp = IterableWrapper(%5BURL%5D)%0A%0A    cache_dp = url_dp.on_disk_cache(%0A        filepath_fn=lambda x: os.path.join(root, _PATH),%0A        hash_dict=%7Bos.path.join(root, _PATH): MD5%7D, hash_type=%22md5%22
 %0A    
+)%0A
     
-                              _EXTRACTED_FILES_MD5%5Bsplit%5D, hash_type=%22md5
+cache_dp = GDriveReader(cache_dp).end_caching(mode=%22wb%22, same_filepath_fn=True)%0A    cache_dp = FileOpener(cache_dp, mode=%22b
 %22)%0A
+%0A
     
-logging.info('Creating %7B%7D data'.format(split))%0A    return _RawTextIterableDataset(DATASET_NAME, NUM_LINES%5Bsplit%5D,%0A                                   _create_data_from_csv(path
+extracted_files = cache_dp.read_from_tar()%0A%0A    filter_extracted_files = extracted_files.filter(lambda x: _EXTRACTED_FILES%5Bsplit%5D in x%5B0%5D)%0A%0A    return filter_extracted_files.parse_csv().map(fn=lambda t: (int(t%5B0%5D), %22 %22.join(t%5B1:%5D)
 ))%0A

add initial pass at migrating Amazon Review Full to datapipes. (#1499)