Revize 81980e82

Re #8160 new dataset

modules/crawler/pipeline.py

@@ -7 +7 @@
 import logging
 from datetime import date
 
-
 # Path to crawled data
 CRAWLED_DATA_PATH = "CrawledData/"
 # Path to processed data
@@ -19 +18 @@
 # Path to dataset processor implementations
 PROCESSOR_LIB_PATH = "DatasetProcessing."
 
-
 #logger
-logging.basicConfig(filename=CRAWLER_LOGS_PATH + 'Applicationlog-' + date.today().strftime("%b-%Y") + '.log',
-                    level=logging.INFO,
-                    format='%(asctime)s %(message)s'
-                    )
+logging.basicConfig(filename=CRAWLER_LOGS_PATH + 'Applicationlog-' +
+                               date.today().strftime("%b-%Y") + '.log',
+                               level=logging.INFO,
+                               format='%(asctime)s %(message)s')
 
 
 def check_last_update(config):
@@ -47 +45 @@
 
     if config["update-period"] <= last_update:
         logging.info("Dataset " + dataset_name + " is being updated today")
-        database_record_logs.update_updated(dataset_name,0)
+        database_record_logs.update_updated(dataset_name, 0)
         return True
     else:
         last_update_days = last_update + 1
-        logging.info("Dataset " + dataset_name + " will be updated in " + str(int(config["update-period"]) - last_update_days) + "days")
-        database_record_logs.update_updated(dataset_name,last_update + 1)
+        logging.info("Dataset " + dataset_name + " will be updated in " +
+                     str(int(config["update-period"]) - last_update_days) +
+                     "days")
+        database_record_logs.update_updated(dataset_name, last_update + 1)
         return False
 
 
-
 def crawl_data(config):
     """
       Imports dataset crawler in DatasetCrawler/"dataset_name"_crawler.py
@@ -67 +66 @@
     """
     dataset_name = config["dataset-name"]
 
-    crawl_func = __import__(CRAWLER_LIB_PATH + dataset_name + "_crawler", globals(), locals(), ['crawl']).crawl
+    crawl_func = __import__(CRAWLER_LIB_PATH + dataset_name + "_crawler",
+                            globals(), locals(), ['crawl']).crawl
     crawl_func(config)
 
     dataset_name += '/'
@@ -86 +86 @@
     dataset_name = config["dataset-name"]
     dataset_path = dataset_name + '/'
 
-    process_file_func = __import__(PROCESSOR_LIB_PATH + dataset_name + "_processor", globals(), locals(),
-                                   ['process_file']).process_file
+    process_file_func = __import__(
+        PROCESSOR_LIB_PATH + dataset_name + "_processor", globals(), locals(),
+        ['process_file']).process_file
 
     ignore_set = database_record_logs.load_ignore_set_processed(dataset_name)
-    not_processed_files = folder_processor.list_of_all_new_files(ignore_set,CRAWLED_DATA_PATH + dataset_path)
-    logging.info(dataset_name + " found " + str(len(not_processed_files)) + " not processed files")
+    not_processed_files = folder_processor.list_of_all_new_files(
+        ignore_set, CRAWLED_DATA_PATH + dataset_path)
+    logging.info(dataset_name + " found " + str(len(not_processed_files)) +
+                 " not processed files")
 
     for not_processed_file in not_processed_files:
         path = CRAWLED_DATA_PATH + dataset_path + not_processed_file
         date_dic = process_file_func(path)
         csv_utils.export_data_to_csv(path, date_dic)
-        database_record_logs.update_ignore_set_processed(dataset_name, not_processed_file)
+        database_record_logs.update_ignore_set_processed(
+            dataset_name, not_processed_file)
 
-    logging.info(dataset_name + " has processed " + str(len(not_processed_files)) + " newly crawled files")
+    logging.info(dataset_name + " has processed " +
+                 str(len(not_processed_files)) + " newly crawled files")
 
 
 def process_data_crone(config):
@@ -120 +125 @@
     dataset_name = config["dataset-name"]
     dataset_path = dataset_name + '/'
 
-    process_file_func = __import__(PROCESSOR_LIB_PATH + dataset_name + "_processor", globals(), locals(),
-                                   ['process_file']).process_file
+    process_file_func = __import__(
+        PROCESSOR_LIB_PATH + dataset_name + "_processor", globals(), locals(),
+        ['process_file']).process_file
 
     ignore_set = database_record_logs.load_ignore_set_processed(dataset_name)
-    not_processed_files = folder_processor.list_of_all_new_files(ignore_set,CRAWLED_DATA_PATH + dataset_path)
-    logging.info(dataset_name + " found " + str(len(not_processed_files)) + " not processed files")
+    not_processed_files = folder_processor.list_of_all_new_files(
+        ignore_set, CRAWLED_DATA_PATH + dataset_path)
+    logging.info(dataset_name + " found " + str(len(not_processed_files)) +
+                 " not processed files")
 
     for not_processed_file in not_processed_files:
         path = CRAWLED_DATA_PATH + dataset_path + not_processed_file
         date_dic = process_file_func(path)
         csv_utils.export_data_to_csv(path, date_dic)
-        database_record_logs.update_ignore_set_processed(dataset_name, not_processed_file)
+        database_record_logs.update_ignore_set_processed(
+            dataset_name, not_processed_file)
 
-    logging.info(dataset_name + " has processed " + str(len(not_processed_files)) + " newly crawled files")
+    logging.info(dataset_name + " has processed " +
+                 str(len(not_processed_files)) + " newly crawled files")
 
 
 def validate_process_data(config):
