Feat: Check if item version is already preserved before bagging (Issue #102)

.env.sample.ini

@@ -5,6 +5,15 @@ retries = 3
 retries_wait = 10
 institution = 1077
 
+[aptrust_api]
+url =
+user =
+token =
+items_per_page =
+alt_identifier_starts_with =
+retries = 3
+retries_wait = 10
+
 [system]
 preservation_storage_location = 
 logs_location = 

Config.py

@@ -9,6 +9,9 @@ def __init__(self, fileName):
     def figshare_config(self):
         return self.config['figshare_api']
 
+    def aptrust_config(self):
+        return self.config['aptrust_api']
+
     def system_config(self):
         return self.config['system']
 

README.md

@@ -22,14 +22,23 @@ ReBACH is run via the command line as outlined in the 'How to Run' section of th
 ## How to run:
 - Copy the .env.sample.ini file and give it a name of your choice (e.g. .env.ini).
 - Fill out the .env.ini file (IMPORTANT: Make sure not to commit this file to Github)
-    - url - required: The figshare API url
-    - token - required: Your auth token to your organization's API
-    - retries - required: Number of times the script should retry API or file system calls if it is unable to connect. Defaults to 3
-    - retries_wait - required: Number of seconds the script should wait between call retries if it is unable to connect. Defaults to 10
-    - institution - required: The Figshare Institution ID for your organization
+    - figshare_api
+	    - url - required: The figshare API url
+	    - token - required: Your auth token to your organization's API
+	    - retries - required: Number of times the script should retry API or file system calls if it is unable to connect. Defaults to 3
+	    - retries_wait - required: Number of seconds the script should wait between call retries if it is unable to connect. Defaults to 10
+	    - institution - required: The Figshare Institution ID for your organization
+    - aptrust_api
+	    - url - required: The AP Trust API url (identifier__starts_with filter must be included in the url)
+		- user - required: Your user email address on AP Trust
+		- token - required: Your user secret token on AP Trust
+        - items_per_page - Maximum number of object to be return per page by the API
+        - alt_identifier_starts_with - Prefix for alternate identifier in AP Trust 
+		- retries - required: Number of times the script should retry API or file system calls if it is unable to connect. Defaults to 3
+		- retries_wait - required: Number of seconds the script should wait between call retries if it is unable to connect. Defaults to 10
     - preservation_storage_location - required: The file system location where the preservation folders/packages should be created
     - logs_location - required: The file system location where logs should be created. This value will override the one in `bagger/config/default.toml` when bagger is used for post-processing (see post_process_script_command setting below).
-    - additional_precentage_required - required: How much extra space the preservation storage location should have in order to handle files as a percent. This percent is applied to the total storage needed for all files. I.e. if the value of this field is 10 and the amount of storage needed for files is 1 GB, the script will make sure that the preservation storage location has at least 1.1 GB free. Defaults to 10
+    - additional_percentage_required - required: How much extra space the preservation storage location should have in order to handle files as a percent. This percent is applied to the total storage needed for all files. I.e. if the value of this field is 10 and the amount of storage needed for files is 1 GB, the script will make sure that the preservation storage location has at least 1.1 GB free. Defaults to 10
     - pre_process_script_command - optional: The terminal command (including arguments) to invoke a script to be run BEFORE the files are copied and logic applied to the preservation storage (note: this action is not currently implemented)
     - post_process_script_command - required: Specifies the method of performing post-processing steps. This can take only two values: the string 'Bagger', or the path to an external script. If the value is set to 'Bagger', the post-processing steps will consist of running the internal `bagger` module. If the value is set to a path to an external script, the post-processing steps will be executed by invoking the external script through the function 'post_process_script_function'. The post-processing steps are executed AFTER the files are copied and logic applied to the preservation storage.
     - curation_storage_location - required: The file system location where the Curation files reside

app.py

@@ -146,11 +146,14 @@ def main():
                           "Fetching articles...",
                           True)
     article_obj = Article(config, log, args.ids)
-    article_data = article_obj.get_articles()
+    article_data, already_preserved_counts_dict = article_obj.get_articles()
 
