Spaces:
Paused
Paused
File size: 1,909 Bytes
f030d3a 65db57d f030d3a 65db57d f030d3a 65db57d 23e50b1 65db57d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | import os
import urllib.request as request
import zipfile
from textSummarizer.logging import logger
from textSummarizer.utils.common import get_size
from pathlib import Path
from textSummarizer.entity import DataIngestionConfig
class DataIngestion:
def __init__(self, config: DataIngestionConfig):
self.config = config
def download_data(self):
try:
if not self.config.local_data_file.exists():
filename, headers = request.urlretrieve(
url=self.config.source_URL,
filename=str(self.config.local_data_file)
)
logger.info(f"{filename} downloaded with info: {headers}")
else:
logger.info(f"File already exists of size: {get_size(self.config.local_data_file)}")
except Exception as e:
logger.error(f"Error downloading data: {e}")
raise e
def extract_zip_file(self):
"""
Extracts the zip file located at `local_data_file` to the directory `unzip_dir`.
"""
try:
unzip_path = self.config.unzip_dir
os.makedirs(unzip_path, exist_ok=True)
# Check if data already extracted (look for samsum_dataset folder)
expected_data_dir = Path(unzip_path) / "samsum_dataset"
if expected_data_dir.exists():
logger.info(f"Dataset already extracted at {expected_data_dir}. Skipping extraction.")
return
logger.info(f"Extracting {self.config.local_data_file} to {unzip_path}")
with zipfile.ZipFile(self.config.local_data_file, 'r') as zip_ref:
zip_ref.extractall(unzip_path)
logger.info("Extraction completed.")
except Exception as e:
logger.error(f"Error extracting zip file: {e}")
raise e
|