fix: xmlupload crashes without writing id2iri mapping (DEV-813)

knora/dsplib/utils/xml_upload.py

@@ -5,12 +5,15 @@
 import json
 import os
 import re
+import time
 import uuid
 from datetime import datetime
 from pathlib import Path
-from typing import Optional, Union, cast, Tuple
+from typing import Optional, Union, cast, Tuple, Any
 from urllib.parse import quote_plus
 
+import requests
+
 from lxml import etree
 
 from knora.dsplib.models.connection import Connection
@@ -767,7 +770,15 @@ def update_xml_texts(
                         "@context": context
                     }
                     jsondata = json.dumps(jsonobj, indent=4, separators=(',', ': '), cls=KnoraStandoffXmlEncoder)
-                    new_value = con.put(path='/v2/values', jsondata=jsondata)
+                    new_value = None
+                    for _ in range(20):
+                        try:
+                            new_value = con.put(path='/v2/values', jsondata=jsondata)
+                            break
+                        except BaseException:
+                            print(f'{datetime.now().isoformat()}: Try reconnecting to DSP server...')
+                            time.sleep(1)
+                            continue
                     if not new_value:
                         print(f'ERROR while updating the xml text of {link_prop.name} of resource {resource.id}')
                     elif verbose:
@@ -799,7 +810,15 @@ def update_resptr_props(
                 '@context': context
             }
             jsondata = json.dumps(jsonobj, indent=4, separators=(',', ': '))
-            new_value = con.post(path='/v2/values', jsondata=jsondata)
+            new_value = None
+            for _ in range(20):
+                try:
+                    new_value = con.post(path='/v2/values', jsondata=jsondata)
+                    break
+                except BaseException:
+                    print(f'{datetime.now().isoformat()}: Try reconnecting to DSP server...')
+                    time.sleep(1)
+                    continue
             if not new_value:
                 print(f'ERROR while updating the resptr prop of {link_prop.name} of resource {resource.id}')
             elif verbose:
