Merge pull request #24 from OpenIsraeliSupermarkets/v0.1.8

V0.1.8

Author: erlichsefi

il_supermarket_parsers/documents/xml_dataframe_parser.py

@@ -12,6 +12,8 @@ class XmlDataFrameConverter(BaseXMLParser):
 
     def reduce_size(self, data):
         """reduce the size"""
+        data = data.fillna("")
+        # remove duplicate columns
         for col in data.columns:
             data[col] = data[col].mask(data[col] == data[col].shift())
         return data

il_supermarket_parsers/multiprocess_pharser.py

@@ -1,6 +1,8 @@
 import itertools
 import json
+import datetime
 import os
+import pytz
 from .raw_parsing_pipeline import RawParsingPipeline
 from .utils.multi_processing import MultiProcessor, ProcessJob
 from .parser_factory import ParserFactory
@@ -20,9 +22,10 @@ def job(self, **kwargs):
         parser_name = kwargs.pop("store_enum")
         output_folder = kwargs.pop("output_folder")
         limit = kwargs.pop("limit")
+        when_date = kwargs.pop("when_date")
 
         return RawParsingPipeline(
-            drop_folder, parser_name, file_type, output_folder
+            drop_folder, parser_name, file_type, output_folder, when_date
         ).process(limit=limit)
 
 
@@ -36,12 +39,14 @@ def __init__(
         enabled_file_types=None,
         multiprocessing=6,
         output_folder="output",
+        when_date=datetime.datetime.now(pytz.timezone("Asia/Jerusalem")),
     ):
         super().__init__(multiprocessing=multiprocessing)
         self.data_folder = data_folder
         self.enabled_parsers = enabled_parsers
         self.enabled_file_types = enabled_file_types
         self.output_folder = output_folder
+        self.when_date = when_date
 
     def task_to_execute(self):
         """the task to execute"""
@@ -67,6 +72,7 @@ def get_arguments_list(self, limit=None):
             "file_type",
             "data_folder",
             "output_folder",
+            "when_date",
         ]
         combinations = list(
             itertools.product(
@@ -75,6 +81,7 @@ def get_arguments_list(self, limit=None):
                 all_file_types,
                 [self.data_folder],
                 [self.output_folder],
+                [self.when_date.strftime("%Y-%m-%d %H:%M:%S %z")]
             )
         )
         task_can_executed_independently = [
@@ -84,10 +91,15 @@ def get_arguments_list(self, limit=None):
 
     def post(self, results):
         """post process the results"""
-        with open(
-            os.path.join(self.output_folder, "parser-status.json"),
-            "w",
-            encoding="utf-8",
-        ) as file:
-            json.dump(results, file)
+        status_file = os.path.join(self.output_folder, "parser-status.json")
+        if os.path.exists(status_file):
+            with open(status_file, "r", encoding="utf-8") as file:
+                existing_results = json.load(file)
+        else:
+            existing_results = []
+
+        existing_results.extend(results)
+
+        with open(status_file, "w", encoding="utf-8") as file:
+            json.dump(existing_results, file)
         return super().post(results)

il_supermarket_parsers/raw_parsing_pipeline.py

@@ -14,11 +14,12 @@ class RawParsingPipeline:
     processing files to dataframe
     """
 
-    def __init__(self, folder, store_name, file_type, output_folder) -> None:
+    def __init__(self, folder, store_name, file_type, output_folder, when_date) -> None:
         self.store_name = store_name
         self.file_type = file_type
         self.folder = folder
         self.output_folder = output_folder
+        self.when_date = when_date
 
     def append_columns_to_csv(self, existing_file, new_columns):
         """Append new columns to an existing CSV file"""
@@ -125,6 +126,7 @@ def process(self, limit=None):
             "status": True,
             "store_name": self.store_name,
             "files_types": self.file_type,
+            "when_date": self.when_date,
             "processed_files": len(files_to_process) > 0,
             "execution_errors": execution_errors > 0,
             "file_was_created": os.path.exists(create_csv),

il_supermarket_parsers/task.py

@@ -1,7 +1,10 @@
+import datetime
+import pytz
 from .multiprocess_pharser import ParallelParser
 from .utils.logger import Logger
 
 
+
 class ConvertingTask:
     """main convert task"""
 
@@ -12,6 +15,7 @@ def __init__(
         files_types=None,
         multiprocessing=6,
         limit=None,
+        when_date=datetime.datetime.now(pytz.timezone("Asia/Jerusalem")),
         output_folder="outputs",
     ):
         Logger.info(
@@ -21,13 +25,15 @@ def __init__(
             f"files_types = {files_types}"
             f"output_folder={output_folder}"
             f"limit={limit}"
+            f"when_date={when_date}"
         )
         self.runner = ParallelParser(
             data_folder,
             enabled_parsers=enabled_parsers,
             enabled_file_types=files_types,
             multiprocessing=multiprocessing,
             output_folder=output_folder,
+            when_date=when_date,
         )
         self.limit = limit
 

setup.py

@@ -31,7 +31,7 @@
     tests_require=dev_required,
     extras_require={"test": ["pytest", "pytest-xdist"]},
     # *strongly* suggested for sharing
-    version="0.1.7",
+    version="0.1.8",
     # The license can be anything you like
     license="MIT",
     description="python package that process the data dumped by the israeli supermarket",