+    already_preserved_articles_count = len(already_preserved_counts_dict['already_preserved_article_ids'])
+    already_preserved_versions_count = already_preserved_counts_dict['already_preserved_versions']
     published_articles_count = 0
     published_articles_versions_count = 0
     published_unpublished_count = 0
+
     for i, (k, v) in enumerate(article_data.items()):
         published_unpublished_count += 1
         if len(v) > 0:
@@ -159,8 +162,8 @@ def main():
 
     log.write_log_in_file('info', "Fetched: "
                           + f"Total articles: {published_unpublished_count}, "
-                          + f"Published articles: {published_articles_count}, "
-                          + f"Published article versions: {published_articles_versions_count}",
+                          + f"Published articles: {published_articles_count + already_preserved_articles_count}, "
+                          + f"Published article versions: {published_articles_versions_count + already_preserved_versions_count}",
                           True)
     print(" ")
 
@@ -181,18 +184,26 @@ def main():
     print(" ")
 
     # Start articles processing after completing fetching data from API
-    processed_articles_versions_count = article_obj.process_articles(article_data)
+    processed_articles_versions_count, ap_trust_preserved_article_version_count, wasabi_preserved_versions \
+        = article_obj.process_articles(article_data)
 
     # Start collections processing after completing fetching data from API and articles processing.
-    processed_collections_versions_count = collection_obj.process_collections(collection_data)
+    # processed_collections_versions_count, already_preserved_collection_versions_count, preserved_collection_versions_in_wasabi \
+    #     = collection_obj.process_collections(collection_data)
+    processed_collections_versions_count, already_preserved_collections_counts = collection_obj.process_collections(collection_data)
+    already_preserved_collections = len(already_preserved_collections_counts['already_preserved_collection_ids'])
+    already_preserved_collection_versions = already_preserved_collections_counts['already_preserved_versions']
+    preserved_collection_versions_in_wasabi = already_preserved_collections_counts['wasabi_preserved_versions']
+    preserved_collection_versions_in_ap_trust = already_preserved_collections_counts['ap_trust_preserved_versions']
 
     log.write_log_in_file('info', '------- Summary -------')
     log.write_log_in_file('info',
                           "Total articles/published articles: \t\t\t\t\t\t"
-                          + f'{published_unpublished_count} / {published_articles_count}',
+                          + f'{published_unpublished_count} / {published_articles_count + already_preserved_articles_count}',
                           True)
+
     log.write_log_in_file('info',
-                          "Total processed articles bags already in preservation storage: \t\t\t"
+                          "Total processed articles bags already in preservation storage found by Bagger: \t\t\t"
                           + f'{article_obj.processor.duplicate_bag_in_preservation_storage_count}',
                           True)
     log.write_log_in_file('info',
@@ -203,12 +214,31 @@ def main():
                           "Total articles versions processed/matched: \t\t\t\t\t"
                           + f'{processed_articles_versions_count} / {article_obj.no_matched}',
                           True)
+    log.write_log_in_file('info',
+                          "Total count of already preserved articles versions in AP Trust: \t\t\t\t\t\t"
+                          + f'{ap_trust_preserved_article_version_count}',
+                          True)
+    log.write_log_in_file('info',
+                          "Total count of already preserved articles versions in Wasabi: \t\t\t\t\t\t"
+                          + f'{wasabi_preserved_versions}',
+                          True)
+
+    log.write_log_in_file('info',
+                          "Total count of already preserved articles: \t\t\t\t\t\t"
+                          + f'{already_preserved_articles_count}',
+                          True)
+
+    log.write_log_in_file('info',
+                          "Total count of already preserved article versions: \t\t\t\t\t\t"
+                          + f'{already_preserved_versions_count}',
+                          True)
+
     log.write_log_in_file('info',
                           "Total articles versions unmatched (published-matched): \t\t\t\t"
                           + f'{article_obj.no_unmatched}',
                           True)
     log.write_log_in_file('info',
-                          "Total processed articles bags successfully preserved \t\t\t\t"
+                          "Total processed articles bags successfully preserved: \t\t\t\t"
                           + f'{article_obj.processor.bag_preserved_count}',
                           True)
     log.write_log_in_file('info',
@@ -221,9 +251,27 @@ def main():
                           True)
     log.write_log_in_file('info',
                           "Total collections already preserved: \t\t\t\t\t\t"
+                          + f'{already_preserved_collections}',
+                          True)
+    log.write_log_in_file('info',
+                          "Total collection versions already preserved: \t\t\t\t\t\t"
+                          + f'{already_preserved_collection_versions}',
+                          True)
+    log.write_log_in_file('info',
+                          "Total collections already preserved found by Bagger: \t\t\t\t\t\t"
                           + f'{collection_obj.processor.duplicate_bag_in_preservation_storage_count}',
                           True)
 