@@ -880,6 +899,9 @@ def xml_upload(input_file: str, server: str, user: str, password: str, imgdir: s
     # temporarily remove circular references, but only if not an incremental upload
     if not incremental:
         resources, stashed_xml_texts, stashed_resptr_props = remove_circular_references(resources, verbose)
+    else:
+        stashed_xml_texts = dict()
+        stashed_resptr_props = dict()
 
     sipi_server = Sipi(sipi, con.get_token())
 
@@ -897,8 +919,55 @@ def xml_upload(input_file: str, server: str, user: str, password: str, imgdir: s
         resclass_name_2_type[res_class_name] = project.get_resclass_type(res_class_name)
 
     res_iri_lookup: dict[str, str] = {}
+    failed_uploads: list[str] = []
+
+    try:
+        upload_resources(verbose, resources, imgdir, sipi_server, permissions_lookup,
+                         resclass_name_2_type, res_iri_lookup, con, failed_uploads)
+    except BaseException as err:
+        handle_upload_error(err, input_file, res_iri_lookup, failed_uploads, stashed_xml_texts, stashed_resptr_props)
+        exit(1)
+
+    # update the resources with the stashed XML texts
+    if len(stashed_xml_texts) > 0:
+        update_stashed_xml_texts(verbose, res_iri_lookup, con, stashed_xml_texts)
 
-    failed_uploads = []
+    # update the resources with the stashed resptrs
+    if len(stashed_resptr_props) > 0:
+        update_stashed_resptr_props(verbose, res_iri_lookup, con, stashed_resptr_props)
+
+    write_id2iri_mapping(input_file, res_iri_lookup, datetime.now().strftime("%Y%m%d-%H%M%S"))
+    if failed_uploads:
+        print(f"Could not upload the following resources: {failed_uploads}")
+
+
+def try_sipi_upload(sipi_server: Sipi, filepath: str) -> dict[Any, Any]:
+    img = None
+    for _ in range(20):
+        try:
+            img = sipi_server.upload_bitstream(filepath)
+            break
+        except BaseException:
+            print(f'{datetime.now().isoformat()}: Try reconnecting to DSP server (SIPI)...')
+            time.sleep(1)
+            continue
+    if img:
+        return img
+    else:
+        return sipi_server.upload_bitstream(filepath)
+
+
+def upload_resources(
+    verbose: bool,
+    resources: list[XMLResource],
+    imgdir: str,
+    sipi_server: Sipi,
+    permissions_lookup: dict[str, Permissions],
+    resclass_name_2_type: dict[str, type],
+    res_iri_lookup: dict[str, str],
+    con: Connection,
+    failed_uploads: list[str]
+) -> None:
     for resource in resources:
         if verbose:
             resource.print()
@@ -909,42 +978,49 @@ def xml_upload(input_file: str, server: str, user: str, password: str, imgdir: s
 
         resource_bitstream = None
         if resource.bitstream:
-            img = sipi_server.upload_bitstream(os.path.join(imgdir, resource.bitstream.value))
+            img = try_sipi_upload(sipi_server=sipi_server, filepath=os.path.join(imgdir, resource.bitstream.value))
             internal_file_name_bitstream = img['uploadedFiles'][0]['internalFilename']
             resource_bitstream = resource.get_bitstream(internal_file_name_bitstream, permissions_lookup)
 
         permissions_tmp = permissions_lookup.get(resource.permissions)
 
-        try:
-            # create a resource instance (ResourceInstance) from the given resource in the XML (XMLResource)
-            resclass_type = resclass_name_2_type[resource.restype]
-            properties = resource.get_propvals(res_iri_lookup, permissions_lookup)
-            resclass_instance: ResourceInstance = resclass_type(
-                con=con,
-                label=resource.label,
-                iri=resource_iri,
-                permissions=permissions_tmp,
-                bitstream=resource_bitstream,
-                values=properties
-            )
-            resclass_instance = resclass_instance.create()
-        except BaseError as err:
-            print(f"ERROR while trying to create resource '{resource.label}' ({resource.id}). "
-                  f"The error message was: {err.message}")
-            failed_uploads.append(resource.id)
-            continue
-        except Exception as exception:
-            print(f"EXCEPTION while trying to create resource '{resource.label}' ({resource.id}). "
-                  f"The exception message was: {exception}")
+        resclass_instance = None
+
+        for _ in range(20):
+            try:
+                # create a resource instance (ResourceInstance) from the given resource in the XML (XMLResource)
+                resclass_type = resclass_name_2_type[resource.restype]
+                properties = resource.get_propvals(res_iri_lookup, permissions_lookup)
+                resclass_instance: ResourceInstance = resclass_type(
+                    con=con,
+                    label=resource.label,
+                    iri=resource_iri,
+                    permissions=permissions_tmp,
+                    bitstream=resource_bitstream,
+                    values=properties
+                )
+                resclass_instance = resclass_instance.create()
+                break
+            except BaseError:
+                print(f'{datetime.now().isoformat()}: Try reconnecting to DSP server...')
+                time.sleep(1)
+                continue
+
+        if not resclass_instance:
+            print(f"ERROR while trying to create resource '{resource.label}' ({resource.id}). ")
             failed_uploads.append(resource.id)
-            continue
+        else:
+            res_iri_lookup[resource.id] = resclass_instance.iri
+            print(f"Created resource '{resclass_instance.label}' ({resource.id}) with IRI '{resclass_instance.iri}'")
 
-        res_iri_lookup[resource.id] = resclass_instance.iri
-        print(f"Created resource '{resclass_instance.label}' ({resource.id}) with IRI '{resclass_instance.iri}'")
 
-    # update the resources with the stashed XML texts
-    if len(stashed_xml_texts) > 0:
-        print('Update the stashed XML texts...')
+def update_stashed_xml_texts(
+    verbose: bool,
+    res_iri_lookup: dict[str, str],
+    con: Connection,
+    stashed_xml_texts: dict[XMLResource, dict[XMLProperty, dict[str, KnoraStandoffXml]]]
+) -> None:
+    print('Update the stashed XML texts...')
     for resource, link_props in stashed_xml_texts.items():
         print(f'Update XML text(s) of resource "{resource.id}"...')
         res_iri = res_iri_lookup[resource.id]
@@ -964,9 +1040,14 @@ def xml_upload(input_file: str, server: str, user: str, password: str, imgdir: s
             print(f'Exception while updating an XML text of resource "{resource.id}": {exception}')
             continue
 
-    # update the resources with the stashed resptrs
-    if len(stashed_resptr_props) > 0:
-        print('Update the stashed resptrs...')
+
+def update_stashed_resptr_props(
+    verbose: bool,
+    res_iri_lookup: dict[str, str],
+    con: Connection,
+    stashed_resptr_props: dict[XMLResource, dict[XMLProperty, list[str]]]
+) -> None:
+    print('Update the stashed resptrs...')
     for resource, prop_2_resptrs in stashed_resptr_props.items():
         print(f'Update resptrs of resource "{resource.id}"...')
         res_iri = res_iri_lookup[resource.id]
@@ -986,15 +1067,42 @@ def xml_upload(input_file: str, server: str, user: str, password: str, imgdir: s
             print(f'Exception while updating an XML text of resource "{resource.id}": {exception}')
             continue
 
-    # write mapping of internal IDs to IRIs to file with timestamp
-    timestamp_now = datetime.now()
-    timestamp_str = timestamp_now.strftime("%Y%m%d-%H%M%S")
 
+def handle_upload_error(
+    err: Any,
+    input_file: str,
+    res_iri_lookup: dict[str, str],
+    failed_uploads: list[str],
+    stashed_xml_texts: dict[XMLResource, dict[XMLProperty, dict[str, KnoraStandoffXml]]],
+    stashed_resptr_props: dict[XMLResource, dict[XMLProperty, list[str]]]
+) -> None:
+    print(f'xmlupload must be aborted because of the following error: {err}')
+    timestamp_str = datetime.now().strftime("%Y%m%d-%H%M%S")
+    write_id2iri_mapping(input_file, res_iri_lookup, timestamp_str)
+    if len(stashed_xml_texts) > 0:
+        with open(f'stashed_xml_texts_{timestamp_str}.txt', 'a') as f:
+            for res, props in stashed_xml_texts.items():
+                f.write(f'\n{res.id}\n---------\n')
+                for prop, stashed_texts in props.items():
+                    f.write(f'{prop.name}\n')
+                    for name, standoff in stashed_texts.items():
+                        f.write(f'\t{name}: {standoff}')
+    if len(stashed_resptr_props) > 0:
+        with open(f'stashed_resptr_props_{timestamp_str}.txt', 'a') as f:
+            for res, props_ in stashed_resptr_props.items():
+                f.write(f'\n{res.id}\n---------\n')
+                for prop, stashed_props in props_.items():
+                    f.write(f'{prop.name}\n\t{stashed_props}')
+
+    if failed_uploads:
+        print(f"Independently of this error, there were some resources that could not be uploaded: "
+              f"{failed_uploads}")
+
+
+def write_id2iri_mapping(input_file: str, res_iri_lookup: dict[str, str], timestamp_str: str) -> None:
+    # write mapping of internal IDs to IRIs to file with timestamp
     xml_file_name = Path(input_file).stem
     res_iri_lookup_file = "id2iri_" + xml_file_name + "_mapping_" + timestamp_str + ".json"
     with open(res_iri_lookup_file, "w") as outfile:
         print(f"============\nThe mapping of internal IDs to IRIs was written to {res_iri_lookup_file}")
         outfile.write(json.dumps(res_iri_lookup))
-
-    if failed_uploads:
-        print(f"Could not upload the following resources: {failed_uploads}")