data ingestion added

.gitignore

@@ -160,3 +160,4 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+artifacts/*

config/config.yaml

@@ -0,0 +1,8 @@
+artifacts_root: artifacts
+
+
+data_ingestion:
+  root_dir: artifacts/data_ingestion
+  source_URL: https://github.com/entbappy/Branching-tutorial/raw/master/summarizer-data.zip
+  local_data_file: artifacts/data_ingestion/data.zip
+  unzip_dir: artifacts/data_ingestion

main.py

@@ -0,0 +1,12 @@
+from textsummarizer.pipeline.stage_01_data_ingestion import DataIngestionPipeline
+from textsummarizer.logging import logger
+
+STAGE_NAME = "Data Ingestion stage"
+try:
+   logger.info(f">>>>>> stage {STAGE_NAME} started <<<<<<") 
+   data_ingestion = DataIngestionPipeline()
+   data_ingestion.main()
+   logger.info(f">>>>>> stage {STAGE_NAME} completed <<<<<<\n\nx==========x")
+except Exception as e:
+        logger.exception(e)
+        raise e

params.yaml

@@ -0,0 +1 @@
+key : val

src/textsummarizer/config/configuration.py

@@ -0,0 +1,31 @@
+from textsummarizer.constants import *
+from textsummarizer.utils.common import read_yaml, create_directories
+from textsummarizer.entity.config_entity import DataIngestionConfig
+
+class ConfigurationManager:
+    def __init__(
+        self,
+        config_filepath = CONFIG_FILE_PATH,
+        params_filepath = PARAMS_FILE_PATH):
+
+        self.config = read_yaml(config_filepath)
+        self.params = read_yaml(params_filepath)
+
+        create_directories([self.config.artifacts_root])
+
+    
+
+    def get_data_ingestion_config(self) -> DataIngestionConfig:
+        config = self.config.data_ingestion
+
+        create_directories([config.root_dir])
+
+        data_ingestion_config = DataIngestionConfig(
+            root_dir=config.root_dir,
+            source_URL=config.source_URL,
+            local_data_file=config.local_data_file,
+            unzip_dir=config.unzip_dir 
+        )
+
+        return data_ingestion_config
+        

src/textsummarizer/conponents/data_ingestion.py

@@ -0,0 +1,32 @@
+import os
+import urllib.request as request
+import zipfile
+from textsummarizer.logging import logger
+from textsummarizer.utils.common import get_size
+from textsummarizer.entity.config_entity import DataIngestionConfig
+from pathlib import Path
+
+
+class DataIngestion:
+    def __init__(self, config: DataIngestionConfig):
+        self.config = config
+        
+        
+    def download_file(self):
+        if not os.path.exists(self.config.local_data_file):
+            filename, headers = request.urlretrieve(
+                url=self.config.source_URL,
+                filename=self.config.local_data_file
+            )
+            logger.info(f'Download! with following info: \n{headers}')
+        else:
+            logger.info(f'File already exist of size: {get_size(Path(self.config.local_data_file))}')
+            
+            
+            
+    def extract_file(self):
+        unzip_dir = self.config.unzip_dir
+        os.makedirs(unzip_dir, exist_ok=True)
+        with zipfile.ZipFile(self.config.local_data_file, 'r') as file :
+            file.extractall(unzip_dir)   
+    

src/textsummarizer/constants/__init__.py

@@ -0,0 +1,4 @@
+from pathlib import  Path
+
+CONFIG_FILE_PATH = Path('config/config.yaml')
+PARAMS_FILE_PATH = Path('params.yaml')

src/textsummarizer/entity/config_entity.py

@@ -0,0 +1,11 @@
+from dataclasses import dataclass
+from pathlib import  Path
+
+@dataclass(frozen=True)
+class DataIngestionConfig:
+    root_dir : Path
+    source_URL : str
+    local_data_file : Path
+    unzip_dir : Path
+    
+    

src/textsummarizer/logging/__init__.py

@@ -0,0 +1,22 @@
+import os
+import sys
+import logging
+
+logging_str = "[%(asctime)s: %(levelname)s: %(module)s: %(message)s]"
+log_dir = "logs"
+log_filepath = os.path.join(log_dir,"running_logs.log")
+os.makedirs(log_dir, exist_ok=True)
+
+
+
+logging.basicConfig(
+    level= logging.INFO,
+    format= logging_str,
+
+    handlers=[
+        logging.FileHandler(log_filepath),
+        logging.StreamHandler(sys.stdout)
+    ]
+)
+
+logger = logging.getLogger("textsummarizerLogger")

src/textsummarizer/pipeline/stage_01_data_ingestion.py

@@ -0,0 +1,17 @@
+from textsummarizer.config.configuration import ConfigurationManager
+from textsummarizer.entity.config_entity import DataIngestionConfig
+from textsummarizer.conponents.data_ingestion import DataIngestion
+
+
+class DataIngestionPipeline:
+    def __init__(self):
+        pass
+    
+    def main(self):
+        config = ConfigurationManager()
+        data_ingestion_config = config.get_data_ingestion_config()
+        data_ingestion = DataIngestion(config=data_ingestion_config)
+        data_ingestion.download_file()
+        data_ingestion.extract_file()
+        
+        

src/textsummarizer/utils/common.py

@@ -0,0 +1,66 @@
+import os
+from box.exceptions import BoxValueError
+import yaml
+from textsummarizer.logging import logger
+from ensure import ensure_annotations
+from box import ConfigBox
+from pathlib import Path
+from typing import Any
+
+
+
+@ensure_annotations
+def read_yaml(path_to_yaml: Path) -> ConfigBox:
+    """reads yaml file and returns
+
+    Args:
+        path_to_yaml (str): path like input
+
+    Raises:
+        ValueError: if yaml file is empty
+        e: empty file
+
+    Returns:
+        ConfigBox: ConfigBox type
+    """
+    try:
+        with open(path_to_yaml) as yaml_file:
+            content = yaml.safe_load(yaml_file)
+            logger.info(f"yaml file: {path_to_yaml} loaded successfully")
+            return ConfigBox(content)
+    except BoxValueError:
+        raise ValueError("yaml file is empty")
+    except Exception as e:
+        raise e
+    
+
+
+@ensure_annotations
+def create_directories(path_to_directories: list, verbose=True):
+    """create list of directories
+
+    Args:
+        path_to_directories (list): list of path of directories
+        ignore_log (bool, optional): ignore if multiple dirs is to be created. Defaults to False.
+    """
+    for path in path_to_directories:
+        os.makedirs(path, exist_ok=True)
+        if verbose:
+            logger.info(f"created directory at: {path}")
+
+
+
+@ensure_annotations
+def get_size(path: Path) -> str:
+    """get size in KB
+
+    Args:
+        path (Path): path of the file
+
+    Returns:
+        str: size in KB
+    """
+    size_in_kb = round(os.path.getsize(path)/1024)
+    return f"~ {size_in_kb} KB"
+
+