+    log.write_log_in_file('info',
+                          "Total count of already preserved collections versions in AP Trust: \t\t\t\t\t\t"
+                          + f'{preserved_collection_versions_in_ap_trust}',
+                          True)
+
+    log.write_log_in_file('info',
+                          "Total count of already preserved collections versions in Wasabi: \t\t\t\t\t\t"
+                          + f'{preserved_collection_versions_in_wasabi}',
+                          True)
+
     if processed_articles_versions_count != published_articles_versions_count or processed_collections_versions_count != collections_versions_count:
         log.write_log_in_file('warning',
                               'The number of articles versions or collections versions sucessfully processed is different'

figshare/Article.py

@@ -7,9 +7,13 @@
 import hashlib
 import re
 from figshare.Integration import Integration
+from figshare.Utils import standardize_api_result, sorter_api_result, get_preserved_version_hash_and_size
+from figshare.Utils import compare_hash, check_wasabi
 from slugify import slugify
 from requests.adapters import HTTPAdapter, Retry
 
+from ReBACH.figshare.Utils import calculate_payload_size
+
 
 class Article:
     api_endpoint = ""
@@ -25,6 +29,7 @@ class Article:
     def __init__(self, config, log, ids):
         self.config_obj = config
         figshare_config = self.config_obj.figshare_config()
+        self.aptrust_config = self.config_obj.aptrust_config()
         self.system_config = self.config_obj.system_config()
         self.api_endpoint = figshare_config["url"]
         self.api_token = figshare_config["token"]
@@ -47,6 +52,8 @@ def __init__(self, config, log, ids):
         self.matched_curation_folder_list = []
         self.no_matched = 0
         self.no_unmatched = 0
+        self.already_preserved_counts_dict = {'already_preserved_article_ids': set(), 'already_preserved_versions': 0,
+                                              'wasabi_preserved_versions': 0, 'ap_trust_preserved_versions': 0}
         self.processor = Integration(self.config_obj, self.logs)
 
     """
@@ -109,7 +116,7 @@ def get_articles(self):
                 if (retries > self.retries):
                     break
 
-        return article_data
+        return article_data, self.already_preserved_counts_dict
 
     def article_loop(self, articles, page_size, page, article_data):
         no_of_article = 0
@@ -149,6 +156,8 @@ def __get_article_versions(self, article):
                                 self.logs.write_log_in_file("info",
                                                             f"Fetching article {article['id']} version {version['version']}.", True)
                                 version_data = self.__get_article_metadata_by_version(version, article['id'])
+                                if version_data is None:
+                                    continue
                                 metadata.append(version_data)
                         else:
                             # This branch is for cases where the item has a total embargo and no versions are available via the public API
@@ -226,6 +235,8 @@ def private_article_for_data(self, private_url, article_id):
     :param version object value.
     :param article_id int value.
     On successful response from url_public_api API, metadata array will be setup for response.
+    Metadata hash is calculated and matched against preserved version copy's has if any, if
+    a match is found, the version will not be processed.
     If files aren't found and size is > 0 in public API response then
     private api will be called for files.
     No. of tries implemented in while loop, loop will exit if API is not giving
@@ -235,6 +246,7 @@ def private_article_for_data(self, private_url, article_id):
     def __get_article_metadata_by_version(self, version, article_id):
         retries = 1
         success = False
+        already_preserved = in_ap_trust = False
 
         while not success and retries <= int(self.retries):
             try:
@@ -243,6 +255,7 @@ def __get_article_metadata_by_version(self, version, article_id):
                     get_response = requests.get(public_url)
                     if (get_response.status_code == 200):
                         version_data = get_response.json()
+                        payload_size = calculate_payload_size(self.system_config, version_data)
                         total_file_size = version_data['size']
                         files = []
                         error = ""
@@ -258,14 +271,54 @@ def __get_article_metadata_by_version(self, version, article_id):
                             files = version_data['files']
 
                         version_md5 = ''
+                        version_data = standardize_api_result(version_data)
+                        version_data = sorter_api_result(version_data)
                         json_data = json.dumps(version_data).encode("utf-8")
                         version_md5 = hashlib.md5(json_data).hexdigest()
+                        preserved_version_md5, preserved_version_size \
+                            = get_preserved_version_hash_and_size(self.aptrust_config, article_id, version['version'])
+                        wasabi_preserved_version_md5, wasabi_preserved_size = check_wasabi(article_id, version['version'])
+
+                        # Compare hashes
+                        if compare_hash(version_md5, wasabi_preserved_version_md5) and compare_hash(version_md5, preserved_version_md5):
+                            already_preserved = in_ap_trust = True
+                            self.already_preserved_counts_dict['already_preserved_versions'] += 1
+                            self.already_preserved_counts_dict['wasabi_preserved_versions'] += 1
+                            self.already_preserved_counts_dict['ap_trust_preserved_versions'] += 1
+                            self.logs.write_log_in_file("info",
+                                                        f"Article {article_id} version {version['version']} \
+                                                        already preserved in Wasabi and AP Trust.", True)
+
+                        elif compare_hash(version_md5, wasabi_preserved_version_md5):  # Wasabi only check
+                            already_preserved = True
+                            in_ap_trust = False
+                            self.already_preserved_counts_dict['already_preserved_versions'] += 1
+                            self.already_preserved_counts_dict['wasabi_preserved_versions'] += 1
+                            self.logs.write_log_in_file("info", f"Article {article_id} version {version['version']} \
+                             already preserved in Wasabi.", True)
+
+                        elif compare_hash(version_md5, preserved_version_md5):  # AP Trust only check
+                            already_preserved = in_ap_trust = True
+                            self.already_preserved_counts_dict['already_preserved_versions'] += 1
+                            self.already_preserved_counts_dict['ap_trust_preserved_versions'] += 1
+                            self.logs.write_log_in_file("info",
+                                                        f"Article {article_id} version {version['version']} \
+                                                        already preserved in AP Trust.", True)
+
+                        if already_preserved:
+                            self.already_preserved_counts_dict['already_preserved_article_ids'].add(article_id)
+                            if in_ap_trust and preserved_version_size != payload_size:
+                                self.logs.write_log_in_file("warning",
+                                                            f"Article {article_id} version {version['version']} "
+                                                            f"found in AP Trust but sizes do not match.", True)
+                            return None
 
                         version_metadata = self.set_version_metadata(version_data, files, private_version_no, version_md5, total_file_size)
                         version_data['total_num_files'] = file_len
                         version_data['file_size_sum'] = total_file_size
                         version_data['version_md5'] = version_md5
-                        if (error):
+
+                        if error:
                             version_metadata['errors'] = []
                             version_metadata['errors'].append(error)
 
@@ -977,7 +1030,8 @@ def process_articles(self, articles):
                             if (value_post_process != 0):
                                 self.logs.write_log_in_file("error", f"{version_data['id']} version {version_data['version']} - "
                                                             + "Post-processing script failed.", True)
-        return processed_count
+        return processed_count, self.already_preserved_counts_dict['ap_trust_preserved_versions'], \
+            self.already_preserved_counts_dict['wasabi_preserved_versions']
 
     """
     Preservation and Curation directory access check while processing.