allow to use ingestor for TPZ online items and optionally send them to...

allow to use ingestor for TPZ online items and optionally send them to preprocess_queue instead of md_queue (if string provided)

ingestor/Dockerfile

@@ -42,20 +42,10 @@ RUN mkdir /ingestor
 COPY app.py config.py filedaemon.py /
 COPY ingestor/ /ingestor
 
-ENV COLLECTION_ID= \
-    INSTANCE_ID="prism-view-server_ingestor" \
-    RENDERER_HOST= \
-    COLLECTION= \
+ENV INSTANCE_ID="prism-view-server_ingestor" \
     REDIS_HOST= \
     REDIS_PORT="6379" \
     REDIS_PREPROCESS_MD_QUEUE_KEY="preprocess-md_queue" \
-    INOTIFY_WATCH_DIR="/mnt/data" \
-    ST_AUTH_VERSION=3 \
-    OS_AUTH_URL= \
-    OS_USERNAME= \
-    OS_PASSWORD= \
-    OS_TENANT_NAME= \
-    OS_TENANT_ID= \
-    OS_REGION_NAME=
+    INOTIFY_WATCH_DIR="/mnt/data"
 
 CMD ["gunicorn3", "-c", "config.py", "app"]

ingestor/app.py

@@ -83,9 +83,13 @@ def ingest():
         request.get_data()
         browse_report = parse_browse_report(request.data)
         logger.debug(browse_report)
-        client.lpush(queue_name, json.dumps(
-            browse_report, default=converter
-        ))
+        if isinstance(browse_report, str):
+            queue_content = browse_report
+        else:
+            queue_content = json.dumps(
+                browse_report, default=converter
+            )
+        client.lpush(queue_name, queue_content)
         return Response(status=202)
 
     except Exception as e:

ingestor/filedaemon.py

@@ -86,9 +86,13 @@ class EventHandler(pyinotify.ProcessEvent):
             with open(event.pathname) as f:
                 browse_report = parse_browse_report(f)
             logger.debug(browse_report)
-            client.lpush(queue_name, json.dumps(
-                browse_report, default=converter
-            ))
+            if isinstance(browse_report, str):
+                queue_content = browse_report
+            else:
+                queue_content = json.dumps(
+                    browse_report, default=converter
+                )
+            client.lpush(queue_name, queue_content)
         except Exception as e:
             logger.exception(e)
 

ingestor/ingestor/browse_report.py

@@ -36,28 +36,30 @@ from .util import pairwise
 class BrowseReportParserError(ValueError):
     pass
 
+
 NS_REP = 'http://ngeo.eo.esa.int/ngEO/browseReport/1.0'
 NS_REP_OLD = 'http://ngeo.eo.esa.int/schema/browseReport'
 NS_BSI = 'http://ngeo.eo.esa.int/schema/browse/ingestion'
-
-
-nsmap = {
-    'rep': NS_REP,
-    'bsi': NS_BSI
-}
+NS_TPZ = 'http://www.telespazio.com/CSCDA/CDD/PDAS'
 
 
 def rep(tag):
     return f'{{{NS_REP}}}{tag}'
 
+
 def rep_old(tag):
     return f'{{{NS_REP_OLD}}}{tag}'
 
+
 def bsi(tag):
     return f'{{{NS_BSI}}}{tag}'
 
 
-ALLOWED_ROOT_TAGS = {rep('browseReport'), rep_old('browseReport'), bsi('ingestBrowse')}
+def tpz(tag):
+    return f'{{{NS_TPZ}}}{tag}'
+
+
+ALLOWED_ROOT_TAGS = {rep('browseReport'), rep_old('browseReport'), bsi('ingestBrowse'), tpz('OnLineArchiveItem')}
 
 
 def parse_browse_report(input_file):
@@ -66,7 +68,6 @@ def parse_browse_report(input_file):
     """
     if isinstance(input_file, bytes):
         input_file = io.BytesIO(input_file)
-
     try:
         tree = etree.parse(input_file)
     except etree.XMLSyntaxError as e:
@@ -80,6 +81,12 @@ def parse_browse_report(input_file):
 
     if root.tag == rep_old('browseReport'):
         used_rep = rep_old
+    elif root.tag == tpz('OnLineArchiveItem'):
+        if root.findtext(tpz('status')) in ['added', 'updated']:
+            path_el = root.find(tpz('path')).findtext(tpz('URL'))
+            obs_path = path_el.replace('\n', '').replace('/', '', 1)
+            return obs_path
+        raise BrowseReportParserError('Status not in "added" or "updated", aborting.')
     else:
         used_rep = rep
 

ingestor/tests/data/tpzonlineitem.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<!--Sample XML file generated by XMLSpy v2013 sp1 (http://www.altova.com)-->
+<OnLineArchiveItem xsi:schemaLocation="http://www.telespazio.com/CSCDA/CDD/PDAS PDAS_interfaces%2020190924_1916.xsd" xmlns="http://www.telespazio.com/CSCDA/CDD/PDAS" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+	<identifier>urn:eop:SPOT:MULTISPECTRAL_4m:DS_SPOT6_201807121048474_FR1_FR1_FR1_FR1_E007N61_04388_projected_6f78
+</identifier>
+	<path>
+		<Service>OS</Service>
+		<URL>/data25/OA/SP06/1.0/58/urn:eop:SPOT:MULTISPECTRAL_4m:DS_SPOT6_201807121048474_FR1_FR1_FR1_FR1_E007N61_04388_projected_6f78/0000/SP06_NAO_MS4_2A_20180712T104910_20180712T104914_TOU_1234_6f78.DIMA.tar
+</URL>
+	</path>
+	<status>added</status>
+</OnLineArchiveItem>