@@ -150 +160 @@
     """
     dataset_name = config["dataset-name"]
 
-    processed_devices_set = folder_processor.get_devices_set(dataset_name,PROCESSED_DATA_PATH +dataset_name + '/')
-    unknown_devices_set = folder_processor.get_unknown_devices_set(config, processed_devices_set)
+    processed_devices_set = folder_processor.get_devices_set(
+        dataset_name, PROCESSED_DATA_PATH + dataset_name + '/')
+    unknown_devices_set = folder_processor.get_unknown_devices_set(
+        config, processed_devices_set)
     unknown_devices_size = len(unknown_devices_set)
 
     if unknown_devices_size != 0:
-        logging.info("There is " + str(unknown_devices_size) + " unknown devices")
+        logging.info("There is " + str(unknown_devices_size) +
+                     " unknown devices")
         logging.info("Adding devices to " + dataset_name + " config file")
-        configure_functions.update_configuration(dataset_name, unknown_devices_set)
+        configure_functions.update_configuration(dataset_name,
+                                                 unknown_devices_set)
         return False
 
     for device in config["devices"]:
         device = config["devices"][device]
         if device["x"] == "UNKNOWN!" or device["y"] == "UNKNOWN!":
-            logging.info(dataset_name + " config file contains devices with UNKOWN! values please update them!!")
-            return False
+            logging.info(
+                dataset_name +
+                " config file contains devices with UNKOWN! values please update them!!"
+            )
+            #return False
 
     return True
 
@@ -184 +201 @@
 
     database_connection = database_loader.create_database_connection()
 
-    database_loader.check_or_update_datasets_collection(database_connection,config)
+    database_loader.check_or_update_datasets_collection(
+        database_connection, config)
 
     changes_in_devices = database_loader.update_devices_collection(config)
 
     if changes_in_devices == True:
-        logging.info(dataset_name + " contains changes in devices configuration. Deleting old data and preparing new")
+        logging.info(
+            dataset_name +
+            " contains changes in devices configuration. Deleting old data and preparing new"
+        )
         database_loader.reset_dataset_database(dataset_name)
 
     # get all unprocessed files from dataset
     ignore_set = database_record_logs.load_ignore_set_loaded(dataset_name)
-    not_loaded_files = folder_processor.list_of_all_new_files(ignore_set,PROCESSED_DATA_PATH + dataset_path)
+    not_loaded_files = folder_processor.list_of_all_new_files(
+        ignore_set, PROCESSED_DATA_PATH + dataset_path)
 
     # load every file
     for not_loaded_file in not_loaded_files:
         # load processed data
-        processed_data = database_loader.get_data_from_file(not_loaded_file, config)
+        processed_data = database_loader.get_data_from_file(
+            not_loaded_file, config)
         # load processed data to database
-        database_loader.load_data_to_database(database_connection, dataset_name, processed_data, not_loaded_file)
-        database_record_logs.update_ignore_set_loaded(dataset_name, not_loaded_file)
+        database_loader.load_data_to_database(database_connection,
+                                              dataset_name, processed_data,
+                                              not_loaded_file)
+        database_record_logs.update_ignore_set_loaded(dataset_name,
+                                                      not_loaded_file)
 
-    logging.info(dataset_name + " has loaded to database " + str(len(not_loaded_files)) + " newly processed files.")
+    logging.info(dataset_name + " has loaded to database " +
+                 str(len(not_loaded_files)) + " newly processed files.")
 
     client = pymongo.MongoClient()
     client.close()
@@ -231 +258 @@
 
     # get all unprocessed files from dataset
     ignore_set = database_record_logs.load_ignore_set_loaded(dataset_name)
-    not_loaded_files = folder_processor.list_of_all_new_files(ignore_set,PROCESSED_DATA_PATH + dataset_path)
+    not_loaded_files = folder_processor.list_of_all_new_files(
+        ignore_set, PROCESSED_DATA_PATH + dataset_path)
 
     # load every file
     for not_loaded_file in not_loaded_files:
         # load processed data
-        processed_data = database_loader.get_data_from_file(not_loaded_file, config)
+        processed_data = database_loader.get_data_from_file(
+            not_loaded_file, config)
         # load processed data to database
-        database_loader.load_data_to_database(database_connection, dataset_name, processed_data, not_loaded_file)
-        database_record_logs.update_ignore_set_loaded(dataset_name, not_loaded_file)
+        database_loader.load_data_to_database(database_connection,
+                                              dataset_name, processed_data,
+                                              not_loaded_file)
+        database_record_logs.update_ignore_set_loaded(dataset_name,
+                                                      not_loaded_file)
 
-    logging.info(dataset_name + " has loaded to database " + str(len(not_loaded_files)) + " newly processed files.")
+    logging.info(dataset_name + " has loaded to database " +
+                 str(len(not_loaded_files)) + " newly processed files.")
 
     client = pymongo.MongoClient()
     client.close()
 
 
-
 def run_full_pipeline(dataset_name):
     """
     Loads config file and starts full pipeline
@@ -259 +291 @@
         dataset_name: name of dataset that has existing configuration file
     """
     logging.info("Starting pipeline for dataset " + dataset_name)
-    print("Zpracovávám dataset " + dataset_name + " průběh lze sledovat v logu umístěném v in CrawlerLogs folder")
-    
+    print("Zpracovávám dataset " + dataset_name +
+          " průběh lze sledovat v logu umístěném v in CrawlerLogs folder")
+
     config = configure_functions.load_configuration(dataset_name)
     crawl_data(config)
     process_data(config)
@@ -293 +326 @@
 
         if validation_test:
             load_data_to_database_crone(config)
-